alg/process_data.py

"""
process the data to satisfy the lightgbm
"""
import datetime
import sys
import os
import json
import asyncio
import argparse
import time

from tqdm import tqdm
import jieba.analyse
import pandas as pd

sys.path.append(os.getcwd())

from functions import generate_label_date, generate_daily_strings, MysqlClient, MySQLClientSpider


class DataProcessor(object):
    """
    Insert some information to lightgbm_data
    """

    def __init__(self, ):
        self.client = MysqlClient()
        self.client_spider = MySQLClientSpider()
        self.label_data = {}

    def generate_train_label(self, item, y_ori_data, cate):
        """
        生成训练数据，用 np.array矩阵的方式返回，
        :return: x_train, 训练数据， y_train, 训练 label
        """
        video_id = item["video_id"]
        dt = item["dt"]
        useful_features = [
            "uid",
            "type",
            "channel",
            "fans",
            "view_count_user_30days",
            "share_count_user_30days",
            "return_count_user_30days",
            "rov_user",
            "str_user",
            "out_user_id",
            "mode",
            "out_play_cnt",
            "out_like_cnt",
            "out_share_cnt",
            "out_collection_cnt",
        ]
        spider_features = [
            "channel",
            "out_user_id",
            "mode",
            "out_play_cnt",
            "out_like_cnt",
            "out_share_cnt"
        ]
        user_features = [
            "uid",
            "channel",
            "fans",
            "view_count_user_30days",
            "share_count_user_30days",
            "return_count_user_30days",
            "rov_user",
            "str_user"
        ]
        match self.ll:
            case "all":
                item_features = [item[i] for i in useful_features]
            case "user":
                if item['type'] == "userupload":
                    item_features = [item[i] for i in user_features]
                else:
                    return None, None
            case "spider":
                if item['type'] == "spider":
                    item_features = [item[i] for i in spider_features]
                    lop, duration = self.cal_lop(video_id)
                    item_features.append(lop)
                    item_features.append(duration)
                else:
                    return None, None
        keywords_textrank = self.title_processor(video_id)
        if keywords_textrank:
            for i in range(3):
                try:
                    item_features.append(keywords_textrank[i])
                except:
                    item_features.append(None)
        else:
            item_features.append(None)
            item_features.append(None)
            item_features.append(None)

        label_dt = generate_label_date(dt)
        label_obj = y_ori_data.get(label_dt, {}).get(video_id)
        if label_obj:
            label = int(label_obj[cate]) if label_obj[cate] else 0
        else:
            label = 0
        return label, item_features

    def producer(self):
        """
        生成数据
        :return:none
        """
        # 把 label， video_title, daily_dt_str, 存储到 mysql 数据库中去
        label_path = "data/train_data/daily-label-20240101-20240325.json"
        with open(label_path, encoding="utf-8") as f:
            self.label_data = json.loads(f.read())

        def read_title(client, video_id):
            """
            read_title_from mysql
            """
            sql = f"""SELECT title from wx_video where id = {video_id};"""
            # print("title", sql)
            try:
                title = client.select(sql)[0][0]
                return title.strip()
            except Exception as e:
                print(video_id, "\t", e)
                return ""

        def generate_label(video_id, hourly_dt_str, label_info):
            """
            generate label daily_dt_str for mysql
            :param label_info:
            :param video_id:
            :param hourly_dt_str:
            :return: label, daily_dt_str
            """
            label_dt = generate_label_date(hourly_dt_str)
            label_obj = label_info.get(label_dt, {}).get(video_id)
            if label_obj:
                label = int(label_obj["total_return"]) if label_obj["total_return"] else 0
                # print(label)
            else:
                label = 0
            return label, label_dt

        def process_info(item_):
            """
            Insert data into MySql
            :param item_:
            """
            video_id, hour_dt = item_
            # print(type(video_id))
            label_info = self.label_data
            title = read_title(client=self.client, video_id=video_id)
            label, dt_daily = generate_label(str(video_id), hour_dt, label_info)

            insert_sql = f"""UPDATE lightgbm_data set video_title = '{title}',  label = '{label}', daily_dt_str = '{dt_daily}' where video_id = '{video_id}';"""
            # print(insert_sql)
            self.client_spider.update(insert_sql)

        select_sql = "SELECT video_id, hour_dt_str FROM lightgbm_data where label is NULL and hour_dt_str < '20240327';"
        init_data_tuple = self.client_spider.select(select_sql)
        init_list = list(init_data_tuple)
        for item in tqdm(init_list):
            try:
                process_info(item)
            except Exception as e:
                print("操作失败", e)
        #     time.sleep(0.5)
        # with ThreadPoolExecutor(max_workers=8) as Pool:
        #     Pool.map(process_info, init_list)


