"""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") os.environ["DEMAND_WEIGHT_DATA_DIR"] = str(run_root / "intermediate" / "data") 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 os.environ.setdefault("DEMAND_ITEMSET_DETAIL_MAX_IDS", "12") os.environ.setdefault("DEMAND_ITEMSET_DETAIL_MAX_POST_IDS", "20") 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()