mzinfo/V2/app/tasks/UserActivateDegreeAnalyze.py

#!/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
        第四步：将数据映射成<CategorySortKey,info>格式的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