#!/usr/bin/python # -*- coding=utf-8 -*- # @Create Time: 2024-01-22 13:47:11 # @Last Modified time: 2024-02-23 10:28:27 import os, json 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'] ) self.loadHeatMapByJson() # 设置变量名称无关化 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 read_from_json(self) -> None: self.DataStream = pd.read_json(f'/root/mzinfo/V2/{self.user}_data.json', orient='records', lines=True) self.DataStream['time'] = self.DataStream['record_Time'] self.DataStream['cur_url'] = self.DataStream['current_Url'] self.DataStream['tar_url'] = self.DataStream['target_Url'] self.DataStream['ifr_url'] = self.DataStream['iframe_Url'] self.DataStream['req_met'] = self.DataStream['request_Method'] self.DataStream['for_dat'] = self.DataStream['form_Data'] self.DataStream['but_txt'] = self.DataStream['button_Text'] self.DataStream['is_op'] = False self.DataStream['is_bus'] = False self.DataStream['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.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() # 设定按键文本 curbut = '' # 结果缓存 result = 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', 'cur_que', 'tar_que', 'ifr_que', 'acc']) # 遍历数据 for idx, item in enumerate(CopyDataStream.itertuples()): # 当前地址发生变化，页面跳转（真实可靠） if item.cur_url != current: # 当前地址发生变化，页面跳转 if target.get(current): # 查看是否有请求当前URL的历史记录，有历史记录则该历史记录就为页面变更记录 result.loc[len(result.index)] = CopyDataStream.iloc[target.get(current)] result.at[len(result.index) - 1, 'is_op'] = True # 变更记录为人工 elif '4a.gd.csg.local' in item.cur_url: # 未找到历史记录，该页面并非由请求获得，判定是否首次登录 # print("*********是首次登录*********") # 首次登录 result.loc[len(result.index)] = CopyDataStream.iloc[idx]# 记录当前变更项 result.at[len(result.index) - 1, 'is_op'] = True # 变更记录为人工 # print(result.loc[ len(result.index) -1 ]) else: # 非正常跳转（既不是由统一认证平台进入、也不是直接点击跳转进入） # print("*********非正常跳转*********") # 非正常跳转 result.loc[len(result.index)] = CopyDataStream.iloc[idx]# 记录当前变更项 # print(result.loc[ len(result.index) -1 ]) result.at[len(result.index) - 1, 'acc'] = 'url' curbut = item.but_txt # 当前文本修改 target = dict() # 清除历史记录 target[item.tar_url] = idx # 添加当前记录 current = item.cur_url # 当前地址修改 # 点击文本发生变动，操作发生变化（无法判断第一条数据的真实可靠性） elif item.but_txt != curbut: curbut = item.but_txt # 当前文本修改 result.loc[len(result.index)] = CopyDataStream.iloc[idx] # 记录当前变更项 result.at[len(result.index) - 1, 'is_op'] = True # 变更记录为人工 result.at[len(result.index) - 1, 'acc'] = 'btx' # 判断是否重新上线（无法保证数据的可靠性） elif (item.time - curtime).seconds > 600: # 请求时间距离上次请求相隔大于600秒，判断上次请求以离线 result.loc[len(result.index)] = CopyDataStream.iloc[idx -1] # 将上次的请求记录下来 result.at[len(result.index) - 1, 'is_op'] = True # 变更记录为人工 result.at[len(result.index) - 1, 'is_bias'] = True # 用户超时行为偏离 result.at[len(result.index) - 1, 'acc'] = 'tim' # print(result.loc[ len(result.index) -1 ]) result.loc[len(result.index)] = CopyDataStream.iloc[idx] # 将新请求记录下来（新记录为人工操作） result.at[len(result.index) - 1, 'is_op'] = True # 变更记录为人工 result.at[len(result.index) - 1, 'acc'] = 'tim' # print(result.loc[ len(result.index) -1 ]) curbut = item.but_txt # 当前文本修改 current = item.cur_url # 当前地址修改 target = dict() # 清除历史记录 target[item.tar_url] = idx # 添加当前记录 # 判断是否营销域（真实可靠） elif '_INVOKE_FUNC_TITLE_' in item.for_dat: current = item.cur_url curbut = item.but_txt # 当前文本修改 result.loc[len(result.index)] = CopyDataStream.iloc[idx] # 记录当前变更项 result.at[len(result.index) - 1, 'is_op'] = True # 变更记录为人工 result.at[len(result.index) - 1, 'is_bus'] = True # 变更记录为人工 result.at[len(result.index) - 1, 'acc'] = 'form' # 判断是否appCode # elif 'appCode' in item.for_dat and item.but_txt != curbut: # current = item.cur_url # result.loc[len(result.index)] = CopyDataStream.iloc[idx] # result.at[len(result.index) - 1, 'is_op'] = True # 变更记录为人工 # result.at[len(result.index) - 1, 'is_bus'] = True # 变更记录为人工 # 当前地址没变，页面无跳转（需判断AJAX，需判断请求时间） elif item.cur_url == current: # 当前地址没变，页面无跳转（需判断AJAX，需判断请求时间） if item.tar_url not in target: # 目标URL不在临时目录 target[item.tar_url] = idx # 添加目标URL到临时目录 curtime = item.time # 修正时间戳 # # 保存文件 # result.to_csv(f'{self.user}.csv', sep=',') # 结果暂存 self.result = result # 结果返回 return result # 生成行为概率图 def getClickCategory2Count(self) -> None: """获取各分类点击次数 1、过滤出点击行为数据，点击数据只占总数据的一小部分，所以过滤后数据可能不均匀 2、对点击行为进行计数 3、保存计数器 """ for row in self.result.itertuples(): # 1、过滤出点击行为数据 key = row.cur_url.split("?")[0] + '-' + row.tar_url.split("?")[0] + '-' + row.acc # 2、对点击行为进行计数 if self.heatmap.get(key): self.heatmap[key] += 1 self.heatmap['total'] += 1 else: self.heatmap[key] = 1 self.heatmap['total'] += 1 # 3、保存计数器 self.saveHeatMap2Json() # 加载或初始化行为概率图 def loadHeatMapByJson(self) -> None: """加载点击行为计数器 如果有概率图，就加载概率图，如果没有，就初始化概率图 """ if os.path.exists('heatmap.json'): with open('heatmap.json', 'r', encoding='utf-8') as fp: self.heatmap = json.load(fp) else: self.heatmap = dict() # 设置非零数保证除法正常 self.heatmap['total'] = 1 # 保存行为概率图 def saveHeatMap2Json(self) -> None: """保存点击行为计数器 """ with open('heatmap.json', 'w', encoding='utf-8') as fp: json.dump(self.heatmap, fp) # 获取某个行为的概率(无权重计算) def getHeatMapValue(self, key: str) -> float: """获取某个行为的概率 """ value = self.heatmap.get(key) total = self.heatmap['total'] if value: return value / total else: return 0. def getTopKSession(self): """获取每个分类top10活跃用户 1、将topK热门分类的id，生成DF 2、计算topK分类被各用户访问的次数 3、分组取TopK算法实现，获取每个分类的topK活跃用户 4、获取topK活跃用户的明细数据，并写入DM """ pass 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