supply_cross_transform/workers/consumption_work.py

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"[处理] 开始写入票圈后台")
        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=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())