@@ -60,6 +60,16 @@
+
diff --git a/docs/DATABASE_SCHEMA.md b/docs/DATABASE_SCHEMA.md
index 4e52a97..2e5f8d6 100644
--- a/docs/DATABASE_SCHEMA.md
+++ b/docs/DATABASE_SCHEMA.md
@@ -1,5 +1,23 @@
# 数据库结构说明
+## 2026-08-29：内容诊断与实验闭环
+
+迁移账本版本为 `20260829_02_content_feedback_loop`。迁移只新增表和索引，不删除或改写现有作品快照、发布记录、排期或 Prompt 版本；应用前继续通过 SQLite Online Backup 生成 `workflow-before-content-feedback-loop-*` 备份。
+
+### `content_improvement_experiments`
+
+- 以 `account_id + recommendation_id` 唯一标识一次基于固定证据批次的实验，保存诊断类型、假设、唯一动作、主指标方向、护栏指标、基线批次及冻结的中位数/四分位数 JSON。
+- 固定门槛为实验 20 条、对照 20 条、3 个不同官方导出周；状态为 `active / completed / cancelled`，人工结论为 `keep / revert / inconclusive`。
+- `baseline_batch_id` 外键指向官方导入批次；创建后不会随着后续数据重算或改写冻结基线。
+
+### `content_improvement_experiment_items`
+
+- 关联实验与 `publish_jobs`，`publish_job_id` 全局唯一，保证一个作品同一时刻只有一个实验归属。
+- 只允许在 `DRAFT / WAITING / SCHEDULED` 且没有 `claimed_at / started_at / attempt_count` 执行证据时关联、切换或解除；执行开始后永久冻结该作品的实验归属。
+- 发送中心使用原子设置接口完成新增、切换和解除，不会留下“旧关联已删、新关联未写”的中间状态。
+
+诊断本身为只读查询：只读取当前账号最新的 `douyin_item_export` 快照，未匹配、多个候选或核心指标缺失的作品标记为证据不足。官方导出周按 `captured_at` 的北京时间 ISO 自然周去重；账号趋势表不参与周数。Prompt 统计同样按账号和官方作品来源隔离。
+
## 2026-08-29：抖音官方作品报表完整指标
- 新迁移账本版本：`20260829_01_douyin_official_item_export`。它只追加字段，不修改已应用的 `20260828_01_content_review_v1` 名称、定义或 checksum。
diff --git a/docs/UI_REFERENCE.md b/docs/UI_REFERENCE.md
index e6b648c..d07b1d0 100644
--- a/docs/UI_REFERENCE.md
+++ b/docs/UI_REFERENCE.md
@@ -1,5 +1,16 @@
# UI 参考说明
+## 2026-08-29 更新：内容复盘可执行闭环
+
+- 自动导出卡固定展示精确北京时间和本批作品/匹配统计；文件上传使用蓝色自定义按钮、独立文件名和禁用态预览按钮，不展示浏览器原生文件控件。
+- 账号趋势指标卡常显，每日明细默认折叠；2 秒跳出率是反向指标，下降为绿色、上升为红色。作品归因默认折叠，标题区常显“已匹配 / 待确认 / 未匹配”。
+- 所有折叠区用原生 `details/summary`，状态保存到当前浏览器 `localStorage`；无 JavaScript 时仍可手动展开，不依赖前端框架。
+- “本周改进建议”卡显示来源作品、证据指标、同类对照的 P25/中位数/P75、唯一建议动作、主指标与采纳按钮；证据不足和缺失封面指标只说明原因，不生成伪结论。
+- 实验卡显示已标记、已有官方指标、冻结对照和官方导出周；10 条前不展示结果，10 条起标为“早期趋势”，达到 20+20+3 周后才展示人工决定按钮。
+- “Prompt 与实验依据”降级为默认折叠的高级区，显示真实官方导出周和每个版本缺少的准确作品数。
+- 发送中心内容卡增加可选实验下拉，只展示当前发布账号可用的实验；保存时原子更新实验归属，提示执行开始后冻结。
+- 桌面 1440px 与窄屏 390px 均保持页面无整体横向溢出；宽作品表只在自己的滚动容器内滚动。
+
## 2026-08-29 更新：官方作品报表全量同步
- 内容复盘左侧人工导入卡支持账号趋势表和官方 `作品列表导出.xlsx`，预览区明确显示识别类型、日期范围、行数和归因口径，再由用户确认导入。
diff --git a/tests/test_content_review.py b/tests/test_content_review.py
index 869c14b..31398bc 100644
--- a/tests/test_content_review.py
+++ b/tests/test_content_review.py
@@ -1,7 +1,10 @@
from __future__ import annotations
import csv
+from datetime import datetime
+import hashlib
import io
+import json
from uuid import uuid4
from fastapi.testclient import TestClient
@@ -34,6 +37,19 @@ def _cleanup() -> None:
).fetchall()
account_ids = [row["id"] for row in account_rows]
for account_id in account_ids:
+ experiment_rows = connection.execute(
+ "SELECT id FROM content_improvement_experiments WHERE account_id = ?",
+ (account_id,),
+ ).fetchall()
+ for experiment in experiment_rows:
+ connection.execute(
+ "DELETE FROM content_improvement_experiment_items WHERE experiment_id = ?",
+ (experiment["id"],),
+ )
+ connection.execute(
+ "DELETE FROM content_improvement_experiments WHERE account_id = ?",
+ (account_id,),
+ )
batch_rows = connection.execute(
"SELECT id FROM content_metric_import_batches WHERE account_id = ?",
(account_id,),
@@ -58,6 +74,8 @@ def _cleanup() -> None:
connection.execute("DELETE FROM clip_candidates WHERE task_id LIKE ?", (f"{PREFIX}%",))
connection.execute("DELETE FROM ai_analysis_runs WHERE task_id LIKE ?", (f"{PREFIX}%",))
connection.execute("DELETE FROM tasks WHERE id LIKE ?", (f"{PREFIX}%",))
+ connection.execute("DELETE FROM ai_prompt_versions WHERE preset_id LIKE ?", (f"{PREFIX}%",))
+ connection.execute("DELETE FROM ai_prompt_presets WHERE id LIKE ?", (f"{PREFIX}%",))
connection.execute("DELETE FROM publish_accounts WHERE id LIKE ?", (f"{PREFIX}%",))
connection.commit()
@@ -389,6 +407,7 @@ def _insert_publish_job(
duration_seconds: int = 60,
description: str = "",
caption: str = "",
+ status: str = "PUBLISHED",
) -> str:
suffix = uuid4().hex[:8]
task_id = f"{PREFIX}task-{suffix}"
@@ -414,7 +433,7 @@ def _insert_publish_job(
INSERT INTO publish_jobs (
id, task_id, output_clip_id, account_id, platform, title, description,
caption, status, platform_item_id, published_at, created_at, updated_at
- ) VALUES (?, ?, ?, ?, 'douyin', ?, ?, ?, 'PUBLISHED', ?, ?, ?, ?)
+ ) VALUES (?, ?, ?, ?, 'douyin', ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
job_id,
@@ -424,6 +443,7 @@ def _insert_publish_job(
title,
description,
caption,
+ status,
platform_item_id,
published_at,
now,
@@ -434,6 +454,129 @@ def _insert_publish_job(
return job_id
+def _diagnosis_item(index: int, *, title: str, published_at: str, **overrides) -> dict:
+ item = {
+ "aweme_id": f"export:test-{index}-{uuid4().hex[:6]}",
+ "title": title,
+ "published_at": published_at,
+ "duration_seconds": 60,
+ "content_genre": "视频",
+ "audit_status": "审核通过",
+ "play_count": 1000 + index * 100,
+ "completion_rate": 0.45 + index * 0.005,
+ "five_second_completion_rate": 0.60 + index * 0.005,
+ "two_second_bounce_rate": 0.20 + index * 0.003,
+ "average_watch_seconds": 30 + index,
+ "cover_click_rate": None,
+ "like_count": 20,
+ "comment_count": 5,
+ "share_count": 3,
+ }
+ item.update(overrides)
+ return item
+
+
+def _seed_diagnosis_baseline(account_id: str, *, captured_at: str = "2026-08-29T12:00:00+08:00") -> dict:
+ items = []
+ for index in range(9):
+ published_at = f"2026-08-2{index}T10:00:00+08:00"
+ title = f"诊断基线作品 {index}"
+ _insert_publish_job(
+ account_id,
+ title=title,
+ published_at=published_at,
+ duration_seconds=60,
+ )
+ overrides = {}
+ if index == 8:
+ overrides = {
+ "five_second_completion_rate": 0.30,
+ "two_second_bounce_rate": 0.55,
+ "completion_rate": 0.12,
+ "average_watch_seconds": 12,
+ }
+ items.append(
+ _diagnosis_item(
+ index,
+ title=title,
+ published_at=published_at,
+ **overrides,
+ )
+ )
+ items.append(
+ _diagnosis_item(
+ 99,
+ title="没有候选的作品",
+ published_at="2026-08-28T18:00:00+08:00",
+ )
+ )
+ return content_review_service.commit_douyin_item_export(
+ account_id=account_id,
+ items=items,
+ captured_at=captured_at,
+ source_filename="作品列表导出.xlsx",
+ )
+
+
+def _attach_prompt_chain(job_id: str, label: str) -> str:
+ suffix = uuid4().hex[:8]
+ preset_id = f"{PREFIX}preset-{label}-{suffix}"
+ version_id = f"{PREFIX}version-{label}-{suffix}"
+ run_id = f"{PREFIX}run-{label}-{suffix}"
+ candidate_id = f"{PREFIX}candidate-{label}-{suffix}"
+ prompt_text = f"{label} prompt"
+ prompt_hash = hashlib.sha256(prompt_text.encode("utf-8")).hexdigest()
+ now = "2026-08-28T10:00:00+08:00"
+ with get_connection() as connection:
+ job = connection.execute(
+ "SELECT task_id, output_clip_id FROM publish_jobs WHERE id = ?",
+ (job_id,),
+ ).fetchone()
+ slot = int(connection.execute("SELECT COALESCE(MAX(slot), 1000) + 1 FROM ai_prompt_presets").fetchone()[0])
+ connection.execute(
+ """
+ INSERT INTO ai_prompt_presets (
+ id, slot, name, prompt_text, is_default, created_at, updated_at
+ ) VALUES (?, ?, ?, ?, 0, ?, ?)
+ """,
+ (preset_id, slot, label, prompt_text, now, now),
+ )
+ connection.execute(
+ """
+ INSERT INTO ai_prompt_versions (
+ id, preset_id, version_number, preset_name_snapshot,
+ prompt_text, prompt_sha256, created_at
+ ) VALUES (?, ?, 1, ?, ?, ?, ?)
+ """,
+ (version_id, preset_id, label, prompt_text, prompt_hash, now),
+ )
+ connection.execute(
+ """
+ INSERT INTO ai_analysis_runs (
+ id, task_id, run_number, provider, provider_label, model,
+ prompt_version_id, prompt_text_sha256, requested_clip_count,
+ clip_count, analysis_payload_json, created_at
+ ) VALUES (?, ?, 1, 'test', 'Test', 'test-model', ?, ?, 1, 1, ?, ?)
+ """,
+ (run_id, job["task_id"], version_id, prompt_hash, json.dumps({"clips": []}), now),
+ )
+ connection.execute(
+ """
+ INSERT INTO clip_candidates (
+ id, task_id, clip_key, title, start_time, end_time,
+ duration_seconds, source_analysis_run_id, created_at, updated_at
+ ) VALUES (?, ?, 'clip_001', ?, '00:00:00', '00:01:00', 60, ?, ?, ?)
+ """,
+ (candidate_id, job["task_id"], label, run_id, now, now),
+ )
+ connection.execute(
+ "UPDATE output_clip SET clip_candidate_id = ? WHERE id = ?",
+ (candidate_id, job["output_clip_id"]),
+ )
+ connection.commit()
+ return version_id
+
+
def test_item_matching_exact_unique_ambiguous_and_manual_confirmation():
account_id = _insert_account()
exact_job = _insert_publish_job(
@@ -501,6 +644,8 @@ def test_item_matching_exact_unique_ambiguous_and_manual_confirmation():
],
)
assert sync["ambiguous_count"] == 1
+ works = content_review_service.list_content_review_works(account_id)
+ assert works[0]["match_label"] == "待人工确认·多个候选"
with get_connection() as connection:
snapshot_id = connection.execute(
"SELECT id FROM douyin_item_metric_snapshots WHERE title = '同名作品'"
@@ -508,6 +653,7 @@ def test_item_matching_exact_unique_ambiguous_and_manual_confirmation():
content_review_service.set_item_match(snapshot_id, ambiguous_a)
works = content_review_service.list_content_review_works(account_id)
assert works[0]["match_status"] == "confirmed_manual"
+ assert works[0]["match_label"] == "已匹配·人工确认"
assert works[0]["publish_job_id"] == ambiguous_a
content_review_service.delete_item_match(snapshot_id)
works = content_review_service.list_content_review_works(account_id)
@@ -704,3 +850,332 @@ def rate_limited(_self, *, account_id):
)
assert response.status_code == 429
assert response.json()["detail"]["error_code"] == "RATE_LIMITED"
+
+
+def test_summary_uses_latest_official_export_and_deduplicates_same_week(monkeypatch):
+ account_id = _insert_account()
+ job_id = _insert_publish_job(
+ account_id,
+ title="周口径测试作品",
+ published_at="2026-08-28T10:00:00+08:00",
+ )
+ clock = {"value": datetime.fromisoformat("2026-08-29T00:48:37+08:00")}
+ monkeypatch.setattr(content_review_service, "_now", lambda: clock["value"])
+ first = content_review_service.commit_douyin_item_export(
+ account_id=account_id,
+ items=[
+ _diagnosis_item(
+ 1,
+ title="周口径测试作品",
+ published_at="2026-08-28T10:00:00+08:00",
+ )
+ ],
+ captured_at="2026-08-29T00:47:00+08:00",
+ source_filename="作品列表导出.xlsx",
+ )
+ clock["value"] = datetime.fromisoformat("2026-08-30T08:15:00+08:00")
+ second = content_review_service.commit_douyin_item_export(
+ account_id=account_id,
+ items=[
+ _diagnosis_item(
+ 1,
+ title="周口径测试作品",
+ published_at="2026-08-28T10:00:00+08:00",
+ play_count=2222,
+ )
+ ],
+ captured_at="2026-08-30T08:14:00+08:00",
+ source_filename="作品列表导出.xlsx",
+ )
+ clock["value"] = datetime.fromisoformat("2026-09-01T09:00:00+08:00")
+ preview = content_review_service.preview_metric_import(
+ account_id=account_id,
+ filename="账号趋势.xlsx",
+ content=_xlsx_bytes(),
+ )
+ content_review_service.commit_metric_import(preview["batch_id"])
+
+ summary = content_review_service.get_content_review_summary(account_id)
+
+ assert first["matched_count"] == second["matched_count"] == 1
+ assert summary["last_export_batch_id"] == second["batch_id"]
+ assert summary["last_export_committed_at"] == "2026-08-30T08:15:00+08:00"
+ assert summary["last_export_row_count"] == 1
+ assert summary["last_export_matched_count"] == 1
+ assert summary["last_export_unmatched_count"] == 0
+ assert summary["official_export_weeks"] == 1
+ assert summary["completed_cycles"] == 1
+ assert summary["last_sync_at"] == "2026-09-01T09:00:00+08:00"
+ assert job_id
+
+
+def test_diagnosis_uses_matched_metrics_and_skips_missing_cover_or_unmatched_works():
+ account_id = _insert_account()
+ export = _seed_diagnosis_baseline(account_id)
+
+ insights = content_review_service.get_content_review_insights(account_id)
+ works = content_review_service.list_content_review_works(account_id)
+
+ assert export["matched_count"] == 9
+ assert export["ambiguous_count"] == 0
+ assert insights["summary"] == {
+ "total_works": 10,
+ "eligible_works": 9,
+ "insufficient_works": 1,
+ "unmatched_works": 1,
+ "missing_metric_works": 0,
+ "recommendation_count": len(insights["recommendations"]),
+ "cover_metric_available": False,
+ "note": "作品级封面点击率缺失时不会生成封面建议。",
+ }
+ weak_opening = next(
+ item for item in insights["recommendations"] if item["diagnosis_code"] == "weak_opening"
+ )
+ assert weak_opening["data_sufficiency"] == "sufficient"
+ assert weak_opening["source_work"]["title"] == "诊断基线作品 8"
+ assert weak_opening["comparison_interval"]["two_second_bounce_rate"]["p75"] is not None
+ assert all(item["source_work"]["title"] != "没有候选的作品" for item in insights["recommendations"])
+ labels = {work["match_status"]: work["match_label"] for work in works}
+ assert labels["matched_unique"] == "已匹配·唯一证据"
+ assert labels["unmatched"] == "未匹配·没有候选"
+
+
+def test_prompt_comparison_only_includes_versions_used_by_selected_account():
+ first_account = _insert_account()
+ second_account = _insert_account()
+ first_job = _insert_publish_job(
+ first_account,
+ title="账号一作品",
+ published_at="2026-08-28T10:00:00+08:00",
+ )
+ second_job = _insert_publish_job(
+ second_account,
+ title="账号二作品",
+ published_at="2026-08-28T11:00:00+08:00",
+ )
+ first_version = _attach_prompt_chain(first_job, "账号一 Prompt")
+ second_version = _attach_prompt_chain(second_job, "账号二 Prompt")
+
+ first = content_review_service.get_prompt_comparison(first_account)
+ second = content_review_service.get_prompt_comparison(second_account)
+
+ assert [item["prompt_version_id"] for item in first["versions"]] == [first_version]
+ assert [item["prompt_version_id"] for item in second["versions"]] == [second_version]
+ assert first["completed_cycles"] == 0
+ assert second["completed_cycles"] == 0
+
+
+@pytest.mark.parametrize(
+ ("candidate_duration", "expected_ratio"),
+ [(60, 0.5), (0, 1 / 3)],
+)
+def test_prompt_comparison_uses_effective_duration_for_official_export_watch_ratio(
+ candidate_duration: int,
+ expected_ratio: float,
+):
+ account_id = _insert_account()
+ published_at = "2026-08-28T10:00:00+08:00"
+ job_id = _insert_publish_job(
+ account_id,
+ title="官方导出时长回退作品",
+ published_at=published_at,
+ duration_seconds=90,
+ )
+ version_id = _attach_prompt_chain(job_id, "时长回退 Prompt")
+ with get_connection() as connection:
+ connection.execute(
+ """
+ UPDATE clip_candidates
+ SET duration_seconds = ?
+ WHERE id = (
+ SELECT oc.clip_candidate_id
+ FROM publish_jobs pj
+ JOIN output_clip oc ON oc.id = pj.output_clip_id
+ WHERE pj.id = ?
+ )
+ """,
+ (candidate_duration, job_id),
+ )
+ connection.commit()
+ exported = content_review_service.commit_douyin_item_export(
+ account_id=account_id,
+ captured_at="2026-08-29T12:00:00+08:00",
+ source_filename="作品列表导出.xlsx",
+ items=[
+ _diagnosis_item(
+ 301,
+ title="官方导出时长回退作品",
+ published_at=published_at,
+ duration_seconds=None,
+ average_watch_seconds=30,
+ )
+ ],
+ )
+
+ comparison = content_review_service.get_prompt_comparison(account_id)
+ version = next(item for item in comparison["versions"] if item["prompt_version_id"] == version_id)
+
+ assert exported["matched_count"] == 1
+ assert version["average_watch_ratio"] == pytest.approx(expected_ratio)
+
+
+def test_experiment_assignment_is_unique_atomic_and_freezes_after_execution():
+ account_id = _insert_account()
+ _seed_diagnosis_baseline(account_id)
+ insights = content_review_service.get_content_review_insights(account_id)
+ recommendation = next(
+ item for item in insights["recommendations"] if item["diagnosis_code"] == "weak_opening"
+ )
+ created = content_review_service.create_content_experiment(
+ account_id,
+ recommendation["recommendation_id"],
+ )
+ first_experiment_id = created["experiment_id"]
+ assert created["experiment"]["progress"]["trend_visible"] is False
+ second_experiment_id = f"{PREFIX}experiment-second-{uuid4().hex[:8]}"
+ with get_connection() as connection:
+ first = connection.execute(
+ "SELECT * FROM content_improvement_experiments WHERE id = ?",
+ (first_experiment_id,),
+ ).fetchone()
+ connection.execute(
+ """
+ INSERT INTO content_improvement_experiments (
+ id, account_id, recommendation_id, diagnosis_code, title,
+ hypothesis, action_text, primary_metric, primary_direction,
+ guardrail_metrics_json, baseline_batch_id, baseline_json,
+ status, created_at, updated_at
+ ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'active', ?, ?)
+ """,
+ (
+ second_experiment_id,
+ account_id,
+ f"second-{uuid4().hex[:8]}",
+ first["diagnosis_code"],
+ "第二个实验",
+ first["hypothesis"],
+ first["action_text"],
+ first["primary_metric"],
+ first["primary_direction"],
+ first["guardrail_metrics_json"],
+ first["baseline_batch_id"],
+ first["baseline_json"],
+ first["created_at"],
+ first["updated_at"],
+ ),
+ )
+ connection.commit()
+ job_id = _insert_publish_job(
+ account_id,
+ title="实验作品",
+ published_at="2026-08-30T10:00:00+08:00",
+ status="WAITING",
+ )
+
+ assigned = content_review_service.set_publish_job_experiment(job_id, first_experiment_id)
+ reassigned = content_review_service.set_publish_job_experiment(job_id, second_experiment_id)
+ assert assigned["status"] == "assigned"
+ assert reassigned["status"] == "reassigned"
+ with get_connection() as connection:
+ count = connection.execute(
+ "SELECT COUNT(*) FROM content_improvement_experiment_items WHERE publish_job_id = ?",
+ (job_id,),
+ ).fetchone()[0]
+ owner = connection.execute(
+ "SELECT experiment_id FROM content_improvement_experiment_items WHERE publish_job_id = ?",
+ (job_id,),
+ ).fetchone()[0]
+ connection.execute(
+ """
+ UPDATE publish_jobs
+ SET status = 'PUBLISHING', claimed_at = '2026-08-30T10:05:00+08:00',
+ started_at = '2026-08-30T10:05:00+08:00', attempt_count = 1
+ WHERE id = ?
+ """,
+ (job_id,),
+ )
+ connection.commit()
+ assert count == 1
+ assert owner == second_experiment_id
+ with pytest.raises(content_review_service.ContentReviewError, match="已经开始"):
+ content_review_service.set_publish_job_experiment(job_id, "")
+ with pytest.raises(content_review_service.ContentReviewError, match="样本或官方导出周数还不足"):
+ content_review_service.update_content_experiment(first_experiment_id, "keep")
+
+
+def test_experiment_requires_twenty_treatment_twenty_baseline_and_three_export_weeks():
+ account_id = _insert_account()
+ _seed_diagnosis_baseline(account_id, captured_at="2026-08-29T12:00:00+08:00")
+ recommendation = next(
+ item
+ for item in content_review_service.get_content_review_insights(account_id)["recommendations"]
+ if item["diagnosis_code"] == "weak_opening"
+ )
+ created = content_review_service.create_content_experiment(
+ account_id,
+ recommendation["recommendation_id"],
+ )
+ experiment_id = created["experiment_id"]
+ with get_connection() as connection:
+ baseline = json.loads(
+ connection.execute(
+ "SELECT baseline_json FROM content_improvement_experiments WHERE id = ?",
+ (experiment_id,),
+ ).fetchone()[0]
+ )
+ baseline["work_count"] = 20
+ connection.execute(
+ "UPDATE content_improvement_experiments SET baseline_json = ? WHERE id = ?",
+ (json.dumps(baseline, ensure_ascii=False), experiment_id),
+ )
+ connection.commit()
+
+ treatment_items = []
+ for index in range(20):
+ published_at = f"2026-08-30T09:{index:02d}:00+08:00"
+ title = f"实验门槛作品 {index}"
+ job_id = _insert_publish_job(
+ account_id,
+ title=title,
+ published_at=published_at,
+ status="WAITING",
+ )
+ content_review_service.set_publish_job_experiment(job_id, experiment_id)
+ treatment_items.append(
+ _diagnosis_item(index + 200, title=title, published_at=published_at)
+ )
+
+ for week_index, captured_at in enumerate(
+ [
+ "2026-09-07T12:00:00+08:00",
+ "2026-09-14T12:00:00+08:00",
+ "2026-09-21T12:00:00+08:00",
+ ]
+ ):
+ weekly_items = [
+ {**item, "play_count": int(item["play_count"]) + week_index}
+ for item in treatment_items
+ ]
+ content_review_service.commit_douyin_item_export(
+ account_id=account_id,
+ items=weekly_items,
+ captured_at=captured_at,
+ source_filename="作品列表导出.xlsx",
+ )
+
+ experiment = next(
+ item
+ for item in content_review_service.list_content_experiments(account_id)
+ if item["id"] == experiment_id
+ )
+ assert experiment["progress"]["treatment_count"] == 20
+ assert experiment["progress"]["baseline_count"] == 20
+ assert experiment["progress"]["official_export_weeks"] == 3
+ assert experiment["progress"]["trend_visible"] is True
+ assert experiment["progress"]["decision_ready"] is True
+ decided = content_review_service.update_content_experiment(experiment_id, "inconclusive")
+ assert decided == {
+ "status": "completed",
+ "decision": "inconclusive",
+ "message": "实验结论已记录。",
+ }
diff --git a/tests/test_content_review_browser.py b/tests/test_content_review_browser.py
index 7807d6f..ccfcbbd 100644
--- a/tests/test_content_review_browser.py
+++ b/tests/test_content_review_browser.py
@@ -1,6 +1,7 @@
from __future__ import annotations
import os
+from datetime import datetime
from pathlib import Path
import socket
import threading
@@ -14,6 +15,7 @@
from app.db.database import get_connection, init_db # noqa: E402
from app.main import app # noqa: E402
+from app.services import content_review_service # noqa: E402
def _free_port() -> int:
@@ -38,9 +40,33 @@ def _seed_account() -> str:
@pytest.mark.parametrize("width", [1440, 390])
-def test_content_review_page_has_no_document_overflow(width: int):
+def test_content_review_page_has_no_document_overflow(width: int, monkeypatch, tmp_path):
init_db()
account_id = _seed_account()
+ monkeypatch.setattr(
+ content_review_service,
+ "_now",
+ lambda: datetime.fromisoformat("2026-08-29T00:48:37+08:00"),
+ )
+ content_review_service.commit_douyin_item_export(
+ account_id=account_id,
+ items=[
+ {
+ "aweme_id": "export:browser-test",
+ "title": "页面测试作品",
+ "published_at": "2026-08-28T20:00:00+08:00",
+ "duration_seconds": 60,
+ "play_count": 100,
+ "completion_rate": 0.4,
+ "five_second_completion_rate": 0.6,
+ "two_second_bounce_rate": 0.2,
+ "average_watch_seconds": 24,
+ "content_genre": "视频",
+ }
+ ],
+ captured_at="2026-08-29T00:47:00+08:00",
+ source_filename="作品列表导出.xlsx",
+ )
port = _free_port()
server = uvicorn.Server(
uvicorn.Config(
@@ -79,6 +105,63 @@ def test_content_review_page_has_no_document_overflow(width: int):
assert page.locator("h1").filter(has_text="内容复盘").is_visible()
assert page.locator("#content-review-import-form").is_visible()
assert page.locator("#content-review-sync").is_visible()
+ page.locator("#content-review-last-export").filter(
+ has_text="上次成功导出：北京时间 2026-08-29 00:48"
+ ).wait_for()
+ assert page.locator("#content-review-last-export-stats").inner_text() == (
+ "1 条作品 · 已匹配 0 · 待确认 0 · 未匹配 1"
+ )
+ assert page.locator("label.file-upload-button").is_visible()
+ assert not page.locator("#content-review-file").is_visible()
+ assert page.locator("#content-review-preview-button").is_disabled()
+ sample_file = tmp_path / "页面测试.xlsx"
+ sample_file.write_bytes(b"not-uploaded")
+ page.locator("#content-review-file").set_input_files(str(sample_file))
+ assert page.locator("#content-review-file-name").inner_text() == "页面测试.xlsx"
+ assert page.locator("#content-review-preview-button").is_enabled()
+ account_details = page.locator('[data-content-review-disclosure="account-history"]')
+ work_details = page.locator('[data-content-review-disclosure="work-attribution"]')
+ assert account_details.get_attribute("open") is None
+ assert work_details.get_attribute("open") is None
+ account_details.locator("summary").click()
+ assert account_details.get_attribute("open") is not None
+ page.reload(wait_until="networkidle")
+ assert page.locator(
+ '[data-content-review-disclosure="account-history"]'
+ ).get_attribute("open") is not None
+ assert page.locator(
+ '[data-content-review-disclosure="work-attribution"]'
+ ).get_attribute("open") is None
+ page.evaluate(
+ """
+ renderSummary({
+ latest_metric_date: '2026-08-28',
+ last_export_committed_at: '2026-08-29T00:48:37+08:00',
+ current_period: {two_second_bounce_rate: 0.30},
+ previous_period: {two_second_bounce_rate: 0.20},
+ comparisons: {two_second_bounce_rate: 0.50},
+ history: [],
+ match_summary: {}
+ })
+ """
+ )
+ bounce_delta = page.locator(
+ '[data-metric="two_second_bounce_rate"] [data-delta]'
+ )
+ assert "is-down" in (bounce_delta.get_attribute("class") or "")
+ page.evaluate(
+ """
+ renderSummary({
+ latest_metric_date: '2026-08-28',
+ current_period: {two_second_bounce_rate: 0.18},
+ previous_period: {two_second_bounce_rate: 0.20},
+ comparisons: {two_second_bounce_rate: -0.10},
+ history: [],
+ match_summary: {}
+ })
+ """
+ )
+ assert "is-up" in (bounce_delta.get_attribute("class") or "")
overflow = page.evaluate(
"document.documentElement.scrollWidth - window.innerWidth"
)
@@ -89,6 +172,19 @@ def test_content_review_page_has_no_document_overflow(width: int):
server.should_exit = True
thread.join(timeout=10)
with get_connection() as connection:
+ batch_rows = connection.execute(
+ "SELECT id FROM content_metric_import_batches WHERE account_id = ?",
+ (account_id,),
+ ).fetchall()
+ for batch in batch_rows:
+ connection.execute(
+ "DELETE FROM douyin_item_metric_snapshots WHERE batch_id = ?",
+ (batch["id"],),
+ )
+ connection.execute(
+ "DELETE FROM content_metric_import_batches WHERE account_id = ?",
+ (account_id,),
+ )
connection.execute(
"DELETE FROM publish_accounts WHERE id = ?",
(account_id,),
diff --git a/tests/test_content_review_foundation.py b/tests/test_content_review_foundation.py
index 2c0b3f1..ec0be17 100644
--- a/tests/test_content_review_foundation.py
+++ b/tests/test_content_review_foundation.py
@@ -66,6 +66,24 @@ def _insert_task_and_candidate(*, enabled: bool = True, reviewed: bool = False)
return task_id, clip_id
+def _insert_analysis_run(task_id: str, run_number: int, *, is_active: bool) -> str:
+ run_id = f"{task_id}-run-{run_number}"
+ now = "2026-08-28T10:00:00"
+ with get_connection() as connection:
+ connection.execute(
+ """
+ INSERT INTO ai_analysis_runs (
+ id, task_id, run_number, provider, provider_label, model,
+ requested_clip_count, clip_count, analysis_payload_json,
+ created_at, is_active
+ ) VALUES (?, ?, ?, 'test', 'Test', 'test-model', 1, 1, '{}', ?, ?)
+ """,
+ (run_id, task_id, run_number, now, int(is_active)),
+ )
+ connection.commit()
+ return run_id
+
+
def _payload(clip_id: str, *, enabled: bool, reason: str | None = None) -> ClipCandidateBatchItem:
return ClipCandidateBatchItem(
id=clip_id,
@@ -162,6 +180,64 @@ def test_legacy_feedback_endpoint_marks_source_and_latest_context_uses_one_per_c
assert task_context[0]["decision"] == "keep"
+def test_explicit_feedback_uses_candidate_source_run_instead_of_active_run():
+ task_id, clip_id = _insert_task_and_candidate(enabled=True, reviewed=False)
+ source_run_id = _insert_analysis_run(task_id, 1, is_active=False)
+ active_run_id = _insert_analysis_run(task_id, 2, is_active=True)
+ with get_connection() as connection:
+ connection.execute(
+ "UPDATE clip_candidates SET source_analysis_run_id = ? WHERE id = ?",
+ (source_run_id, clip_id),
+ )
+ connection.commit()
+
+ save_clip_feedback(
+ task_id,
+ clip_id,
+ ClipFeedbackCreate(decision="reject", reason_code="not_funny"),
+ )
+
+ with get_connection() as connection:
+ feedback = connection.execute(
+ "SELECT analysis_run_id FROM clip_feedback WHERE clip_candidate_id = ?",
+ (clip_id,),
+ ).fetchone()
+ assert feedback["analysis_run_id"] == source_run_id
+ assert feedback["analysis_run_id"] != active_run_id
+
+
+@pytest.mark.parametrize("source_kind", ["missing", "unknown", "cross_task"])
+def test_explicit_feedback_without_trustworthy_source_run_stays_unattributed(source_kind: str):
+ task_id, clip_id = _insert_task_and_candidate(enabled=True, reviewed=False)
+ active_run_id = _insert_analysis_run(task_id, 2, is_active=True)
+ source_run_id = None
+ if source_kind == "unknown":
+ source_run_id = f"{task_id}-missing-run"
+ elif source_kind == "cross_task":
+ other_task_id, _ = _insert_task_and_candidate(enabled=True, reviewed=False)
+ source_run_id = _insert_analysis_run(other_task_id, 1, is_active=True)
+ with get_connection() as connection:
+ connection.execute(
+ "UPDATE clip_candidates SET source_analysis_run_id = ? WHERE id = ?",
+ (source_run_id, clip_id),
+ )
+ connection.commit()
+
+ save_clip_feedback(
+ task_id,
+ clip_id,
+ ClipFeedbackCreate(decision="keep", reason_code="worth_publishing"),
+ )
+
+ with get_connection() as connection:
+ feedback = connection.execute(
+ "SELECT analysis_run_id FROM clip_feedback WHERE clip_candidate_id = ?",
+ (clip_id,),
+ ).fetchone()
+ assert feedback["analysis_run_id"] is None
+ assert feedback["analysis_run_id"] != active_run_id
+
+
def test_prompt_version_changes_only_when_prompt_content_changes():
suffix = uuid4().hex[:8]
preset_id = f"{PREFIX}preset-{suffix}"
diff --git a/tests/test_long_live_selection.py b/tests/test_long_live_selection.py
index 4b688f6..7357d58 100644
--- a/tests/test_long_live_selection.py
+++ b/tests/test_long_live_selection.py
@@ -267,6 +267,7 @@ def test_ambiguous_window_failure_is_not_retried_after_restart(tmp_path):
assert second.meta["failed_window_count"] == 1
assert second.meta["reused_window_count"] == second.meta["window_count"] - 1
assert second.meta["analysis_incomplete"] is True
+ assert second.meta["quality_degraded"] is False
def test_corrupt_completed_window_checkpoint_is_not_rebilled(tmp_path):
@@ -328,6 +329,8 @@ def test_safe_rate_limit_window_is_retried_within_bound(tmp_path):
result = analyze_long_live_talk(request, provider=provider, sleep_fn=lambda _seconds: None)
assert provider.calls[1] == 3
assert result.meta["failed_window_count"] == 0
+ assert result.meta["analysis_incomplete"] is False
+ assert result.meta["quality_degraded"] is False
def test_stale_worker_cannot_update_long_live_checkpoint():
diff --git a/tests/test_partial_ai_analysis.py b/tests/test_partial_ai_analysis.py
index 5614cc5..b18e992 100644
--- a/tests/test_partial_ai_analysis.py
+++ b/tests/test_partial_ai_analysis.py
@@ -236,6 +236,24 @@ def test_quality_degraded_analysis_is_manual_review_only(monkeypatch):
process_task_video_cuts("test-quality-cut")
+def test_complete_long_live_analysis_meta_passes_cut_validation():
+ meta = {
+ "schema_version": 2,
+ "selection_profile": "long_live_talk",
+ "analysis_incomplete": False,
+ "quality_degraded": False,
+ "coverage_ratio": 1.0,
+ "coverage_percent": 100.0,
+ "invalid_item_count": 0,
+ "window_count": 2,
+ "completed_window_count": 2,
+ "failed_window_count": 0,
+ "failed_windows": [],
+ }
+
+ assert validate_ai_analysis_meta_for_cut(meta, "long_live_talk") == meta
+
+
def test_missing_analysis_meta_blocks_manual_and_auto_cut(monkeypatch):
engine = PipelineEngine()
monkeypatch.setattr(engine, "_get_task", lambda _task_id: {"selection_profile": "general"})
diff --git a/tests/test_schema_migration_ledger.py b/tests/test_schema_migration_ledger.py
index d198e40..4b81da8 100644
--- a/tests/test_schema_migration_ledger.py
+++ b/tests/test_schema_migration_ledger.py
@@ -48,7 +48,7 @@ def test_init_records_migration_once_and_switches_unique_index(isolated_database
).fetchall()
indexes = _index_names(connection)
- assert len(migrations) == 4
+ assert len(migrations) == 5
migrations_by_version = {row["version"]: row for row in migrations}
publish_migration = migrations_by_version[database_module.PUBLISH_ACTIVE_INDEX_MIGRATION_VERSION]
assert publish_migration["name"] == database_module.PUBLISH_ACTIVE_INDEX_MIGRATION_NAME
@@ -66,15 +66,27 @@ def test_init_records_migration_once_and_switches_unique_index(isolated_database
assert export_migration["name"] == database_module.DOUYIN_ITEM_EXPORT_MIGRATION_NAME
assert export_migration["checksum"] == database_module.DOUYIN_ITEM_EXPORT_MIGRATION_CHECKSUM
assert export_migration["applied_at"]
+ feedback_migration = migrations_by_version[database_module.CONTENT_FEEDBACK_LOOP_MIGRATION_VERSION]
+ assert feedback_migration["name"] == database_module.CONTENT_FEEDBACK_LOOP_MIGRATION_NAME
+ assert feedback_migration["checksum"] == database_module.CONTENT_FEEDBACK_LOOP_MIGRATION_CHECKSUM
+ assert feedback_migration["applied_at"]
assert database_module.PUBLISH_ACTIVE_UNIQUE_INDEX_NAME in indexes
assert database_module.PUBLISH_ACTIVE_UNIQUE_INDEX_LEGACY_NAME not in indexes
assert set(database_module.CONTENT_REVIEW_REQUIRED_INDEXES) <= indexes
+ assert set(database_module.CONTENT_FEEDBACK_LOOP_REQUIRED_INDEXES) <= indexes
with _connect(isolated_database) as connection:
item_columns = {
row["name"]
for row in connection.execute("PRAGMA table_info(douyin_item_metric_snapshots)")
}
+ tables = {
+ row["name"]
+ for row in connection.execute(
+ "SELECT name FROM sqlite_master WHERE type = 'table'"
+ )
+ }
assert set(database_module.DOUYIN_ITEM_EXPORT_COLUMNS) <= item_columns
+ assert set(database_module.CONTENT_FEEDBACK_LOOP_REQUIRED_TABLES) <= tables
def test_checksum_drift_refuses_startup(isolated_database):