SupplyAgent/scripts/verify_find_agent_flow.py

"""Smoke-check find_agent demand flow without calling the LLM."""
from __future__ import annotations

import json
import sys

from agents.find_agent.demand_run import (
    build_find_agent_user_input,
    build_run_input_payload,
    filter_pending_contexts,
    list_find_demand_contexts,
    pick_find_demand_context,
    prepare_video_discovery_run,
)
from agents.find_agent.async_runner import _run_coroutine


def main() -> int:
    biz_dt, contexts = list_find_demand_contexts()
    pending, stats = filter_pending_contexts(contexts, biz_dt)
    ctx = pick_find_demand_context(biz_dt, index=0)
    if ctx is None:
        print(json.dumps({"ok": False, "error": "no context"}, ensure_ascii=False))
        return 1

    run_id, skip_reason = prepare_video_discovery_run(ctx, force=True)
    if skip_reason or not run_id:
        print(json.dumps({"ok": False, "skip_reason": skip_reason}, ensure_ascii=False))
        return 1

    prompt = build_find_agent_user_input(ctx, run_id)
    payload = build_run_input_payload(ctx)

    async def _noop():
        return None

    # 验证主线程 asyncio 运行器可用
    _run_coroutine(_noop())

    print(
        json.dumps(
            {
                "ok": True,
                "biz_dt": biz_dt,
                "total": len(contexts),
                "pending": len(pending),
                "stats": stats,
                "run_id": run_id,
                "prompt_lines": len(prompt.splitlines()),
                "payload_keys": list(payload.keys()),
            },
            ensure_ascii=False,
        )
    )
    return 0


if __name__ == "__main__":
    sys.exit(main())