cat-shark
2338 lines
105 KiB
Python
2338 lines
105 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
字幕封面生成器 - 简化版本（不依赖spacy和rembg）
|
||
Subtitle & Cover Generator - Simplified Version
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import json
|
||
import argparse
|
||
import subprocess
|
||
from typing import Dict, List, Any, Optional
|
||
from pathlib import Path
|
||
|
||
# ⚠️ 修复模块路径：确保 Python 能找到 modules 和其他本地模块
|
||
# 将脚本所在目录（python 目录）添加到 sys.path
|
||
script_dir = os.path.dirname(os.path.abspath(__file__))
|
||
if script_dir not in sys.path:
|
||
sys.path.insert(0, script_dir)
|
||
print(f"[PATH] Added to sys.path: {script_dir}", file=sys.stderr)
|
||
|
||
try:
|
||
from modules.font_manager import get_font_manager
|
||
FONT_MANAGER_AVAILABLE = True
|
||
print("[PATH] font_manager loaded successfully", file=sys.stderr)
|
||
except ImportError as e:
|
||
FONT_MANAGER_AVAILABLE = False
|
||
print(f"[PATH] Warning: font_manager module not available: {e}", file=sys.stderr)
|
||
|
||
# 导入封面模板
|
||
try:
|
||
from modules.cover_templates import get_template, get_all_templates
|
||
COVER_TEMPLATES_AVAILABLE = True
|
||
except ImportError:
|
||
COVER_TEMPLATES_AVAILABLE = False
|
||
print("Warning: cover_templates module not available", file=sys.stderr)
|
||
|
||
# 导入基础库
|
||
try:
|
||
import cv2
|
||
import numpy as np
|
||
from PIL import Image, ImageDraw, ImageFont, ImageFilter, ImageEnhance, ImageFont, ImageOps
|
||
except ImportError as e:
|
||
print(f"Error: Missing required library - {e}", file=sys.stderr)
|
||
sys.exit(1)
|
||
|
||
# 尝试导入 RMBG（人像抠图）
|
||
try:
|
||
from rembg import remove as rembg_remove
|
||
REMBG_AVAILABLE = True
|
||
print("RMBG library loaded successfully", file=sys.stderr)
|
||
except ImportError:
|
||
REMBG_AVAILABLE = False
|
||
print("Warning: RMBG library not available, portrait extraction will be disabled", file=sys.stderr)
|
||
|
||
# 尝试导入 Jieba（中文NLP - 轻量级）
|
||
try:
|
||
import jieba
|
||
import jieba.analyse
|
||
NLP_AVAILABLE = True
|
||
print("Jieba NLP library loaded successfully", file=sys.stderr)
|
||
except ImportError:
|
||
NLP_AVAILABLE = False
|
||
print("Warning: Jieba library not available, NLP features will be disabled", file=sys.stderr)
|
||
|
||
# 尝试导入 Transformers（深度学习情感分析）
|
||
try:
|
||
from transformers import pipeline
|
||
import torch
|
||
TRANSFORMERS_AVAILABLE = True
|
||
print("Transformers library loaded successfully", file=sys.stderr)
|
||
# 检查是否支持FP16
|
||
FP16_AVAILABLE = torch.cuda.is_available()
|
||
if FP16_AVAILABLE:
|
||
print("CUDA available, FP16 optimization enabled", file=sys.stderr)
|
||
else:
|
||
print("CUDA not available, using CPU mode", file=sys.stderr)
|
||
except ImportError:
|
||
TRANSFORMERS_AVAILABLE = False
|
||
FP16_AVAILABLE = False
|
||
print("Warning: Transformers library not available, advanced sentiment analysis will be disabled", file=sys.stderr)
|
||
|
||
# 尝试导入 FUNASR（语音识别）
|
||
try:
|
||
from funasr import AutoModel
|
||
FUNASR_AVAILABLE = True
|
||
print("FUNASR library loaded successfully", file=sys.stderr)
|
||
except ImportError:
|
||
FUNASR_AVAILABLE = False
|
||
print("Warning: FUNASR library not available, speech recognition will be disabled", file=sys.stderr)
|
||
|
||
|
||
class SubtitleCoverGenerator:
|
||
"""字幕封面生成器主类（简化版）"""
|
||
|
||
def __init__(self, config: Dict[str, Any]):
|
||
"""
|
||
初始化生成器
|
||
|
||
Args:
|
||
config: 配置字典，包含视频路径、文案、样式等
|
||
"""
|
||
self.config = config
|
||
self.video_path = config.get('videoPath')
|
||
self.script_text = config.get('scriptText', '')
|
||
self.subtitle_style = config.get('subtitleStyle', 'default')
|
||
self.cover_style = config.get('coverStyle', 'default')
|
||
self.output_dir = config.get('outputDir', './output')
|
||
|
||
# 新增：关键词分组和效果配置
|
||
self.keyword_groups = config.get('keywordGroups', []) # 关键词分组列表
|
||
self.keyword_markers = config.get('keywordMarkers', []) # 关键词标记
|
||
print(f"[SubtitleGenerator] Loaded {len(self.keyword_groups)} keyword groups and {len(self.keyword_markers)} keyword markers", file=sys.stderr)
|
||
|
||
# ✅ 详细日志：显示关键词分组信息
|
||
if self.keyword_groups:
|
||
print(f"[SubtitleGenerator] === Keyword Groups Details ===", file=sys.stderr)
|
||
for idx, group in enumerate(self.keyword_groups):
|
||
style_override = group.get('styleOverride', {})
|
||
print(f"[SubtitleGenerator] Group {idx}: id={group.get('id')}", file=sys.stderr)
|
||
print(f" name={group.get('name')}", file=sys.stderr)
|
||
print(f" effectId={group.get('effectId')}", file=sys.stderr)
|
||
print(f" styleOverride={style_override}", file=sys.stderr)
|
||
print(f" fontColor={style_override.get('fontColor')}", file=sys.stderr)
|
||
print(f" outlineColor={style_override.get('outlineColor')}", file=sys.stderr)
|
||
print(f" fontSize={style_override.get('fontSize')}", file=sys.stderr)
|
||
else:
|
||
print("[SubtitleGenerator] WARNING: No keyword groups received!", file=sys.stderr)
|
||
|
||
# ✅ 详细日志：显示关键词标记信息
|
||
if self.keyword_markers:
|
||
for marker in self.keyword_markers:
|
||
print(f"[SubtitleGenerator] Keyword Marker: text={marker.get('text')}, groupId={marker.get('groupId')}", file=sys.stderr)
|
||
|
||
# 新增：字体配置（从subtitleStyle中提取）
|
||
subtitle_style = config.get('subtitleStyle', {})
|
||
self.subtitle_font_name = subtitle_style.get('fontName', 'NotoSerifCJK-VF') # 字体名称
|
||
self.subtitle_font_size = subtitle_style.get('fontSize', 24) # 字体大小
|
||
self.subtitle_font_color = subtitle_style.get('fontColor', '#FFFFFF') # 字体颜色
|
||
print(f"Subtitle font config: {self.subtitle_font_name}, {self.subtitle_font_size}px, {self.subtitle_font_color}", file=sys.stderr)
|
||
|
||
# 获取用户配置的文件存储路径，如果没有则使用默认输出目录
|
||
self.hub_root = config.get('hubRoot', None)
|
||
if self.hub_root:
|
||
# 确保hub_root路径存在
|
||
os.makedirs(self.hub_root, exist_ok=True)
|
||
# 使用hub_root作为输出目录
|
||
self.output_dir = self.hub_root
|
||
else:
|
||
# 没有hub_root配置，使用提供的outputDir或默认值
|
||
self.output_dir = config.get('outputDir', './output')
|
||
|
||
# 确保输出目录存在
|
||
os.makedirs(self.output_dir, exist_ok=True)
|
||
|
||
# 初始化 ziti 字体目录路径（用于 FFmpeg fontconfig）
|
||
self.ziti_dir = self._find_ziti_directory()
|
||
print(f"[SubtitleGenerator] Ziti font directory: {self.ziti_dir}", file=sys.stderr)
|
||
|
||
self.use_funasr = config.get('useFunasr', False) # 是否使用FUNASR
|
||
self.funasr_result = config.get('funasrResult', None) # FUNASR识别结果（可选）
|
||
self.cover_only = config.get('coverOnly', False) # 是否仅生成封面
|
||
self.use_extracted_frame = config.get('useExtractedFrame', False) # 是否使用提取的帧
|
||
self.extracted_frame_path = config.get('extractedFramePath', None) # 提取的帧路径
|
||
|
||
# 封面参数配置
|
||
self.cover_text = config.get('coverText', '') # 封面文字
|
||
self.cover_text_position = config.get('coverTextPosition', 'bottom') # 文字位置
|
||
self.cover_text_color = config.get('coverTextColor', '#FFFFFF') # 文字颜色
|
||
self.cover_text_size = config.get('coverTextSize', 50) # 文字大小
|
||
self.outline_color = config.get('outlineColor', '#FFD700') # 描边颜色
|
||
self.outline_width = config.get('outlineWidth', 3) # 描边宽度
|
||
self.blur_radius = config.get('blurRadius', 15) # 背景模糊度
|
||
self.zoom_level = config.get('zoomLevel', 1.0) # 封面放大倍数
|
||
|
||
# 自定义模板配置
|
||
self.custom_template = config.get('customTemplate', False) # 是否使用自定义模板
|
||
self.template = config.get('template', None) # 自定义模板对象
|
||
|
||
# 初始化情感分析模型（延迟加载）
|
||
self.sentiment_analyzer = None
|
||
if TRANSFORMERS_AVAILABLE:
|
||
try:
|
||
print("Loading sentiment analysis model...", file=sys.stderr)
|
||
# 使用中文情感分析模型，支持FP16优化
|
||
device = 0 if FP16_AVAILABLE else -1 # 0 = GPU, -1 = CPU
|
||
torch_dtype = torch.float16 if FP16_AVAILABLE else torch.float32
|
||
self.sentiment_analyzer = pipeline(
|
||
'sentiment-analysis',
|
||
model='uer/roberta-base-finetuned-jd-binary-chinese', # 中文情感分析模型
|
||
device=device,
|
||
torch_dtype=torch_dtype
|
||
)
|
||
print(f"Sentiment analysis model loaded (device={device}, dtype={torch_dtype})", file=sys.stderr)
|
||
except Exception as e:
|
||
print(f"Failed to load sentiment analyzer: {e}", file=sys.stderr)
|
||
self.sentiment_analyzer = None
|
||
|
||
# 初始化FUNASR模型（延迟加载）
|
||
self.funasr_model = None
|
||
if FUNASR_AVAILABLE and self.use_funasr:
|
||
try:
|
||
print("Loading FUNASR model...", file=sys.stderr)
|
||
self.funasr_model = AutoModel(
|
||
model="paraformer-zh", # 中文语音识别模型
|
||
model_revision="v2.0.4",
|
||
vad_model="fsmn-vad",
|
||
vad_model_revision="v2.0.4",
|
||
punc_model="ct-punc-c",
|
||
punc_model_revision="v2.0.4",
|
||
)
|
||
print("FUNASR model loaded successfully", file=sys.stderr)
|
||
except Exception as e:
|
||
print(f"Failed to load FUNASR model: {e}", file=sys.stderr)
|
||
self.funasr_model = None
|
||
|
||
def _find_ziti_directory(self) -> Optional[str]:
|
||
"""
|
||
查找 ziti 字体目录的绝对路径（支持开发环境和打包环境）
|
||
|
||
Returns:
|
||
ziti 目录路径，如果未找到返回 None
|
||
"""
|
||
# 可能的 ziti 目录路径（按优先级排列）
|
||
possible_paths = []
|
||
|
||
# 1. 从 APP_ROOT 环境变量查找（打包环境）
|
||
app_root = os.environ.get('APP_ROOT', None)
|
||
if app_root:
|
||
# 打包后的结构: resources/app.asar.unpacked/ziti
|
||
possible_paths.append(os.path.join(app_root, 'app.asar.unpacked', 'ziti'))
|
||
# 备选结构: resources/extra/common/fonts/ziti
|
||
possible_paths.append(os.path.join(app_root, 'extra', 'common', 'fonts', 'ziti'))
|
||
# 直接在 app_root 下
|
||
possible_paths.append(os.path.join(app_root, 'ziti'))
|
||
|
||
# 2. 相对于脚本位置查找（开发环境）
|
||
script_dir = os.path.dirname(os.path.abspath(__file__))
|
||
possible_paths.append(os.path.join(script_dir, '..', 'ziti')) # python/../ziti
|
||
possible_paths.append(os.path.join(script_dir, '..', '..', 'ziti')) # python/../../ziti
|
||
|
||
# 3. 当前工作目录
|
||
possible_paths.append(os.path.join(os.getcwd(), 'ziti'))
|
||
|
||
# 4. 用户主目录
|
||
possible_paths.append(os.path.expanduser('~/ziti'))
|
||
|
||
# 遍历查找存在的目录
|
||
for path in possible_paths:
|
||
normalized_path = os.path.normpath(os.path.abspath(path))
|
||
if os.path.isdir(normalized_path):
|
||
# 检查目录中是否有字体文件
|
||
try:
|
||
fonts = [f for f in os.listdir(normalized_path)
|
||
if f.endswith(('.ttf', '.otf', '.ttc'))]
|
||
if fonts:
|
||
print(f"[_find_ziti_directory] ✅ Found ziti directory with {len(fonts)} fonts: {normalized_path}", file=sys.stderr)
|
||
return normalized_path
|
||
except Exception as e:
|
||
print(f"[_find_ziti_directory] ⚠ Error scanning {normalized_path}: {e}", file=sys.stderr)
|
||
|
||
print(f"[_find_ziti_directory] ❌ ziti directory not found, searched: {possible_paths[:3]}...", file=sys.stderr)
|
||
return None
|
||
|
||
def _get_ffmpeg_fontconfig(self) -> str:
|
||
"""
|
||
生成 FFmpeg fontconfig 参数字符串
|
||
|
||
Returns:
|
||
FFmpeg fontconfig 参数，如 ":fontconfig='fontdir=/path/to/ziti:'"
|
||
"""
|
||
if self.ziti_dir and os.path.isdir(self.ziti_dir):
|
||
# 将 Windows 路径转换为正斜杠格式（FFmpeg 需要）
|
||
ziti_path = self.ziti_dir.replace('\\', '/')
|
||
fontconfig = f":fontconfig='fontdir={ziti_path}:'"
|
||
print(f"[_get_ffmpeg_fontconfig] Using ziti path: {ziti_path}", file=sys.stderr)
|
||
return fontconfig
|
||
else:
|
||
# 降级：使用当前工作目录
|
||
print(f"[_get_ffmpeg_fontconfig] ⚠ Ziti directory not available, using fallback '.'", file=sys.stderr)
|
||
return ":fontconfig='fontdir=.'"
|
||
|
||
def extract_audio_for_asr(self) -> str:
|
||
"""
|
||
从视频中提取音频为WAV格式,用于语音识别
|
||
|
||
Returns:
|
||
音频文件路径
|
||
"""
|
||
try:
|
||
audio_path = os.path.join(self.output_dir, 'extracted_audio.wav')
|
||
|
||
# 使用FFmpeg提取音频
|
||
cmd = [
|
||
'ffmpeg', '-y',
|
||
'-i', self.video_path,
|
||
'-vn', # 不要视频
|
||
'-acodec', 'pcm_s16le', # WAV格式
|
||
'-ar', '16000', # 采样率16kHz (FUNASR推荐)
|
||
'-ac', '1', # 单声道
|
||
audio_path
|
||
]
|
||
|
||
print(f"Extracting audio from video: {' '.join(cmd)}", file=sys.stderr)
|
||
result = subprocess.run(cmd, capture_output=True, text=True, encoding='utf-8', errors='ignore', timeout=300)
|
||
|
||
if result.returncode == 0 and os.path.exists(audio_path):
|
||
print(f"Audio extracted successfully: {audio_path}", file=sys.stderr)
|
||
return audio_path
|
||
else:
|
||
raise Exception(f"FFmpeg failed: {result.stderr}")
|
||
|
||
except subprocess.TimeoutExpired:
|
||
raise Exception("Audio extraction timeout (exceeded 5 minutes)")
|
||
except Exception as e:
|
||
raise Exception(f"Error extracting audio: {str(e)}")
|
||
|
||
def transcribe_audio_with_funasr(self, audio_path: str) -> Dict:
|
||
"""
|
||
使用FUNASR进行语音识别
|
||
|
||
Args:
|
||
audio_path: 音频文件路径
|
||
|
||
Returns:
|
||
识别结果字典，包含segments列表
|
||
"""
|
||
if not FUNASR_AVAILABLE or self.funasr_model is None:
|
||
raise Exception("FUNASR not available or model not loaded")
|
||
|
||
try:
|
||
print(f"Transcribing audio with FUNASR: {audio_path}", file=sys.stderr)
|
||
|
||
# 调用FUNASR模型进行识别
|
||
result = self.funasr_model.generate(
|
||
input=audio_path,
|
||
batch_size_s=300, # 批处理大小
|
||
hotword='', # 热词
|
||
)
|
||
|
||
print(f"FUNASR transcription completed", file=sys.stderr)
|
||
|
||
# 解析FUNASR结果
|
||
segments = []
|
||
if result and len(result) > 0:
|
||
for i, item in enumerate(result):
|
||
# FUNASR返回格式: {'text': '...', 'timestamp': [[start_ms, end_ms], ...]}
|
||
text = item.get('text', '')
|
||
timestamp = item.get('timestamp', [])
|
||
|
||
# 如果有时间戳，使用时间戳
|
||
if timestamp and len(timestamp) > 0:
|
||
for j, (start_ms, end_ms) in enumerate(timestamp):
|
||
# 提取对应的文本片段
|
||
words = text.split()
|
||
if j < len(words):
|
||
segment_text = words[j]
|
||
else:
|
||
segment_text = text
|
||
|
||
segments.append({
|
||
'text': segment_text,
|
||
'start': start_ms / 1000.0, # 转换为秒
|
||
'end': end_ms / 1000.0
|
||
})
|
||
else:
|
||
# 没有时间戳，按句子分割，平均分配时间
|
||
sentences = text.split('。')
|
||
for k, sentence in enumerate(sentences):
|
||
if sentence.strip():
|
||
segments.append({
|
||
'text': sentence.strip() + '。',
|
||
'start': k * 3.0,
|
||
'end': (k + 1) * 3.0
|
||
})
|
||
|
||
print(f"Generated {len(segments)} segments from FUNASR", file=sys.stderr)
|
||
|
||
return {
|
||
'success': True,
|
||
'segments': segments
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f"FUNASR transcription failed: {e}", file=sys.stderr)
|
||
return {
|
||
'success': False,
|
||
'error': str(e),
|
||
'segments': []
|
||
}
|
||
|
||
def extract_frame(self, timestamp: float = 1.0) -> tuple[np.ndarray, tuple[int, int]]:
|
||
"""
|
||
从视频中提取关键帧
|
||
|
||
Args:
|
||
timestamp: 提取帧的时间戳（秒）
|
||
|
||
Returns:
|
||
提取的帧（numpy数组）和视频尺寸（宽度, 高度）
|
||
"""
|
||
try:
|
||
cap = cv2.VideoCapture(self.video_path)
|
||
|
||
# 获取视频尺寸
|
||
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
|
||
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
|
||
video_size = (width, height)
|
||
|
||
# 设置到指定时间戳
|
||
fps = cap.get(cv2.CAP_PROP_FPS)
|
||
frame_number = int(timestamp * fps)
|
||
cap.set(cv2.CAP_PROP_POS_FRAMES, frame_number)
|
||
|
||
ret, frame = cap.read()
|
||
cap.release()
|
||
|
||
if not ret:
|
||
raise Exception("Failed to extract frame from video")
|
||
|
||
return frame, video_size
|
||
except Exception as e:
|
||
raise Exception(f"Error extracting frame: {str(e)}")
|
||
|
||
def generate_cover(self) -> str:
|
||
"""
|
||
生成视频封面
|
||
|
||
Returns:
|
||
生成的封面图片路径
|
||
"""
|
||
try:
|
||
# 如果使用提取的帧，直接使用已提取的帧
|
||
if self.use_extracted_frame and self.extracted_frame_path and os.path.exists(self.extracted_frame_path):
|
||
print(f"Using extracted frame: {self.extracted_frame_path}", file=sys.stderr)
|
||
frame = cv2.imread(self.extracted_frame_path)
|
||
self.video_width, self.video_height = frame.shape[1], frame.shape[0]
|
||
else:
|
||
# 提取关键帧和视频尺寸
|
||
frame, video_size = self.extract_frame(1.0)
|
||
self.video_width, self.video_height = video_size
|
||
|
||
# 检查是否使用自定义模板
|
||
if self.custom_template and self.template:
|
||
print("Using custom template for cover generation", file=sys.stderr)
|
||
cover_path = self._generate_custom_template_cover(frame)
|
||
return cover_path
|
||
|
||
if self.cover_style == 'default':
|
||
# 默认封面：直接使用提取的帧，保持原始尺寸
|
||
cover_path = os.path.join(self.output_dir, 'cover.png')
|
||
cv2.imwrite(cover_path, frame)
|
||
|
||
# 应用缩放和文字叠加
|
||
image = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
|
||
image = self._apply_zoom(image)
|
||
image = self._add_text_overlay(image)
|
||
image.save(cover_path, quality=95)
|
||
|
||
elif self.cover_style == 'blur-bg':
|
||
# 背景虚化效果（简化版）
|
||
cover_path = self._generate_blur_cover(frame)
|
||
|
||
elif self.cover_style == 'outline':
|
||
# 人物描边效果（简化版）
|
||
cover_path = self._generate_outline_cover(frame)
|
||
|
||
elif self.cover_style == 'gradient':
|
||
# 渐变背景效果
|
||
cover_path = self._generate_gradient_cover(frame)
|
||
|
||
elif self.cover_style == 'split':
|
||
# 斜切设计效果
|
||
cover_path = self._generate_split_cover(frame)
|
||
else:
|
||
cover_path = os.path.join(self.output_dir, 'cover.png')
|
||
cv2.imwrite(cover_path, frame)
|
||
|
||
return cover_path
|
||
except Exception as e:
|
||
raise Exception(f"Error generating cover: {str(e)}")
|
||
|
||
def _generate_blur_cover(self, frame: np.ndarray) -> str:
|
||
"""生成背景虚化封面（人物突出，背景虚化）"""
|
||
try:
|
||
print("Starting blur cover generation...", file=sys.stderr)
|
||
# 转换为PIL
|
||
image = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
|
||
width, height = image.size
|
||
|
||
# 确保图像格式正确
|
||
print(f"Image size: {width}x{height}, mode: {image.mode}", file=sys.stderr)
|
||
|
||
# 使用视频的原始尺寸
|
||
video_width = getattr(self, 'video_width', width)
|
||
video_height = getattr(self, 'video_height', height)
|
||
|
||
if REMBG_AVAILABLE:
|
||
try:
|
||
print("Creating background blur effect with portrait segmentation...", file=sys.stderr)
|
||
# 使用RMBG提取人像
|
||
portrait = rmbg_remove(image)
|
||
|
||
print(f"Portrait extracted, mode: {portrait.mode if hasattr(portrait, 'mode') else 'unknown'}", file=sys.stderr)
|
||
|
||
# 创建强力背景虚化
|
||
print("Creating blurred background...", file=sys.stderr)
|
||
blurred_bg = image.filter(ImageFilter.GaussianBlur(radius=self.blur_radius))
|
||
|
||
# 创建结果：虚化背景 + 清晰人物
|
||
print(f"Creating result image with size: {video_width}x{video_height}", file=sys.stderr)
|
||
result = Image.new('RGB', (video_width, video_height))
|
||
result.paste(blurred_bg, (0, 0))
|
||
|
||
# 将清晰的人物合成到虚化背景上
|
||
if portrait.mode == 'RGBA':
|
||
# 计算居中位置
|
||
x_offset = (video_width - width) // 2
|
||
y_offset = (video_height - height) // 2
|
||
print(f"Pasting portrait at offset: ({x_offset}, {y_offset})", file=sys.stderr)
|
||
result.paste(portrait, (x_offset, y_offset), portrait)
|
||
print("Portrait background blur effect successful", file=sys.stderr)
|
||
else:
|
||
# 如果抠图失败，使用降低透明度的原始图像
|
||
print(f"Warning: Portrait mode is {portrait.mode}, expected RGBA", file=sys.stderr)
|
||
overlay = Image.new('RGBA', (width, height), (0, 0, 0, 0))
|
||
overlay.paste(image, (0, 0))
|
||
result.paste(overlay, (0, 0), overlay)
|
||
|
||
# 应用缩放和文字叠加
|
||
print("Applying zoom and text overlay...", file=sys.stderr)
|
||
result = self._apply_zoom(result)
|
||
result = self._add_text_overlay(result)
|
||
except Exception as e:
|
||
print(f"RMBG failed with error: {str(e)}", file=sys.stderr)
|
||
import traceback
|
||
traceback.print_exc(file=sys.stderr)
|
||
print("Falling back to software blur method", file=sys.stderr)
|
||
# 软件方法：创建人物区域mask
|
||
# 这里简化实现：假设人物在中心区域
|
||
result = image.copy()
|
||
# 降低背景亮度并模糊
|
||
bg_overlay = Image.new('RGB', (video_width, video_height), (0, 0, 0))
|
||
blurred = image.filter(ImageFilter.GaussianBlur(radius=self.blur_radius))
|
||
bg_overlay.paste(blurred, (0, 0))
|
||
# 中心人物区域保持清晰
|
||
center_mask = Image.new('L', (width, height), 0)
|
||
draw = ImageDraw.Draw(center_mask)
|
||
draw.ellipse([width//4, height//4, 3*width//4, 3*height//4], fill=255)
|
||
result = Image.composite(image, bg_overlay, center_mask)
|
||
|
||
# 应用缩放和文字叠加
|
||
result = self._apply_zoom(result)
|
||
result = self._add_text_overlay(result)
|
||
else:
|
||
print("RMBG not available, using simple blur effect", file=sys.stderr)
|
||
# 简单方法：整体轻微模糊 + 中心清晰区域
|
||
blurred = image.filter(ImageFilter.GaussianBlur(radius=self.blur_radius))
|
||
center_region = image.crop((width//4, height//4, 3*width//4, 3*width//4))
|
||
blurred.paste(center_region, (width//4, height//4))
|
||
result = blurred.resize((video_width, video_height))
|
||
|
||
# 应用缩放和文字叠加
|
||
result = self._apply_zoom(result)
|
||
result = self._add_text_overlay(result)
|
||
|
||
# 保存
|
||
print("Saving cover image...", file=sys.stderr)
|
||
cover_path = os.path.join(self.output_dir, 'cover_blur.png')
|
||
result.save(cover_path, quality=95)
|
||
print(f"Cover saved successfully: {cover_path}", file=sys.stderr)
|
||
return cover_path
|
||
except Exception as e:
|
||
print(f"Error in _generate_blur_cover: {str(e)}", file=sys.stderr)
|
||
import traceback
|
||
traceback.print_exc(file=sys.stderr)
|
||
raise
|
||
|
||
def _generate_outline_cover(self, frame: np.ndarray) -> str:
|
||
"""生成人物描边封面（使用RMBG人像抠图 + 描边效果）"""
|
||
# 转换为PIL
|
||
image = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
|
||
width, height = image.size
|
||
|
||
if REMBG_AVAILABLE:
|
||
try:
|
||
# 使用更精确的人像分割模型
|
||
print("Extracting portrait using RMBG for outline effect...", file=sys.stderr)
|
||
|
||
# 尝试使用专门的人像分割模型
|
||
try:
|
||
from rembg import new_session
|
||
session = new_session("u2net_human_seg")
|
||
portrait = rmbg_remove(image, session=session, post_process_mask=True)
|
||
print("Used specialized human segmentation model", file=sys.stderr)
|
||
except:
|
||
# 降级到默认模型
|
||
portrait = rmbg_remove(image)
|
||
print("Used default rembg model", file=sys.stderr)
|
||
|
||
# 获取alpha通道作为mask
|
||
if portrait.mode == 'RGBA':
|
||
alpha = portrait.split()[3]
|
||
alpha_np = np.array(alpha)
|
||
|
||
# 改进的描边方法：使用固定迭代次数和可变 kernel 大小
|
||
# 这样描边宽度可控，但紧贴轮廓
|
||
kernel_size = max(3, self.outline_width * 2 + 1) # 确保是奇数
|
||
kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (kernel_size, kernel_size))
|
||
|
||
# 只膨胀1次，避免离轮廓太远
|
||
alpha_dilated = cv2.dilate(alpha_np, kernel, iterations=1)
|
||
|
||
# 描边 = 膨胀后的轮廓 - 原始轮廓（这样得到紧贴的描边）
|
||
outline_mask = cv2.subtract(alpha_dilated, alpha_np)
|
||
|
||
# 转换描边颜色
|
||
outline_rgb = self._hex_to_rgb(self.outline_color)
|
||
outline_layer = Image.new('RGBA', (width, height), (*outline_rgb, 0))
|
||
|
||
# 应用描边
|
||
outline_pil = Image.fromarray(outline_mask)
|
||
outline_colored = Image.new('RGBA', (width, height), (*outline_rgb, 255))
|
||
outline_colored.putalpha(outline_pil)
|
||
outline_layer = Image.alpha_composite(outline_layer, outline_colored)
|
||
|
||
# 创建结果：背景 + 描边 + 人像
|
||
result = Image.new('RGB', (width, height), (0, 0, 0))
|
||
|
||
# 先放原图作为背景
|
||
result.paste(image, (0, 0))
|
||
|
||
# 应用缩放到人像和描边
|
||
if self.zoom_level != 1.0:
|
||
# 只缩放人像和描边，保持背景不变
|
||
portrait_zoomed = self._apply_zoom(portrait)
|
||
outline_zoomed = self._apply_zoom(outline_layer)
|
||
|
||
# 计算居中位置
|
||
p_width, p_height = portrait_zoomed.size
|
||
if portrait_zoomed.mode == 'RGBA':
|
||
result.paste(portrait_zoomed,
|
||
((width - p_width) // 2, (height - p_height) // 2),
|
||
portrait_zoomed)
|
||
|
||
result.paste(outline_zoomed,
|
||
((width - p_width) // 2, (height - p_height) // 2),
|
||
outline_zoomed)
|
||
else:
|
||
result.paste(outline_layer, (0, 0), outline_layer)
|
||
result.paste(portrait, (0, 0), portrait)
|
||
|
||
print("Enhanced portrait outline extraction successful", file=sys.stderr)
|
||
else:
|
||
raise Exception("Portrait extraction did not return RGBA image")
|
||
|
||
except Exception as e:
|
||
print(f"Enhanced RMBG portrait outline failed: {e}, using fallback method", file=sys.stderr)
|
||
# 降级方案：简单的边缘检测
|
||
image_np = np.array(image.convert('L'))
|
||
edges = cv2.Canny(image_np, 50, 150)
|
||
outline_rgb = self._hex_to_rgb(self.outline_color)
|
||
edges_colored = np.zeros((*edges.shape, 3), dtype=np.uint8)
|
||
edges_colored[edges > 0] = outline_rgb
|
||
edges_pil = Image.fromarray(edges_colored)
|
||
|
||
result = Image.blend(image, edges_pil, 0.3)
|
||
result = self._apply_zoom(result)
|
||
else:
|
||
# 降级方案：简单的边缘检测
|
||
image_np = np.array(image.convert('L'))
|
||
edges = cv2.Canny(image_np, 50, 150)
|
||
outline_rgb = self._hex_to_rgb(self.outline_color)
|
||
edges_colored = np.zeros((*edges.shape, 3), dtype=np.uint8)
|
||
edges_colored[edges > 0] = outline_rgb
|
||
edges_pil = Image.fromarray(edges_colored)
|
||
|
||
result = Image.blend(image, edges_pil, 0.3)
|
||
result = self._apply_zoom(result)
|
||
|
||
# 应用文字叠加
|
||
result = self._add_text_overlay(result)
|
||
|
||
# 保存
|
||
cover_path = os.path.join(self.output_dir, 'cover_outline.png')
|
||
result.save(cover_path, quality=95)
|
||
|
||
return cover_path
|
||
|
||
def _generate_gradient_cover(self, frame: np.ndarray) -> str:
|
||
"""生成渐变背景封面"""
|
||
# 转换为PIL
|
||
image = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
|
||
width, height = image.size
|
||
|
||
# 创建渐变背景
|
||
gradient = Image.new('RGB', (width, height), color=(0, 0, 0))
|
||
draw = ImageDraw.Draw(gradient)
|
||
|
||
# 绘制渐变
|
||
for y in range(height):
|
||
r = int(103 + (118 - 103) * y / height)
|
||
g = int(126 + (74 - 126) * y / height)
|
||
b = int(234 + (162 - 234) * y / height)
|
||
draw.line([(0, y), (width, y)], fill=(r, g, b))
|
||
|
||
# 应用半透明图片
|
||
image_enhanced = ImageEnhance.Brightness(image).enhance(0.8)
|
||
gradient.paste(image_enhanced, (0, 0))
|
||
|
||
# 应用缩放和文字叠加
|
||
gradient = self._apply_zoom(gradient)
|
||
gradient = self._add_text_overlay(gradient)
|
||
|
||
# 保存
|
||
cover_path = os.path.join(self.output_dir, 'cover_gradient.png')
|
||
gradient.save(cover_path, quality=95)
|
||
|
||
return cover_path
|
||
|
||
def _generate_split_cover(self, frame: np.ndarray) -> str:
|
||
"""生成斜切设计封面"""
|
||
# 转换为PIL
|
||
image = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
|
||
width, height = image.size
|
||
|
||
# 创建斜切背景
|
||
background = Image.new('RGB', (width, height), color=(255, 255, 255))
|
||
draw = ImageDraw.Draw(background)
|
||
|
||
# 绘制斜切色块
|
||
draw.polygon([(0, 0), (width * 0.6, 0), (width * 0.4, height), (0, height)],
|
||
fill=(103, 126, 234))
|
||
draw.polygon([(width * 0.6, 0), (width, 0), (width, height), (width * 0.4, height)],
|
||
fill=(118, 74, 162))
|
||
|
||
# 应用半透明图片
|
||
image_enhanced = ImageEnhance.Brightness(image).enhance(0.8)
|
||
background.paste(image_enhanced, (0, 0))
|
||
|
||
# 应用缩放和文字叠加
|
||
background = self._apply_zoom(background)
|
||
background = self._add_text_overlay(background)
|
||
|
||
# 保存
|
||
cover_path = os.path.join(self.output_dir, 'cover_split.png')
|
||
background.save(cover_path, quality=95)
|
||
|
||
return cover_path
|
||
|
||
def _apply_zoom(self, image: Image.Image, zoom_level: float = None) -> Image.Image:
|
||
"""
|
||
对图像应用缩放效果
|
||
|
||
Args:
|
||
image: 输入图像
|
||
zoom_level: 缩放倍数，如果为None则使用self.zoom_level
|
||
|
||
Returns:
|
||
缩放后的图像
|
||
"""
|
||
if zoom_level is None:
|
||
zoom_level = self.zoom_level
|
||
|
||
if zoom_level == 1.0:
|
||
return image
|
||
|
||
width, height = image.size
|
||
new_width = int(width * zoom_level)
|
||
new_height = int(height * zoom_level)
|
||
|
||
print(f"Applying zoom level {zoom_level}x: {width}x{height} -> {new_width}x{new_height}", file=sys.stderr)
|
||
|
||
# 使用高质量重采样
|
||
zoomed = image.resize((new_width, new_height), Image.LANCZOS)
|
||
|
||
# 如果放大，需要裁剪中心区域
|
||
if zoom_level > 1.0:
|
||
left = (new_width - width) // 2
|
||
top = (new_height - height) // 2
|
||
zoomed = zoomed.crop((left, top, left + width, top + height))
|
||
print(f"Cropped zoomed image to center: {width}x{height}", file=sys.stderr)
|
||
# 如果缩小，需要填充边缘
|
||
elif zoom_level < 1.0:
|
||
result = Image.new('RGB', (width, height), (0, 0, 0))
|
||
left = (width - new_width) // 2
|
||
top = (height - new_height) // 2
|
||
result.paste(zoomed, (left, top))
|
||
print(f"Padded zoomed image to original size: {width}x{height}", file=sys.stderr)
|
||
return result
|
||
|
||
return zoomed
|
||
|
||
def _hex_to_rgb(self, hex_color: str) -> tuple:
|
||
"""
|
||
将十六进制颜色转换为RGB元组
|
||
|
||
Args:
|
||
hex_color: 十六进制颜色字符串，如 '#FFFFFF' 或 'FFFFFF'
|
||
|
||
Returns:
|
||
RGB元组，如 (255, 255, 255)
|
||
"""
|
||
hex_color = hex_color.lstrip('#')
|
||
return tuple(int(hex_color[i:i+2], 16) for i in (0, 2, 4))
|
||
|
||
def _draw_char_with_effects(self, image: Image.Image, char: str, x: int, y: int, font,
|
||
color: tuple, title_config: dict) -> Image.Image:
|
||
"""
|
||
绘制带阴影和描边效果的单个字符
|
||
|
||
Args:
|
||
image: PIL图像（RGBA模式）
|
||
char: 要绘制的字符
|
||
x, y: 字符位置
|
||
font: 字体对象
|
||
color: 文字颜色（RGB元组）
|
||
title_config: 标题配置（包含阴影和描边参数）
|
||
|
||
Returns:
|
||
绘制后的图像
|
||
"""
|
||
# 获取阴影参数
|
||
shadow_color = self._hex_to_rgb(title_config.get('shadowColor', '#000000'))
|
||
shadow_offset_x = title_config.get('shadowOffsetX', 0)
|
||
shadow_offset_y = title_config.get('shadowOffsetY', 0)
|
||
shadow_blur = title_config.get('shadowBlur', 0)
|
||
|
||
# 获取描边参数
|
||
stroke_width = title_config.get('strokeWidth', 0)
|
||
stroke_color = self._hex_to_rgb(title_config.get('strokeColor', '#000000'))
|
||
|
||
# 1. 绘制阴影（如果有偏移）
|
||
if shadow_offset_x != 0 or shadow_offset_y != 0:
|
||
shadow_x = x + shadow_offset_x
|
||
shadow_y = y + shadow_offset_y
|
||
|
||
# 调试：打印阴影位置计算
|
||
print(f"[DEBUG] Shadow for char '{char}': text_pos=({x}, {y}), offset=({shadow_offset_x}, {shadow_offset_y}), shadow_pos=({shadow_x}, {shadow_y})", file=sys.stderr)
|
||
|
||
if shadow_blur > 0:
|
||
# 创建阴影图层并模糊
|
||
shadow_layer = Image.new('RGBA', image.size, (0, 0, 0, 0))
|
||
shadow_draw = ImageDraw.Draw(shadow_layer)
|
||
shadow_draw.text((shadow_x, shadow_y), char, font=font, fill=(*shadow_color, 180))
|
||
shadow_layer = shadow_layer.filter(ImageFilter.GaussianBlur(radius=shadow_blur))
|
||
image = Image.alpha_composite(image, shadow_layer)
|
||
else:
|
||
draw = ImageDraw.Draw(image)
|
||
draw.text((shadow_x, shadow_y), char, font=font, fill=(*shadow_color, 180))
|
||
|
||
# 2. 绘制描边
|
||
if stroke_width > 0:
|
||
draw = ImageDraw.Draw(image)
|
||
for offset_x in range(-stroke_width, stroke_width + 1):
|
||
for offset_y in range(-stroke_width, stroke_width + 1):
|
||
if offset_x == 0 and offset_y == 0:
|
||
continue
|
||
draw.text((x + offset_x, y + offset_y), char, font=font, fill=(*stroke_color, 255))
|
||
|
||
# 3. 绘制主体文字
|
||
draw = ImageDraw.Draw(image)
|
||
draw.text((x, y), char, font=font, fill=(*color, 255))
|
||
|
||
return image
|
||
|
||
def _load_font(self, font_name: str = None, font_size: int = 24, font_weight: int = 400) -> ImageFont.FreeTypeFont:
|
||
"""
|
||
加载字体，支持自定义字体、ziti目录、系统字体
|
||
优先顺序：ziti目录 -> FontManager -> 系统字体 -> 默认字体
|
||
|
||
Args:
|
||
font_name: 字体名称（如 "微软雅黑"、"墨趣古风体" 等）
|
||
font_size: 字体大小
|
||
font_weight: 字体粗细 (100-1000)
|
||
|
||
Returns:
|
||
字体对象
|
||
"""
|
||
import platform
|
||
|
||
# 1. 首先直接尝试在 ziti 目录查找字体文件
|
||
if font_name:
|
||
# 多个可能的 ziti 目录路径（相对和绝对）
|
||
possible_ziti_paths = [
|
||
os.path.join(os.getcwd(), 'ziti'), # 当前工作目录
|
||
os.path.join(os.path.dirname(__file__), '..', 'ziti'), # 相对于脚本
|
||
os.path.join(os.path.dirname(__file__), '..', '..', 'ziti'), # 项目根目录
|
||
'ziti', # 相对路径
|
||
os.path.expanduser('~/ziti'), # 用户主目录
|
||
]
|
||
|
||
# 尝试找到 ziti 目录
|
||
for ziti_dir in possible_ziti_paths:
|
||
if os.path.isdir(ziti_dir):
|
||
print(f"Found ziti directory: {ziti_dir}", file=sys.stderr)
|
||
|
||
# 在 ziti 目录中查找字体文件
|
||
for ext in ['.ttf', '.otf', '.ttc']:
|
||
# 直接按字体名称查找
|
||
font_path = os.path.join(ziti_dir, font_name + ext)
|
||
if os.path.exists(font_path):
|
||
try:
|
||
font = ImageFont.truetype(font_path, font_size)
|
||
print(f"✓ Successfully loaded ziti font '{font_name}': {font_path}", file=sys.stderr)
|
||
return font
|
||
except Exception as e:
|
||
print(f"✗ Failed to load ziti font from {font_path}: {e}", file=sys.stderr)
|
||
|
||
# 如果没有找到，扫描目录中所有文件进行模糊匹配
|
||
try:
|
||
for file in os.listdir(ziti_dir):
|
||
if not file.startswith('.') and (file.endswith('.ttf') or file.endswith('.otf') or file.endswith('.ttc')):
|
||
file_base = os.path.splitext(file)[0]
|
||
# 检查是否匹配（精确或包含）
|
||
if file_base == font_name or font_name in file_base:
|
||
font_path = os.path.join(ziti_dir, file)
|
||
try:
|
||
font = ImageFont.truetype(font_path, font_size)
|
||
print(f"✓ Successfully loaded ziti font '{font_name}' (matched as {file}): {font_path}", file=sys.stderr)
|
||
return font
|
||
except Exception as e:
|
||
print(f"✗ Failed to load matched font {font_path}: {e}", file=sys.stderr)
|
||
except Exception as e:
|
||
print(f"✗ Error scanning ziti directory: {e}", file=sys.stderr)
|
||
|
||
# 2. 如果 ziti 目录查找失败，使用 FontManager
|
||
if font_name and FONT_MANAGER_AVAILABLE:
|
||
try:
|
||
font_manager = get_font_manager()
|
||
font_path = font_manager.find_font(font_name, font_weight)
|
||
|
||
if font_path and os.path.exists(font_path):
|
||
try:
|
||
font = ImageFont.truetype(font_path, font_size)
|
||
print(f"✓ Successfully loaded font via FontManager '{font_name}': {font_path}", file=sys.stderr)
|
||
return font
|
||
except Exception as e:
|
||
print(f"✗ Failed to load font from {font_path}: {e}", file=sys.stderr)
|
||
except Exception as e:
|
||
print(f"✗ FontManager error: {e}", file=sys.stderr)
|
||
|
||
# 3. 降级：根据操作系统加载默认系统字体
|
||
print(f"⚠ Font '{font_name}' not found, trying system fonts...", file=sys.stderr)
|
||
system = platform.system()
|
||
font_paths = []
|
||
|
||
if system == 'Windows':
|
||
font_dir = None
|
||
font_paths = [
|
||
os.path.join(font_dir, 'NotoSerifCJK-VF'), # 微软雅黑
|
||
'', # 微软雅黑 Bold
|
||
'', # 黑体
|
||
'', # 宋体
|
||
os.path.join(font_dir, 'arial.ttf'), # Arial
|
||
]
|
||
elif system == 'Darwin': # macOS
|
||
font_paths = [
|
||
'/Library/Fonts/PingFang.ttc',
|
||
'/System/Library/Fonts/PingFang.ttc',
|
||
'/Library/Fonts/Arial.ttf',
|
||
]
|
||
else: # Linux
|
||
font_paths = [
|
||
'/usr/share/fonts/truetype/wqy/wqy-zenhei.ttc',
|
||
'/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf',
|
||
'/usr/share/fonts/truetype/liberation/LiberationSans-Regular.ttf',
|
||
]
|
||
|
||
# 尝试加载每个系统字体
|
||
for font_path in font_paths:
|
||
if os.path.exists(font_path):
|
||
try:
|
||
font = ImageFont.truetype(font_path, font_size)
|
||
print(f"✓ Successfully loaded system font: {font_path}", file=sys.stderr)
|
||
return font
|
||
except Exception as e:
|
||
print(f"✗ Failed to load system font {font_path}: {e}", file=sys.stderr)
|
||
continue
|
||
|
||
# 4. 最后的降级方案：使用 PIL 默认字体
|
||
print("⚠ All font loading attempts failed, using PIL default font (text may not render correctly for CJK)", file=sys.stderr)
|
||
return ImageFont.load_default()
|
||
|
||
def _add_text_overlay(self, image: Image.Image) -> Image.Image:
|
||
"""
|
||
在封面上添加文字叠加
|
||
|
||
Args:
|
||
image: 输入图像
|
||
|
||
Returns:
|
||
添加了文字的图像
|
||
"""
|
||
if not self.cover_text or not self.cover_text.strip():
|
||
return image
|
||
|
||
print(f"Adding text overlay: '{self.cover_text}'", file=sys.stderr)
|
||
|
||
# 创建绘图对象
|
||
draw = ImageDraw.Draw(image)
|
||
|
||
# 获取字体名称（从配置中优先获取，否则使用微软雅黑作为默认）
|
||
font_name = self.config.get('coverTextFont', 'NotoSerifCJK-VF')
|
||
|
||
# 加载字体 - 改进的字体加载逻辑，支持自定义字体
|
||
font = self._load_font(font_name, self.cover_text_size)
|
||
|
||
# 计算文字边界框
|
||
text_bbox = draw.textbbox((0, 0), self.cover_text, font=font)
|
||
text_width = text_bbox[2] - text_bbox[0]
|
||
text_height = text_bbox[3] - text_bbox[1]
|
||
|
||
# 根据位置参数定位文字
|
||
x = (image.width - text_width) // 2
|
||
if self.cover_text_position == 'top':
|
||
y = 50
|
||
elif self.cover_text_position == 'center':
|
||
y = (image.height - text_height) // 2
|
||
else: # bottom
|
||
y = image.height - text_height - 50
|
||
|
||
print(f"Text position: ({x}, {y}), size: {text_width}x{text_height}", file=sys.stderr)
|
||
|
||
# 转换颜色
|
||
text_color = self._hex_to_rgb(self.cover_text_color)
|
||
outline_rgb = self._hex_to_rgb(self.outline_color)
|
||
|
||
# 绘制文字描边
|
||
if self.outline_width > 0:
|
||
for offset_x in range(-self.outline_width, self.outline_width + 1):
|
||
for offset_y in range(-self.outline_width, self.outline_width + 1):
|
||
if offset_x == 0 and offset_y == 0:
|
||
continue
|
||
draw.text((x + offset_x, y + offset_y), self.cover_text,
|
||
font=font, fill=outline_rgb)
|
||
|
||
# 绘制文字主体
|
||
draw.text((x, y), self.cover_text, font=font, fill=text_color)
|
||
|
||
print("Text overlay added successfully", file=sys.stderr)
|
||
return image
|
||
|
||
def _generate_custom_template_cover(self, frame: np.ndarray) -> str:
|
||
"""
|
||
使用自定义模板生成封面
|
||
|
||
Args:
|
||
frame: 视频帧
|
||
|
||
Returns:
|
||
生成的封面路径
|
||
"""
|
||
try:
|
||
print("Generating custom template cover...", file=sys.stderr)
|
||
template = self.template
|
||
|
||
# 转换为PIL图像
|
||
image = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
|
||
width, height = image.size
|
||
|
||
# 1. 应用背景模糊
|
||
# 支持两种格式：旧格式 template['background']['blurEnabled'] 和新格式 template['backgroundBlurEnabled']
|
||
blur_enabled = template.get('backgroundBlurEnabled', template.get('background', {}).get('blurEnabled', False))
|
||
if blur_enabled:
|
||
blur_radius = template.get('backgroundBlurIntensity', template.get('background', {}).get('blurRadius', 15))
|
||
print(f"Applying background blur: {blur_radius}", file=sys.stderr)
|
||
image = image.filter(ImageFilter.GaussianBlur(radius=blur_radius))
|
||
|
||
# 2. 处理人像描边（如果启用RMBG）
|
||
# 支持两种格式：旧格式 template['portrait']['strokeEnabled'] 和新格式 template['personBorderEnabled']
|
||
stroke_enabled = template.get('personBorderEnabled', template.get('portrait', {}).get('strokeEnabled', False))
|
||
if stroke_enabled and REMBG_AVAILABLE:
|
||
try:
|
||
print("Extracting portrait for stroke effect...", file=sys.stderr)
|
||
portrait = rmbg_remove(image)
|
||
|
||
# 创建描边效果
|
||
# 支持两种格式
|
||
stroke_color = template.get('personBorderColor', template.get('portrait', {}).get('strokeColor', '#FFD700'))
|
||
stroke_width = template.get('personBorderWidth', template.get('portrait', {}).get('strokeWidth', 3))
|
||
stroke_type = template.get('personBorderStyle', template.get('portrait', {}).get('strokeType', 'solid'))
|
||
|
||
print(f"Stroke: color={stroke_color}, width={stroke_width}, type={stroke_type}", file=sys.stderr)
|
||
|
||
# 合成带描边的人像
|
||
if portrait.mode == 'RGBA':
|
||
# 获取alpha通道作为mask
|
||
alpha = portrait.split()[3]
|
||
alpha_np = np.array(alpha)
|
||
|
||
# 改进的描边方法：使用固定迭代次数和可变 kernel 大小
|
||
# 这样描边宽度可控，但紧贴轮廓
|
||
kernel_size = max(3, stroke_width * 2 + 1) # 确保是奇数
|
||
kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (kernel_size, kernel_size))
|
||
|
||
# 只膨胀1次，避免离轮廓太远
|
||
alpha_dilated = cv2.dilate(alpha_np, kernel, iterations=1)
|
||
|
||
# 描边 = 膨胀后的轮廓 - 原始轮廓（这样得到紧贴的描边）
|
||
outline_mask = cv2.subtract(alpha_dilated, alpha_np)
|
||
|
||
# 转换描边颜色
|
||
stroke_rgb = self._hex_to_rgb(stroke_color)
|
||
outline_layer = Image.new('RGBA', (width, height), (*stroke_rgb, 0))
|
||
|
||
# 应用描边
|
||
outline_pil = Image.fromarray(outline_mask)
|
||
outline_colored = Image.new('RGBA', (width, height), (*stroke_rgb, 255))
|
||
outline_colored.putalpha(outline_pil)
|
||
outline_layer = Image.alpha_composite(outline_layer, outline_colored)
|
||
|
||
# 创建结果：背景 + 描边 + 人像
|
||
result = Image.new('RGB', (width, height))
|
||
result.paste(image, (0, 0))
|
||
|
||
# 将描边和人像合成到背景上
|
||
result_rgba = result.convert('RGBA')
|
||
result_rgba = Image.alpha_composite(result_rgba, outline_layer)
|
||
result_rgba = Image.alpha_composite(result_rgba, portrait)
|
||
image = result_rgba.convert('RGB')
|
||
|
||
print("Portrait stroke effect applied successfully", file=sys.stderr)
|
||
except Exception as e:
|
||
print(f"Portrait stroke failed: {e}", file=sys.stderr)
|
||
import traceback
|
||
traceback.print_exc(file=sys.stderr)
|
||
|
||
|
||
# 3. 添加蒙版（如果有）
|
||
mask_config = template.get('mask', {})
|
||
mask_path = mask_config.get('imagePath', '')
|
||
if mask_path and os.path.exists(mask_path):
|
||
try:
|
||
print(f"Adding mask from: {mask_path}", file=sys.stderr)
|
||
mask_img = Image.open(mask_path).convert('RGBA')
|
||
|
||
# 获取蒙版位置和大小
|
||
mask_pos = mask_config.get('position', {'x': 10, 'y': 10})
|
||
mask_size = mask_config.get('size', {'width': 30, 'height': 20})
|
||
mask_opacity = mask_config.get('opacity', 0.8)
|
||
|
||
mask_width = int(mask_size['width'] / 100 * width)
|
||
mask_height = int(mask_size['height'] / 100 * height)
|
||
mask_x = int(mask_pos['x'] / 100 * width)
|
||
mask_y = int(mask_pos['y'] / 100 * height)
|
||
|
||
# 调整蒙版大小和透明度
|
||
mask_img = mask_img.resize((mask_width, mask_height), Image.LANCZOS)
|
||
if mask_opacity < 1.0:
|
||
alpha = mask_img.split()[3]
|
||
alpha = alpha.point(lambda p: int(p * mask_opacity))
|
||
mask_img.putalpha(alpha)
|
||
|
||
# 将蒙版粘贴到图像上
|
||
image_rgba = image.convert('RGBA')
|
||
image_rgba.paste(mask_img, (mask_x - mask_width // 2, mask_y - mask_height // 2), mask_img)
|
||
image = image_rgba.convert('RGB')
|
||
|
||
print("Mask applied successfully", file=sys.stderr)
|
||
except Exception as e:
|
||
print(f"Failed to add mask: {e}", file=sys.stderr)
|
||
import traceback
|
||
traceback.print_exc(file=sys.stderr)
|
||
|
||
# 4. 添加标题
|
||
# 转换为RGBA以支持透明图层
|
||
if image.mode != 'RGBA':
|
||
image = image.convert('RGBA')
|
||
|
||
# 加载字体 - 使用模板中的字体设置
|
||
main_font_family = template['titles']['main'].get('fontFamily', 'NotoSerifCJK-VF')
|
||
main_font_size = template['titles']['main'].get('fontSize', 60)
|
||
main_font_weight = template['titles']['main'].get('fontWeight', 400)
|
||
|
||
sub_font_family = template['titles']['sub'].get('fontFamily', 'NotoSerifCJK-VF')
|
||
sub_font_size = template['titles']['sub'].get('fontSize', 40)
|
||
sub_font_weight = template['titles']['sub'].get('fontWeight', 400)
|
||
|
||
print(f"Loading main font: {main_font_family}, size={main_font_size}, weight={main_font_weight}", file=sys.stderr)
|
||
print(f"Loading sub font: {sub_font_family}, size={sub_font_size}, weight={sub_font_weight}", file=sys.stderr)
|
||
|
||
font_main = self._load_font(main_font_family, main_font_size, main_font_weight)
|
||
font_sub = self._load_font(sub_font_family, sub_font_size, sub_font_weight)
|
||
|
||
draw = ImageDraw.Draw(image)
|
||
|
||
# 绘制主标题（支持自动换行）
|
||
main_title = template['titles']['main']
|
||
if main_title.get('text'):
|
||
main_text = main_title['text']
|
||
main_color = self._hex_to_rgb(main_title.get('color', '#FFFFFF'))
|
||
main_pos_x = int(width * main_title['position']['x'] / 100)
|
||
main_pos_y = int(height * main_title['position']['y'] / 100)
|
||
|
||
# 调试：打印接收到的位置参数
|
||
print(f"[DEBUG] Main title position - Received from frontend: x={main_title['position']['x']}%, y={main_title['position']['y']}%", file=sys.stderr)
|
||
print(f"[DEBUG] Main title position - Calculated pixels: x={main_pos_x}px, y={main_pos_y}px", file=sys.stderr)
|
||
print(f"[DEBUG] Canvas size: {width}x{height}", file=sys.stderr)
|
||
|
||
max_chars_per_line = main_title.get('maxLength', 4) # 每行最多字符数
|
||
|
||
# 获取文字方向和间距参数
|
||
direction = main_title.get('direction', 'horizontal')
|
||
font_size = main_title.get('fontSize', 60)
|
||
char_spacing = main_title.get('charSpacing', int(font_size * 0.2)) # 字符间距
|
||
line_spacing = main_title.get('lineSpacing', int(font_size * 1.2)) # 行间距
|
||
|
||
# 调试日志
|
||
print(f"[DEBUG] Main title config:", file=sys.stderr)
|
||
print(f" Text: '{main_text}' (length: {len(main_text)})", file=sys.stderr)
|
||
print(f" maxLength: {max_chars_per_line}", file=sys.stderr)
|
||
print(f" Direction: {direction}", file=sys.stderr)
|
||
print(f" Char spacing: {char_spacing}px, Line spacing: {line_spacing}px", file=sys.stderr)
|
||
print(f" Position: ({main_pos_x}, {main_pos_y})", file=sys.stderr)
|
||
|
||
# 将文字分行
|
||
lines = []
|
||
for i in range(0, len(main_text), max_chars_per_line):
|
||
lines.append(main_text[i:i + max_chars_per_line])
|
||
|
||
print(f" Split into {len(lines)} lines: {lines}", file=sys.stderr)
|
||
|
||
if direction == 'vertical':
|
||
# 竖排：从右到左排列，每列从上到下
|
||
# 计算总宽度和高度
|
||
total_width = (len(lines) - 1) * line_spacing + font_size
|
||
|
||
# 找出最长的列
|
||
max_col_chars = max(len(line) for line in lines) if lines else 1
|
||
total_height = (max_col_chars - 1) * (font_size + char_spacing) + font_size
|
||
|
||
# 计算起始位置（整体居中）
|
||
start_x = main_pos_x + total_width // 2 - font_size // 2
|
||
start_y = main_pos_y - total_height // 2
|
||
|
||
# 逐列绘制（从右到左）
|
||
for col_idx, line in enumerate(lines):
|
||
col_x = start_x - col_idx * line_spacing
|
||
|
||
# 逐字符绘制（从上到下）
|
||
for char_idx, char in enumerate(line):
|
||
char_y = start_y + char_idx * (font_size + char_spacing)
|
||
|
||
# 计算字符实际宽度以水平居中
|
||
try:
|
||
bbox = draw.textbbox((0, 0), char, font=font_main)
|
||
char_w = bbox[2] - bbox[0]
|
||
char_x = col_x - char_w // 2 + font_size // 2
|
||
except:
|
||
char_x = col_x
|
||
|
||
# 使用带阴影和描边效果的绘制函数
|
||
image = self._draw_char_with_effects(
|
||
image, char, char_x, char_y, font_main, main_color, main_title
|
||
)
|
||
draw = ImageDraw.Draw(image)
|
||
|
||
print(f" Char '{char}' at ({char_x}, {char_y})", file=sys.stderr)
|
||
else:
|
||
# 横排：从左到右，从上到下（默认）
|
||
# 计算总高度
|
||
total_height = (len(lines) - 1) * line_spacing + font_size
|
||
start_y = main_pos_y - total_height // 2
|
||
|
||
# 逐行绘制
|
||
for row_idx, line in enumerate(lines):
|
||
line_y = start_y + row_idx * line_spacing
|
||
|
||
# 计算该行的总宽度
|
||
line_width = (len(line) - 1) * (font_size + char_spacing) + font_size
|
||
start_x = main_pos_x - line_width // 2
|
||
|
||
# 逐字符绘制（从左到右）
|
||
for char_idx, char in enumerate(line):
|
||
char_x = start_x + char_idx * (font_size + char_spacing)
|
||
|
||
# 使用带阴影和描边效果的绘制函数
|
||
image = self._draw_char_with_effects(
|
||
image, char, char_x, line_y, font_main, main_color, main_title
|
||
)
|
||
draw = ImageDraw.Draw(image)
|
||
|
||
print(f" Char '{char}' at ({char_x}, {line_y})", file=sys.stderr)
|
||
|
||
print(f"Drew main title: '{main_text}' ({len(lines)} lines) at ({main_pos_x}, {main_pos_y})", file=sys.stderr)
|
||
|
||
# 绘制副标题（支持自动换行）
|
||
sub_title = template['titles']['sub']
|
||
if sub_title.get('text'):
|
||
sub_text = sub_title['text']
|
||
sub_color = self._hex_to_rgb(sub_title.get('color', '#FFFFFF'))
|
||
sub_pos_x = int(width * sub_title['position']['x'] / 100)
|
||
sub_pos_y = int(height * sub_title['position']['y'] / 100)
|
||
|
||
# 调试：打印接收到的位置参数
|
||
print(f"[DEBUG] Subtitle position - Received from frontend: x={sub_title['position']['x']}%, y={sub_title['position']['y']}%", file=sys.stderr)
|
||
print(f"[DEBUG] Subtitle position - Calculated pixels: x={sub_pos_x}px, y={sub_pos_y}px", file=sys.stderr)
|
||
|
||
max_chars_per_line = sub_title.get('maxLength', 3) # 每行最多字符数
|
||
|
||
# 获取文字方向和间距参数
|
||
direction = sub_title.get('direction', 'horizontal')
|
||
font_size = sub_title.get('fontSize', 40)
|
||
char_spacing = sub_title.get('charSpacing', int(font_size * 0.2)) # 字符间距
|
||
line_spacing = sub_title.get('lineSpacing', int(font_size * 1.2)) # 行间距
|
||
|
||
# 调试日志
|
||
print(f"[DEBUG] Sub title config:", file=sys.stderr)
|
||
print(f" Text: '{sub_text}' (length: {len(sub_text)})", file=sys.stderr)
|
||
print(f" maxLength: {max_chars_per_line}", file=sys.stderr)
|
||
print(f" Direction: '{direction}' (type: {type(direction).__name__}, equals 'vertical': {direction == 'vertical'})", file=sys.stderr)
|
||
print(f" Char spacing: {char_spacing}px, Line spacing: {line_spacing}px", file=sys.stderr)
|
||
print(f" Position: ({sub_pos_x}, {sub_pos_y})", file=sys.stderr)
|
||
print(f"[DEBUG] sub_title dict keys: {list(sub_title.keys())}", file=sys.stderr)
|
||
|
||
# 将文字分行
|
||
lines = []
|
||
for i in range(0, len(sub_text), max_chars_per_line):
|
||
lines.append(sub_text[i:i + max_chars_per_line])
|
||
|
||
print(f" Split into {len(lines)} lines: {lines}", file=sys.stderr)
|
||
|
||
if direction == 'vertical':
|
||
# 竖排：从右到左排列，每列从上到下
|
||
# 计算总宽度和高度
|
||
total_width = (len(lines) - 1) * line_spacing + font_size
|
||
|
||
# 找出最长的列
|
||
max_col_chars = max(len(line) for line in lines) if lines else 1
|
||
total_height = (max_col_chars - 1) * (font_size + char_spacing) + font_size
|
||
|
||
# 计算起始位置（整体居中）
|
||
start_x = sub_pos_x + total_width // 2 - font_size // 2
|
||
start_y = sub_pos_y - total_height // 2
|
||
|
||
# 逐列绘制（从右到左）
|
||
for col_idx, line in enumerate(lines):
|
||
col_x = start_x - col_idx * line_spacing
|
||
|
||
# 逐字符绘制（从上到下）
|
||
for char_idx, char in enumerate(line):
|
||
char_y = start_y + char_idx * (font_size + char_spacing)
|
||
|
||
# 计算字符实际宽度以水平居中
|
||
try:
|
||
bbox = draw.textbbox((0, 0), char, font=font_sub)
|
||
char_w = bbox[2] - bbox[0]
|
||
char_x = col_x - char_w // 2 + font_size // 2
|
||
except:
|
||
char_x = col_x
|
||
|
||
# 使用带阴影和描边效果的绘制函数
|
||
image = self._draw_char_with_effects(
|
||
image, char, char_x, char_y, font_sub, sub_color, sub_title
|
||
)
|
||
draw = ImageDraw.Draw(image)
|
||
|
||
print(f" Char '{char}' at ({char_x}, {char_y})", file=sys.stderr)
|
||
else:
|
||
# 横排：从左到右，从上到下（默认）
|
||
# 计算总高度
|
||
total_height = (len(lines) - 1) * line_spacing + font_size
|
||
start_y = sub_pos_y - total_height // 2
|
||
|
||
# 逐行绘制
|
||
for row_idx, line in enumerate(lines):
|
||
line_y = start_y + row_idx * line_spacing
|
||
|
||
# 计算该行的总宽度
|
||
line_width = (len(line) - 1) * (font_size + char_spacing) + font_size
|
||
start_x = sub_pos_x - line_width // 2
|
||
|
||
# 逐字符绘制（从左到右）
|
||
for char_idx, char in enumerate(line):
|
||
char_x = start_x + char_idx * (font_size + char_spacing)
|
||
char_y = line_y # ✅ 添加char_y变量以保持一致性
|
||
|
||
# 使用带阴影和描边效果的绘制函数
|
||
image = self._draw_char_with_effects(
|
||
image, char, char_x, char_y, font_sub, sub_color, sub_title
|
||
)
|
||
draw = ImageDraw.Draw(image)
|
||
|
||
print(f" Char '{char}' at ({char_x}, {line_y})", file=sys.stderr)
|
||
|
||
print(f"Drew sub title: '{sub_text}' ({len(lines)} lines) at ({sub_pos_x}, {sub_pos_y})", file=sys.stderr)
|
||
|
||
# 转换回RGB并保存封面
|
||
if image.mode == 'RGBA':
|
||
image = image.convert('RGB')
|
||
cover_path = os.path.join(self.output_dir, 'cover_custom_template.png')
|
||
image.save(cover_path, quality=95)
|
||
|
||
print(f"Custom template cover generated: {cover_path}", file=sys.stderr)
|
||
return cover_path
|
||
|
||
except Exception as e:
|
||
print(f"Error generating custom template cover: {e}", file=sys.stderr)
|
||
raise Exception(f"Custom template cover generation failed: {str(e)}")
|
||
|
||
def generate_subtitles(self) -> List[Dict[str, Any]]:
|
||
"""
|
||
生成字幕列表
|
||
支持两种模式：
|
||
1. FUNASR模式：使用语音识别结果的时间戳
|
||
2. 文本模式：根据文案按行分割，平均分配时间
|
||
|
||
现在支持：逐句情感分析 + 动态字幕样式
|
||
|
||
Returns:
|
||
字幕列表，每个字幕包含文本、开始时间、结束时间、情感、样式等
|
||
"""
|
||
try:
|
||
subtitles = []
|
||
|
||
# 模式1: 使用FUNASR识别结果
|
||
if self.use_funasr and self.funasr_result:
|
||
print("Generating subtitles from FUNASR result with sentiment analysis...", file=sys.stderr)
|
||
segments = self.funasr_result.get('segments', [])
|
||
|
||
for i, segment in enumerate(segments):
|
||
text = segment.get('text', '')
|
||
|
||
# 对每句话进行情感分析
|
||
sentiment_analysis = self.analyze_sentence_sentiment(text)
|
||
sentiment = sentiment_analysis['sentiment']
|
||
score = sentiment_analysis['score']
|
||
|
||
# 根据情感选择样式
|
||
dynamic_style = self.get_style_by_sentiment(sentiment, score)
|
||
|
||
subtitle = {
|
||
'index': i + 1,
|
||
'text': text,
|
||
'start_time': segment.get('start', i * 3.0),
|
||
'end_time': segment.get('end', (i + 1) * 3.0),
|
||
'style': dynamic_style, # 动态样式
|
||
'sentiment': sentiment, # 情感类型
|
||
'sentiment_score': score, # 情感分数
|
||
'fontName': self.subtitle_font_name, # 加入字体名称
|
||
'fontSize': self.subtitle_font_size, # 加入字体大小
|
||
}
|
||
subtitles.append(subtitle)
|
||
|
||
print(f"Generated {len(subtitles)} subtitles from FUNASR with sentiment-based styles", file=sys.stderr)
|
||
|
||
# 模式2: 基于文本的简单分割（支持情感分析）
|
||
else:
|
||
print("Generating subtitles from script text with sentiment analysis...", file=sys.stderr)
|
||
# 先按行分割
|
||
lines = [line.strip() for line in self.script_text.split('\n') if line.strip()]
|
||
|
||
# 如果只有一行,则按句子分割（使用句号、问号、感叹号等标点）
|
||
if len(lines) == 1:
|
||
import re
|
||
# 使用正则表达式按中英文标点分割句子
|
||
sentences = re.split(r'[。！？\.\!\?]+', lines[0])
|
||
lines = [s.strip() for s in sentences if s.strip()]
|
||
print(f"Single line detected, split into {len(lines)} sentences", file=sys.stderr)
|
||
|
||
# 生成字幕列表（简单示例，每句3秒）
|
||
for i, line in enumerate(lines):
|
||
# 对每句话进行情感分析
|
||
sentiment_analysis = self.analyze_sentence_sentiment(line)
|
||
sentiment = sentiment_analysis['sentiment']
|
||
score = sentiment_analysis['score']
|
||
|
||
# 根据情感选择样式
|
||
dynamic_style = self.get_style_by_sentiment(sentiment, score)
|
||
|
||
subtitle = {
|
||
'index': i + 1,
|
||
'text': line,
|
||
'start_time': i * 3.0,
|
||
'end_time': (i + 1) * 3.0,
|
||
'style': dynamic_style, # 动态样式
|
||
'sentiment': sentiment, # 情感类型
|
||
'sentiment_score': score, # 情感分数
|
||
'fontName': self.subtitle_font_name, # 加入字体名称
|
||
'fontSize': self.subtitle_font_size, # 加入字体大小
|
||
}
|
||
subtitles.append(subtitle)
|
||
|
||
print(f"Generated {len(subtitles)} subtitles from text with sentiment-based styles", file=sys.stderr)
|
||
|
||
return self.normalize_subtitles(subtitles)
|
||
except Exception as e:
|
||
raise Exception(f"Error generating subtitles: {str(e)}")
|
||
|
||
def normalize_subtitles(self, subtitles: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
||
min_duration = 0.18
|
||
overlap_gap = 0.03
|
||
duplicate_time_threshold = 0.2
|
||
normalized: List[Dict[str, Any]] = []
|
||
|
||
for original in subtitles:
|
||
text = str(original.get('text', '')).replace('\r', ' ').replace('\n', ' ').strip()
|
||
if not text:
|
||
continue
|
||
|
||
try:
|
||
start_time = float(original.get('start_time', 0.0) or 0.0)
|
||
except (TypeError, ValueError):
|
||
start_time = 0.0
|
||
|
||
try:
|
||
end_time = float(original.get('end_time', start_time + min_duration) or (start_time + min_duration))
|
||
except (TypeError, ValueError):
|
||
end_time = start_time + min_duration
|
||
|
||
start_time = max(0.0, start_time)
|
||
end_time = max(start_time + min_duration, end_time)
|
||
|
||
subtitle = dict(original)
|
||
subtitle['text'] = text
|
||
subtitle['start_time'] = start_time
|
||
subtitle['end_time'] = end_time
|
||
normalized.append(subtitle)
|
||
|
||
normalized.sort(key=lambda item: (item['start_time'], item['end_time'], item.get('index', 0)))
|
||
|
||
cleaned: List[Dict[str, Any]] = []
|
||
adjusted_count = 0
|
||
|
||
for subtitle in normalized:
|
||
if cleaned:
|
||
previous = cleaned[-1]
|
||
|
||
is_duplicate = (
|
||
subtitle['text'] == previous['text']
|
||
and abs(subtitle['start_time'] - previous['start_time']) <= duplicate_time_threshold
|
||
)
|
||
if is_duplicate:
|
||
previous['end_time'] = max(previous['end_time'], subtitle['end_time'])
|
||
adjusted_count += 1
|
||
continue
|
||
|
||
if subtitle['start_time'] < previous['end_time']:
|
||
preferred_previous_end = max(
|
||
previous['start_time'] + min_duration,
|
||
subtitle['start_time'] - overlap_gap
|
||
)
|
||
|
||
if preferred_previous_end < previous['end_time']:
|
||
previous['end_time'] = preferred_previous_end
|
||
adjusted_count += 1
|
||
|
||
if subtitle['start_time'] < previous['end_time']:
|
||
subtitle['start_time'] = previous['end_time'] + overlap_gap
|
||
subtitle['end_time'] = max(subtitle['start_time'] + min_duration, subtitle['end_time'])
|
||
adjusted_count += 1
|
||
|
||
cleaned.append(subtitle)
|
||
|
||
for index, subtitle in enumerate(cleaned, start=1):
|
||
subtitle['index'] = index
|
||
|
||
if adjusted_count > 0:
|
||
print(
|
||
f"Normalized subtitles: {len(subtitles)} -> {len(cleaned)}, adjusted {adjusted_count} entries",
|
||
file=sys.stderr
|
||
)
|
||
|
||
return cleaned
|
||
|
||
def save_srt(self, subtitles: List[Dict[str, Any]]) -> str:
|
||
"""
|
||
保存SRT字幕文件
|
||
|
||
Args:
|
||
subtitles: 字幕列表
|
||
|
||
Returns:
|
||
SRT文件路径
|
||
"""
|
||
try:
|
||
srt_path = os.path.join(self.output_dir, 'subtitles.srt')
|
||
|
||
with open(srt_path, 'w', encoding='utf-8') as f:
|
||
for sub in subtitles:
|
||
# SRT格式
|
||
f.write(f"{sub['index']}\n")
|
||
f.write(f"{self._format_time(sub['start_time'])} --> {self._format_time(sub['end_time'])}\n")
|
||
f.write(f"{sub['text']}\n\n")
|
||
|
||
return srt_path
|
||
except Exception as e:
|
||
raise Exception(f"Error saving SRT file: {str(e)}")
|
||
|
||
def _format_time(self, seconds: float) -> str:
|
||
"""格式化时间为SRT格式"""
|
||
hours = int(seconds // 3600)
|
||
minutes = int((seconds % 3600) // 60)
|
||
secs = int(seconds % 60)
|
||
millis = int((seconds % 1) * 1000)
|
||
|
||
return f"{hours:02d}:{minutes:02d}:{secs:02d},{millis:03d}"
|
||
|
||
def _apply_keyword_effects_to_subtitle(self, subtitle: Dict[str, Any]) -> Dict[str, Any]:
|
||
"""
|
||
对字幕应用关键词分组效果
|
||
|
||
Args:
|
||
subtitle: 字幕对象
|
||
|
||
Returns:
|
||
应用了效果的字幕对象
|
||
"""
|
||
# 查找该字幕中的关键词标记
|
||
text = subtitle.get('text', '')
|
||
|
||
# ✅ 修复：正确处理关键词标记的结构
|
||
# 前端发送的关键词标记结构：{ text: '关键词', groupId: 'xxx-xxx-xxx', positions: [...], count: 1 }
|
||
if not self.keyword_markers:
|
||
print(f"[_apply_keyword_effects_to_subtitle] No keyword markers found", file=sys.stderr)
|
||
return subtitle
|
||
|
||
# 检查是否有相关的关键词标记
|
||
for marker in self.keyword_markers:
|
||
# ✅ 尝试多种方式获取标记文本
|
||
marker_text = marker.get('text') or marker.get('keyword') or ''
|
||
|
||
if not marker_text:
|
||
continue
|
||
|
||
# 检查字幕中是否包含此关键词
|
||
if marker_text in text:
|
||
group_id = marker.get('groupId')
|
||
|
||
# 查找对应的分组
|
||
if group_id:
|
||
matched_group = None
|
||
for group in self.keyword_groups:
|
||
if group.get('id') == group_id:
|
||
matched_group = group
|
||
break
|
||
|
||
if matched_group:
|
||
# 应用分组的样式覆盖
|
||
style_override = matched_group.get('styleOverride', {})
|
||
|
||
# ✅ 即使没有 styleOverride，也要应用 effectId
|
||
effect_id = matched_group.get('effectId', 'none')
|
||
|
||
# 更新字幕的样式信息
|
||
subtitle['keywordEffect'] = {
|
||
'groupId': group_id,
|
||
'groupName': matched_group.get('name', ''),
|
||
'fontColor': style_override.get('fontColor', '#FFFFFF'),
|
||
'outlineColor': style_override.get('outlineColor', '#000000'),
|
||
'fontSize': style_override.get('fontSize', 24),
|
||
'outlineWidth': style_override.get('outlineWidth', 2),
|
||
'effectId': effect_id,
|
||
'effectDuration': style_override.get('effectDuration', 2.0),
|
||
}
|
||
print(f"[_apply_keyword_effects] Matched keyword '{marker_text}' -> group '{matched_group.get('name')}': effect={effect_id}, color={style_override.get('fontColor')}", file=sys.stderr)
|
||
# ✅ 找到匹配后，只返回，不继续搜索（一个字幕一个特效）
|
||
return subtitle
|
||
else:
|
||
print(f"[_apply_keyword_effects] Keyword '{marker_text}' found but no groupId", file=sys.stderr)
|
||
|
||
return subtitle
|
||
|
||
def save_ass(self, subtitles: List[Dict[str, Any]]) -> str:
|
||
"""
|
||
保存ASS字幕文件（Advanced Substation Alpha）
|
||
支持每个字幕不同的样式、字体和关键词分组效果
|
||
|
||
Args:
|
||
subtitles: 字幕列表
|
||
|
||
Returns:
|
||
ASS文件路径
|
||
"""
|
||
try:
|
||
ass_path = os.path.join(self.output_dir, 'subtitles.ass')
|
||
|
||
with open(ass_path, 'w', encoding='utf-8-sig') as f:
|
||
# ASS头部信息
|
||
f.write("""[Script Info]
|
||
ScriptType: v4.00+
|
||
Collisions: Normal
|
||
PlayResX: 1920
|
||
PlayResY: 1080
|
||
Timer: 100.0000
|
||
WrapStyle: 0
|
||
|
||
[V4+ Styles]
|
||
Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding
|
||
""")
|
||
|
||
# 收集所有唯一的字体名称和关键词分组样式（从字幕中提取）
|
||
unique_fonts = set()
|
||
keyword_group_styles = {} # 存储关键词分组的样式信息
|
||
|
||
for sub in subtitles:
|
||
font_name = sub.get('fontName', 'NotoSerifCJK-VF')
|
||
unique_fonts.add(font_name)
|
||
|
||
# 如果有关键词效果，记录其样式信息
|
||
if sub.get('keywordEffect'):
|
||
keyword_effect = sub.get('keywordEffect')
|
||
group_id = keyword_effect.get('groupId')
|
||
group_name = keyword_effect.get('groupName', '')
|
||
|
||
# 为关键词分组创建唯一的样式
|
||
style_key = f"group_{group_id}" if group_id else None
|
||
if style_key and style_key not in keyword_group_styles:
|
||
# 转换颜色格式：#RRGGBB → &H00BBGGRR&
|
||
font_color = keyword_effect.get('fontColor', '#FFFFFF')
|
||
outline_color = keyword_effect.get('outlineColor', '#000000')
|
||
outline_width = keyword_effect.get('outlineWidth', 2)
|
||
font_size = keyword_effect.get('fontSize', 72)
|
||
|
||
# 转换十六进制颜色
|
||
if font_color.startswith('#'):
|
||
hex_c = font_color.lstrip('#')
|
||
primary_color = f"&H00{hex_c[4:6]}{hex_c[2:4]}{hex_c[0:2]}&"
|
||
else:
|
||
primary_color = "&H00FFFFFF&"
|
||
|
||
if outline_color.startswith('#'):
|
||
hex_c = outline_color.lstrip('#')
|
||
outline_color_ass = f"&H00{hex_c[4:6]}{hex_c[2:4]}{hex_c[0:2]}&"
|
||
else:
|
||
outline_color_ass = "&H00000000&"
|
||
|
||
keyword_group_styles[style_key] = {
|
||
'font_size': font_size,
|
||
'primary_color': primary_color,
|
||
'outline_color': outline_color_ass,
|
||
'outline_width': outline_width,
|
||
'style_def': f'Style: {style_key},{font_name.replace(" ", "_")},{font_size},{primary_color},&H000000FF,{outline_color_ass},&H80000000,-1,0,0,0,100,100,0,0,1,{int(outline_width)},1,2,10,10,20,1'
|
||
}
|
||
|
||
print(f"Found {len(unique_fonts)} unique fonts in subtitles: {unique_fonts}", file=sys.stderr)
|
||
print(f"Found {len(keyword_group_styles)} keyword group styles", file=sys.stderr)
|
||
|
||
# 为每个唯一字体创建样式
|
||
styles = {}
|
||
default_font = 'NotoSerifCJK-VF'
|
||
|
||
# 先添加默认样式（使用默认字体）
|
||
default_font_to_use = list(unique_fonts)[0] if unique_fonts else default_font
|
||
styles['default'] = f'Style: default,{default_font_to_use},24,&H00FFFFFF,&H000000FF,&H00000000,&H80000000,-1,0,0,0,100,100,0,0,1,2,1,2,10,10,20,1'
|
||
|
||
# 为其他唯一字体创建额外样式
|
||
for idx, font_name in enumerate(unique_fonts, 1):
|
||
# 清理字体名称用作样式名（只允许字母和数字）
|
||
style_name = f'font_{idx}' if font_name == default_font_to_use or font_name not in ['NotoSerifCJK-VF', 'NotoSerifCJK-VF', 'NotoSerifCJK-VF'] else font_name.replace(' ', '_')
|
||
styles[style_name] = f'Style: {style_name},{font_name},24,&H00FFFFFF,&H000000FF,&H00000000,&H80000000,-1,0,0,0,100,100,0,0,1,2,1,2,10,10,20,1'
|
||
|
||
# 添加关键词分组的样式
|
||
for style_key, style_info in keyword_group_styles.items():
|
||
styles[style_key] = style_info['style_def']
|
||
|
||
# 输出所有样式定义
|
||
for style_def in styles.values():
|
||
f.write(f"{style_def}\n")
|
||
|
||
f.write("""
|
||
[Events]
|
||
Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
|
||
""")
|
||
|
||
# 添加字幕事件
|
||
for sub in subtitles:
|
||
# 应用关键词分组效果
|
||
sub = self._apply_keyword_effects_to_subtitle(sub)
|
||
|
||
start_time = self._format_ass_time(sub['start_time'])
|
||
end_time = self._format_ass_time(sub['end_time'])
|
||
style = sub.get('style', 'default')
|
||
text = sub['text']
|
||
font_name = sub.get('fontName') # 从字幕对象获取字体名称
|
||
|
||
# ✅ 如果有关键词效果，优先使用关键词分组的样式
|
||
keyword_effect = sub.get('keywordEffect', None)
|
||
if keyword_effect:
|
||
group_id = keyword_effect.get('groupId')
|
||
style_key = f"group_{group_id}" if group_id else None
|
||
if style_key and style_key in styles:
|
||
style = style_key
|
||
print(f"Subtitle '{text[:20]}...' using keyword group style: {style}", file=sys.stderr)
|
||
else:
|
||
print(f"Subtitle '{text[:20]}...' keyword group style '{style_key}' not found, using default", file=sys.stderr)
|
||
# 否则，根据字体名称选择对应的样式
|
||
elif font_name:
|
||
# 尝试使用字体名称作为样式名
|
||
style_for_font = font_name.replace(' ', '_')
|
||
# 如果样式不存在，使用default
|
||
if style_for_font not in styles:
|
||
style = 'default'
|
||
else:
|
||
style = style_for_font
|
||
print(f"Subtitle '{text[:20]}...' using font: {font_name}, style: {style}", file=sys.stderr)
|
||
else:
|
||
print(f"Subtitle '{text[:20]}...' using default font and style", file=sys.stderr)
|
||
|
||
# 如果有关键词效果，添加ASS特效标签和颜色
|
||
if keyword_effect:
|
||
effect_id = keyword_effect.get('effectId', 'none')
|
||
effect_duration = keyword_effect.get('effectDuration', 2.0)
|
||
font_color = keyword_effect.get('fontColor', '#FFFFFF')
|
||
|
||
# 转换十六进制颜色为ASS格式（&HAABBGGRR）
|
||
if font_color.startswith('#'):
|
||
hex_color = font_color.lstrip('#')
|
||
# 反转RGB为BGR
|
||
ass_color = f"&H00{hex_color[4:6]}{hex_color[2:4]}{hex_color[0:2]}&"
|
||
else:
|
||
ass_color = "&H00FFFFFF&"
|
||
|
||
# ✅ 根据前端支持的所有特效类型添加ASS标签
|
||
# 前端支持的特效ID: zoom-in, pulse, bounce, glow, shake, slide-up, fade, spin, wave, flip
|
||
if effect_id in ['bounce', 'pulse', 'pulse_scale']:
|
||
# 弹跳/脉冲：使用缩放变换
|
||
text = f"{{\\t(0,{int(effect_duration*1000)},\\t(0,{int(effect_duration*500)},\\fscx110)\\t({int(effect_duration*500)},{int(effect_duration*1000)},\\fscx100))}}{text}"
|
||
elif effect_id in ['zoom-in']:
|
||
# 缩放弹出
|
||
text = f"{{\\t(0,{int(effect_duration*1000)},\\fscx130\\fscy130)}}{text}"
|
||
elif effect_id in ['glow', 'glow_effect']:
|
||
# 发光效果
|
||
text = f"{{\\c{ass_color}\\blur2\\bord2}}{text}"
|
||
elif effect_id in ['shake']:
|
||
# 抖动效果（使用多个 pos 标签模拟）
|
||
text = f"{{\\fad(100,{int(effect_duration*500)})}}{text}"
|
||
elif effect_id in ['slide-up', 'slide_in_up']:
|
||
# 从下向上滑入
|
||
text = f"{{\\move(0,{int(1080 * 0.3)},0,{int(1080 * 0.2)},0,{int(effect_duration*1000)})}}{text}"
|
||
elif effect_id in ['fade']:
|
||
# 渐变高亮
|
||
text = f"{{\\t(0,{int(effect_duration*500)},\\c{ass_color})\\t({int(effect_duration*500)},{int(effect_duration*1000)},\\c&H00FFFFFF&)}}{text}"
|
||
elif effect_id in ['spin']:
|
||
# 旋转效果
|
||
text = f"{{\\t(0,{int(effect_duration*1000)},\\frz360)}}{text}"
|
||
elif effect_id in ['wave']:
|
||
# 彩虹波浪（使用多个颜色变换）
|
||
text = f"{{\\t(0,{int(effect_duration*1000)},\\c&H0000FF&)\\t({int(effect_duration*250)},{int(effect_duration*500)},\\c&H00FF00&)\\t({int(effect_duration*500)},{int(effect_duration*750)},\\c&HFF0000&)}}{text}"
|
||
elif effect_id in ['flip']:
|
||
# 3D翻转效果
|
||
text = f"{{\\t(0,{int(effect_duration*1000)},\\frx360)}}{text}"
|
||
# 其他特效或 'none' 不添加标签
|
||
|
||
f.write(f"Dialogue: 0,{start_time},{end_time},{style},,0,0,0,,{text}\n")
|
||
|
||
print(f"ASS subtitle file saved: {ass_path}", file=sys.stderr)
|
||
return ass_path
|
||
except Exception as e:
|
||
raise Exception(f"Error saving ASS file: {str(e)}")
|
||
|
||
def _format_ass_time(self, seconds: float) -> str:
|
||
"""格式化时间为ASS格式 (H:MM:SS.cc)"""
|
||
hours = int(seconds // 3600)
|
||
minutes = int((seconds % 3600) // 60)
|
||
secs = int(seconds % 60)
|
||
centisecs = int((seconds % 1) * 100)
|
||
|
||
return f"{hours}:{minutes:02d}:{secs:02d}.{centisecs:02d}"
|
||
|
||
def analyze_script(self) -> Dict[str, Any]:
|
||
"""
|
||
使用轻量级NLP分析文案
|
||
提取标题、关键词、列表项、情感等信息
|
||
|
||
Returns:
|
||
分析结果字典
|
||
"""
|
||
if not NLP_AVAILABLE:
|
||
print("NLP not available, returning basic analysis", file=sys.stderr)
|
||
return {
|
||
'title': None,
|
||
'keywords': [],
|
||
'list_items': [],
|
||
'sentiment': 'neutral'
|
||
}
|
||
|
||
try:
|
||
result = {}
|
||
|
||
# 分割文本为行
|
||
lines = [line.strip() for line in self.script_text.split('\n') if line.strip()]
|
||
|
||
# 1. 提取标题（取第一行，如果长度适中）
|
||
if lines and len(lines[0]) <= 30:
|
||
result['title'] = lines[0]
|
||
else:
|
||
result['title'] = None
|
||
|
||
# 2. 提取列表项（包含"第一"、"第二"、"首先"、"其次"等序号词的行）
|
||
list_markers = ['第一', '第二', '第三', '第四', '第五', '首先', '其次', '再次', '最后', '1.', '2.', '3.', '4.', '5.']
|
||
list_items = []
|
||
for line in lines:
|
||
if any(marker in line for marker in list_markers):
|
||
list_items.append(line)
|
||
result['list_items'] = list_items
|
||
|
||
# 3. 提取关键词（使用TF-IDF）
|
||
full_text = ' '.join(lines)
|
||
try:
|
||
# 使用 jieba 提取关键词（基于 TF-IDF）
|
||
keywords = jieba.analyse.extract_tags(full_text, topK=10, withWeight=False)
|
||
result['keywords'] = keywords
|
||
print(f"Extracted keywords: {keywords}", file=sys.stderr)
|
||
except Exception as e:
|
||
print(f"Keyword extraction failed: {e}", file=sys.stderr)
|
||
result['keywords'] = []
|
||
|
||
# 4. 情感分析（优先使用Transformer，降级到关键词匹配）
|
||
if self.sentiment_analyzer is not None:
|
||
try:
|
||
# 使用Transformer模型进行深度情感分析
|
||
print("Using Transformer for sentiment analysis...", file=sys.stderr)
|
||
# 截取前512个字符（模型限制）
|
||
text_for_analysis = full_text[:512] if len(full_text) > 512 else full_text
|
||
sentiment_result = self.sentiment_analyzer(text_for_analysis)[0]
|
||
|
||
# 转换标签为统一格式
|
||
label = sentiment_result['label'].lower()
|
||
score = sentiment_result['score']
|
||
|
||
if 'pos' in label or 'positive' in label:
|
||
result['sentiment'] = 'positive'
|
||
elif 'neg' in label or 'negative' in label:
|
||
result['sentiment'] = 'negative'
|
||
else:
|
||
result['sentiment'] = 'neutral'
|
||
|
||
result['sentiment_score'] = score
|
||
print(f"Transformer sentiment: {result['sentiment']} (score: {score:.4f})", file=sys.stderr)
|
||
except Exception as e:
|
||
print(f"Transformer sentiment analysis failed: {e}, using fallback", file=sys.stderr)
|
||
# 降级到关键词匹配
|
||
result['sentiment'] = self._fallback_sentiment_analysis(full_text)
|
||
result['sentiment_score'] = 0.5
|
||
else:
|
||
# 降级到关键词匹配
|
||
result['sentiment'] = self._fallback_sentiment_analysis(full_text)
|
||
result['sentiment_score'] = 0.5
|
||
|
||
print(f"Script analysis result: {result}", file=sys.stderr)
|
||
return result
|
||
|
||
except Exception as e:
|
||
print(f"Error analyzing script: {e}", file=sys.stderr)
|
||
return {
|
||
'title': None,
|
||
'keywords': [],
|
||
'list_items': [],
|
||
'sentiment': 'neutral',
|
||
'sentiment_score': 0.5
|
||
}
|
||
|
||
def _fallback_sentiment_analysis(self, text: str) -> str:
|
||
"""基于关键词的简单情感分析（降级方案）"""
|
||
positive_words = ['好', '棒', '优秀', '成功', '喜欢', '开心', '快乐', '美好', '精彩', '完美']
|
||
negative_words = ['差', '糟', '失败', '讨厌', '难过', '痛苦', '糟糕', '失望', '困难']
|
||
|
||
positive_count = sum(1 for word in positive_words if word in text)
|
||
negative_count = sum(1 for word in negative_words if word in text)
|
||
|
||
if positive_count > negative_count:
|
||
return 'positive'
|
||
elif negative_count > positive_count:
|
||
return 'negative'
|
||
else:
|
||
return 'neutral'
|
||
|
||
def analyze_sentence_sentiment(self, text: str) -> Dict[str, Any]:
|
||
"""
|
||
分析单句话的情感
|
||
|
||
Args:
|
||
text: 待分析的句子
|
||
|
||
Returns:
|
||
包含情感和分数的字典
|
||
"""
|
||
# 如果句子太短（少于3个字），返回中性
|
||
if len(text.strip()) < 3:
|
||
return {
|
||
'sentiment': 'neutral',
|
||
'score': 0.5
|
||
}
|
||
|
||
# 优先使用Transformer模型
|
||
if self.sentiment_analyzer is not None:
|
||
try:
|
||
# 截取前512个字符（模型限制）
|
||
text_for_analysis = text[:512] if len(text) > 512 else text
|
||
sentiment_result = self.sentiment_analyzer(text_for_analysis)[0]
|
||
|
||
# 转换标签为统一格式
|
||
label = sentiment_result['label'].lower()
|
||
score = sentiment_result['score']
|
||
|
||
if 'pos' in label or 'positive' in label:
|
||
sentiment = 'positive'
|
||
elif 'neg' in label or 'negative' in label:
|
||
sentiment = 'negative'
|
||
else:
|
||
sentiment = 'neutral'
|
||
|
||
return {
|
||
'sentiment': sentiment,
|
||
'score': score
|
||
}
|
||
except Exception as e:
|
||
print(f"Transformer sentiment analysis failed for sentence: {e}", file=sys.stderr)
|
||
# 降级到关键词匹配
|
||
sentiment = self._fallback_sentiment_analysis(text)
|
||
return {
|
||
'sentiment': sentiment,
|
||
'score': 0.5
|
||
}
|
||
else:
|
||
# 降级到关键词匹配
|
||
sentiment = self._fallback_sentiment_analysis(text)
|
||
return {
|
||
'sentiment': sentiment,
|
||
'score': 0.5
|
||
}
|
||
|
||
def get_style_by_sentiment(self, sentiment: str, score: float) -> str:
|
||
"""
|
||
根据情感选择字幕样式
|
||
|
||
Args:
|
||
sentiment: 情感类型 (positive/negative/neutral)
|
||
score: 情感分数 (0-1)
|
||
|
||
Returns:
|
||
字幕样式名称
|
||
"""
|
||
# 如果用户强制指定了样式，优先使用用户指定的
|
||
if hasattr(self, 'force_style') and self.force_style:
|
||
return self.subtitle_style
|
||
|
||
# 处理emotion-mixed样式：使用情感分析自动选择
|
||
if self.subtitle_style == 'emotion-mixed':
|
||
# 根据情感自动选择样式
|
||
if sentiment == 'positive':
|
||
if score > 0.8: # 强烈积极
|
||
return 'neon' # 霓虹样式 - 明亮彩色
|
||
else:
|
||
return 'highlight' # 高亮样式 - 黄色背景
|
||
elif sentiment == 'negative':
|
||
if score > 0.8: # 强烈消极
|
||
return 'minimal' # 极简样式 - 低调
|
||
else:
|
||
return 'card' # 卡片样式 - 半透明背景
|
||
else:
|
||
return 'default' # 中性 - 默认样式
|
||
|
||
# 普通样式映射
|
||
if sentiment == 'positive':
|
||
if score > 0.8: # 强烈积极
|
||
return 'neon' # 霓虹样式 - 明亮彩色
|
||
else:
|
||
return 'highlight' # 高亮样式 - 黄色背景
|
||
elif sentiment == 'negative':
|
||
if score > 0.8: # 强烈消极
|
||
return 'minimal' # 极简样式 - 低调
|
||
else:
|
||
return 'card' # 卡片样式 - 半透明背景
|
||
else:
|
||
return 'default' # 中性 - 默认样式
|
||
|
||
def burn_subtitles_to_video(self, subtitles: List[Dict[str, Any]]) -> str:
|
||
"""
|
||
使用 FFmpeg 将字幕烧录到视频
|
||
现在支持每个字幕不同的样式（使用ASS格式）
|
||
|
||
Args:
|
||
subtitles: 字幕列表（包含样式信息）
|
||
|
||
Returns:
|
||
输出视频路径
|
||
"""
|
||
try:
|
||
output_video = os.path.join(self.output_dir, 'video_with_subtitles.mp4')
|
||
|
||
# 保存ASS字幕文件（支持多样式）
|
||
ass_path = self.save_ass(subtitles)
|
||
|
||
# Windows上FFmpeg无法正确解析绝对路径中的ASS文件
|
||
# 使用相对路径执行FFmpeg
|
||
ass_filename = os.path.basename(ass_path)
|
||
working_dir = os.path.dirname(ass_path)
|
||
|
||
# 构建 FFmpeg 命令
|
||
# 🔧 修复：使用 ziti 目录的绝对路径，而不是相对路径 '.'
|
||
fontconfig = self._get_ffmpeg_fontconfig()
|
||
print(f"Adding fontconfig to FFmpeg: {fontconfig}", file=sys.stderr)
|
||
|
||
# 构建字幕过滤器参数
|
||
subtitle_filter = f"subtitles={ass_filename}{fontconfig}"
|
||
|
||
cmd = [
|
||
'ffmpeg', '-y',
|
||
'-i', self.video_path,
|
||
'-vf', subtitle_filter,
|
||
'-c:v', 'libx264',
|
||
'-preset', 'medium',
|
||
'-crf', '23',
|
||
'-c:a', 'copy',
|
||
output_video
|
||
]
|
||
|
||
print(f"Running FFmpeg command with ASS subtitles: {' '.join(cmd)}", file=sys.stderr)
|
||
print(f"Working directory: {working_dir}", file=sys.stderr)
|
||
|
||
# 执行 FFmpeg 命令 (使用UTF-8编码避免GBK解码错误，从工作目录执行)
|
||
result = subprocess.run(cmd, capture_output=True, text=True, encoding='utf-8', errors='ignore', timeout=300, cwd=working_dir)
|
||
|
||
if result.returncode == 0 and os.path.exists(output_video):
|
||
print(f"Video with dynamic sentiment-based subtitles generated successfully: {output_video}", file=sys.stderr)
|
||
return output_video
|
||
else:
|
||
print(f"FFmpeg error: {result.stderr}", file=sys.stderr)
|
||
# 尝试使用subtitles过滤器作为fallback
|
||
print("Trying fallback with subtitles filter...", file=sys.stderr)
|
||
return self._burn_with_subtitles_filter(ass_path, output_video)
|
||
|
||
except subprocess.TimeoutExpired:
|
||
print("FFmpeg timeout, trying fallback method...", file=sys.stderr)
|
||
return self._burn_with_srt_fallback(subtitles, output_video)
|
||
except Exception as e:
|
||
print(f"FFmpeg failed: {e}, trying fallback method...", file=sys.stderr)
|
||
return self._burn_with_srt_fallback(subtitles, output_video)
|
||
|
||
def _burn_with_subtitles_filter(self, ass_path: str, output_video: str) -> str:
|
||
"""使用subtitles过滤器的fallback方法"""
|
||
try:
|
||
# 使用相对路径执行FFmpeg
|
||
ass_filename = os.path.basename(ass_path)
|
||
working_dir = os.path.dirname(ass_path)
|
||
|
||
# 构建 FFmpeg 命令（添加字体路径支持）
|
||
# 🔧 修复：使用 ziti 目录的绝对路径，而不是相对路径 '.'
|
||
fontconfig = self._get_ffmpeg_fontconfig()
|
||
print(f"Fallback: Adding fontconfig to FFmpeg: {fontconfig}", file=sys.stderr)
|
||
|
||
# 使用subtitles过滤器而不是ass过滤器
|
||
subtitle_filter = f"subtitles={ass_filename}{fontconfig}"
|
||
|
||
cmd = [
|
||
'ffmpeg', '-y',
|
||
'-i', self.video_path,
|
||
'-vf', subtitle_filter,
|
||
'-c:v', 'libx264',
|
||
'-preset', 'medium',
|
||
'-crf', '23',
|
||
'-c:a', 'copy',
|
||
output_video
|
||
]
|
||
|
||
print(f"Running fallback FFmpeg command: {' '.join(cmd)}", file=sys.stderr)
|
||
result = subprocess.run(cmd, capture_output=True, text=True, encoding='utf-8', errors='ignore', timeout=300, cwd=working_dir)
|
||
|
||
if result.returncode == 0 and os.path.exists(output_video):
|
||
print(f"Fallback method successful: {output_video}", file=sys.stderr)
|
||
return output_video
|
||
else:
|
||
print(f"Fallback method failed: {result.stderr}", file=sys.stderr)
|
||
raise Exception(f"Both FFmpeg methods failed")
|
||
except Exception as e:
|
||
raise Exception(f"FFmpeg fallback also failed: {str(e)}")
|
||
|
||
def _burn_with_srt_fallback(self, subtitles: List[Dict[str, Any]], output_video: str) -> str:
|
||
"""最终fallback：使用SRT格式"""
|
||
try:
|
||
# 保存SRT文件（所有字幕用default样式）
|
||
srt_path = self.save_srt(subtitles)
|
||
|
||
# 使用相对路径执行FFmpeg
|
||
srt_filename = os.path.basename(srt_path)
|
||
working_dir = os.path.dirname(srt_path)
|
||
|
||
# 构建 FFmpeg 命令（添加字体路径支持）
|
||
# 🔧 修复：使用 ziti 目录的绝对路径，而不是相对路径 '.'
|
||
fontconfig = self._get_ffmpeg_fontconfig()
|
||
print(f"SRT Fallback: Adding fontconfig to FFmpeg: {fontconfig}", file=sys.stderr)
|
||
|
||
# 使用subtitles过滤器
|
||
subtitle_filter = f"subtitles={srt_filename}{fontconfig}"
|
||
|
||
cmd = [
|
||
'ffmpeg', '-y',
|
||
'-i', self.video_path,
|
||
'-vf', subtitle_filter,
|
||
'-c:v', 'libx264',
|
||
'-preset', 'medium',
|
||
'-crf', '23',
|
||
'-c:a', 'copy',
|
||
output_video
|
||
]
|
||
|
||
print(f"Running SRT fallback command: {' '.join(cmd)}", file=sys.stderr)
|
||
result = subprocess.run(cmd, capture_output=True, text=True, encoding='utf-8', errors='ignore', timeout=300, cwd=working_dir)
|
||
|
||
if result.returncode == 0 and os.path.exists(output_video):
|
||
print(f"SRT fallback successful: {output_video}", file=sys.stderr)
|
||
return output_video
|
||
else:
|
||
raise Exception(f"All methods failed: {result.stderr}")
|
||
except Exception as e:
|
||
raise Exception(f"SRT fallback failed: {str(e)}")
|
||
|
||
def _get_subtitle_filter(self, srt_path: str) -> str:
|
||
"""
|
||
根据样式生成 FFmpeg 字幕过滤器
|
||
|
||
Args:
|
||
srt_path: SRT文件路径
|
||
|
||
Returns:
|
||
FFmpeg 过滤器字符串
|
||
"""
|
||
# 使用双引号包围 Windows 路径
|
||
safe_path = f'"{srt_path}"'
|
||
|
||
if self.subtitle_style == 'highlight':
|
||
# 高亮样式：黄色背景
|
||
return f"subtitles={safe_path}:force_style='FontName=NotoSerifCJK-VF,FontSize=24,PrimaryColour=&H000000&,BackColour=&H00FFFF&,Bold=1,Outline=0,Shadow=0,MarginV=20'"
|
||
|
||
elif self.subtitle_style == 'card':
|
||
# 卡片样式：带背景框
|
||
return f"subtitles={safe_path}:force_style='FontName=NotoSerifCJK-VF,FontSize=24,PrimaryColour=&HFFFFFF&,BackColour=&H80000000&,Bold=1,Outline=2,Shadow=1,MarginV=20'"
|
||
|
||
elif self.subtitle_style == 'neon':
|
||
# 霓虹样式：彩色描边
|
||
return f"subtitles={safe_path}:force_style='FontName=NotoSerifCJK-VF,FontSize=26,PrimaryColour=&H00FFFF&,OutlineColour=&HFF00FF&,Bold=1,Outline=3,Shadow=0,MarginV=20'"
|
||
|
||
elif self.subtitle_style == 'minimal':
|
||
# 极简样式：细描边
|
||
return f"subtitles={safe_path}:force_style='FontName=NotoSerifCJK-VF,FontSize=22,PrimaryColour=&HFFFFFF&,OutlineColour=&H000000&,Bold=0,Outline=1,Shadow=0,MarginV=20'"
|
||
|
||
else:
|
||
# 默认样式：白色文字，黑色描边
|
||
return f"subtitles={safe_path}:force_style='FontName=NotoSerifCJK-VF,FontSize=24,PrimaryColour=&HFFFFFF&,OutlineColour=&H000000&,Bold=1,Outline=2,Shadow=1,MarginV=20'"
|
||
|
||
def generate_all(self) -> Dict[str, Any]:
|
||
"""
|
||
一键生成所有内容
|
||
|
||
Returns:
|
||
包含生成结果的字典
|
||
"""
|
||
try:
|
||
# 如果仅生成封面模式
|
||
if self.cover_only:
|
||
print("Cover-only mode: Generating cover only...", file=sys.stderr)
|
||
|
||
# 生成封面
|
||
print("Generating cover...", file=sys.stderr)
|
||
cover_path = self.generate_cover()
|
||
|
||
result = {
|
||
'success': True,
|
||
'coverPath': cover_path,
|
||
'message': 'Cover generated successfully'
|
||
}
|
||
|
||
return result
|
||
|
||
# 完整生成模式
|
||
# 分析文案（NLP）
|
||
print("Analyzing script with NLP...", file=sys.stderr)
|
||
nlp_analysis = self.analyze_script()
|
||
|
||
# 生成封面
|
||
print("Generating cover...", file=sys.stderr)
|
||
cover_path = self.generate_cover()
|
||
|
||
# 生成字幕
|
||
print("Generating subtitles...", file=sys.stderr)
|
||
subtitles = self.generate_subtitles()
|
||
|
||
# 烧录字幕到视频（传递字幕列表而不是SRT路径）
|
||
print("Burning subtitles to video...", file=sys.stderr)
|
||
video_path = self.burn_subtitles_to_video(subtitles)
|
||
|
||
result = {
|
||
'success': True,
|
||
'coverPath': cover_path,
|
||
'videoPath': video_path,
|
||
'subtitles': subtitles,
|
||
'nlpAnalysis': nlp_analysis, # 添加NLP分析结果
|
||
'message': 'Generated successfully with sentiment-based subtitles'
|
||
}
|
||
|
||
return result
|
||
except Exception as e:
|
||
return {
|
||
'success': False,
|
||
'error': str(e),
|
||
'message': f'Generation failed: {str(e)}'
|
||
}
|
||
|
||
|
||
def main():
|
||
"""主函数"""
|
||
parser = argparse.ArgumentParser(description='Subtitle & Cover Generator (Simplified)')
|
||
parser.add_argument('config_file', help='Configuration JSON file path')
|
||
args = parser.parse_args()
|
||
|
||
try:
|
||
# 读取配置文件
|
||
with open(args.config_file, 'r', encoding='utf-8') as f:
|
||
config = json.load(f)
|
||
|
||
# 创建生成器
|
||
generator = SubtitleCoverGenerator(config)
|
||
|
||
# 执行生成
|
||
result = generator.generate_all()
|
||
|
||
# 输出结果
|
||
print(json.dumps(result, ensure_ascii=False))
|
||
|
||
sys.exit(0 if result['success'] else 1)
|
||
|
||
except Exception as e:
|
||
error_result = {
|
||
'success': False,
|
||
'error': str(e),
|
||
'message': f'Error: {str(e)}'
|
||
}
|
||
print(json.dumps(error_result, ensure_ascii=False))
|
||
sys.exit(1)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|