#!/usr/bin/python # -*- coding=utf-8 -*- # @Create Time: 2024-01-22 13:47:11 # @Last Modified time: 2024-02-02 13:20:51 import pandas as pd # 用户活跃度分析 # 第一个功能：统计指定时间范围内的访问次数最多的10个用户 # 第二个功能：统计最近一个周期相对上一个周期访问次数增长最多的10个用户 class UserActivateDegreeAnalyzeSpace(object): """用户访问分析 接收用户创建的分析任务 时间范围：起始时间~结束时间 在接收用户创建的任务请求之后，会将任务信息插入MD的task表中，任务参数以JSON格式封装 执行submit，将taskid作为参数传递给submit """ def __init__(self, user: str): self.user = user self.pool_size = 10 self.DataStream = pd.DataFrame(columns=['time', 'cur_url', 'tar_url', 'dev_id', 'req_met', 'eve_typ', 'for_dat', 'sta_cod', 'ifr_url', 'but_txt', 'is_op', 'is_bus', 'is_bias']) # 设置变量名称无关化 def push(self, item: dict) -> None: self.DataStream.loc[len(self.DataStream.index)] = { "time": item['record_time'], # 请求时间 "cur_url": item['current_url'], # 当前URL "tar_url": item['target_url'], # 目标URL "ifr_url": item['iframe_url'], # 内嵌URL "req_met": item['request_method'], # 请求方法 "for_dat": item['form_data'], # 表单数据 "but_txt": item['button_text'], # 按钮文本 "is_op": False, # 是否为人工操作 "is_bus": False, # 是否为业务项 "is_bias": False, # 是否偏离 } # 获取指定日期范围内的用户访问行为数据 def getActionByDateRange(self) -> pd.DataFrame: """获取指定日期范围内的用户访问行为数据 """ # 1、拷贝数据,保存数据并清除数据 CopyDataStream = self.DataStream.copy(deep=True) # 清除数据 self.DataStream.drop(self.DataStream.index, inplace=True) # 保存数据到 HDF # CopyDataStream.to_hdf('DataStream.h5', key=self.user) CopyDataStream.to_json(f'DataStream-{self.user}.h5', orient='records', lines=True, force_ascii=False) # 2、URL分解，将URL的地址和参数进行分解 # print(CopyDataStream) # 当前URL参数分离 CopyDataStream['cur_que'] = CopyDataStream['cur_url'].apply(lambda x: x.split("?")[-1]) CopyDataStream['cur_url'] = CopyDataStream['cur_url'].apply(lambda x: x.split("?")[0]) # 目标URL参数分离 CopyDataStream['tar_que'] = CopyDataStream['tar_url'].apply(lambda x: x.split("?")[-1]) CopyDataStream['tar_url'] = CopyDataStream['tar_url'].apply(lambda x: x.split("?")[0]) # 内嵌URL参数分离 CopyDataStream['ifr_que'] = CopyDataStream['ifr_url'].apply(lambda x: x.split("?")[-1]) CopyDataStream['ifr_url'] = CopyDataStream['ifr_url'].apply(lambda x: x.split("?")[0]) # print(CopyDataStream) # 3、时序排列，将请求按照时间进行排序 # 设置时间格式 CopyDataStream['time'] = pd.to_datetime(CopyDataStream['time']) CopyDataStream['idx'] = pd.to_datetime(CopyDataStream['time']) # 时间排序 # CopyDataStream = CopyDataStream.sort_values(by='time') # CopyDataStream = CopyDataStream.reset_index() CopyDataStream = CopyDataStream.set_index('idx') CopyDataStream.to_csv('debug.csv', index=True, columns=['cur_url', 'tar_url']) # 4、功能判定 # 设定当前时间 curtime = CopyDataStream.iloc[0]['time'] # 设定当前URL # current = CopyDataStream.iloc[0]['cur_url'] current = '' # 设定目标 URL 集 target = dict() # 结果缓存 result = pd.DataFrame() # 将第一个请求放入结果 result = result.append([CopyDataStream.iloc[0]]) result.iloc[ len(result.index) -1 ]['is_op'] = True # 遍历数据 for idx, item in enumerate(CopyDataStream.itertuples()): # 判断是否营销域 if ('10.150.23.1:8010' in item.cur_url) and ('_INVOKE_FUNC_TITLE_' in item.for_dat): current = item.cur_url result = result.append([CopyDataStream.iloc[idx]]) # 判断是否重新上线 elif (item.time - curtime).seconds > 300: # 请求时间距离上次请求相隔大于300秒，判断上次请求以离线 result = result.append([CopyDataStream.iloc[idx -1]]) # 将上次的请求记录下来 result.iloc[ len(result.index) -1 ]['is_op'] = True # 变更记录为人工 result = result.append([CopyDataStream.iloc[idx]]) # 将新请求记录下来（新记录为人工操作） result.iloc[ len(result.index) -1 ]['is_op'] = True # 变更记录为人工 current = item.cur_url # 当前地址修改 target = dict() # 清除历史记录 target[item.tar_url] = idx # 添加当前记录 elif item.cur_url == current: # 当前地址没变，页面无跳转（需判断AJAX，需判断请求时间） if item.tar_url not in target: # 目标URL不在临时目录 target[item.tar_url] = idx # 添加目标URL到临时目录 else: # 当前地址发生变化，页面跳转 if target.get(current): # 查看是否有请求当前URL的历史记录，有历史记录则该历史记录就为页面变更记录 result = result.append([CopyDataStream.iloc[target.get(current)]]) result.iloc[ len(result.index) -1 ]['is_op'] = True # 变更记录为人工 elif '4a.gd.csg.local' in item.cur_url: # 未找到历史记录，该页面并非由请求获得，判定是否首次登录 print("*********是首次登录*********") # 首次登录 result = result.append([CopyDataStream.iloc[idx]]) # 记录当前变更项 result.iloc[ len(result.index) -1 ]['is_op'] = True # 变更记录为人工 else: # 非正常跳转（既不是由统一认证平台进入、也不是直接点击跳转进入） print("*********非正常跳转*********") # 非正常跳转 target = dict() # 清除历史记录 target[item.tar_url] = idx # 添加当前记录 current = item.cur_url # 当前地址修改 curtime = item.time # 修正时间戳 # 结果返回 print(result) return result def getSession2Action(self): """获取sessionid到访问行为数据的映射的 """ pass def aggregateBySession(self): """对行为数据按session粒度进行聚合,并将用户信息join后聚合 """ pass def filterSessionAndAggrStat(self): """过滤session数据，并进行计数器值累加 """ pass def getSession2detail(self): """获取通过筛选条件的session的访问明细数据 """ pass def getTopKCategory(self): """获取topK热门分类 第一步：获取符合条件的访问过的所有品类 第二步：计算各品类的点击的次数 第三步：自定义二次排序key 第四步：将数据映射成格式的RDD，然后进行二次排序（降序） 第五步：取出top10热门品类，并写入DM """ pass def getClickCategory2Count(self): """获取各分类点击次数 1、过滤出点击行为数据，点击数据只占总数据的一小部分，所以过滤后数据可能不均匀 """ pass def getTopKSession(self): """获取每个分类top10活跃用户 1、将topK热门分类的id，生成DF 2、计算topK分类被各用户访问的次数 3、分组取TopK算法实现，获取每个分类的topK活跃用户 4、获取topK活跃用户的明细数据，并写入DM """ pass