diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6a3d596..7be2dee 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -177,6 +177,7 @@ jobs: run: | docker exec niuma-studio-ci python -m scripts.seed_demo_data --reset docker exec niuma-studio-ci python -c "import sqlite3; db=sqlite3.connect('/app/ci-data/workflow.sqlite3'); assert db.execute(\"select count(*) from tasks where id like 'demo_%'\").fetchone()[0] == 3; assert db.execute(\"select count(*) from clip_candidates where id like 'demo_%'\").fetchone()[0] == 6; assert db.execute(\"select count(*) from publish_jobs where id like 'demo_%'\").fetchone()[0] == 6" + curl --fail --silent --show-error 'http://127.0.0.1:8001/api/system/readiness?deep=1' | python -c "import json, sys; report=json.load(sys.stdin); assert report['status'] != 'not_ready', report; assert report['checks']['database']['status'] == 'ok', report" - name: 验证主要页面 run: | @@ -184,6 +185,7 @@ jobs: curl --fail --silent --show-error http://127.0.0.1:8001/tasks >/dev/null curl --fail --silent --show-error http://127.0.0.1:8001/clips >/dev/null curl --fail --silent --show-error http://127.0.0.1:8001/publish >/dev/null + curl --fail --silent --show-error http://127.0.0.1:8001/content-review >/dev/null - name: 输出失败日志 if: failure() diff --git a/CHANGELOG.md b/CHANGELOG.md index b5f7233..475bf81 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,23 @@ - 在 Windows 10/11 + Docker Desktop 实机生成当前 `master` 的脱敏验收报告。 - 完成抖音 / B站真实发布灰度验证矩阵。 +## 2.2.0 - 2026-08-28 + +### Added + +- 新增独立“内容复盘”页面，支持抖音账号级日汇总 `.xlsx/.csv` 两阶段导入、近 50 条作品的一键同步和人工确认匹配。 +- 建立作品指标到发布记录、输出切片、候选片段、AI 分析与不可变 Prompt 版本的完整归因链，并提供带样本门槛的 Prompt 版本对比。 +- 新增 `/api/system/readiness?deep=0|1`，统一报告数据库、迁移账本、存储、调度、Worker 和 FFmpeg 状态。 + +### Changed + +- 保存审片选择时自动生成保留或淘汰反馈；重复状态不重复写入，原因变化保留为新事件，AI 只读取每个候选最新反馈。 +- 备份恢复增加 `integrity_check`、`foreign_key_check`、迁移 checksum、关键索引、活动服务和独占锁校验；危险旧目录迁移入口永久停用。 + +### Security + +- 抖音作品同步只复用本机 Chrome 登录态并保存指标白名单；登录失效、验证码、限流或页面变化时立即停止，不绕过平台验证，也不自动重试。 + ## 2.1.0 - 2026-08-23 ### Added diff --git a/DEVELOPMENT_LOG.md b/DEVELOPMENT_LOG.md index ef3799e..3206f0e 100644 --- a/DEVELOPMENT_LOG.md +++ b/DEVELOPMENT_LOG.md @@ -1,5 +1,18 @@ # Development Log +## 2026-08-28 v2.2.0 抖音内容复盘与 Prompt 归因 + +- 审片页以“保存审核选择”为唯一写入边界：首次保存自动形成 `keep/reject`，相同状态不重复，决策或淘汰原因变化时新增事件；AI 历史上下文只采用每个候选最新反馈。 +- 新增不可变 `ai_prompt_versions`，候选片段关联来源 AI Run，AI Run 固定关联 Prompt 版本与 SHA-256；历史记录只做可唯一确定的候选回填，不猜测 Prompt。 +- 新增 `/content-review`：支持抖音账号日汇总 `.xlsx/.csv` 预览与确认导入、7 天环比、作品归因表、人工确认和带 3 周期/30 条/双版本 20 条门槛的 Prompt 对比。 +- Windows Chrome Worker 新增最近 50 条抖音作品指标同步；复用现有登录态和账号锁，只保存指标白名单，登录失效、验证码、429、页面变化时立即停止且不自动重试。 +- 数据库恢复统一执行完整性、外键、迁移账本 checksum 和关键索引校验；正式库恢复前检查活动服务和独占锁。旧任务目录迁移入口永久停用，新增 `/api/system/readiness?deep=0|1`。 +- 14:21 的表结构检查调用了 `init_db()`，活动库因此执行 `20260828_01_content_review_v1`；迁移器先自动生成 `data/backups/workflow-before-content-review-v1-20260828-142159-923806-112221a7.sqlite3`。事后只读复核为 `integrity_check=ok`、`foreign_key_check=0`、备份同样通过完整诊断；Scheduler 正在运行，复核时 `PUBLISHING=0`、保留原有 `SCHEDULED=6`，未触发投稿或更改排期。该提前升级偏离了“先停服务再人工升级”的理想操作顺序，已在交付说明中明确保留。 +- Docker 隔离冒烟发现并修复演示数据仍缺少 2.2 归因的问题；Demo AI Run 现在固定关联 Prompt 版本及哈希，6 条候选均关联来源 Run。重建后深度 readiness 为 `database=ok`、`integrity_check=ok`、`foreign_key_check=0`、迁移 3 条、`PUBLISHING=0`，`/`、`/tasks`、`/clips`、`/publish`、`/content-review` 均返回 HTTP 200；隔离临时容器已删除。 +- 最终隔离回归为 `836 passed`、0 失败；Ruff、Compileall、5 个 JavaScript、20 个 PowerShell、3 套 Compose、`pip check` 和 `git diff --check` 全部通过。内容复盘页使用真实 Chrome 在 1440px 与 390px 验证，无页面级横向溢出。Docker 官方 Debian 源首次受本机代理 500/502 影响，改用 Dockerfile 已支持的阿里云镜像参数后构建及上述冒烟全部通过。 +- `acceptance.ps1` 与 `release_gate.ps1` 保留到 2.2 PR 合并后的干净 `master` 执行，避免功能分支报告被误当成正式发布证据。 +- 版本统一升级为 `2.2.0`。本轮不自动修改 Prompt、不自动发布、不绕过平台验证，也不对单体服务、前端框架或数据库层做全面重构。 + ## 2026-08-28 2.1 集成 PR 与 Docker 冒烟修复 - 将当前线性领先 `master` 的 26 个提交完整保留到 `codex/integrate-v2.1-stable`，新增独立空白清理提交并创建顶层集成 PR #60；不 rebase、不 squash，也不自动合并。 diff --git a/NEXT_STEPS.md b/NEXT_STEPS.md index 1fc8210..0aae27e 100644 --- a/NEXT_STEPS.md +++ b/NEXT_STEPS.md @@ -1,11 +1,18 @@ # Next Steps -## 2026-08-28 2.1 集成收口 +## 2026-08-28 v2.2 内容复盘验收与后续使用 -1. 等待 PR #60 的 Linux、Windows 与 Docker 三组 CI 全部通过；Docker 页面冒烟必须确认 `/`、`/tasks`、`/clips`、`/publish` 均返回成功。 -2. CI 全绿后由用户在 GitHub 使用普通 merge commit 手动合并到 `master`，不要 squash、rebase 或自动合并，以保留原 26 个提交历史。 -3. 合并后确认 `master` 包含提交 `c3e52be`，再关闭被顶层 PR 覆盖的堆叠 PR；不要删除分支，也不要处理无关的 Dependabot #44/#45。 -4. 只有 2.1 基线合并完成，才从最新 `master` 创建 `codex/douyin-content-review` 开始 2.2 内容复盘开发。 +1. 合并 2.2 PR 前确认 Linux、Windows host smoke 与 Docker image smoke 全绿；使用普通 merge commit，由用户手动合并，不自动删除功能分支。 +2. 活动库已在迁移器自动备份后完成 2.2 Schema 升级并通过完整性、外键和迁移账本校验；不要重复手工迁移。合并后安排受控重启前仍要确认没有 `PUBLISHING`，重启后访问 `/api/system/readiness?deep=1`，必须没有 `not_ready`。 +3. 在“内容复盘”选择本人抖音账号，先上传日汇总表预览，再人工确认导入；账号级历史数据只作为未归因趋势基线。 +4. 点击“同步最近 50 条作品”时保持本机 Chrome 已登录；验证码、登录失效、限流或页面变化时按错误提示人工处理，不连续重试。 +5. 对未匹配或多候选作品逐条人工确认。累积 3 个完整周同步、当前 Prompt 至少 30 条准确归因作品后再评估；两个版本各至少 20 条才查看版本间比较。 +6. 系统只给出 Prompt 调整建议。用户确认并实际修改 Prompt 后才创建新版本，继续观察下一个周期，不凭单条爆款直接下结论。 + +## 2026-08-28 2.1 集成收口（已合并） + +1. PR #60 已由用户使用普通 merge commit 合并，原 26 个提交历史保留，2.2 分支已从合并后的 `master` 创建。 +2. 顶层 PR 覆盖的旧堆叠 PR 仅做关闭归档，不删除分支；无关的 Dependabot #44/#45 保持不动。 ## 2026-08-26 页面动效检查 @@ -68,7 +75,7 @@ - [x] P1.4：已统一 AI/远程转写/FFmpeg/FFprobe 超时、错误 JSON、429/5xx 与重试幂等边界；计费不确定请求默认禁止自动重发。 - [x] P1.4b：AI 候选/run/任务终态已原子提交；人工 AI 已使用持久 Job/lease；普通/综艺/长直播具备单元恢复、计费不确定和不完整/质量降级门禁，父 Worker 取消与异常收尾已明确。 - [x] P1.5：配置读取 DTO 已脱敏，非本机请求受 Bearer 门禁保护，写请求校验同源；设置、平台 URL 与动态 HTML 输入边界已收紧。 -- [ ] P2：先替换并继续停用非 WAL-aware 的旧任务目录迁移脚本，再拆分 God Service、去除查询重复、补核心集成/故障测试和可观测性；不做全面重构。 +- [x] P2 必要护栏：危险旧任务目录迁移入口已无条件停用，恢复完整校验与 readiness 已补齐。God Service 拆分、查询去重和全面重构继续延后，按实际数据反馈再安排。 ### 本轮人工检查 diff --git a/README.en.md b/README.en.md index 2e3559d..c5d42db 100644 --- a/README.en.md +++ b/README.en.md @@ -9,7 +9,7 @@ A Windows-first, local AI video highlight workspace for livestream recordings, i [中文](README.md) · [English](README.en.md) · [Quick Start](docs/PROJECT_GUIDE.md) · [Portable Setup](docs/PORTABLE_SETUP.md) · [Backup & Restore](docs/BACKUP_AND_RESTORE.md) · [Technical Reference](docs/TECHNICAL_REFERENCE.md) · [Roadmap](ROADMAP.md) ![CI](https://github.com/damingishere-coder/Ai-Clip-Workflow/actions/workflows/ci.yml/badge.svg) -![Version](https://img.shields.io/badge/version-2.1.0-0969da) +![Version](https://img.shields.io/badge/version-2.2.0-0969da) ![Python](https://img.shields.io/badge/Python-3.12%2B-3776AB) ![Platform](https://img.shields.io/badge/platform-Windows-0078D4) ![FastAPI](https://img.shields.io/badge/backend-FastAPI-009688) @@ -91,6 +91,7 @@ flowchart LR | Transcription | Volcengine remote transcription and local faster-whisper | ✅ Available | | AI analysis | General-value mode, comedy-first mode, segmented long-video analysis | ✅ Available | | Review and clipping | Edit candidates, save selections, generate versioned clips | ✅ Available | +| Content review | Douyin daily imports, recent-50 item sync, attribution, and prompt comparison | ✅ Available | | Content preparation | Titles, descriptions, tags, cover frames, accounts, visibility | ✅ Available | | Scheduling | Batch preview, overnight windows, calendar view, timeline continuation | ✅ Available | | Data protection | Verified SQLite snapshots, manifests, restore rollback, upgrade protection | ✅ Available | diff --git a/README.md b/README.md index b562b62..55a43d5 100644 --- a/README.md +++ b/README.md @@ -9,7 +9,7 @@ Windows 本地运行的 AI 视频高光生产工作台，面向直播录像、 [中文](README.md) · [English](README.en.md) · [快速开始](docs/PROJECT_GUIDE.md) · [通用启动](docs/PORTABLE_SETUP.md) · [备份恢复](docs/BACKUP_AND_RESTORE.md) · [技术说明](docs/TECHNICAL_REFERENCE.md) · [路线图](ROADMAP.md) ![CI](https://github.com/damingishere-coder/Ai-Clip-Workflow/actions/workflows/ci.yml/badge.svg) -![Version](https://img.shields.io/badge/version-2.1.0-0969da) +![Version](https://img.shields.io/badge/version-2.2.0-0969da) ![Python](https://img.shields.io/badge/Python-3.12%2B-3776AB) ![Platform](https://img.shields.io/badge/platform-Windows-0078D4) ![FastAPI](https://img.shields.io/badge/backend-FastAPI-009688) @@ -91,6 +91,7 @@ flowchart LR | 语音转写 | 火山引擎远程转写、faster-whisper 本地转写 | ✅ 可用 | | AI 选片 | 通用内容价值、综艺笑点优先、长内容分段分析 | ✅ 可用 | | 审核切片 | 编辑候选、保存选择、按需生成新切片版本 | ✅ 可用 | +| 内容复盘 | 抖音日汇总导入、近 50 条作品同步、归因链和 Prompt 对比 | ✅ 可用 | | 内容准备 | 标题、简介、话题、封面帧、账号和可见范围 | ✅ 可用 | | 排期计划 | 批量预览、跨午夜窗口、月历、续接最晚排期 | ✅ 可用 | | 数据保护 | SQLite 一致性备份、清单校验、恢复前回滚与升级保护 | ✅ 可用 | diff --git a/VERSION b/VERSION index 7ec1d6d..ccbccc3 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -2.1.0 +2.2.0 diff --git a/app/db/database.py b/app/db/database.py index cfd57c7..c0a06e0 100644 --- a/app/db/database.py +++ b/app/db/database.py @@ -62,6 +62,37 @@ f"{TASK_UPLOAD_ONLY_MIGRATION_SQL}" ).encode("utf-8") ).hexdigest() +CONTENT_REVIEW_MIGRATION_VERSION = "20260828_01_content_review_v1" +CONTENT_REVIEW_MIGRATION_NAME = "内容复盘归因与指标快照基础结构" +CONTENT_REVIEW_REQUIRED_INDEXES = ( + "idx_clip_candidates_source_analysis_run", + "idx_ai_analysis_runs_prompt_version", + "idx_clip_feedback_candidate_created", + "idx_content_metric_import_batches_account_created", + "idx_douyin_account_daily_account_date", + "idx_douyin_item_metrics_account_published", + "idx_douyin_item_metrics_match_status", +) +CONTENT_REVIEW_MIGRATION_SPEC = "\n".join( + ( + CONTENT_REVIEW_MIGRATION_VERSION, + CONTENT_REVIEW_MIGRATION_NAME, + "clip_candidates.source_analysis_run_id", + "ai_analysis_runs.prompt_version_id", + "ai_analysis_runs.prompt_text_sha256", + "clip_feedback.decision_source", + "ai_prompt_versions", + "content_metric_import_batches", + "douyin_account_daily_metric_snapshots", + "douyin_item_metric_snapshots", + *CONTENT_REVIEW_REQUIRED_INDEXES, + "backfill-candidate-only-when-one-analysis-run", + "do-not-guess-historical-prompt-version", + ) +) +CONTENT_REVIEW_MIGRATION_CHECKSUM = hashlib.sha256( + CONTENT_REVIEW_MIGRATION_SPEC.encode("utf-8") +).hexdigest() class SchemaMigrationError(RuntimeError): @@ -101,6 +132,7 @@ def init_db() -> None: needs_subtitle_auto_backup = _requires_subtitle_auto_schema_migration(settings.database_path) needs_publish_index_backup = _requires_publish_active_index_migration(settings.database_path) needs_task_upload_only_backup = _requires_task_upload_only_migration(settings.database_path) + needs_content_review_backup = _requires_content_review_schema_migration(settings.database_path) if needs_long_live_backup: create_schema_migration_backup( settings.database_path, @@ -144,6 +176,12 @@ def init_db() -> None: settings.data_dir / "backups", "task-upload-only", ) + if needs_content_review_backup: + create_schema_migration_backup( + settings.database_path, + settings.data_dir / "backups", + "content-review-v1", + ) with get_connection() as connection: connection.executescript( @@ -211,6 +249,7 @@ def init_db() -> None: selected_by_default INTEGER NOT NULL DEFAULT 1, enabled INTEGER NOT NULL DEFAULT 1, reviewed INTEGER NOT NULL DEFAULT 0, + source_analysis_run_id TEXT, is_deleted INTEGER NOT NULL DEFAULT 0, deleted_at TEXT, created_at TEXT NOT NULL, @@ -247,6 +286,19 @@ def init_db() -> None: updated_at TEXT NOT NULL ); + CREATE TABLE IF NOT EXISTS ai_prompt_versions ( + id TEXT PRIMARY KEY, + preset_id TEXT NOT NULL, + version_number INTEGER NOT NULL, + preset_name_snapshot TEXT NOT NULL, + prompt_text TEXT NOT NULL, + prompt_sha256 TEXT NOT NULL, + created_at TEXT NOT NULL, + UNIQUE(preset_id, version_number), + UNIQUE(preset_id, prompt_sha256), + FOREIGN KEY(preset_id) REFERENCES ai_prompt_presets(id) + ); + CREATE TABLE IF NOT EXISTS ai_analysis_runs ( id TEXT PRIMARY KEY, task_id TEXT NOT NULL, @@ -256,13 +308,16 @@ def init_db() -> None: model TEXT NOT NULL, ai_prompt_preset_id TEXT, ai_prompt_preset_name TEXT, + prompt_version_id TEXT, + prompt_text_sha256 TEXT, requested_clip_count INTEGER NOT NULL DEFAULT 5, clip_count INTEGER NOT NULL DEFAULT 0, analysis_summary TEXT, fallback_notice TEXT, analysis_payload_json TEXT NOT NULL, created_at TEXT NOT NULL, - FOREIGN KEY(task_id) REFERENCES tasks(id) + FOREIGN KEY(task_id) REFERENCES tasks(id), + FOREIGN KEY(prompt_version_id) REFERENCES ai_prompt_versions(id) ); CREATE TABLE IF NOT EXISTS subtitle_style_presets ( @@ -600,6 +655,7 @@ def init_db() -> None: selection_profile TEXT NOT NULL DEFAULT 'general', decision TEXT NOT NULL, reason_code TEXT NOT NULL, + decision_source TEXT NOT NULL DEFAULT 'explicit_feedback', note TEXT, title_snapshot TEXT, summary_snapshot TEXT, @@ -609,6 +665,75 @@ def init_db() -> None: FOREIGN KEY(task_id) REFERENCES tasks(id), FOREIGN KEY(analysis_run_id) REFERENCES ai_analysis_runs(id) ); + + CREATE TABLE IF NOT EXISTS content_metric_import_batches ( + id TEXT PRIMARY KEY, + account_id TEXT NOT NULL, + source_kind TEXT NOT NULL, + source_filename TEXT NOT NULL, + source_sha256 TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'previewed', + period_start TEXT, + period_end TEXT, + normalized_payload_json TEXT NOT NULL DEFAULT '[]', + row_count INTEGER NOT NULL DEFAULT 0, + matched_count INTEGER NOT NULL DEFAULT 0, + ambiguous_count INTEGER NOT NULL DEFAULT 0, + invalid_count INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL, + committed_at TEXT, + expires_at TEXT, + UNIQUE(account_id, source_kind, source_sha256), + FOREIGN KEY(account_id) REFERENCES publish_accounts(id) + ); + + CREATE TABLE IF NOT EXISTS douyin_account_daily_metric_snapshots ( + id TEXT PRIMARY KEY, + batch_id TEXT NOT NULL, + account_id TEXT NOT NULL, + metric_date TEXT NOT NULL, + post_count INTEGER NOT NULL DEFAULT 0, + play_count INTEGER NOT NULL DEFAULT 0, + like_count INTEGER NOT NULL DEFAULT 0, + share_count INTEGER NOT NULL DEFAULT 0, + comment_count INTEGER NOT NULL DEFAULT 0, + five_second_completion_rate REAL, + two_second_bounce_rate REAL, + cover_click_rate REAL, + average_watch_seconds REAL, + created_at TEXT NOT NULL, + UNIQUE(batch_id, metric_date), + FOREIGN KEY(batch_id) REFERENCES content_metric_import_batches(id) ON DELETE CASCADE, + FOREIGN KEY(account_id) REFERENCES publish_accounts(id) + ); + + CREATE TABLE IF NOT EXISTS douyin_item_metric_snapshots ( + id TEXT PRIMARY KEY, + batch_id TEXT NOT NULL, + publish_job_id TEXT, + account_id TEXT NOT NULL, + aweme_id TEXT NOT NULL, + title TEXT NOT NULL DEFAULT '', + published_at TEXT, + duration_seconds REAL, + captured_at TEXT NOT NULL, + play_count INTEGER, + like_count INTEGER, + comment_count INTEGER, + share_count INTEGER, + collect_count INTEGER, + five_second_completion_rate REAL, + two_second_bounce_rate REAL, + cover_click_rate REAL, + average_watch_seconds REAL, + match_status TEXT NOT NULL DEFAULT 'unmatched', + match_method TEXT, + created_at TEXT NOT NULL, + UNIQUE(batch_id, aweme_id), + FOREIGN KEY(batch_id) REFERENCES content_metric_import_batches(id) ON DELETE CASCADE, + FOREIGN KEY(publish_job_id) REFERENCES publish_jobs(id), + FOREIGN KEY(account_id) REFERENCES publish_accounts(id) + ); """ ) _migrate_tasks_table(connection) @@ -774,6 +899,51 @@ def _requires_task_upload_only_migration(database_path) -> bool: connection.close() +def _requires_content_review_schema_migration(database_path) -> bool: + """已有库缺少内容复盘基础结构时，账本迁移前先创建可恢复备份。""" + if not database_path.exists() or database_path.stat().st_size == 0: + return False + connection = None + try: + connection = sqlite3.connect(f"{database_path.resolve().as_uri()}?mode=ro", uri=True, timeout=10) + table_names = { + row[0] for row in connection.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall() + } + if "clip_candidates" not in table_names: + return False + clip_columns = {row[1] for row in connection.execute("PRAGMA table_info(clip_candidates)").fetchall()} + run_columns = {row[1] for row in connection.execute("PRAGMA table_info(ai_analysis_runs)").fetchall()} + feedback_columns = {row[1] for row in connection.execute("PRAGMA table_info(clip_feedback)").fetchall()} + index_names = { + row[0] for row in connection.execute("SELECT name FROM sqlite_master WHERE type='index'").fetchall() + } + ledger_row = None + if "schema_migrations" in table_names: + try: + ledger_row = connection.execute( + "SELECT 1 FROM schema_migrations WHERE version = ? AND checksum = ?", + (CONTENT_REVIEW_MIGRATION_VERSION, CONTENT_REVIEW_MIGRATION_CHECKSUM), + ).fetchone() + except sqlite3.Error: + return True + return ( + "source_analysis_run_id" not in clip_columns + or not {"prompt_version_id", "prompt_text_sha256"} <= run_columns + or "decision_source" not in feedback_columns + or not { + "ai_prompt_versions", + "content_metric_import_batches", + "douyin_account_daily_metric_snapshots", + "douyin_item_metric_snapshots", + } <= table_names + or not set(CONTENT_REVIEW_REQUIRED_INDEXES) <= index_names + or ledger_row is None + ) + finally: + if connection is not None: + connection.close() + + def _get_table_columns(connection: sqlite3.Connection, table_name: str) -> set[str]: rows = connection.execute(f"PRAGMA table_info({table_name})").fetchall() return {row["name"] for row in rows} @@ -927,6 +1097,233 @@ def _verify_task_upload_only_migration(connection: sqlite3.Connection) -> None: raise SchemaMigrationError("仍存在未归一化的 NAS 视频来源记录") +def _apply_content_review_migration(connection: sqlite3.Connection) -> None: + clip_columns = _get_table_columns(connection, "clip_candidates") + if "source_analysis_run_id" not in clip_columns: + connection.execute("ALTER TABLE clip_candidates ADD COLUMN source_analysis_run_id TEXT") + + run_columns = _get_table_columns(connection, "ai_analysis_runs") + if "prompt_version_id" not in run_columns: + connection.execute("ALTER TABLE ai_analysis_runs ADD COLUMN prompt_version_id TEXT") + if "prompt_text_sha256" not in run_columns: + connection.execute("ALTER TABLE ai_analysis_runs ADD COLUMN prompt_text_sha256 TEXT") + + feedback_columns = _get_table_columns(connection, "clip_feedback") + if "decision_source" not in feedback_columns: + connection.execute( + "ALTER TABLE clip_feedback ADD COLUMN decision_source " + "TEXT NOT NULL DEFAULT 'explicit_feedback'" + ) + + schema_sql = """ + CREATE TABLE IF NOT EXISTS ai_prompt_versions ( + id TEXT PRIMARY KEY, + preset_id TEXT NOT NULL, + version_number INTEGER NOT NULL, + preset_name_snapshot TEXT NOT NULL, + prompt_text TEXT NOT NULL, + prompt_sha256 TEXT NOT NULL, + created_at TEXT NOT NULL, + UNIQUE(preset_id, version_number), + UNIQUE(preset_id, prompt_sha256), + FOREIGN KEY(preset_id) REFERENCES ai_prompt_presets(id) + ); + + CREATE TABLE IF NOT EXISTS content_metric_import_batches ( + id TEXT PRIMARY KEY, + account_id TEXT NOT NULL, + source_kind TEXT NOT NULL, + source_filename TEXT NOT NULL, + source_sha256 TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'previewed', + period_start TEXT, + period_end TEXT, + normalized_payload_json TEXT NOT NULL DEFAULT '[]', + row_count INTEGER NOT NULL DEFAULT 0, + matched_count INTEGER NOT NULL DEFAULT 0, + ambiguous_count INTEGER NOT NULL DEFAULT 0, + invalid_count INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL, + committed_at TEXT, + expires_at TEXT, + UNIQUE(account_id, source_kind, source_sha256), + FOREIGN KEY(account_id) REFERENCES publish_accounts(id) + ); + + CREATE TABLE IF NOT EXISTS douyin_account_daily_metric_snapshots ( + id TEXT PRIMARY KEY, + batch_id TEXT NOT NULL, + account_id TEXT NOT NULL, + metric_date TEXT NOT NULL, + post_count INTEGER NOT NULL DEFAULT 0, + play_count INTEGER NOT NULL DEFAULT 0, + like_count INTEGER NOT NULL DEFAULT 0, + share_count INTEGER NOT NULL DEFAULT 0, + comment_count INTEGER NOT NULL DEFAULT 0, + five_second_completion_rate REAL, + two_second_bounce_rate REAL, + cover_click_rate REAL, + average_watch_seconds REAL, + created_at TEXT NOT NULL, + UNIQUE(batch_id, metric_date), + FOREIGN KEY(batch_id) REFERENCES content_metric_import_batches(id) ON DELETE CASCADE, + FOREIGN KEY(account_id) REFERENCES publish_accounts(id) + ); + + CREATE TABLE IF NOT EXISTS douyin_item_metric_snapshots ( + id TEXT PRIMARY KEY, + batch_id TEXT NOT NULL, + publish_job_id TEXT, + account_id TEXT NOT NULL, + aweme_id TEXT NOT NULL, + title TEXT NOT NULL DEFAULT '', + published_at TEXT, + duration_seconds REAL, + captured_at TEXT NOT NULL, + play_count INTEGER, + like_count INTEGER, + comment_count INTEGER, + share_count INTEGER, + collect_count INTEGER, + five_second_completion_rate REAL, + two_second_bounce_rate REAL, + cover_click_rate REAL, + average_watch_seconds REAL, + match_status TEXT NOT NULL DEFAULT 'unmatched', + match_method TEXT, + created_at TEXT NOT NULL, + UNIQUE(batch_id, aweme_id), + FOREIGN KEY(batch_id) REFERENCES content_metric_import_batches(id) ON DELETE CASCADE, + FOREIGN KEY(publish_job_id) REFERENCES publish_jobs(id), + FOREIGN KEY(account_id) REFERENCES publish_accounts(id) + ); + + CREATE INDEX IF NOT EXISTS idx_clip_candidates_source_analysis_run + ON clip_candidates(source_analysis_run_id); + CREATE INDEX IF NOT EXISTS idx_ai_analysis_runs_prompt_version + ON ai_analysis_runs(prompt_version_id); + CREATE INDEX IF NOT EXISTS idx_clip_feedback_candidate_created + ON clip_feedback(clip_candidate_id, created_at DESC); + CREATE INDEX IF NOT EXISTS idx_content_metric_import_batches_account_created + ON content_metric_import_batches(account_id, created_at DESC); + CREATE INDEX IF NOT EXISTS idx_douyin_account_daily_account_date + ON douyin_account_daily_metric_snapshots(account_id, metric_date DESC); + CREATE INDEX IF NOT EXISTS idx_douyin_item_metrics_account_published + ON douyin_item_metric_snapshots(account_id, published_at DESC); + CREATE INDEX IF NOT EXISTS idx_douyin_item_metrics_match_status + ON douyin_item_metric_snapshots(match_status, created_at DESC); + """ + # sqlite3.executescript() 会隐式提交，账本迁移必须逐条执行以保持同一事务。 + for statement in schema_sql.split(";"): + normalized = statement.strip() + if normalized: + connection.execute(normalized) + + now = datetime.now().astimezone().isoformat(timespec="seconds") + presets = connection.execute( + "SELECT id, name, prompt_text FROM ai_prompt_presets ORDER BY slot" + ).fetchall() + for preset in presets: + prompt_text = str(preset["prompt_text"] or "").strip() + prompt_sha256 = hashlib.sha256(prompt_text.encode("utf-8")).hexdigest() + existing = connection.execute( + "SELECT 1 FROM ai_prompt_versions WHERE preset_id = ? AND prompt_sha256 = ?", + (preset["id"], prompt_sha256), + ).fetchone() + if existing is not None: + continue + version_number = int( + connection.execute( + "SELECT COALESCE(MAX(version_number), 0) + 1 FROM ai_prompt_versions WHERE preset_id = ?", + (preset["id"],), + ).fetchone()[0] + ) + version_id = f"promptv_{preset['id']}_{version_number:03d}" + connection.execute( + """ + INSERT INTO ai_prompt_versions ( + id, preset_id, version_number, preset_name_snapshot, + prompt_text, prompt_sha256, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?) + """, + ( + version_id, + preset["id"], + version_number, + str(preset["name"] or "未命名方案"), + prompt_text, + prompt_sha256, + now, + ), + ) + + connection.execute( + """ + UPDATE clip_candidates + SET source_analysis_run_id = ( + SELECT MIN(r.id) FROM ai_analysis_runs r WHERE r.task_id = clip_candidates.task_id + ) + WHERE source_analysis_run_id IS NULL + AND 1 = ( + SELECT COUNT(*) FROM ai_analysis_runs r WHERE r.task_id = clip_candidates.task_id + ) + """ + ) + + +def _verify_content_review_migration(connection: sqlite3.Connection) -> None: + required_columns = { + "clip_candidates": {"source_analysis_run_id"}, + "ai_analysis_runs": {"prompt_version_id", "prompt_text_sha256"}, + "clip_feedback": {"decision_source"}, + } + for table_name, expected in required_columns.items(): + missing = expected - _get_table_columns(connection, table_name) + if missing: + raise SchemaMigrationError( + f"内容复盘迁移后的 {table_name} 缺少字段：{', '.join(sorted(missing))}" + ) + + required_tables = { + "ai_prompt_versions", + "content_metric_import_batches", + "douyin_account_daily_metric_snapshots", + "douyin_item_metric_snapshots", + } + actual_tables = { + row[0] + for row in connection.execute("SELECT name FROM sqlite_master WHERE type = 'table'").fetchall() + } + missing_tables = sorted(required_tables - actual_tables) + if missing_tables: + raise SchemaMigrationError("内容复盘迁移缺少数据表：" + ", ".join(missing_tables)) + + actual_indexes = { + row[0] + for row in connection.execute("SELECT name FROM sqlite_master WHERE type = 'index'").fetchall() + } + missing_indexes = sorted(set(CONTENT_REVIEW_REQUIRED_INDEXES) - actual_indexes) + if missing_indexes: + raise SchemaMigrationError("内容复盘迁移缺少索引：" + ", ".join(missing_indexes)) + + stale_candidates = connection.execute( + """ + SELECT 1 + FROM clip_candidates c + WHERE c.source_analysis_run_id IS NULL + AND 1 = (SELECT COUNT(*) FROM ai_analysis_runs r WHERE r.task_id = c.task_id) + LIMIT 1 + """ + ).fetchone() + if stale_candidates is not None: + raise SchemaMigrationError("存在可唯一归因但尚未关联 AI 分析记录的历史候选片段") + + for row in connection.execute("SELECT id, prompt_text, prompt_sha256 FROM ai_prompt_versions"): + actual_hash = hashlib.sha256(str(row["prompt_text"] or "").encode("utf-8")).hexdigest() + if actual_hash != row["prompt_sha256"]: + raise SchemaMigrationError(f"Prompt 版本 {row['id']} 的 SHA-256 校验失败") + + def _registered_schema_migrations() -> tuple[SchemaMigration, ...]: return ( SchemaMigration( @@ -943,6 +1340,13 @@ def _registered_schema_migrations() -> tuple[SchemaMigration, ...]: apply=_apply_task_upload_only_migration, verify=_verify_task_upload_only_migration, ), + SchemaMigration( + version=CONTENT_REVIEW_MIGRATION_VERSION, + name=CONTENT_REVIEW_MIGRATION_NAME, + checksum=CONTENT_REVIEW_MIGRATION_CHECKSUM, + apply=_apply_content_review_migration, + verify=_verify_content_review_migration, + ), ) diff --git a/app/main.py b/app/main.py index f952b75..2aea627 100644 --- a/app/main.py +++ b/app/main.py @@ -12,7 +12,17 @@ from app.core.config import settings from app.db.database import init_db -from app.routers import ai_prompts, media, pages, publish, settings as settings_router, subtitles, tasks +from app.routers import ( + ai_prompts, + content_review, + media, + pages, + publish, + settings as settings_router, + subtitles, + system, + tasks, +) from app.services.publish_scheduler import start_scheduler_background from app.services.storage_service import configure_runtime_media_storage from app.services.job_worker import WorkflowJobRunner @@ -145,7 +155,7 @@ async def lifespan(app: FastAPI): app = FastAPI( title=settings.app_name, description=settings.app_description, - version="2.1.0", + version="2.2.0", lifespan=lifespan, ) @@ -237,7 +247,9 @@ async def security_middleware(request: Request, call_next): app.include_router(subtitles.router) app.include_router(media.router) app.include_router(publish.router) +app.include_router(content_review.router) app.include_router(settings_router.router) +app.include_router(system.router) @app.get("/favicon.ico", include_in_schema=False) diff --git a/app/models/content_review.py b/app/models/content_review.py new file mode 100644 index 0000000..05a651f --- /dev/null +++ b/app/models/content_review.py @@ -0,0 +1,14 @@ +from pydantic import BaseModel, Field + + +class ContentMetricImportCommitRequest(BaseModel): + confirm: bool = True + + +class DouyinAnalyticsSyncPreviewRequest(BaseModel): + account_id: str = Field(default="", max_length=120) + limit: int = Field(default=50, ge=1, le=50) + + +class ContentItemMatchUpdate(BaseModel): + publish_job_id: str = Field(..., min_length=1, max_length=120) diff --git a/app/models/task.py b/app/models/task.py index 7a2c197..e6e0350 100644 --- a/app/models/task.py +++ b/app/models/task.py @@ -396,6 +396,9 @@ class ClipCandidateUpdate(BaseModel): end_time: str = Field(..., min_length=1, max_length=16) enabled: bool = True summary: Optional[str] = Field(default=None, max_length=1000) + feedback_reason_code: Optional[ + Literal["not_funny", "fragmented", "missing_setup", "duplicate", "dragging", "other"] + ] = None class ClipCandidateBatchItem(ClipCandidateUpdate): diff --git a/app/routers/content_review.py b/app/routers/content_review.py new file mode 100644 index 0000000..2f58ee7 --- /dev/null +++ b/app/routers/content_review.py @@ -0,0 +1,138 @@ +from fastapi import APIRouter, File, Form, HTTPException, Query, UploadFile + +from app.models.content_review import ( + ContentItemMatchUpdate, + ContentMetricImportCommitRequest, + DouyinAnalyticsSyncPreviewRequest, +) +from app.services import content_review_service +from app.services.publishers.base import PublishError, PublishWorkerUnavailable +from app.services.publishers.worker_client import PublishWorkerClient + + +router = APIRouter(prefix="/api/content-review", tags=["content-review"]) + + +def _raise_content_review_http(exc: content_review_service.ContentReviewError): + raise HTTPException(status_code=exc.status_code, detail=str(exc)) from exc + + +@router.get("/accounts") +async def list_accounts() -> dict: + return {"accounts": content_review_service.list_douyin_accounts()} + + +@router.post("/imports/preview") +async def preview_import( + file: UploadFile = File(...), + account_id: str = Form(default=""), +) -> dict: + try: + content = await file.read(content_review_service.MAX_IMPORT_BYTES + 1) + return content_review_service.preview_metric_import( + account_id=account_id, + filename=file.filename or "data", + content=content, + ) + except content_review_service.ContentReviewError as exc: + _raise_content_review_http(exc) + finally: + await file.close() + + +@router.post("/imports/{batch_id}/commit") +async def commit_import( + batch_id: str, + payload: ContentMetricImportCommitRequest, +) -> dict: + if not payload.confirm: + raise HTTPException(status_code=400, detail="请确认后再导入") + try: + return content_review_service.commit_metric_import(batch_id) + except content_review_service.ContentReviewError as exc: + _raise_content_review_http(exc) + + +@router.post("/douyin/sync-preview") +async def sync_douyin_items(payload: DouyinAnalyticsSyncPreviewRequest) -> dict: + try: + account_id = content_review_service._resolve_douyin_account_id(payload.account_id) + worker_result = PublishWorkerClient().analytics_sync(account_id=account_id, limit=payload.limit) + return content_review_service.stage_douyin_item_sync( + account_id=account_id, + items=list(worker_result.get("items") or []), + captured_at=str(worker_result.get("captured_at") or content_review_service._now_iso()), + ) + except content_review_service.ContentReviewError as exc: + _raise_content_review_http(exc) + except (PublishError, PublishWorkerUnavailable) as exc: + error_code = str(getattr(exc, "error_code", "") or "WORKER_UNAVAILABLE") + status_code = { + "LOGIN_REQUIRED": 409, + "VERIFICATION_REQUIRED": 409, + "RATE_LIMITED": 429, + "PAGE_CHANGED": 422, + "WORKER_UNAVAILABLE": 503, + "publish_worker_unavailable": 503, + }.get(error_code, 409) + raise HTTPException( + status_code=status_code, + detail={"message": str(exc), "error_code": error_code}, + ) from exc + + +@router.get("/summary") +async def summary( + account_id: str = Query(default="", max_length=120), + days: int = Query(default=28, ge=14, le=180), +) -> dict: + try: + return content_review_service.get_content_review_summary(account_id, days) + except content_review_service.ContentReviewError as exc: + _raise_content_review_http(exc) + + +@router.get("/works") +async def works( + account_id: str = Query(default="", max_length=120), + limit: int = Query(default=100, ge=1, le=200), +) -> dict: + try: + return {"works": content_review_service.list_content_review_works(account_id, limit)} + except content_review_service.ContentReviewError as exc: + _raise_content_review_http(exc) + + +@router.get("/prompt-comparison") +async def prompt_comparison(account_id: str = Query(default="", max_length=120)) -> dict: + try: + return content_review_service.get_prompt_comparison(account_id) + except content_review_service.ContentReviewError as exc: + _raise_content_review_http(exc) + + +@router.get("/imports") +async def imports( + account_id: str = Query(default="", max_length=120), + limit: int = Query(default=20, ge=1, le=100), +) -> dict: + try: + return {"imports": content_review_service.list_import_batches(account_id, limit)} + except content_review_service.ContentReviewError as exc: + _raise_content_review_http(exc) + + +@router.put("/item-matches/{snapshot_id}") +async def update_item_match(snapshot_id: str, payload: ContentItemMatchUpdate) -> dict: + try: + return content_review_service.set_item_match(snapshot_id, payload.publish_job_id) + except content_review_service.ContentReviewError as exc: + _raise_content_review_http(exc) + + +@router.delete("/item-matches/{snapshot_id}") +async def remove_item_match(snapshot_id: str) -> dict: + try: + return content_review_service.delete_item_match(snapshot_id) + except content_review_service.ContentReviewError as exc: + _raise_content_review_http(exc) diff --git a/app/routers/pages.py b/app/routers/pages.py index 8fd6e24..d5e48ef 100644 --- a/app/routers/pages.py +++ b/app/routers/pages.py @@ -5,6 +5,7 @@ from app.core.config import settings from app.services.ai_prompt_preset_service import list_ai_prompt_presets +from app.services.content_review_service import list_douyin_accounts from app.services.publish_service import ( get_publish_center_context, get_publish_link_states, @@ -289,3 +290,18 @@ async def system_status_page(request: Request): **get_system_status_context(), }, ) + + +@router.get("/content-review") +async def content_review_page(request: Request): + accounts = list_douyin_accounts() + return templates.TemplateResponse( + name="content_review.html", + request=request, + context={ + "request": request, + "active_page": "content_review", + "settings": settings, + "accounts": accounts, + }, + ) diff --git a/app/routers/system.py b/app/routers/system.py new file mode 100644 index 0000000..bf539b9 --- /dev/null +++ b/app/routers/system.py @@ -0,0 +1,13 @@ +from fastapi import APIRouter, Query + +from app.services.system_readiness_service import build_system_readiness + + +router = APIRouter(prefix="/api/system", tags=["system"]) + + +@router.get("/readiness") +async def get_system_readiness( + deep: bool = Query(default=False), +) -> dict: + return build_system_readiness(deep=deep) diff --git a/app/services/ai_analysis_workflow_service.py b/app/services/ai_analysis_workflow_service.py index d135b87..b3679ed 100644 --- a/app/services/ai_analysis_workflow_service.py +++ b/app/services/ai_analysis_workflow_service.py @@ -29,7 +29,7 @@ ) from app.services.ai.diagnostics import ensure_local_ai_ready from app.services.ai.base import AIProviderError -from app.services.ai_prompt_preset_service import get_task_ai_prompt_preset +from app.services.ai_prompt_preset_service import get_task_ai_prompt_preset, get_task_ai_prompt_snapshot from app.services.storage_service import get_artifact_paths from app.services.task_log_service import append_task_log, read_task_log_tail @@ -319,16 +319,33 @@ def _clear_clip_candidates(task_id: str) -> None: connection.commit() -def _insert_clip_candidates(task_id: str, clips: list[dict]) -> None: +def _insert_clip_candidates( + task_id: str, + clips: list[dict], + source_analysis_run_id: str | None = None, +) -> None: from app.services.task_service import _now_iso now = _now_iso() with get_connection() as connection: - _insert_clip_candidates_with_connection(connection, task_id, clips, now) + _insert_clip_candidates_with_connection( + connection, + task_id, + clips, + now, + source_analysis_run_id=source_analysis_run_id, + ) connection.commit() -def _insert_clip_candidates_with_connection(connection, task_id: str, clips: list[dict], now: str) -> None: +def _insert_clip_candidates_with_connection( + connection, + task_id: str, + clips: list[dict], + now: str, + *, + source_analysis_run_id: str | None = None, +) -> None: for index, clip in enumerate(clips, start=1): clip_key = str(clip["clip_id"]) database_id = f"{task_id}_{clip_key}"[:120] @@ -341,9 +358,10 @@ def _insert_clip_candidates_with_connection(connection, task_id: str, clips: lis confidence_score, quality_tier, quality_score, text_quality_score, humor_score, completeness_score, audio_reaction_score, topic_key, key_moment_time, quality_evidence_json, rejection_reason, - selected_by_default, enabled, reviewed, created_at, updated_at + selected_by_default, enabled, reviewed, source_analysis_run_id, + created_at, updated_at ) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( database_id or f"{task_id}_clip_{index:03d}", @@ -373,20 +391,31 @@ def _insert_clip_candidates_with_connection(connection, task_id: str, clips: lis 1 if selected_by_default else 0, 1 if selected_by_default else 0, 0, + source_analysis_run_id, now, now, ), ) -def _replace_clip_candidates(task_id: str, clips: list[dict]) -> None: +def _replace_clip_candidates( + task_id: str, + clips: list[dict], + source_analysis_run_id: str | None = None, +) -> None: """在同一个事务里替换候选片段，失败时保留原结果。""" from app.services.task_service import _now_iso now = _now_iso() with get_connection() as connection: connection.execute("BEGIN IMMEDIATE") - _replace_clip_candidates_with_connection(connection, task_id, clips, now) + _replace_clip_candidates_with_connection( + connection, + task_id, + clips, + now, + source_analysis_run_id=source_analysis_run_id, + ) connection.commit() @@ -395,6 +424,8 @@ def _replace_clip_candidates_with_connection( task_id: str, clips: list[dict], now: str, + *, + source_analysis_run_id: str | None = None, ) -> None: referenced = connection.execute( """ @@ -411,7 +442,13 @@ def _replace_clip_candidates_with_connection( "请在片段审核页修改现有候选并重新切片。" ) connection.execute("DELETE FROM clip_candidates WHERE task_id = ?", (task_id,)) - _insert_clip_candidates_with_connection(connection, task_id, clips, now) + _insert_clip_candidates_with_connection( + connection, + task_id, + clips, + now, + source_analysis_run_id=source_analysis_run_id, + ) def _assert_ai_task_can_start(connection, task_id: str, current_status: str, *, current_job_id: str = "") -> None: @@ -702,6 +739,8 @@ def _analysis_run_row_to_dict(row: Row, include_payload: bool = False) -> dict: "model": run.get("model") or "", "ai_prompt_preset_id": run.get("ai_prompt_preset_id") or "", "ai_prompt_preset_name": run.get("ai_prompt_preset_name") or "", + "prompt_version_id": run.get("prompt_version_id") or "", + "prompt_text_sha256": run.get("prompt_text_sha256") or "", "requested_clip_count": int(run.get("requested_clip_count") or 0), "clip_count": int(run.get("clip_count") or 0), "analysis_summary": run.get("analysis_summary") or "", @@ -736,6 +775,8 @@ def _analysis_payload_to_preview(task_id: str, payload: dict, fallback: dict | N "model": model, "ai_prompt_preset_id": fallback.get("ai_prompt_preset_id") or "", "ai_prompt_preset_name": fallback.get("ai_prompt_preset_name") or "", + "prompt_version_id": meta.get("prompt_version_id") or fallback.get("prompt_version_id") or "", + "prompt_text_sha256": meta.get("prompt_sha256") or fallback.get("prompt_text_sha256") or "", "requested_clip_count": int(fallback.get("requested_clip_count") or len(clips)), "clip_count": len(clips), "analysis_summary": payload.get("analysis_summary") or fallback.get("analysis_summary") or "", @@ -842,8 +883,9 @@ def _insert_ai_analysis_run_with_connection( prompt_preset: dict, requested_clip_count: int, now: str, + run_id: str | None = None, ) -> str: - run_id = uuid4().hex[:12] + run_id = run_id or uuid4().hex[:12] clips = analysis_payload.get("clips") or [] run_number = _next_ai_analysis_run_number(connection, task_id) connection.execute("UPDATE ai_analysis_runs SET is_active = 0 WHERE task_id = ?", (task_id,)) @@ -851,11 +893,12 @@ def _insert_ai_analysis_run_with_connection( """ INSERT INTO ai_analysis_runs ( id, task_id, run_number, provider, provider_label, model, - ai_prompt_preset_id, ai_prompt_preset_name, requested_clip_count, + ai_prompt_preset_id, ai_prompt_preset_name, + prompt_version_id, prompt_text_sha256, requested_clip_count, clip_count, analysis_summary, fallback_notice, analysis_payload_json, is_active, created_at ) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( run_id, @@ -866,6 +909,8 @@ def _insert_ai_analysis_run_with_connection( model, prompt_preset.get("id") or "", prompt_preset.get("name") or "", + prompt_preset.get("prompt_version_id"), + prompt_preset.get("prompt_sha256"), requested_clip_count, len(clips), analysis_payload.get("analysis_summary") or "", @@ -1045,6 +1090,7 @@ def restore_ai_analysis_run(task_id: str, run_id: str) -> dict: task_id, payload.get("clips") or [], now, + source_analysis_run_id=run_id, ) connection.execute("UPDATE ai_analysis_runs SET is_active = 0 WHERE task_id = ?", (task_id,)) cursor = connection.execute( @@ -1086,8 +1132,13 @@ def restore_ai_analysis_run(task_id: str, run_id: str) -> dict: # ---------- AI 分析核心流程 ---------- -def _analyze_with_provider(task_id: str, task: dict, paths: dict[str, Path], provider_name: str): - prompt_preset = get_task_ai_prompt_preset(task_id) +def _analyze_with_provider( + task_id: str, + task: dict, + paths: dict[str, Path], + provider_name: str, + prompt_preset: dict, +): prompt_template = (prompt_preset.get("prompt_text") or "").strip() if not prompt_template: raise AIAnalysisError(f"当前选择的 AI Prompt 方案\"{prompt_preset.get('name')}\"还没有填写 Prompt 内容") @@ -1212,6 +1263,7 @@ def _commit_ai_analysis_result( from app.services.task_service import STATUS_PROGRESS, _now_iso now = _now_iso() + run_id = uuid4().hex[:12] with get_connection() as connection: connection.execute("BEGIN IMMEDIATE") try: @@ -1237,8 +1289,9 @@ def _commit_ai_analysis_result( task_id, analysis_payload.get("clips") or [], now, + source_analysis_run_id=run_id, ) - run_id = _insert_ai_analysis_run_with_connection( + _insert_ai_analysis_run_with_connection( connection, task_id=task_id, analysis_payload=analysis_payload, @@ -1249,6 +1302,7 @@ def _commit_ai_analysis_result( prompt_preset=prompt_preset, requested_clip_count=requested_clip_count, now=now, + run_id=run_id, ) cursor = connection.execute( """ @@ -1391,8 +1445,15 @@ def process_task_ai_analysis(task_id: str, provider: str | None = None) -> dict: try: if not paths["transcript_path"].exists(): raise AIAnalysisError("请先生成带时间戳的转写 Markdown，再开始 AI 分析") + prompt_preset = get_task_ai_prompt_snapshot(task_id) try: - analysis = _analyze_with_provider(task_id, task, paths, provider_name) + analysis = _analyze_with_provider( + task_id, + task, + paths, + provider_name, + prompt_preset, + ) except Exception as provider_exc: provider_error = ( provider_exc.checkpoint_message() @@ -1422,9 +1483,10 @@ def process_task_ai_analysis(task_id: str, provider: str | None = None) -> dict: "final_clip_target": int(task.get("final_clip_target") or 5), "generated_at": _now_iso(), "workflow_job_id": str(job["id"]), + "prompt_version_id": prompt_preset.get("prompt_version_id"), + "prompt_sha256": prompt_preset.get("prompt_sha256"), **long_live_meta, } - prompt_preset = get_task_ai_prompt_preset(task_id) provider_label = _ai_provider_label(used_provider) model_name = _ai_model_name(used_provider) analysis_run = _commit_ai_analysis_result( diff --git a/app/services/ai_prompt_preset_service.py b/app/services/ai_prompt_preset_service.py index 6085409..b588fa7 100644 --- a/app/services/ai_prompt_preset_service.py +++ b/app/services/ai_prompt_preset_service.py @@ -1,3 +1,4 @@ +import hashlib from datetime import datetime from app.db.database import DEFAULT_AI_PROMPT_PRESET_ID, get_connection @@ -8,6 +9,69 @@ def _now_iso() -> str: return datetime.now().isoformat(timespec="seconds") +def _prompt_sha256(prompt_text: str) -> str: + return hashlib.sha256((prompt_text or "").strip().encode("utf-8")).hexdigest() + + +def ensure_ai_prompt_version_with_connection( + connection, + *, + preset_id: str, + preset_name: str, + prompt_text: str, + now: str | None = None, +) -> dict: + """按 Prompt 实际内容复用或创建不可变版本；仅改名称不会增加版本号。""" + normalized_prompt = (prompt_text or "").strip() + prompt_sha256 = _prompt_sha256(normalized_prompt) + existing = connection.execute( + """ + SELECT id, preset_id, version_number, preset_name_snapshot, + prompt_text, prompt_sha256, created_at + FROM ai_prompt_versions + WHERE preset_id = ? AND prompt_sha256 = ? + """, + (preset_id, prompt_sha256), + ).fetchone() + if existing is not None: + return dict(existing) + + version_number = int( + connection.execute( + "SELECT COALESCE(MAX(version_number), 0) + 1 FROM ai_prompt_versions WHERE preset_id = ?", + (preset_id,), + ).fetchone()[0] + ) + version_id = f"promptv_{preset_id}_{version_number:03d}" + created_at = now or _now_iso() + connection.execute( + """ + INSERT INTO ai_prompt_versions ( + id, preset_id, version_number, preset_name_snapshot, + prompt_text, prompt_sha256, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?) + """, + ( + version_id, + preset_id, + version_number, + (preset_name or "未命名方案").strip() or "未命名方案", + normalized_prompt, + prompt_sha256, + created_at, + ), + ) + return { + "id": version_id, + "preset_id": preset_id, + "version_number": version_number, + "preset_name_snapshot": (preset_name or "未命名方案").strip() or "未命名方案", + "prompt_text": normalized_prompt, + "prompt_sha256": prompt_sha256, + "created_at": created_at, + } + + def list_ai_prompt_presets() -> list[dict]: with get_connection() as connection: rows = connection.execute( @@ -50,6 +114,7 @@ def update_ai_prompt_preset(preset_id: str, payload: AIPromptPresetUpdate) -> di normalized_name = payload.name.strip() or "未命名方案" prompt_text = payload.prompt_text.strip() with get_connection() as connection: + connection.execute("BEGIN IMMEDIATE") cursor = connection.execute( """ UPDATE ai_prompt_presets @@ -58,9 +123,17 @@ def update_ai_prompt_preset(preset_id: str, payload: AIPromptPresetUpdate) -> di """, (normalized_name, prompt_text, now, preset_id), ) + if cursor.rowcount == 0: + connection.rollback() + raise ValueError("AI Prompt 方案不存在") + ensure_ai_prompt_version_with_connection( + connection, + preset_id=preset_id, + preset_name=normalized_name, + prompt_text=prompt_text, + now=now, + ) connection.commit() - if cursor.rowcount == 0: - raise ValueError("AI Prompt 方案不存在") return { "status": "ok", "message": "AI Prompt 方案已保存。", @@ -117,6 +190,66 @@ def get_task_ai_prompt_preset(task_id: str) -> dict: return fallback +def get_task_ai_prompt_snapshot(task_id: str) -> dict: + """一次读取任务 Prompt 并绑定不可变版本，供整次 AI 分析复用。""" + now = _now_iso() + with get_connection() as connection: + connection.execute("BEGIN IMMEDIATE") + row = connection.execute( + """ + SELECT p.id, p.slot, p.name, p.prompt_text, p.is_default, + p.created_at, p.updated_at + FROM tasks t + LEFT JOIN ai_prompt_presets p ON p.id = t.ai_prompt_preset_id + WHERE t.id = ? + """, + (task_id,), + ).fetchone() + if row is None: + connection.rollback() + raise ValueError("任务不存在") + if not row["id"]: + row = connection.execute( + """ + SELECT id, slot, name, prompt_text, is_default, created_at, updated_at + FROM ai_prompt_presets WHERE id = ? + """, + (DEFAULT_AI_PROMPT_PRESET_ID,), + ).fetchone() + if row is None: + connection.rollback() + raise ValueError("默认 AI Prompt 方案不存在") + + preset = dict(row) + version = ensure_ai_prompt_version_with_connection( + connection, + preset_id=str(preset["id"]), + preset_name=str(preset["name"] or ""), + prompt_text=str(preset["prompt_text"] or ""), + now=now, + ) + connection.commit() + preset["is_default"] = bool(preset["is_default"]) + preset["prompt_preview"] = _prompt_preview(preset["prompt_text"]) + preset["prompt_version_id"] = version["id"] + preset["prompt_version_number"] = version["version_number"] + preset["prompt_sha256"] = version["prompt_sha256"] + return preset + + +def list_ai_prompt_versions() -> list[dict]: + with get_connection() as connection: + rows = connection.execute( + """ + SELECT id, preset_id, version_number, preset_name_snapshot, + prompt_sha256, created_at + FROM ai_prompt_versions + ORDER BY preset_id, version_number DESC + """ + ).fetchall() + return [dict(row) for row in rows] + + def _prompt_preview(prompt_text: str) -> str: compact = " ".join((prompt_text or "").split()) if len(compact) <= 80: diff --git a/app/services/clip_feedback_service.py b/app/services/clip_feedback_service.py index ef7ef63..f8687e0 100644 --- a/app/services/clip_feedback_service.py +++ b/app/services/clip_feedback_service.py @@ -15,14 +15,69 @@ FEEDBACK_REASON_LABELS = { "worth_publishing": "值得发", "not_funny": "不好笑", - "fragmented": "内容太碎", - "missing_setup": "铺垫不足", + "fragmented": "片段不完整", + "missing_setup": "铺垫缺失", "duplicate": "内容重复", "dragging": "节奏拖沓", "other": "其他", } +def record_review_toggle_feedback_with_connection( + connection, + *, + task_id: str, + clip: dict, + selection_profile: str, + enabled: bool, + reason_code: str | None, + now: str, +) -> bool: + """在候选保存事务中记录开关反馈；相同状态和原因不会重复写入。""" + decision = "keep" if enabled else "reject" + normalized_reason = "worth_publishing" if enabled else (reason_code or "other") + latest = connection.execute( + """ + SELECT decision, reason_code + FROM clip_feedback + WHERE task_id = ? AND clip_candidate_id = ? + ORDER BY created_at DESC, rowid DESC + LIMIT 1 + """, + (task_id, clip["id"]), + ).fetchone() + if latest is not None and ( + str(latest["decision"] or "") == decision + and str(latest["reason_code"] or "") == normalized_reason + ): + return False + + connection.execute( + """ + INSERT INTO clip_feedback ( + id, task_id, clip_candidate_id, analysis_run_id, selection_profile, + decision, reason_code, decision_source, note, title_snapshot, + summary_snapshot, start_time, end_time, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, 'review_toggle', '', ?, ?, ?, ?, ?) + """, + ( + uuid4().hex[:12], + task_id, + clip["id"], + clip.get("source_analysis_run_id"), + selection_profile or "general", + decision, + normalized_reason, + clip.get("title") or "", + clip.get("summary") or "", + clip.get("start_time") or "", + clip.get("end_time") or "", + now, + ), + ) + return True + + def save_clip_feedback(task_id: str, clip_id: str, payload: ClipFeedbackCreate) -> dict: from app.services.task_service import _now_iso, get_clip_candidate, get_task # noqa: F811 @@ -45,9 +100,9 @@ def save_clip_feedback(task_id: str, clip_id: str, payload: ClipFeedbackCreate) """ INSERT INTO clip_feedback ( id, task_id, clip_candidate_id, analysis_run_id, selection_profile, - decision, reason_code, note, title_snapshot, summary_snapshot, + decision, reason_code, decision_source, note, title_snapshot, summary_snapshot, start_time, end_time, created_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ) VALUES (?, ?, ?, ?, ?, ?, ?, 'explicit_feedback', ?, ?, ?, ?, ?, ?) """, ( uuid4().hex[:12], @@ -93,9 +148,17 @@ def list_recent_feedback_context(selection_profile: str, limit: int = 20) -> lis rows = connection.execute( """ SELECT decision, reason_code, note, title_snapshot, summary_snapshot, - start_time, end_time, created_at - FROM clip_feedback - WHERE selection_profile = ? + start_time, end_time, created_at, decision_source + FROM ( + SELECT f.*, + ROW_NUMBER() OVER ( + PARTITION BY f.task_id, f.clip_candidate_id + ORDER BY f.created_at DESC, f.rowid DESC + ) AS feedback_rank + FROM clip_feedback f + WHERE f.selection_profile = ? + ) + WHERE feedback_rank = 1 ORDER BY created_at DESC LIMIT ? """, @@ -104,4 +167,9 @@ def list_recent_feedback_context(selection_profile: str, limit: int = 20) -> lis return [dict(row) for row in rows] -__all__ = ["FEEDBACK_REASON_LABELS", "list_recent_feedback_context", "save_clip_feedback"] +__all__ = [ + "FEEDBACK_REASON_LABELS", + "list_recent_feedback_context", + "record_review_toggle_feedback_with_connection", + "save_clip_feedback", +] diff --git a/app/services/content_review_service.py b/app/services/content_review_service.py new file mode 100644 index 0000000..c278780 --- /dev/null +++ b/app/services/content_review_service.py @@ -0,0 +1,1058 @@ +from __future__ import annotations + +import csv +import hashlib +import io +import json +import math +import re +import statistics +from datetime import date, datetime, timedelta +from pathlib import Path +from uuid import uuid4 + +from app.db.database import get_connection + + +MAX_IMPORT_BYTES = 10 * 1024 * 1024 +MAX_IMPORT_ROWS = 10_000 +MAX_IMPORT_COLUMNS = 50 +PREVIEW_TTL_HOURS = 24 +DAILY_SOURCE_KIND = "account_daily_file" +DOUYIN_SYNC_SOURCE_KIND = "douyin_item_sync" + +DAILY_HEADERS = ( + "日期", + "投稿量", + "总播放量", + "总点赞量", + "总分享量", + "总评论量", + "5秒完播率", + "2秒跳出率", + "封面点击率", + "平均播放时长", +) + +COUNT_FIELDS = { + "投稿量": "post_count", + "总播放量": "play_count", + "总点赞量": "like_count", + "总分享量": "share_count", + "总评论量": "comment_count", +} +RATE_FIELDS = { + "5秒完播率": "five_second_completion_rate", + "2秒跳出率": "two_second_bounce_rate", + "封面点击率": "cover_click_rate", +} +REJECT_REASON_LABELS = { + "not_funny": "不好笑", + "missing_setup": "铺垫缺失", + "fragmented": "片段不完整", + "duplicate": "重复", + "dragging": "节奏拖沓", + "other": "其他", + "worth_publishing": "保留", +} +MATCHED_STATUSES = {"matched_exact", "matched_unique", "confirmed_manual"} + + +class ContentReviewError(ValueError): + def __init__(self, message: str, *, status_code: int = 400): + super().__init__(message) + self.status_code = status_code + + +def _now() -> datetime: + return datetime.now().astimezone() + + +def _now_iso() -> str: + return _now().isoformat(timespec="seconds") + + +def _resolve_douyin_account_id(account_id: str = "") -> str: + normalized = str(account_id or "").strip() + with get_connection() as connection: + if normalized: + row = connection.execute( + "SELECT id FROM publish_accounts WHERE id = ? AND platform = 'douyin'", + (normalized,), + ).fetchone() + if row is None: + raise ContentReviewError("没有找到这个抖音账号，请先在发送中心添加账号") + return normalized + rows = connection.execute( + "SELECT id FROM publish_accounts WHERE platform = 'douyin' ORDER BY created_at, id" + ).fetchall() + if len(rows) == 1: + return str(rows[0]["id"]) + if not rows: + raise ContentReviewError("还没有抖音账号，请先在发送中心添加账号") + raise ContentReviewError("存在多个抖音账号，请先选择本次数据对应的账号") + + +def list_douyin_accounts() -> list[dict]: + with get_connection() as connection: + rows = connection.execute( + """ + SELECT id, account_name, account_uid, login_status, login_checked_at + FROM publish_accounts + WHERE platform = 'douyin' + ORDER BY created_at, id + """ + ).fetchall() + return [dict(row) for row in rows] + + +def _normalize_header(value) -> str: + return re.sub(r"\s+", "", str(value or "").replace("\ufeff", "").strip()) + + +def _trim_row(row: tuple | list) -> list: + values = list(row) + while values and (values[-1] is None or str(values[-1]).strip() == ""): + values.pop() + return values + + +def _load_xlsx_rows(content: bytes) -> list[list]: + try: + from openpyxl import load_workbook + + workbook = load_workbook( + io.BytesIO(content), + read_only=True, + data_only=True, + keep_links=False, + ) + except Exception as exc: + raise ContentReviewError("Excel 文件无法读取，可能已损坏或不是有效的 .xlsx 文件") from exc + + matching_sheets: list[tuple[str, list[list]]] = [] + try: + for worksheet in workbook.worksheets: + if int(worksheet.max_column or 0) > MAX_IMPORT_COLUMNS: + raise ContentReviewError(f"工作表“{worksheet.title}”超过 {MAX_IMPORT_COLUMNS} 列限制") + rows: list[list] = [] + for raw_row in worksheet.iter_rows(values_only=True): + row = _trim_row(raw_row) + if not row or all(value is None or str(value).strip() == "" for value in row): + continue + rows.append(row) + if len(rows) > MAX_IMPORT_ROWS + 1: + raise ContentReviewError(f"工作表“{worksheet.title}”超过 {MAX_IMPORT_ROWS} 行数据限制") + if not rows: + continue + normalized = {_normalize_header(value) for value in rows[0]} + if set(DAILY_HEADERS) <= normalized: + matching_sheets.append((worksheet.title, rows)) + finally: + workbook.close() + + if not matching_sheets: + raise ContentReviewError("没有找到包含完整抖音日汇总表头的工作表") + if len(matching_sheets) > 1: + names = "、".join(name for name, _ in matching_sheets) + raise ContentReviewError(f"发现多个可导入工作表（{names}），请只保留一个数据工作表") + return matching_sheets[0][1] + + +def _load_csv_rows(content: bytes) -> list[list]: + decoded = None + for encoding in ("utf-8-sig", "gb18030"): + try: + decoded = content.decode(encoding) + break + except UnicodeDecodeError: + continue + if decoded is None: + raise ContentReviewError("CSV 编码无法识别，请使用 UTF-8 或 GB18030") + try: + reader = csv.reader(io.StringIO(decoded, newline="")) + rows = [_trim_row(row) for row in reader] + except csv.Error as exc: + raise ContentReviewError("CSV 文件格式损坏，无法读取") from exc + rows = [row for row in rows if row and any(str(value).strip() for value in row)] + if not rows: + raise ContentReviewError("CSV 文件没有数据") + if len(rows) - 1 > MAX_IMPORT_ROWS: + raise ContentReviewError(f"CSV 超过 {MAX_IMPORT_ROWS} 行数据限制") + if max(len(row) for row in rows) > MAX_IMPORT_COLUMNS: + raise ContentReviewError(f"CSV 超过 {MAX_IMPORT_COLUMNS} 列限制") + return rows + + +def _parse_date(value, *, row_number: int) -> str: + if isinstance(value, datetime): + parsed = value.date() + elif isinstance(value, date): + parsed = value + else: + text = str(value or "").strip() + parsed = None + for fmt in ("%Y-%m-%d", "%Y/%m/%d", "%Y.%m.%d", "%Y年%m月%d日"): + try: + parsed = datetime.strptime(text, fmt).date() + break + except ValueError: + continue + if parsed is None: + raise ContentReviewError(f"第 {row_number} 行日期无法识别：{text or '空值'}") + if parsed > _now().date(): + raise ContentReviewError(f"第 {row_number} 行日期晚于今天：{parsed.isoformat()}") + return parsed.isoformat() + + +def _parse_count(value, *, header: str, row_number: int) -> int: + text = ("" if value is None else str(value)).replace(",", "").strip() + try: + number = float(text) + except (TypeError, ValueError) as exc: + raise ContentReviewError(f"第 {row_number} 行“{header}”不是有效数字") from exc + if not math.isfinite(number) or number < 0 or number > 1_000_000_000_000: + raise ContentReviewError(f"第 {row_number} 行“{header}”超出合理范围") + if not number.is_integer(): + raise ContentReviewError(f"第 {row_number} 行“{header}”必须是整数") + return int(number) + + +def _parse_rate(value, *, header: str, row_number: int) -> float: + text = ("" if value is None else str(value)).replace(",", "").strip() + is_percent = text.endswith("%") + if is_percent: + text = text[:-1].strip() + try: + number = float(text) + except (TypeError, ValueError) as exc: + raise ContentReviewError(f"第 {row_number} 行“{header}”不是有效百分比") from exc + if is_percent or number > 1: + number /= 100 + if not math.isfinite(number) or not 0 <= number <= 1: + raise ContentReviewError(f"第 {row_number} 行“{header}”必须在 0% 到 100% 之间") + return round(number, 8) + + +def _parse_seconds(value, *, row_number: int) -> float: + if isinstance(value, (int, float)) and not isinstance(value, bool): + seconds = float(value) + else: + text = str(value or "").strip().lower().replace("秒", "").replace("s", "") + try: + seconds = float(text) + except ValueError as exc: + raise ContentReviewError(f"第 {row_number} 行“平均播放时长”不是有效秒数") from exc + if not math.isfinite(seconds) or not 0 <= seconds <= 86_400: + raise ContentReviewError(f"第 {row_number} 行“平均播放时长”超出合理范围") + return round(seconds, 4) + + +def _normalize_daily_rows(rows: list[list]) -> list[dict]: + if len(rows) < 2: + raise ContentReviewError("文件只有表头，没有可导入的数据行") + normalized_headers = [_normalize_header(value) for value in rows[0]] + duplicates = sorted({header for header in normalized_headers if normalized_headers.count(header) > 1}) + if duplicates: + raise ContentReviewError("表头存在重复列：" + "、".join(duplicates)) + missing = [header for header in DAILY_HEADERS if header not in normalized_headers] + if missing: + raise ContentReviewError("缺少必需表头：" + "、".join(missing)) + header_index = {header: normalized_headers.index(header) for header in DAILY_HEADERS} + + result: list[dict] = [] + seen_dates: set[str] = set() + for row_number, row in enumerate(rows[1:], start=2): + if not row or all(value is None or str(value).strip() == "" for value in row): + continue + + def value_for(header: str): + index = header_index[header] + return row[index] if index < len(row) else None + + metric_date = _parse_date(value_for("日期"), row_number=row_number) + if metric_date in seen_dates: + raise ContentReviewError(f"日期重复：{metric_date}（第 {row_number} 行）") + seen_dates.add(metric_date) + item = {"metric_date": metric_date} + for header, field in COUNT_FIELDS.items(): + item[field] = _parse_count(value_for(header), header=header, row_number=row_number) + for header, field in RATE_FIELDS.items(): + item[field] = _parse_rate(value_for(header), header=header, row_number=row_number) + item["average_watch_seconds"] = _parse_seconds( + value_for("平均播放时长"), + row_number=row_number, + ) + result.append(item) + if not result: + raise ContentReviewError("没有可导入的有效数据行") + result.sort(key=lambda item: item["metric_date"]) + return result + + +def preview_metric_import(*, account_id: str, filename: str, content: bytes) -> dict: + resolved_account_id = _resolve_douyin_account_id(account_id) + if not content: + raise ContentReviewError("请选择要导入的数据文件") + if len(content) > MAX_IMPORT_BYTES: + raise ContentReviewError("文件超过 10MB 限制") + safe_filename = Path(str(filename or "data")).name[:180] + extension = Path(safe_filename).suffix.lower() + if extension not in {".xlsx", ".csv"}: + raise ContentReviewError("仅支持 .xlsx 和 .csv；不支持旧 .xls、宏文件或其他格式") + rows = _load_xlsx_rows(content) if extension == ".xlsx" else _load_csv_rows(content) + normalized_rows = _normalize_daily_rows(rows) + source_sha256 = hashlib.sha256(content).hexdigest() + now = _now() + now_iso = now.isoformat(timespec="seconds") + expires_at = (now + timedelta(hours=PREVIEW_TTL_HOURS)).isoformat(timespec="seconds") + normalized_json = json.dumps(normalized_rows, ensure_ascii=False, separators=(",", ":")) + + with get_connection() as connection: + connection.execute("BEGIN IMMEDIATE") + existing = connection.execute( + """ + SELECT id, status, created_at, committed_at, expires_at + FROM content_metric_import_batches + WHERE account_id = ? AND source_kind = ? AND source_sha256 = ? + """, + (resolved_account_id, DAILY_SOURCE_KIND, source_sha256), + ).fetchone() + if existing is not None and existing["status"] == "committed": + connection.commit() + return { + "status": "already_imported", + "already_imported": True, + "message": "这个账号已经导入过相同文件，无需重复导入。", + "batch_id": existing["id"], + "account_id": resolved_account_id, + "row_count": len(normalized_rows), + "period_start": normalized_rows[0]["metric_date"], + "period_end": normalized_rows[-1]["metric_date"], + } + if existing is not None: + batch_id = str(existing["id"]) + connection.execute( + """ + UPDATE content_metric_import_batches + SET source_filename = ?, status = 'previewed', period_start = ?, period_end = ?, + normalized_payload_json = ?, row_count = ?, invalid_count = 0, + created_at = ?, committed_at = NULL, expires_at = ? + WHERE id = ? + """, + ( + safe_filename, + normalized_rows[0]["metric_date"], + normalized_rows[-1]["metric_date"], + normalized_json, + len(normalized_rows), + now_iso, + expires_at, + batch_id, + ), + ) + else: + batch_id = f"metric-{uuid4().hex[:16]}" + connection.execute( + """ + INSERT INTO content_metric_import_batches ( + id, account_id, source_kind, source_filename, source_sha256, status, + period_start, period_end, normalized_payload_json, row_count, + invalid_count, created_at, expires_at + ) VALUES (?, ?, ?, ?, ?, 'previewed', ?, ?, ?, ?, 0, ?, ?) + """, + ( + batch_id, + resolved_account_id, + DAILY_SOURCE_KIND, + safe_filename, + source_sha256, + normalized_rows[0]["metric_date"], + normalized_rows[-1]["metric_date"], + normalized_json, + len(normalized_rows), + now_iso, + expires_at, + ), + ) + connection.commit() + return { + "status": "previewed", + "already_imported": False, + "message": "预览校验通过；确认后才会写入账号级历史基线。", + "batch_id": batch_id, + "account_id": resolved_account_id, + "filename": safe_filename, + "row_count": len(normalized_rows), + "period_start": normalized_rows[0]["metric_date"], + "period_end": normalized_rows[-1]["metric_date"], + "expires_at": expires_at, + "sample_rows": normalized_rows[:5], + "attribution": "unattributed_historical_baseline", + } + + +def commit_metric_import(batch_id: str) -> dict: + now = _now() + now_iso = now.isoformat(timespec="seconds") + with get_connection() as connection: + connection.execute("BEGIN IMMEDIATE") + batch = connection.execute( + "SELECT * FROM content_metric_import_batches WHERE id = ?", + (batch_id,), + ).fetchone() + if batch is None: + connection.rollback() + raise ContentReviewError("导入预览不存在或已被清理", status_code=404) + if batch["status"] == "committed": + connection.commit() + return { + "status": "already_imported", + "already_imported": True, + "batch_id": batch_id, + "message": "这批数据已经确认导入。", + } + if batch["status"] != "previewed": + connection.rollback() + raise ContentReviewError("这批预览当前不能确认导入", status_code=409) + expires_at = datetime.fromisoformat(str(batch["expires_at"])) + if expires_at < now: + connection.execute( + "UPDATE content_metric_import_batches SET status = 'expired' WHERE id = ?", + (batch_id,), + ) + connection.commit() + raise ContentReviewError("预览已超过 24 小时，请重新选择文件预览", status_code=409) + try: + rows = json.loads(str(batch["normalized_payload_json"] or "[]")) + except json.JSONDecodeError as exc: + connection.rollback() + raise ContentReviewError("预览数据损坏，请重新上传文件", status_code=409) from exc + if not isinstance(rows, list) or not rows: + connection.rollback() + raise ContentReviewError("预览没有可导入数据，请重新上传文件", status_code=409) + for row in rows: + connection.execute( + """ + INSERT INTO douyin_account_daily_metric_snapshots ( + id, batch_id, account_id, metric_date, post_count, play_count, + like_count, share_count, comment_count, five_second_completion_rate, + two_second_bounce_rate, cover_click_rate, average_watch_seconds, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + f"daily-{uuid4().hex[:16]}", + batch_id, + batch["account_id"], + row["metric_date"], + row["post_count"], + row["play_count"], + row["like_count"], + row["share_count"], + row["comment_count"], + row["five_second_completion_rate"], + row["two_second_bounce_rate"], + row["cover_click_rate"], + row["average_watch_seconds"], + now_iso, + ), + ) + connection.execute( + """ + UPDATE content_metric_import_batches + SET status = 'committed', committed_at = ?, expires_at = NULL + WHERE id = ? + """, + (now_iso, batch_id), + ) + connection.commit() + return { + "status": "committed", + "already_imported": False, + "batch_id": batch_id, + "row_count": len(rows), + "message": f"已导入 {len(rows)} 天账号级数据，并标记为未归因历史基线。", + "attribution": "unattributed_historical_baseline", + } + + +def list_import_batches(account_id: str = "", limit: int = 20) -> list[dict]: + resolved = _resolve_douyin_account_id(account_id) + safe_limit = max(1, min(100, int(limit))) + with get_connection() as connection: + rows = connection.execute( + """ + SELECT id, account_id, source_kind, source_filename, source_sha256, status, + period_start, period_end, row_count, matched_count, ambiguous_count, + invalid_count, created_at, committed_at, expires_at + FROM content_metric_import_batches + WHERE account_id = ? + ORDER BY created_at DESC LIMIT ? + """, + (resolved, safe_limit), + ).fetchall() + return [dict(row) for row in rows] + + +def _aggregate_period(rows: list[dict]) -> dict: + if not rows: + return { + "post_count": 0, + "play_count": 0, + "interaction_count": 0, + "five_second_completion_rate": None, + "two_second_bounce_rate": None, + "cover_click_rate": None, + "average_watch_seconds": None, + } + rate_fields = ( + "five_second_completion_rate", + "two_second_bounce_rate", + "cover_click_rate", + "average_watch_seconds", + ) + result = { + "post_count": sum(int(row["post_count"] or 0) for row in rows), + "play_count": sum(int(row["play_count"] or 0) for row in rows), + "interaction_count": sum( + int(row["like_count"] or 0) + + int(row["share_count"] or 0) + + int(row["comment_count"] or 0) + for row in rows + ), + } + for field in rate_fields: + values = [float(row[field]) for row in rows if row[field] is not None] + result[field] = round(sum(values) / len(values), 6) if values else None + return result + + +def _comparison_delta(current, previous) -> float | None: + if current is None or previous in (None, 0): + return None + return round((float(current) - float(previous)) / abs(float(previous)), 6) + + +def get_content_review_summary(account_id: str = "", days: int = 28) -> dict: + resolved = _resolve_douyin_account_id(account_id) + safe_days = max(14, min(180, int(days))) + with get_connection() as connection: + rows = connection.execute( + """ + WITH ranked AS ( + SELECT s.*, b.committed_at, + ROW_NUMBER() OVER ( + PARTITION BY s.metric_date + ORDER BY b.committed_at DESC, s.created_at DESC, s.rowid DESC + ) AS metric_rank + FROM douyin_account_daily_metric_snapshots s + JOIN content_metric_import_batches b ON b.id = s.batch_id + WHERE s.account_id = ? AND b.status = 'committed' + ) + SELECT * FROM ranked WHERE metric_rank = 1 ORDER BY metric_date ASC + """, + (resolved,), + ).fetchall() + sync_row = connection.execute( + """ + SELECT MAX(committed_at) AS last_sync_at, + COUNT(*) AS completed_cycles + FROM content_metric_import_batches + WHERE account_id = ? AND status = 'committed' + """, + (resolved,), + ).fetchone() + history = [dict(row) for row in rows] + if not history: + return { + "account_id": resolved, + "has_data": False, + "message": "还没有确认导入的账号级数据。", + "last_sync_at": None, + "days_since_sync": None, + "completed_cycles": 0, + "current_period": _aggregate_period([]), + "previous_period": _aggregate_period([]), + "comparisons": {}, + "history": [], + } + + latest_date = date.fromisoformat(history[-1]["metric_date"]) + current_start = latest_date - timedelta(days=6) + previous_end = current_start - timedelta(days=1) + previous_start = previous_end - timedelta(days=6) + current_rows = [row for row in history if current_start <= date.fromisoformat(row["metric_date"]) <= latest_date] + previous_rows = [ + row for row in history if previous_start <= date.fromisoformat(row["metric_date"]) <= previous_end + ] + current = _aggregate_period(current_rows) + previous = _aggregate_period(previous_rows) + comparisons = { + key: _comparison_delta(current.get(key), previous.get(key)) + for key in current + } + last_sync_at = sync_row["last_sync_at"] if sync_row else None + days_since = None + if last_sync_at: + days_since = max(0, (_now().date() - datetime.fromisoformat(str(last_sync_at)).date()).days) + history_start = latest_date - timedelta(days=safe_days - 1) + visible_history = [row for row in history if date.fromisoformat(row["metric_date"]) >= history_start] + return { + "account_id": resolved, + "has_data": True, + "latest_metric_date": latest_date.isoformat(), + "last_sync_at": last_sync_at, + "days_since_sync": days_since, + "completed_cycles": int(sync_row["completed_cycles"] or 0), + "current_period": current, + "previous_period": previous, + "comparisons": comparisons, + "history": visible_history, + "attribution": "unattributed_account_daily_baseline", + } + + +def list_content_review_works(account_id: str = "", limit: int = 100) -> list[dict]: + resolved = _resolve_douyin_account_id(account_id) + safe_limit = max(1, min(200, int(limit))) + with get_connection() as connection: + rows = connection.execute( + """ + WITH latest_items AS ( + SELECT i.*, + ROW_NUMBER() OVER ( + PARTITION BY i.aweme_id + ORDER BY i.captured_at DESC, i.created_at DESC, i.rowid DESC + ) AS item_rank + FROM douyin_item_metric_snapshots i + JOIN content_metric_import_batches b ON b.id = i.batch_id + WHERE i.account_id = ? AND b.status = 'committed' + ) + SELECT i.*, pj.title AS publish_title, pj.published_at AS job_published_at, + pj.status AS publish_status, pj.output_clip_id, oc.clip_candidate_id, + c.title AS candidate_title, c.source_analysis_run_id, + ar.run_number AS analysis_run_number, ar.provider_label, ar.model, + ar.prompt_version_id, pv.version_number AS prompt_version_number, + pv.preset_name_snapshot AS prompt_name, + ( + SELECT f.decision FROM clip_feedback f + WHERE f.clip_candidate_id = c.id + ORDER BY f.created_at DESC, f.rowid DESC LIMIT 1 + ) AS review_decision, + ( + SELECT f.reason_code FROM clip_feedback f + WHERE f.clip_candidate_id = c.id + ORDER BY f.created_at DESC, f.rowid DESC LIMIT 1 + ) AS review_reason_code + FROM latest_items i + LEFT JOIN publish_jobs pj ON pj.id = i.publish_job_id + LEFT JOIN output_clip oc ON oc.id = pj.output_clip_id + LEFT JOIN clip_candidates c ON c.id = oc.clip_candidate_id + LEFT JOIN ai_analysis_runs ar ON ar.id = c.source_analysis_run_id + LEFT JOIN ai_prompt_versions pv ON pv.id = ar.prompt_version_id + WHERE i.item_rank = 1 + ORDER BY COALESCE(i.published_at, i.captured_at) DESC + LIMIT ? + """, + (resolved, safe_limit), + ).fetchall() + works = [] + for row in rows: + work = dict(row) + work["review_reason_label"] = REJECT_REASON_LABELS.get( + str(work.get("review_reason_code") or ""), + str(work.get("review_reason_code") or ""), + ) + work["attribution_complete"] = bool( + work.get("publish_job_id") + and work.get("clip_candidate_id") + and work.get("source_analysis_run_id") + and work.get("prompt_version_id") + ) + works.append(work) + return works + + +def _safe_ratio(numerator: float, denominator: float) -> float | None: + if not denominator: + return None + return round(numerator / denominator, 6) + + +def get_prompt_comparison(account_id: str = "") -> dict: + resolved = _resolve_douyin_account_id(account_id) + with get_connection() as connection: + candidate_rows = connection.execute( + """ + SELECT pv.id AS prompt_version_id, pv.preset_id, pv.version_number, + pv.preset_name_snapshot, pv.created_at, + c.id AS candidate_id, c.enabled, + ( + SELECT f.decision FROM clip_feedback f + WHERE f.clip_candidate_id = c.id + ORDER BY f.created_at DESC, f.rowid DESC LIMIT 1 + ) AS latest_decision + FROM ai_prompt_versions pv + LEFT JOIN ai_analysis_runs ar ON ar.prompt_version_id = pv.id + LEFT JOIN clip_candidates c ON c.source_analysis_run_id = ar.id AND c.is_deleted = 0 + ORDER BY pv.created_at, pv.preset_id, pv.version_number + """ + ).fetchall() + work_rows = connection.execute( + """ + WITH latest_items AS ( + SELECT i.*, + ROW_NUMBER() OVER ( + PARTITION BY i.aweme_id + ORDER BY i.captured_at DESC, i.created_at DESC, i.rowid DESC + ) AS item_rank + FROM douyin_item_metric_snapshots i + JOIN content_metric_import_batches b ON b.id = i.batch_id + WHERE i.account_id = ? AND b.status = 'committed' + AND i.match_status IN ('matched_exact', 'matched_unique', 'confirmed_manual') + ) + SELECT i.*, c.id AS candidate_id, ar.prompt_version_id + FROM latest_items i + JOIN publish_jobs pj ON pj.id = i.publish_job_id + JOIN output_clip oc ON oc.id = pj.output_clip_id + JOIN clip_candidates c ON c.id = oc.clip_candidate_id + JOIN ai_analysis_runs ar ON ar.id = c.source_analysis_run_id + WHERE i.item_rank = 1 AND ar.prompt_version_id IS NOT NULL + """, + (resolved,), + ).fetchall() + cycle_row = connection.execute( + """ + SELECT COUNT(*) FROM content_metric_import_batches + WHERE account_id = ? AND status = 'committed' + """, + (resolved,), + ).fetchone() + + groups: dict[str, dict] = {} + for row in candidate_rows: + version_id = str(row["prompt_version_id"]) + group = groups.setdefault( + version_id, + { + "prompt_version_id": version_id, + "preset_id": row["preset_id"], + "version_number": int(row["version_number"]), + "prompt_name": row["preset_name_snapshot"], + "created_at": row["created_at"], + "candidate_ids": set(), + "kept_ids": set(), + "published_candidate_ids": set(), + "works": [], + }, + ) + if row["candidate_id"]: + candidate_id = str(row["candidate_id"]) + group["candidate_ids"].add(candidate_id) + if row["latest_decision"] == "keep" or ( + row["latest_decision"] is None and int(row["enabled"] or 0) == 1 + ): + group["kept_ids"].add(candidate_id) + + for row in work_rows: + version_id = str(row["prompt_version_id"]) + group = groups.get(version_id) + if group is None: + continue + group["works"].append(dict(row)) + group["published_candidate_ids"].add(str(row["candidate_id"])) + + versions = [] + for group in groups.values(): + candidate_count = len(group["candidate_ids"]) + works = group["works"] + plays = [int(work["play_count"] or 0) for work in works] + five_rates = [float(work["five_second_completion_rate"]) for work in works if work["five_second_completion_rate"] is not None] + bounce_rates = [float(work["two_second_bounce_rate"]) for work in works if work["two_second_bounce_rate"] is not None] + watch_ratios = [ + float(work["average_watch_seconds"]) / float(work["duration_seconds"]) + for work in works + if work["average_watch_seconds"] is not None and float(work["duration_seconds"] or 0) > 0 + ] + interactions = sum( + int(work["like_count"] or 0) + + int(work["comment_count"] or 0) + + int(work["share_count"] or 0) + for work in works + ) + total_plays = sum(plays) + versions.append( + { + "prompt_version_id": group["prompt_version_id"], + "preset_id": group["preset_id"], + "version_number": group["version_number"], + "prompt_name": group["prompt_name"], + "created_at": group["created_at"], + "candidate_count": candidate_count, + "accurate_published_count": len(works), + "keep_rate": _safe_ratio(len(group["kept_ids"]), candidate_count), + "publish_rate": _safe_ratio(len(group["published_candidate_ids"]), candidate_count), + "median_play_count": statistics.median(plays) if plays else None, + "five_second_completion_rate": statistics.fmean(five_rates) if five_rates else None, + "two_second_bounce_rate": statistics.fmean(bounce_rates) if bounce_rates else None, + "average_watch_ratio": statistics.fmean(watch_ratios) if watch_ratios else None, + "interaction_rate": _safe_ratio(interactions, total_plays), + "evaluable": len(works) >= 30, + } + ) + versions.sort(key=lambda item: (item["created_at"], item["preset_id"], item["version_number"])) + comparisons = [] + for previous, current in zip(versions, versions[1:], strict=False): + if previous["accurate_published_count"] < 20 or current["accurate_published_count"] < 20: + continue + comparisons.append( + { + "from_prompt_version_id": previous["prompt_version_id"], + "to_prompt_version_id": current["prompt_version_id"], + "note": "仅相关性，不代表因果；请结合选题、发布时间和样本结构判断。", + } + ) + completed_cycles = int(cycle_row[0] or 0) if cycle_row else 0 + return { + "account_id": resolved, + "completed_cycles": completed_cycles, + "minimum_cycles": 3, + "can_review_prompt": completed_cycles >= 3 and any(item["evaluable"] for item in versions), + "versions": versions, + "comparisons": comparisons, + "message": ( + "已达到评估门槛，系统只提供建议，不会自动修改 Prompt。" + if completed_cycles >= 3 and any(item["evaluable"] for item in versions) + else "数据不足：默认需要 3 个完整周期，且当前 Prompt 至少 30 条准确关联作品。" + ), + "causality_notice": "所有对比仅表示相关性，不代表因果。", + } + + +def set_item_match(snapshot_id: str, publish_job_id: str) -> dict: + now = _now_iso() + with get_connection() as connection: + connection.execute("BEGIN IMMEDIATE") + snapshot = connection.execute( + "SELECT id, account_id FROM douyin_item_metric_snapshots WHERE id = ?", + (snapshot_id,), + ).fetchone() + if snapshot is None: + connection.rollback() + raise ContentReviewError("作品指标记录不存在", status_code=404) + job = connection.execute( + """ + SELECT id FROM publish_jobs + WHERE id = ? AND platform = 'douyin' AND account_id = ? + """, + (publish_job_id, snapshot["account_id"]), + ).fetchone() + if job is None: + connection.rollback() + raise ContentReviewError("发布记录不存在，或不属于同一个抖音账号", status_code=409) + connection.execute( + """ + UPDATE douyin_item_metric_snapshots + SET publish_job_id = ?, match_status = 'confirmed_manual', + match_method = 'manual_confirmation', created_at = ? + WHERE id = ? + """, + (publish_job_id, now, snapshot_id), + ) + connection.commit() + return {"status": "ok", "snapshot_id": snapshot_id, "publish_job_id": publish_job_id} + + +def delete_item_match(snapshot_id: str) -> dict: + with get_connection() as connection: + cursor = connection.execute( + """ + UPDATE douyin_item_metric_snapshots + SET publish_job_id = NULL, match_status = 'unmatched', match_method = NULL + WHERE id = ? + """, + (snapshot_id,), + ) + connection.commit() + if cursor.rowcount != 1: + raise ContentReviewError("作品指标记录不存在", status_code=404) + return {"status": "ok", "snapshot_id": snapshot_id, "message": "已解除错误关联。"} + + +def normalize_title(value: str) -> str: + return re.sub(r"[^0-9a-z\u4e00-\u9fff]+", "", str(value or "").lower()) + + +def match_douyin_item_with_connection(connection, *, account_id: str, item: dict) -> dict: + aweme_id = str(item.get("aweme_id") or "").strip() + exact_rows = connection.execute( + """ + SELECT id FROM publish_jobs + WHERE platform = 'douyin' AND account_id = ? + AND (platform_item_id = ? OR remote_video_id = ?) + """, + (account_id, aweme_id, aweme_id), + ).fetchall() + if len(exact_rows) == 1: + return {"status": "matched_exact", "method": "platform_item_id", "publish_job_id": exact_rows[0]["id"]} + if len(exact_rows) > 1: + return {"status": "ambiguous", "method": "platform_item_id", "publish_job_id": None} + + normalized_title = normalize_title(str(item.get("title") or "")) + published_at = item.get("published_at") + if not normalized_title or not published_at: + return {"status": "unmatched", "method": None, "publish_job_id": None} + try: + item_time = datetime.fromisoformat(str(published_at).replace("Z", "+00:00")) + except ValueError: + return {"status": "unmatched", "method": None, "publish_job_id": None} + candidates = connection.execute( + """ + SELECT pj.id, pj.title, pj.published_at, oc.source_duration_ms + FROM publish_jobs pj + LEFT JOIN output_clip oc ON oc.id = pj.output_clip_id + WHERE pj.platform = 'douyin' AND pj.account_id = ? AND pj.published_at IS NOT NULL + """, + (account_id,), + ).fetchall() + matches = [] + for candidate in candidates: + if normalize_title(str(candidate["title"] or "")) != normalized_title: + continue + try: + candidate_time = datetime.fromisoformat(str(candidate["published_at"]).replace("Z", "+00:00")) + seconds_apart = abs((candidate_time - item_time).total_seconds()) + except (TypeError, ValueError): + continue + if seconds_apart > 600: + continue + item_duration = float(item.get("duration_seconds") or 0) + candidate_duration = float(candidate["source_duration_ms"] or 0) / 1000 + if item_duration and candidate_duration and abs(item_duration - candidate_duration) > 3: + continue + matches.append(candidate) + if len(matches) == 1: + return {"status": "matched_unique", "method": "title_time_duration", "publish_job_id": matches[0]["id"]} + if len(matches) > 1: + return {"status": "ambiguous", "method": "title_time_duration", "publish_job_id": None} + return {"status": "unmatched", "method": None, "publish_job_id": None} + + +def stage_douyin_item_sync(*, account_id: str, items: list[dict], captured_at: str) -> dict: + resolved = _resolve_douyin_account_id(account_id) + if len(items) > 50: + raise ContentReviewError("单次最多同步最近 50 条作品") + normalized_items = [] + seen_ids = set() + allowed_count_fields = ("play_count", "like_count", "comment_count", "share_count", "collect_count") + allowed_rate_fields = ( + "five_second_completion_rate", + "two_second_bounce_rate", + "cover_click_rate", + "average_watch_seconds", + ) + for raw in items: + aweme_id = str(raw.get("aweme_id") or "").strip() + if not aweme_id or aweme_id in seen_ids: + continue + seen_ids.add(aweme_id) + item = { + "aweme_id": aweme_id[:120], + "title": str(raw.get("title") or "")[:240], + "published_at": raw.get("published_at"), + "duration_seconds": raw.get("duration_seconds"), + } + for field in allowed_count_fields: + value = raw.get(field) + item[field] = max(0, int(value)) if value is not None else None + for field in allowed_rate_fields: + value = raw.get(field) + item[field] = float(value) if value is not None else None + normalized_items.append(item) + payload_json = json.dumps(normalized_items, ensure_ascii=False, separators=(",", ":")) + payload_hash = hashlib.sha256(payload_json.encode("utf-8")).hexdigest() + batch_id = f"sync-{uuid4().hex[:16]}" + now_iso = _now_iso() + matched = ambiguous = 0 + with get_connection() as connection: + connection.execute("BEGIN IMMEDIATE") + existing = connection.execute( + """ + SELECT id FROM content_metric_import_batches + WHERE account_id = ? AND source_kind = ? AND source_sha256 = ? + """, + (resolved, DOUYIN_SYNC_SOURCE_KIND, payload_hash), + ).fetchone() + if existing is not None: + connection.commit() + return { + "status": "already_imported", + "batch_id": existing["id"], + "message": "最近作品指标没有变化，无需重复保存。", + } + connection.execute( + """ + INSERT INTO content_metric_import_batches ( + id, account_id, source_kind, source_filename, source_sha256, status, + normalized_payload_json, row_count, created_at, committed_at + ) VALUES (?, ?, ?, 'douyin-worker', ?, 'committed', ?, ?, ?, ?) + """, + (batch_id, resolved, DOUYIN_SYNC_SOURCE_KIND, payload_hash, payload_json, len(normalized_items), now_iso, now_iso), + ) + for item in normalized_items: + match = match_douyin_item_with_connection(connection, account_id=resolved, item=item) + if match["status"] in MATCHED_STATUSES: + matched += 1 + elif match["status"] == "ambiguous": + ambiguous += 1 + connection.execute( + """ + INSERT INTO douyin_item_metric_snapshots ( + id, batch_id, publish_job_id, account_id, aweme_id, title, + published_at, duration_seconds, captured_at, play_count, like_count, + comment_count, share_count, collect_count, five_second_completion_rate, + two_second_bounce_rate, cover_click_rate, average_watch_seconds, + match_status, match_method, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + f"item-{uuid4().hex[:16]}", + batch_id, + match["publish_job_id"], + resolved, + item["aweme_id"], + item["title"], + item["published_at"], + item["duration_seconds"], + captured_at, + item["play_count"], + item["like_count"], + item["comment_count"], + item["share_count"], + item["collect_count"], + item["five_second_completion_rate"], + item["two_second_bounce_rate"], + item["cover_click_rate"], + item["average_watch_seconds"], + match["status"], + match["method"], + now_iso, + ), + ) + connection.execute( + """ + UPDATE content_metric_import_batches + SET matched_count = ?, ambiguous_count = ? WHERE id = ? + """, + (matched, ambiguous, batch_id), + ) + connection.commit() + return { + "status": "committed", + "batch_id": batch_id, + "row_count": len(normalized_items), + "matched_count": matched, + "ambiguous_count": ambiguous, + "unmatched_count": len(normalized_items) - matched - ambiguous, + "message": f"已同步 {len(normalized_items)} 条作品指标；存在歧义的记录等待人工确认。", + } diff --git a/app/services/database_backup_service.py b/app/services/database_backup_service.py index 68d1c99..4ea8c8e 100644 --- a/app/services/database_backup_service.py +++ b/app/services/database_backup_service.py @@ -5,6 +5,7 @@ from dataclasses import dataclass from datetime import datetime, timedelta from pathlib import Path +from typing import Any from uuid import uuid4 from zoneinfo import ZoneInfo @@ -70,6 +71,140 @@ def sqlite_quick_check(database_path: Path) -> str: connection.close() +def sqlite_diagnostic_report( + database_path: Path, + *, + deep: bool = False, +) -> dict[str, Any]: + """只读检查数据库、迁移账本及已应用迁移的不变量。""" + path = database_path.resolve() + report: dict[str, Any] = { + "status": "ok", + "database_path": str(path), + "deep": bool(deep), + "readable": False, + "integrity_check": "skipped", + "foreign_key_violation_count": 0, + "foreign_key_violations": [], + "migration_count": 0, + "migration_errors": [], + "errors": [], + } + errors: list[str] = report["errors"] + migration_errors: list[str] = report["migration_errors"] + if not path.is_file(): + errors.append("数据库文件不存在") + report["status"] = "error" + return report + + connection: sqlite3.Connection | None = None + try: + connection = sqlite3.connect( + f"{path.as_uri()}?mode=ro", + uri=True, + timeout=3, + ) + connection.row_factory = sqlite3.Row + connection.execute("PRAGMA query_only = ON") + connection.execute("SELECT 1").fetchone() + report["readable"] = True + + table_info = connection.execute( + "PRAGMA table_info(schema_migrations)" + ).fetchall() + required_ledger_columns = {"version", "name", "checksum", "applied_at"} + actual_ledger_columns = {str(row["name"]) for row in table_info} + missing_columns = sorted(required_ledger_columns - actual_ledger_columns) + if not table_info: + migration_errors.append("缺少 schema_migrations 迁移账本") + elif missing_columns: + migration_errors.append( + "迁移账本缺少字段：" + ", ".join(missing_columns) + ) + else: + version_column = next( + row for row in table_info if row["name"] == "version" + ) + if int(version_column["pk"] or 0) != 1: + migration_errors.append("迁移账本 version 字段不是主键") + + ledger_rows = connection.execute( + "SELECT version, name, checksum FROM schema_migrations ORDER BY version" + ).fetchall() + report["migration_count"] = len(ledger_rows) + + # 延迟导入可避免 database.py 在加载本服务时产生循环导入。 + from app.db import database as database_module + + registered = { + migration.version: migration + for migration in database_module._registered_schema_migrations() + } + for row in ledger_rows: + version = str(row["version"]) + migration = registered.get(version) + if migration is None: + migration_errors.append(f"存在当前程序无法识别的迁移：{version}") + continue + if str(row["name"]) != migration.name: + migration_errors.append(f"迁移 {version} 的名称与程序定义不一致") + if str(row["checksum"]) != migration.checksum: + migration_errors.append(f"迁移 {version} 的 checksum 与程序定义不一致") + continue + try: + migration.verify(connection) + except Exception as exc: + migration_errors.append(f"迁移 {version} 校验失败：{exc}") + + if deep: + integrity_rows = connection.execute("PRAGMA integrity_check").fetchall() + integrity_messages = [str(row[0]) for row in integrity_rows] + report["integrity_check"] = ( + "ok" if integrity_messages == ["ok"] else "; ".join(integrity_messages) + ) + if report["integrity_check"] != "ok": + errors.append(f"integrity_check 失败：{report['integrity_check']}") + + foreign_key_rows = connection.execute("PRAGMA foreign_key_check").fetchall() + report["foreign_key_violation_count"] = len(foreign_key_rows) + report["foreign_key_violations"] = [ + { + "table": str(row[0]), + "rowid": row[1], + "parent": str(row[2]), + "foreign_key_id": row[3], + } + for row in foreign_key_rows[:20] + ] + if foreign_key_rows: + errors.append(f"foreign_key_check 发现 {len(foreign_key_rows)} 条异常") + except sqlite3.Error as exc: + errors.append(f"数据库读取失败：{exc}") + except Exception as exc: + errors.append(f"数据库诊断失败：{exc}") + finally: + if connection is not None: + connection.close() + + errors.extend(migration_errors) + if errors: + report["status"] = "error" + return report + + +def assert_sqlite_database_ready( + database_path: Path, + *, + deep: bool = True, + label: str = "数据库", +) -> dict[str, Any]: + report = sqlite_diagnostic_report(database_path, deep=deep) + if report["status"] != "ok": + details = "；".join(str(item) for item in report["errors"]) + raise BackupSafetyError(f"{label}校验失败：{details}") + return report + + def _ensure_safe_backup_path(path: Path, backup_dir: Path) -> None: resolved_dir = backup_dir.resolve() absolute_parent = Path(os.path.abspath(path.parent)) diff --git a/app/services/publishers/worker_client.py b/app/services/publishers/worker_client.py index 56db7d2..6875778 100644 --- a/app/services/publishers/worker_client.py +++ b/app/services/publishers/worker_client.py @@ -69,9 +69,21 @@ def _request(self, method: str, path: str, payload: dict[str, Any] | None = None detail = json.loads(raw).get("detail") or raw except json.JSONDecodeError: detail = raw or str(exc) + detail_code = "" + if isinstance(detail, dict): + detail_code = str(detail.get("error_code") or "") + detail = str(detail.get("message") or detail_code or "Worker 拒绝了请求") if exc.code in {401, 403}: raise PublishError(str(detail), "publish_worker_unauthorized") from exc - if exc.code in {409, 422}: + if detail_code in { + "LOGIN_REQUIRED", + "VERIFICATION_REQUIRED", + "RATE_LIMITED", + "PAGE_CHANGED", + "WORKER_UNAVAILABLE", + }: + raise PublishError(str(detail), detail_code) from exc + if exc.code in {409, 422, 429}: raise PublishError(str(detail), "publish_worker_rejected") from exc raise PublishWorkerUnavailable( f"Windows 发布 Worker 返回 HTTP {exc.code}：{detail}", @@ -121,6 +133,15 @@ def open_creator_center(self, platform: str, account_id: str) -> dict[str, Any]: account_id = validate_worker_identifier(account_id, "account_id", max_length=120) return self._request("POST", "/v1/accounts/open-center", {"platform": platform, "account_id": account_id}) + def analytics_sync(self, account_id: str, limit: int = 50) -> dict[str, Any]: + account_id = validate_worker_identifier(account_id, "account_id", max_length=120) + safe_limit = max(1, min(50, int(limit))) + return self._request( + "POST", + "/v1/analytics/douyin/sync", + {"account_id": account_id, "limit": safe_limit}, + ) + def publish(self, payload: dict[str, Any]) -> PublishResult: safe_payload = dict(payload) safe_payload["job_id"] = validate_worker_identifier( diff --git a/app/services/system_readiness_service.py b/app/services/system_readiness_service.py new file mode 100644 index 0000000..4e05ce2 --- /dev/null +++ b/app/services/system_readiness_service.py @@ -0,0 +1,96 @@ +from __future__ import annotations + +from datetime import datetime +import os +from pathlib import Path +import shutil +from typing import Any + +from app.core.config import settings +from app.services.database_backup_service import sqlite_diagnostic_report +from app.services.publish_scheduler import scheduler_health + + +def _directory_readiness(path: Path, label: str) -> dict[str, Any]: + resolved = path.resolve() + exists = resolved.is_dir() + readable = exists and os.access(resolved, os.R_OK) + writable = exists and os.access(resolved, os.W_OK) + return { + "label": label, + "path": str(resolved), + "status": "ok" if readable and writable else "error", + "exists": exists, + "readable": readable, + "writable": writable, + } + + +def build_system_readiness(*, deep: bool = False) -> dict[str, Any]: + """汇总启动就绪条件；只读检查，不创建目录也不修复状态。""" + database = sqlite_diagnostic_report(settings.database_path, deep=deep) + storage = [ + _directory_readiness(settings.data_dir, "数据目录"), + _directory_readiness(settings.tasks_dir, "任务存储目录"), + ] + + scheduler: dict[str, Any] + try: + scheduler = scheduler_health() + scheduler["status"] = "ok" + if scheduler.get("enabled") and ( + not scheduler.get("running") or scheduler.get("last_error_code") + ): + scheduler["status"] = "degraded" + except Exception as exc: + scheduler = { + "status": "error", + "worker_available": False, + "message": f"调度状态读取失败：{exc}", + } + + ffmpeg_path = shutil.which("ffmpeg") + ffmpeg = { + "status": "ok" if ffmpeg_path else "degraded", + "available": bool(ffmpeg_path), + "path": ffmpeg_path or "", + } + + critical_errors: list[str] = [] + degraded_reasons: list[str] = [] + if database["status"] != "ok": + critical_errors.extend(str(item) for item in database["errors"]) + for item in storage: + if item["status"] != "ok": + critical_errors.append(f"{item['label']}不可读写") + if scheduler["status"] == "error": + critical_errors.append(str(scheduler.get("message") or "调度状态异常")) + elif scheduler["status"] == "degraded": + degraded_reasons.append("发布调度未运行或存在错误") + if not scheduler.get("worker_available", False): + degraded_reasons.append( + str(scheduler.get("worker_message") or "Windows 发布 Worker 不可用") + ) + if not ffmpeg_path: + degraded_reasons.append("FFmpeg 不可用") + + if critical_errors: + status = "not_ready" + elif degraded_reasons: + status = "degraded" + else: + status = "ready" + + return { + "status": status, + "deep": bool(deep), + "checked_at": datetime.now().astimezone().isoformat(timespec="seconds"), + "critical_errors": critical_errors, + "degraded_reasons": degraded_reasons, + "checks": { + "database": database, + "storage": storage, + "scheduler": scheduler, + "ffmpeg": ffmpeg, + }, + } diff --git a/app/services/task_service.py b/app/services/task_service.py index 8805e82..a4ae503 100644 --- a/app/services/task_service.py +++ b/app/services/task_service.py @@ -43,7 +43,10 @@ queue_task_ai_analysis, restore_ai_analysis_run, ) -from app.services.clip_feedback_service import save_clip_feedback +from app.services.clip_feedback_service import ( + record_review_toggle_feedback_with_connection, + save_clip_feedback, +) from app.services.storage_service import ( get_artifact_paths, get_source_video_path, @@ -897,14 +900,21 @@ def list_clip_candidates(task_id: str) -> list[dict]: with get_connection() as connection: rows = connection.execute( """ - SELECT id, task_id, clip_key, title, start_time, end_time, duration_seconds, cover_time_seconds, + SELECT c.id, c.task_id, c.clip_key, c.title, c.start_time, c.end_time, + c.duration_seconds, c.cover_time_seconds, summary, reason, highlight_reason, spread_value, suggested_editing, confidence_score, quality_tier, quality_score, text_quality_score, humor_score, completeness_score, audio_reaction_score, topic_key, key_moment_time, quality_evidence_json, rejection_reason, - selected_by_default, enabled, reviewed, is_deleted, deleted_at, created_at, updated_at - FROM clip_candidates - WHERE task_id = ? AND is_deleted = 0 - ORDER BY start_time ASC + selected_by_default, enabled, reviewed, source_analysis_run_id, + is_deleted, deleted_at, c.created_at, c.updated_at, + ( + SELECT f.reason_code FROM clip_feedback f + WHERE f.task_id = c.task_id AND f.clip_candidate_id = c.id + ORDER BY f.created_at DESC, f.rowid DESC LIMIT 1 + ) AS feedback_reason_code + FROM clip_candidates c + WHERE c.task_id = ? AND c.is_deleted = 0 + ORDER BY c.start_time ASC """, (task_id,), ).fetchall() @@ -939,6 +949,12 @@ def list_clip_candidates(task_id: str) -> list[dict]: "key_moment_time": clip.get("key_moment_time") or "", "quality_evidence": quality_evidence, "rejection_reason": clip.get("rejection_reason") or "", + "feedback_reason_code": ( + clip.get("feedback_reason_code") + if clip.get("feedback_reason_code") + in {"not_funny", "fragmented", "missing_setup", "duplicate", "dragging", "other"} + else "" + ), "ai_source_label": ai_source_label, "selected_by_default": bool(clip.get("selected_by_default")), "enabled": bool(clip.get("enabled")), @@ -1035,6 +1051,18 @@ def update_clip_candidate(task_id: str, clip_id: str, payload: ClipCandidateUpda data = _validate_clip_update(task, payload) now = _now_iso() with get_connection() as connection: + connection.execute("BEGIN IMMEDIATE") + current = connection.execute( + """ + SELECT id, source_analysis_run_id + FROM clip_candidates + WHERE id = ? AND task_id = ? AND is_deleted = 0 + """, + (clip_id, task_id), + ).fetchone() + if current is None: + connection.rollback() + raise ValueError("候选片段不存在") cursor = connection.execute( """ UPDATE clip_candidates @@ -1054,6 +1082,19 @@ def update_clip_candidate(task_id: str, clip_id: str, payload: ClipCandidateUpda task_id, ), ) + record_review_toggle_feedback_with_connection( + connection, + task_id=task_id, + clip={ + "id": clip_id, + "source_analysis_run_id": current["source_analysis_run_id"], + **data, + }, + selection_profile=task.get("selection_profile") or "general", + enabled=bool(data["enabled"]), + reason_code=payload.feedback_reason_code, + now=now, + ) connection.execute("UPDATE tasks SET updated_at = ? WHERE id = ?", (now, task_id)) connection.commit() @@ -1072,61 +1113,83 @@ def update_clip_candidates_batch(task_id: str, payloads: list[ClipCandidateBatch validated = [] for payload in payloads: - validated.append((payload.id, _validate_clip_update(task, payload))) + validated.append((payload.id, _validate_clip_update(task, payload), payload.feedback_reason_code)) now = _now_iso() changed_count = 0 + feedback_count = 0 with get_connection() as connection: - for clip_id, data in validated: - current = connection.execute( - """ - SELECT title, start_time, end_time, duration_seconds, enabled, summary - FROM clip_candidates - WHERE id = ? AND task_id = ? AND is_deleted = 0 - """, - (clip_id, task_id), - ).fetchone() - if current is None: - raise ValueError(f"候选片段不存在：{clip_id}") - if any( - ( - str(current["title"] or "") != data["title"], - str(current["start_time"] or "") != data["start_time"], - str(current["end_time"] or "") != data["end_time"], - int(current["duration_seconds"] or 0) != int(data["duration_seconds"]), - int(current["enabled"] or 0) != int(data["enabled"]), - str(current["summary"] or "") != data["summary"], + connection.execute("BEGIN IMMEDIATE") + try: + for clip_id, data, feedback_reason_code in validated: + current = connection.execute( + """ + SELECT title, start_time, end_time, duration_seconds, enabled, summary, + source_analysis_run_id + FROM clip_candidates + WHERE id = ? AND task_id = ? AND is_deleted = 0 + """, + (clip_id, task_id), + ).fetchone() + if current is None: + raise ValueError(f"候选片段不存在：{clip_id}") + if any( + ( + str(current["title"] or "") != data["title"], + str(current["start_time"] or "") != data["start_time"], + str(current["end_time"] or "") != data["end_time"], + int(current["duration_seconds"] or 0) != int(data["duration_seconds"]), + int(current["enabled"] or 0) != int(data["enabled"]), + str(current["summary"] or "") != data["summary"], + ) + ): + changed_count += 1 + cursor = connection.execute( + """ + UPDATE clip_candidates + SET title = ?, start_time = ?, end_time = ?, duration_seconds = ?, + enabled = ?, summary = ?, reviewed = 1, updated_at = ? + WHERE id = ? AND task_id = ? AND is_deleted = 0 + """, + ( + data["title"], + data["start_time"], + data["end_time"], + data["duration_seconds"], + data["enabled"], + data["summary"], + now, + clip_id, + task_id, + ), ) - ): - changed_count += 1 - cursor = connection.execute( - """ - UPDATE clip_candidates - SET title = ?, start_time = ?, end_time = ?, duration_seconds = ?, - enabled = ?, summary = ?, reviewed = 1, updated_at = ? - WHERE id = ? AND task_id = ? AND is_deleted = 0 - """, - ( - data["title"], - data["start_time"], - data["end_time"], - data["duration_seconds"], - data["enabled"], - data["summary"], - now, - clip_id, - task_id, - ), - ) - if cursor.rowcount == 0: - raise ValueError(f"候选片段不存在：{clip_id}") - connection.execute("UPDATE tasks SET updated_at = ? WHERE id = ?", (now, task_id)) - connection.commit() + if cursor.rowcount == 0: + raise ValueError(f"候选片段不存在：{clip_id}") + if record_review_toggle_feedback_with_connection( + connection, + task_id=task_id, + clip={ + "id": clip_id, + "source_analysis_run_id": current["source_analysis_run_id"], + **data, + }, + selection_profile=task.get("selection_profile") or "general", + enabled=bool(data["enabled"]), + reason_code=feedback_reason_code, + now=now, + ): + feedback_count += 1 + connection.execute("UPDATE tasks SET updated_at = ? WHERE id = ?", (now, task_id)) + connection.commit() + except Exception: + connection.rollback() + raise _append_task_log(task_id, f"已批量保存 {len(validated)} 条候选片段审核修改") return { "message": f"已保存 {len(validated)} 条候选片段，任务状态仍保持 AI 结果待检查。", "changed_count": changed_count, + "feedback_count": feedback_count, "task": get_task(task_id, include_video_probe=False), "clips": list_clip_candidates(task_id), } diff --git a/app/static/css/content-review.css b/app/static/css/content-review.css new file mode 100644 index 0000000..9e974f9 --- /dev/null +++ b/app/static/css/content-review.css @@ -0,0 +1,282 @@ +.content-review-heading { + align-items: end; +} + +.content-review-account-picker { + min-width: min(320px, 100%); +} + +.content-review-account-picker span { + display: block; + margin-bottom: 8px; + color: var(--muted-text); + font-size: 12px; + font-weight: 700; +} + +.content-review-sync-grid, +.content-review-bottom-grid { + display: grid; + grid-template-columns: repeat(2, minmax(0, 1fr)); + gap: 18px; + margin-bottom: 18px; +} + +.content-review-sync-card { + display: grid; + align-content: start; + gap: 16px; +} + +.content-review-card-title, +.content-review-section-heading { + display: flex; + align-items: start; + justify-content: space-between; + gap: 16px; +} + +.content-review-card-title h2, +.content-review-section-heading h2 { + margin: 3px 0 0; +} + +.content-review-import-form { + display: flex; + flex-wrap: wrap; + gap: 10px; +} + +.content-review-import-form input[type="file"] { + flex: 1 1 260px; + min-height: 44px; + padding: 9px 12px; + border: 1px solid var(--border-color); + border-radius: 12px; + background: rgba(255, 255, 255, 0.78); +} + +.content-review-preview { + padding: 14px; + border: 1px solid rgba(0, 122, 255, 0.18); + border-radius: 14px; + background: rgba(0, 122, 255, 0.055); +} + +.content-review-preview dl { + display: grid; + grid-template-columns: repeat(2, minmax(0, 1fr)); + gap: 10px; + margin: 0 0 14px; +} + +.content-review-preview dl div { + display: grid; + gap: 3px; +} + +.content-review-preview dt, +.content-review-preview dd { + margin: 0; +} + +.content-review-preview dt { + color: var(--muted-text); + font-size: 12px; +} + +.content-review-preview dd { + font-weight: 700; +} + +.content-review-summary-panel, +.content-review-works-panel { + margin-bottom: 18px; +} + +.content-review-metric-grid { + display: grid; + grid-template-columns: repeat(4, minmax(0, 1fr)); + gap: 12px; + margin-top: 18px; +} + +.content-review-metric { + display: grid; + gap: 5px; + min-height: 132px; + padding: 16px; + border: 1px solid var(--border-color); + border-radius: 16px; + background: linear-gradient(145deg, rgba(255, 255, 255, 0.95), rgba(242, 247, 255, 0.8)); +} + +.content-review-metric > span, +.content-review-metric em { + color: var(--muted-text); + font-size: 12px; + font-style: normal; +} + +.content-review-metric strong { + font-size: clamp(24px, 3vw, 34px); + letter-spacing: -0.04em; +} + +.content-review-metric small.is-up { + color: #11733f; +} + +.content-review-metric small.is-down { + color: #b04b3e; +} + +.content-review-history-wrap { + overflow-x: auto; + margin-top: 18px; +} + +.content-review-table { + width: 100%; + min-width: 820px; + border-collapse: collapse; +} + +.content-review-table th, +.content-review-table td { + padding: 13px 12px; + border-bottom: 1px solid var(--border-color); + text-align: left; + vertical-align: top; +} + +.content-review-table th { + color: var(--muted-text); + font-size: 12px; + white-space: nowrap; +} + +.content-review-table td { + font-size: 13px; +} + +.content-review-table td strong, +.content-review-table td span, +.content-review-table td small { + display: block; +} + +.content-review-table td small { + margin-top: 4px; + color: var(--muted-text); +} + +.content-review-works-table { + min-width: 1120px; +} + +.content-review-match-tools { + display: grid; + gap: 7px; + min-width: 190px; +} + +.content-review-match-tools input { + min-height: 36px; + padding: 7px 9px; + border: 1px solid var(--border-color); + border-radius: 9px; +} + +.content-review-prompt-list, +.content-review-import-list { + display: grid; + gap: 10px; + margin-top: 14px; +} + +.content-review-prompt-item, +.content-review-import-item { + padding: 14px; + border: 1px solid var(--border-color); + border-radius: 14px; + background: rgba(255, 255, 255, 0.66); +} + +.content-review-prompt-item header, +.content-review-import-item header { + display: flex; + justify-content: space-between; + gap: 12px; +} + +.content-review-prompt-item dl { + display: grid; + grid-template-columns: repeat(3, minmax(0, 1fr)); + gap: 8px; + margin: 12px 0 0; +} + +.content-review-prompt-item dt, +.content-review-prompt-item dd { + margin: 0; +} + +.content-review-prompt-item dt { + color: var(--muted-text); + font-size: 11px; +} + +.content-review-prompt-item dd { + margin-top: 3px; + font-weight: 700; +} + +.content-review-note, +.content-review-causality { + color: var(--muted-text); +} + +.content-review-causality { + margin: 14px 0 0; + font-size: 12px; +} + +.content-review-empty { + margin-top: 18px; +} + +@media (max-width: 980px) { + .content-review-sync-grid, + .content-review-bottom-grid { + grid-template-columns: 1fr; + } + + .content-review-metric-grid { + grid-template-columns: repeat(2, minmax(0, 1fr)); + } +} + +@media (max-width: 520px) { + .content-review-heading, + .content-review-card-title, + .content-review-section-heading { + align-items: stretch; + flex-direction: column; + } + + .content-review-account-picker { + min-width: 0; + width: 100%; + } + + .content-review-metric-grid, + .content-review-preview dl, + .content-review-prompt-item dl { + grid-template-columns: 1fr; + } + + .content-review-metric { + min-height: 118px; + } +} diff --git a/app/static/css/styles.css b/app/static/css/styles.css index 8ccd4ef..8511c8b 100644 --- a/app/static/css/styles.css +++ b/app/static/css/styles.css @@ -1763,12 +1763,6 @@ fieldset input.visually-hidden-file { background: rgba(0, 122, 255, 0.1); } -.clip-feedback .feedback-positive.is-active { - border-color: rgba(52, 199, 89, 0.55); - color: #11643f; - background: rgba(52, 199, 89, 0.12); -} - .compact-button { min-height: 34px; padding: 0 12px; diff --git a/app/static/js/app.js b/app/static/js/app.js index 8c3cfa4..6282155 100644 --- a/app/static/js/app.js +++ b/app/static/js/app.js @@ -920,14 +920,18 @@ function updateClipReviewActionState() { function collectClipReviewPayload() { const cards = getClipReviewCards(); - return cards.map((card) => ({ - id: card.dataset.clipId, - title: card.querySelector("[name='title']").value.trim(), - start_time: card.querySelector("[name='start_time']").value.trim(), - end_time: card.querySelector("[name='end_time']").value.trim(), - enabled: card.querySelector("[name='enabled']").checked, - summary: card.querySelector("[name='summary']").value.trim(), - })); + return cards.map((card) => { + const enabled = card.querySelector("[name='enabled']").checked; + return { + id: card.dataset.clipId, + title: card.querySelector("[name='title']").value.trim(), + start_time: card.querySelector("[name='start_time']").value.trim(), + end_time: card.querySelector("[name='end_time']").value.trim(), + enabled, + summary: card.querySelector("[name='summary']").value.trim(), + feedback_reason_code: enabled ? null : (card.dataset.feedbackReason || null), + }; + }); } async function persistClipReviewChanges() { @@ -981,38 +985,11 @@ async function deleteClipCard(card, button) { } } -async function saveClipFeedback(card, button) { - if (!clipReviewForm || !card || !button) return; - const taskId = clipReviewForm.dataset.taskId; - const clipId = card.dataset.clipId; - const decision = button.dataset.feedbackDecision; - const reasonCode = button.dataset.feedbackReason; - const feedbackButtons = Array.from(card.querySelectorAll("[data-feedback-decision]")); - feedbackButtons.forEach((item) => { item.disabled = true; }); - showClipReviewMessage("正在记录你的审片判断...", "info"); - - try { - const response = await fetch(`/api/tasks/${taskId}/clips/${clipId}/feedback`, { - method: "POST", - headers: { "Content-Type": "application/json" }, - body: JSON.stringify({ decision, reason_code: reasonCode }), - }); - const data = await response.json(); - if (!response.ok) { - throw new Error(data.detail || "保存反馈失败"); - } - feedbackButtons.forEach((item) => { - item.classList.toggle("is-active", item === button); - item.setAttribute("aria-pressed", item === button ? "true" : "false"); - }); - const enabledInput = card.querySelector("[name='enabled']"); - if (enabledInput) enabledInput.checked = Boolean(data.enabled); - showClipReviewMessage(data.message || "反馈已保存。", "success"); - } catch (error) { - showClipReviewMessage(`保存反馈失败：${error.message}`, "error"); - } finally { - feedbackButtons.forEach((item) => { item.disabled = false; }); - } +function syncRejectReasonVisibility(card) { + if (!card) return; + const enabledInput = card.querySelector("[name='enabled']"); + const rejectFeedback = card.querySelector("[data-reject-feedback]"); + if (rejectFeedback && enabledInput) rejectFeedback.hidden = enabledInput.checked; } function timeTextToSeconds(value) { @@ -1302,7 +1279,10 @@ document.querySelectorAll("[data-clip-card] input[name='start_time'], [data-clip if (clipSelectAll) { clipSelectAll.addEventListener("change", () => { const shouldEnable = clipSelectAll.checked; - getClipEnableCheckboxes().forEach((checkbox) => { checkbox.checked = shouldEnable; }); + getClipEnableCheckboxes().forEach((checkbox) => { + checkbox.checked = shouldEnable; + syncRejectReasonVisibility(checkbox.closest("[data-clip-card]")); + }); updateClipSelectAllUi(); showClipReviewMessage( shouldEnable @@ -1314,7 +1294,10 @@ if (clipSelectAll) { } clipReviewForm?.addEventListener("change", (event) => { - if (event.target.matches("[data-clip-card] input[name='enabled']")) updateClipSelectAllUi(); + if (event.target.matches("[data-clip-card] input[name='enabled']")) { + syncRejectReasonVisibility(event.target.closest("[data-clip-card]")); + updateClipSelectAllUi(); + } }); updateClipSelectAllUi(); @@ -1459,9 +1442,18 @@ document.querySelectorAll("[data-delete-trigger]").forEach((button) => { }); }); -document.querySelectorAll("[data-feedback-decision]").forEach((button) => { +document.querySelectorAll("[data-reject-reason]").forEach((button) => { button.addEventListener("click", () => { - saveClipFeedback(button.closest("[data-clip-card]"), button); + const card = button.closest("[data-clip-card]"); + if (!card) return; + const selected = card.dataset.feedbackReason === button.dataset.rejectReason; + card.dataset.feedbackReason = selected ? "" : button.dataset.rejectReason; + card.querySelectorAll("[data-reject-reason]").forEach((item) => { + const active = !selected && item === button; + item.classList.toggle("is-active", active); + item.setAttribute("aria-pressed", active ? "true" : "false"); + }); + showClipReviewMessage("淘汰原因已暂存；点击“保存修改”后统一写入。", "info"); }); }); diff --git a/app/static/js/content-review.js b/app/static/js/content-review.js new file mode 100644 index 0000000..e8e41fd --- /dev/null +++ b/app/static/js/content-review.js @@ -0,0 +1,428 @@ +const contentReviewAccount = document.querySelector("#content-review-account"); +const contentReviewMessage = document.querySelector("#content-review-message"); +const contentReviewImportForm = document.querySelector("#content-review-import-form"); +const contentReviewFile = document.querySelector("#content-review-file"); +const contentReviewPreview = document.querySelector("#content-review-preview"); +const contentReviewCommit = document.querySelector("#content-review-commit"); +const contentReviewSync = document.querySelector("#content-review-sync"); + +let previewBatchId = ""; + +function currentAccountId() { + return contentReviewAccount?.value || ""; +} + +function showContentReviewMessage(message, tone = "info") { + if (!contentReviewMessage) return; + contentReviewMessage.hidden = !message; + contentReviewMessage.textContent = message || ""; + contentReviewMessage.className = `page-alert ${tone}`; +} + +function errorMessage(data, fallback) { + if (!data) return fallback; + if (typeof data.detail === "string") return data.detail; + if (data.detail && typeof data.detail.message === "string") { + const code = data.detail.error_code ? `（${data.detail.error_code}）` : ""; + return `${data.detail.message}${code}`; + } + return data.message || fallback; +} + +async function contentReviewApi(path, options = {}) { + const response = await fetch(path, options); + let data = {}; + try { + data = await response.json(); + } catch (_error) { + data = {}; + } + if (!response.ok) throw new Error(errorMessage(data, "请求失败")); + return data; +} + +function textNode(tag, text, className = "") { + const node = document.createElement(tag); + node.textContent = text == null || text === "" ? "—" : String(text); + if (className) node.className = className; + return node; +} + +function formatNumber(value) { + if (value == null) return "—"; + return Number(value).toLocaleString("zh-CN", { maximumFractionDigits: 1 }); +} + +function formatPercent(value) { + if (value == null) return "—"; + return `${(Number(value) * 100).toFixed(1)}%`; +} + +function formatSeconds(value) { + if (value == null) return "—"; + return `${Number(value).toFixed(1)} 秒`; +} + +function formatMetric(key, value) { + if (["five_second_completion_rate", "two_second_bounce_rate", "cover_click_rate"].includes(key)) { + return formatPercent(value); + } + if (key === "average_watch_seconds") return formatSeconds(value); + return formatNumber(value); +} + +function formatDateTime(value) { + if (!value) return "—"; + const parsed = new Date(value); + if (Number.isNaN(parsed.getTime())) return String(value); + return parsed.toLocaleString("zh-CN", { hour12: false }); +} + +function renderSummary(summary) { + const latestDate = document.querySelector("#content-review-latest-date"); + const syncAge = document.querySelector("#content-review-sync-age"); + if (latestDate) latestDate.textContent = summary.latest_metric_date || "暂无数据"; + if (syncAge) { + syncAge.textContent = summary.days_since_sync == null + ? "尚未同步" + : summary.days_since_sync === 0 ? "今天已同步" : `${summary.days_since_sync} 天前同步`; + } + document.querySelectorAll("[data-metric]").forEach((card) => { + const key = card.dataset.metric; + const current = summary.current_period?.[key]; + const previous = summary.previous_period?.[key]; + const delta = summary.comparisons?.[key]; + card.querySelector("[data-current]").textContent = formatMetric(key, current); + card.querySelector("[data-previous]").textContent = formatMetric(key, previous); + const deltaNode = card.querySelector("[data-delta]"); + deltaNode.classList.remove("is-up", "is-down"); + if (delta == null) { + deltaNode.textContent = "暂无可比基线"; + } else { + deltaNode.textContent = `${delta >= 0 ? "↑" : "↓"} ${Math.abs(Number(delta) * 100).toFixed(1)}%`; + deltaNode.classList.add(delta >= 0 ? "is-up" : "is-down"); + } + }); + + const body = document.querySelector("#content-review-history"); + if (!body) return; + body.replaceChildren(); + const history = Array.from(summary.history || []).reverse(); + if (!history.length) { + const row = document.createElement("tr"); + const cell = textNode("td", "暂无账号级数据。请先预览并确认导入抖音数据表。"); + cell.colSpan = 8; + row.append(cell); + body.append(row); + return; + } + history.forEach((item) => { + const row = document.createElement("tr"); + const interaction = Number(item.like_count || 0) + Number(item.share_count || 0) + Number(item.comment_count || 0); + [ + item.metric_date, + formatNumber(item.post_count), + formatNumber(item.play_count), + formatNumber(interaction), + formatPercent(item.five_second_completion_rate), + formatPercent(item.two_second_bounce_rate), + formatPercent(item.cover_click_rate), + formatSeconds(item.average_watch_seconds), + ].forEach((value) => row.append(textNode("td", value))); + body.append(row); + }); +} + +function appendCell(row, primary, secondary = "") { + const cell = document.createElement("td"); + cell.append(textNode("strong", primary)); + if (secondary) cell.append(textNode("small", secondary)); + row.append(cell); + return cell; +} + +function matchLabel(status) { + return { + matched_exact: "作品 ID 精确匹配", + matched_unique: "标题时间唯一匹配", + confirmed_manual: "人工确认", + ambiguous: "存在多个候选", + unmatched: "未匹配", + }[status] || status || "未匹配"; +} + +async function updateItemMatch(snapshotId, publishJobId) { + await contentReviewApi(`/api/content-review/item-matches/${encodeURIComponent(snapshotId)}`, { + method: "PUT", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ publish_job_id: publishJobId }), + }); + showContentReviewMessage("人工关联已保存。", "success"); + await loadContentReviewData(); +} + +async function removeItemMatch(snapshotId) { + await contentReviewApi(`/api/content-review/item-matches/${encodeURIComponent(snapshotId)}`, { + method: "DELETE", + }); + showContentReviewMessage("错误关联已解除。", "success"); + await loadContentReviewData(); +} + +function renderWorks(works) { + const body = document.querySelector("#content-review-works"); + if (!body) return; + body.replaceChildren(); + if (!works.length) { + const row = document.createElement("tr"); + const cell = textNode("td", "暂无作品级指标。请在登录状态正常时手动点击同步最近 50 条作品。"); + cell.colSpan = 6; + row.append(cell); + body.append(row); + return; + } + works.forEach((work) => { + const row = document.createElement("tr"); + appendCell(row, work.title || work.aweme_id, `${work.aweme_id} · ${formatDateTime(work.published_at)}`); + appendCell( + row, + `${formatNumber(work.play_count)} 播放`, + `${formatNumber(work.like_count)} 赞 · ${formatNumber(work.comment_count)} 评 · ${formatNumber(work.share_count)} 转`, + ); + appendCell( + row, + work.publish_title || "未关联发布记录", + work.publish_job_id ? `${work.publish_job_id} · ${work.publish_status || ""}` : "等待匹配", + ); + appendCell( + row, + work.candidate_title || "候选来源不完整", + work.review_decision + ? `${work.review_decision === "keep" ? "保留" : "淘汰"} · ${work.review_reason_label || "未填写原因"}` + : "暂无审核反馈", + ); + appendCell( + row, + work.analysis_run_number ? `第 ${work.analysis_run_number} 次 · ${work.provider_label || "AI"}` : "AI 来源不完整", + work.prompt_version_number ? `${work.prompt_name || "Prompt"} v${work.prompt_version_number}` : "Prompt 来源不完整", + ); + const matchCell = document.createElement("td"); + const tools = document.createElement("div"); + tools.className = "content-review-match-tools"; + tools.append(textNode("strong", matchLabel(work.match_status))); + if (["ambiguous", "unmatched"].includes(work.match_status)) { + const input = document.createElement("input"); + input.type = "text"; + input.placeholder = "输入发布记录 ID"; + input.setAttribute("aria-label", "发布记录 ID"); + const button = textNode("button", "人工确认", "secondary-button compact-button"); + button.type = "button"; + button.addEventListener("click", async () => { + const jobId = input.value.trim(); + if (!jobId) { + showContentReviewMessage("请先填写要关联的发布记录 ID。", "error"); + return; + } + button.disabled = true; + try { + await updateItemMatch(work.id, jobId); + } catch (error) { + showContentReviewMessage(`关联失败：${error.message}`, "error"); + } finally { + button.disabled = false; + } + }); + tools.append(input, button); + } else if (work.publish_job_id) { + const button = textNode("button", "解除关联", "link-button danger"); + button.type = "button"; + button.addEventListener("click", async () => { + button.disabled = true; + try { + await removeItemMatch(work.id); + } catch (error) { + showContentReviewMessage(`解除失败：${error.message}`, "error"); + } finally { + button.disabled = false; + } + }); + tools.append(button); + } + matchCell.append(tools); + row.append(matchCell); + body.append(row); + }); +} + +function promptMetric(label, value) { + const wrapper = document.createElement("div"); + wrapper.append(textNode("dt", label), textNode("dd", value)); + return wrapper; +} + +function renderPromptComparison(data) { + const cycles = document.querySelector("#content-review-cycles"); + const message = document.querySelector("#content-review-prompt-message"); + const list = document.querySelector("#content-review-prompts"); + if (cycles) cycles.textContent = `${data.completed_cycles || 0} / ${data.minimum_cycles || 3} 周期`; + if (message) message.textContent = data.message || "数据不足。"; + if (!list) return; + list.replaceChildren(); + if (!(data.versions || []).length) { + list.append(textNode("p", "还没有带准确作品归因的 Prompt 版本。", "empty-note")); + return; + } + Array.from(data.versions).reverse().forEach((version) => { + const item = document.createElement("article"); + item.className = "content-review-prompt-item"; + const header = document.createElement("header"); + header.append( + textNode("strong", `${version.prompt_name} · v${version.version_number}`), + textNode("span", version.evaluable ? "可评估" : "数据不足", "status-pill"), + ); + const metrics = document.createElement("dl"); + metrics.append( + promptMetric("准确发布", `${version.accurate_published_count} 条`), + promptMetric("保留率", formatPercent(version.keep_rate)), + promptMetric("发布率", formatPercent(version.publish_rate)), + promptMetric("播放中位数", formatNumber(version.median_play_count)), + promptMetric("5 秒完播", formatPercent(version.five_second_completion_rate)), + promptMetric("2 秒跳出", formatPercent(version.two_second_bounce_rate)), + promptMetric("平均观看 / 片长", formatPercent(version.average_watch_ratio)), + promptMetric("互动率", formatPercent(version.interaction_rate)), + ); + item.append(header, metrics); + list.append(item); + }); +} + +function renderImports(imports) { + const list = document.querySelector("#content-review-imports"); + if (!list) return; + list.replaceChildren(); + if (!imports.length) { + list.append(textNode("p", "暂无同步记录。", "empty-note")); + return; + } + imports.forEach((batch) => { + const item = document.createElement("article"); + item.className = "content-review-import-item"; + const header = document.createElement("header"); + header.append( + textNode("strong", batch.source_kind === "douyin_item_sync" ? "作品级同步" : batch.source_filename), + textNode("span", batch.status, "status-pill"), + ); + item.append( + header, + textNode( + "small", + `${batch.row_count || 0} 行 · ${batch.period_start || "—"} 至 ${batch.period_end || "—"} · ${formatDateTime(batch.committed_at || batch.created_at)}`, + ), + ); + list.append(item); + }); +} + +async function loadContentReviewData() { + const accountId = currentAccountId(); + if (!accountId) return; + const query = `account_id=${encodeURIComponent(accountId)}`; + try { + const [summary, works, prompts, imports] = await Promise.all([ + contentReviewApi(`/api/content-review/summary?${query}&days=28`), + contentReviewApi(`/api/content-review/works?${query}&limit=100`), + contentReviewApi(`/api/content-review/prompt-comparison?${query}`), + contentReviewApi(`/api/content-review/imports?${query}&limit=20`), + ]); + renderSummary(summary); + renderWorks(works.works || []); + renderPromptComparison(prompts); + renderImports(imports.imports || []); + } catch (error) { + showContentReviewMessage(`读取复盘数据失败：${error.message}`, "error"); + } +} + +contentReviewImportForm?.addEventListener("submit", async (event) => { + event.preventDefault(); + const file = contentReviewFile?.files?.[0]; + if (!file) { + showContentReviewMessage("请先选择 .xlsx 或 .csv 文件。", "error"); + return; + } + const button = contentReviewImportForm.querySelector("button[type='submit']"); + button.disabled = true; + showContentReviewMessage("正在只读解析并校验数据表…", "info"); + const formData = new FormData(); + formData.append("file", file); + formData.append("account_id", currentAccountId()); + try { + const data = await contentReviewApi("/api/content-review/imports/preview", { + method: "POST", + body: formData, + }); + previewBatchId = data.batch_id || ""; + document.querySelector("#content-review-file-status").textContent = data.already_imported ? "已导入" : "预览通过"; + if (contentReviewPreview) contentReviewPreview.hidden = data.already_imported; + if (!data.already_imported) { + contentReviewPreview.querySelector("[data-preview-filename]").textContent = data.filename || file.name; + contentReviewPreview.querySelector("[data-preview-period]").textContent = `${data.period_start} 至 ${data.period_end}`; + contentReviewPreview.querySelector("[data-preview-rows]").textContent = `${data.row_count} 行`; + } + showContentReviewMessage(data.message, data.already_imported ? "info" : "success"); + } catch (error) { + previewBatchId = ""; + if (contentReviewPreview) contentReviewPreview.hidden = true; + showContentReviewMessage(`预览失败：${error.message}`, "error"); + } finally { + button.disabled = false; + } +}); + +contentReviewCommit?.addEventListener("click", async () => { + if (!previewBatchId) return; + contentReviewCommit.disabled = true; + try { + const data = await contentReviewApi(`/api/content-review/imports/${encodeURIComponent(previewBatchId)}/commit`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ confirm: true }), + }); + showContentReviewMessage(data.message, "success"); + contentReviewPreview.hidden = true; + previewBatchId = ""; + await loadContentReviewData(); + } catch (error) { + showContentReviewMessage(`导入失败：${error.message}`, "error"); + } finally { + contentReviewCommit.disabled = false; + } +}); + +contentReviewSync?.addEventListener("click", async () => { + contentReviewSync.disabled = true; + contentReviewSync.textContent = "同步中…"; + showContentReviewMessage("正在读取最近 50 条作品；不会触发投稿或自动重试。", "info"); + try { + const data = await contentReviewApi("/api/content-review/douyin/sync-preview", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ account_id: currentAccountId(), limit: 50 }), + }); + showContentReviewMessage(data.message, "success"); + await loadContentReviewData(); + } catch (error) { + showContentReviewMessage(`同步已安全停止：${error.message}`, "error"); + } finally { + contentReviewSync.disabled = false; + contentReviewSync.textContent = "一键同步最近 50 条作品"; + } +}); + +contentReviewAccount?.addEventListener("change", () => { + previewBatchId = ""; + if (contentReviewPreview) contentReviewPreview.hidden = true; + loadContentReviewData(); +}); + +loadContentReviewData(); diff --git a/app/templates/base.html b/app/templates/base.html index ed88c20..d303f2b 100644 --- a/app/templates/base.html +++ b/app/templates/base.html @@ -30,6 +30,7 @@ 审片段审核 字字幕推送 发发送中心 + 复内容复盘 设系统状态
diff --git a/app/templates/clip_review.html b/app/templates/clip_review.html index 14171cc..0c40206 100644 --- a/app/templates/clip_review.html +++ b/app/templates/clip_review.html @@ -45,6 +45,7 @@

片段审核 · {{ task.title }}

data-start-seconds="{{ clip.start_seconds }}" data-end-seconds="{{ clip.end_seconds }}" data-title="{{ clip.title }}" + data-feedback-reason="{{ clip.feedback_reason_code or '' }}" >