update

utils/aliyun_oss.py

@@ -0,0 +1,134 @@
+# -*- coding: utf-8 -*-
+# @Time: 2023/12/26
+import time
+import uuid
+from datetime import datetime
+from typing import Dict, Any,  Optional
+import oss2
+import requests
+OSS_ACCESS_KEY_ID = "LTAIP6x1l3DXfSxm"
+OSS_ACCESS_KEY_SECRET = "KbTaM9ars4OX3PMS6Xm7rtxGr1FLon"
+# OSS_BUCKET_ENDPOINT = "oss-cn-hangzhou-internal.aliyuncs.com"# 内网地址
+OSS_BUCKET_ENDPOINT = "oss-cn-hangzhou.aliyuncs.com" # 外网地址
+OSS_BUCKET_NAME = "art-crawler"
+class Oss():
+
+    @classmethod
+    def channel_upload_oss(cls, src_url: str,
+                              video_id: str,
+                              referer: Optional[str] = None) -> Dict[str, Any]:
+        headers = {
+            'Accept': '*/*',
+            'Accept-Language': 'zh-CN,zh;q=0.9',
+            'Cache-Control': 'no-cache',
+            'Pragma': 'no-cache',
+            'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) '
+                          'Chrome/117.0.0.0 Safari/537.36',
+        }
+        if referer:
+            headers.update({'Referer': referer})
+        response = requests.request(url=src_url, method='GET', headers=headers, timeout=30)
+        file_content = response.content
+        content_type = response.headers.get('Content-Type', 'application/octet-stream')
+
+        oss_object_key = f'carry/video/{video_id}'
+        auth = oss2.Auth(OSS_ACCESS_KEY_ID, OSS_ACCESS_KEY_SECRET)
+        bucket = oss2.Bucket(auth, OSS_BUCKET_ENDPOINT, OSS_BUCKET_NAME)
+        response = bucket.put_object(oss_object_key, file_content, headers={'Content-Type': content_type})
+
+        if 'Content-Length' in response.headers:
+            return {
+                'status': response.status,
+                'oss_object_key': oss_object_key}
+        raise AssertionError(f'OSS上传失败，请求ID: \n{response.headers["x-oss-request-id"]}')
+
+    """
+    视频发送到art-pubbucket
+    """
+    @classmethod
+    def stitching_sync_upload_oss(cls, src_url: str,
+                        video_id: str) -> Dict[str, Any]:
+        oss_object_key = f'carry/video/{video_id}'
+        auth = oss2.Auth(OSS_ACCESS_KEY_ID, OSS_ACCESS_KEY_SECRET)
+        bucket = oss2.Bucket(auth, OSS_BUCKET_ENDPOINT, "art-pubbucket")
+        response = bucket.put_object_from_file(oss_object_key, src_url)
+
+        if 'Content-Length' in response.headers:
+            return {
+                'status': response.status,
+                'oss_object_key': oss_object_key,
+                'save_oss_timestamp': int(datetime.now().timestamp() * 1000),
+            }
+        raise AssertionError(f'OSS上传失败，请求ID: \n{response.headers["x-oss-request-id"]}')
+
+    """
+    封面发送到art-pubbucket
+    """
+    @classmethod
+    def stitching_fm_upload_oss(cls, src_url: str,
+                                  video_id: str) -> Dict[str, Any]:
+        oss_object_key = f'jq_oss/jpg/{video_id}.jpg'
+        auth = oss2.Auth(OSS_ACCESS_KEY_ID, OSS_ACCESS_KEY_SECRET)
+        bucket = oss2.Bucket(auth, OSS_BUCKET_ENDPOINT, "art-pubbucket")
+        response = bucket.put_object_from_file(oss_object_key, src_url)
+
+        if 'Content-Length' in response.headers:
+            return {
+                'status': response.status,
+                'oss_object_key': oss_object_key,
+                'save_oss_timestamp': int(datetime.now().timestamp() * 1000),
+            }
+        raise AssertionError(f'OSS上传失败，请求ID: \n{response.headers["x-oss-request-id"]}')
+
+    """
+    封面发送到art-pubbucket
+    """
+    @classmethod
+    def mp3_upload_oss(cls, src_url: str,
+                                video_id: str) -> Dict[str, Any]:
+        oss_object_key = f'jq_audio/audio/{video_id}.mp3'
+        auth = oss2.Auth(OSS_ACCESS_KEY_ID, OSS_ACCESS_KEY_SECRET)
+        bucket = oss2.Bucket(auth, OSS_BUCKET_ENDPOINT, "art-crawler")
+        response = bucket.put_object_from_file(oss_object_key, src_url)
+
+        if 'Content-Length' in response.headers:
+            return {
+                'status': response.status,
+                'oss_object_key': oss_object_key,
+                'save_oss_timestamp': int(datetime.now().timestamp() * 1000),
+            }
+        raise AssertionError(f'OSS上传失败，请求ID: \n{response.headers["x-oss-request-id"]}')
+
+
+    @classmethod
+    def download_video_oss(cls, url, file_path):
+        video_path = file_path + 'video.mp4'
+        oss_object_key = cls.channel_upload_oss(url, str(uuid.uuid4()))
+        time.sleep(2)
+        oss_object = oss_object_key.get("oss_object_key")
+        if oss_object:
+            auth = oss2.Auth(OSS_ACCESS_KEY_ID, OSS_ACCESS_KEY_SECRET)
+            bucket = oss2.Bucket(auth, OSS_BUCKET_ENDPOINT, OSS_BUCKET_NAME)
+            # 获取指定路径下的对象列表
+            bucket.get_object_to_file(oss_object, video_path)
+            time.sleep(5)
+            return video_path
+        else:
+            return video_path
+
+    @classmethod
+    def download_sph_ls(cls, video_url, video_path_url, v_id):
+        if "jpg" in video_url:
+            video_path = video_path_url + str(v_id) + '.jpg'
+        else:
+            video_path = video_path_url + str(v_id) + '.mp4'
+        auth = oss2.Auth(OSS_ACCESS_KEY_ID, OSS_ACCESS_KEY_SECRET)
+        bucket = oss2.Bucket(auth, OSS_BUCKET_ENDPOINT, OSS_BUCKET_NAME)
+        # 获取指定路径下的对象列表
+        bucket.get_object_to_file(video_url, video_path)
+        time.sleep(5)
+        return video_path
+
+
+if __name__ == '__main__':
+    Oss.download_sph_ls('channel/video/sph/14374775553517295881.jpg','asa','1')

utils/download_video.py

@@ -0,0 +1,69 @@
+import os
+import time
+import uuid
+import requests
+
+
+class DownLoad:
+    @classmethod
+    def download_video(cls, video_url, file_path):
+        video = file_path + 'video.mp4'
+        try:
+            for i in range(3):
+                payload = {}
+                headers = {}
+                response = requests.request("GET", video_url, headers=headers, data=payload, timeout=240)
+                if response.status_code == 200:
+                    # 以二进制写入模式打开文件
+
+                    with open(f"{video}", "wb") as file:
+                        # 将响应内容写入文件
+                        file.write(response.content)
+                    return video
+            return video
+        except Exception:
+            return video
+
+    @classmethod
+    def download_m3u8_video(cls ,url, file_path):
+        r = requests.get(url)
+        if r.status_code != 200:
+            return False
+        m3u8_list = r.text.split('\n')
+        m3u8_list = [i for i in m3u8_list if i and i[0] != '#']
+
+        ts_list = []
+        for ts_url in m3u8_list:
+            ts_url = url.rsplit('/', 1)[0] + '/' + ts_url
+            ts_list.append(ts_url)
+        with open(file_path, 'wb') as f:
+            for ts_url in ts_list:
+                r = requests.get(ts_url)
+                if r.status_code == 200:
+                    f.write(r.content)
+        return True
+
+    @classmethod
+    def convert_ts_to_mp4(cls, ts_file_path, mp4_file_path):
+        os.system(f'ffmpeg -i {ts_file_path} -c copy {mp4_file_path}')
+
+    @classmethod
+    def download_pq_video(cls,video_path_url , video_url_list):
+        video_list = []
+        for video_url in video_url_list:
+            video = f'{video_path_url}{str(uuid.uuid4())}.mp4'
+            try:
+                payload = {}
+                headers = {}
+                response = requests.request("GET", video_url, headers=headers, data=payload, timeout=60)
+                if response.status_code == 200:
+                    # 以二进制写入模式打开文件
+
+                    with open(f"{video}", "wb") as file:
+                        # 将响应内容写入文件
+                        file.write(response.content)
+                    video_list.append(video)
+                time.sleep(1)
+            except Exception:
+                continue
+        return video_list

