comprehensive_video_rewriting/workers/consumption_work.py

import os
import random
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.download_video import DownLoad
from utils.feishu_form import Material
from utils.feishu_utils import Feishu
from utils.piaoquan import PQ
from utils.sql_help import sqlCollect
from utils.tag_video import Tag
from channel.dy import DY
from channel.dy_keyword import DyKeyword
from channel.ks import KS
from channel.ks_keyword import KsKeyword
from channel.ks_xcx import KSXCX
from utils.aliyun_oss import Oss
from utils.ffmpeg import FFmpeg
from utils.google_ai_studio import GoogleAI
from utils.gpt4o_mini_help import GPT4oMini
from utils.tts_help import TTS






CACHE_DIR = '/app/cache/'
# CACHE_DIR = '/Users/z/Downloads/'
class ConsumptionRecommend(object):

    @classmethod
    def generate_title(cls, video, title):
        """
        生成新标题
        """
        if video['old_title']:
            new_title = video['old_title'].strip().replace("\n", "") \
                .replace("/", "").replace("\\", "").replace("\r", "") \
                .replace(":", "").replace("*", "").replace("？", "") \
                .replace("?", "").replace('"', "").replace("<", "") \
                .replace(">", "").replace("|", "").replace(" ", "") \
                .replace("&NBSP", "").replace(".", "。").replace(" ", "") \
                .replace("'", "").replace("#", "").replace("Merge", "")
        else:
            return '这个视频，分享给我的老友，祝愿您能幸福安康'
        if title == "原标题":
            if not new_title:
                new_title = '这个视频，分享给我的老友，祝愿您能幸福安康'
        elif title == "AI标题":
            if not new_title:
                new_title = '这个视频，分享给我的老友，祝愿您能幸福安康'
            else:
                new_title = GPT4oMini.get_ai_mini_title(new_title)
        else:
            titles = title.split('/') if '/' in title else [title]
            new_title = random.choice(titles)
        return new_title

    @classmethod
    def insert_pq(cls, task, oss_object_key, title, tags, fs_channel_name, video, voice, fs_mark):
        logger.info(f"[+] 开始写入票圈")
        n_id = str(task["pd_id"])
        code = PQ.insert_piaoquantv(oss_object_key, title, n_id, None)
        if not code:
            logger.error(f"[机器改造] 写入票圈后台失败")
            text = (
                f"**通知类型**: 写入票圈后台失败\n"
                f"**负责人**: {fs_channel_name}\n"
                f"**渠道**: {task['channel']}\n"
                f"**视频主页ID**: {task['channel_url']}\n"
                f"**视频Video_id**: {video['video_id']}\n"
            )
            AliyunLogger.logging(task['channel'], fs_channel_name, task['channel_url'], video['video_id'],
                                 "视频下载失败", "3002", f"video_url:{video['video_url']}")

            Feishu.finish_bot(text,
                              "https://open.feishu.cn/open-apis/bot/v2/hook/e7697dc6-5254-4411-8b59-3cd0742bf703",
                              "【 机器改造通知 】")

        logger.info(f"[机器改造] 写入票圈成功，返回视频id{code}")
        log_data = f"user:{task['channel']},,video_id:{video['video_id']},,video_url:{video['video_url']},,ai_title:{title},,voice:{voice}"
        AliyunLogger.logging(task['channel'], fs_channel_name, task['channel_url'], video['video_id'], "视频改造成功", "1000", log_data, str(code))

        tag_status = Tag.video_tag(code, str(tags))
        if tag_status == 0:
            logger.info(f"[机器改造] 写入标签成功，后台视频ID为{code}")
        current_time = datetime.now()
        formatted_time = current_time.strftime("%Y-%m-%d %H:%M:%S")
        sqlCollect.insert_machine_making_data(fs_channel_name, task['task_mark'], task['channel'], task['channel_url'], video['video_id'], n_id, title, code,
                                              formatted_time, title, oss_object_key)
        sqlCollect.insert_task(task['task_mark'], task['channel_url'], fs_mark, task['channel'])  # 插入数据库

        try:
            current_time = datetime.now()
            formatted_time = current_time.strftime("%Y-%m-%d %H:%M:%S")
            pq_url = f'https://admin.piaoquantv.com/cms/post-detail/{code}/detail'  # 站内视频链接
            if '历史抖音' == task['channel'] or '历史快手' == task['channel']:
                explain = "历史爆款"
            else:
                explain = "新供给"
            values = [
                [
                    fs_channel_name,
                    task['task_mark'],
                    task['channel'],
                    task['channel_url'],
                    str(video['video_id']),
                    str(n_id),
                    video['old_title'],
                    "AI标题",
                    title,
                    str(code),
                    formatted_time,
                    video["rule"],
                    explain,
                    voice,
                    tags,
                    task['keyword_name'],
                    pq_url
                ]
            ]
            if fs_channel_name == "抖音品类账号":
                sheet = "905313"
            elif fs_channel_name == "快手品类账号":
                sheet = "JHVpNK"
            elif fs_channel_name == "抖音关键词搜索":
                sheet = "6nclDV"
            elif fs_channel_name == "快手关键词搜索":
                sheet = "PVd8nj"
            elif fs_channel_name == "快手小程序":
                sheet = "dEjDt1"
            elif fs_channel_name == "top溯源账号":
                sheet = "5tPopP"
            Feishu.insert_columns("L2KGsz5HzhDfyYtV9IRcLHjtnsg", sheet, "ROWS", 1, 2)
            time.sleep(0.5)
            Feishu.update_values("L2KGsz5HzhDfyYtV9IRcLHjtnsg", sheet, "A2:Z2", values)
            logger.info(f"[处理] 写入飞书成功")
        except Exception as e:
            logger.error(f"[处理] 写入飞书失败{e}")
            pass
        return


    @classmethod
    def get_channle_data(cls, task, fs_channel_name):
        channel = task['channel']
        if channel == "抖音" or channel == "历史抖音":
            return DY.get_dy_list(task, fs_channel_name)
        elif channel == "快手" or channel == "历史快手":
            return KS.get_ks_list(task, fs_channel_name)
        elif channel == "抖音搜索":
            return DyKeyword.get_key_word(task, fs_channel_name)
        elif channel == "快手搜索":
            return KsKeyword.get_key_word(task, fs_channel_name)
        elif channel == '快手小程序':
            return KSXCX.get_xcx_date(task, fs_channel_name)


    @classmethod
    def data_handle(cls, task, file_path, fs_channel_name, ai_key, fs_mark):
        logger.info(f"[机器改造] 开始获取视频数据")
        data_list = cls.get_channle_data(task, fs_channel_name)
        if not data_list:
            logger.info(f"[+]  {fs_channel_name}-{task['channel_url']}获取视频完成，没有符合/改造视频")
            AliyunLogger.logging(task['channel'], fs_channel_name, task['channel_url'], "", "无改造视频", "4000")
            text = (
                f"**通知类型**: 没有改造的视频\n"
                f"**负责人**: {fs_channel_name}\n"
                f"**渠道**: {task['channel']}\n"
                f"**视频主页ID**: {task['channel_url']}\n"
            )
            Feishu.finish_bot(text, "https://open.feishu.cn/open-apis/bot/v2/hook/e7697dc6-5254-4411-8b59-3cd0742bf703",
                              "【 机器改造通知 】")
            return
        for video in data_list:
            file_path =  file_path + str(uuid.uuid4())
            logger.info(f"[机器改造] {video['video_url']}视频开始下载")
            video_path = DownLoad.download_video(video["video_url"], file_path, task['channel'], video["video_id"])
            logger.info(f"[机器改造] {video['video_url']}视频下载成功")
            if not os.path.exists(video_path) or os.path.getsize(video_path) == 0:
                text = (
                    f"**通知类型**: 视频下载失败\n"
                    f"**负责人**: {fs_channel_name}\n"
                    f"**渠道**: {task['channel']}\n"
                    f"**视频主页ID**: {task['channel_url']}\n"
                    f"**视频Video_id**: {video['video_id']}\n"
                )
                AliyunLogger.logging(task['channel'], fs_channel_name, task['channel_url'], video['video_id'], "视频下载失败", "3002", f"video_url:{video['video_url']}")

                Feishu.finish_bot(text,
                                  "https://open.feishu.cn/open-apis/bot/v2/hook/e7697dc6-5254-4411-8b59-3cd0742bf703",
                                  "【 机器改造通知 】")
                continue
            logger.info(f"[机器改造] 视频更改分辨率处理")
            width, height = FFmpeg.get_w_h_size(video_path)
            if width < height:  # 判断是否需要修改为竖屏
                video_path = FFmpeg.update_video_h_w(video_path, file_path)
            if task["crop_tool"]:  # 判断是否需要裁剪
                video_path = FFmpeg.video_crop(video_path, file_path)
            if task["gg_duration"]:  # 判断是否需要指定视频时长
                video_path = FFmpeg.video_ggduration(video_path, file_path, task["gg_duration"])
            video_path = FFmpeg.video_640(video_path, file_path)

            logger.info(f"[机器改造] 视频更改分辨率处理成功")
            video_text = GoogleAI.run(ai_key, video_path)
            if not video_text:
                text = (
                    f"**通知类型**: 获取口播文案失败\n"
                    f"**负责人**: {fs_channel_name}\n"
                    f"**渠道**: {task['channel']}\n"
                    f"**视频主页ID**: {task['channel_url']}\n"
                    f"**视频Video_id**: {video['video_id']}\n"
                )
                AliyunLogger.logging(task['channel'], fs_channel_name, task['channel_url'], video['video_id'],
                                     "获取口播文案失败", "3002", f"video_url:{video['video_url']}")

                Feishu.finish_bot(text,
                                  "https://open.feishu.cn/open-apis/bot/v2/hook/e7697dc6-5254-4411-8b59-3cd0742bf703",
                                  "【 机器改造通知 】")
                continue
            pw_srt_text = GPT4oMini.get_content_understanding_pw(video_text)
            voice = task["voice"]
            if voice:
                if ',' in voice:
                    voices = voice.split(',')
                else:
                    voices = [voice]
                voice = random.choice(voices)
            else:
                voice = "zhifeng_emo"
            pw_url = TTS.get_pw_zm(pw_srt_text, voice)
            if not pw_url:
                text = (
                    f"**通知类型**: 获取片尾引导失败\n"
                    f"**负责人**: {fs_channel_name}\n"
                    f"**渠道**: {task['channel']}\n"
                    f"**视频主页ID**: {task['channel_url']}\n"
                    f"**视频Video_id**: {video['video_id']}\n"
                )
                AliyunLogger.logging(task['channel'], fs_channel_name, task['channel_url'], video['video_id'],
                                     "获取片尾引导失败", "3002", f"video_url:{video['video_url']}")

                Feishu.finish_bot(text,
                                  "https://open.feishu.cn/open-apis/bot/v2/hook/e7697dc6-5254-4411-8b59-3cd0742bf703",
                                  "【 机器改造通知 】")
                continue
            pw_srt = TTS.getSrt(pw_url)
            if not pw_srt:
                text = (
                    f"**通知类型**: 获取片尾引导失败\n"
                    f"**负责人**: {fs_channel_name}\n"
                    f"**渠道**: {task['channel']}\n"
                    f"**视频主页ID**: {task['channel_url']}\n"
                    f"**视频Video_id**: {video['video_id']}\n"
                )
                AliyunLogger.logging(task['channel'], fs_channel_name, task['channel_url'], video['video_id'],
                                     "获取片尾引导失败", "3002", f"video_url:{video['video_url']}")

                Feishu.finish_bot(text,
                                  "https://open.feishu.cn/open-apis/bot/v2/hook/e7697dc6-5254-4411-8b59-3cd0742bf703",
                                  "【 机器改造通知 】")
                continue
            logger.info(f"[机器改造] 开始下载音频")
            pw_mp3_path = TTS.download_mp3(pw_url, file_path)
            if not pw_mp3_path:
                text = (
                    f"**通知类型**: 片尾音频下载失败\n"
                    f"**负责人**: {fs_channel_name}\n"
                    f"**渠道**: {task['channel']}\n"
                    f"**视频主页ID**: {task['channel_url']}\n"
                    f"**视频Video_id**: {video['video_id']}\n"
                )
                AliyunLogger.logging(task['channel'], fs_channel_name, task['channel_url'], video['video_id'],
                                     "片尾音频下载失败", "3002", f"video_url:{video['video_url']}")

                Feishu.finish_bot(text,
                                  "https://open.feishu.cn/open-apis/bot/v2/hook/e7697dc6-5254-4411-8b59-3cd0742bf703",
                                  "【 机器改造通知 】")
                continue
            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:
                text = (
                    f"**通知类型**: 生成片尾视频失败\n"
                    f"**负责人**: {fs_channel_name}\n"
                    f"**渠道**: {task['channel']}\n"
                    f"**视频主页ID**: {task['channel_url']}\n"
                    f"**视频Video_id**: {video['video_id']}\n"
                )
                AliyunLogger.logging(task['channel'], fs_channel_name, task['channel_url'], video['video_id'],
                                     "生成片尾视频失败", "3002", f"video_url:{video['video_url']}")

                Feishu.finish_bot(text,
                                  "https://open.feishu.cn/open-apis/bot/v2/hook/e7697dc6-5254-4411-8b59-3cd0742bf703",
                                  "【 机器改造通知 】")
                continue
            logger.info(f"[机器改造] 合并开始拼接")
            video_path = FFmpeg.h_b_video(video_path, pw_path, file_path)
            zm = Material.get_pzsrt_data("summary", "500Oe0", task["video_share"])

            video_path = FFmpeg.single_video(video_path, file_path, zm)

            if not os.path.exists(video_path) or os.path.getsize(video_path) == 0:
                text = (
                    f"**通知类型**: 合并拼接失败\n"
                    f"**负责人**: {fs_channel_name}\n"
                    f"**渠道**: {task['channel']}\n"
                    f"**视频主页ID**: {task['channel_url']}\n"
                    f"**视频Video_id**: {video['video_id']}\n"
                )
                AliyunLogger.logging(task['channel'], fs_channel_name, task['channel_url'], video['video_id'],
                                     "合并拼接失败", "3002", f"video_url:{video['video_url']}")

                Feishu.finish_bot(text,
                                  "https://open.feishu.cn/open-apis/bot/v2/hook/e7697dc6-5254-4411-8b59-3cd0742bf703",
                                  "【 机器改造通知 】")
                continue
            logger.info(f"[机器改造] 视频-开始发送oss")
            oss_object_key = Oss.stitching_sync_upload_oss(video_path, str(uuid.uuid4()))  # 视频发送OSS
            logger.info(f"[机器改造] 数据发送oss成功")
            oss_object_key = oss_object_key.get("oss_object_key")
            base_tags = [task['tag'], f"一级品类_{task['first_category']}",f"来源_{fs_channel_name}"]
            tags = ','.join(filter(None, base_tags))
            title = cls.generate_title(video, task["ai_title"])
            cls.insert_pq(task, oss_object_key,title, tags, fs_channel_name, video, voice, fs_mark)
    @classmethod
    def run(cls):
        uid = str(uuid.uuid4())
        file_path = os.path.join(CACHE_DIR, uid)
        logger.info(f"[机器改造] 开始获取redis数据")
        fs_data = os.getenv("FS_DATA")
        # fs_data = '快手关键词搜索,ks-gjc,B65Gs3bCHhIzj0t7KGWcwZD0nGf,91wp7k,AIzaSyBiwivvBKfqDsxvqAKBrCZyk-wFMhfthXg'
        try:
            fs_data_list = fs_data.split(',')
            fs_channel_name = fs_data_list[0]
            fs_mark = fs_data_list[1]
            fs_id = fs_data_list[2]
            fs_sheet = fs_data_list[3]
            ai_key = fs_data_list[4]
            data = RedisHelper().get_client().rpop(name=f"task:{fs_mark}")
            if not data:
                logger.info('[机器改造] 无待执行的扫描任务，重新获取飞书表格读取任务')
                fs_data = Material.get_carry_data(fs_id, fs_sheet)
                RedisHelper().get_client().rpush(f"task:{fs_mark}", *fs_data)
                return
            task = orjson.loads(data)
            totality_count = task['totality_count']
            if totality_count:
                task_totality_count = sqlCollect.get_mark_count(task["task_mark"])
                if int(totality_count) <= int(task_totality_count[0][0]):
                    AliyunLogger.logging((task["channel"]), fs_channel_name, task["channel_url"], '',
                                         f"{task['task_mark']}标识任务每日指定条数已足够，指定条数{totality_count},实际生成条数{int(task_totality_count[0][0])}",
                                         "1111")
                    logger.info(
                        f"[+] {task['task_mark']}标识任务每日指定条数已足够，指定条数{totality_count},实际生成条数{int(task_totality_count[0][0])}")
                    return

            if fs_channel_name in ["快手关键词搜索","抖音关键词搜索"]:
                gjc_count = RedisHelper().get_client().get(f"{fs_mark}_count")
                RedisHelper().get_client().incrby(f"{fs_mark}_count", 1)
                if int(gjc_count.decode()) >= 300:
                    logger.info(f"[机器改造] {fs_channel_name}上限")
                    return

            cls.data_handle(task, file_path, fs_channel_name, ai_key, fs_mark)
            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=1))  # 每1分钟启动一次
        scheduler.start()
    except KeyboardInterrupt:
        pass
    except Exception as e:
        logger.error(f"[机器改造] 启动异常，异常信息:{e}")
        pass
    finally:
        scheduler.shutdown()


if __name__ == '__main__':
    run()