alg/data_process/read_data_from_odps.py

"""
Created on Mon Mar 18, 2024
@author: luojunhui
Read data from odps and save to json file in local files
"""

import os
import sys
import json
from tqdm import tqdm

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

from functions import PyODPS, generate_hourly_strings, generate_daily_strings
from functions import MySQLClientSpider


class VideoDataGenerator(object):
    """
    生成训练数据，测试数据
    """

    def __init__(self):
        self.oo = PyODPS()
        # self.v_mysql = MysqlClient()
        self.spider_mysql = MySQLClientSpider()

    def insert_into_database(self, dt):
        """
        把小时级数据插入 MySQL 数据库
        :return:
        """
        data_list = self.get_hour_data(dt)
        for obj in tqdm(data_list):
            insert_sql = f"""
                INSERT INTO lightgbm_data
                (video_id, user_id,  type, channel, user_fans, user_view_30, user_share_30, user_return_30, user_rov, user_str, out_user_id, spider_mode, out_play_cnt, out_like_cnt, out_share_cnt, hour_dt_str)
                    values 
                ({obj['video_id']}, '{obj['uid']}', '{obj['type']}', '{obj['channel']}', {obj['fans']}, {obj['view_count_user_30days']}, {obj['share_count_user_30days']}, {obj['return_count_user_30days']}, {obj['rov_user']}, {obj['str_user']}, '{obj['out_user_id']}', '{obj['mode']}', {obj['out_play_cnt']}, {obj['out_like_cnt']}, {obj['out_share_cnt']}, '{obj['dt']}');
            """
            self.spider_mysql.update(insert_sql)

    def get_hour_data(self, dt):
        """
        获取小时级的新视频
        :param dt: 小时参数
        :return:
        """
        sql = f"""select * from loghubods.conten_quality_base_hour where dt = '{dt}';"""
        hour_data = self.oo.select(sql)
        result = []
        for line in hour_data:
            obj = {
                "uid": line["uid"] if line['uid'] else "",
                "video_id": line["videoid"] if line['videoid'] else "",
                "type": line["type"] if line['type'] else "",
                "channel": line["channel"] if line['channel'] else "",
                "fst": line["flowpool_start_type"],
                "fsl": line["flowpool_start_level"],
                "fet": line["flowpool_end_type"],
                "fel": line["flowpool_end_level"],
                "f_view": line["flowpool_distribute_view_times"],
                "f_share": line["flowpool_share_times"],
                "f_return": line["flowpool_return_users"],
                "f3_view": line["flowpool_3days_distribute_view_times"],
                "f3_share": line["flowpool_3days_share_times"],
                "f3_return": line["flowpool_3days_return_users"],
                "ros_dms": line["ros_dms"],
                "rov_dms": line["rov_dms"],
                "ros_sls": line["ros_sls"],
                "rov_sls": line["rov_sls"],
                "fans": line["fans"] if line["fans"] else 0,
                "view_count_user_30days": line["view_cnt_user_30days"] if line["view_cnt_user_30days"] else 0,
                "share_count_user_30days": line["share_cnt_user_30days"] if line["share_cnt_user_30days"] else 0,
                "return_count_user_30days": line["return_cnt_user_30days"] if line["return_cnt_user_30days"] else 0,
                "rov_user": line["rov_user"] if line["rov_user"] else 0,
                "str_user": line["str_user"] if line["str_user"] else 0, 
                "out_user_id": line["out_user_id"] if line["out_user_id"] else "",
                "mode": line["strategy"] if line["strategy"] else "",
                "out_play_cnt": line["out_play_cnt"] if line["out_play_cnt"] else 0,
                "out_like_cnt": line["out_like_cnt"] if line["out_like_cnt"] else 0,
                "out_share_cnt": line["out_share_cnt"] if line["out_share_cnt"] else 0,
                "out_collection_cnt": line["out_collection_cnt"],
                "up_level_time_hour": line["up_level_time_hour"],
                "dt": line["dt"],
            }
            result.append(obj)
        return result

    def get_daily_data(self, dt):
        """
        天级表里面存储了视频的表现 label， 通过小时级的 video_id 去获取视频的表现
        :param dt: 20240101
        :return: data_list
        """
        sql = f"""select * from loghubods.conten_quality_base where dt = '{dt}';"""
        data = self.oo.select(sql)
        result = [
            {
                "video_id": item["videoid"],
                "total_view": item["flowpool_distribute_view_times"],
                "total_share": item["flowpool_share_times"],
                "total_return": item["flowpool_return_users"],
                "3day_view": item["flowpool_3days_distribute_view_times"],
                "3day_share": item["flowpool_3days_share_times"],
                "3day_return": item["flowpool_3days_return_users"],
                "3day_up_level": item["up_level_3_days"],
                'uplevel': item['uplevel'],
                "dt": item["dt"],
            }
            for item in data
        ]
        return result

    def get_daily_user_info(self, dt):
        """
        获取用户的信息以及特征
        :param dt:
        """
        sql = f"""select * from loghubods.conten_quality_base_day_new where dt = '{dt}';"""
        data = self.oo.select(sql)
        result = [
            {
                "uid": item["uid"],
                "video_id": item["videoid"],
                "address": item["city"],
                "return_3days": item["return_count_user_3days"],
                "view_3days": item["view_count_user_3days"],
                "share_3days": item["share_count_3days"],
                "3day_return_500_videos": item["return_count_user_3days_videocnt"],
                "30day_return_2000_videos": item["return_count_user_30days_videocnt"],
                "dt": item["dt"],
            }
            for item in data
        ]
        return result


