video-insight/workers/consumption_work.py

import asyncio
import json
import os
import sys
import orjson
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.interval import IntervalTrigger
from loguru import logger
sys.path.append('/app')

from utils.aliyun_log import AliyunLogger
from utils.google_ai_studio import GoogleAI
from utils.piaoquan import PQ
from utils.redis import RedisHelper, content_video_data





class ConsumptionRecommend(object):
    @classmethod
    async def run(cls):
        logger.info(f"[处理] 开始获取redis数据")

        task = RedisHelper().get_client().rpop(name = 'task:video_insight')

        if not task:
            logger.info('[处理] 无待执行的扫描任务')
            return
        task = orjson.loads(task)
        logger.info(f"[处理] 获取redis数据{task}")
        video_id = task['video_id']
        logger.info(f"[处理] 开始获取原视频OSS地址")
        video_path = PQ.get_pq_oss(video_id)
        if not video_path:
            return
        logger.info(f"[处理] 获取原视频OSS地址，视频链接:{video_path}")
        video_url = f"http://rescdn.yishihui.com/{video_path}"
        logger.info(f"[处理] 开始分析视频")
        api_key = os.getenv("GEMINI_API_KEY")
        # api_key = 'AIzaSyBFLCKMLX-Pf1iXoC2e_rMDLbNhNG23vTk'
        logger.info(f"[处理] 使用的API_KEY:{api_key}")
        text = GoogleAI.run(api_key, video_url)
        if "[异常]" in text:
            content_video_data(json.dumps(task))
        AliyunLogger.logging(str(video_id), orjson.dumps(text).decode())
        logger.info(f"[处理] 写入日志成功")


async def run():
    scheduler = AsyncIOScheduler()
    try:
        logger.info(f"[处理] 开始启动")
        scheduler.add_job(ConsumptionRecommend.run, trigger=IntervalTrigger(minutes=2))  # 每2分钟启动一次
        scheduler.start()
        await asyncio.Event().wait()
    except KeyboardInterrupt:
        pass
    except Exception as e:
        logger.error(f"[处理] 启动异常，异常信息:{e}")
        pass
    finally:
        scheduler.shutdown()


if __name__ == '__main__':
    # asyncio.run(ConsumptionRecommend.run())
    loop = asyncio.get_event_loop()
    loop.run_until_complete(run())