top_automatic_traceability/job_top_day_redis.py

import json
import time
from datetime import datetime
from collections import defaultdict

import schedule
from loguru import logger
import json

from common.feishu_utils import Feishu
from common.redis import insert_job_data, get_top_data, get_llen_list, get_lrange_list


def jab_day_recommend():
    """获取每日每小时top前100数据"""
    try:
        logger.info(f"开始获取溯源到的数据")
        list_task = [{'task:top_data_ks_gjc': '77618314'}, {'task:top_data_dy_gjc':'77618313'}, {'task:top_data_ks_gz':'77618315'}, {'task:top_data_dy_gz':'77618317'}]
        for task in list_task:
            for key, value in task.items():

                count = get_llen_list(key)
                if count > 0:
                    result = []
                    channel = None
                    tag_transport_channel = None
                    top_task = get_lrange_list(key,count)
                    for top in top_task:
                        data = json.loads(top)
                        channel_account_id = data.get("channel_account_id", "")
                        tag_transport_channel = data.get("tag_transport_channel", "")
                        channel = data.get("channel", "")
                        result.append(channel_account_id)
                    result_string = ",".join(result)
                    result_string = ",".join(dict.fromkeys(result_string.split(",")))
                    values = [
                        [
                            "",
                            tag_transport_channel,
                            result_string,
                            "lev-供给,rol-机器,#str-搬运搜索top视频溯源账号_42,genMod-账号",
                            value,
                            "2",
                            "15",
                            "通用-分享到群",
                            "AI片尾引导",
                            "zhifeng_emo,sijia",
                            "",
                            "",
                            "AI标题",
                            "1",
                            "",
                            "王雪珂",
                            channel
                        ]
                    ]
                    Feishu.insert_columns("YqiSsMvvMhr5tCttL7tcRbHTnrd", "d79d48", "ROWS", 1, 2)
                    time.sleep(0.5)
                    Feishu.update_values("YqiSsMvvMhr5tCttL7tcRbHTnrd", "d79d48", "A2:Z2", values)
                    logger.info(f"[+] 成功写入改造飞书表格")

            else:
                continue
        logger.info(f"写入top前100数据成功")
        return
    except Exception as e:
        logger.error(f"写入飞书异常,异常信息{e}")
        return
def schedule_tasks():
    schedule.every().day.at("00:40").do(jab_day_recommend)


if __name__ == "__main__":
    # jab_day_recommend()
    schedule_tasks()  # 调用任务调度函数
    while True:
        schedule.run_pending()
        time.sleep(1)  # 每秒钟检查一次