"""Ai-DOP 第三方对接模拟器 · 本地控制服务（任务书 P1-B / P1-G）。 - 只绑定 127.0.0.1，页面与 API 仅本机可用； - Secret/密码/Token 只从环境变量读取，任何 API 不回显； - 所有对外目标 host 走白名单硬阻断（默认拒绝）； - 运行记录（RunStore）写入前统一脱敏； - 内嵌管理 Mock API（providers.mock_api），默认端口 8018，可启停/切鉴权/注入错误； - 三层报告：代码层 SUPPORTED/UNSUPPORTED、配置层 READY/NOT_*、数据层 NOT_RUN/ACCEPTED/TRANSFORM_PENDING/RUN_FAILED。 启动：python tools/integration-simulator/app.py （详见 README.md） """ from __future__ import annotations import asyncio import os import socket import sys import time import uuid from pathlib import Path SIM_ROOT = Path(__file__).resolve().parent if str(SIM_ROOT) not in sys.path: sys.path.insert(0, str(SIM_ROOT)) import uvicorn from fastapi import FastAPI, HTTPException from fastapi.responses import FileResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel from catalog import sample_adapters as sa from clients import aidop_probe from clients.inbound_client import InboundClient from clients.redaction import mask_access_key, redact_obj from providers import mock_api, mock_db DEFAULT_CONTROL_PORT = 8017 MAX_RUNS = 200 # 状态枚举（任务书 §4） CONFIG_STATES = ("READY", "CONFIG_NOT_REGISTERED", "CONFIG_NOT_ENABLED", "GRANT_MISSING", "SOURCE_UNREACHABLE", "RUN_FAILED") def _env(name: str) -> str: return os.environ.get(name, "") or "" def _secrets() -> list[str]: return [v for v in (_env("AIDOP_SIM_INBOUND_SECRET"), _env("AIDOP_SIM_MYSQL_PASSWORD"), _env("AIDOP_SIM_ADMIN_TOKEN")) if v] def load_config() -> dict: cfg = { "aidopBaseUrl": "http://127.0.0.1:5005", "mockApiPort": 8018, "mysqlDatabase": mock_db.SIM_DB_DEFAULT, "allowedTargetHosts": ["127.0.0.1", "localhost"], } cfg_file = SIM_ROOT / "config.json" if cfg_file.exists(): import json with cfg_file.open("r", encoding="utf-8") as f: cfg.update(json.load(f)) return cfg CONFIG = load_config() MOCK_STATE = mock_api.MockApiState() MOCK_APP = mock_api.create_app(MOCK_STATE) class RunStore: def __init__(self, limit: int = MAX_RUNS): self.limit = limit self.runs: list[dict] = [] def add(self, run_id: str, object_code: str, channel: str, request_summary: dict, response: object, layers: dict, followup_sql: list[str] | None = None, gaps: list[str] | None = None, scenario_id: str | None = None, nodes: list[dict] | None = None) -> dict: entry = { "runId": run_id, "time": time.strftime("%Y-%m-%dT%H:%M:%S%z"), "objectCode": object_code, "channel": channel, "scenarioId": scenario_id, "nodes": nodes, "request": redact_obj(request_summary, _secrets()), "response": redact_obj(response, _secrets()), "layers": layers, "followupSql": followup_sql or [], "knownGaps": gaps or [], } self.runs.append(entry) if len(self.runs) > self.limit: del self.runs[: len(self.runs) - self.limit] return entry def recent(self) -> list[dict]: return list(reversed(self.runs[-50:])) RUNS = RunStore() app = FastAPI(title="Ai-DOP Integration Simulator", version="1.0.0") # --------------------------------------------------------------------------- # Mock API 内嵌启停 # --------------------------------------------------------------------------- class _MockServerHandle: def __init__(self): self.server: uvicorn.Server | None = None self.task: asyncio.Task | None = None self.last_error: str = "" MOCK_HANDLE = _MockServerHandle() def _port_in_use(port: int) -> bool: """真实探测本机端口是否已被占用。 不依赖 uvicorn 内部属性：0.30+ 的 Server 已无 startup_error， 读它会抛 AttributeError 并把已启动的服务误报成 500。 """ with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe: probe.settimeout(0.3) return probe.connect_ex(("127.0.0.1", int(port))) == 0 async def _start_mock_api() -> dict: port = int(CONFIG["mockApiPort"]) if MOCK_HANDLE.task and not MOCK_HANDLE.task.done(): return {"running": True, "port": port, "alreadyRunning": True} # 端口预检必须在建 task 之前：否则失败路径会留下一个已赋值的句柄， # 使 _mock_running() 与实际监听状态不一致。 if _port_in_use(port): MOCK_HANDLE.last_error = f"端口 {port} 已被占用（可能是另一个模拟器实例）" return {"running": False, "error": MOCK_HANDLE.last_error} config = uvicorn.Config(MOCK_APP, host="127.0.0.1", port=port, log_level="warning") server = uvicorn.Server(config) MOCK_HANDLE.server = server MOCK_HANDLE.task = asyncio.get_running_loop().create_task(server.serve()) # 轮询库的公开状态位，上限 ~2s：比固定 sleep 既不会过早返回未就绪，也不会白等。 for _ in range(40): await asyncio.sleep(0.05) if server.started: MOCK_HANDLE.last_error = "" return {"running": True, "port": port} if server.should_exit or MOCK_HANDLE.task.done(): break MOCK_HANDLE.last_error = f"Mock API 在端口 {port} 上启动失败" failed_task = MOCK_HANDLE.task MOCK_HANDLE.server = None MOCK_HANDLE.task = None if failed_task and not failed_task.done(): failed_task.cancel() return {"running": False, "error": MOCK_HANDLE.last_error} async def _stop_mock_api() -> dict: if MOCK_HANDLE.server and MOCK_HANDLE.task and not MOCK_HANDLE.task.done(): MOCK_HANDLE.server.should_exit = True try: await asyncio.wait_for(MOCK_HANDLE.task, timeout=5) except asyncio.TimeoutError: MOCK_HANDLE.task.cancel() return {"running": False} return {"running": False, "alreadyStopped": True} def _mock_running() -> bool: return bool(MOCK_HANDLE.task and not MOCK_HANDLE.task.done() and MOCK_HANDLE.server and MOCK_HANDLE.server.started) # --------------------------------------------------------------------------- # 基础模型 # --------------------------------------------------------------------------- class RowsPayload(BaseModel): objectCode: str runId: str | None = None rows: list[dict] | None = None class InboundExecPayload(BaseModel): objectCode: str | None = None entityCode: str | None = None operation: str = "push" # schema|push|receipt|bulk|snapshot_open|snapshot_commit|digest runId: str | None = None rows: list[dict] | None = None contractVersion: str | None = None # 用例与操作参数 case: str | None = None # happy|missing-idem|tamper-body|expired-ts|nonce-replay|same-key-replay|same-key-diff-body|bad-signature syncBatchId: str | None = None snapshotId: str | None = None force: bool = False bizDate: str | None = None class ScenarioPayload(BaseModel): scenarioId: str runId: str | None = None channelOverride: dict[str, str] | None = None # --------------------------------------------------------------------------- # 公共辅助 # --------------------------------------------------------------------------- def _require_obj(object_code: str) -> dict: obj = sa.find_object(object_code) if not obj: raise HTTPException(status_code=404, detail=f"unknown objectCode: {object_code}") return obj def _build_inbound_client() -> InboundClient: base = CONFIG["aidopBaseUrl"] try: aidop_probe.assert_host_allowed(base, CONFIG["allowedTargetHosts"]) except aidop_probe.HostNotAllowedError as ex: raise HTTPException(status_code=400, detail=str(ex)) access_key = _env("AIDOP_SIM_INBOUND_ACCESS_KEY") secret = _env("AIDOP_SIM_INBOUND_SECRET") missing = [n for n, v in (("AIDOP_SIM_INBOUND_ACCESS_KEY", access_key), ("AIDOP_SIM_INBOUND_SECRET", secret)) if not v] if missing: raise HTTPException(status_code=400, detail=f"缺少环境变量: {', '.join(missing)}（Secret 只从环境变量读取，不落盘）") return InboundClient(base, access_key, secret) def _resolve_entity(payload: InboundExecPayload) -> tuple[dict | None, str]: """解析本次执行的对象与 entityCode。仅给 entityCode 时允许无目录对象直发（rows 必填）。""" if payload.objectCode: obj = _require_obj(payload.objectCode) else: obj = _require_obj_by_entity(payload.entityCode or "") if payload.entityCode: entity = payload.entityCode.upper() else: if not obj: raise HTTPException(status_code=400, detail="need objectCode or entityCode") inbound = obj.get("apiInbound") or {} if not inbound.get("supported"): raise HTTPException(status_code=400, detail=f"{obj['objectCode']} 不支持 API_INBOUND: {inbound.get('reason', '无契约')}") entity = inbound["entityCode"] return obj, entity def _require_obj_by_entity(entity_code: str) -> dict | None: for obj in sa.load_catalog()["objects"]: if (obj.get("apiInbound") or {}).get("entityCode") == entity_code.upper(): return obj return None def _layers(code: str, config: str, data: str) -> dict: return {"code": code, "config": config, "data": data} def _followup_sql(obj: dict, channel: str, run_id: str) -> list[str]: prefix = sa.biz_prefix(run_id) db = (obj.get("dbSync") or {}).get("table") sqls = [ "SELECT source_code, status, COUNT(*) cnt, MAX(end_time) latest FROM mdp_sync_log " "WHERE source_code LIKE 'SIM_MODE3%' GROUP BY source_code, status;", "SELECT COUNT(*) FROM mdp_inbound_request WHERE create_time >= CURDATE();", ] if channel == "DB_SYNC" and db: sqls.insert(0, f"-- 模拟库核对：\nSELECT COUNT(*), MAX(sourceUpdatedAt) FROM `{db}` WHERE bizKey LIKE '{prefix}%';") sqls.append(f"SELECT * FROM `{db.replace('sim_', 'mdp_stg_sim_')}` WHERE raw_data LIKE '%{prefix}%' LIMIT 20;") if channel == "API_INBOUND": entity = (obj.get("apiInbound") or {}).get("entityCode") sqls.append(f"SELECT status, accepted_rows, rejected_rows, sync_batch_id FROM mdp_inbound_request " f"WHERE entity_code='{entity}' ORDER BY id DESC LIMIT 10;") sqls.append("-- 标准层/KPI 下游核验属第二阶段自动对账范围；第一阶段报告 TRANSFORM_PENDING，不猜测成功。") return sqls # --------------------------------------------------------------------------- # 状态与目录 # --------------------------------------------------------------------------- @app.get("/api/status") def status(): inbound_key = _env("AIDOP_SIM_INBOUND_ACCESS_KEY") return { "controlPort": DEFAULT_CONTROL_PORT, "aidopBaseUrl": CONFIG["aidopBaseUrl"], "allowedTargetHosts": CONFIG["allowedTargetHosts"], "aidopReachable": None, # 由 /api/precheck 实测，避免每次状态查询打真实请求 "mockApi": {"running": _mock_running(), "port": CONFIG["mockApiPort"], "authMode": MOCK_STATE.settings.auth_mode, "forceStatus": MOCK_STATE.settings.force_status, "lastError": MOCK_HANDLE.last_error}, "mysql": { "database": _env("AIDOP_SIM_MYSQL_DATABASE") or mock_db.SIM_DB_DEFAULT, "passwordLoaded": bool(_env("AIDOP_SIM_MYSQL_PASSWORD")), }, "inbound": { "accessKeyLoaded": bool(inbound_key), "accessKeyMasked": mask_access_key(inbound_key), "secretLoaded": bool(_env("AIDOP_SIM_INBOUND_SECRET")), }, "adminTokenLoaded": bool(_env("AIDOP_SIM_ADMIN_TOKEN")), "objectCounts": _channel_counts(), } def _channel_counts() -> dict: counts = {"DB_SYNC": 0, "API_PULL": 0, "API_INBOUND": 0, "total": 0} for obj in sa.load_catalog()["objects"]: counts["total"] += 1 if (obj.get("dbSync") or {}).get("supported"): counts["DB_SYNC"] += 1 if (obj.get("apiPull") or {}).get("supported"): counts["API_PULL"] += 1 if (obj.get("apiInbound") or {}).get("supported"): counts["API_INBOUND"] += 1 return counts @app.get("/api/catalog") def catalog(): return sa.load_catalog() @app.get("/api/scenarios") def scenarios(): return sa.load_scenarios() @app.get("/api/sample/{object_code}") def get_sample(object_code: str, runId: str | None = None): obj = _require_obj(object_code) run_id = runId or sa.new_run_id() inbound = obj.get("apiInbound") or {} result: dict = {"objectCode": obj["objectCode"], "runId": run_id, "canonical": sa.load_canonical_rows(obj), "dbSync": None, "apiPull": None, "apiInbound": None} if (obj.get("dbSync") or {}).get("supported"): result["dbSync"] = sa.adapt_db_rows(obj, run_id) if (obj.get("apiPull") or {}).get("supported"): result["apiPull"] = sa.adapt_api_rows(obj, run_id) if inbound.get("supported"): adapted = sa.adapt_inbound_rows(obj, run_id) result["apiInbound"] = { "entityCode": inbound["entityCode"], "contractVersion": inbound.get("contractVersion", "v1"), "rows": adapted["rows"], "missingRequired": adapted["missing_required"], "envelope": sa.inbound_envelope(adapted["rows"]), } return result # --------------------------------------------------------------------------- # Mock API 控制 # --------------------------------------------------------------------------- @app.post("/api/mock-api/start") async def mock_api_start(): return await _start_mock_api() @app.post("/api/mock-api/stop") async def mock_api_stop(): return await _stop_mock_api() @app.get("/api/mock-api/settings") def mock_api_settings(): s = MOCK_STATE.settings return {"running": _mock_running(), "authMode": s.auth_mode, "delaySeconds": s.delay_seconds, "forceStatus": s.force_status, "responsePath": s.response_path, "apiKeyHeader": s.api_key_header, "overrides": sorted(MOCK_STATE.overrides.keys())} @app.post("/api/mock-api/control") def mock_api_control(payload: dict): """统一控制入口：auth_mode/token/basic/apikey/delay/force_status/response_path/override。""" s = MOCK_STATE.settings if "auth_mode" in payload: mode = str(payload["auth_mode"]).upper() if mode not in ("NONE", "TOKEN", "BASIC", "APIKEY", "OAUTH2"): raise HTTPException(status_code=400, detail=f"unknown auth_mode: {mode}") s.auth_mode = mode for key, attr in (("token", "token"), ("basic_user", "basic_user"), ("basic_password", "basic_password"), ("api_key_header", "api_key_header"), ("api_key_value", "api_key_value"), ("response_path", "response_path")): if payload.get(key): setattr(s, attr, str(payload[key])) if "delay_seconds" in payload: s.delay_seconds = max(0.0, float(payload["delay_seconds"])) if "force_status" in payload: s.force_status = int(payload["force_status"]) if payload["force_status"] else None if payload.get("override_path") and isinstance(payload.get("override_rows"), list): MOCK_STATE.overrides[payload["override_path"]] = payload["override_rows"] if payload.get("clear_overrides"): MOCK_STATE.overrides.clear() return {"ok": True, "settings": mock_api_settings()} @app.get("/api/mock-api/audit") def mock_api_audit(): return {"entries": redact_obj(MOCK_STATE.audit[-50:], _secrets())} # --------------------------------------------------------------------------- # 模拟 MySQL 控制 # --------------------------------------------------------------------------- @app.post("/api/db/ensure") def db_ensure(payload: RowsPayload): """初始化模拟 schema 并确保对象源表存在（DDL 由样例推断）。""" obj = _require_obj(payload.objectCode) db_spec = obj.get("dbSync") or {} if not db_spec.get("supported"): raise HTTPException(status_code=400, detail=f"{obj['objectCode']} 不支持 DB_SYNC") if not _env("AIDOP_SIM_MYSQL_PASSWORD"): raise HTTPException(status_code=400, detail="缺少环境变量 AIDOP_SIM_MYSQL_PASSWORD") run_id = payload.runId or sa.new_run_id() rows = payload.rows if payload.rows is not None else sa.adapt_db_rows(obj, run_id) try: result = mock_db.init_schema({db_spec["table"]: { "rows": rows, "businessKey": db_spec.get("businessKey"), "incrementColumn": db_spec.get("incrementColumn", "sourceUpdatedAt")}}) return {"ok": True, "runId": run_id, **result} except mock_db.BlockedDatabaseError as ex: raise HTTPException(status_code=400, detail=str(ex)) except Exception as ex: raise HTTPException(status_code=502, detail=f"MySQL 不可达或执行失败: {type(ex).__name__}: {ex}") @app.post("/api/db/load") def db_load(payload: RowsPayload): obj = _require_obj(payload.objectCode) db_spec = obj.get("dbSync") or {} if not db_spec.get("supported"): raise HTTPException(status_code=400, detail=f"{obj['objectCode']} 不支持 DB_SYNC") run_id = payload.runId or sa.new_run_id() rows = payload.rows if payload.rows is not None else sa.adapt_db_rows(obj, run_id) try: count = mock_db.upsert_rows(db_spec["table"], rows) stats = mock_db.table_stats(db_spec["table"], db_spec.get("incrementColumn", "sourceUpdatedAt")) except Exception as ex: raise HTTPException(status_code=502, detail=f"装载失败（先执行 /api/db/ensure？）: {type(ex).__name__}: {ex}") layers = _layers("SUPPORTED", "READY", "ACCEPTED") RUNS.add(run_id, obj["objectCode"], "DB_SYNC", {"action": "load", "table": db_spec["table"], "rowCount": count}, stats, layers, _followup_sql(obj, "DB_SYNC", run_id), gaps=["Ai-DOP 侧同步须由 MdpDbPullExecutor 真实执行；本操作只装载第三方模拟源库"]) return {"ok": True, "runId": run_id, "loaded": count, "stats": stats, "layers": layers} @app.post("/api/db/reset") def db_reset(payload: RowsPayload): obj = _require_obj(payload.objectCode) table = (obj.get("dbSync") or {}).get("table") if not table: raise HTTPException(status_code=400, detail="object has no dbSync table") try: mock_db.reset_table(table) return {"ok": True, "table": table} except Exception as ex: raise HTTPException(status_code=502, detail=f"{type(ex).__name__}: {ex}") @app.post("/api/db/update") def db_update(payload: dict): obj = _require_obj(payload.get("objectCode", "")) db_spec = obj.get("dbSync") or {} table = db_spec.get("table") updates = payload.get("updates") or {} where = payload.get("where") or {} if not table or not updates or not where: raise HTTPException(status_code=400, detail="need objectCode/updates/where") try: affected = mock_db.update_rows(table, updates, where, db_spec.get("incrementColumn", "sourceUpdatedAt"), payload.get("newIncrement")) return {"ok": True, "table": table, "affected": affected} except Exception as ex: raise HTTPException(status_code=502, detail=f"{type(ex).__name__}: {ex}") @app.get("/api/db/stats/{object_code}") def db_stats(object_code: str): obj = _require_obj(object_code) db_spec = obj.get("dbSync") or {} if not db_spec.get("table"): raise HTTPException(status_code=400, detail="object has no dbSync table") try: return mock_db.table_stats(db_spec["table"], db_spec.get("incrementColumn", "sourceUpdatedAt")) except Exception as ex: raise HTTPException(status_code=502, detail=f"{type(ex).__name__}: {ex}") @app.get("/api/db/aidop-config/{object_code}") def db_aidop_config(object_code: str): obj = _require_obj(object_code) base = f"http://127.0.0.1:{CONFIG['mockApiPort']}" return mock_db.generate_aidop_config(obj, base) # --------------------------------------------------------------------------- # 预检与三层报告（P1-G） # --------------------------------------------------------------------------- @app.get("/api/precheck") def precheck(): base = CONFIG["aidopBaseUrl"] reach = aidop_probe.check_reachable(base, CONFIG["allowedTargetHosts"]) inbound_ready = bool(_env("AIDOP_SIM_INBOUND_ACCESS_KEY") and _env("AIDOP_SIM_INBOUND_SECRET")) env_missing = [n for n in ("AIDOP_SIM_INBOUND_ACCESS_KEY", "AIDOP_SIM_INBOUND_SECRET", "AIDOP_SIM_MYSQL_PASSWORD") if not _env(n)] return { "aidopBaseUrl": base, "aidop": reach, "mockApi": {"running": _mock_running(), "port": CONFIG["mockApiPort"]}, "mysql": {"database": _env("AIDOP_SIM_MYSQL_DATABASE") or mock_db.SIM_DB_DEFAULT, "passwordLoaded": bool(_env("AIDOP_SIM_MYSQL_PASSWORD"))}, "inboundCredentialsLoaded": inbound_ready, "envMissing": env_missing, "testSources": mock_db.SIM_SOURCES, "note": "DB_SYNC/API_PULL 的来源、实体、租户登记须人工在 Ai-DOP 配置页核对（生成清单见 /api/db/aidop-config/{objectCode}）", } @app.get("/api/precheck/inbound") def precheck_inbound(): """26 个代码契约逐一 schema 预检（真实签名，只读）。""" client = _build_inbound_client() codes = sorted({(o.get("apiInbound") or {}).get("entityCode") for o in sa.load_catalog()["objects"] if (o.get("apiInbound") or {}).get("supported")}) results = aidop_probe.precheck_all_inbound(client, codes, CONFIG["allowedTargetHosts"]) return {"accessKey": mask_access_key(client.key), "count": len(results), "results": results} @app.get("/api/precheck/inbound/{entity_code}") def precheck_inbound_one(entity_code: str): client = _build_inbound_client() return aidop_probe.precheck_inbound_entity(client, entity_code.upper(), CONFIG["allowedTargetHosts"]) # --------------------------------------------------------------------------- # API_INBOUND 执行（P1-E） # --------------------------------------------------------------------------- @app.post("/api/inbound/exec") def inbound_exec(payload: InboundExecPayload): obj, entity = _resolve_entity(payload) client = _build_inbound_client() run_id = payload.runId or sa.new_run_id() contract_version = payload.contractVersion if obj is None: # 仅按 entityCode 直发（无目录样例），rows 必填 rows = payload.rows if rows is None and payload.operation in ("push", "bulk"): raise HTTPException(status_code=400, detail="entityCode-only exec requires rows") envelope = sa.inbound_envelope(rows or []) else: inbound = obj.get("apiInbound") or {} contract_version = contract_version or inbound.get("contractVersion") rows = payload.rows if payload.rows is not None else sa.adapt_inbound_rows(obj, run_id)["rows"] envelope = sa.inbound_envelope(rows) object_code = obj["objectCode"] if obj else entity op = payload.operation result: dict = {"runId": run_id, "entityCode": entity, "operation": op, "case": payload.case} def record(status: int, body, headers, layers: dict, extra: dict | None = None): result.update({"httpStatus": status, "response": body, "requestHeaders": headers, **(extra or {})}) RUNS.add(run_id, object_code, "API_INBOUND", {"operation": op, "case": payload.case, "entityCode": entity, "rowCount": len(rows or []), "contractVersion": contract_version}, {"httpStatus": status, "body": body, "headers": headers}, layers, _followup_sql(obj or {"objectCode": entity, "apiInbound": {"entityCode": entity}}, "API_INBOUND", run_id)) return result # 预检（配置层） pre = aidop_probe.precheck_inbound_entity(client, entity, CONFIG["allowedTargetHosts"], contract_version) config_state = pre["configState"] result["precheck"] = {"configState": config_state, "httpStatus": pre["httpStatus"]} if op == "schema": status, body, headers = client.op_schema(entity, contract_version) layers = _layers("SUPPORTED", aidop_probe.classify_inbound_precheck(status, body), "NOT_RUN") return record(status, body, headers, layers) if config_state != "READY" and op in ("push", "bulk", "snapshot_open"): layers = _layers("SUPPORTED", config_state, "NOT_RUN") raise HTTPException(status_code=409, detail={ "message": f"配置层 {config_state}：普通推数只对 READY 实体执行（不伪造成功）", "precheck": pre, "layers": layers}) if op == "push": case = payload.case or "happy" if case == "happy": status, body, headers = client.op_push(entity, envelope, contract_version=contract_version) elif case == "missing-idem": status, body, headers = client.case_missing_idem(entity, envelope) elif case == "tamper-body": status, body, headers = client.case_tamper_body(entity, envelope) elif case == "expired-ts": status, body, headers = client.case_expired_ts(entity, envelope) elif case == "bad-signature": status, body, headers = client.case_bad_signature(entity, envelope) elif case == "nonce-replay": (s1, d1), (s2, d2) = client.case_nonce_replay(entity, envelope) status, body, headers = s2, {"first": [s1, d1], "replay": [s2, d2]}, {} elif case == "same-key-replay": (s1, d1), (s2, d2) = client.case_same_key_replay(entity, envelope) status, body, headers = s2, {"first": [s1, d1], "replay": [s2, d2]}, {} elif case == "same-key-diff-body": env2 = sa.inbound_envelope([{**r, "sourceUpdatedAt": sa.now_iso()} for r in (rows or [])]) (s1, d1), (s2, d2) = client.case_same_key_diff_body(entity, envelope, env2) status, body, headers = s2, {"first": [s1, d1], "conflict": [s2, d2]}, {} else: raise HTTPException(status_code=400, detail=f"unknown case: {case}") data_state = "ACCEPTED" if status == 202 else ("RUN_FAILED" if status >= 400 else "NOT_RUN") layers = _layers("SUPPORTED", config_state, data_state) return record(status, body, headers, layers, extra={"gaps": ["HTTP 202 仅代表摄取受理；贴源/标准层核验见 followupSql（TRANSFORM_PENDING）"]}) if op == "receipt": if not payload.syncBatchId: raise HTTPException(status_code=400, detail="receipt requires syncBatchId") status, body, headers = client.op_receipt(payload.syncBatchId) layers = _layers("SUPPORTED", config_state, "ACCEPTED" if status == 200 else "RUN_FAILED") return record(status, body, headers, layers) if op == "bulk": status, body, headers = client.op_bulk(entity, rows or [], contract_version=contract_version) layers = _layers("SUPPORTED", config_state, "ACCEPTED" if status == 202 else "RUN_FAILED") return record(status, body, headers, layers) if op == "snapshot_open": status, body, headers = client.op_snapshot_open(entity) layers = _layers("SUPPORTED", config_state, "ACCEPTED" if status in (200, 201) else "RUN_FAILED") return record(status, body, headers, layers) if op == "snapshot_commit": if not payload.snapshotId: raise HTTPException(status_code=400, detail="snapshot_commit requires snapshotId") status, body, headers = client.op_snapshot_commit(entity, payload.snapshotId, payload.force) layers = _layers("SUPPORTED", config_state, "ACCEPTED" if status in (200, 202) else "RUN_FAILED") return record(status, body, headers, layers) if op == "digest": biz_date = payload.bizDate or time.strftime("%Y-%m-%d") status, body, headers = client.op_digest(entity, biz_date) layers = _layers("SUPPORTED", config_state, "ACCEPTED" if status == 200 else "RUN_FAILED") return record(status, body, headers, layers) raise HTTPException(status_code=400, detail=f"unknown operation: {op}") # --------------------------------------------------------------------------- # 场景执行（P1-F） # --------------------------------------------------------------------------- @app.post("/api/scenarios/run") def scenario_run(payload: ScenarioPayload): scenario = next((s for s in sa.load_scenarios()["scenarios"] if s["scenarioId"] == payload.scenarioId), None) if not scenario: raise HTTPException(status_code=404, detail=f"unknown scenario: {payload.scenarioId}") run_id = payload.runId or sa.new_run_id() overrides = payload.channelOverride or {} nodes_out = [] for node in scenario["nodes"]: obj = sa.find_object(node["objectCode"]) channel = overrides.get(node["objectCode"], node["channel"]) node_result = {"objectCode": node["objectCode"], "channel": channel, "relationKey": node.get("relationKey")} if not obj: node_result.update({"state": "CODE_UNSUPPORTED", "message": "目录中无此对象"}) nodes_out.append(node_result) continue spec = {"DB_SYNC": obj.get("dbSync"), "API_PULL": obj.get("apiPull"), "API_INBOUND": obj.get("apiInbound")}.get(channel) if not spec or not spec.get("supported"): reason = (spec or {}).get("reason") or node.get("inboundNote") or "该通道不支持" node_result.update({"state": "CODE_UNSUPPORTED", "message": reason}) nodes_out.append(node_result) continue try: if channel == "DB_SYNC": rows = sa.adapt_db_rows(obj, run_id) if not _env("AIDOP_SIM_MYSQL_PASSWORD"): raise RuntimeError("缺少 AIDOP_SIM_MYSQL_PASSWORD") mock_db.init_schema({spec["table"]: {"rows": rows, "businessKey": spec.get("businessKey"), "incrementColumn": spec.get("incrementColumn")}}) count = mock_db.upsert_rows(spec["table"], rows) node_result.update({"state": "ACCEPTED", "loaded": count, "table": spec["table"]}) elif channel == "API_PULL": if not _mock_running(): raise RuntimeError("Mock API 未启动（先 POST /api/mock-api/start）") rows = sa.adapt_api_rows(obj, run_id) MOCK_STATE.overrides[spec["path"]] = rows node_result.update({"state": "ACCEPTED", "overridePath": spec["path"], "rowCount": len(rows), "note": "已在 Mock API 挂载适配样例；由 Ai-DOP MdpApiPullExecutor 真实拉取"}) else: client = _build_inbound_client() entity = spec["entityCode"] version = node.get("contractVersion") or spec.get("contractVersion") pre = aidop_probe.precheck_inbound_entity(client, entity, CONFIG["allowedTargetHosts"], version) if pre["configState"] != "READY": node_result.update({"state": pre["configState"], "message": f"配置层 {pre['configState']}，未推数（不伪造成功）"}) else: rows = sa.adapt_inbound_rows(obj, run_id)["rows"] status, body, _ = client.op_push(entity, sa.inbound_envelope(rows), contract_version=version) node_result.update({"state": "ACCEPTED" if status == 202 else "RUN_FAILED", "httpStatus": status, "response": body}) except HTTPException: raise except aidop_probe.HostNotAllowedError as ex: node_result.update({"state": "SOURCE_UNREACHABLE", "message": str(ex)}) except Exception as ex: node_result.update({"state": "RUN_FAILED", "message": f"{type(ex).__name__}: {ex}"}) nodes_out.append(node_result) unsupported = [n["objectCode"] for n in nodes_out if n["state"] == "CODE_UNSUPPORTED"] layers = _layers("SUPPORTED", "READY" if all(n["state"] not in ("SOURCE_UNREACHABLE",) for n in nodes_out) else "SOURCE_UNREACHABLE", "ACCEPTED" if all(n["state"] in ("ACCEPTED", "CODE_UNSUPPORTED") for n in nodes_out) else "RUN_FAILED") entry = RUNS.add(run_id, ",".join(n["objectCode"] for n in nodes_out), f"SCENARIO:{payload.scenarioId}", {"scenario": scenario["name"]}, {"nodes": nodes_out}, layers, _followup_sql({"objectCode": "SCENARIO", "dbSync": {}}, "SCENARIO", run_id), gaps=([f"以下节点该通道不支持：{', '.join(unsupported)}（链路不完整，不得宣称全链路跑通）"] if unsupported else []), scenario_id=payload.scenarioId, nodes=nodes_out) return entry # --------------------------------------------------------------------------- # 运行记录 # --------------------------------------------------------------------------- @app.get("/api/runs") def runs(): return {"runs": RUNS.recent()} # --------------------------------------------------------------------------- # 静态页面 # --------------------------------------------------------------------------- @app.get("/") def index(): return FileResponse(SIM_ROOT / "static" / "index.html") app.mount("/static", StaticFiles(directory=SIM_ROOT / "static"), name="static") def main() -> None: print(f"[simulator] control panel: http://127.0.0.1:{DEFAULT_CONTROL_PORT}/") print(f"[simulator] aidop target : {CONFIG['aidopBaseUrl']} (allowed hosts: {CONFIG['allowedTargetHosts']})") uvicorn.run(app, host="127.0.0.1", port=DEFAULT_CONTROL_PORT, log_level="info") if __name__ == "__main__": main()