diff --git a/app/api/v1/python_scripts.py b/app/api/v1/python_scripts.py new file mode 100644 index 0000000..a9e8e87 --- /dev/null +++ b/app/api/v1/python_scripts.py @@ -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}) diff --git a/app/api/v1/router.py b/app/api/v1/router.py index 95898a2..e384099 100644 --- a/app/api/v1/router.py +++ b/app/api/v1/router.py @@ -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) diff --git a/scripts/cover/cover_generate.py b/scripts/cover/cover_generate.py new file mode 100644 index 0000000..157ffe4 --- /dev/null +++ b/scripts/cover/cover_generate.py @@ -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)} diff --git a/scripts/cover/modules/extractor.py b/scripts/cover/modules/extractor.py new file mode 100644 index 0000000..ed6c104 --- /dev/null +++ b/scripts/cover/modules/extractor.py @@ -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