class SpiderProcess(object):
    """
    Spider Data Process and Process data for lightgbm training
    """

    def __init__(self):
        self.client_spider = MySQLClientSpider()
        self.spider_features = [
            "channel",
            "out_user_id",
            "mode",
            "out_play_cnt",
            "out_like_cnt",
            "out_share_cnt"
        ]

    def spider_lop(self, video_id):
        """
        Spider lop = like / play
        :param video_id:
        :return:
        """
        sql = f"""SELECT like_cnt, play_cnt, duration from crawler_video where video_id = '{video_id}';"""
        try:
            like_cnt, play_cnt, duration = self.client_spider.select(sql)[0]
            lop = (like_cnt + 700) / (play_cnt + 18000)
            return lop, duration
        except Exception as e:
            print(video_id, "\t", e)
            return 0, 0

    def spider_data_produce(self):
        """
        把 spider_duration 存储到数据库中
        :return:
        """
        select_sql = "SELECT video_id, video_title, channel, out_user_id, spider_mode, out_play_cnt, out_like_cnt, out_share_cnt FROM lightgbm_data WHERE type = 'spider';"
        data_list = self.client_spider.select(select_sql)
        df = []
        for line in tqdm(data_list):
            try:
                temp = list(line)
                video_id = line[0]
                title = line[1]
                lop, duration = self.spider_lop(video_id)
                title_tags = list(jieba.analyse.textrank(title, topK=3))
                temp.append(lop)
                temp.append(duration)
                if title_tags:
                    for i in range(3):
                        try:
                            temp.append(title_tags[i])
                        except:
                            temp.append(None)
                else:
                    temp.append(None)
                    temp.append(None)
                    temp.append(None)

                df.append(temp[1:])
            except:
                continue
        df = pd.DataFrame(df, columns=['title', 'channel', 'out_user_id', 'mode', 'out_play_cnt', 'out_like_cnt',
                                       'out_share_cnt', 'lop', 'duration', 'tag1', 'tag2', 'tag3'])
        df.to_excel("data/train_data/spider_data_{}.xlsx".format(datetime.datetime.today().strftime("y%m%d")))


class UserProcess(object):
    """
        User Data Process
        """

    def __init__(self):
        self.client = MysqlClient()
        self.user_features = [
            "uid",
            "channel",
            "user_fans",
            "user_view_30",
            "user_share_30",
            "user_return_30",
            "user_rov",
            "user_str",
            "user_return_videos_30",
            "user_return_videos_3",
            "user_return_3",
            "user_view_3",
            "user_share_3",
            "address"
        ]

    def title_processor(self, video_id):
        """
            通过 video_id 去获取title， 然后通过 title 再分词，把关键词作为 feature
            :param video_id: the video id
            :return: tag_list [tag, tag, tag, tag......]
            """
        sql = f"""SELECT title from wx_video where id = {video_id};"""
        try:
            title = self.client.select(sql)[0][0]
            keywords_textrank = jieba.analyse.textrank(title, topK=3)
            return list(keywords_textrank)
        except Exception as e:
            print(video_id, "\t", e)
            return []

    def user_data_process(self):
        """
        把 user_return_3, user_view_3, user_share_3
        user_return_videos_3, user_return_videos_30
        address 存储到 mysql 数据库中
        :return:
        """
        user_path = '/data'


if __name__ == "__main__":
    # D = DataProcessor()
    # D.producer()
    # parser = argparse.ArgumentParser()  # 新建参数解释器对象
    # parser.add_argument("--mode")
    # parser.add_argument("--category")
    # parser.add_argument("--dtype", default="whole")
    # args = parser.parse_args()
    # mode = args.mode
    # category = args.category
    # dtype = args.dtype
    # D = DataProcessor()
    # D.producer()
    S = SpiderProcess()
    S.spider_data_produce()
    # if mode == "train":
    #     print("Loading data and process for training.....")
    #     D = DataProcessor(flag="train", ll=category)
    #     D.producer("whole")
    # elif mode == "predict":
    #     print("Loading data and process for prediction for each day......")
    #     D = DataProcessor(flag="predict", ll=category)
    #     if dtype == "single":
    #         date_str = str(input("Please enter the date of the prediction"))
    #         D.producer(date_str)
    #     elif dtype == "days":
    #         start_date_str = str(input("Please enter the start date of the prediction"))
    #         end_date_str = str(input("Please enter the end date of the prediction"))
    #         dt_list = generate_daily_strings(start_date=start_date_str, end_date=end_date_str)
    #         for d in dt_list:
    #             D.producer()