11
This commit is contained in:
61
app/api/v1/python_scripts.py
Normal file
61
app/api/v1/python_scripts.py
Normal file
@@ -0,0 +1,61 @@
|
||||
"""封面等 Python 脚本:按相对路径下发源码,供桌面端拉取后在 bundled python.exe 中执行。"""
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
from fastapi import APIRouter
|
||||
|
||||
from app.schemas.common import ApiResponse
|
||||
|
||||
router = APIRouter(prefix="/python-scripts", tags=["python"])
|
||||
|
||||
# pythonbackend/scripts/cover/
|
||||
_SCRIPTS_DIR = Path(__file__).resolve().parents[3] / "scripts" / "cover"
|
||||
|
||||
_SAFE_NAME = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9._/-]*\.py$")
|
||||
|
||||
|
||||
def _validate_name(name: str) -> str | None:
|
||||
raw = name.strip().replace("\\", "/")
|
||||
if not raw or not _SAFE_NAME.fullmatch(raw):
|
||||
return None
|
||||
if ".." in raw.split("/"):
|
||||
return None
|
||||
return raw
|
||||
|
||||
|
||||
def _resolve_script_path(safe: str) -> Path | None:
|
||||
path = _SCRIPTS_DIR / safe
|
||||
try:
|
||||
base_resolved = _SCRIPTS_DIR.resolve()
|
||||
resolved = path.resolve()
|
||||
resolved.relative_to(base_resolved)
|
||||
except (OSError, ValueError):
|
||||
return None
|
||||
if not path.is_file():
|
||||
return None
|
||||
return path
|
||||
|
||||
|
||||
@router.get("", response_model=ApiResponse[dict])
|
||||
async def get_python_script(name: str) -> ApiResponse[dict]:
|
||||
safe = _validate_name(name)
|
||||
if not safe:
|
||||
return ApiResponse(ok=False, message="无效的脚本名称", data=None)
|
||||
path = _resolve_script_path(safe)
|
||||
if path is None:
|
||||
return ApiResponse(ok=False, message=f"脚本不存在: {name}", data=None)
|
||||
source = path.read_text(encoding="utf-8")
|
||||
return ApiResponse(ok=True, message="", data={"source": source})
|
||||
|
||||
|
||||
@router.get("/list", response_model=ApiResponse[dict])
|
||||
async def list_python_scripts() -> ApiResponse[dict]:
|
||||
if not _SCRIPTS_DIR.is_dir():
|
||||
return ApiResponse(ok=True, message="", data={"names": []})
|
||||
names: list[str] = []
|
||||
for p in sorted(_SCRIPTS_DIR.rglob("*.py")):
|
||||
if p.is_file():
|
||||
rel = p.relative_to(_SCRIPTS_DIR).as_posix()
|
||||
names.append(rel)
|
||||
return ApiResponse(ok=True, message="", data={"names": names})
|
||||
@@ -17,6 +17,7 @@ from app.api.v1 import (
|
||||
oem_users,
|
||||
auth,
|
||||
nodejs_scripts,
|
||||
python_scripts,
|
||||
quickjs_scripts,
|
||||
)
|
||||
|
||||
@@ -37,3 +38,4 @@ api_router.include_router(oem_oem_branding.router)
|
||||
api_router.include_router(app_config.router)
|
||||
api_router.include_router(quickjs_scripts.router)
|
||||
api_router.include_router(nodejs_scripts.router)
|
||||
api_router.include_router(python_scripts.router)
|
||||
|
||||
328
scripts/cover/cover_generate.py
Normal file
328
scripts/cover/cover_generate.py
Normal file
@@ -0,0 +1,328 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
桌面端封面生成入口(由 Tauri pythonscript runner 调用 __python_main)。
|
||||
优先高级模板 → PIL → ffmpeg drawtext 回退。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
SCRIPT_DIR = Path(__file__).resolve().parent
|
||||
if str(SCRIPT_DIR) not in sys.path:
|
||||
sys.path.insert(0, str(SCRIPT_DIR))
|
||||
|
||||
OUTPUT_DIR = Path(tempfile.gettempdir()) / "aiclient-cover"
|
||||
IMAGE_EXT = {".jpg", ".jpeg", ".png", ".bmp", ".webp"}
|
||||
|
||||
|
||||
def _emit_progress(status: str) -> None:
|
||||
native = globals().get("__native")
|
||||
if native is not None and hasattr(native, "emit_progress"):
|
||||
native.emit_progress(status)
|
||||
|
||||
|
||||
def _locate_ffmpeg() -> str | None:
|
||||
env = os.environ.get("AICLIENT_FFMPEG_PATH", "").strip()
|
||||
if env and os.path.isfile(env):
|
||||
return env
|
||||
return None
|
||||
|
||||
|
||||
def _normalize_path(p: str) -> str:
|
||||
return str(p).replace("\\", "/")
|
||||
|
||||
|
||||
def _is_image(path: str) -> bool:
|
||||
return Path(path).suffix.lower() in IMAGE_EXT
|
||||
|
||||
|
||||
def _ensure_output_dir() -> Path:
|
||||
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
|
||||
return OUTPUT_DIR
|
||||
|
||||
|
||||
def _exec_ffmpeg(args: list[str]) -> None:
|
||||
ffmpeg = _locate_ffmpeg()
|
||||
if not ffmpeg:
|
||||
raise RuntimeError(
|
||||
"未找到 ffmpeg,请配置 AICLIENT_FFMPEG_PATH 或 src-tauri/binaries/ffmpeg.exe"
|
||||
)
|
||||
r = subprocess.run(
|
||||
[ffmpeg, *args],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
encoding="utf-8",
|
||||
errors="replace",
|
||||
)
|
||||
if r.returncode != 0:
|
||||
msg = (r.stderr or r.stdout or "").strip() or f"ffmpeg 退出码 {r.returncode}"
|
||||
raise RuntimeError(msg)
|
||||
|
||||
|
||||
def _locate_cjk_font() -> str | None:
|
||||
scripts_dir = os.environ.get("AICLIENT_COVER_SCRIPTS_DIR") or str(SCRIPT_DIR)
|
||||
windir = os.environ.get("WINDIR", r"C:\Windows")
|
||||
candidates = [
|
||||
os.path.join(scripts_dir, "fonts", "simhei.ttf"),
|
||||
os.path.join(scripts_dir, "fonts", "msyh.ttc"),
|
||||
os.path.join(windir, "Fonts", "simhei.ttf"),
|
||||
os.path.join(windir, "Fonts", "msyh.ttc"),
|
||||
os.path.join(windir, "Fonts", "simsun.ttc"),
|
||||
]
|
||||
for p in candidates:
|
||||
if p and os.path.isfile(p):
|
||||
return p
|
||||
return None
|
||||
|
||||
|
||||
def _prepare_frame(source_path: str, temp_frame_path: str) -> None:
|
||||
if _is_image(source_path):
|
||||
_emit_progress("正在加载图片素材…")
|
||||
_exec_ffmpeg(
|
||||
[
|
||||
"-i",
|
||||
source_path,
|
||||
"-vf",
|
||||
"scale=1080:1920:force_original_aspect_ratio=decrease,"
|
||||
"pad=1080:1920:(ow-iw)/2:(oh-ih)/2:color=black",
|
||||
"-y",
|
||||
temp_frame_path,
|
||||
]
|
||||
)
|
||||
return
|
||||
_emit_progress("正在提取视频帧…")
|
||||
_exec_ffmpeg(
|
||||
[
|
||||
"-ss",
|
||||
"00:00:03",
|
||||
"-i",
|
||||
source_path,
|
||||
"-vframes",
|
||||
"1",
|
||||
"-pix_fmt",
|
||||
"yuvj420p",
|
||||
"-c:v",
|
||||
"mjpeg",
|
||||
"-strict:v",
|
||||
"unofficial",
|
||||
"-f",
|
||||
"image2",
|
||||
"-y",
|
||||
temp_frame_path,
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
def _escape_ffmpeg_path(file_path: str) -> str:
|
||||
return file_path.replace("\\", "/").replace(":", "\\:")
|
||||
|
||||
|
||||
def _run_ffmpeg_cover(
|
||||
source_video: str,
|
||||
title_text: str,
|
||||
effect_style: dict,
|
||||
cover_path: str,
|
||||
temp_frame_path: str,
|
||||
) -> None:
|
||||
_prepare_frame(source_video, temp_frame_path)
|
||||
title_font_size = effect_style.get("titleFontSize") or 90
|
||||
title_font_color = effect_style.get("titleFontColor") or effect_style.get("titleColor") or "#FFFFFF"
|
||||
title_position = effect_style.get("titlePosition")
|
||||
if isinstance(title_position, dict):
|
||||
text_pos = "y=h-text_h-50"
|
||||
elif title_position == "top":
|
||||
text_pos = "y=50"
|
||||
elif title_position == "center":
|
||||
text_pos = "y=(h-text_h)/2"
|
||||
else:
|
||||
text_pos = "y=h-text_h-50"
|
||||
|
||||
font_file = _locate_cjk_font()
|
||||
if not font_file:
|
||||
raise RuntimeError(
|
||||
"未找到中文字体文件,无法使用 ffmpeg 绘制标题"
|
||||
)
|
||||
|
||||
title_file = OUTPUT_DIR / f"cover_title_{int(time.time() * 1000)}.txt"
|
||||
title_file.write_text(str(title_text), encoding="utf-8")
|
||||
try:
|
||||
font_opt = f"fontfile='{_escape_ffmpeg_path(font_file)}'"
|
||||
text_opt = f"textfile='{_escape_ffmpeg_path(str(title_file))}'"
|
||||
vf = (
|
||||
f"drawtext={font_opt}:{text_opt}:fontsize={title_font_size}:"
|
||||
f"fontcolor={title_font_color}:x=(w-text_w)/2:{text_pos}:"
|
||||
"shadowcolor=black:shadowx=2:shadowy=2"
|
||||
)
|
||||
_emit_progress("正在叠加标题文字(ffmpeg)…")
|
||||
_exec_ffmpeg(["-i", temp_frame_path, "-vf", vf, "-y", cover_path])
|
||||
finally:
|
||||
try:
|
||||
title_file.unlink(missing_ok=True)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
def _is_advanced_cover_config(config: dict | None) -> bool:
|
||||
if not config or not isinstance(config, dict):
|
||||
return False
|
||||
return (
|
||||
config.get("blurBackground") is not None
|
||||
or config.get("extractPerson") is not None
|
||||
or config.get("personOutlineColor") is not None
|
||||
or config.get("personOutlineWidth") is not None
|
||||
or config.get("maskImagePath") is not None
|
||||
or bool(config.get("titleFontFamily"))
|
||||
or config.get("titleBackgroundEnabled") is not None
|
||||
or config.get("personSize") is not None
|
||||
or config.get("backgroundBlurEnabled") is not None
|
||||
)
|
||||
|
||||
|
||||
def _should_use_pil_cover(effect_style: dict) -> bool:
|
||||
if _is_advanced_cover_config(effect_style):
|
||||
return False
|
||||
return (
|
||||
bool(effect_style.get("titleStrokeWidth"))
|
||||
or bool(effect_style.get("titleFontFamily"))
|
||||
or bool(effect_style.get("titleColor"))
|
||||
or (effect_style.get("titleShadowBlur") or 0) > 0
|
||||
)
|
||||
|
||||
|
||||
def _run_advanced(
|
||||
video_path: str,
|
||||
title_text: str,
|
||||
output_path: str,
|
||||
config: dict,
|
||||
) -> dict:
|
||||
from advanced_cover_generator import AdvancedCoverGenerator
|
||||
|
||||
_emit_progress("正在使用高级封面生成器…")
|
||||
merged = {
|
||||
**config,
|
||||
"output": output_path,
|
||||
"video": video_path,
|
||||
"title": title_text,
|
||||
}
|
||||
generator = AdvancedCoverGenerator(merged)
|
||||
result = generator.generate()
|
||||
if not result.get("success"):
|
||||
return {
|
||||
"success": False,
|
||||
"error": result.get("error") or result.get("message") or "高级封面生成失败",
|
||||
}
|
||||
if not os.path.isfile(output_path):
|
||||
return {"success": False, "error": "封面文件未生成"}
|
||||
return {
|
||||
"success": True,
|
||||
"coverPath": result.get("coverPath") or output_path,
|
||||
"previewImages": result.get("previewImages") or [],
|
||||
"message": result.get("message") or "高级封面生成完成",
|
||||
"mode": "advanced_python",
|
||||
}
|
||||
|
||||
|
||||
def _run_pil(
|
||||
video_path: str,
|
||||
title_text: str,
|
||||
output_path: str,
|
||||
config: dict,
|
||||
timestamp: float = 3,
|
||||
) -> dict:
|
||||
import cover_generator as cg
|
||||
|
||||
_emit_progress("正在使用 PIL 封面生成器…")
|
||||
frame = cg.extract_frame_from_video(video_path, timestamp)
|
||||
if frame is None:
|
||||
return {"success": False, "error": "无法从视频提取帧"}
|
||||
width, height = frame.size
|
||||
max_width = config.get("maxWidth", 1920)
|
||||
if width > max_width:
|
||||
scale = max_width / width
|
||||
frame = frame.resize(
|
||||
(max_width, int(height * scale)),
|
||||
cg.Image.Resampling.LANCZOS,
|
||||
)
|
||||
result = cg.add_text_to_image(frame, title_text, config)
|
||||
out_dir = os.path.dirname(output_path)
|
||||
if out_dir:
|
||||
os.makedirs(out_dir, exist_ok=True)
|
||||
result.save(output_path, quality=95)
|
||||
if not os.path.isfile(output_path):
|
||||
return {"success": False, "error": "封面文件未生成"}
|
||||
return {
|
||||
"success": True,
|
||||
"coverPath": output_path,
|
||||
"message": "封面生成完成",
|
||||
"mode": "pil_python",
|
||||
}
|
||||
|
||||
|
||||
def __python_main(params: dict | None) -> dict:
|
||||
p = params or {}
|
||||
video_path = str(p.get("videoPath") or "").strip()
|
||||
title_text = str(p.get("titleText") or "").strip()
|
||||
effect_style = p.get("effectStyle") or {}
|
||||
if not isinstance(effect_style, dict):
|
||||
effect_style = {}
|
||||
|
||||
if not video_path or not os.path.isfile(video_path):
|
||||
return {
|
||||
"success": False,
|
||||
"error": "视频文件不存在,请先在步骤 02/03/05 生成视频",
|
||||
}
|
||||
if not title_text:
|
||||
return {
|
||||
"success": False,
|
||||
"error": "标题文本不能为空,请先在步骤 04 生成标题",
|
||||
}
|
||||
|
||||
_ensure_output_dir()
|
||||
ts = int(time.time() * 1000)
|
||||
temp_frame_path = str(OUTPUT_DIR / f"temp_frame_{ts}.jpg")
|
||||
cover_path = str(OUTPUT_DIR / f"cover_{ts}.jpg")
|
||||
|
||||
source_video = video_path
|
||||
custom = str(effect_style.get("customVideoPath") or "").strip()
|
||||
if custom and os.path.isfile(custom):
|
||||
source_video = custom
|
||||
|
||||
try:
|
||||
if _is_advanced_cover_config(effect_style):
|
||||
adv = _run_advanced(source_video, title_text, cover_path, effect_style)
|
||||
if adv.get("success"):
|
||||
return adv
|
||||
_emit_progress(
|
||||
f"高级生成失败,尝试回退:{str(adv.get('error') or '')[:80]}"
|
||||
)
|
||||
elif _should_use_pil_cover(effect_style):
|
||||
pil = _run_pil(source_video, title_text, cover_path, effect_style)
|
||||
if pil.get("success"):
|
||||
return pil
|
||||
_emit_progress("PIL 生成失败,尝试 ffmpeg 回退…")
|
||||
|
||||
_run_ffmpeg_cover(
|
||||
source_video, title_text, effect_style, cover_path, temp_frame_path
|
||||
)
|
||||
if os.path.isfile(temp_frame_path):
|
||||
try:
|
||||
os.unlink(temp_frame_path)
|
||||
except OSError:
|
||||
pass
|
||||
if not os.path.isfile(cover_path):
|
||||
return {"success": False, "error": "封面文件未生成"}
|
||||
return {
|
||||
"success": True,
|
||||
"coverPath": cover_path,
|
||||
"message": "封面生成完成",
|
||||
"mode": "ffmpeg",
|
||||
}
|
||||
except Exception as e:
|
||||
return {"success": False, "error": str(e) or repr(e)}
|
||||
34
scripts/cover/modules/extractor.py
Normal file
34
scripts/cover/modules/extractor.py
Normal file
@@ -0,0 +1,34 @@
|
||||
"""
|
||||
视频帧提取(供 modules.segment 的 segment_person_from_video_frame 使用)
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import cv2
|
||||
import numpy as np
|
||||
|
||||
|
||||
class VideoExtractor:
|
||||
"""从视频中按时间戳抽取一帧(RGB numpy 数组)。"""
|
||||
|
||||
def __init__(self, video_path: str) -> None:
|
||||
self.video_path = video_path
|
||||
self._cap = cv2.VideoCapture(video_path)
|
||||
if not self._cap.isOpened():
|
||||
raise RuntimeError(f"无法打开视频: {video_path}")
|
||||
|
||||
def extract_frame_at_time(self, time_seconds: float) -> np.ndarray:
|
||||
self._cap.set(cv2.CAP_PROP_POS_MSEC, max(0.0, float(time_seconds)) * 1000.0)
|
||||
ret, frame = self._cap.read()
|
||||
if not ret or frame is None:
|
||||
self._cap.release()
|
||||
raise RuntimeError(f"无法在 {time_seconds}s 处提取视频帧: {self.video_path}")
|
||||
rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
|
||||
self._cap.release()
|
||||
return rgb
|
||||
|
||||
def __del__(self) -> None:
|
||||
try:
|
||||
if hasattr(self, "_cap") and self._cap is not None:
|
||||
self._cap.release()
|
||||
except Exception:
|
||||
pass
|
||||
Reference in New Issue
Block a user