video-image-production-build/run_production.py

"""从 GlobalDataStageDelivery 启动或恢复 Production 0.7 正式闭环。"""

from __future__ import annotations

import argparse
import json
from pathlib import Path
from uuid import uuid4

from dotenv import load_dotenv

from production_build_agents.production.runtime import run_production_graph


PROJECT_ROOT = Path(__file__).resolve().parent


def run_full_production(
    global_data_delivery_path: Path,
    *,
    output_root: Path,
    thread_id: str,
    shared_visual_anchor_path: Path,
    **runtime_options,
):
    """Production 0.7 CLI façade；正式控制流只存在于 LangGraph。"""

    load_dotenv(PROJECT_ROOT / ".env", override=False)
    return run_production_graph(
        global_data_delivery_path,
        output_root=output_root,
        thread_id=thread_id,
        shared_visual_anchor_path=shared_visual_anchor_path,
        **runtime_options,
    )


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description=(
            "运行或恢复 Production 0.7：Planner → Segment DAG → "
            "Assembly → Production Validator → Replan"
        )
    )
    parser.add_argument(
        "global_data_delivery",
        type=Path,
        help="Global Data 阶段正式 GlobalDataStageDelivery JSON",
    )
    parser.add_argument(
        "--output-dir",
        type=Path,
        default=Path("demo_output"),
    )
    parser.add_argument("--thread-id", default=None)
    parser.add_argument(
        "--shared-visual-anchor",
        type=Path,
        required=True,
        help="已在 Production 外部封印并接纳的共享视觉锚点图片",
    )
    parser.add_argument(
        "--max-plan-revisions",
        type=int,
        default=5,
        choices=range(0, 6),
        metavar="{0..5}",
        help="本次运行允许的 ADAPT/REPLAN 总次数；协议硬上限为 5",
    )
    parser.add_argument(
        "--pause-after-progress-review",
        action="store_true",
        help=(
            "在下一次 Segment Progress Review 完成并保存 checkpoint 后"
            "以 RUNNING 状态安全暂停；恢复时省略此参数"
        ),
    )
    return parser


def main() -> None:
    args = _build_parser().parse_args()
    result = run_full_production(
        args.global_data_delivery,
        output_root=args.output_dir,
        thread_id=args.thread_id or f"production-{uuid4()}",
        shared_visual_anchor_path=args.shared_visual_anchor,
        max_plan_revisions=args.max_plan_revisions,
        pause_after_progress_review=args.pause_after_progress_review,
    )
    print(json.dumps(result, ensure_ascii=False, indent=2))
    if result.get("status") == "COMPLETED":
        return
    if (
        args.pause_after_progress_review
        and result.get("status") == "RUNNING"
    ):
        return
    raise SystemExit(1)


if __name__ == "__main__":
    main()