video_editor_agc/video_agc/agc_video.py

import configparser
import os
import random
import subprocess
import sys
import time
import shutil
from datetime import datetime
from common.sql_help import sqlHelp
sys.path.append(os.getcwd())
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, mark):
        video_path_url = config['PATHS']['VIDEO_PATH'] + mark + "/"
        # 获取文件夹中所有扩展名为 '.mp4' 的文件路径列表
        if os.path.exists(video_path_url):
            # 列出目录中的所有文件和文件夹
            for filename in os.listdir(video_path_url):
                file_path = os.path.join(video_path_url, filename)
                try:
                    # 如果是文件，则删除
                    if os.path.isfile(file_path) or os.path.islink(file_path):
                        os.unlink(file_path)
                    # 如果是文件夹，则删除文件夹及其内容
                    elif os.path.isdir(file_path):
                        shutil.rmtree(file_path)
                except Exception as e:
                    print(f"Failed to delete {file_path}. Reason: {e}")
            print(f"文件已清空。")
        else:
            print(f"文件已清空。")
    """
    站外视频拼接
    """
    @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))
        count = 0
        while True:
            # 清空所有文件
            cls.clear_mp4_files(mark)
            s_path, v_path, video_path_url, v_oss_path = cls.create_folders(mark)

            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 and len(srt.strip()) >= 3:
                #     # 创建临时字幕文件
                #     cls.create_subtitle_file(srt, s_path)
                #     Common.logger("bk_video").info(f"S{mark} 文件目录创建成功")
                # else:
                srt_new = SRT.getSrt(int(uid))
                Common.logger("video").info(f"S{mark}的{platform}渠道音频ID")

                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}渠道SRT 文件目录创建成功")
                # 获取音频
                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 = str(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}渠道视频添加到对应用户成功")
                    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")
                # 清空所有mp4数据
                cls.clear_mp4_files(mark)
                continue
        if "-" in mark:
            name = mark.split("-")[0]
        else:
            name = mark
        Feishu.bot('recommend', 'AGC完成通知', f'今日{platform}任务拼接任务完成,共{count}条', name, mark_name)