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
from datetime import datetime


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


def _load_project_env() -> None:
    try:
        from dotenv import load_dotenv
    except ImportError:
        return
    load_dotenv(Path(__file__).parent.parent.parent / ".env")


def _safe_path_segment(value: str) -> str:
    return "".join(ch if ch.isalnum() or ch in {"-", "_"} else "_" for ch in value).strip("_") or "run"


def _prepare_run_output_paths(execution_id: int, run_label: str) -> None:
    run_root = Path(__file__).parent / "test_output_data" / "mysql_runs" / _safe_path_segment(run_label)
    result_base_dir = run_root / "intermediate" / "result"
    os.environ["DEMAND_RESULT_BASE_DIR"] = str(result_base_dir)
    os.environ["DEMAND_OUTPUT_BASE_DIR"] = str(run_root / "output")
    os.environ["DEMAND_TRACE_STORE_PATH"] = str(run_root / ".trace")

    demand_items_path = result_base_dir / str(execution_id) / f"execution_id_{execution_id}_demand_items.json"
    if demand_items_path.exists():
        demand_items_path.unlink()


def _configure_mysql_env(run_label: str) -> None:
    os.environ["DEMAND_OUTPUT_MODE"] = "mysql_demand_content"
    os.environ["DEMAND_MYSQL_ENTRYPOINT"] = "run_existing_execution_mysql"
    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:
    _load_project_env()
    args = parse_args()
    run_label = args.run_label or f"mysql_{args.execution_id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
    _configure_mysql_env(run_label)
    _prepare_run_output_paths(args.execution_id, 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"] = 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()