# -*- coding: utf-8 -*- import os import random import subprocess from pathlib import Path from typing import List, Tuple from concurrent.futures import ThreadPoolExecutor from pydub import AudioSegment import tempfile import time from datetime import datetime import re import json import re from auto_media_publisher.utils.logger import get_logger from auto_media_publisher.config.conf_video_mashup import VIDEO_DIR,FINAL_VIDEO_DIR,VIDEO_LOGO_PATH,VIDEO_SEGMENT_COUNT,VIDEO_SEGMENT_DURATION,VIDEO_RESOLUTION_DEFAULT_WIDTH,VIDEO_RESOLUTION_DEFAULT_HEIGHT,VIDEO_FONT_PATH,VIDEO_FONT_SIZE,VIDEO_FONT_COLOR from auto_media_publisher.generators.media_generator_base import MediaGenerator class VideoGeneratorMashup(MediaGenerator): """ 视频生成器类，用于根据参数生成短视频。 Attributes: **kwargs : 其中包括 user_id,platform,proxy_full 和其他用得着addon 参数。 本类中,addon 参数是 video_type 具体参考 conf_video_mashup.py """ required_keys = ['user_id', 'scheduled_time','proxy_full'] def __init__(self, *args, **kwargs): if args: raise TypeError("不支持位置参数，请全部使用关键字参数") for key in self.required_keys: if key not in kwargs: raise ValueError(f"缺少必要参数：{key}") setattr(self, key, kwargs[key]) # 允许设置额外字段 for key, value in kwargs.items(): if key not in self.required_keys: setattr(self, key, value) self.logger = get_logger(self.__class__.__name__,"admin") if not self.video_width : self.video_width = VIDEO_RESOLUTION_DEFAULT_WIDTH if not self.video_height : self.video_height = VIDEO_RESOLUTION_DEFAULT_HEIGHT if not self.font_size : self.font_size = VIDEO_FONT_SIZE def get_random_mp4_files(self,video_dir: str, num_files: int) -> List[str]: """ 递归 获取目录下所有的mp4和mov文件清单 Attributes: video_dir : 顶级目录 num_files : 最大文件数 """ all_mp4s = [] for root, dirs, files in os.walk(video_dir): for file in files: if file.lower().endswith((".mp4", ".mov")): all_mp4s.append(os.path.join(root, file)) return random.sample(all_mp4s, min(num_files, len(all_mp4s))) def get_video_files_and_segments(self,video_dir: str, num_segs=10) -> List[str]: """ 返回一个列表，列表长度 == num_segs 当视频文件数量 > num_segs，取前 num_segs 个文件各一个 segment， 当文件数量 <= num_segs，随机选文件，文件可以重复，凑够 num_segs。 """ all_mp4s = self.get_random_mp4_files(video_dir,num_segs) if len(all_mp4s) >= num_segs: # 直接取前num_segs个文件 return all_mp4s[:num_segs] else: # 不够10个，随机挑选文件（允许重复），直到凑够num_segs chosen = [] for _ in range(num_segs): chosen.append(random.choice(all_mp4s)) return chosen def get_random_text_line(self,txt_dir: str) -> str: """ 递归 获取目录下所有的txt文件,并从中随机地选取一行 Attributes: txt_dir : 顶级目录 """ lines = [] for root, dirs, files in os.walk(txt_dir): for file in files: if file.lower().endswith(".txt"): with open(os.path.join(root, file), 'r', encoding='utf-8') as f: lines.extend([line.strip() for line in f if line.strip()]) return random.choice(lines) if lines else "" def get_all_audio_files(self,audio_dir: str) -> List[str]: """ 递归 获取目录下所有的mp3 和 wav 文件清单 Attributes: audio_dir : 顶级目录 """ audio_files = [] for root, dirs, files in os.walk(audio_dir): for file in files: if file.lower().endswith((".mp3", ".wav")): audio_files.append(os.path.join(root, file)) return audio_files def get_random_speed(self) -> float: """ 获取一个随机的播放速度，实现视频片段的慢放和快放 """ roll = random.random() if roll < 0.15: return 0.5 elif roll < 0.30: return 1.2 else: return 1.0 def extract_segment_cmd(self,input_path: str, start: float, duration: float, speed: float, resolution: Tuple[int, int], flip: bool = False, zoom: bool = False) -> List[str]: """ 调用ffmpeg, 获取一个视频片段,输出到 pipe:1 。 使用 ts 格式，而不是 kmv 格式 。 """ filters = [] # 缩放 filters.append(f"scale={resolution[0]}:{resolution[1]}") # 镜头推进：zoompan 滤镜（保持帧率和缩放同步） if zoom: # 使用 zoompan 实现从中心推进的效果，zoom=1.0→1.2 线性过渡 # 注意：需设置帧率和持续帧数（duration × fps），否则默认1fps zoom_filter = ( f"zoompan=" f"z='zoom+0.002':" f"x='iw/2-(iw/zoom/2)':" f"y='ih/2-(ih/zoom/2)':" f"d=1:fps=25:" f"s={resolution[0]}x{resolution[1]}" ) filters.append(zoom_filter) # 变速 if speed != 1.0: filters.append(f"setpts={1/speed}*PTS") # 水平翻转 if flip: filters.append("hflip") vf = ",".join(filters) return [ "ffmpeg", "-loglevel", "error", "-ss", str(start), "-t", str(duration), "-i", input_path, "-an", "-vf", vf, #"-f", "matroska", "-codec:v", "mpeg1video", "-q:v", "1", "pipe:1" "-f", "mpegts", "-c:v", "libx264", "-preset", "veryfast", "-crf", "18", "pipe:1" ] def process_video_segment(self,input_path: str, duration_range: Tuple[float, float], resolution: Tuple[int, int]) -> bytes: """ 给每一个视频片段添加一些随机的动态效果。 """ try: duration = random.uniform(*duration_range) speed = self.get_random_speed() start = random.uniform(0, max(1.0, self.get_video_duration(input_path) - duration)) flip = random.random() > 0.5 zoom = random.random() > 0.5 cmd = self.extract_segment_cmd(input_path, start, duration, speed, resolution, flip=flip, zoom=zoom) return subprocess.check_output(cmd) except subprocess.CalledProcessError as e: self.logger.error(f"提取视频片段失败：{input_path}，错误：{e}") return b"" def get_video_duration(self,path: str) -> float: """ 调用ffprobe 获取视频的长度 秒 """ cmd = [ "ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=noprint_wrappers=1:nokey=1", path ] output = subprocess.check_output(cmd).decode().strip() return float(output) def merge_segments_to_ts(self,segment_datas: List[bytes]) -> bytes: return b"".join(segment_datas) def wrap_text_cn_advanced(self,text, max_chars=20): # 常用中文标点符号，含中英文括号引号等 punctuation = r'[，。！？；：“”‘’（）()、,.!?;:"\'\[\]]' # 先判断空格 if ' ' in text: words = text.split() lines = [] current_line = "" for w in words: if len(current_line) + len(w) + 1 <= max_chars: current_line += (w + " ") else: lines.append(current_line.strip()) current_line = w + " " lines.append(current_line.strip()) return "\n".join(lines) else: # 没空格，先用正则把文本按标点符号切成小段，标点本身保留 parts = re.split(f'({punctuation})', text) # parts 里奇数位是标点，偶数位是文本 segments = [] buffer = "" for i, part in enumerate(parts): if part == "": continue if re.match(punctuation, part): buffer += part # 标点加到buffer末尾 segments.append(buffer) buffer = "" else: buffer += part if buffer: segments.append(buffer) # 现在segments里每个元素是一个小段（带标点） lines = [] current_line = "" for seg in segments: if len(current_line) + len(seg) <= max_chars: current_line += seg else: if current_line: lines.append(current_line) # 如果单个segment比max_chars还长，则强制切断 while len(seg) > max_chars: lines.append(seg[:max_chars]) seg = seg[max_chars:] current_line = seg if current_line: lines.append(current_line) return "\n".join(lines) def compose_final_video( self, video_dir: str, text_line: str, audio_dir: str, logo_path: str, output_path: str, num_segments: int, segment_duration: Tuple[float, float], max_workers: int, resolution: Tuple[int, int], font_path: str, font_size: int, text_color: str ): """ 生成一个拼接的视频。 支持 添加左上角logo 支持 添加居中的文字 """ #selected_medias = get_random_mp4_files(video_dir, num_segments) selected_medias = self.get_video_files_and_segments(video_dir,num_segments) audio_files = self.get_all_audio_files(audio_dir) audio_file = random.choice(audio_files) if audio_files else None with ThreadPoolExecutor(max_workers=max_workers) as executor: futures = [executor.submit(self.process_video_segment, str(v), segment_duration, resolution) for v in selected_medias] segments = [f.result() for f in futures if f.result()] ts_data = self.merge_segments_to_ts(segments) self.cleanup_old_temp_files() with tempfile.NamedTemporaryFile(delete=False, suffix=".ts") as temp_ts: temp_ts.write(ts_data) temp_ts.flush() #filters = [f"movie={logo_path} [logo]; [0:v][logo] overlay=30:30:format=auto [v1]; [v1] drawtext=fontfile={font_path}:text='{text_line}':fontsize={font_size}:fontcolor={text_color}:x=40:y=300 [vout]"] filters = [ f"movie={logo_path}, scale=iw*0.3:-1, format=rgba, colorchannelmixer=aa=0.8 [logo];" f"[0:v][logo] overlay=100:100:format=auto [v1];" f"[v1] drawtext=fontfile={font_path}:text='{text_line}':fontsize={font_size}:fontcolor={text_color}:" f"box=1:boxcolor=black@0.5:boxborderw=5:x=(w-text_w)/2:y=300 [vout]" ] cmd = ["ffmpeg", "-loglevel", "error", "-y", "-f", "mpegts", "-i", temp_ts.name] if audio_file: required_duration = num_segments * segment_duration[1] def get_valid_audio_offset(audio_path, max_attempts=5): sound = AudioSegment.from_file(audio_path) total_duration = len(sound) / 1000 # in seconds for _ in range(max_attempts): offset = random.uniform(0, total_duration - required_duration) if (total_duration - offset) >= required_duration: return round(offset, 2) return 0.0 # fallback to beginning cmd += ["-ss", str(get_valid_audio_offset(audio_file)), "-i", audio_file, "-filter_complex", "; ".join(filters), "-map", "[vout]", "-map", "1:a"] else: cmd += ["-filter_complex", "; ".join(filters), "-map", "[vout]"] cmd += ["-shortest", "-c:v", "libx264", "-crf", "18", "-preset", "veryfast", "-c:a", "aac", output_path] # CPU: libx264 # Nivida GPU : h264_nvenc 或 hevc_nvenc # AMD GPU: h264_amf 或 hevc_amf subprocess.run(cmd, check=True) time.sleep(5) # 等待操作系统释放文件句柄 try: os.unlink(temp_ts.name) except PermissionError: print(f"Warning: Could not delete temp file {temp_ts.name} because it is in use.") def extract_cover_frame(self,input_path, output_path, frame_number=10): """ 从mp4视频中获取 一帧保存成png,作为视频封面。 """ cmd = [ "ffmpeg", "-y", # 自动覆盖输出文件 "-loglevel", "error", "-i", input_path, "-vf", f"select=eq(n\\,{frame_number})", "-vframes", "1", "-q:v", "1", output_path ] subprocess.run(cmd, check=True) def generate(self): """ 被外部调用。 生成一个短视频mp4; 生成一个png封面; 生成一个视频的配置json文件; 三个文件都 保存到 video_final / user_id / 下面 """ # from 2025-07-19 08:00:00 to 2025-07-19 08:00 try: tmp_datetime = datetime.strptime(self.scheduled_time, "%Y-%m-%d %H:%M:%S") scheduled_time_str = tmp_datetime.strftime("%Y%m%d%H%M%S") scheduled_time_pub = tmp_datetime.strftime("%Y-%m-%d %H:%M") self.logger.info(f"VideoGeneratorMashup 正在生成 media：{self.user_id},{self.scheduled_time} .mp4") os.makedirs(os.path.join(FINAL_VIDEO_DIR,self.user_id),exist_ok=True) output_mp4_path =f"{FINAL_VIDEO_DIR}/{self.user_id}/{scheduled_time_str}.mp4" output_png_path =f"{FINAL_VIDEO_DIR}/{self.user_id}/{scheduled_time_str}.png" output_json_path = f"{FINAL_VIDEO_DIR}/{self.user_id}/{scheduled_time_str}.json" # 针对不同的平台，如果配置了相同的发布时间，默认是使用同一个视频。 # 如果某个用户在不同的平台上，需要展示不同的内容，要 人为 地 设置最终发布时间是不同的时间。 if os.path.exists(output_json_path): return True VIDEO_DIR_SUB = VIDEO_DIR + "/"+ self.video_type text_line = self.wrap_text_cn_advanced(self.get_random_text_line(VIDEO_DIR_SUB), max_chars=45) # 换行符用 \n 传给ffmpeg需要写成 \\n self.compose_final_video( video_dir=VIDEO_DIR_SUB, text_line=text_line, audio_dir=VIDEO_DIR_SUB, logo_path=VIDEO_LOGO_PATH, output_path=output_mp4_path, num_segments=VIDEO_SEGMENT_COUNT, segment_duration=VIDEO_SEGMENT_DURATION, max_workers=4, resolution=(self.video_width,self.video_height), font_path=VIDEO_FONT_PATH, font_size=self.font_size, text_color=VIDEO_FONT_COLOR ) self.extract_cover_frame(output_mp4_path,output_png_path) video_json ={ "user": self.user_id, "title": "", "description": text_line, "path_video": output_mp4_path, "tags": ["感恩相遇","再来一个","太好看了"], "publish_datetime": scheduled_time_pub, "location": "", "category": "", "path_cover": output_png_path, "proxy": self.proxy_full } # 保存 with open(output_json_path, "w", encoding="utf-8") as f: json.dump(video_json, f, ensure_ascii=False, indent=2) return True except Exception as e: self.logger.error(f"VideoGeneratorMashup error: {e}") return False