import asyncio import json import os import re from datetime import datetime from typing import Any, Callable from loguru import logger from app.config import config from app.services.documentary.frame_analysis_models import FrameBatchResult from app.services.generate_narration_script import generate_narration, parse_frame_analysis_to_markdown from app.services.llm.migration_adapter import create_vision_analyzer from app.utils import utils, video_processor class DocumentaryFrameAnalysisService: PROMPT_TEMPLATE = """ 我提供了 {frame_count} 张视频帧，它们按时间顺序排列，代表一个连续的视频片段。 首先，请详细描述每一帧的关键视觉信息（包含：主要内容、人物、动作和场景）。 然后，基于所有帧的分析，请用简洁的语言总结整个视频片段中发生的主要活动或事件流程。 请务必使用 JSON 格式输出。 JSON 必须包含以下键： - frame_observations: 数组，且长度必须为 {frame_count} - overall_activity_summary: 字符串，描述整个批次主要活动 示例结构： {{ "frame_observations": [ {{"timestamp": "00:00:00,000", "observation": "画面描述"}} ], "overall_activity_summary": "本批次主要活动总结" }} 请务必不要遗漏视频帧，我提供了 {frame_count} 张视频帧，frame_observations 必须包含 {frame_count} 个元素 请只返回 JSON 字符串，不要附加解释文字。 """.strip() async def generate_documentary_script( self, *, video_path: str, video_theme: str = "", custom_prompt: str = "", frame_interval_input: int | float | None = None, vision_batch_size: int | None = None, vision_llm_provider: str | None = None, progress_callback: Callable[[float, str], None] | None = None, vision_api_key: str | None = None, vision_model_name: str | None = None, vision_base_url: str | None = None, max_concurrency: int | None = None, ) -> list[dict]: progress = progress_callback or (lambda _p, _m: None) analysis_result = await self.analyze_video( video_path=video_path, video_theme=video_theme, custom_prompt=custom_prompt, frame_interval_input=frame_interval_input, vision_batch_size=vision_batch_size, vision_llm_provider=vision_llm_provider, progress_callback=progress_callback, vision_api_key=vision_api_key, vision_model_name=vision_model_name, vision_base_url=vision_base_url, max_concurrency=max_concurrency, ) analysis_json_path = analysis_result["analysis_json_path"] progress(80, "正在生成解说文案...") text_provider = config.app.get("text_llm_provider", "openai").lower() text_api_key = config.app.get(f"text_{text_provider}_api_key") text_model = config.app.get(f"text_{text_provider}_model_name") text_base_url = config.app.get(f"text_{text_provider}_base_url") if not text_api_key or not text_model: raise ValueError( f"未配置 {text_provider} 的文本模型参数。" f"请在设置中配置 text_{text_provider}_api_key 和 text_{text_provider}_model_name" ) markdown_output = parse_frame_analysis_to_markdown(analysis_json_path) narration_input = self._build_narration_input( markdown_output=markdown_output, video_theme=video_theme, custom_prompt=custom_prompt, ) narration_raw = generate_narration( narration_input, text_api_key, base_url=text_base_url, model=text_model, ) narration_items = self._parse_narration_items(narration_raw) final_script = [{**item, "OST": 2} for item in narration_items] progress(100, "脚本生成完成") return final_script async def analyze_video( self, *, video_path: str, video_theme: str = "", custom_prompt: str = "", frame_interval_input: int | float | None = None, vision_batch_size: int | None = None, vision_llm_provider: str | None = None, progress_callback: Callable[[float, str], None] | None = None, vision_api_key: str | None = None, vision_model_name: str | None = None, vision_base_url: str | None = None, max_concurrency: int | None = None, ) -> dict[str, Any]: progress = progress_callback or (lambda _p, _m: None) if not video_path or not os.path.exists(video_path): raise FileNotFoundError(f"视频文件不存在: {video_path}") frame_interval_seconds = self._resolve_frame_interval(frame_interval_input) batch_size = self._resolve_batch_size(vision_batch_size) concurrency = self._resolve_max_concurrency(max_concurrency) provider = (vision_llm_provider or config.app.get("vision_llm_provider", "openai")).lower() api_key = vision_api_key if vision_api_key is not None else config.app.get(f"vision_{provider}_api_key") model_name = ( vision_model_name if vision_model_name is not None else config.app.get(f"vision_{provider}_model_name") ) base_url = vision_base_url if vision_base_url is not None else config.app.get(f"vision_{provider}_base_url", "") if not api_key or not model_name: raise ValueError( f"未配置 {provider} 的 API Key 或模型名称。" f"请在设置中配置 vision_{provider}_api_key 和 vision_{provider}_model_name" ) progress(10, "正在提取关键帧...") keyframe_files = self._load_or_extract_keyframes(video_path, frame_interval_seconds) progress(25, f"关键帧准备完成，共 {len(keyframe_files)} 帧") progress(30, "正在初始化视觉分析器...") analyzer = create_vision_analyzer( provider=provider, api_key=api_key, model=model_name, base_url=base_url, ) batches = self._chunk_keyframes(keyframe_files, batch_size=batch_size) if not batches: raise RuntimeError("未能构建任何关键帧批次") progress(40, f"正在分析关键帧，共 {len(batches)} 个批次...") batch_results = await self._analyze_batches( analyzer=analyzer, batches=batches, custom_prompt=custom_prompt, video_theme=video_theme, max_concurrency=concurrency, progress_callback=progress, ) progress(65, "正在整理分析结果...") sorted_batches = self._sort_batch_results(batch_results) artifact = self._build_analysis_artifact( sorted_batches, video_path=video_path, frame_interval_seconds=frame_interval_seconds, vision_batch_size=batch_size, vision_llm_provider=provider, vision_model_name=model_name, max_concurrency=concurrency, ) analysis_json_path = self._save_analysis_artifact(artifact) video_clip_json = self._build_video_clip_json(sorted_batches) progress(75, "逐帧分析完成") return { "analysis_json_path": analysis_json_path, "analysis_artifact": artifact, "video_clip_json": video_clip_json, "keyframe_files": keyframe_files, } def _parse_narration_items(self, narration_raw: str) -> list[dict[str, Any]]: parsed = self._repair_narration_payload(narration_raw) items: list[dict[str, Any]] = [] if isinstance(parsed, dict): raw_items = parsed.get("items") if isinstance(raw_items, list): items = [item for item in raw_items if isinstance(item, dict)] if not items: raise ValueError("解说文案格式错误，无法解析JSON或缺少items字段") return items def _build_narration_input(self, *, markdown_output: str, video_theme: str, custom_prompt: str) -> str: context_lines: list[str] = [] if (video_theme or "").strip(): context_lines.append(f"视频主题：{video_theme.strip()}") if (custom_prompt or "").strip(): context_lines.append(f"补充创作要求：{custom_prompt.strip()}") if not context_lines: return markdown_output context_block = "\n".join(f"- {line}" for line in context_lines) return f"{markdown_output.rstrip()}\n\n## 创作上下文\n{context_block}\n" def _repair_narration_payload(self, narration_raw: str) -> dict[str, Any] | None: def load_json_candidate(payload: str) -> dict[str, Any] | None: try: parsed = json.loads(payload) return parsed if isinstance(parsed, dict) else None except Exception: return None cleaned = (narration_raw or "").strip() if not cleaned: return None candidates: list[str] = [cleaned] candidates.append(cleaned.replace("{{", "{").replace("}}", "}")) json_block = re.search(r"```json\s*(.*?)\s*```", cleaned, re.DOTALL) if json_block: candidates.append(json_block.group(1).strip()) start = cleaned.find("{") end = cleaned.rfind("}") if start >= 0 and end > start: candidates.append(cleaned[start : end + 1]) for candidate in candidates: parsed = load_json_candidate(candidate) if parsed is not None: return parsed fixed = cleaned.replace("{{", "{").replace("}}", "}") fixed_start = fixed.find("{") fixed_end = fixed.rfind("}") if fixed_start >= 0 and fixed_end > fixed_start: fixed = fixed[fixed_start : fixed_end + 1] fixed = re.sub(r"^\s*#.*$", "", fixed, flags=re.MULTILINE) fixed = re.sub(r"^\s*//.*$", "", fixed, flags=re.MULTILINE) fixed = re.sub(r",\s*}", "}", fixed) fixed = re.sub(r",\s*]", "]", fixed) fixed = re.sub(r"'([^']*)'\s*:", r'"\1":', fixed) fixed = re.sub(r'([{\[,]\s*)([A-Za-z_][\w\u4e00-\u9fff]*)(\s*:)', r'\1"\2"\3', fixed) fixed = re.sub(r'""([^"]*?)""', r'"\1"', fixed) return load_json_candidate(fixed) def _resolve_frame_interval(self, frame_interval_input: int | float | None) -> float: interval = frame_interval_input if interval in (None, ""): interval = config.frames.get("frame_interval_input", 3) try: value = float(interval) except (TypeError, ValueError): value = 3.0 if value <= 0: raise ValueError("frame_interval_input must be > 0") return value def _resolve_batch_size(self, vision_batch_size: int | None) -> int: size = vision_batch_size or config.frames.get("vision_batch_size", 10) try: value = int(size) except (TypeError, ValueError): value = 10 if value <= 0: raise ValueError("vision_batch_size must be > 0") return value def _resolve_max_concurrency(self, max_concurrency: int | None) -> int: value = max_concurrency if max_concurrency is not None else config.frames.get("vision_max_concurrency", 2) try: parsed = int(value) except (TypeError, ValueError): parsed = 1 return max(1, parsed) def _load_or_extract_keyframes(self, video_path: str, frame_interval_seconds: float) -> list[str]: keyframes_root = os.path.join(utils.temp_dir(), "keyframes") os.makedirs(keyframes_root, exist_ok=True) cache_key = self._build_keyframe_cache_key(video_path, frame_interval_seconds) cache_dir = os.path.join(keyframes_root, cache_key) os.makedirs(cache_dir, exist_ok=True) cached_files = self._collect_keyframe_paths(cache_dir) if cached_files: logger.info(f"使用已缓存关键帧: {cache_dir}, 共 {len(cached_files)} 帧") return cached_files processor = video_processor.VideoProcessor(video_path) extracted = processor.extract_frames_by_interval_with_fallback( output_dir=cache_dir, interval_seconds=frame_interval_seconds, ) keyframe_files = sorted(str(path) for path in extracted if str(path).endswith(".jpg")) if not keyframe_files: keyframe_files = self._collect_keyframe_paths(cache_dir) if not keyframe_files: raise RuntimeError("未提取到任何关键帧") logger.info(f"关键帧提取完成: {cache_dir}, 共 {len(keyframe_files)} 帧") return keyframe_files def _build_keyframe_cache_key(self, video_path: str, frame_interval_seconds: float) -> str: try: video_mtime = os.path.getmtime(video_path) except OSError: video_mtime = 0 legacy_prefix = utils.md5(f"{video_path}{video_mtime}") payload = "|".join( [ str(video_path), str(video_mtime), str(frame_interval_seconds), "documentary-keyframes-v2", ] ) return f"{legacy_prefix}_{utils.md5(payload)}" @staticmethod def _collect_keyframe_paths(cache_dir: str) -> list[str]: if not os.path.exists(cache_dir): return [] return sorted( os.path.join(cache_dir, name) for name in os.listdir(cache_dir) if re.fullmatch(r"keyframe_\d{6}_\d{9}\.jpg", name) ) @staticmethod def _chunk_keyframes(keyframe_files: list[str], batch_size: int) -> list[list[str]]: return [keyframe_files[index : index + batch_size] for index in range(0, len(keyframe_files), batch_size)] async def _analyze_batches( self, *, analyzer: Any, batches: list[list[str]], custom_prompt: str, video_theme: str, max_concurrency: int, progress_callback: Callable[[float, str], None], ) -> list[FrameBatchResult]: semaphore = asyncio.Semaphore(max(1, max_concurrency)) total = len(batches) done = 0 done_lock = asyncio.Lock() batch_time_ranges: list[str] = [] previous_batch_files: list[str] | None = None for batch_files in batches: _, _, time_range = self._get_batch_timestamps(batch_files, previous_batch_files) batch_time_ranges.append(time_range) previous_batch_files = batch_files async def run_single(batch_index: int, frame_paths: list[str], time_range: str) -> FrameBatchResult: nonlocal done prompt = self._build_batch_prompt( frame_count=len(frame_paths), video_theme=video_theme, custom_prompt=custom_prompt, ) try: async with semaphore: raw_results = await analyzer.analyze_images( images=frame_paths, prompt=prompt, batch_size=max(1, len(frame_paths)), max_concurrency=1, ) raw_response, error_message = self._extract_batch_response(raw_results) if error_message: return self._build_failed_batch_result( batch_index=batch_index, raw_response=raw_response, error_message=error_message, frame_paths=frame_paths, time_range=time_range, ) return self._parse_batch_response( batch_index=batch_index, raw_response=raw_response, frame_paths=frame_paths, time_range=time_range, ) except Exception as exc: return self._build_failed_batch_result( batch_index=batch_index, raw_response="", error_message=str(exc), frame_paths=frame_paths, time_range=time_range, ) finally: async with done_lock: done += 1 progress = 40 + (done / max(1, total)) * 25 progress_callback(progress, f"正在分析关键帧批次 ({done}/{total})...") tasks = [ run_single(batch_index=index, frame_paths=batch_files, time_range=batch_time_ranges[index]) for index, batch_files in enumerate(batches) ] return await asyncio.gather(*tasks) def _build_batch_prompt(self, *, frame_count: int, video_theme: str, custom_prompt: str) -> str: prompt = self._build_analysis_prompt(frame_count=frame_count) extra_lines: list[str] = [] if (video_theme or "").strip(): extra_lines.append(f"视频主题：{video_theme.strip()}") if (custom_prompt or "").strip(): extra_lines.append(custom_prompt.strip()) if not extra_lines: return prompt extras = "\n".join(f"- {line}" for line in extra_lines) return f"{prompt}\n\n补充分析要求：\n{extras}" def _extract_batch_response(self, raw_results: Any) -> tuple[str, str]: if not raw_results: return "", "Batch response is empty" first_result = raw_results[0] if isinstance(raw_results, list) else raw_results if isinstance(first_result, dict): raw_response = str(first_result.get("response", "") or "") error_message = str(first_result.get("error", "") or "") if error_message: if not raw_response: raw_response = error_message return raw_response, error_message if not raw_response.strip(): return raw_response, "Batch response is empty" return raw_response, "" raw_response = str(first_result or "") if not raw_response.strip(): return raw_response, "Batch response is empty" return raw_response, "" def _sort_batch_results(self, batch_results: list[FrameBatchResult]) -> list[FrameBatchResult]: return sorted(batch_results, key=lambda item: (self._time_range_sort_key(item.time_range), item.batch_index)) def _build_analysis_artifact( self, batch_results: list[FrameBatchResult], *, video_path: str, frame_interval_seconds: float, vision_batch_size: int, vision_llm_provider: str, vision_model_name: str, max_concurrency: int, ) -> dict[str, Any]: sorted_batches = self._sort_batch_results(batch_results) batch_dicts: list[dict[str, Any]] = [] frame_observations: list[dict[str, Any]] = [] overall_activity_summaries: list[dict[str, Any]] = [] for batch in sorted_batches: batch_payload = { "batch_index": batch.batch_index, "status": batch.status, "time_range": batch.time_range, "raw_response": batch.raw_response, "frame_paths": list(batch.frame_paths), "frame_observations": list(batch.frame_observations), "overall_activity_summary": batch.overall_activity_summary, "fallback_summary": batch.fallback_summary, "error_message": batch.error_message, } batch_dicts.append(batch_payload) for observation in batch.frame_observations: observation_payload = dict(observation) observation_payload["batch_index"] = batch.batch_index observation_payload["time_range"] = batch.time_range frame_observations.append(observation_payload) summary_text = (batch.overall_activity_summary or batch.fallback_summary or "").strip() if summary_text: overall_activity_summaries.append( { "batch_index": batch.batch_index, "time_range": batch.time_range, "summary": summary_text, } ) return { "artifact_version": "documentary-frame-analysis-v2", "generated_at": datetime.now().isoformat(), "video_path": video_path, "frame_interval_seconds": frame_interval_seconds, "vision_batch_size": vision_batch_size, "vision_llm_provider": vision_llm_provider, "vision_model_name": vision_model_name, "vision_max_concurrency": max_concurrency, "batches": batch_dicts, # 向后兼容旧解析器结构 "frame_observations": frame_observations, "overall_activity_summaries": overall_activity_summaries, } def _save_analysis_artifact(self, artifact: dict[str, Any]) -> str: analysis_dir = os.path.join(utils.storage_dir(), "temp", "analysis") os.makedirs(analysis_dir, exist_ok=True) filename = f"frame_analysis_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" file_path = os.path.join(analysis_dir, filename) suffix = 1 while os.path.exists(file_path): filename = f"frame_analysis_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{suffix:02d}.json" file_path = os.path.join(analysis_dir, filename) suffix += 1 with open(file_path, "w", encoding="utf-8") as fp: json.dump(artifact, fp, ensure_ascii=False, indent=2) logger.info(f"分析结果已保存到: {file_path}") return file_path def _build_video_clip_json(self, batch_results: list[FrameBatchResult]) -> list[dict]: clips: list[dict] = [] for batch in self._sort_batch_results(batch_results): picture = self._build_batch_picture(batch) clips.append( { "timestamp": batch.time_range, "picture": picture, "narration": "", "OST": 2, } ) return clips def _build_batch_picture(self, batch: FrameBatchResult) -> str: summary = (batch.overall_activity_summary or "").strip() if summary: return summary fallback = (batch.fallback_summary or "").strip() if fallback: return fallback observation_lines = [] for frame in batch.frame_observations: timestamp = str(frame.get("timestamp", "") or "").strip() observation = str(frame.get("observation", "") or "").strip() if timestamp and observation: observation_lines.append(f"{timestamp}: {observation}") elif observation: observation_lines.append(observation) if observation_lines: return " ".join(observation_lines) raw_response = (batch.raw_response or "").strip() if raw_response: return raw_response[:200] return "该批次分析失败，未返回可用描述。" def _time_range_sort_key(self, time_range: str) -> tuple[int, str]: start = (time_range or "").split("-", 1)[0].strip() return self._timestamp_to_milliseconds(start), time_range @staticmethod def _timestamp_to_milliseconds(timestamp: str) -> int: text = (timestamp or "").strip() try: if "," in text: time_part, ms_part = text.split(",", 1) milliseconds = int(ms_part) else: time_part = text milliseconds = 0 parts = [int(part) for part in time_part.split(":") if part] while len(parts) < 3: parts.insert(0, 0) hours, minutes, seconds = parts[-3], parts[-2], parts[-1] return ((hours * 3600 + minutes * 60 + seconds) * 1000) + milliseconds except Exception: return 0 def _get_batch_timestamps( self, batch_files: list[str], prev_batch_files: list[str] | None = None, ) -> tuple[str, str, str]: if not batch_files: return "00:00:00,000", "00:00:00,000", "00:00:00,000-00:00:00,000" if len(batch_files) == 1 and prev_batch_files: first_frame = os.path.basename(prev_batch_files[-1]) last_frame = os.path.basename(batch_files[0]) else: first_frame = os.path.basename(batch_files[0]) last_frame = os.path.basename(batch_files[-1]) first_timestamp = self._timestamp_from_keyframe_name(first_frame) last_timestamp = self._timestamp_from_keyframe_name(last_frame) return first_timestamp, last_timestamp, f"{first_timestamp}-{last_timestamp}" def _timestamp_from_keyframe_name(self, filename: str) -> str: match = re.search(r"keyframe_\d{6}_(\d{9})\.jpg$", filename) if not match: return "00:00:00,000" token = match.group(1) hours = int(token[0:2]) minutes = int(token[2:4]) seconds = int(token[4:6]) milliseconds = int(token[6:9]) return f"{hours:02d}:{minutes:02d}:{seconds:02d},{milliseconds:03d}" def _build_analysis_prompt(self, frame_count: int) -> str: return self.PROMPT_TEMPLATE.format(frame_count=frame_count) def _build_failed_batch_result( self, *, batch_index: int, raw_response: str, error_message: str, frame_paths: list[str], time_range: str, ) -> FrameBatchResult: fallback_summary = (raw_response or "").strip()[:200] if not fallback_summary: fallback_summary = f"Batch {batch_index} analysis failed: {error_message or 'unknown error'}" return FrameBatchResult( batch_index=batch_index, status="failed", time_range=time_range, raw_response=raw_response, frame_paths=list(frame_paths), fallback_summary=fallback_summary, error_message=error_message, ) def _build_cache_key( self, video_path: str, interval_seconds: float, prompt_version: str, model_name: str, batch_size: int, max_concurrency: int, ) -> str: try: video_mtime = os.path.getmtime(video_path) except OSError: video_mtime = 0 legacy_prefix = utils.md5(f"{video_path}{video_mtime}") payload = "|".join( [ str(video_path), str(video_mtime), str(interval_seconds), str(prompt_version), str(model_name), str(batch_size), str(max_concurrency), "documentary-frame-analysis-v2", ] ) return f"{legacy_prefix}_{utils.md5(payload)}" def _strip_code_fence(self, response_text: str) -> str: cleaned = (response_text or "").strip() cleaned = re.sub(r"^```[a-zA-Z0-9_-]*\s*", "", cleaned) cleaned = re.sub(r"\s*```$", "", cleaned) return cleaned.strip() def _parse_batch_response( self, *, batch_index: int, raw_response: str, frame_paths: list[str], time_range: str, ) -> FrameBatchResult: try: payload = json.loads(self._strip_code_fence(raw_response)) except Exception as exc: return self._build_failed_batch_result( batch_index=batch_index, raw_response=raw_response, error_message=str(exc), frame_paths=frame_paths, time_range=time_range, ) validation_error = self._validate_batch_payload_contract(payload, expected_frame_count=len(frame_paths)) if validation_error: return self._build_failed_batch_result( batch_index=batch_index, raw_response=raw_response, error_message=validation_error, frame_paths=frame_paths, time_range=time_range, ) raw_observations = payload["frame_observations"] frame_observations: list[dict] = [] for index, frame_path in enumerate(frame_paths): entry = raw_observations[index] if index < len(raw_observations) else {} if isinstance(entry, dict): observation = str(entry.get("observation", "") or "") timestamp = str(entry.get("timestamp", "") or "") else: observation = str(entry or "") timestamp = "" frame_observations.append( { "frame_path": frame_path, "timestamp": timestamp, "observation": observation, } ) raw_summary = payload.get("overall_activity_summary", "") if isinstance(raw_summary, str): summary = raw_summary elif raw_summary is None: summary = "" else: summary = str(raw_summary) return FrameBatchResult( batch_index=batch_index, status="success", time_range=time_range, raw_response=raw_response, frame_paths=list(frame_paths), frame_observations=frame_observations, overall_activity_summary=summary, ) def _validate_batch_payload_contract(self, payload: object, *, expected_frame_count: int) -> str: if not isinstance(payload, dict): return "Batch response JSON payload must be an object" if "frame_observations" not in payload or not isinstance(payload["frame_observations"], list): return "Batch response must include frame_observations as a list" if len(payload["frame_observations"]) < expected_frame_count: return ( "Batch response frame_observations length is shorter than provided frame_paths: " f"{len(payload['frame_observations'])} < {expected_frame_count}" ) return ""