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())