import asyncio import os import sys import time from datetime import datetime import orjson from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.triggers.interval import IntervalTrigger from loguru import logger sys.path.append('/app') from utils.feishu_utils import Feishu from utils.google_ai_studio import GoogleAI from utils.gpt4o_mimi import GPT4oMini from utils.piaoquan import PQ from utils.redis import RedisHelper class ConsumptionRecommend(object): @classmethod async def run(cls): logger.info(f"[处理] 开始获取redis数据") task = await RedisHelper().get_client().rpop(name = 'gong_ji_heng_ceng:scan_tasks') 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, cover_path, old_title = PQ.get_pq_oss(video_id) if not video_path: return logger.info(f"[处理] 获取原视频OSS地址,视频链接:{video_path},封面链接:{cover_path}") video_url = f"http://rescdn.yishihui.com/{video_path}" logger.info(f"[处理] 开始提取口播文案") api_key = os.getenv("GEMINI_API_KEY") logger.info(f"[处理] 使用的API_KEY:{api_key}") text = GoogleAI.run(api_key, video_url) if not text: logger.error(f"[处理] 提取口播文案失败") return if "视频下载失败" == text: logger.error(f"[处理] 视频下载失败") return logger.info(f"[处理] 提取口播文案完成") logger.info(f"[处理] 口播文案通过gpt开始生成标题") new_title = GPT4oMini.get_ai_mini_title(text) if not new_title: logger.error(f"[处理] 口播文案通过gpt无法生成标题") return logger.info(f"[处理] 口播文案通过gpt生成标题完成,{new_title}") n_id = "78354423" logger.info(f"[处理] 开始写入票圈后台") # logger.info(f"[处理] 开始修改票圈后台标题") # PQ.update_pq_title(video_id, new_title) # logger.info(f"[处理] 修改票圈后台标题成功") code_vid = PQ.install_tj_pq(video_id, video_path, new_title, n_id, cover_path) if not code_vid: logger.error(f"[处理] 写入票圈后台失败") return logger.info(f"[处理] 写入票圈后台成功,视频id:{code_vid}") tag = f"lev-供给,rol-机器,#str-内容理解优化标题_51,{video_id}" PQ.video_tag(code_vid,tag) logger.info(f"[处理] 视频写入标签成功") current_time = datetime.now() formatted_time = current_time.strftime("%Y-%m-%d %H:%M:%S") logger.info(f"[处理] 开始写入飞书表格") values = [[video_id, code_vid ,f"https://admin.piaoquantv.com/cms/post-detail/{code_vid}/detail",old_title,new_title,task['channel'],task["time"],task["partition"],formatted_time]] Feishu.insert_columns("JY4esfYvShLbTkthHEqcTw5qnsh", "qFpmD4", "ROWS", 1, 2) time.sleep(0.5) Feishu.update_values("JY4esfYvShLbTkthHEqcTw5qnsh", "qFpmD4", "A2:Z2", values) logger.info(f"[处理] 写入飞书表格成功") async def run(): scheduler = AsyncIOScheduler() try: logger.info(f"[处理] 开始启动") scheduler.add_job(ConsumptionRecommend.run, trigger=IntervalTrigger(minutes=5)) # 每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())