From aa193c40a6eebdfd84f2ab79ddd55725c9a4c222 Mon Sep 17 00:00:00 2001 From: Codex Date: Sun, 23 Aug 2026 22:49:58 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E=EF=BC=9A=E5=AE=9E=E7=8E=B0?= =?UTF-8?q?=E9=95=BF=E7=9B=B4=E6=92=AD=E5=88=86=E5=B1=82=E9=AB=98=E5=85=89?= =?UTF-8?q?=E9=80=89=E7=89=87?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- DEVELOPMENT_LOG.md | 10 + NEXT_STEPS.md | 4 +- app/db/database.py | 50 ++ app/services/ai/long_live_talk_analyzer.py | 745 ++++++++++++++++++ app/services/ai_analysis_workflow_service.py | 100 ++- app/services/pipeline_engine.py | 16 +- app/services/task_service.py | 1 + app/services/video_cut_workflow_service.py | 13 + docs/AI_ANALYSIS.md | 11 + docs/ARCHITECTURE.md | 18 + docs/DATABASE_SCHEMA.md | 19 + docs/TASK_FLOW.md | 9 + docs/UI_REFERENCE.md | 6 + .../2026-08-23-long-live-selection.md | 62 ++ tests/test_auto_pipeline.py | 2 + tests/test_long_live_selection.py | 249 ++++++ 16 files changed, 1304 insertions(+), 11 deletions(-) create mode 100644 app/services/ai/long_live_talk_analyzer.py create mode 100644 docs/agent_tasks/2026-08-23-long-live-selection.md create mode 100644 tests/test_long_live_selection.py diff --git a/DEVELOPMENT_LOG.md b/DEVELOPMENT_LOG.md index 9ecb3f0..5347a60 100644 --- a/DEVELOPMENT_LOG.md +++ b/DEVELOPMENT_LOG.md @@ -1,5 +1,15 @@ # Development Log +## 2026-08-23 长直播分层高光选片（PR 2） + +- `long_live_talk` 不再进入通用选片，改用固定约 300 秒、重叠 60 秒的语言高光窗口。 +- 新增窗口级 SQLite checkpoint：状态、累计尝试次数、结果校验和、错误和下次重试时间均独立保存；同一转写指纹下成功窗口跨进程复用。 +- 远程窗口单轮最多尝试 3 次并指数退避；失败窗口不会抹掉其他窗口结果，再次分析只请求缺失窗口。 +- 高光按金句观点、故事经历、情绪峰值、冲突反转、实用知识、互动幽默六类召回，跨窗口按时间与文本语义合并。 +- 最终结果先执行每小时密度上限，再轮询各小时执行总量上限，避免前半场提前占满全部名额。 +- 时间轴覆盖不足 90% 时保存为“分析不完整”，自动流水线和手动切片均会在生成文件、同步发送中心之前停止。 +- 新增六小时结构化时间轴、跨窗口去重、三次重试、断点复用、覆盖率门禁和幂等数据库测试。 + ## 2026-08-23 长直播基础设施（PR 1） - 新建任务移除综艺隐藏默认值，页面、multipart 上传与 JSON API 都要求显式选择三种模式；历史任务保留原模式。 diff --git a/NEXT_STEPS.md b/NEXT_STEPS.md index a4e75a2..628d7f2 100644 --- a/NEXT_STEPS.md +++ b/NEXT_STEPS.md @@ -3,11 +3,11 @@ ## 2026-08-23 长直播四阶段进度 - [x] PR 1：模式必选、已有文件入口、媒体/磁盘预检、持久化重型 Job、转写断点与词级时间戳。 -- [ ] PR 2：`long_live_talk` 5 分钟重叠窗口、每小时覆盖、全局配额、去重、90% 覆盖门禁和窗口级恢复。 +- [x] PR 2：`long_live_talk` 5 分钟重叠窗口、每小时覆盖、全局配额、去重、90% 覆盖门禁和窗口级恢复。 - [ ] PR 3：统一字幕 track/revision/cue、pysubs2 导入导出、wavesurfer 波形与专业编辑器。 - [ ] PR 4：字幕审核暂停、AI 建议 revision、异步批量烧录、NVENC 回退和发送中心门禁。 -当前验证重点：完成 PR 1 专项与全量回归；不要在 PR 1 宣称长直播选片或新字幕系统已经可用。 +当前验证重点：PR 2 完成专项与全量回归后进入字幕数据层；在 PR 3/PR 4 完成前，不宣称新字幕审核和批量烧录已经可用。 ## 2026-08-23 v2.1.0 主线同步后检查 diff --git a/app/db/database.py b/app/db/database.py index 8bbab47..c72dc79 100644 --- a/app/db/database.py +++ b/app/db/database.py @@ -369,6 +369,31 @@ def init_db() -> None: FOREIGN KEY(task_id) REFERENCES tasks(id) ); + CREATE TABLE IF NOT EXISTS ai_analysis_windows ( + id TEXT PRIMARY KEY, + task_id TEXT NOT NULL, + transcript_fingerprint TEXT NOT NULL, + provider TEXT NOT NULL, + model TEXT NOT NULL DEFAULT '', + window_index INTEGER NOT NULL, + start_seconds INTEGER NOT NULL, + end_seconds INTEGER NOT NULL, + status TEXT NOT NULL DEFAULT 'queued', + attempt_count INTEGER NOT NULL DEFAULT 0, + result_json TEXT, + result_checksum TEXT, + error_message TEXT, + next_retry_at TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + completed_at TEXT, + UNIQUE( + task_id, transcript_fingerprint, provider, model, + window_index, start_seconds, end_seconds + ), + FOREIGN KEY(task_id) REFERENCES tasks(id) + ); + CREATE TABLE IF NOT EXISTS cut_runs ( id TEXT PRIMARY KEY, task_id TEXT NOT NULL, @@ -413,6 +438,7 @@ def init_db() -> None: _restore_legacy_user_cancelled_publish_jobs(connection) _migrate_workflow_jobs_table(connection) _migrate_transcription_tables(connection) + _migrate_ai_analysis_windows_table(connection) _migrate_cut_runs_table(connection) _seed_ai_prompt_presets(connection) _seed_subtitle_style_preset(connection) @@ -438,6 +464,7 @@ def _requires_long_live_schema_migration(database_path) -> bool: or "lease_owner" not in job_columns or "transcription_runs" not in table_names or "transcription_chunks" not in table_names + or "ai_analysis_windows" not in table_names ) @@ -476,6 +503,7 @@ def _create_indexes(connection: sqlite3.Connection) -> None: "CREATE INDEX IF NOT EXISTS idx_workflow_jobs_task_type_status ON workflow_jobs(task_id, job_type, status)", "CREATE INDEX IF NOT EXISTS idx_transcription_runs_task_active ON transcription_runs(task_id, is_active, updated_at)", "CREATE INDEX IF NOT EXISTS idx_transcription_chunks_run_status ON transcription_chunks(run_id, status, chunk_index)", + "CREATE INDEX IF NOT EXISTS idx_ai_analysis_windows_resume ON ai_analysis_windows(task_id, transcript_fingerprint, provider, model, status, window_index)", """CREATE UNIQUE INDEX IF NOT EXISTS uq_publish_jobs_active_clip_platform_mode ON publish_jobs(output_clip_id, platform, publish_mode) WHERE status IN ('DRAFT', 'WAITING', 'SCHEDULED', 'PUBLISHING', 'NEED_REVIEW') @@ -1263,6 +1291,28 @@ def _migrate_transcription_tables(connection: sqlite3.Connection) -> None: ) +def _migrate_ai_analysis_windows_table(connection: sqlite3.Connection) -> None: + """创建长直播 AI 窗口 checkpoint 表；成功窗口可跨进程复用。""" + connection.executescript( + """ + CREATE TABLE IF NOT EXISTS ai_analysis_windows ( + id TEXT PRIMARY KEY, task_id TEXT NOT NULL, + transcript_fingerprint TEXT NOT NULL, provider TEXT NOT NULL, + model TEXT NOT NULL DEFAULT '', window_index INTEGER NOT NULL, + start_seconds INTEGER NOT NULL, end_seconds INTEGER NOT NULL, + status TEXT NOT NULL DEFAULT 'queued', attempt_count INTEGER NOT NULL DEFAULT 0, + result_json TEXT, result_checksum TEXT, error_message TEXT, next_retry_at TEXT, + created_at TEXT NOT NULL, updated_at TEXT NOT NULL, completed_at TEXT, + UNIQUE( + task_id, transcript_fingerprint, provider, model, + window_index, start_seconds, end_seconds + ), + FOREIGN KEY(task_id) REFERENCES tasks(id) + ); + """ + ) + + def _migrate_cut_runs_table(connection: sqlite3.Connection) -> None: """cut_runs 表的列级迁移，兼容未来新增字段""" columns = _get_table_columns(connection, "cut_runs") diff --git a/app/services/ai/long_live_talk_analyzer.py b/app/services/ai/long_live_talk_analyzer.py new file mode 100644 index 0000000..1c68834 --- /dev/null +++ b/app/services/ai/long_live_talk_analyzer.py @@ -0,0 +1,745 @@ +"""语言类长直播的可恢复分层高光选片。""" + +from __future__ import annotations + +from collections import defaultdict +from dataclasses import dataclass +from datetime import datetime, timedelta +import hashlib +import json +import math +from pathlib import Path +import re +import time +from typing import Any, Callable +from uuid import uuid4 + +from app.db.database import get_connection +from app.models.task import AIClipAnalysisResult +from app.services.ai.ai_clip_analyzer import ( + AIAnalysisError, + TranscriptRow, + _extract_transcript_rows, + _loads_ai_json, + _read_transcript, + _seconds_to_time, + _time_to_seconds, + build_provider, +) +from app.services.ai.base import AIProvider + + +WINDOW_SECONDS = 300 +WINDOW_OVERLAP_SECONDS = 60 +WINDOW_CHAR_BUDGET = 12_000 +WINDOW_RECALL_LIMIT = 5 +MIN_COMPLETE_COVERAGE = 0.90 +ALLOWED_CATEGORIES = ( + "quote_opinion", + "story_experience", + "emotional_peak", + "conflict_reversal", + "practical_knowledge", + "interactive_humor", +) +CATEGORY_LABELS = { + "quote_opinion": "金句观点", + "story_experience": "故事经历", + "emotional_peak": "情绪峰值", + "conflict_reversal": "冲突反转", + "practical_knowledge": "实用知识", + "interactive_humor": "互动幽默", +} + + +@dataclass(frozen=True) +class LongLiveAnalysisRequest: + task_id: str + transcript_path: Path + provider_name: str + model_name: str + density_per_hour: int = 4 + total_limit: int = 30 + ai_preference: str = "" + prompt_template: str | None = None + + +@dataclass(frozen=True) +class LongLiveWindow: + index: int + total: int + start_seconds: int + end_seconds: int + rows: tuple[TranscriptRow, ...] + text: str + + +@dataclass(frozen=True) +class LongLiveAnalysisOutcome: + result: AIClipAnalysisResult + meta: dict[str, Any] + + +def analyze_long_live_talk( + request: LongLiveAnalysisRequest, + *, + provider: AIProvider | None = None, + sleep_fn: Callable[[float], None] = time.sleep, + progress_callback: Callable[[dict[str, Any]], None] | None = None, +) -> LongLiveAnalysisOutcome: + transcript_text = _read_transcript(request.transcript_path) + rows = _extract_transcript_rows(transcript_text) + if not rows: + raise AIAnalysisError("长直播分析失败：转写中没有可识别的逐句时间戳") + + windows = build_long_live_windows(rows) + if not windows: + raise AIAnalysisError("长直播分析失败：没有生成可分析窗口") + + transcript_fingerprint = hashlib.sha256(transcript_text.encode("utf-8")).hexdigest() + provider = provider or build_provider(request.provider_name) + preference = _preference_summary(request.prompt_template or "", request.ai_preference) + successful_payloads: list[dict[str, Any]] = [] + completed_windows: list[LongLiveWindow] = [] + failed_windows: list[dict[str, Any]] = [] + reused_count = 0 + + for window in windows: + checkpoint = _get_or_create_checkpoint(request, transcript_fingerprint, window) + if checkpoint.get("status") == "completed" and checkpoint.get("result_json"): + payload = _load_verified_checkpoint_payload(checkpoint) + if isinstance(payload, dict): + successful_payloads.append(payload) + completed_windows.append(window) + reused_count += 1 + _report_progress(progress_callback, window, "reused", len(completed_windows)) + continue + + max_attempts = 3 if request.provider_name == "remote" else 1 + last_error = "" + payload = None + for attempt in range(1, max_attempts + 1): + _mark_checkpoint_running(checkpoint["id"]) + try: + raw = provider.generate_json(_window_prompt(window, preference)) + payload = _parse_window_payload(raw, window) + if not payload.get("moments"): + payload = {"moments": []} + _mark_checkpoint_completed(checkpoint["id"], payload) + successful_payloads.append(payload) + completed_windows.append(window) + _report_progress(progress_callback, window, "completed", len(completed_windows)) + break + except Exception as exc: # 每个窗口必须独立记录，不能丢失前面成功结果 + last_error = " ".join(str(exc).split())[:1000] or "未知错误" + should_retry = attempt < max_attempts + delay_seconds = 2 ** (attempt - 1) if should_retry else 0 + _mark_checkpoint_failed(checkpoint["id"], last_error, delay_seconds) + if should_retry: + sleep_fn(delay_seconds) + if payload is None: + failed_windows.append( + { + "window_index": window.index, + "start_seconds": window.start_seconds, + "end_seconds": window.end_seconds, + "error": last_error, + } + ) + _report_progress(progress_callback, window, "failed", len(completed_windows)) + + if not completed_windows: + detail = "；".join(item["error"] for item in failed_windows[:3]) or "全部窗口均失败" + raise AIAnalysisError(f"长直播分析没有完成任何窗口：{detail}") + + moments: list[dict[str, Any]] = [] + for payload in successful_payloads: + moments.extend(payload.get("moments") or []) + deduplicated = deduplicate_long_live_moments(moments) + density = max(1, min(10, int(request.density_per_hour or 4))) + total_limit = max(1, min(50, int(request.total_limit or 30))) + selected = select_temporally_balanced_highlights(deduplicated, density, total_limit) + clips = [_moment_to_clip(moment, index) for index, moment in enumerate(selected, start=1)] + + transcript_start = rows[0].start_seconds + transcript_end = rows[-1].end_seconds + coverage_ratio = calculate_window_coverage( + [(window.start_seconds, window.end_seconds) for window in completed_windows], + transcript_start, + transcript_end, + ) + incomplete = coverage_ratio < MIN_COMPLETE_COVERAGE + coverage_percent = round(coverage_ratio * 100, 2) + summary = ( + f"长直播高光已完成 {len(completed_windows)}/{len(windows)} 个重叠窗口，" + f"时间轴覆盖 {coverage_percent:.2f}%，去重后保留 {len(clips)} 条候选。" + ) + if incomplete: + summary += " 当前分析不完整，必须补齐失败窗口后才能进入自动切片。" + + meta = { + "transcript_fingerprint": transcript_fingerprint, + "window_seconds": WINDOW_SECONDS, + "window_overlap_seconds": WINDOW_OVERLAP_SECONDS, + "window_count": len(windows), + "completed_window_count": len(completed_windows), + "failed_window_count": len(failed_windows), + "failed_windows": failed_windows, + "reused_window_count": reused_count, + "coverage_ratio": round(coverage_ratio, 6), + "coverage_percent": coverage_percent, + "analysis_incomplete": incomplete, + "minimum_complete_coverage": MIN_COMPLETE_COVERAGE, + "highlight_density_per_hour": density, + "highlight_total_limit": total_limit, + "deduplicated_moment_count": len(deduplicated), + "selected_highlight_count": len(clips), + } + return LongLiveAnalysisOutcome( + result=AIClipAnalysisResult(task_id=request.task_id, analysis_summary=summary, clips=clips), + meta=meta, + ) + + +def build_long_live_windows(rows: list[TranscriptRow]) -> list[LongLiveWindow]: + """按约 5 分钟、60 秒重叠构造窗口，并兼顾 prompt 字符预算。""" + if not rows: + return [] + raw_windows: list[tuple[TranscriptRow, ...]] = [] + start_index = 0 + while start_index < len(rows): + start_seconds = rows[start_index].start_seconds + current: list[TranscriptRow] = [] + current_chars = 0 + end_index = start_index + while end_index < len(rows): + row = rows[end_index] + line = _format_row(row) + exceeds_time = bool(current) and row.end_seconds - start_seconds > WINDOW_SECONDS + exceeds_chars = bool(current) and current_chars + len(line) + 1 > WINDOW_CHAR_BUDGET + if exceeds_time or exceeds_chars: + break + current.append(row) + current_chars += len(line) + 1 + end_index += 1 + if not current: + current = [rows[start_index]] + end_index = start_index + 1 + raw_windows.append(tuple(current)) + if end_index >= len(rows): + break + next_time = max(current[0].start_seconds + 1, current[-1].end_seconds - WINDOW_OVERLAP_SECONDS) + next_index = start_index + 1 + while next_index < end_index and rows[next_index].start_seconds < next_time: + next_index += 1 + start_index = max(start_index + 1, next_index) + + total = len(raw_windows) + return [ + LongLiveWindow( + index=index, + total=total, + start_seconds=window_rows[0].start_seconds, + end_seconds=window_rows[-1].end_seconds, + rows=window_rows, + text="\n".join(_format_row(row) for row in window_rows), + ) + for index, window_rows in enumerate(raw_windows, start=1) + ] + + +def calculate_window_coverage( + intervals: list[tuple[int, int]], + timeline_start: int, + timeline_end: int, +) -> float: + if timeline_end <= timeline_start: + return 1.0 if intervals else 0.0 + clipped = sorted( + (max(timeline_start, start), min(timeline_end, end)) + for start, end in intervals + if end > timeline_start and start < timeline_end and end > start + ) + if not clipped: + return 0.0 + merged: list[list[int]] = [] + for start, end in clipped: + if not merged or start > merged[-1][1]: + merged.append([start, end]) + else: + merged[-1][1] = max(merged[-1][1], end) + covered = sum(end - start for start, end in merged) + return min(1.0, covered / (timeline_end - timeline_start)) + + +def deduplicate_long_live_moments(moments: list[dict[str, Any]]) -> list[dict[str, Any]]: + ranked = sorted(moments, key=lambda item: float(item.get("score") or 0), reverse=True) + selected: list[dict[str, Any]] = [] + for raw in ranked: + moment = _normalize_moment(raw) + if not moment: + continue + duplicate_index = None + for index, existing in enumerate(selected): + if _moments_are_duplicate(moment, existing): + duplicate_index = index + break + if duplicate_index is None: + selected.append(moment) + else: + selected[duplicate_index] = _merge_moments(selected[duplicate_index], moment) + return sorted(selected, key=lambda item: int(item["start_seconds"])) + + +def select_temporally_balanced_highlights( + moments: list[dict[str, Any]], + density_per_hour: int, + total_limit: int, +) -> list[dict[str, Any]]: + density = max(1, min(10, int(density_per_hour or 4))) + limit = max(1, min(50, int(total_limit or 30))) + buckets: dict[int, list[dict[str, Any]]] = defaultdict(list) + for moment in moments: + midpoint = (int(moment["start_seconds"]) + int(moment["end_seconds"])) / 2 + buckets[int(midpoint // 3600)].append(moment) + for bucket in buckets.values(): + bucket.sort(key=lambda item: float(item.get("score") or 0), reverse=True) + del bucket[density:] + + hours = sorted(buckets) + if len(hours) > limit: + indexes = _evenly_spaced_indexes(len(hours), limit) + hours = [hours[index] for index in indexes] + + selected: list[dict[str, Any]] = [] + for rank in range(density): + for hour in hours: + if rank < len(buckets[hour]): + selected.append(buckets[hour][rank]) + if len(selected) >= limit: + return sorted(selected, key=lambda item: int(item["start_seconds"])) + return sorted(selected, key=lambda item: int(item["start_seconds"])) + + +def list_long_live_window_checkpoints(task_id: str) -> list[dict[str, Any]]: + with get_connection() as connection: + rows = connection.execute( + """ + SELECT * FROM ai_analysis_windows + WHERE task_id = ? + ORDER BY updated_at DESC, window_index ASC + """, + (task_id,), + ).fetchall() + return [dict(row) for row in rows] + + +def get_latest_long_live_window_status(task_id: str) -> dict[str, Any]: + """返回最新一组窗口的轻量统计，不把历史转写指纹混入当前进度。""" + with get_connection() as connection: + latest = connection.execute( + """ + SELECT transcript_fingerprint, provider, model + FROM ai_analysis_windows + WHERE task_id = ? + ORDER BY updated_at DESC + LIMIT 1 + """, + (task_id,), + ).fetchone() + if not latest: + return {} + rows = connection.execute( + """ + SELECT window_index, start_seconds, end_seconds, status, + attempt_count, error_message, updated_at + FROM ai_analysis_windows + WHERE task_id = ? AND transcript_fingerprint = ? AND provider = ? AND model = ? + ORDER BY window_index ASC + """, + (task_id, latest["transcript_fingerprint"], latest["provider"], latest["model"]), + ).fetchall() + items = [dict(row) for row in rows] + completed = sum(1 for item in items if item["status"] == "completed") + failed = [item for item in items if item["status"] == "failed"] + return { + "provider": latest["provider"], + "model": latest["model"], + "window_count": len(items), + "completed_window_count": completed, + "failed_window_count": len(failed), + "failed_windows": failed, + "percent": round(completed / len(items) * 100) if items else 0, + } + + +def _get_or_create_checkpoint( + request: LongLiveAnalysisRequest, + fingerprint: str, + window: LongLiveWindow, +) -> dict[str, Any]: + now = _now_iso() + checkpoint_id = uuid4().hex + with get_connection() as connection: + connection.execute( + """ + INSERT OR IGNORE INTO ai_analysis_windows ( + id, task_id, transcript_fingerprint, provider, model, + window_index, start_seconds, end_seconds, status, + attempt_count, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'queued', 0, ?, ?) + """, + ( + checkpoint_id, + request.task_id, + fingerprint, + request.provider_name, + request.model_name, + window.index, + window.start_seconds, + window.end_seconds, + now, + now, + ), + ) + row = connection.execute( + """ + SELECT * FROM ai_analysis_windows + WHERE task_id = ? AND transcript_fingerprint = ? AND provider = ? AND model = ? + AND window_index = ? AND start_seconds = ? AND end_seconds = ? + """, + ( + request.task_id, + fingerprint, + request.provider_name, + request.model_name, + window.index, + window.start_seconds, + window.end_seconds, + ), + ).fetchone() + connection.commit() + if not row: + raise AIAnalysisError(f"无法创建长直播窗口 checkpoint：{window.index}") + return dict(row) + + +def _mark_checkpoint_running(checkpoint_id: str) -> None: + with get_connection() as connection: + connection.execute( + """ + UPDATE ai_analysis_windows + SET status = 'running', attempt_count = attempt_count + 1, + error_message = NULL, next_retry_at = NULL, updated_at = ? + WHERE id = ? + """, + (_now_iso(), checkpoint_id), + ) + connection.commit() + + +def _mark_checkpoint_completed(checkpoint_id: str, payload: dict[str, Any]) -> None: + now = _now_iso() + result_json = json.dumps(payload, ensure_ascii=False, sort_keys=True) + checksum = hashlib.sha256(result_json.encode("utf-8")).hexdigest() + with get_connection() as connection: + connection.execute( + """ + UPDATE ai_analysis_windows + SET status = 'completed', result_json = ?, result_checksum = ?, + error_message = NULL, next_retry_at = NULL, + updated_at = ?, completed_at = ? + WHERE id = ? + """, + (result_json, checksum, now, now, checkpoint_id), + ) + connection.commit() + + +def _mark_checkpoint_failed(checkpoint_id: str, error: str, delay_seconds: int) -> None: + now = datetime.now().astimezone() + next_retry_at = (now + timedelta(seconds=delay_seconds)).isoformat(timespec="seconds") if delay_seconds else None + with get_connection() as connection: + connection.execute( + """ + UPDATE ai_analysis_windows + SET status = 'failed', error_message = ?, next_retry_at = ?, updated_at = ? + WHERE id = ? + """, + (error, next_retry_at, now.isoformat(timespec="seconds"), checkpoint_id), + ) + connection.commit() + + +def _load_verified_checkpoint_payload(checkpoint: dict[str, Any]) -> dict[str, Any] | None: + result_json = str(checkpoint.get("result_json") or "") + expected_checksum = str(checkpoint.get("result_checksum") or "") + if not result_json or not expected_checksum: + return None + actual_checksum = hashlib.sha256(result_json.encode("utf-8")).hexdigest() + if actual_checksum != expected_checksum: + return None + try: + payload = json.loads(result_json) + except json.JSONDecodeError: + return None + return payload if isinstance(payload, dict) else None + + +def _parse_window_payload(raw: str, window: LongLiveWindow) -> dict[str, Any]: + payload = _loads_ai_json(raw) + if isinstance(payload, list): + payload = {"moments": payload} + if not isinstance(payload, dict): + raise AIAnalysisError("长直播窗口输出必须是 JSON 对象") + moments = payload.get("moments") + if moments is None: + moments = payload.get("clips") or payload.get("candidates") or [] + if not isinstance(moments, list): + raise AIAnalysisError("长直播窗口输出缺少 moments 数组") + normalized: list[dict[str, Any]] = [] + for item in moments[:WINDOW_RECALL_LIMIT]: + if not isinstance(item, dict): + continue + moment = _normalize_moment(item, window=window) + if moment: + normalized.append(moment) + return {"moments": normalized} + + +def _normalize_moment( + item: dict[str, Any], + *, + window: LongLiveWindow | None = None, +) -> dict[str, Any] | None: + try: + start_seconds = _coerce_seconds(item.get("start_seconds"), item.get("start_time")) + end_seconds = _coerce_seconds(item.get("end_seconds"), item.get("end_time")) + except (TypeError, ValueError): + return None + if end_seconds <= start_seconds: + return None + if window and (start_seconds < window.start_seconds - 5 or end_seconds > window.end_seconds + 5): + return None + duration = end_seconds - start_seconds + if duration < 15 or duration > 300: + return None + category = str(item.get("category") or "quote_opinion").strip().lower() + if category not in ALLOWED_CATEGORIES: + category = "quote_opinion" + score = _bounded_float(item.get("score") or item.get("confidence_score") or 70, 0, 100) + title = _clean_text(item.get("title"), "长直播高光", 160) + summary = _clean_text(item.get("summary"), title, 1000) + reason = _clean_text(item.get("highlight_reason") or item.get("reason"), CATEGORY_LABELS[category], 1000) + topic_key = _clean_text(item.get("topic_key"), title, 120) + key_seconds = _coerce_seconds(item.get("key_seconds"), item.get("key_time"), fallback=(start_seconds + end_seconds) // 2) + key_seconds = max(start_seconds, min(end_seconds, key_seconds)) + return { + "title": title, + "start_seconds": start_seconds, + "end_seconds": end_seconds, + "key_seconds": key_seconds, + "summary": summary, + "highlight_reason": reason, + "suggested_editing": _clean_text( + item.get("suggested_editing"), + "保留观点或故事闭环，剪掉明显停顿和重复表达。", + 1000, + ), + "category": category, + "topic_key": topic_key, + "score": score, + "source_window_indexes": [window.index] if window else list(item.get("source_window_indexes") or []), + } + + +def _moments_are_duplicate(first: dict[str, Any], second: dict[str, Any]) -> bool: + overlap = max( + 0, + min(int(first["end_seconds"]), int(second["end_seconds"])) + - max(int(first["start_seconds"]), int(second["start_seconds"])), + ) + shorter = max( + 1, + min( + int(first["end_seconds"]) - int(first["start_seconds"]), + int(second["end_seconds"]) - int(second["start_seconds"]), + ), + ) + if overlap / shorter >= 0.35: + return True + key_distance = abs(int(first["key_seconds"]) - int(second["key_seconds"])) + if key_distance <= 30: + return True + semantic = _semantic_similarity( + f"{first.get('topic_key', '')}{first.get('title', '')}{first.get('summary', '')}", + f"{second.get('topic_key', '')}{second.get('title', '')}{second.get('summary', '')}", + ) + return key_distance <= 120 and semantic >= 0.34 + + +def _merge_moments(first: dict[str, Any], second: dict[str, Any]) -> dict[str, Any]: + stronger, other = (first, second) if float(first["score"]) >= float(second["score"]) else (second, first) + merged = dict(stronger) + merged["start_seconds"] = min(int(first["start_seconds"]), int(second["start_seconds"])) + merged["end_seconds"] = max(int(first["end_seconds"]), int(second["end_seconds"])) + merged["source_window_indexes"] = sorted( + set(first.get("source_window_indexes") or []) | set(second.get("source_window_indexes") or []) + ) + if len(str(other.get("summary") or "")) > len(str(merged.get("summary") or "")): + merged["summary"] = other["summary"] + return merged + + +def _moment_to_clip(moment: dict[str, Any], index: int) -> dict[str, Any]: + start = int(moment["start_seconds"]) + end = int(moment["end_seconds"]) + duration = max(1, end - start) + score = float(moment.get("score") or 0) + quality_tier = "A" if score >= 80 else "B" if score >= 65 else "C" + return { + "clip_id": f"long_live_{index:03d}", + "title": moment["title"], + "start_time": _seconds_to_time(start), + "end_time": _seconds_to_time(end), + "duration_seconds": duration, + "cover_time_seconds": max(0.0, min(duration - 0.001, float(moment["key_seconds"] - start))), + "summary": moment["summary"], + "highlight_reason": f"{CATEGORY_LABELS[moment['category']]}：{moment['highlight_reason']}", + "spread_value": "高" if score >= 80 else "中", + "suggested_editing": moment["suggested_editing"], + "confidence_score": round(score / 100, 4), + "selected_by_default": True, + "quality_tier": quality_tier, + "quality_score": score, + "text_quality_score": score, + "humor_score": score if moment["category"] == "interactive_humor" else 0, + "completeness_score": score, + "audio_reaction_score": 0, + "topic_key": moment["topic_key"], + "key_moment_time": _seconds_to_time(int(moment["key_seconds"])), + "quality_evidence": { + "highlight_category": moment["category"], + "highlight_category_label": CATEGORY_LABELS[moment["category"]], + "source_window_indexes": moment.get("source_window_indexes") or [], + }, + "rejection_reason": "", + } + + +def _window_prompt(window: LongLiveWindow, preference: str) -> str: + return f"""你是语言内容型长直播的高光召回编辑。只分析当前约 5 分钟窗口，不评价整场直播。 +从下列六类中找出 0-{WINDOW_RECALL_LIMIT} 个可独立理解的高光： +1. quote_opinion 金句观点；2. story_experience 故事经历；3. emotional_peak 情绪峰值； +4. conflict_reversal 冲突反转；5. practical_knowledge 实用知识；6. interactive_humor 互动幽默。 +不要为凑数输出寒暄、重复表达、纯过场。片段建议 45-180 秒，必要时可为 15-300 秒。 +{preference} +只返回严格 JSON：{{"moments":[{{"title":"标题","category":"六类英文值之一","start_time":"HH:MM:SS","end_time":"HH:MM:SS","key_time":"HH:MM:SS","topic_key":"稳定话题标识","summary":"完整内容闭环","highlight_reason":"具体价值","suggested_editing":"剪辑建议","score":0}}]}} +score 为 0-100；时间必须来自窗口转写。没有高光就返回空数组。 + +窗口 {window.index}/{window.total}，范围 {_seconds_to_time(window.start_seconds)}-{_seconds_to_time(window.end_seconds)}： +{window.text}""" + + +def _preference_summary(prompt_template: str, ai_preference: str) -> str: + prompt = (prompt_template or "").replace("{{AI_PREFERENCE}}", ai_preference or "") + for marker in ("# Output Format", "【输出格式】", "输出 JSON", "转写文本：", "# Transcript", "{{TRANSCRIPT_TEXT}}"): + if marker in prompt: + prompt = prompt.split(marker, 1)[0] + prompt = " ".join(prompt.split())[:2000] + extra = " ".join((ai_preference or "").split())[:500] + parts = [] + if prompt: + parts.append(f"沿用本任务内容偏好：{prompt}") + if extra and extra not in prompt: + parts.append(f"用户补充偏好：{extra}") + return "\n".join(parts) + + +def _semantic_similarity(first: str, second: str) -> float: + first_tokens = _semantic_tokens(first) + second_tokens = _semantic_tokens(second) + if not first_tokens or not second_tokens: + return 0.0 + return len(first_tokens & second_tokens) / len(first_tokens | second_tokens) + + +def _semantic_tokens(value: str) -> set[str]: + normalized = re.sub(r"\s+", "", str(value or "").lower()) + chinese = "".join(re.findall(r"[\u4e00-\u9fff]", normalized)) + tokens = {chinese[index : index + 2] for index in range(max(0, len(chinese) - 1))} + tokens.update(re.findall(r"[a-z0-9]{2,}", normalized)) + return tokens + + +def _coerce_seconds(value: Any, time_text: Any, fallback: int | None = None) -> int: + if value not in (None, ""): + return int(float(value)) + if time_text not in (None, ""): + return _time_to_seconds(str(time_text)) + if fallback is not None: + return fallback + raise ValueError("缺少时间") + + +def _bounded_float(value: Any, minimum: float, maximum: float) -> float: + try: + number = float(value) + except (TypeError, ValueError): + number = minimum + return round(max(minimum, min(maximum, number)), 2) + + +def _clean_text(value: Any, fallback: str, limit: int) -> str: + text = " ".join(str(value or "").split()) or fallback + return text[:limit] + + +def _format_row(row: TranscriptRow) -> str: + return f"{row.start_time} - {row.end_time} {row.text}" + + +def _evenly_spaced_indexes(length: int, count: int) -> list[int]: + if count >= length: + return list(range(length)) + if count <= 1: + return [length // 2] + return sorted({round(index * (length - 1) / (count - 1)) for index in range(count)}) + + +def _report_progress( + callback: Callable[[dict[str, Any]], None] | None, + window: LongLiveWindow, + status: str, + completed_count: int, +) -> None: + if callback: + callback( + { + "window_index": window.index, + "window_count": window.total, + "status": status, + "completed_count": completed_count, + "percent": min(99, math.floor(window.index / max(1, window.total) * 100)), + } + ) + + +def _now_iso() -> str: + return datetime.now().astimezone().isoformat(timespec="seconds") + + +__all__ = [ + "ALLOWED_CATEGORIES", + "LongLiveAnalysisOutcome", + "LongLiveAnalysisRequest", + "LongLiveWindow", + "MIN_COMPLETE_COVERAGE", + "analyze_long_live_talk", + "build_long_live_windows", + "calculate_window_coverage", + "deduplicate_long_live_moments", + "get_latest_long_live_window_status", + "list_long_live_window_checkpoints", + "select_temporally_balanced_highlights", +] diff --git a/app/services/ai_analysis_workflow_service.py b/app/services/ai_analysis_workflow_service.py index 6f0e487..110d7f9 100644 --- a/app/services/ai_analysis_workflow_service.py +++ b/app/services/ai_analysis_workflow_service.py @@ -19,6 +19,12 @@ result_to_jsonable, ) from app.services.ai.variety_comedy_analyzer import ComedyAnalysisRequest, analyze_variety_comedy +from app.services.ai.long_live_talk_analyzer import ( + LongLiveAnalysisOutcome, + LongLiveAnalysisRequest, + analyze_long_live_talk, + get_latest_long_live_window_status, +) from app.services.ai.diagnostics import ensure_local_ai_ready from app.services.ai_prompt_preset_service import get_task_ai_prompt_preset from app.services.storage_service import get_artifact_paths @@ -73,6 +79,11 @@ def _read_analysis_meta(task_id: str) -> dict: return payload.get("analysis_meta") or {} +def get_task_ai_analysis_meta(task_id: str) -> dict: + """读取当前生效分析的元数据，供长直播覆盖率门禁复用。""" + return dict(_read_analysis_meta(task_id)) + + def _read_latest_ai_provider_from_log(task_id: str) -> str: paths = get_artifact_paths(task_id) if not paths["log_path"].exists(): @@ -114,14 +125,24 @@ def get_task_ai_analysis_status(task_id: str) -> dict: log_lines = read_task_log_tail(task_id) is_running = task.get("status") == TaskStatus.ai_analyzing.value has_analysis = paths["analysis_path"].exists() + window_status = ( + get_latest_long_live_window_status(task_id) + if task.get("selection_profile") == "long_live_talk" + else {} + ) percent = 0 message = "等待开始 AI 分析" status = "idle" if is_running: status = "running" - percent = 48 - message = "AI 正在分析转写文本，请保持页面打开。" + percent = int(window_status.get("percent") or 0) or 1 + message = ( + f"长直播 AI 正在处理窗口：已完成 {window_status.get('completed_window_count', 0)}" + f"/{window_status.get('window_count', 0)}。" + if window_status + else "AI 正在分析转写文本，请保持页面打开。" + ) if any("将使用分段分析" in line for line in log_lines): percent = 62 message = "AI 已读取 Prompt 和转写文本，正在分段生成候选片段。" @@ -129,9 +150,15 @@ def get_task_ai_analysis_status(task_id: str) -> dict: percent = 72 message = "远程 AI 分析接口暂不可用，已暂停等待你确认下一步。" elif task.get("status") == TaskStatus.pending_review.value and has_analysis: - status = "completed" - percent = 100 - message = "AI 分析完成，候选片段已生成，可检查后直接生成切片。" + meta = _read_analysis_meta(task_id) + status = "incomplete" if meta.get("analysis_incomplete") else "completed" + percent = int(float(meta.get("coverage_percent") or 100)) + message = ( + f"长直播分析覆盖 {float(meta.get('coverage_percent') or 0):.2f}%，" + "仍有窗口失败；请重试 AI 分析补齐窗口。" + if meta.get("analysis_incomplete") + else "AI 分析完成，候选片段已生成，可检查后直接生成切片。" + ) elif task.get("status") == TaskStatus.failed.value and any("AI 分析失败" in line for line in log_lines): status = "failed" percent = 100 @@ -153,6 +180,7 @@ def get_task_ai_analysis_status(task_id: str) -> dict: "log_path": str(paths["log_path"]), "log_lines": log_lines, "error_message": task.get("error_message") or "", + "window_status": window_status, } @@ -266,6 +294,7 @@ def _analysis_run_row_to_dict(row: Row, include_payload: bool = False) -> dict: payload = {} clips = payload.get("clips") or [] + analysis_meta = payload.get("analysis_meta") or {} return { "id": run.get("id"), "task_id": run.get("task_id"), @@ -285,6 +314,9 @@ def _analysis_run_row_to_dict(row: Row, include_payload: bool = False) -> dict: "review_url": f"/tasks/{run.get('task_id')}/clips/review", "clips": clips if include_payload else [], "clip_summaries": _summarize_analysis_clips(clips) if include_payload else [], + "analysis_meta": analysis_meta if include_payload else {}, + "analysis_incomplete": bool(analysis_meta.get("analysis_incomplete")), + "coverage_ratio": float(analysis_meta.get("coverage_ratio") or 0), } @@ -573,6 +605,37 @@ def _analyze_with_provider(task_id: str, task: dict, paths: dict[str, Path], pro ) ) + if task.get("selection_profile") == "long_live_talk": + density = max(1, min(10, int(task.get("highlight_density_per_hour") or 4))) + total_limit = max(1, min(50, int(task.get("highlight_total_limit") or 30))) + append_task_log( + task_id, + "长直播高光：使用约 5 分钟、重叠 60 秒的可恢复窗口；" + f"每小时最多 {density} 条，总计最多 {total_limit} 条。", + ) + + def report_progress(progress: dict) -> None: + window_index = int(progress.get("window_index") or 0) + window_count = int(progress.get("window_count") or 0) + status = str(progress.get("status") or "") + if status in {"failed", "reused"} or window_index in {1, window_count} or window_index % 10 == 0: + label = {"failed": "失败", "reused": "复用", "completed": "完成"}.get(status, status) + append_task_log(task_id, f"长直播 AI 窗口 {window_index}/{window_count}：{label}") + + return analyze_long_live_talk( + LongLiveAnalysisRequest( + task_id=task_id, + transcript_path=paths["transcript_path"], + provider_name=provider_name, + model_name=_ai_model_name(provider_name), + density_per_hour=density, + total_limit=total_limit, + ai_preference=task.get("ai_preference") or "", + prompt_template=prompt_template, + ), + progress_callback=report_progress, + ) + request = AnalysisRequest( task_id=task_id, transcript_path=paths["transcript_path"], @@ -655,6 +718,10 @@ def process_task_ai_analysis(task_id: str, provider: str | None = None) -> dict: f"{provider_error}。如需使用本地模型，请点击\"本地 AI 分析\"。" ) from provider_exc raise + long_live_meta = {} + if isinstance(analysis, LongLiveAnalysisOutcome): + long_live_meta = analysis.meta + analysis = analysis.result analysis_payload = result_to_jsonable(analysis) analysis_payload["analysis_meta"] = { "provider": used_provider, @@ -663,6 +730,7 @@ def process_task_ai_analysis(task_id: str, provider: str | None = None) -> dict: "selection_profile": task.get("selection_profile") or "general", "final_clip_target": int(task.get("final_clip_target") or 5), "generated_at": _now_iso(), + **long_live_meta, } prompt_preset = get_task_ai_prompt_preset(task_id) provider_label = _ai_provider_label(used_provider) @@ -678,7 +746,11 @@ def process_task_ai_analysis(task_id: str, provider: str | None = None) -> dict: model=model_name, fallback_notice=fallback_notice, prompt_preset=prompt_preset, - requested_clip_count=int(task["candidate_clip_count"]), + requested_clip_count=( + int(task.get("highlight_total_limit") or 30) + if task.get("selection_profile") == "long_live_talk" + else int(task["candidate_clip_count"]) + ), ) _append_ai_clip_quality_warnings(task_id, analysis_payload["clips"]) except (AIAnalysisError, Exception) as exc: @@ -689,8 +761,20 @@ def process_task_ai_analysis(task_id: str, provider: str | None = None) -> dict: raise ValueError(user_error) from exc update_task_status(task_id, TaskStatus.pending_review) - append_task_log(task_id, f"AI 分析完成，Provider：{used_provider}，生成候选片段：{len(analysis_payload['clips'])} 条") - message = f"AI 分析完成，已生成 {len(analysis_payload['clips'])} 条可直接切片的候选片段，可进入片段审核检查或直接生成切片。" + incomplete = bool(analysis_payload.get("analysis_meta", {}).get("analysis_incomplete")) + if incomplete: + coverage = float(analysis_payload["analysis_meta"].get("coverage_percent") or 0) + append_task_log( + task_id, + f"长直播 AI 分析不完整：覆盖率 {coverage:.2f}%，已保留成功窗口，自动切片已锁定。", + ) + message = ( + f"长直播分析覆盖率为 {coverage:.2f}%，低于 90%。" + "成功窗口已保存，请重试 AI 分析补齐缺失窗口；补齐前不能进入自动切片。" + ) + else: + append_task_log(task_id, f"AI 分析完成，Provider：{used_provider}，生成候选片段：{len(analysis_payload['clips'])} 条") + message = f"AI 分析完成，已生成 {len(analysis_payload['clips'])} 条可直接切片的候选片段，可进入片段审核检查或直接生成切片。" if fallback_notice: message = f"{fallback_notice} {message}" return { diff --git a/app/services/pipeline_engine.py b/app/services/pipeline_engine.py index 26a181b..64b7387 100644 --- a/app/services/pipeline_engine.py +++ b/app/services/pipeline_engine.py @@ -225,11 +225,23 @@ def _run_ai_analysis(self, task_id: str, context: dict) -> dict: result = task_service.process_task_ai_analysis(task_id) clip_count = len(result.get("clips") or []) append_task_log(task_id, f"全自动 AI 分析完成，候选片段：{clip_count} 条") - return {"clip_count": clip_count, "analysis_path": result.get("analysis_path") or ""} + return { + "clip_count": clip_count, + "analysis_path": result.get("analysis_path") or "", + "analysis_meta": (result.get("analysis_run") or {}).get("analysis_meta") or {}, + } def _select_clips(self, task_id: str, context: dict) -> dict: task = self._get_task(task_id) config = context["config"] + if task.get("selection_profile") == "long_live_talk": + meta = task_service.get_task_ai_analysis_meta(task_id) + if meta.get("analysis_incomplete") or float(meta.get("coverage_ratio") or 0) < 0.90: + coverage = float(meta.get("coverage_percent") or 0) + raise ValueError( + f"长直播分析覆盖率仅 {coverage:.2f}%，低于 90%；" + "请重试 AI 分析补齐缺失窗口，当前不会进入自动切片或发送中心。" + ) candidates = self._list_raw_candidates(task_id) if not candidates: latest_run = task_service.get_latest_ai_analysis_run(task_id) @@ -404,6 +416,8 @@ def _list_raw_candidates(self, task_id: str) -> list[dict]: def _resolve_target_count(self, task: dict, config: dict) -> int: if task.get("selection_profile") == "variety_comedy": return max(1, min(12, int(task.get("final_clip_target") or 5))) + if task.get("selection_profile") == "long_live_talk": + return max(1, min(50, int(task.get("highlight_total_limit") or 30))) return max(1, min(50, int(task.get("candidate_clip_count") or 12))) def _update_selected_clips(self, task_id: str, selected_ids: set[str]) -> None: diff --git a/app/services/task_service.py b/app/services/task_service.py index 6111301..8e3b6fe 100644 --- a/app/services/task_service.py +++ b/app/services/task_service.py @@ -32,6 +32,7 @@ _write_analysis_payload, get_ai_analysis_run, get_latest_ai_analysis_run, + get_task_ai_analysis_meta, get_task_ai_analysis_status, get_task_ai_source_label, list_ai_analysis_runs, diff --git a/app/services/video_cut_workflow_service.py b/app/services/video_cut_workflow_service.py index 881cef5..37c3bd7 100644 --- a/app/services/video_cut_workflow_service.py +++ b/app/services/video_cut_workflow_service.py @@ -167,6 +167,19 @@ def process_task_video_cuts(task_id: str, *, sync_publish_jobs: bool = True) -> if not task: raise ValueError("任务不存在") + if task.get("selection_profile") == "long_live_talk": + from app.services.ai_analysis_workflow_service import get_task_ai_analysis_meta + + meta = get_task_ai_analysis_meta(task_id) + if meta.get("analysis_incomplete") or float(meta.get("coverage_ratio") or 0) < 0.90: + coverage = float(meta.get("coverage_percent") or 0) + error = ( + f"长直播分析覆盖率仅 {coverage:.2f}%，低于 90%；" + "请先重试 AI 分析补齐缺失窗口，当前不会生成切片或同步发送中心。" + ) + append_task_log(task_id, f"视频切割已阻止：{error}") + raise ValueError(error) + source_path = get_source_video_path(task) valid, error_message = validate_source_video_path(str(source_path) if source_path else None) if not valid: diff --git a/docs/AI_ANALYSIS.md b/docs/AI_ANALYSIS.md index 7b1d7cf..1004fc7 100644 --- a/docs/AI_ANALYSIS.md +++ b/docs/AI_ANALYSIS.md @@ -1,5 +1,16 @@ # AI 片段分析说明 +## 2026-08-23：长直播语言高光模式 + +- 模式键：`long_live_talk`。 +- 窗口：300 秒，重叠 60 秒；单窗口最多召回 5 个高光，允许返回空数组，不强制凑数。 +- 类别：`quote_opinion`、`story_experience`、`emotional_peak`、`conflict_reversal`、`practical_knowledge`、`interactive_humor`。 +- 恢复：checkpoint 绑定转写 SHA-256、Provider、模型和窗口边界；任一键变化都会创建新的窗口集合。 +- 完整性：以成功窗口区间并集计算 `coverage_ratio`，小于 `0.90` 写入 `analysis_incomplete=true`。 +- 排序：先跨窗口去重，再按小时最多 N 条，最后跨小时轮询到总量上限；默认 4 条/小时、总计 30 条。 + +`general` 和 `variety_comedy` 继续使用原算法；本次没有迁移或重算既有分析历史。 + ## 2026-05-23：AI Prompt 方案 任务详情页现在使用“AI Prompt 方案”管理 AI 分析 Prompt： diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 36fcc3b..7a80917 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -1,5 +1,23 @@ # 系统架构 +## 2026-08-23：长直播选片层 + +`long_live_talk` 使用独立的 `long_live_talk_analyzer`： + +```text +结构化转写 +→ 300 秒窗口（60 秒重叠） +→ 每窗口独立 AI 召回与 SQLite checkpoint +→ 成功窗口时间轴并集 / 完整转写时间轴 = coverage_ratio +→ 时间重叠 + 语义相似去重 +→ 每小时密度筛选 +→ 跨小时轮询合并 +→ 总量上限 +→ analysis.json + ai_analysis_runs + clip_candidates +``` + +自动流水线在 `CLIP_SELECTING` 入口读取当前 `analysis_meta`。`analysis_incomplete=true` 或 `coverage_ratio < 0.90` 会直接中断，所以后续 `VIDEO_CUTTING`、内容准备和发送任务创建均不会执行。手动切片入口使用相同门禁。 + ## 2026-08-23：长直播基础层 - 新建任务必须显式选择 `general`、`variety_comedy` 或 `long_live_talk`；数据库的 `general` 默认值只用于旧数据兼容。 diff --git a/docs/DATABASE_SCHEMA.md b/docs/DATABASE_SCHEMA.md index d034b1d..542c346 100644 --- a/docs/DATABASE_SCHEMA.md +++ b/docs/DATABASE_SCHEMA.md @@ -1,5 +1,24 @@ # 数据库结构说明 +## 2026-08-23：长直播 AI 窗口 checkpoint + +新增 `ai_analysis_windows`： + +| 字段 | 说明 | +| --- | --- | +| `task_id` | 所属任务 | +| `transcript_fingerprint` | 完整转写内容 SHA-256；内容变化后不会复用旧窗口 | +| `provider` / `model` | Provider 与模型隔离键 | +| `window_index` | 当前窗口序号 | +| `start_seconds` / `end_seconds` | 原片主时间轴范围 | +| `status` | `queued / running / completed / failed` | +| `attempt_count` | 累计真实请求次数 | +| `result_json` / `result_checksum` | 成功结果与 SHA-256 校验和 | +| `error_message` / `next_retry_at` | 最后错误与退避时间 | +| `created_at / updated_at / completed_at` | 生命周期时间 | + +唯一键由任务、转写指纹、Provider、模型、窗口序号和起止时间组成。迁移使用 `CREATE TABLE/INDEX IF NOT EXISTS`，已有数据库在变更前继续执行 SQLite 在线备份。 + ## 2026-08-23：长直播基础设施迁移 - `tasks` 增加 `highlight_density_per_hour INTEGER NOT NULL DEFAULT 4` 与 `highlight_total_limit INTEGER NOT NULL DEFAULT 30`；历史任务模式和值不改写。 diff --git a/docs/TASK_FLOW.md b/docs/TASK_FLOW.md index 6c80361..9cc6727 100644 --- a/docs/TASK_FLOW.md +++ b/docs/TASK_FLOW.md @@ -1,5 +1,14 @@ # 任务状态流转 +## 2026-08-23：长直播高光流程 + +1. 逐句时间戳转写按约 5 分钟、60 秒重叠生成窗口。 +2. 每个窗口分别召回金句观点、故事经历、情绪峰值、冲突反转、实用知识和互动幽默；远程失败单轮最多尝试 3 次。 +3. 成功窗口立即保存，重启或重试时直接复用；只重新请求未成功窗口。 +4. 跨窗口重复事件按时间重叠、关键时刻距离和文本语义合并。 +5. 每小时最多使用任务的 `highlight_density_per_hour`，再按跨小时轮询应用 `highlight_total_limit`。 +6. 成功窗口覆盖完整时间轴不足 90% 时，任务保留候选供查看但标记“分析不完整”；自动/手动切片和发送中心同步均被阻止。 + ## 2026-08-23：长直播基础流程 `明确选择模式 → 上传或引用已有文件 → 媒体/磁盘预检 → SQLite Job 排队 → 音频提取 → 分块转写 checkpoint → transcript.md 兼容导出`。 diff --git a/docs/UI_REFERENCE.md b/docs/UI_REFERENCE.md index ffa35f9..4e7cc6a 100644 --- a/docs/UI_REFERENCE.md +++ b/docs/UI_REFERENCE.md @@ -1,5 +1,11 @@ # UI 参考说明 +## 2026-08-23 更新：长直播分析完整性状态 + +- 长直播 AI 窗口失败时，任务详情的 AI 状态显示“分析不完整”和当前覆盖率，不把部分结果冒充为已完整分析。 +- 用户可再次点击原 AI 分析按钮；系统复用成功窗口，只补失败窗口。 +- 覆盖率低于 90% 时，自动流水线和“生成切片”返回中文说明，不进入发送中心；页面布局和现有通用/综艺审核卡片保持不变。 + ## 2026-08-23 更新：新建任务与长直播入口 1. “视频来源”提供上传和本地/NAS已有文件。上传默认最大 4 GB；更大素材使用已有文件入口。 diff --git a/docs/agent_tasks/2026-08-23-long-live-selection.md b/docs/agent_tasks/2026-08-23-long-live-selection.md new file mode 100644 index 0000000..754a5fc --- /dev/null +++ b/docs/agent_tasks/2026-08-23-long-live-selection.md @@ -0,0 +1,62 @@ +# PR 2 执行任务：长直播高光分层选片 + +## 背景 + +PR 1 已完成长直播任务字段、媒体预检、持久化 Job 与转写分块 checkpoint。现有通用/综艺 AI 分析仍是请求内顺序处理，不能按窗口恢复，也没有长直播的时段覆盖和总量门禁。 + +## 目标 + +1. 为 `long_live_talk` 新增独立的 5 分钟重叠窗口召回。 +2. 将每个 AI 窗口的状态、尝试次数、结果和错误持久化，成功窗口可跨重启复用。 +3. 在每小时内去重排序，再进行全局合并，默认每小时 4 条、总计最多 30 条。 +4. 支持金句观点、故事经历、情绪峰值、冲突反转、实用知识、互动幽默六类高光。 +5. 覆盖率低于 90% 时明确标记分析不完整，禁止自动切片和发送中心同步。 + +## 允许修改范围 + +- `app/db/database.py` +- `app/services/ai/` 下的长直播分析实现及必要导出 +- `app/services/ai_analysis_workflow_service.py` +- `app/services/pipeline_engine.py` +- 与分析完成门禁直接相关的视频切片工作流 +- PR 2 专项测试、数据库/架构/任务流程/UI 文档、开发日志与下一步 + +## 禁止修改范围 + +- 不改变 `general` 与 `variety_comedy` 的算法和历史结果。 +- 不实现字幕数据层、字幕编辑器或字幕渲染；这些属于 PR 3/PR 4。 +- 不读取或写入任何 `.env`、密钥、Token、Cookie。 +- 不合并 PR，不删除分支，不改写 Git 历史。 + +## 已确定实现要求 + +- 窗口默认 300 秒、重叠 60 秒。 +- 远程窗口失败最多 3 次，指数退避；本地失败也记录错误但不做隐藏兜底。 +- checkpoint 以任务、转写内容指纹、Provider、模型及窗口参数隔离。 +- 同一事件用时间重叠/接近度与文本语义相似度合并。 +- 先按小时分桶和配额，再按时间排序输出，不能让前段高分候选耗尽全局名额。 +- 覆盖率按成功窗口对原时间轴的并集计算；不足 90% 保留成功结果供查看，但设置 `analysis_incomplete=true`。 +- 自动流水线在选片前检查最新分析元数据；不完整时停止，不进入切片、内容准备或发送中心。 + +## 验收标准 + +- 六小时结构化时间轴能生成完整窗口清单，并验证覆盖率、每小时密度、总上限和时间覆盖。 +- 模拟第 N 窗口失败后，失败状态和错误可查询；再次运行只请求缺失窗口。 +- 模拟跨窗口重复事件可以合并。 +- 覆盖率不足 90% 时自动流水线明确报错且不调用切片。 +- 原有通用和综艺测试不回归。 + +## 测试命令 + +```powershell +python -m pytest tests/test_long_live_selection.py tests/test_auto_pipeline.py -q +python -m pytest tests/ -q +python -m compileall app tests +``` + +## 返回格式 + +- 修改文件与关键设计 +- 专项及全量测试结果 +- Git diff/敏感信息检查结论 +- 中文 commit、推送分支、堆叠 PR 链接 diff --git a/tests/test_auto_pipeline.py b/tests/test_auto_pipeline.py index 673e9ec..09c9565 100644 --- a/tests/test_auto_pipeline.py +++ b/tests/test_auto_pipeline.py @@ -30,6 +30,7 @@ def auto_pipeline_db_cleanup(monkeypatch): ) init_db() with get_connection() as connection: + connection.execute("DELETE FROM ai_analysis_windows WHERE task_id LIKE 'test-auto-%'") connection.execute("DELETE FROM publish_jobs WHERE task_id LIKE 'test-auto-%'") connection.execute("DELETE FROM subtitle_jobs WHERE task_id LIKE 'test-auto-%'") connection.execute("DELETE FROM output_clip WHERE task_id LIKE 'test-auto-%'") @@ -40,6 +41,7 @@ def auto_pipeline_db_cleanup(monkeypatch): connection.commit() yield with get_connection() as connection: + connection.execute("DELETE FROM ai_analysis_windows WHERE task_id LIKE 'test-auto-%'") connection.execute("DELETE FROM publish_jobs WHERE task_id LIKE 'test-auto-%'") connection.execute("DELETE FROM subtitle_jobs WHERE task_id LIKE 'test-auto-%'") connection.execute("DELETE FROM output_clip WHERE task_id LIKE 'test-auto-%'") diff --git a/tests/test_long_live_selection.py b/tests/test_long_live_selection.py new file mode 100644 index 0000000..bafbef1 --- /dev/null +++ b/tests/test_long_live_selection.py @@ -0,0 +1,249 @@ +from __future__ import annotations + +from collections import Counter +from datetime import datetime, timezone +import json +import re + +import pytest + +from app.db.database import get_connection, init_db +from app.services.ai.ai_clip_analyzer import TranscriptRow +from app.services.ai.long_live_talk_analyzer import ( + LongLiveAnalysisRequest, + analyze_long_live_talk, + build_long_live_windows, + calculate_window_coverage, + deduplicate_long_live_moments, + list_long_live_window_checkpoints, + select_temporally_balanced_highlights, +) +from app.services.pipeline_engine import PipelineEngine +from app.services.storage_service import get_artifact_paths + + +PREFIX = "test-long-selection-" + + +@pytest.fixture(autouse=True) +def cleanup_long_live_selection_rows(): + init_db() + _cleanup() + yield + _cleanup() + + +def _cleanup() -> None: + with get_connection() as connection: + connection.execute("DELETE FROM ai_analysis_windows WHERE task_id LIKE ?", (f"{PREFIX}%",)) + connection.execute("DELETE FROM clip_candidates WHERE task_id LIKE ?", (f"{PREFIX}%",)) + connection.execute("DELETE FROM ai_analysis_runs WHERE task_id LIKE ?", (f"{PREFIX}%",)) + connection.execute("DELETE FROM tasks WHERE id LIKE ?", (f"{PREFIX}%",)) + connection.commit() + + +def _time_text(seconds: int) -> str: + hours, remainder = divmod(seconds, 3600) + minutes, value = divmod(remainder, 60) + return f"{hours:02d}:{minutes:02d}:{value:02d}" + + +def _rows_for_duration(duration_seconds: int, step_seconds: int = 30) -> list[TranscriptRow]: + rows = [] + for start in range(0, duration_seconds, step_seconds): + end = min(duration_seconds, start + step_seconds) + rows.append( + TranscriptRow( + start_time=_time_text(start), + end_time=_time_text(end), + start_seconds=start, + end_seconds=end, + text=f"第 {start // step_seconds + 1} 句结构化直播转写", + ) + ) + return rows + + +def _write_transcript(task_id: str, duration_seconds: int) -> None: + path = get_artifact_paths(task_id)["transcript_path"] + path.parent.mkdir(parents=True, exist_ok=True) + lines = ["| 开始 | 结束 | 文本 |", "| --- | --- | --- |"] + lines.extend( + f"| {row.start_time} | {row.end_time} | {row.text} |" + for row in _rows_for_duration(duration_seconds) + ) + path.write_text("\n".join(lines), encoding="utf-8") + + +def _create_task(task_id: str) -> None: + now = datetime.now(timezone.utc).isoformat(timespec="seconds") + with get_connection() as connection: + connection.execute( + """ + INSERT INTO tasks ( + id, task_name, task_dir_name, source_type, platform, selection_profile, + highlight_density_per_hour, highlight_total_limit, + status, progress, is_deleted, created_at, updated_at + ) VALUES (?, ?, ?, 'upload', 'general', 'long_live_talk', 4, 30, + 'pending_processing', 0, 0, ?, ?) + """, + (task_id, task_id, task_id, now, now), + ) + connection.commit() + + +def _moment(start: int, *, score: float = 80, title: str = "观点") -> dict: + return { + "title": title, + "start_seconds": start, + "end_seconds": start + 60, + "key_seconds": start + 30, + "summary": f"{title}完整内容", + "highlight_reason": "有明确内容价值", + "suggested_editing": "保留完整表达", + "category": "quote_opinion", + "topic_key": title, + "score": score, + "source_window_indexes": [], + } + + +class FakeWindowProvider: + def __init__(self, failed_indexes: set[int] | None = None): + self.failed_indexes = failed_indexes or set() + self.calls: Counter[int] = Counter() + + def generate_json(self, prompt: str, retry_instruction: str | None = None) -> str: + del retry_instruction + match = re.search(r"窗口 (\d+)/(\d+)，范围 (\d{2}:\d{2}:\d{2})-(\d{2}:\d{2}:\d{2})", prompt) + assert match + index = int(match.group(1)) + self.calls[index] += 1 + if index in self.failed_indexes: + raise RuntimeError(f"模拟窗口 {index} 网络失败") + start = sum( + value * factor + for value, factor in zip(map(int, match.group(3).split(":")), (3600, 60, 1), strict=True) + ) + end = sum( + value * factor + for value, factor in zip(map(int, match.group(4).split(":")), (3600, 60, 1), strict=True) + ) + clip_end = min(end, start + 60) + return json.dumps( + { + "moments": [ + { + "title": f"窗口 {index} 的观点", + "category": "quote_opinion", + "start_time": _time_text(start), + "end_time": _time_text(clip_end), + "key_time": _time_text((start + clip_end) // 2), + "topic_key": f"topic-{index}", + "summary": "完整观点", + "highlight_reason": "可独立传播", + "score": 80, + } + ] + }, + ensure_ascii=False, + ) + + +def test_six_hour_structured_timeline_window_coverage_density_and_total_limit(): + rows = _rows_for_duration(6 * 3600) + windows = build_long_live_windows(rows) + coverage = calculate_window_coverage( + [(window.start_seconds, window.end_seconds) for window in windows], + 0, + 6 * 3600, + ) + assert 85 <= len(windows) <= 95 + assert coverage == pytest.approx(1.0) + + moments = [] + for hour in range(6): + for position in range(7): + moments.append(_moment(hour * 3600 + position * 420, score=100 - position, title=f"{hour}-{position}")) + selected = select_temporally_balanced_highlights(moments, density_per_hour=4, total_limit=30) + hour_counts = Counter(((item["start_seconds"] + item["end_seconds"]) // 2) // 3600 for item in selected) + assert len(selected) == 24 + assert set(hour_counts) == set(range(6)) + assert max(hour_counts.values()) == 4 + + +def test_cross_window_duplicate_uses_time_and_semantic_merge(): + first = _moment(100, score=82, title="创业失败后的转折") + first["source_window_indexes"] = [1] + second = _moment(125, score=91, title="创业失败以后如何翻身") + second["source_window_indexes"] = [2] + unique = deduplicate_long_live_moments([first, second, _moment(1000, title="独立知识点")]) + assert len(unique) == 2 + merged = unique[0] + assert merged["score"] == 91 + assert merged["start_seconds"] == 100 + assert merged["end_seconds"] == 185 + assert merged["source_window_indexes"] == [1, 2] + + +def test_failed_window_retries_three_times_and_next_run_reuses_success(tmp_path): + task_id = f"{PREFIX}resume" + _create_task(task_id) + _write_transcript(task_id, 30 * 60) + request = LongLiveAnalysisRequest( + task_id=task_id, + transcript_path=get_artifact_paths(task_id)["transcript_path"], + provider_name="remote", + model_name="test-model", + density_per_hour=4, + total_limit=30, + ) + first_provider = FakeWindowProvider({3}) + first = analyze_long_live_talk(request, provider=first_provider, sleep_fn=lambda _seconds: None) + assert first_provider.calls[3] == 3 + assert first.meta["failed_window_count"] == 1 + checkpoints = list_long_live_window_checkpoints(task_id) + failed = [item for item in checkpoints if item["window_index"] == 3][0] + assert failed["status"] == "failed" + assert failed["attempt_count"] == 3 + assert "模拟窗口 3 网络失败" in failed["error_message"] + + second_provider = FakeWindowProvider() + second = analyze_long_live_talk(request, provider=second_provider, sleep_fn=lambda _seconds: None) + assert second_provider.calls == Counter({3: 1}) + assert second.meta["failed_window_count"] == 0 + assert second.meta["reused_window_count"] == second.meta["window_count"] - 1 + assert second.meta["coverage_ratio"] == pytest.approx(1.0) + + +def test_pipeline_blocks_incomplete_long_live_before_candidate_selection(): + task_id = f"{PREFIX}gate" + _create_task(task_id) + path = get_artifact_paths(task_id)["analysis_path"] + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps( + { + "clips": [], + "analysis_meta": { + "selection_profile": "long_live_talk", + "analysis_incomplete": True, + "coverage_ratio": 0.72, + "coverage_percent": 72, + }, + } + ), + encoding="utf-8", + ) + with pytest.raises(ValueError, match="低于 90%"): + PipelineEngine()._select_clips(task_id, {"config": {}}) + + +def test_ai_analysis_windows_schema_is_idempotent(): + init_db() + init_db() + with get_connection() as connection: + columns = {row["name"] for row in connection.execute("PRAGMA table_info(ai_analysis_windows)")} + indexes = {row["name"] for row in connection.execute("PRAGMA index_list(ai_analysis_windows)")} + assert {"transcript_fingerprint", "window_index", "attempt_count", "result_checksum", "next_retry_at"} <= columns + assert "idx_ai_analysis_windows_resume" in indexes