def save_daily_data(start_date, end_date, save_path):
    """
    获取日期范围内数据，并且保存到指定路径
    :param start_date:
    :param end_date:
    :param save_path:
    :return:
    """
    date_list = generate_daily_strings(start_date, end_date)
    V = VideoDataGenerator()
    L = {}
    for date_str in tqdm(date_list):
        L[date_str] = {}
        data_list = V.get_daily_data(date_str)
        for obj in tqdm(data_list):
            video_id = obj["video_id"]
            L[date_str][video_id] = obj
    with open(save_path, "w") as f:
        f.write(json.dumps(L, ensure_ascii=False))


def download_hour_video_data(date_str):
    """
    获取日期参数
    :param date_str:
    :return:
    """
    V = VideoDataGenerator()
    data_list = V.get_hour_data(date_str)
    L = []
    for obj in tqdm(data_list):
        L.append(obj)
    temp_path = "data/temp_data/hour_{}.json".format(date_str)
    with open(temp_path, "w") as f:
        f.write(json.dumps(L, ensure_ascii=False))


def save_hourly_data(start_date, end_date):
    """
    save hourly data
    :param start_date:
    :param end_date:
    :param save_path:
    :return:
    """
    # print(save_path)
    V = VideoDataGenerator()
    date_list = generate_hourly_strings(start_date, end_date)
    for obj in tqdm(date_list):
        V.insert_into_database(obj)
    # with ThreadPoolExecutor(max_workers=5) as Pool:
    #     Pool.map(V.insert_into_database, date_list)


def save_daily_user_info(start_date, end_date, save_path):
    """
    save daily user_info
    :param start_date:
    :param end_date:
    :param save_path:
    :return:
    """
    date_list = generate_daily_strings(start_date, end_date)
    V = VideoDataGenerator()
    L = {}
    for date_str in tqdm(date_list):
        L[date_str] = {}
        data_list = V.get_daily_user_info(date_str)
        for obj in tqdm(data_list):
            video_id = obj["video_id"]
            L[date_str][video_id] = obj
    with open(save_path, "w") as f:
        f.write(json.dumps(L, ensure_ascii=False))


if __name__ == "__main__":
    flag = int(
        input(
            "请输入标识符，输入 1 获取小时级数据\n输入 2 获取天级数据\n输入 3 获取用户信息数据: \n"
        )
    )
    if flag == 1:
        start = str(input("请输入开始字符串， 格式为 yyyymmddhh:\n"))
        end = str(input("请输入结束字符串, 格式为 yyyymmddhh: \n"))
        if len(start) == 10 and len(end) == 10:
            save_hourly_data(start, end)
        else:
            print("Time format is not ok")
    elif flag == 2:
        start = str(input("请输入开始字符串， 格式为 yyyymmdd:\n"))
        end = str(input("请输入结束字符串, 格式为 yymmdd: \n"))
        save_p = "data/train_data/daily-label-{}-{}.json".format(start, end)
        if len(start) == 8 and len(end) == 8:
            save_daily_data(start, end, save_p)
        else:
            print("Time format is not ok")
    elif flag == 3:
        start = str(input("请输入开始字符串， 格式为 yyyymmdd:\n"))
        end = str(input("请输入结束字符串, 格式为 yymmdd: \n"))
        save_p = "data/train_data/daily-user-info-{}-{}.json".format(start, end)
        if len(start) == 8 and len(end) == 8:
            save_daily_user_info(start, end, save_p)
        else:
            print("Time format is not ok")
    else:
        print("Input Error ! Make sure your input is 1 or 2 or 3!!")