video_editor_agc/video_agc/agc_video.py

import configparser
import glob
import os
import random
import re
import subprocess
import sys
import time
import urllib.parse
import json

import requests
from datetime import datetime, timedelta
from urllib.parse import urlencode

from common.sql_help import sqlHelp

sys.path.append(os.getcwd())
from common.db import MysqlHelper
from common.material import Material
from common import Common, Oss, Feishu, PQ
from common.srt import SRT

config = configparser.ConfigParser()
config.read('./config.ini')  # 替换为您的配置文件路径


class AGC():
    """清除文件下所有mp4文件"""
    @classmethod
    def clear_mp4_files(cls, folder_path):
        # 获取文件夹中所有扩展名为 '.mp4' 的文件路径列表
        mp4_files = glob.glob(os.path.join(folder_path, '*.mp4'))
        if not mp4_files:
            return
        # 遍历并删除所有 .mp4 文件
        for mp4_file in mp4_files:
            os.remove(mp4_file)
        print(f"文件夹 '{folder_path}' 中的所有 .mp4 文件已清空。")
    """
    站外视频拼接
    """
    @classmethod
    def zw_concatenate_videos(cls, videos, audio_duration, audio_video, platform, s_path, v_path, mark, v_oss_path):

        video_files = cls.concat_videos_with_subtitles(videos, audio_duration, platform, mark)
        Common.logger("video").info(f"{mark}的{platform}视频文件:{video_files}")

        if video_files == "":
            return ""
        print(f"{mark}的{platform}:开始拼接视频喽～～～")
        Common.logger("video").info(f"{mark}的{platform}:开始拼接视频喽～～～")
        if os.path.exists(s_path):
            # subtitle_cmd = f"subtitles={s_path}:force_style='Fontsize=11,Fontname=Hiragino Sans GB,Outline=0,PrimaryColour=&H000000,SecondaryColour=&H000000'"
            subtitle_cmd = f"subtitles={s_path}:force_style='Fontsize=12,Fontname=wqy-zenhei,Bold=1,Outline=0,PrimaryColour=&H000000,SecondaryColour=&H000000'"
        else:
            start_time = cls.seconds_to_srt_time(0)
            end_time = cls.seconds_to_srt_time(audio_duration)
            with open(s_path, 'w') as f:
                f.write(f"1\n{start_time} --> {end_time}\n分享、转发给群友\n")
            # subtitle_cmd = "drawtext=text='分享、转发给群友':fontsize=28:fontcolor=black:x=(w-text_w)/2:y=h-text_h-15"
            subtitle_cmd = f"subtitles={s_path}:force_style='Fontsize=12,Fontname=wqy-zenhei,Bold=1,Outline=0,PrimaryColour=&H000000,SecondaryColour=&H000000'"
        # 背景色参数
        background_cmd = "drawbox=y=ih-65:color=yellow@1.0:width=iw:height=0:t=fill"

        VIDEO_COUNTER = 0
        FF_INPUT = ""
        FF_SCALE = ""
        FF_FILTER = ""
        ffmpeg_cmd = ["ffmpeg"]
        for videos in video_files:
            Common.logger("video").info(f"{mark}的{platform}视频:{videos[3]}")
            # 添加输入文件
            FF_INPUT += f" -i {videos[3]}"
            # 为每个视频文件统一长宽，并设置SAR(采样宽高比)
            FF_SCALE += f"[{VIDEO_COUNTER}:v]scale=320x480,setsar=1[v{VIDEO_COUNTER}];"
            # 为每个视频文件创建一个输入流，并添加到-filter_complex参数中
            FF_FILTER += f"[v{VIDEO_COUNTER}][{VIDEO_COUNTER}:a]"
            # 增加视频计数器
            VIDEO_COUNTER += 1
        # 构建最终的FFmpeg命令
        ffmpeg_cmd.extend(FF_INPUT.split())
        ffmpeg_cmd.extend(["-filter_complex", f"{FF_SCALE}{FF_FILTER}concat=n={VIDEO_COUNTER}:v=1:a=1[v][a]",
                           "-map", "[v]", "-map", "[a]", v_path])
        # 多线程数
        num_threads = 4
        # 构建 FFmpeg 命令,生成视频
        ffmpeg_cmd_oss = [
            "ffmpeg",
            "-i", v_path,  # 视频文件列表
            "-i", audio_video,  # 音频文件
            "-c:v", "libx264",  # 复制视频流
            "-c:a", "aac",  # 编码音频流为AAC
            "-threads", str(num_threads),
            "-vf", f"{background_cmd},{subtitle_cmd}",  # 添加背景色和字幕
            "-t", str(int(audio_duration)),  # 保持与音频时长一致
            "-map", "0:v:0",  # 映射第一个输入的视频流
            "-map", "1:a:0",  # 映射第二个输入的音频流
            "-y",  # 覆盖输出文件
            v_oss_path
        ]
        try:
            timeout_seconds = 25 * 60
            subprocess.run(ffmpeg_cmd,  timeout=timeout_seconds)
            if os.path.isfile(v_path):
                subprocess.run(ffmpeg_cmd_oss,  timeout=timeout_seconds)
            print("视频处理完成！")
        except subprocess.TimeoutExpired:
            print("视频处理超时，处理失败！")
            return ""
        except subprocess.CalledProcessError as e:
            print(f"视频处理失败：{e}")
            return ""
        print(f"{mark}的{platform}:视频拼接成功啦～～～")
        Common.logger("video").info(f"{mark}的{platform}:视频拼接成功啦～～～")
        return video_files

    """视频秒数转换"""
    @classmethod
    def seconds_to_srt_time(cls, seconds):
        hours = int(seconds // 3600)
        minutes = int((seconds % 3600) // 60)
        seconds = seconds % 60
        milliseconds = int((seconds - int(seconds)) * 1000)
        return f"{hours:02d}:{minutes:02d}:{int(seconds):02d},{milliseconds:03d}"

    """
    获取视频文件的时长（秒）
    """
    @classmethod
    def get_video_duration(cls, video_file):
        result = subprocess.run(
            ["ffprobe", "-v", "error", "-show_entries", "format=duration",
             "-of", "default=noprint_wrappers=1:nokey=1", video_file],
            capture_output=True, text=True)
        return float(result.stdout)

    """计算需要拼接的视频"""
    @classmethod
    def concat_videos_with_subtitles(cls, videos, audio_duration, platform, mark):
        # 计算视频文件列表总时长
        if platform == "爆款" or platform == "跟随":
            total_video_duration = sum(cls.get_video_duration(video_file) for video_file in videos)
        else:
            total_video_duration = sum(cls.get_video_duration(video_file[3]) for video_file in videos)
        if platform == "爆款" or platform == "跟随":
            # 视频时长大于音频时长
            if total_video_duration > audio_duration:
                return videos
            # 计算音频秒数与视频秒数的比率，然后加一得到需要的视频数量
            video_audio_ratio = audio_duration / total_video_duration
            videos_needed = int(video_audio_ratio) + 2
            trimmed_video_list = videos * videos_needed
            return trimmed_video_list
        else:
            # 如果视频总时长小于音频时长，则不做拼接
            if total_video_duration < audio_duration:
                Common.logger("video").info(f"{mark}的{platform}渠道时长小于等于目标时长，不做视频拼接")
                return ""
            # 如果视频总时长大于音频时长，则截断视频
            trimmed_video_list = []
            remaining_duration = audio_duration
            for video_file in videos:
                video_duration = cls.get_video_duration(video_file[3])
                if video_duration <= remaining_duration:
                    # 如果视频时长小于或等于剩余时长，则将整个视频添加到列表中
                    trimmed_video_list.append(video_file)
                    remaining_duration -= video_duration
                else:
                    trimmed_video_list.append(video_file)
                    break
            return trimmed_video_list

    """
    text文件没有则创建目录
    """
    @classmethod
    def bk_text_folders(cls, mark):
        oss_id = cls.random_id()
        v_text_url = config['PATHS']['VIDEO_PATH'] + mark + "/text/"
        if not os.path.exists(v_text_url):
            os.makedirs(v_text_url)
        # srt 文件地址
        text_path = v_text_url + mark + f"{str(oss_id)}.text"
        return text_path

    """
    站内视频拼接
    """
    @classmethod
    def zn_concatenate_videos(cls, videos, audio_duration, audio_video, platform, s_path, mark, v_oss_path):
        text_ptah = cls.bk_text_folders(mark)
        video_files = cls.concat_videos_with_subtitles(videos, audio_duration, platform, mark)
        with open(text_ptah, 'w') as f:
            for file in video_files:
                f.write(f"file '{file}'\n")
        Common.logger("video").info(f"{mark}的{platform}视频文件:{video_files}")
        if video_files == "":
            return ""
        print(f"{mark}的{platform}:开始拼接视频喽～～～")
        Common.logger("video").info(f"{mark}的{platform}:开始拼接视频喽～～～")
        if os.path.exists(s_path):
            # subtitle_cmd = f"subtitles={s_path}:force_style='Fontsize=11,Fontname=Hiragino Sans GB,Outline=0,PrimaryColour=&H000000,SecondaryColour=&H000000'"
            subtitle_cmd = f"subtitles={s_path}:force_style='Fontsize=12,Fontname=wqy-zenhei,Bold=1,Outline=0,PrimaryColour=&H000000,SecondaryColour=&H000000'"
        else:
            start_time = cls.seconds_to_srt_time(0)
            end_time = cls.seconds_to_srt_time(audio_duration)
            with open(s_path, 'w') as f:
                f.write(f"1\n{start_time} --> {end_time}\n分享、转发给群友\n")
            # subtitle_cmd = "drawtext=text='分享、转发给群友':fontsize=28:fontcolor=black:x=(w-text_w)/2:y=h-text_h-15"
            subtitle_cmd = f"subtitles={s_path}:force_style='Fontsize=12,Fontname=wqy-zenhei,Bold=1,Outline=0,PrimaryColour=&H000000,SecondaryColour=&H000000'"
        # 背景色参数
        background_cmd = "drawbox=y=ih-65:color=yellow@1.0:width=iw:height=0:t=fill"
        # 多线程数
        num_threads = 4
        # 构建 FFmpeg 命令,生成视频
        ffmpeg_cmd_oss = [
            "ffmpeg",
            "-f", "concat",
            "-safe", "0",
            "-i", f"{text_ptah}",  # 视频文件列表
            "-i", audio_video,  # 音频文件
            "-c:v", "libx264",
            "-c:a", "aac",
            "-threads", str(num_threads),
            "-vf", f"scale=320x480,{background_cmd},{subtitle_cmd}",  # 添加背景色和字幕
            "-t", str(int(audio_duration)),  # 保持与音频时长一致
            "-map", "0:v:0",  # 映射第一个输入的视频流
            "-map", "1:a:0",  # 映射第二个输入的音频流
            "-y",  # 覆盖输出文件
            v_oss_path
        ]
        try:
            subprocess.run(ffmpeg_cmd_oss)
            print("视频处理完成！")
            if os.path.isfile(text_ptah):
                os.remove(text_ptah)
        except subprocess.CalledProcessError as e:
            print(f"视频处理失败：{e}")
        print(f"{mark}:视频拼接成功啦～～～")
        Common.logger("video").info(f"{mark}:视频拼接成功啦～～～")
        return v_oss_path

    """
    获取视频时长
    """
    @classmethod
    def get_audio_duration(cls, video_url):
        ffprobe_cmd = [
            "ffprobe",
            "-i", video_url,
            "-show_entries", "format=duration",
            "-v", "quiet",
            "-of", "csv=p=0"
        ]
        output = subprocess.check_output(ffprobe_cmd).decode("utf-8").strip()
        return float(output)
    """
    创建临时字幕
    """
    @classmethod
    def create_subtitle_file(cls, srt, s_path):
        with open(s_path, 'w') as f:
            f.write(srt)
    """
    随机生成id
    """
    @classmethod
    def random_id(cls):
        now = datetime.now()
        rand_num = random.randint(10000, 99999)
        oss_id = "{}{}".format(now.strftime("%Y%m%d%H%M%S"), rand_num)
        return oss_id
    """
    文件没有则创建目录
    """
    @classmethod
    def create_folders(cls, mark):
        oss_id = cls.random_id()

        video_path_url = config['PATHS']['VIDEO_PATH'] + mark + "/"
        # srt 目录
        s_path_url = config['PATHS']['VIDEO_PATH'] + mark + "/srt/"
        # oss 目录
        v_path_url = config['PATHS']['VIDEO_PATH'] + mark + "/oss/"

        if not os.path.exists(video_path_url):
            os.makedirs(video_path_url)
        if not os.path.exists(s_path_url):
            os.makedirs(s_path_url)
        if not os.path.exists(v_path_url):
            os.makedirs(v_path_url)
        # srt 文件地址
        s_path = s_path_url + mark + f"{str(oss_id)}.srt"
        # 最终生成视频地址
        v_path = v_path_url + mark + f"{str(oss_id)}.mp4"
        v_oss_path = v_path_url + mark + f"{str(oss_id)}oss.mp4"
        return s_path, v_path, video_path_url, v_oss_path

    """
    获取未使用的数据
    """
    @classmethod
    def get_unique_uid_data(cls, data, count):
        unique_data_dict = {item['uid']: item for item in data}
        unique_data = list(unique_data_dict.values())

        if count >= len(unique_data):
            return unique_data
        else:
            selected_items = []
            selected_uids = set()

            while len(selected_items) < count:
                # 随机选择一个元素
                item = random.choice(unique_data)
                uid = item['uid']

                if uid not in selected_uids:
                    # 如果该uid还未被选择过，则将该元素添加到选中项列表中，并记录已选择的uid
                    selected_items.append(item)
                    selected_uids.add(uid)
            return selected_items

    """
    任务处理
    """
    @classmethod
    def video(cls, data, platform):
        mark_name = data['mark_name']  # 负责人
        if platform == "爆款":
            pq_ids = data["pq_id"]
            pq_ids_list = pq_ids.split(',')  # 账号ID
            mark = data["mark"]  # 标示
            feishu_id = data["feishu_id"]  # 飞书文档ID
            video_call = data["video_call"]  # 脚本sheet
            list_data = Material.get_allbk_data(feishu_id, video_call)
            if len(list_data) == 0:
                Feishu.bot('recommend', 'AGC完成通知', f'爆款任务数为0,不做拼接', mark, mark_name)
                return mark
        elif platform == "常规":
            pq_ids = data["pq_id"]# 账号ID
            pq_ids_list = pq_ids.split(',')
            mark = data["mark"]
            feishu_id = data["feishu_id"]  # 飞书文档ID
            video_call = data["video_call"]  # 脚本sheet
            video_count = data["video_count"]
            if int(video_count) == 0:
                Feishu.bot('recommend', 'AGC完成通知', f'常规任务数为{video_count},不做拼接', mark, mark_name)
                return mark
            data_list, videos_mark = Material.get_all_data(feishu_id, video_call, mark)
            list_data = cls.get_unique_uid_data(data_list, int(video_count))
        elif platform == "跟随":
            pq_ids = data["pq_id"]
            pq_ids_list = pq_ids.split(',')  # 账号ID
            mark = data["mark"]  # 标示
            feishu_id = data["feishu_id"]  # 飞书文档ID
            video_call = data["video_call"]
            video_count = data["video_count"]
            if int(video_count) == 0:
                Feishu.bot('recommend', 'AGC完成通知', f'跟随任务数为{video_count},不做拼接', mark, mark_name)
                return mark
            data_list, videos = Material.get_all_data(feishu_id, video_call, mark)
            list_data = cls.get_unique_uid_data(data_list, int(video_count))
        s_path, v_path, video_path_url, v_oss_path = cls.create_folders(mark)
        count = 0
        while True:
            if count == len(list_data):
                break
        # for d_list in list_data:
            try:
                d_list = list_data[count]
                uid = d_list['uid']  # 音频id
                srt = d_list['text']  # srt
                cover = d_list['cover']
                audio_title = d_list['title']
                if srt:
                    # 创建临时字幕文件
                    cls.create_subtitle_file(srt, s_path)
                    Common.logger("bk_video").info(f"S{mark} 文件目录创建成功")
                else:
                    srt_new = SRT.getSrt(int(uid))
                    if srt_new:
                        current_time = datetime.now()
                        formatted_time = current_time.strftime("%Y-%m-%d %H:%M:%S")
                        values = [[mark, str(uid), srt_new, formatted_time]]
                        random_wait_time = random.uniform(0.5, 2.5)
                        time.sleep(random_wait_time)
                        Feishu.insert_columns("IbVVsKCpbhxhSJtwYOUc8S1jnWb", "jd9qD9", "ROWS", 1, 2)
                        time.sleep(random_wait_time)
                        Feishu.update_values("IbVVsKCpbhxhSJtwYOUc8S1jnWb", "jd9qD9", "A2:Z2", values)
                        # 创建临时字幕文件
                        cls.create_subtitle_file(srt_new, s_path)
                        Common.logger("video").info(f"S{mark}的{platform}渠道RT 文件目录创建成功")
                # 获取音频
                audio_video = PQ.get_audio_url(uid)
                Common.logger("video").info(f"{mark}的{platform}渠道获音频成功")
                audio_duration = cls.get_audio_duration(audio_video)
                Common.logger("video").info(f"{mark}的{platform}渠道获取需要拼接的音频秒数为:{audio_duration}")
                if platform != "常规":
                    if platform == "爆款":
                        videos = d_list['video']
                        if ',' in videos:
                            videos = str(videos).split(',')
                        else:
                            videos = [str(videos)]
                    video_id = random.choice(videos)

                    video_url = PQ.get_audio_url(video_id)
                    download_video = Oss.download_url(video_url, video_path_url, str(video_id))
                    if download_video:
                        video_files = cls.zn_concatenate_videos(download_video, audio_duration, audio_video, platform,
                                                                s_path, mark, v_oss_path)

                        if os.path.isfile(v_oss_path):
                            Common.logger("video").info(f"{mark}的{platform}渠道新视频生成成功")
                        else:
                            Common.logger("video").info(f"{mark}的{platform}渠道新视频生成失败")
                            continue
                else:
                    # chnnel_count = int(len(list_data)/2)
                    channels = ["douyin", "kuaishou"]
                    channel = random.choice(channels)
                    user_id = sqlHelp.get_user_id(channel, mark)
                    url_list, user = sqlHelp.get_url_list(user_id, mark, "35")
                    videos = [list(item) for item in url_list]
                    videos = Oss.get_oss_url(videos, video_path_url)
                    video_files = cls.zw_concatenate_videos(videos, audio_duration, audio_video, platform, s_path, v_path,
                                                         mark, v_oss_path)
                    if video_files == "":
                        Common.logger("video").info(f"{mark}的{platform}渠道使用拼接视频为空")
                        continue
                    if os.path.isfile(v_oss_path):
                        Common.logger("video").info(f"{mark}的{platform}渠道新视频生成成功")
                    else:
                        Common.logger("video").info(f"{mark}的{platform}渠道新视频生成失败")
                        continue
                # 随机生成视频oss_id
                oss_id = cls.random_id()
                Common.logger("video").info(f"{mark}的{platform}渠道上传到 OSS 生成视频id为：{oss_id}")
                oss_object_key = Oss.stitching_sync_upload_oss(v_oss_path, oss_id)
                status = oss_object_key.get("status")
                if status == 200:
                    # 获取 oss 视频地址
                    oss_object_key = oss_object_key.get("oss_object_key")
                    Common.logger("video").info(f"{mark}的{platform}渠道拼接视频发送成功，OSS 地址：{oss_object_key}")
                    time.sleep(10)
                    if platform == "常规":
                        # 已使用视频存入数据库
                        Common.logger("video").info(f"{mark}的{platform}渠道开始已使用视频存入数据库")
                        sqlHelp.insert_videoAudio(video_files, uid, platform, mark)
                        Common.logger("video").info(f"{mark}的{platform}渠道完成已使用视频存入数据库")
                    Common.logger("video").info(f"{mark}的{platform}渠道开始视频添加到对应用户")
                    new_video_id, title = PQ.insert_piaoquantv(oss_object_key, audio_title, pq_ids_list, cover, uid)
                    if new_video_id:
                        Common.logger("video").info(f"{mark}的{platform}渠道视频添加到对应用户成功")
                    if os.path.isfile(v_oss_path):
                        os.remove(v_oss_path)
                    if os.path.isfile(v_path):
                        os.remove(v_path)
                    if os.path.isfile(s_path):
                        os.remove(s_path)
                    # 清空所有mp4数据
                    cls.clear_mp4_files(video_path_url)
                    count += 1
                    if mark_name == "穆新艺":
                        sheet = '50b8a1'
                    elif mark_name == "信欣":
                        sheet = 'UyVK7y'
                    elif mark_name == "范军":
                        sheet = 'uP3zbf'
                    elif mark_name == "鲁涛":
                        sheet = 'iDTHt4'
                    elif mark_name == "余海涛":
                        sheet = 'R1jIeT'
                    elif mark_name == "罗情":
                        sheet = 'iuxfAt'
                    current_time = datetime.now()
                    formatted_time = current_time.strftime("%Y-%m-%d %H:%M:%S")
                    if platform == "常规":
                        third_chars = [j[2] for j in video_files]
                        data = ",".join(third_chars)
                        user_id = user
                    else:
                        user_id = video_id
                        data = ''
                    values = [[mark, str(uid), str(user_id), data, title, new_video_id, formatted_time]]
                    Feishu.insert_columns("LAn9so7E0hxRYht2UMEcK5wpnMj", sheet, "ROWS", 1, 2)
                    random_wait_time = random.uniform(0.5, 2.5)
                    time.sleep(random_wait_time)
                    Feishu.update_values("LAn9so7E0hxRYht2UMEcK5wpnMj", sheet, "A2:Z2", values)
            except Exception as e:
                Common.logger("bk_video").warning(f"{mark}的视频拼接失败:{e}\n")
                if os.path.isfile(v_oss_path):
                    os.remove(v_oss_path)
                if os.path.isfile(v_path):
                    os.remove(v_path)
                if os.path.isfile(s_path):
                    os.remove(s_path)
                # 清空所有mp4数据
                cls.clear_mp4_files(video_path_url)
                continue
        if "-" in mark:
            name = mark.split("-")[0]
        else:
            name = mark
        Feishu.bot('recommend', 'AGC完成通知', f'今日{platform}任务拼接任务完成,共{count}条', name, mark_name)