carry_efficiency_tool/workers/consumption_work_studio.py

import os

import sys
import time
import uuid
from datetime import datetime

import orjson
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.interval import IntervalTrigger
from loguru import logger
sys.path.append('/app')
from utils.redis import RedisHelper
from utils.aliyun_log import AliyunLogger
from utils.aliyun_oss import Oss
from utils.download_video import DownLoad
from utils.feishu_utils import Feishu
from utils.ffmpeg import FFmpeg
from utils.google_ai_studio import GoogleAI
from utils.gpt4o_mini_help import GPT4oMini
from utils.piaoquan import PQ
from utils.sql_help import sqlCollect
from utils.tag_video import Tag
from utils.tts_help import TTS





CACHE_DIR = '/app/cache/'
# CACHE_DIR = '/Users/z/Downloads/'

class ConsumptionRecommend(object):
    @classmethod
    def insert_pq(cls, data, oss_object_key, title, cover):
        logger.info(f"[内容分析] 开始写入票圈")

        code = PQ.insert_piaoquantv(oss_object_key, title, '50322062', cover)
        if not code:
            logger.error(f"[内容分析] 写入票圈后台失败")
            text = (
                f"**渠道**: {data['channel']}\n"
                f"**内容**: {data}\n"
                f"**失败信息**: 写入票圈后台失败\n"
            )
            Feishu.finish_bot(text,
                              "https://open.feishu.cn/open-apis/bot/v2/hook/65bc5463-dee9-46d0-bc2d-ec6c49a8f3cd",
                              f"【 内容理解-{data['channel']}失败通知 】")
            return
        logger.info(f"[内容分析] 写入票圈成功，返回视频id{code}")
        tag_status = Tag.video_tag(code, "lev-供给,rol-机器,#str-搬运改造内容理解引导语实验_60")
        Tag.video_tag(data["videoid"], "lev-供给,rol-机器,#str-搬运改造内容理解引导语base_61")

        if tag_status == 0:
            logger.info(f"[内容分析] 写入标签成功，后台视频ID为{code}")
        try:
            current_time = datetime.now()
            formatted_time = current_time.strftime("%Y-%m-%d %H:%M:%S")
            sqlCollect.insert_machine_making_data(data["channel"], data["channel"], data["channel"],
                                                  data["videoid"], data["videoid"], "50322062",
                                                  title,
                                                  code,
                                                  formatted_time, title, oss_object_key)
            pq_url = f'https://admin.piaoquantv.com/cms/post-detail/{code}/detail'  # 站内视频链接
            values = [
                [
                    data["videoid"],
                    code,
                    data["channel"],
                    data["dt"],
                    formatted_time,
                    pq_url
                ]
            ]
            Feishu.insert_columns("R4dLsce8Jhz9oCtDMr9ccpFHnbI", '1Ycd37', "ROWS", 1, 2)
            time.sleep(0.5)
            Feishu.update_values("R4dLsce8Jhz9oCtDMr9ccpFHnbI", '1Ycd37', "A2:Z2", values)
            logger.info(f"[内容分析] 写入飞书成功")
            return
        except Exception as e:
            logger.error(f"[内容分析] 写入飞书失败{e}")
            return



    @classmethod
    def data_handle(cls, data, file_path):
        video_id = data["videoid"]
        AliyunLogger.logging(data["type"], "片尾引导", data["channel"], video_id, "扫描到一条视频", "2001", str(data))
        AliyunLogger.logging(data["type"], "片尾引导", data["channel"], video_id, "符合规则等待改造", "2004", str(data))
        logger.info(f"[内容分析] 获取{video_id}的视频链接")
        video_path, cover_path, old_title = PQ.get_pq_oss_path(video_id)
        if not video_path:
            AliyunLogger.logging(data["type"], "片尾引导", data["channel"], video_id, "没有获取到视频链接", "3001",
                                 str(data))
            text = (
                f"**渠道**: {data['channel']}\n"
                f"**内容**: {data}\n"
                f"**失败信息**: 没有获取到视频链接\n"
            )
            Feishu.finish_bot(text,
                              "https://open.feishu.cn/open-apis/bot/v2/hook/65bc5463-dee9-46d0-bc2d-ec6c49a8f3cd",
                              f"【 内容理解-{data['channel']}失败通知 】")

            return
        video_url = f"http://rescdn.yishihui.com/{video_path}"
        video_path = DownLoad.download_video(video_url, file_path, '', video_id)
        if not os.path.exists(video_path) or os.path.getsize(video_path) == 0:
            logger.error(f"[内容分析] {video_url}下载失败")
            AliyunLogger.logging(data["type"], "片尾引导", data["channel"], video_id, "视频下载失败等待重新处理",
                                 "3002",
                                 str(data))
            text = (
                f"**渠道**: {data['channel']}\n"
                f"**内容**: {data}\n"
                f"**失败信息**: 视频下载失败\n"
            )
            Feishu.finish_bot(text,
                              "https://open.feishu.cn/open-apis/bot/v2/hook/65bc5463-dee9-46d0-bc2d-ec6c49a8f3cd",
                              f"【 内容理解-{data['channel']}失败通知 】")
            return
        logger.info(f"[内容分析] {video_url}视频下载成功")
        logger.info(f"[内容分析] {video_url}开始处理标题")
        video_path = FFmpeg.video_640(video_path, file_path)
        logger.info(f"[内容分析] 视频更改分辨率处理成功")
        logger.info(f"[内容分析] 片尾引导-开始获取视频口播内容")
        video_text = GoogleAI.run("AIzaSyCor0q5w37Dy6fGxloLlCT7KqyEFU3PWP8", video_path)
        if not video_text:
            AliyunLogger.logging(data["type"], "片尾引导", data["channel"], video_id, "片尾引导,获取口播文案失败",
                                 "3003",
                                 str(data))
            text = (
                f"**渠道**: {data['channel']}\n"
                f"**内容**: {data}\n"
                f"**失败信息**: 获取口播文案失败\n"
            )
            Feishu.finish_bot(text,
                              "https://open.feishu.cn/open-apis/bot/v2/hook/65bc5463-dee9-46d0-bc2d-ec6c49a8f3cd",
                              f"【 内容理解-{data['channel']}失败通知 】")
            return
        logger.info(f"[内容分析] 片尾引导-开始获取AI片尾")
        pw_srt_text = GPT4oMini.get_content_understanding_pw(video_text)
        pw_url = TTS.get_pw_zm(pw_srt_text, 'zhifeng_emo')
        if not pw_url:
            logger.error(f"[内容分析] 片尾引导-片尾获取失败")
            data["transform_rule"] = "仅改造"
            AliyunLogger.logging(data["type"], "片尾引导", data["channel"], video_id, "片尾引导,片尾获取失败",
                                 "3003",
                                 str(data))
            text = (
                f"**渠道**: {data['channel']}\n"
                f"**内容**: {data}\n"
                f"**失败信息**: 片尾获取失败\n"
            )
            Feishu.finish_bot(text,
                              "https://open.feishu.cn/open-apis/bot/v2/hook/65bc5463-dee9-46d0-bc2d-ec6c49a8f3cd",
                              f"【 内容理解-{data['channel']}失败通知 】")

            return
        logger.info(f"[内容分析] 片尾引导-片尾获取成功")
        pw_srt = TTS.getSrt(pw_url)
        if not pw_srt:
            AliyunLogger.logging(data["type"], "片尾引导", data["channel"], video_id, "片尾引导,片尾音频获取失败",
                                 "3003",
                                 str(data))
            text = (
                f"**渠道**: {data['channel']}\n"
                f"**内容**: {data}\n"
                f"**失败信息**: 片尾音频获取失败\n"
            )
            Feishu.finish_bot(text,
                              "https://open.feishu.cn/open-apis/bot/v2/hook/65bc5463-dee9-46d0-bc2d-ec6c49a8f3cd",
                              f"【 内容理解-{data['channel']}失败通知 】")
            return
        pw_mp3_path = TTS.download_mp3(pw_url, file_path)
        if not pw_mp3_path:
            AliyunLogger.logging(data["type"], "片尾引导", data["channel"], video_id, "片尾引导,片尾音频下载失败",
                                 "3003",
                                 str(data))
            text = (
                f"**渠道**: {data['channel']}\n"
                f"**内容**: {data}\n"
                f"**失败信息**: 片尾音频下载失败\n"
            )
            Feishu.finish_bot(text,
                              "https://open.feishu.cn/open-apis/bot/v2/hook/65bc5463-dee9-46d0-bc2d-ec6c49a8f3cd",
                              f"【 内容理解-{data['channel']}失败通知 】")
            return
        logger.info(f"[内容分析] 片尾引导-片尾音频下载成功")
        logger.info(f"[内容分析] 片尾引导-片尾获取最后一帧成功")
        jpg_path = FFmpeg.video_png(video_path, file_path)  # 生成视频最后一帧jpg
        pw_path = FFmpeg.pw_video(jpg_path, file_path, pw_mp3_path, pw_srt)  # 生成片尾视频
        if not os.path.exists(pw_path) or os.path.getsize(pw_path) == 0:
            logger.error(f"[内容分析] 片尾引导-片尾拼接失败")
            AliyunLogger.logging(data["type"], "片尾引导", "", data["video_url"],
                                 "片尾引导,片尾拼接失败", "3003", str(data))
            text = (
                f"**渠道**: {data['channel']}\n"
                f"**内容**: {data}\n"
                f"**失败信息**: 片尾拼接失败\n"
            )
            Feishu.finish_bot(text,
                              "https://open.feishu.cn/open-apis/bot/v2/hook/65bc5463-dee9-46d0-bc2d-ec6c49a8f3cd",
                              f"【 内容理解-{data['channel']}失败通知 】")
            return
        logger.info(f"[内容分析] 片尾引导-合并开始拼接")
        video_path = FFmpeg.h_b_video(video_path, pw_path, file_path)
        if not os.path.exists(video_path) or os.path.getsize(video_path) == 0:
            logger.error(f"[内容分析] 片尾引导-添加片尾失败")
            text = (
                f"**渠道**: {data['channel']}\n"
                f"**内容**: {data}\n"
                f"**失败信息**: 添加片尾失败\n"
            )
            Feishu.finish_bot(text,
                              "https://open.feishu.cn/open-apis/bot/v2/hook/65bc5463-dee9-46d0-bc2d-ec6c49a8f3cd",
                              f"【 内容理解-{data['channel']}失败通知 】")
            return
        logger.info(f"[内容分析] 片尾引导-开始发送oss")
        oss_object_key = Oss.stitching_sync_upload_oss(video_path, str(uuid.uuid4()))  # 视频发送OSS
        status = oss_object_key.get("status")
        if status != 200:
            logger.error(f"[内容分析] 片尾引导-发送oss失败")
            AliyunLogger.logging(data["type"], "片尾引导", data["channel"], video_id, "片尾引导,发送oss失败",
                                 "3003",
                                 str(data))
            text = (
                f"**渠道**: {data['channel']}\n"
                f"**内容**: {data}\n"
                f"**失败信息**: 发送oss失败\n"
            )
            Feishu.finish_bot(text,
                              "https://open.feishu.cn/open-apis/bot/v2/hook/65bc5463-dee9-46d0-bc2d-ec6c49a8f3cd",
                              f"【 内容理解-{data['channel']}失败通知 】")
            return
        logger.info(f"[内容分析] 片尾引导-发送oss成功")
        oss_object_key = oss_object_key.get("oss_object_key")
        cls.insert_pq(data, oss_object_key, old_title, cover_path)
        return


    @classmethod
    def run(cls):
        logger.info(f"[内容分析] 开始获取redis数据")
        data = RedisHelper().get_client().rpop(name = "task:carry_redis_by_nrfx")
        if not data:
            logger.info('[内容分析] 无待执行的扫描任务')
            return
        data = orjson.loads(data)
        uid = str(uuid.uuid4())
        file_path = os.path.join(CACHE_DIR, uid)
        try:
            cls.data_handle(data, file_path)
            for filename in os.listdir(CACHE_DIR):
                # 检查文件名是否包含关键字
                if uid in filename:
                    file_path = os.path.join(CACHE_DIR, filename)
                    try:
                        # 删除文件
                        os.remove(file_path)
                        logger.info(f"已删除文件: {file_path}")
                    except Exception as e:
                        logger.error(f"删除文件时出错: {file_path}, 错误: {e}")
            return

        except Exception as e:
            for filename in os.listdir(CACHE_DIR):
                # 检查文件名是否包含关键字
                if uid in filename:
                    file_path = os.path.join(CACHE_DIR, filename)
                    try:
                        # 删除文件
                        os.remove(file_path)
                        logger.info(f"已删除文件: {file_path}")
                    except Exception as e:
                        logger.error(f"删除文件时出错: {file_path}, 错误: {e}")
            return


def run():
    scheduler = BlockingScheduler()
    try:
        logger.info(f"[内容分析] 开始启动")
        scheduler.add_job(ConsumptionRecommend.run, trigger=IntervalTrigger(minutes=5))  # 每5分钟启动一次
        scheduler.start()
    except KeyboardInterrupt:
        pass
    except Exception as e:
        logger.error(f"[内容分析] 启动异常，异常信息:{e}")
        pass
    finally:
        scheduler.shutdown()


if __name__ == '__main__':
    run()