Add daily Douyin AIGC automation

scripts/daily_auto_douyin_and_aigc.py

@@ -0,0 +1,826 @@
+"""Daily CFA Douyin automation: select demands, run real Douyin jobs, push pooled items.
+
+Default mode is dry-run. Add ``--live`` to run CFA and call AIGC.
+The script is intentionally idempotent for systemd use:
+
+* reuse an already completed same-day run for the selected demand by default;
+* skip AIGC push when the same source run set was already registered;
+* filter platform content IDs that were already pushed before by default.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+import time
+import traceback
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+from aigc_push_registry.registry import AigcPushRegistry
+from content_agent.integrations.aigc_platform import AigcPlatformClient
+from content_agent.integrations.database_runtime import ContentSupplyDbConfig
+from content_agent.run_service import RunService, _merged_project_env
+from content_agent.schemas import RunStartRequest
+
+
+DEFAULT_PLATFORM = "douyin"
+DEFAULT_TOP_N = 2
+DEFAULT_STRATEGY_VERSION = "V4"
+DEFAULT_PLATFORM_MODE = "real"
+COMPLETED_RUN_STATUSES = ("success", "partial_success")
+
+
+@dataclass(frozen=True)
+class DemandPick:
+    demand_content_id: int
+    category: str
+    name: str
+    score: float
+    category_cnt: int | None = None
+
+
+@dataclass(frozen=True)
+class RunJob:
+    demand: DemandPick
+    platform: str
+
+
+@dataclass(frozen=True)
+class RunResult:
+    run_id: str
+    demand_content_id: int
+    category: str
+    platform: str
+    status: str | None
+    current_step: str | None
+    reused_existing: bool
+    elapsed_s: float | None = None
+
+
+def _today() -> str:
+    return datetime.now().strftime("%Y%m%d")
+
+
+def _utc_ts() -> str:
+    return datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
+
+
+def _emit(handle: Any, record: dict[str, Any]) -> None:
+    payload = dict(record)
+    payload["ts"] = datetime.now(timezone.utc).isoformat()
+    line = json.dumps(payload, ensure_ascii=False, default=str)
+    print(line, flush=True)
+    if handle is not None:
+        handle.write(line + "\n")
+        handle.flush()
+
+
+def _split_half(items: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
+    mid = (len(items) + 1) // 2
+    return items[:mid], items[mid:]
+
+
+def _dedupe_items(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    seen: set[str] = set()
+    result: list[dict[str, Any]] = []
+    for item in items:
+        content_id = str(item["platform_content_id"])
+        if content_id in seen:
+            continue
+        seen.add(content_id)
+        result.append({**item, "platform_content_id": content_id})
+    return result
+
+
+def _connect() -> Any:
+    return ContentSupplyDbConfig.from_env().connect()
+
+
+def _select_top_category_demands(date: str, top_n: int) -> list[DemandPick]:
+    sql_categories = """
+    SELECT merge_leve2 AS category, COUNT(*) AS cnt, MAX(score) AS top_score
+    FROM demand_content
+    WHERE dt = %s
+    GROUP BY merge_leve2
+    ORDER BY cnt DESC, top_score DESC, merge_leve2 ASC
+    LIMIT %s
+    """
+    sql_pick = """
+    SELECT id, merge_leve2 AS category, name, score
+    FROM demand_content
+    WHERE dt = %s AND merge_leve2 = %s
+    ORDER BY score DESC, id ASC
+    LIMIT 1
+    """
+    picks: list[DemandPick] = []
+    with _connect() as conn:
+        with conn.cursor() as cur:
+            cur.execute(sql_categories, (date, top_n))
+            for category_row in cur.fetchall():
+                cur.execute(sql_pick, (date, category_row["category"]))
+                row = cur.fetchone()
+                if row:
+                    picks.append(
+                        DemandPick(
+                            demand_content_id=int(row["id"]),
+                            category=str(row["category"]),
+                            name=str(row["name"]),
+                            score=float(row["score"]),
+                            category_cnt=int(category_row["cnt"]),
+                        )
+                    )
+    return picks
+
+
+def _select_top_score_demands(date: str, top_n: int) -> list[DemandPick]:
+    sql = """
+    SELECT id, merge_leve2 AS category, name, score
+    FROM demand_content
+    WHERE dt = %s
+    ORDER BY score DESC, id ASC
+    """
+    picks: list[DemandPick] = []
+    seen_categories: set[str] = set()
+    with _connect() as conn:
+        with conn.cursor() as cur:
+            cur.execute(sql, (date,))
+            for row in cur.fetchall():
+                category = str(row["category"])
+                if category in seen_categories:
+                    continue
+                seen_categories.add(category)
+                picks.append(
+                    DemandPick(
+                        demand_content_id=int(row["id"]),
+                        category=category,
+                        name=str(row["name"]),
+                        score=float(row["score"]),
+                    )
+                )
+                if len(picks) >= top_n:
+                    break
+    return picks
+
+
+def _select_explicit_demands(demand_ids: list[int]) -> list[DemandPick]:
+    if not demand_ids:
+        return []
+    placeholders = ", ".join(["%s"] * len(demand_ids))
+    sql = f"""
+    SELECT id, merge_leve2 AS category, name, score
+    FROM demand_content
+    WHERE id IN ({placeholders})
+    """
+    rows_by_id: dict[int, dict[str, Any]] = {}
+    with _connect() as conn:
+        with conn.cursor() as cur:
+            cur.execute(sql, tuple(demand_ids))
+            for row in cur.fetchall():
+                rows_by_id[int(row["id"])] = row
+    missing = [str(demand_id) for demand_id in demand_ids if demand_id not in rows_by_id]
+    if missing:
+        raise SystemExit(f"Demand ids not found: {', '.join(missing)}")
+    return [
+        DemandPick(
+            demand_content_id=int(row["id"]),
+            category=str(row["category"]),
+            name=str(row["name"]),
+            score=float(row["score"]),
+        )
+        for row in (rows_by_id[demand_id] for demand_id in demand_ids)
+    ]
+
+
+def select_demands(
+    *,
+    date: str,
+    top_n: int,
+    selection_mode: str,
+    demand_ids: list[int],
+) -> list[DemandPick]:
+    if demand_ids:
+        return _select_explicit_demands(demand_ids)
+    if selection_mode == "top-score-distinct-category":
+        return _select_top_score_demands(date, top_n)
+    return _select_top_category_demands(date, top_n)
+
+
+def _find_existing_completed_run(
+    *,
+    date: str,
+    demand_content_id: int,
+    platform: str,
+) -> dict[str, Any] | None:
+    sql = """
+    SELECT run_id, demand_content_id, platform, status, current_step, validation_status,
+           created_at, started_at, completed_at
+    FROM content_agent_runs
+    WHERE demand_content_id = %s
+      AND platform = %s
+      AND platform_mode = %s
+      AND status IN ('success', 'partial_success')
+      AND DATE(created_at) = STR_TO_DATE(%s, '%%Y%%m%%d')
+    ORDER BY completed_at DESC, id DESC
+    LIMIT 1
+    """
+    with _connect() as conn:
+        with conn.cursor() as cur:
+            cur.execute(sql, (demand_content_id, platform, DEFAULT_PLATFORM_MODE, date))
+            return cur.fetchone()
+
+
+def _find_active_run(*, demand_content_id: int, platform: str) -> dict[str, Any] | None:
+    sql = """
+    SELECT run_id, demand_content_id, platform, status, current_step, created_at, started_at
+    FROM content_agent_runs
+    WHERE demand_content_id = %s
+      AND platform = %s
+      AND platform_mode = %s
+      AND status IN ('created', 'running')
+    ORDER BY id DESC
+    LIMIT 1
+    """
+    with _connect() as conn:
+        with conn.cursor() as cur:
+            cur.execute(sql, (demand_content_id, platform, DEFAULT_PLATFORM_MODE))
+            return cur.fetchone()
+
+
+def run_jobs(
+    *,
+    jobs: list[RunJob],
+    date: str,
+    live: bool,
+    reuse_existing_runs: bool,
+    strategy_version: str,
+    log_handle: Any,
+) -> list[RunResult]:
+    results: list[RunResult] = []
+    service = RunService.from_env() if live else None
+    batch_ts = _utc_ts()
+    for idx, job in enumerate(jobs, start=1):
+        active = _find_active_run(
+            demand_content_id=job.demand.demand_content_id,
+            platform=job.platform,
+        )
+        if active:
+            _emit(log_handle, {"event": "run_active_skip", "job_index": idx, **active})
+            results.append(
+                RunResult(
+                    run_id=str(active["run_id"]),
+                    demand_content_id=job.demand.demand_content_id,
+                    category=job.demand.category,
+                    platform=job.platform,
+                    status=str(active["status"]),
+                    current_step=str(active.get("current_step")),
+                    reused_existing=True,
+                )
+            )
+            continue
+
+        existing = None
+        if reuse_existing_runs:
+            existing = _find_existing_completed_run(
+                date=date,
+                demand_content_id=job.demand.demand_content_id,
+                platform=job.platform,
+            )
+        if existing:
+            _emit(log_handle, {"event": "run_reuse_existing", "job_index": idx, **existing})
+            results.append(
+                RunResult(
+                    run_id=str(existing["run_id"]),
+                    demand_content_id=job.demand.demand_content_id,
+                    category=job.demand.category,
+                    platform=job.platform,
+                    status=str(existing["status"]),
+                    current_step=str(existing.get("current_step")),
+                    reused_existing=True,
+                )
+            )
+            continue
+
+        run_id = (
+            f"v1_run_{job.demand.demand_content_id}_{job.platform}_"
+            f"daily_auto_{date}_{batch_ts}_{idx:02d}"
+        )
+        _emit(
+            log_handle,
+            {
+                "event": "run_planned" if not live else "run_start",
+                "job_index": idx,
+                "run_id": run_id,
+                "demand_content_id": job.demand.demand_content_id,
+                "category": job.demand.category,
+                "name": job.demand.name,
+                "platform": job.platform,
+            },
+        )
+        if not live:
+            results.append(
+                RunResult(
+                    run_id=run_id,
+                    demand_content_id=job.demand.demand_content_id,
+                    category=job.demand.category,
+                    platform=job.platform,
+                    status="planned",
+                    current_step=None,
+                    reused_existing=False,
+                )
+            )
+            continue
+
+        start = time.time()
+        try:
+            assert service is not None
+            state = service.start_run(
+                RunStartRequest(
+                    run_id=run_id,
+                    demand_content_id=job.demand.demand_content_id,
+                    platform=job.platform,
+                    platform_mode=DEFAULT_PLATFORM_MODE,
+                    strategy_version=strategy_version,
+                )
+            )
+            elapsed_s = round(time.time() - start, 1)
+            _emit(
+                log_handle,
+                {
+                    "event": "run_done",
+                    "job_index": idx,
+                    "run_id": state.get("run_id"),
+                    "demand_content_id": job.demand.demand_content_id,
+                    "category": job.demand.category,
+                    "platform": job.platform,
+                    "status": state.get("status"),
+                    "current_step": state.get("current_step"),
+                    "error_code": state.get("error_code"),
+                    "error_message": state.get("error_message"),
+                    "elapsed_s": elapsed_s,
+                },
+            )
+            results.append(
+                RunResult(
+                    run_id=str(state["run_id"]),
+                    demand_content_id=job.demand.demand_content_id,
+                    category=job.demand.category,
+                    platform=job.platform,
+                    status=state.get("status"),
+                    current_step=state.get("current_step"),
+                    reused_existing=False,
+                    elapsed_s=elapsed_s,
+                )
+            )
+        except Exception as exc:  # noqa: BLE001 - keep the daily batch moving.
+            elapsed_s = round(time.time() - start, 1)
+            _emit(
+                log_handle,
+                {
+                    "event": "run_error",
+                    "job_index": idx,
+                    "run_id": run_id,
+                    "demand_content_id": job.demand.demand_content_id,
+                    "category": job.demand.category,
+                    "platform": job.platform,
+                    "error": str(exc),
+                    "traceback": traceback.format_exc()[-3000:],
+                    "elapsed_s": elapsed_s,
+                },
+            )
+    return results
+
+
+def _pooled_items_for_runs(run_results: list[RunResult], platform: str) -> list[dict[str, Any]]:
+    run_ids = [
+        result.run_id
+        for result in run_results
+        if result.platform == platform and result.status in COMPLETED_RUN_STATUSES
+    ]
+    if not run_ids:
+        return []
+    field_placeholders = ", ".join(["%s"] * len(run_ids))
+    sql = """
+    SELECT
+      rd.run_id,
+      di.id AS discovered_id,
+      di.platform_content_id,
+      di.description AS content_title,
+      di.author_display_name,
+      rd.decision_id,
+      rd.id AS rule_decision_row_id
+    FROM content_agent_rule_decisions rd
+    JOIN content_agent_discovered_content_items di
+      ON rd.run_id = di.run_id AND rd.decision_target_id = di.platform_content_id
+    WHERE rd.run_id IN %s
+      AND di.platform = %s
+      AND rd.decision_action = 'ADD_TO_CONTENT_POOL'
+    ORDER BY FIELD(rd.run_id, """ + field_placeholders + """), di.id
+    """
+    with _connect() as conn:
+        with conn.cursor() as cur:
+            cur.execute(sql, (tuple(run_ids), platform, *run_ids))
+            return [dict(row) for row in cur.fetchall()]
+
+
+def _already_pushed_content_ids(platform: str, content_ids: list[str]) -> set[str]:
+    if not content_ids:
+        return set()
+    with _connect() as conn:
+        with conn.cursor() as cur:
+            found: set[str] = set()
+            chunk_size = 500
+            for offset in range(0, len(content_ids), chunk_size):
+                chunk = content_ids[offset : offset + chunk_size]
+                cur.execute(
+                    """
+                    SELECT DISTINCT i.platform_content_id
+                    FROM content_agent_aigc_push_items i
+                    JOIN content_agent_aigc_push_records r
+                      ON r.push_record_id = i.push_record_id
+                    WHERE r.platform = %s
+                      AND r.push_status = 'pushed'
+                      AND i.platform_content_id IN %s
+                    """,
+                    (platform, tuple(chunk)),
+                )
+                found.update(str(row["platform_content_id"]) for row in cur.fetchall())
+            return found
+
+
+def _find_existing_push_for_sources(
+    *,
+    merged_run_id: str,
+    source_run_ids: list[str],
+    push_record_prefix: str,
+) -> list[dict[str, Any]]:
+    clauses = ["run_id = %s", "push_record_id LIKE %s"]
+    params: list[Any] = [merged_run_id, f"{push_record_prefix}%"]
+    for run_id in source_run_ids:
+        clauses.append("(raw_payload LIKE %s OR source_ref LIKE %s)")
+        params.extend([f"%{run_id}%", f"%{run_id}%"])
+    sql = f"""
+    SELECT push_record_id, run_id, platform, push_status, pushed_at, source_ref
+    FROM content_agent_aigc_push_records
+    WHERE push_status = 'pushed'
+      AND ({' OR '.join(clauses)})
+    ORDER BY pushed_at DESC, id DESC
+    """
+    with _connect() as conn:
+        with conn.cursor() as cur:
+            cur.execute(sql, tuple(params))
+            return [dict(row) for row in cur.fetchall()]
+
+
+def _safe_create_enabled(env: dict[str, str], force_create_plan: bool) -> bool:
+    if force_create_plan:
+        return True
+    value = str(os.environ.get("CAN_NOT_CREATE_PLAN") or env.get("CAN_NOT_CREATE_PLAN", "true"))
+    return value.strip().lower() not in ("1", "true", "yes")
+
+
+def push_aigc(
+    *,
+    date: str,
+    platform: str,
+    run_results: list[RunResult],
+    live: bool,
+    history_dedupe: bool,
+    force_push: bool,
+    force_create_plan: bool,
+    log_handle: Any,
+) -> dict[str, Any]:
+    source_run_ids = [result.run_id for result in run_results if result.platform == platform]
+    completed_source_run_ids = [
+        result.run_id
+        for result in run_results
+        if result.platform == platform and result.status in COMPLETED_RUN_STATUSES
+    ]
+    merged_run_id = f"merged_{platform}_{date}_daily_auto"
+    push_record_prefix = f"aigc_push_{merged_run_id}_"
+    existing = _find_existing_push_for_sources(
+        merged_run_id=merged_run_id,
+        source_run_ids=source_run_ids,
+        push_record_prefix=push_record_prefix,
+    )
+    if existing and not force_push:
+        _emit(
+            log_handle,
+            {
+                "event": "aigc_push_skip_existing",
+                "run_id": merged_run_id,
+                "source_run_ids": source_run_ids,
+                "matches": existing,
+            },
+        )
+        return {"status": "skipped_existing", "matches": existing, "content_count": 0}
+
+    if not completed_source_run_ids:
+        _emit(
+            log_handle,
+            {
+                "event": "aigc_push_skip_no_completed_runs",
+                "run_id": merged_run_id,
+                "source_run_ids": source_run_ids,
+            },
+        )
+        return {"status": "skipped_no_completed_runs", "content_count": 0}
+
+    pooled = _dedupe_items(_pooled_items_for_runs(run_results, platform))
+    original_count = len(pooled)
+    already_pushed: set[str] = set()
+    if history_dedupe:
+        already_pushed = _already_pushed_content_ids(
+            platform,
+            [str(item["platform_content_id"]) for item in pooled],
+        )
+        pooled = [item for item in pooled if str(item["platform_content_id"]) not in already_pushed]
+    if not pooled:
+        _emit(
+            log_handle,
+            {
+                "event": "aigc_push_skip_no_new_items",
+                "run_id": merged_run_id,
+                "source_run_ids": source_run_ids,
+                "pooled_content_count": original_count,
+                "history_dedupe_applied": history_dedupe,
+                "previously_pushed_count": len(already_pushed),
+            },
+        )
+        return {"status": "skipped_no_new_items", "content_count": 0}
+
+    env = _merged_project_env()
+    plan_id_1 = env.get(f"CONTENTFIND_AIGC_PRODUCE_PLAN_ID_{platform.upper()}_1")
+    plan_id_2 = env.get(f"CONTENTFIND_AIGC_PRODUCE_PLAN_ID_{platform.upper()}_2")
+    if not plan_id_1 or not plan_id_2:
+        raise SystemExit(f"Missing AIGC produce plan ids for {platform}")
+
+    first, second = _split_half(pooled)
+    parts = [
+        {"plan_part": "1", "produce_plan_id": plan_id_1, "items": first},
+        {"plan_part": "2", "produce_plan_id": plan_id_2, "items": second},
+    ]
+    ts = _utc_ts()
+    push_record_id = f"{push_record_prefix}{ts}"
+    _emit(
+        log_handle,
+        {
+            "event": "aigc_push_plan",
+            "mode": "live" if live else "dry_run",
+            "push_record_id": push_record_id,
+            "run_id": merged_run_id,
+            "source_run_ids": source_run_ids,
+            "platform": platform,
+            "total_content_count": len(pooled),
+            "split_counts": [len(part["items"]) for part in parts],
+            "history_dedupe_applied": history_dedupe,
+            "previously_pushed_count": len(already_pushed),
+        },
+    )
+    if not live:
+        return {
+            "status": "planned",
+            "push_record_id": push_record_id,
+            "content_count": len(pooled),
+            "split_counts": [len(part["items"]) for part in parts],
+        }
+
+    if not _safe_create_enabled(env, force_create_plan):
+        raise SystemExit(
+            "CAN_NOT_CREATE_PLAN is true. Set CAN_NOT_CREATE_PLAN=false or pass --force-create-plan."
+        )
+
+    client = AigcPlatformClient.from_env({**env, "CAN_NOT_CREATE_PLAN": "false"})
+    pushed_at = None
+    for part in parts:
+        content_ids = [str(item["platform_content_id"]) for item in part["items"]]
+        name = f"【内容寻找Agent自动创建】抖音视频直接抓取-{ts}-{merged_run_id}-{part['plan_part']}"
+        label = f"原始帖子-视频-抖音-内容添加计划-{merged_run_id}-{part['plan_part']}"
+        _emit(
+            log_handle,
+            {
+                "event": "crawler_plan_create_start",
+                "push_record_id": push_record_id,
+                "run_id": merged_run_id,
+                "source_run_ids": source_run_ids,
+                "platform": platform,
+                "half": part["plan_part"],
+                "produce_plan_id": part["produce_plan_id"],
+                "content_count": len(content_ids),
+                "content_ids": content_ids,
+            },
+        )
+        crawler_plan_id = client.create_crawler_plan(content_ids, platform=platform, name=name)
+        client.bind_crawler_to_produce(
+            crawler_plan_id=crawler_plan_id,
+            produce_plan_id=str(part["produce_plan_id"]),
+            label=label,
+            platform=platform,
+        )
+        now = datetime.now(timezone.utc).isoformat()
+        pushed_at = now
+        part["crawler_plan_id"] = crawler_plan_id
+        part["pushed_at"] = now
+        _emit(
+            log_handle,
+            {
+                "event": "crawler_plan_bound",
+                "status": "ok",
+                "push_record_id": push_record_id,
+                "run_id": merged_run_id,
+                "source_run_ids": source_run_ids,
+                "platform": platform,
+                "half": part["plan_part"],
+                "produce_plan_id": part["produce_plan_id"],
+                "crawler_plan_id": crawler_plan_id,
+                "content_count": len(content_ids),
+                "content_ids": content_ids,
+            },
+        )
+
+    assert pushed_at is not None
+    record = {
+        "push_record_id": push_record_id,
+        "run_id": merged_run_id,
+        "platform": platform,
+        "demand_name": "daily_auto",
+        "demand_dt": date,
+        "hive_merge_leve2": ",".join(dict.fromkeys(result.category for result in run_results)),
+        "evidence_platform": platform,
+        "run_label": merged_run_id,
+        "run_status": "pushed",
+        "pushed_at": pushed_at,
+        "push_status": "pushed",
+        "source_kind": "daily_auto_live",
+        "source_ref": ";".join(source_run_ids),
+        "uploaded_by": "daily_auto_douyin_and_aigc",
+        "notes": "Daily automated merged push of ADD_TO_CONTENT_POOL items.",
+        "raw_payload": {
+            "source_run_ids": source_run_ids,
+            "date": date,
+            "platform": platform,
+            "total_content_count": len(pooled),
+            "split_counts": [len(part["items"]) for part in parts],
+            "history_dedupe_applied": history_dedupe,
+            "previously_pushed_count": len(already_pushed),
+        },
+        "plans": [],
+    }
+    for part in parts:
+        record["plans"].append(
+            {
+                "plan_part": part["plan_part"],
+                "produce_plan_id": part["produce_plan_id"],
+                "crawler_plan_id": part["crawler_plan_id"],
+                "content_count": len(part["items"]),
+                "pushed_at": part["pushed_at"],
+                "items": [
+                    {
+                        "platform_content_id": str(item["platform_content_id"]),
+                        "decision_id": item.get("decision_id"),
+                        "content_title": item.get("content_title"),
+                        "author_display_name": item.get("author_display_name"),
+                        "raw_payload": {
+                            "source_run_id": item.get("run_id"),
+                            "discovered_id": item.get("discovered_id"),
+                            "rule_decision_row_id": item.get("rule_decision_row_id"),
+                        },
+                    }
+                    for item in part["items"]
+                ],
+                "raw_payload": {"source_run_ids": source_run_ids},
+            }
+        )
+    registry = AigcPushRegistry.from_env(".env")
+    registry.init_db()
+    registry.upsert_record(record)
+    _emit(
+        log_handle,
+        {
+            "event": "registry_upserted",
+            "status": "ok",
+            "push_record_id": push_record_id,
+            "run_id": merged_run_id,
+            "platform": platform,
+            "content_count": len(pooled),
+            "plan_count": len(parts),
+        },
+    )
+    return {
+        "status": "pushed",
+        "push_record_id": push_record_id,
+        "content_count": len(pooled),
+        "plans": [
+            {
+                "plan_part": part["plan_part"],
+                "produce_plan_id": part["produce_plan_id"],
+                "crawler_plan_id": part["crawler_plan_id"],
+                "content_count": len(part["items"]),
+            }
+            for part in parts
+        ],
+    }
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Daily automation: select demand_content, run Douyin CFA, then push pooled content to AIGC."
+    )
+    parser.add_argument("--date", default=_today(), help="demand_content.dt, default today")
+    parser.add_argument("--top-n", type=int, default=DEFAULT_TOP_N, help="number of demands to select")
+    parser.add_argument(
+        "--selection-mode",
+        choices=("top-category-count", "top-score-distinct-category"),
+        default="top-category-count",
+        help="demand selection rule when --demand-id is not provided",
+    )
+    parser.add_argument("--demand-id", type=int, action="append", default=[], help="explicit demand id; repeatable")
+    parser.add_argument("--platform", default=DEFAULT_PLATFORM, choices=("douyin",), help="currently only douyin")
+    parser.add_argument("--strategy-version", default=DEFAULT_STRATEGY_VERSION)
+    parser.add_argument("--live", action="store_true", help="run CFA and push AIGC; default is dry-run")
+    parser.add_argument("--no-reuse-existing-runs", action="store_true")
+    parser.add_argument("--skip-push", action="store_true", help="run CFA only, do not push AIGC")
+    parser.add_argument("--no-history-dedupe", action="store_true", help="do not filter previously pushed content ids")
+    parser.add_argument("--force-push", action="store_true", help="ignore existing source-run registry matches")
+    parser.add_argument("--force-create-plan", action="store_true", help="bypass CAN_NOT_CREATE_PLAN safety switch")
+    args = parser.parse_args()
+
+    if args.top_n <= 0:
+        raise SystemExit("--top-n must be positive")
+
+    picks = select_demands(
+        date=args.date,
+        top_n=args.top_n,
+        selection_mode=args.selection_mode,
+        demand_ids=args.demand_id,
+    )
+    if not picks:
+        print(f"[skip] demand_content.dt={args.date} has no selected demands")
+        return
+    jobs = [RunJob(demand=pick, platform=args.platform) for pick in picks]
+
+    ts = _utc_ts()
+    log_path = Path("e2e_logs") / f"daily_auto_{args.platform}_{args.date}_{ts}.jsonl"
+    log_path.parent.mkdir(parents=True, exist_ok=True)
+
+    with log_path.open("a", encoding="utf-8") as handle:
+        _emit(
+            handle,
+            {
+                "event": "batch_start",
+                "mode": "live" if args.live else "dry_run",
+                "date": args.date,
+                "platform": args.platform,
+                "selection_mode": args.selection_mode,
+                "top_n": args.top_n,
+                "demand_ids": args.demand_id,
+                "log_path": str(log_path),
+                "picks": [pick.__dict__ for pick in picks],
+            },
+        )
+        run_results = run_jobs(
+            jobs=jobs,
+            date=args.date,
+            live=args.live,
+            reuse_existing_runs=not args.no_reuse_existing_runs,
+            strategy_version=args.strategy_version,
+            log_handle=handle,
+        )
+        push_result = None
+        if args.skip_push:
+            _emit(handle, {"event": "aigc_push_skipped_by_flag"})
+        else:
+            push_result = push_aigc(
+                date=args.date,
+                platform=args.platform,
+                run_results=run_results,
+                live=args.live,
+                history_dedupe=not args.no_history_dedupe,
+                force_push=args.force_push,
+                force_create_plan=args.force_create_plan,
+                log_handle=handle,
+            )
+        _emit(
+            handle,
+            {
+                "event": "batch_done",
+                "mode": "live" if args.live else "dry_run",
+                "date": args.date,
+                "run_results": [result.__dict__ for result in run_results],
+                "push_result": push_result,
+                "log_path": str(log_path),
+            },
+        )
+
+    print(f"完成。日志:{log_path}")
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except KeyboardInterrupt:
+        sys.exit(130)

tests/test_daily_auto_douyin_and_aigc.py

@@ -0,0 +1,25 @@
+from scripts.daily_auto_douyin_and_aigc import _dedupe_items, _split_half
+
+
+def test_split_half_rounds_first_half_up() -> None:
+    items = [{"platform_content_id": str(idx)} for idx in range(5)]
+
+    first, second = _split_half(items)
+
+    assert [item["platform_content_id"] for item in first] == ["0", "1", "2"]
+    assert [item["platform_content_id"] for item in second] == ["3", "4"]
+
+
+def test_dedupe_items_preserves_first_occurrence_order() -> None:
+    items = [
+        {"platform_content_id": "a", "source": "first"},
+        {"platform_content_id": "b", "source": "first"},
+        {"platform_content_id": "a", "source": "second"},
+    ]
+
+    deduped = _dedupe_items(items)
+
+    assert deduped == [
+        {"platform_content_id": "a", "source": "first"},
+        {"platform_content_id": "b", "source": "first"},
+    ]