video-insight/workers/test_select.py

import datetime
import sys
import time
import os

import schedule
from loguru import logger
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))

logger.add("/app/logs/select.log", rotation="10 MB")

sys.path.append('/app')
import os
print("Current working directory:", os.getcwd())

from utils.odps_data import OdpsDataCount
from utils.redis import RedisHelper

def requirement_insight():
    """视频需求点洞察"""
    try:
        dt = (datetime.datetime.now() - datetime.timedelta(days=1)).strftime('%Y%m%d')
        logger.info(f"视频需求点洞察")
        redis_task = "task:video_insight"
        redis_trigger_task = "task:video_trigger_insight"
        sql =f'select clickobjectid as video_id from loghubods.user_share_log where dt = {dt} and topic = "click" group by clickobjectid order by count(distinct machinecode) desc limit 1'
        data = OdpsDataCount.main(sql)
        if not data:
            return
        RedisHelper().get_client().rpush(redis_task, *data)
        RedisHelper().get_client().rpush(redis_trigger_task, *data)
        logger.info(f"[R] 写入Redis 成功 共写入 {len(data)} 条")
    except Exception as e:
        logger.error(f"[R] 写入Redis写入失败，失败信息{e}")

def schedule_tasks():
    schedule.every().day.at("01:00").do(requirement_insight)


if __name__ == "__main__":
    requirement_insight()