demand-agent-new/examples/demand/run_existing_execution_mysql.py

"""Cloud MySQL entrypoint for an existing PG Pattern V2 execution.

This entrypoint intentionally writes only content-deconstruction-supply.demand_content.
It never runs prepare/mining and does not write demand_task, Hive, PG Pattern, or
legacy MySQL Pattern tables.
"""

from __future__ import annotations

import argparse
import asyncio
import json
import os
import sys
from pathlib import Path


sys.path.insert(0, str(Path(__file__).parent.parent.parent))


def _configure_mysql_env(run_label: str | None) -> None:
    os.environ["DEMAND_OUTPUT_MODE"] = "mysql_demand_content"
    os.environ["DEMAND_MYSQL_ENTRYPOINT"] = "run_existing_execution_mysql"
    if run_label:
        os.environ["DEMAND_RUN_LABEL"] = run_label


def _validate_execution_success(execution_id: int) -> None:
    from examples.demand.db_manager import query_execution_for_evidence

    execution = query_execution_for_evidence(execution_id)
    if not execution:
        raise ValueError(f"execution_id={execution_id} 不存在，MySQL 写入入口只允许跑已有 execution")
    if str(execution.get("status") or "").strip().lower() != "success":
        raise ValueError(
            f"execution_id={execution_id} status={execution.get('status')}，不是 success，拒绝生成需求"
        )


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Run DemandAgent against an existing PG Pattern V2 execution_id and write only MySQL demand_content."
    )
    parser.add_argument("--execution-id", type=int, required=True, help="Existing PG pattern_mining_execution.id")
    parser.add_argument("--merge-level2", required=True, help="Current merge_leve2 / cluster name")
    parser.add_argument("--platform-type", default="piaoquan", help="Platform label")
    parser.add_argument("--count", type=int, default=5, help="Approximate demand count requested from the prompt")
    parser.add_argument("--run-label", default=None, help="Label stored at ext_data.run_label for verification")
    return parser.parse_args()


async def async_main() -> dict:
    args = parse_args()
    _configure_mysql_env(args.run_label)
    _validate_execution_success(args.execution_id)

    from examples.demand.run import main

    result = await main(
        cluster_name=args.merge_level2,
        platform_type=args.platform_type,
        count=args.count,
        execution_id=args.execution_id,
        task_id=None,
    )
    result["run_label"] = args.run_label
    return result


def main() -> None:
    result = asyncio.run(async_main())
    print(json.dumps(result, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()