# -*- 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
|
|
from auto_media_publisher.utils.logger import get_logger
|
from auto_media_publisher.generators.media_generator_base import MediaGenerator
|
|
class VideoGeneratorConcat(MediaGenerator):
|
"""
|
视频生成器类，单纯的拼接，不做任何解码处理：如zoom，flip，add text/logo etc
|
不写磁盘，不做编码处理，速度快！！！
|
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")
|
|
|
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文件，以节省磁盘空间，提高速度。
|
|
"""
|
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"
|
]
|
|
|
def process_video_segment(self,input_path: str, duration_range: Tuple[float, float], resolution: Tuple[int, int]) -> bytes:
|
"""
|
给每一个视频片段添加一些随机的动态效果。
|
|
"""
|
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)
|
|
|
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)
|
|
|
import re
|
|
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]
|
|
mkv_data = self.merge_segments_to_ts(segments)
|
|
with tempfile.NamedTemporaryFile(delete=False, suffix=".mkv") as temp_mkv:
|
temp_mkv.write(mkv_data)
|
temp_mkv.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", "-i", temp_mkv.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(2) # 等待操作系统释放文件句柄
|
try:
|
os.unlink(temp_mkv.name)
|
except PermissionError:
|
print(f"Warning: Could not delete temp file {temp_mkv.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文件;
|
三个文件都 保存到 media_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"正在生成 media：{self.user_id},{self.scheduled_time} .mp4")
|
|
output_mp4_path =f"{VIDEO_DIR_VIDEO_DST}/{self.user_id}/{scheduled_time_str}.mp4"
|
output_png_path =f"{VIDEO_DIR_VIDEO_DST}/{self.user_id}/{scheduled_time_str}.png"
|
output_json_path = f"{VIDEO_DIR_VIDEO_DST}/{self.user_id}/{scheduled_time_str}.json"
|
|
text_line = self.wrap_text_cn_advanced(self.get_random_text_line(VIDEO_DIR), max_chars=450) # 换行符用 \n 传给ffmpeg需要写成 \\n
|
|
|
VIDEO_DIR_SUB = VIDEO_DIR + "/"+ self.video_type
|
|
self.compose_final_video(
|
video_dir=VIDEO_DIR_SUB,
|
text_line="",
|
audio_dir=VIDEO_DIR,
|
logo_path=VIDEO_LOGO_PATH,
|
output_path=output_mp4_path,
|
num_segments=VIDEO_SEGMENT_COUNT,
|
segment_duration=VIDEO_SEGMENT_DURATION,
|
max_workers=4,
|
resolution=VIDEO_RESOLUTION,
|
font_path=VIDEO_FONT_PATH,
|
font_size=VIDEO_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_media": output_mp4_path,
|
"tags": ["感恩相遇","再来一个","太好看了"],
|
"scheduled_time": 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" error: {e}")
|
return False
|
|