utils/ffmpeg.py

@@ -0,0 +1,550 @@
+import asyncio
+import json
+import os
+import time
+from typing import List
+
+import cv2
+import requests
+from loguru import logger
+from mutagen.mp3 import MP3
+
+
+
+class FFmpeg():
+
+    """
+    时间转换
+    """
+    @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_url):
+        cap = cv2.VideoCapture(video_url)
+        if cap.isOpened():
+            rate = cap.get(5)
+            frame_num = cap.get(7)
+            duration = int(frame_num / rate)
+            return duration
+        return 0
+
+
+    # """
+    # 获取视频文件的时长（秒）
+    # """
+    # @classmethod
+    # def get_videos_duration(cls, video_file):
+    #     result = cls.asyncio_run_subprocess(["ffprobe", "-v", "error", "-show_entries", "format=duration",
+    #          "-of", "default=noprint_wrappers=1:nokey=1", video_file], timeout=10)
+    #     return float(result)
+
+    """
+    获取视频宽高
+    """
+    @classmethod
+    def get_w_h_size(cls, new_video_path):
+        try:
+            # 获取视频的原始宽高信息
+            ffprobe_cmd = cls.asyncio_run_subprocess(["ffprobe", "-v" ,"error" ,"-select_streams" ,"v:0" ,"-show_entries", "stream=width,height" ,"-of" ,"csv=p=0" ,new_video_path],timeout=10)
+            output_decoded = ffprobe_cmd.strip()
+            split_output = [value for value in output_decoded.split(',') if value.strip()]
+            height, width = map(int, split_output)
+            return width, height
+        except ValueError as e:
+            return 1920, 1080
+
+
+    """
+    视频裁剪
+    """
+    @classmethod
+    def video_crop(cls, video_path, file_path):
+        crop_url = file_path + 'crop.mp4'
+        try:
+            # 获取视频的原始宽高信息
+            ffprobe_cmd = cls.asyncio_run_subprocess(
+                ["ffprobe", "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=width,height", "-of",
+                 "csv=p=0", video_path], timeout=10)
+            width, height = map(int, ffprobe_cmd.strip().split(','))
+            # 计算裁剪后的高度
+            new_height = int(height * 0.8)
+
+            # 构建 FFmpeg 命令,裁剪视频高度为原始高度的80%
+            cls.asyncio_run_subprocess(
+                [
+                "ffmpeg",
+                "-i", video_path,
+                "-vf", f"crop={width}:{new_height}",
+                "-c:v", "libx264",
+                "-c:a", "aac",
+                "-y",
+                crop_url
+                ],timeout=240)
+            return crop_url
+        except Exception as e:
+            return crop_url
+
+    """
+    视频截断
+    """
+    @classmethod
+    def video_ggduration(cls, video_path, file_path, gg_duration_total):
+        gg_duration_url = file_path + 'gg_duration.mp4'
+        # 获取视频时长
+        try:
+            total_duration = cls.get_video_duration(video_path)
+            if total_duration == 0:
+                return gg_duration_url
+            duration = int(total_duration) - int(gg_duration_total)
+            if int(total_duration) < int(gg_duration_total):
+                return gg_duration_url
+            cls.asyncio_run_subprocess([
+                "ffmpeg",
+                "-i", video_path,
+                "-c:v", "libx264",
+                "-c:a", "aac",
+                "-t", str(duration),
+                "-y",
+                gg_duration_url
+            ], timeout= 360)
+            return gg_duration_url
+        except Exception as e:
+            return gg_duration_url
+
+    """
+     截取原视频最后一帧
+    """
+    @classmethod
+    def video_png(cls, video_path, file_path):
+        # 获取视频的原始宽高信息
+        jpg_url = file_path + 'png.jpg'
+        try:
+            cls.asyncio_run_subprocess(
+                ["ffmpeg", "-sseof", "-1", '-i', video_path, '-frames:v', '1',  "-y", jpg_url], timeout=120)
+            return jpg_url
+        except Exception as e:
+            return jpg_url
+
+    """
+    获取视频音频
+    """
+    @classmethod
+    def get_video_mp3(cls, video_file, video_path_url, pw_random_id):
+        pw_mp3_path = video_path_url + str(pw_random_id) +'pw_video.mp3'
+        try:
+            cls.asyncio_run_subprocess([
+                'ffmpeg',
+                '-i', video_file,
+                '-q:a', '0',
+                '-map', 'a',
+                pw_mp3_path
+            ], timeout=120)
+            time.sleep(1)
+            return pw_mp3_path
+        except Exception as e:
+            return pw_mp3_path
+
+    @classmethod
+    def get_pw_video_mp3(cls, video_file, video_path_url):
+        bgm_pw_path_mp3 = video_file + 'bgm_pw.mp3'
+        try:
+            cls.asyncio_run_subprocess([
+                'ffmpeg',
+                '-i', video_path_url,      # 输入的视频文件路径
+                '-q:a', '0',           # 设置音频质量为最佳
+                '-map', 'a',
+                '-y',
+                bgm_pw_path_mp3
+            ], timeout=120)
+            time.sleep(1)
+            return bgm_pw_path_mp3
+        except Exception as e:
+            return bgm_pw_path_mp3
+
+
+    """
+    片尾增加bgm
+    """
+    @classmethod
+    def video_add_bgm(cls, video_file, bgm_path, video_path_url):
+        bgm_pw_path = video_path_url + 'bgm_pw.mp4'
+        pw_duration = cls.get_video_duration(video_file)
+
+        try:
+            pw_path_txt = video_path_url + 'bgm_pw_video.txt'
+            with open(pw_path_txt, 'w') as f:
+                f.write(f"file '{video_file}'\n")
+            cls.asyncio_run_subprocess([
+                "ffmpeg",
+                "-f", "concat",
+                "-safe", "0",
+                "-i", f"{pw_path_txt}",  # 视频序列输入的文本文件
+                "-i", bgm_path,  # 音频文件
+                "-c:v", "libx264",  # 视频编码格式
+                "-t", str(pw_duration),  # 输出视频的持续时间
+                '-c:a', 'aac',           # 音频编码格式
+                '-b:v', '260k',          # 视频比特率
+                '-b:a', '96k',           # 音频比特率
+                '-threads', '2',         # 线程数
+                # '-vf', f'{background_cmd},{subtitle_cmd}',  # 视频过滤器（背景和字幕）
+                "-filter_complex", "[1:a]volume=0.6[a1];[0:a][a1]amerge=inputs=2[aout]",  # 混合音频流
+                "-map", "0:v:0",  # 映射视频流来自第一个输入文件（视频）
+                "-map", "[aout]",  # 映射混合后的音频流
+                '-y',                    # 强制覆盖输出文件
+                bgm_pw_path              # 输出文件路径
+            ], timeout=500)
+            time.sleep(1)
+            return bgm_pw_path
+        except Exception as e:
+            return bgm_pw_path
+
+    """横屏视频改为竖屏"""
+    @classmethod
+    def update_video_h_w(cls, video_path, file_path):
+        video_h_w_path = file_path +'video_h_w_video.mp4'
+        try:
+            cls.asyncio_run_subprocess(["ffmpeg" ,"-i" ,video_path ,"-vf" ,"scale=640:ih*640/iw,pad=iw:iw*16/9:(ow-iw)/2:(oh-ih)/2" ,video_h_w_path],timeout=420)
+            return video_h_w_path
+        except Exception as e:
+            return video_h_w_path
+
+    """视频转为640像素"""
+    @classmethod
+    def video_640(cls, video_path, file_path):
+        video_url = file_path + 'pixelvideo.mp4'
+        try:
+            cls.asyncio_run_subprocess(["ffmpeg" ,"-i" ,video_path ,"-vf" ,"scale=360:640" ,video_url],timeout=420)
+            return video_url
+        except Exception as e:
+            return video_url
+
+    @classmethod
+    def concatenate_videos(cls, videos_paths, file_path):
+        video_url = file_path + 'rg_pw.mp4'
+        list_filename = file_path + 'rg_pw.txt'
+        with open(list_filename, "w") as f:
+            for video_path in videos_paths:
+                f.write(f"file '{video_path}'\n")
+        try:
+            cls.asyncio_run_subprocess(
+                ["ffmpeg", "-f", "concat", "-safe", "0", "-i", list_filename, "-c", "copy", video_url], timeout=420)
+            logger.info(f"[+] 视频转为640像素成功")
+            return video_url
+        except Exception as e:
+            return video_url
+
+    """视频拼接到一起"""
+    @classmethod
+    def h_b_video(cls, video_path, pw_path, file_path):
+        video_url = file_path + 'hbvideo.mp4'
+        try:
+            cls.asyncio_run_subprocess(["ffmpeg","-i", video_path, "-i", pw_path, "-filter_complex" ,"[0:v]scale=360:640[v1]; [1:v]scale=360:640[v2]; [v1][0:a][v2][1:a]concat=n=2:v=1:a=1[outv][outa]" ,"-map" ,"[outv]" ,"-map" ,"[outa]" ,video_url],timeout=500)
+            return video_url
+        except Exception as e:
+            return video_url
+
+    """横屏视频顶部增加字幕"""
+    @classmethod
+    def add_video_zm(cls, new_video_path, video_path_url, pw_random_id, new_text):
+        single_video_srt = video_path_url + str(pw_random_id) +'video_zm.srt'
+        single_video_txt = video_path_url + str(pw_random_id) +'video_zm.txt'
+        single_video = video_path_url + str(pw_random_id) +'video_zm.mp4'
+        try:
+            duration = cls.get_video_duration(new_video_path)
+            if duration == 0:
+                return new_video_path
+            start_time = cls.seconds_to_srt_time(0)
+            end_time = cls.seconds_to_srt_time(duration)
+            # zm = '致敬伟大的教员，为整个民族\n感谢老人家历史向一代伟人'
+            with open(single_video_txt, 'w') as f:
+                f.write(f"file '{new_video_path}'\n")
+            with open(single_video_srt, 'w') as f:
+                f.write(f"1\n{start_time} --> {end_time}\n{new_text}\n\n")
+            subtitle_cmd = f"subtitles={single_video_srt}:force_style='Fontsize=12,Fontname=wqy-zenhei,Outline=2,PrimaryColour=&H00FFFF,SecondaryColour=&H000000,Bold=1,MarginV=225'"
+            draw = f"{subtitle_cmd}"
+            cls.asyncio_run_subprocess([
+                "ffmpeg",
+                "-f", "concat",
+                "-safe", "0",
+                "-i", single_video_txt,
+                "-c:v", "libx264",
+                "-c:a", "aac",
+                "-vf", draw,
+                "-y",
+                single_video
+            ],timeout=500)
+            # subprocess.run(ffmpeg_cmd)
+            return single_video
+        except Exception as e:
+            return single_video
+
+    """获取mp3时长"""
+    @classmethod
+    def get_mp3_duration(cls, file_path):
+        audio = MP3(file_path)
+        duration = audio.info.length
+        if duration:
+            return int(duration)
+        return 0
+
+
+    """
+     生成片尾视频
+    """
+    @classmethod
+    def pw_video(cls, jpg_path, file_path, pw_mp3_path, pw_srt):
+        # 添加音频到图片
+        """
+        jpg_url 图片地址
+        pw_video 提供的片尾视频
+        pw_duration  提供的片尾视频时长
+        new_video_path 视频位置
+        subtitle_cmd 字幕
+        pw_url 生成视频地址
+        :return:
+        """
+        pw_srt_path = file_path +'pw_video.srt'
+        with open(pw_srt_path, 'w') as f:
+            f.write(pw_srt)
+        pw_url_path = file_path + 'pw_video.mp4'
+        try:
+            pw_duration = cls.get_mp3_duration(pw_mp3_path)
+            if pw_duration == 0:
+                return pw_url_path
+            time.sleep(2)
+            # 添加字幕 wqy-zenhei  Hiragino Sans GB
+            height = 1080
+            margin_v = int(height) // 8  # 可根据需要调整字幕和背景之间的距离
+            subtitle_cmd = f"subtitles={pw_srt_path}:force_style='Fontsize=13,Fontname=wqy-zenhei,Outline=0,PrimaryColour=&H000000,SecondaryColour=&H000000,Bold=1,MarginV={margin_v}'"
+            bg_position_offset = (int(360) - 360//8) / 1.75
+            background_cmd = f"drawbox=y=(ih-{int(360)}/2-{bg_position_offset}):color=yellow@1.0:width=iw:height={int(360)}/4:t=fill"
+            if "mp4" in jpg_path:
+                pw_path_txt = file_path + 'pw_path_video.txt'
+                with open(pw_path_txt, 'w') as f:
+                    f.write(f"file '{jpg_path}'\n")
+                cls.asyncio_run_subprocess([
+                    "ffmpeg",
+                    "-f", "concat",
+                    "-safe", "0",
+                    "-i", f"{pw_path_txt}",  # 视频序列输入的文本文件
+                    "-i", pw_mp3_path,  # 音频文件
+                    "-c:v", "libx264",  # 视频编码格式
+                    "-t", str(pw_duration),  # 输出视频的持续时间
+                    "-c:a", "aac",  # 音频编码格式
+                    "-b:v", "260k",  # 视频比特率
+                    "-b:a", "96k",  # 音频比特率
+                    "-threads", "2",  # 线程数
+                    "-vf", f"{background_cmd},{subtitle_cmd}",  # 视频过滤器（背景和字幕）
+                    "-map", "0:v:0",  # 映射视频流来自第一个输入文件（视频）
+                    "-map", "1:a:0",  # 映射音频流来自第二个输入文件（音频）
+                    "-y",  # 强制覆盖输出文件
+                    pw_url_path  # 输出文件路径
+                ], timeout=500)
+            else:
+                cls.asyncio_run_subprocess([
+                    'ffmpeg',
+                    '-loop', '1',
+                    '-i', jpg_path,  # 输入的图片文件
+                    '-i', pw_mp3_path,  # 输入的音频文件
+                    '-c:v', 'libx264',  # 视频编码格式
+                    '-t', str(pw_duration),  # 输出视频的持续时间，与音频持续时间相同
+                    '-pix_fmt', 'yuv420p',  # 像素格式
+                    '-c:a', 'aac',  # 音频编码格式
+                    '-strict', 'experimental',  # 使用实验性编码器
+                    '-shortest',  # 确保输出视频的长度与音频一致
+                    '-vf', f"{background_cmd},{subtitle_cmd}",  # 视频过滤器，设置分辨率和其他过滤器
+                    pw_url_path  # 输出的视频文件路径
+                ], timeout=500)
+            if os.path.exists(pw_srt_path):
+                os.remove(pw_srt_path)
+            return pw_url_path
+        except Exception as e:
+            return pw_url_path
+
+
+    """
+    单个视频拼接
+    """
+    @classmethod
+    def single_video(cls, video_path, file_path, zm):
+        single_video_url = file_path + 'single_video.mp4'
+        single_video_srt = file_path + 'single_video.srt'
+        # 获取时长
+        try:
+            duration = cls.get_video_duration(video_path)
+            if duration == 0:
+                return single_video_url
+            start_time = cls.seconds_to_srt_time(2)
+            end_time = cls.seconds_to_srt_time(duration)
+            single_video_txt = file_path + 'single_video.txt'
+            with open(single_video_txt, 'w') as f:
+                f.write(f"file '{video_path}'\n")
+            if zm:
+                with open(single_video_srt, 'w') as f:
+                    f.write(f"1\n{start_time} --> {end_time}\n<font color=\"red\">\u2764\uFE0F</font>{zm}\n\n")
+                subtitle_cmd = f"subtitles={single_video_srt}:force_style='Fontsize=14,Fontname=wqy-zenhei,Outline=2,PrimaryColour=&H00FFFF,SecondaryColour=&H000000,Bold=1,MarginV=20'"
+            else:
+                subtitle_cmd = f"force_style='Fontsize=14,Fontname=wqy-zenhei,Outline=2,PrimaryColour=&H00FFFF,SecondaryColour=&H000000,Bold=1,MarginV=20'"
+            # 多线程数
+            num_threads = 5
+            # 构建 FFmpeg 命令,生成视频
+            cls.asyncio_run_subprocess([
+                    "ffmpeg",
+                    "-f", "concat",
+                    "-safe", "0",
+                    "-i",  f"{single_video_txt}",
+                    "-c:v", "libx264",
+                    "-c:a", "aac",
+                    '-b:v', '260k',
+                    "-b:a", "96k",
+                    "-threads", str(num_threads),
+                    "-vf", subtitle_cmd,
+                    "-y",
+                     single_video_url
+            ], timeout=400)
+            if os.path.exists(single_video_srt):
+                os.remove(single_video_srt)
+            return single_video_url
+        except Exception as e:
+            return single_video_url
+
+    @classmethod
+    def asyncio_run_subprocess(cls, params: List[str], timeout: int = 30) -> str:
+        async def run_subprocess():
+            process = await asyncio.create_subprocess_exec(
+                params[0],
+                *params[1:],
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+            )
+            try:
+                out, err = await asyncio.wait_for(process.communicate(), timeout=timeout)
+                if process.returncode != 0:
+                    raise IOError(err)
+                return out.decode()
+            except asyncio.TimeoutError:
+                process.kill()
+                out, err = await process.communicate()
+                raise IOError(err)
+        return asyncio.run(run_subprocess())
+
+
+    @classmethod
+    def get_http_duration(cls, videos_path):
+        total_duration = 0
+        for video_path in videos_path:
+            url = "http://101.37.24.17:5555/api/v1/ffmpeg/get_meta"
+            payload = json.dumps({
+                "url": video_path,
+                "referer": ""
+            })
+            headers = {
+                'Authorization': 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyX2lkIjoiNGNhMTI4ZGYtYWMzMy00NWQ2LTg3MmEtMDAzOTk4MGVhM2ViIiwibmFtZSI6Inp5IiwiZXhwIjoyMDUwOTI3MjExfQ.k_rvuESjA62RgPDiLniVgJyLJn3Q8C1Y_AGq3CPRuKI',
+                'Content-Type': 'application/json'
+            }
+
+            try:
+                response = requests.request("POST", url, headers=headers, data=payload, timeout=30)
+                response =  response.json()
+                duration = response['data']['streams'][0]['duration']
+                total_duration += int(float(duration))
+            except Exception as e:
+                print(f"Error processing {video_path}: {e}")
+        return total_duration
+
+
+if __name__ == '__main__':
+    file_path = '/Users/z/Downloads/478de0b6-4e52-44a5-a5d4-967b2cf8ce49'
+    jpg_path = '/Users/z/Downloads/478de0b6-4e52-44a5-a5d4-967b2cf8ce49rg_pixelvideo.mp4'
+    mp3_path='/Users/z/Downloads/478de0b6-4e52-44a5-a5d4-967b2cf8ce49pw_video.mp3'
+    pw_srt = """1
+00:00:00,000 --> 00:00:02,842
+这个视频揭示了中国近代历史上
+
+2
+00:00:02,842 --> 00:00:05,685
+一个鲜为人知却又极为重要的故
+
+3
+00:00:05,685 --> 00:00:05,888
+事
+
+4
+00:00:05,888 --> 00:00:07,106
+真是让人震惊
+
+5
+00:00:07,106 --> 00:00:07,715
+看完后
+
+6
+00:00:07,715 --> 00:00:10,354
+我不禁对历史有了更深的思考
+
+7
+00:00:10,354 --> 00:00:12,588
+让我们一起重温这段历史
+
+8
+00:00:12,588 --> 00:00:14,212
+提醒自己珍惜当下
+
+9
+00:00:14,212 --> 00:00:17,055
+我相信很多朋友也会对这个话题
+
+10
+00:00:17,055 --> 00:00:17,664
+感兴趣
+
+11
+00:00:17,664 --> 00:00:20,506
+请把这个视频分享到你们的群聊
+
+12
+00:00:20,506 --> 00:00:20,709
+中
+
+13
+00:00:20,709 --> 00:00:22,740
+让更多人了解这段历史
+
+14
+00:00:22,820 --> 00:00:23,824
+共鸣与反思
+
+15
+00:00:23,824 --> 00:00:25,430
+是我们共同的责任
+
+16
+00:00:25,430 --> 00:00:28,242
+也许我们能从中汲取更多的智慧
+
+17
+00:00:28,242 --> 00:00:28,844
+与力量
+
+18
+00:00:28,844 --> 00:00:29,848
+快动动手指
+
+19
+00:00:29,848 --> 00:00:32,659
+让我们一起分享这段重要的历史
+
+20
+00:00:32,659 --> 00:00:32,860
+吧"""
+    FFmpeg.pw_video(jpg_path, file_path, mp3_path, pw_srt)
+
+

utils/gpt4o_mini_help.py

@@ -0,0 +1,238 @@
+import json
+import time
+
+import requests
+
+
+class GPT4oMini:
+
+    @classmethod
+    def get_ai_mini_pw(cls, title):
+
+        prompt = """"请针对微信平台视频类小程序场景，面向人群是中国中老年人，在单聊、群聊场景，"
+                "根据以下示例生成结尾引导分享的脚本，引导老年用户分享给好友和群，必须有6个带引导分享的句子，以老年人祝福的语气结尾。\n"
+                "示例case1：\n"
+                "标题: 几十年前的老歌，现在很少听到了！\n"
+                "返回分享引导脚本: 这个视频真是难得，太美太好听了，看完真是回忆满满，" "让人怀念那个时代，多么淳朴多么美好，今天大家都在群里转发这个视频，"  "老朋友你还记得吗？如果你也跟我一样深深的怀念，赶紧把这个视频转发到你所有群里去吧，" "有多少群就转发多少群，不要私藏" "‌每一次分享都是一次奉献，一份真诚的情感‌‌" "转发给每一位都怀念当年岁月的老友，他们都会感谢你的" "多多转发，情谊多多！,"
+                "祝每位老友健健康康，平平安安\n\n" "示例case2：\n"
+                "标题: 🔺令人尊敬的老人❗❗这才是我们应该追的明星❗\n"
+                "返回分享引导脚本: 这个视频发出来5天了，转发的人却寥寥无几，"
+                "试想一次明星出轨就全民皆知，这样中国真正的正能量却无人关注，真是太讽刺了！"
+                "如果你看到了这个视频，请分享到你加入的每个群，向杨老这种无私奉献的大爱致敬！" "发给5个群和5个好友，"
+                "千万别让那些低调做好事的人心寒！转发传递下去，让世界都看到，中华民族的正能量！"
+                "我已经转发了，也希望你能帮助传递"
+                 "赠人玫瑰，手留余香，分享好视频，收获好福气\n\n" """
+        url = "http://aigc-api.cybertogether.net/aigc/dev/test/gpt"
+        payload = json.dumps({
+            "imageList": [],
+            "model": "gpt-4o-mini-2024-07-18",
+            "prompt": (
+                prompt+
+                f"请分析该标题，标题为：{title}，返回结尾分享引导脚本。"
+            ),
+            "responseFormat": {
+                "type": "json_schema",
+                "json_schema": {
+                    "strict": True,
+                    "name": "share_script_result",
+                    "schema": {
+                        "type": "object",
+                        "properties": {
+                            "分享引导脚本": {
+                                "type": "string",
+                                "description": "生成的结尾分享引导脚本"
+                            }
+                        },
+                        "required": ["分享引导脚本"],
+                        "additionalProperties": False
+                    }
+                }
+            }
+        })
+        headers = {'Content-Type': 'application/json'}
+        try:
+            response = requests.post(url, headers=headers, data=payload, timeout=30)
+            response_data = response.json()
+
+            data = json.loads(response_data.get('data', '{}'))
+            pw  = data["分享引导脚本"]
+            return pw
+        except Exception as e:
+            pw_srt_text = '老友们，这个视频说的太好了,请你在看完之后,不要私藏,分享给身边的亲人朋友,转发传递这份美好,这不仅是分享一个好视频,更是在传递一份深厚的祝福和好运,愿每个看到这消息的人,都能收获健康长寿和平安,希望您在看完后能够回传给我,并且把这条视频再传给3个群和3个好友,祝愿你我,健健康康,平平安安,因为3个数字代表着健康和平安,这么好的视频,千万不要在你的手里终止了,分享给群友们,他们看到以后一定会感谢你的,感谢分享,感谢传递'
+            return pw_srt_text
+
+    @classmethod
+    def get_ai_mini_title(cls, title):
+        url = "http://aigc-api.cybertogether.net/aigc/dev/test/gpt"
+        payload = json.dumps({
+            "imageList": [],
+            "model": "gpt-4o-mini-2024-07-18",
+            "prompt": (
+                        "CONTEXT"
+                        "你是一名短视频标题优化专家，任务是为短视频生成吸引力高且符合规范的标题。"
+                        "OBJECTIVE"
+                        "基于优秀标题要求及示例，生成新的、吸引用户注意的标题。"
+                        "TIME"
+                        "在标题生成请求提出后的1分钟内生成。"
+                        "ACTOR"
+                        "短视频标题优化专家"
+                        "RESOURCES"
+                        "- 原标题列表"
+                        "- 优秀标题示例"
+                        "- 标题生成规范和约束条件"
+                        "RESPONSE"
+                        "为每个原标题生成符合规范的新标题。"
+                        "生成示例标题案例及要求"
+                        "1. 标题开头包含醒目emoji🔴，整体字符长度必须控制在10-25个字"
+                        "2. 识别需要修改的标题中可能感兴趣的人群，并在标题中表现出来。人群示例如：群友们、退休人员、50~70后等"
+                        "3. 标题中可增加一些对内容的观点/态度，用人格化的方式表达，示例：太香了、老外至今难以相信"
+                        "4. 标题结尾可以根据标题内容增加一些引导语，格式参考示例但不必局限于示例。示例：你们见过吗、你听对不对、说的太好了、请听、太神奇了"
+                        "5. 对于包含#话题和@人的标题，则去除@人元素后，利用剩下的全部信息生成标题。"
+                        "6. 不能编造、新增原标题没有的信息，如原标题没有养老金，不能在生成的标题中出现养老金。"
+                        "7. 标题内不能使用强引导分享点击的词句，如：快来看看、大家都听一听、值得一看、都看看吧、你也来看看吧、大家注意、都听听等。"
+                        "8. 不能使用无实质信息和强烈诱导点击、紧急、夸张、震惊的描述，避免使用“震惊国人”、“速看”、“太震撼了”等类似描述。"
+                        "9. 标题需要简洁、清晰，不要使用网络流行语，如：太燃了、佛系、躺平、内卷等。"
+                    
+                        "Goodcase示例："
+                        "⭕老外至今难以相信，中国人竟能把大桥建到天上，穿入云中"
+                        "🔴未来酒店体验，群友们，请看！"
+                        "⭕六七十年代的《忠字舞》，你们见过吗？"
+                        "🔴哈哈哈！大哥说的太好了！太真实了"
+                        "🔴今天，请记住那1700个集体赴死的年轻人，平均23岁！"
+                        "🔴这才叫老同学聚会，到了这个年纪，还能聚在一起真不容易！"
+                        "🔴百善孝为先，心凉了捂不热了"
+                        "🔴养儿不如养狗，一件真实的事！"
+                        "🔴让人受益的一段话，写得真好!"
+                        "🔴“处暑三劝”！发给最好的朋友劝一劝！"
+                        "🔴世间公道自在人心，善恶有报，人生智慧建议收藏！"
+                        "🔴坐着电梯登上山顶！这怕是只有中国人敢想敢做！"
+                        "🔴人老了！就应该这样去活！"
+                    
+                        "badcase示例："
+                        "🌸绝对不能错过，快打开看看，越快越好"
+                        "所有老年人一定要看"
+                        "天大的好消息，5月开始实施❗"
+                        "就在刚刚，中国突然传出重磅消息，所有人都不敢相信！ 🚩"
+                        "丧尽天良！为什么生病的人越来越多，原来吃的是这些 🎈"
+                        "今年的端午节太特殊，一辈子难遇一次!一定要看!错过别后悔"
+                        "好消息来了，千万别划走！"
+                        "紧急！已爆发，错过就晚了😱"
+                    f"请分析该标题，标题为：{title}，返回新的标题。"
+                    ),
+            "responseFormat": {
+                "type": "json_schema",
+                "json_schema": {
+                    "strict": True,
+                    "name": "share_script_result",
+                    "schema": {
+                        "type": "object",
+                        "properties": {
+                            "新标题": {
+                                "type": "string",
+                                "description": "生成新的标题"
+                            }
+                        },
+                        "required": ["新标题"],
+                        "additionalProperties": False
+                    }
+                }
+            }
+        })
+        headers = {'Content-Type': 'application/json'}
+        max_retries = 3
+        retry_count = 0
+        while retry_count < max_retries:
+            try:
+                response = requests.post(url, headers=headers, data=payload, timeout=30)
+                response_data = response.json()
+
+                data = json.loads(response_data.get('data', '{}'))
+                new_title = data["新标题"]
+                return new_title
+            except Exception as e:
+                retry_count += 1
+                # logger.error(f"尝试第 {retry_count} 次失败，错误: {e}")
+                time.sleep(1)  # 延迟1秒后重试
+        return "这个视频，分享给我的老友，祝愿您能幸福安康"
+
+    @classmethod
+    def get_content_understanding_pw(cls, pw):
+        """AI标题"""
+        url = "http://aigc-api.cybertogether.net//aigc/dev/test/gpt"
+        payload = json.dumps({
+            "imageList": [],
+            "model": "gpt-4o-mini-2024-07-18",
+            "prompt": (
+                "你是一名专业的短视频分析师，工作是帮助短视频平台的视频撰写视频结尾用于引导用户分享视频的文案。视频的主要用户为60岁以上中国老年人，请你理解视频内容，根据规则选择选择模板，并结合视频内容输出用于引导用户分享视频的文案。"
+
+                "请注意："
+                "1、总的内容输出在100~150字之间，绝对不能低于100字，绝对不能超过150字，如果超过150字，可以针对模板进行适当删减。"
+                "2、针对视频核心内容的提炼，内容需要适配老年人的阅读习惯和理解能力，禁止出现太燃了、佛系、躺平、内卷、炸裂等形容词，以及阶级固化等专业名词。"
+                "3、模板中[]内的内容填充，不要超过15个字。"
+                
+                "规则："
+                "1、视频内容是否围绕着健康科普/安全防范/政策解读/生活技巧/情感激励等知识科普/信息通知的主题？若是，则根据视频内容在“科普/信息分享模板”中，选择合适的模板进行输出，无需处理其他规则。若不是，则继续处理规则2。"
+                "2、视频内容是否围绕着对感动/温情/趣味/祝福、罕见画面、社会正能量等正面主题？若是，则根据视频内容在“正面情绪模板”中，选择合适的模板进行输出，无需处理其他规则。若不是，则继续处理规则3。"
+                "3、频内容是否围绕着对社会风气/现象的不满/批判、对生活的不满等主题？若是，则根据视频内容在“负面情绪模板”中，选择合适的模板进行输出，无需处理其他规则。若不是，则继续处理规则4。"
+                "4、若视频同时不符合规则1、2、3，则根据视频内容，在所有模板中选择适配的模板进行输出。"
+                
+                "输出内容包含："
+                "1、视频内容符合规则1/2/3/4中的哪一条？"
+                "2、你选择的是哪一类模板，具体哪一条模板？"
+                "3、引导语文案"
+                
+                "模板List："
+                "“科普/信息分享模板”"
+                "模板1：健康科普类 ""这个视频讲解的[健康知识]真是太实用了，很多人都不知道原来[具体内容]还有这么多讲究。看完后我才明白，平时我们[具体行为]都做错了。群里的朋友们都说，这些知识太及时了，简直是生活中的智慧宝典。老朋友们，你们[相关问题]做对了吗？如果觉得这个视频对你有帮助，请马上转发到3个群里，让更多人了解正确的[健康知识]。记住，健康是最重要的，您的转发可能会帮助一位需要的朋友，让我们一起关爱健康，守护幸福。多多分享，感恩有你。"
+                "模板2：安全防范类 ""各位老友们要注意了，这个视频提醒的[安全隐患]太重要了，现在的[具体威胁]真是防不胜防。看完后我心惊肉跳，原来我们身边竟然有这么多危险。为了家人的安全，请立即把这个视频转发到至少3个群里，让大家都提高警惕。特别是家里有老人和孩子的，一定要多加注意。记住，一个不小心就可能造成无法挽回的损失。您的一次转发，可能就会挽救一个家庭。让我们一起守护平安，转发传递，功德无量。"
+                "模板3：政策解读类 ""重要通知！这个关于[最新政策]的视频太及时了，从[具体日期]开始实施的新规定，将给我们的生活带来重大变化。很多老年朋友还不知道这些新政策，可能会影响到我们的切身利益。为了不让身边的朋友吃亏，请大家一定要把这个视频转发到至少3个群里。让更多人了解这些新变化，提前做好准备。您的分享就是对他人的关爱，让我们互帮互助，共同进步。转发分享，福气多多。"
+                "模板4：生活技巧类 ""看完这个视频，我才知道原来[日常行为]还可以这样做，太神奇了！视频里教的这些[具体技巧]，简直是生活小妙招。群里的朋友们都说学到了，这些方法太实用了。特别是对我们这些[特定人群]来说，掌握这些技巧后，生活会变得更加便利。如果你也觉得有用，请马上转发到3个群，分享给身边的朋友，让大家一起学习这些生活智慧。记住，分享就是快乐，您的转发一定会帮助到需要的人。多多转发，感恩有你。"
+                "模板5：情感激励类 ""这个视频的内容太有意义了，让我深受触动。在这个快节奏的社会里，我们常常忽视了[具体内容]的重要性。视频里说的每一句话都说到心坎里，相信看完的朋友们都会有同感。为了让更多人受益，请把这个视频转发到至少3个群里。特别是那些正在经历[具体困境]的朋友，看了一定会很有启发。记住，您的一次分享，可能就会改变一个人的生活。让我们用爱心传递温暖，共同创造更美好的明天。转发分享，福气满满。"
+                
+                "“正面情绪模板”"
+                "模板1：感动温情类 ""亲爱的老朋友们，这个视频真是太[感动/温馨/暖心]了，看到[具体内容]的画面，让人心里暖暖的，相信您和我一样被深深打动。这份[幸福/感动/温情]太珍贵了，值得与更多人分享。请您一定要把这个视频转发到您所在的3个微信群，让更多的家人朋友感受这份美好。您的分享就是一份关爱，群友们看到后一定会感谢您的。多一次转发，多一份温暖，让我们一起传递这份感动，还在等什么，赶紧行动起来吧！"
+                "模板2：趣味互动类 ""老朋友们，这个[有趣/精彩]的视频里，您觉得[具体内容]哪一个最棒呢？快来说说您的看法吧！看完后千万别私藏，一定要把这么精彩的内容转发分享到您所在的3个微信群，让群友们也一起来欣赏评论。这么有意思的视频，大家看到后一定会眼前一亮，纷纷感谢您的分享。记住，您的每一次转发都能带来好运哦！多多转发，福气多多，快乐多多。动动手指，传递欢乐，一起分享精彩时刻！"
+                "模板3：祝福传递类 ""亲爱的朋友，这个充满[祝福/好运/福气]的视频，我第一时间就想到要分享给您。据说看到这个视频的人，只要转发给八个好友或三个群，就能收获满满的[福气/好运/健康]。我已经把这份祝福传递给了最重要的朋友，其中就包括您。请您也一定要把这份美好传递下去，转发到您所在的微信群，让更多人感受这份祝福。记住，千万不要让祝福在您手中终止，多多转发，幸福多多！"
+                "模板4：文化传承类 ""老友们，这段展现[中华文化/传统技艺/民族精神]的视频太珍贵了！看到[具体内容]，让人深感自豪。作为炎黄子孙，我们要把这份文化瑰宝传承下去。请您一定要把视频转发到3个群，让更多人了解我们的传统文化。您的每一次转发，都是对中华文化的传播，都是对传统的守护。让我们共同努力，让中华文化发扬光大。转发视频，传承文明，让更多人感受中华文化的魅力！"
+                "模板5：稀奇见闻类 ""亲爱的朋友，这个展现[自然奇观/稀有现象/独特事物]的视频实在太难得了！这么[神奇/罕见/精彩]的画面，错过太可惜了。我第一时间就想分享给身边的好友，也包括您。请您看完后也别忘了转发分享到您所在的3个微信群，让群友们也大开眼界。这么稀奇的视频，绝对不能在您这里终止啊！据说转发的人都能沾上好运气呢。多多转发，惊喜多多，让我们一起分享这份难得的奇观，传递这份惊喜！"
+                
+                "“负面情绪模板”"
+                "模板1：真实共鸣型 ""这个视频说出了我们的心里话，[视频核心内容提炼]真是太真实了。相信很多朋友看了都会有同感，这样难得的真话不常见。如果您也觉得说得对，请一定要转发分享给3个群的群友们看看。这样的真实声音，明天可能就听不到了。您的每一次转发，都是在发声，都是在传递真相。让我们一起行动起来，点点转发，让更多人听到这样的心声。"
+                "模板2：善意提醒型 ""老朋友们，这个视频太重要了，[视频核心内容提炼]对我们每个人都很有帮助。看完后请您一定要转发分享到3个群，让所有的家人朋友都能从中受益。您的转发是对身边人的关心，是在传递祝福，他们看完一定会感谢您的。记得多多转发，这样明天您想再看的时候就能马上找到了。让我们一起把这份关爱传递下去，让更多人受益。祝愿您和家人平安健康！"
+                "模板3：社会责任型 ""这个视频提到的[视频核心内容]，关系到每个人的切身利益。如果您也认同，请动动手指转发到3个群，让所有人都能看到。这不仅仅是一个视频的传播，更是一份社会责任。请老友们多多转发，让更多人了解这个问题，引起全社会的重视。千万不要让这个视频在您这里停止传播，您的每一次转发都是在为社会尽一份力。让我们共同行动，让社会变得更好。"
+                "模板4：情感共振型 ""看完这个视频，[视频核心内容提炼]真是让人感触良多。群里的朋友们都在转发，说看完后特别有感触。这样充满智慧和温度的内容值得分享，请您也加入传播的队伍，转发到您的群里，让更多人感受这份温暖。每一次转发都是在传递一份爱与理解，都是在为这个社会增添一份温暖。让我们一起把这份美好传递下去，让更多人受益。"
+                "模板5：紧迫感召型 ""这个视频太重要了，[视频核心内容提炼]说得太对了！如果您看到这个视频，说明大家都在转发支持。请您立即行动起来，转发到您所有的群，一人行动十人知晓，百人传递，千人支持。现在就是行动的最好时机，明天再想看可能就找不到了。让我们一起努力，让这个声音传得更远。您的每一次转发，都是在发声，赶快行动起来！ "
+                f"请分析该内容，视频脚本内容为：{pw}，返回新的片尾。"
+            ),
+            "responseFormat": {
+                "type": "json_schema",
+                "json_schema": {
+                    "strict": True,
+                    "name": "share_script_result",
+                    "schema": {
+                        "type": "object",
+                        "properties": {
+                            "新片尾": {
+                                "type": "string",
+                                "description": "生成新的片尾"
+                            }
+                        },
+                        "required": ["新片尾"],
+                        "additionalProperties": False
+                    }
+                }
+            }
+        })
+        headers = {'Content-Type': 'application/json'}
+        response = requests.post(url, headers=headers, data=payload)
+        response_data = response.json()
+
+        data = json.loads(response_data.get('data', '{}'))
+        new_pw = data["新片尾"]
+        if new_pw:
+            return new_pw
+        else:
+            return None
+
+if __name__ == '__main__':
+    GPT4oMini.get_ai_mini_pw("这段话说出了多少人的心声 #老百姓的心声 #老百姓关心的话题 #农民的心声 #老百姓不容易","AI片尾引导1")

utils/mysql_db.py

@@ -0,0 +1,66 @@
+# -*- coding: utf-8 -*-
+"""
+数据库连接及操作
+"""
+import pymysql
+
+class MysqlHelper:
+    @classmethod
+    def connect_mysql(cls):
+        # 创建一个 Connection 对象，代表了一个数据库连接
+        connection = pymysql.connect(
+            host="rm-bp14529nwwcw75yr1ko.mysql.rds.aliyuncs.com",
+            port=3306,  # 端口号
+            user="changwen_admin",  # mysql用户名
+            passwd="changwen@123456",  # mysql用户登录密码
+            db="long_articles",  # 数据库名
+            # 如果数据库里面的文本是utf8编码的，charset指定是utf8
+            charset="utf8mb4")
+        return connection
+
+    @classmethod
+    def get_values(cls, sql, params=None):
+        try:
+            # 连接数据库
+            connect = cls.connect_mysql()
+            # 返回一个 Cursor对象
+            mysql = connect.cursor()
+
+            if params:
+                # 如果传递了 params 参数
+                mysql.execute(sql, params)
+            else:
+                # 如果没有传递 params 参数
+                mysql.execute(sql)
+            # fetchall方法返回的是一个元组，里面每个元素也是元组，代表一行记录
+            data = mysql.fetchall()
+
+            # 关闭数据库连接
+            connect.close()
+
+            # 返回查询结果，元组
+            return data
+        except Exception as e:
+            print(f"get_values异常:{e}\n")
+
+    @classmethod
+    def update_values(cls, sql):
+        # 连接数据库
+        connect = cls.connect_mysql()
+        # 返回一个 Cursor对象
+        mysql = connect.cursor()
+        try:
+            # 执行 sql 语句
+            res = mysql.execute(sql)
+            # 注意 一定要commit，否则添加数据不生效
+            connect.commit()
+            return res
+        except Exception as e:
+            # 发生错误时回滚
+            connect.rollback()
+        # 关闭数据库连接
+        connect.close()
+
+
+
+

utils/sql_help.py

@@ -0,0 +1,40 @@
+import os
+import sys
+from datetime import datetime
+
+sys.path.append(os.getcwd())
+from utils.mysql_db import MysqlHelper
+
+
+class sqlCollect:
+
+    """
+    获取长文oss地址
+    """
+    @classmethod
+    def get_oss_path(cls):
+        sql = """
+                SELECT title, oss_path
+                FROM video_end_screen_transformation_task
+                WHERE status = 0
+                ORDER BY task_id ASC
+                LIMIT 1;
+            """
+        data = MysqlHelper.get_values(sql)
+        return data
+
+    @classmethod
+    def update_oss_path_status(cls, status, oss_path):
+        sql = f"""UPDATE video_end_screen_transformation_task set status = {status} where oss_path = '{oss_path}'"""
+        res = MysqlHelper.update_values(
+            sql=sql
+        )
+        return res
+
+    @classmethod
+    def add_new_oss_path(cls, new_oss_path, oss_path):
+        sql = f"""UPDATE video_end_screen_transformation_task set new_oss_path = '{new_oss_path}' where oss_path = '{oss_path}'"""
+        res = MysqlHelper.update_values(
+            sql=sql
+        )
+        return res

utils/tts_help.py

@@ -0,0 +1,207 @@
+from datetime import timedelta
+
+import requests
+import json
+import random
+import re
+import time
+
+
+
+class TTS:
+    @classmethod
+    def get_pw_zm(cls, text, voice):
+        max_retries = 3
+        for attempt in range(max_retries):
+            url = "http://api.piaoquantv.com/produce-center/speechSynthesis"
+            payload = json.dumps({
+                "params": {
+                    "text": text,
+                    "voice": voice,
+                    # "vocie": "zhiyuan",
+                    "format": "pcm",
+                    "volume": 90,
+                    "speechRate": 80,
+                    "pitchRate": 0
+                }
+            })
+
+            headers = {
+                'Content-Type': 'application/json'
+            }
+            wait_time = random.uniform(1, 10)
+            time.sleep(wait_time)
+            try:
+                response = requests.request("POST", url, headers=headers, data=payload, timeout=60)
+                response = response.json()
+                code = response["code"]
+                if code == 0:
+                    mp3 = response["data"]
+                    return mp3
+            except Exception:
+                if attempt == max_retries - 1:
+                    return None
+        return None
+
+    """
+    音频下载到本地
+    """
+    @classmethod
+    def download_mp3(cls,  pw_url, file_path):
+        pw_mp3_path = file_path +'pw_video.mp3'
+        for i in range(3):
+            payload = {}
+            headers = {}
+            response = requests.request("GET", pw_url, headers=headers, data=payload, timeout= 30)
+            if response.status_code == 200:
+                # 以二进制写入模式打开文件
+                with open(f"{pw_mp3_path}", "wb") as file:
+                    # 将响应内容写入文件
+                    file.write(response.content)
+                return pw_mp3_path
+        return None
+
+    @classmethod
+    def get_srt_format(cls, pw_srt_text, pw_url_sec):
+        segments = re.split(r'(，|。|！|？)', pw_srt_text)
+        segments = [segments[i] + segments[i + 1] for i in range(0, len(segments) - 1, 2)]
+        pw_url_sec = int(pw_url_sec) + 1
+        # 确定每段显示时间
+        num_segments = len(segments)
+        duration_per_segment = pw_url_sec / num_segments
+        srt_content = ""
+        start_time = 0.0
+        for i, segment in enumerate(segments):
+            end_time = start_time + duration_per_segment
+            srt_content += f"{i + 1}\n"
+            srt_content += f"{int(start_time // 3600):02}:{int((start_time % 3600) // 60):02}:{int(start_time % 60):02},{int((start_time % 1) * 1000):03} --> "
+            srt_content += f"{int(end_time // 3600):02}:{int((end_time % 3600) // 60):02}:{int(end_time % 60):02},{int((end_time % 1) * 1000):03}\n"
+            srt_content += f"{segment.strip()}\n\n"
+            start_time = end_time
+
+        print(srt_content)
+        return srt_content
+
+    @classmethod
+    def process_srt(cls, srt):
+        lines = srt.strip().split('\n')
+        processed_lines = []
+
+        for line in lines:
+            if re.match(r'^\d+$', line):
+                processed_lines.append(line)
+            elif re.match(r'^\d{2}:\d{2}:\d{2}\.\d{1,3}-->\d{2}:\d{2}:\d{2}\.\d{1,3}$', line):
+                processed_lines.append(line.replace('-->', ' --> '))
+            else:
+                line = re.sub(r'[，。！？；、]$', '', line)
+                # 添加换行符
+                processed_lines.append(line + '\n')
+
+        return '\n'.join(processed_lines)
+
+    @classmethod
+    def parse_timecode(cls, timecode):
+        h, m, s = map(float, timecode.replace(',', '.').split(':'))
+        return timedelta(hours=h, minutes=m, seconds=s)
+
+    @classmethod
+    def format_timecode(cls, delta):
+        total_seconds = delta.total_seconds()
+        hours, remainder = divmod(total_seconds, 3600)
+        minutes, seconds = divmod(remainder, 60)
+        return f"{int(hours):02}:{int(minutes):02}:{seconds:06.3f}".replace('.', ',')
+
+    @classmethod
+    def split_subtitle(cls, subtitle_string):
+        max_len = 14
+        lines = subtitle_string.strip().split('\n')
+        subtitles = []
+        for i in range(0, len(lines), 4):
+            sub_id = int(lines[i].strip())
+            timecode_line = lines[i + 1].strip()
+            start_time, end_time = timecode_line.split(' --> ')
+            text = lines[i + 2].strip()
+            if re.search(r'[a-zA-Z]', text):
+                text = re.sub(r'[a-zA-Z]', '', text)
+            start_delta = cls.parse_timecode(start_time)
+            end_delta = cls.parse_timecode(end_time)
+            total_duration = (end_delta - start_delta).total_seconds()
+            char_duration = total_duration / len(text)
+
+            current_start = start_delta
+            for j in range(0, len(text), max_len):
+                segment = text[j:j + max_len]
+                current_end = current_start + timedelta(seconds=char_duration * len(segment))
+                subtitles.append((sub_id, current_start, current_end, segment))
+                current_start = current_end
+                sub_id += 1
+
+        return subtitles
+
+    @classmethod
+    def generate_srt(cls, subtitles):
+        srt_content = ''
+        for idx, sub in enumerate(subtitles, start=1):
+            srt_content += f"{idx}\n"
+            srt_content += f"{cls.format_timecode(sub[1])} --> {cls.format_timecode(sub[2])}\n"
+            srt_content += f"{sub[3]}\n\n"
+        return srt_content.strip()
+
+    @classmethod
+    def getSrt(cls, mp3_id):
+        url = "http://api.piaoquantv.com/produce-center/srt/get/content"
+
+        payload = json.dumps({
+            "params": {
+                "resourceChannel": "outer",
+                "videoPath": mp3_id
+            }
+        })
+        headers = {
+            'User-Agent': 'Apifox/1.0.0 (https://apifox.com)',
+            'Content-Type': 'application/json',
+            'Accept': '*/*',
+            'Host': 'api-internal.piaoquantv.com',
+            'Connection': 'keep-alive'
+        }
+
+        response = requests.request("POST", url, headers=headers, data=payload, timeout=30)
+        time.sleep(1)
+        data_list = response.json()
+        code = data_list["code"]
+        if code == 0:
+            srt = data_list["data"]
+            if srt:
+                srt = srt.replace("/n", "\n")
+                new_srt = cls.process_srt(srt)
+                result = cls.split_subtitle(new_srt)
+                # 生成SRT格式内容
+                srt_content = cls.generate_srt(result)
+                return srt_content
+            else:
+                return None
+        else:
+            return None
+
+
+if __name__ == '__main__':
+    # text = "真是太实用了，分享给身边的准妈妈们吧！这些孕期禁忌一定要记住，赶紧转发给更多人，帮助更多的宝妈们。一起为宝宝的健康加油！"
+    # mp3 = TTS.get_pw_zm(text)
+    # print(mp3)
+    # command = [
+    #   'ffmpeg',
+    #   '-i', mp3,
+    #   '-q:a', '0',
+    #   '-map', 'a',
+    #   # '-codec:a', 'libmp3lame',  # 指定 MP3 编码器
+    #   "/Users/tzld/Desktop/video_rewriting/path/pw_video.mp3"
+    # ]
+    # subprocess.run(command)
+    # print("完成")
+    video_file = 'http://clipres.yishihui.com/longvideo/crawler/voice/pre/20240821/37fbb8cfc7f1439b8d8a032a1d01d37f1724219959925.mp3'
+    TTS.getSrt(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)
+    # print(float(result.stdout))

workers/consumption_work.py

@@ -0,0 +1,155 @@
+import os
+import sys
+import uuid
+from apscheduler.schedulers.blocking import BlockingScheduler
+from apscheduler.triggers.interval import IntervalTrigger
+from loguru import logger
+
+from utils.gpt4o_mini_help import GPT4oMini
+from utils.tts_help import TTS
+
+sys.path.append('/app')
+from utils.download_video import DownLoad
+from utils.sql_help import sqlCollect
+from utils.aliyun_oss import Oss
+from utils.ffmpeg import FFmpeg
+
+
+
+
+
+
+
+# CACHE_DIR = '/app/cache/'
+CACHE_DIR = '/Users/z/Downloads/'
+class ConsumptionRecommend(object):
+
+
+
+    @classmethod
+    def data_handle(cls, oss_url, title, file_path):
+        video_url = f"https://rescdn.yishihui.com/{oss_url}"
+        logger.info(f"[长文] {video_url}视频开始下载")
+        video_path = DownLoad.download_video(video_url, file_path)
+        if not os.path.exists(video_path) or os.path.getsize(video_path) == 0:
+            """下载失败修改状态"""
+            sqlCollect.update_oss_path_status(99, oss_url)
+            return
+        logger.info(f"[长文] {video_url}视频下载成功")
+        width, height = FFmpeg.get_w_h_size(video_path)
+        if width < height:  # 判断是否需要修改为竖屏
+            logger.info(f"[长文] {video_url}视频修改为竖屏")
+            video_path = FFmpeg.update_video_h_w(video_path, file_path)
+        logger.info(f"[长文] 视频更改分辨率处理")
+        video_path = FFmpeg.video_640(video_path, file_path)
+
+        logger.info(f"[长文] 视频更改分辨率处理成功")
+
+        pw_srt_text = GPT4oMini.get_ai_mini_pw(title)
+
+        pw_url = TTS.get_pw_zm(pw_srt_text, "zhifeng_emo")
+        if not pw_url:
+            """获取片尾引导失败"""
+            sqlCollect.update_oss_path_status(99, oss_url)
+            return
+        pw_srt = TTS.getSrt(pw_url)
+        if not pw_srt:
+            """获取片尾引导失败"""
+            sqlCollect.update_oss_path_status(99, oss_url)
+            return
+        logger.info(f"[长文] 开始下载音频")
+        pw_mp3_path = TTS.download_mp3(pw_url, file_path)
+        if not pw_mp3_path:
+            """片尾音频下载失败"""
+            sqlCollect.update_oss_path_status(99, oss_url)
+            return
+        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:
+            """片尾生成失败"""
+            sqlCollect.update_oss_path_status(99, oss_url)
+
+            return
+        logger.info(f"[长文] 合并开始拼接")
+        video_path = FFmpeg.h_b_video(video_path, pw_path, file_path)
+        zm = "温馨提示：\n点击下方按钮，传递好运"
+
+        video_path = FFmpeg.single_video(video_path, file_path, zm)
+
+        if not os.path.exists(video_path) or os.path.getsize(video_path) == 0:
+            """视频合并失败"""
+            sqlCollect.update_oss_path_status(99, oss_url)
+
+            return
+        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")
+        logger.info(f"[长文] oss新地址写入数据")
+        sqlCollect.add_new_oss_path(oss_object_key,oss_url)
+        logger.info(f"[长文] oss新地址写入成功")
+        logger.info(f"[长文] 修改数据库状态")
+        sqlCollect.update_oss_path_status(2, oss_url)
+        logger.info(f"[长文] 修改数据库状态成功")
+        return
+
+
+    @classmethod
+    def run(cls):
+        uid = str(uuid.uuid4())
+        file_path = os.path.join(CACHE_DIR, uid)
+        logger.info(f"[长文] 查询数据")
+        data = sqlCollect.get_oss_path()
+        title = data[0][0]
+        oss_url = data[0][1]
+        if not oss_url:
+            logger.info(f"[长文] 无待处理的数据")
+            return
+        try:
+
+            sqlCollect.update_oss_path_status(1, oss_url)
+            cls.data_handle(oss_url, title, 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=1))  # 每1分钟启动一次
+        scheduler.start()
+    except KeyboardInterrupt:
+        pass
+    except Exception as e:
+        logger.error(f"[长文] 启动异常，异常信息:{e}")
+        pass
+    finally:
+        scheduler.shutdown()
+
+
+if __name__ == '__main__':
+    ConsumptionRecommend.run()