feat(elevator): 对齐 V1 lib 的 Davinci/扫描/事件与部署配置

- davinci-manager-storage：FilePart 路径与基址按 V1 JAR（/portal/file、/part/*、GET /download）
- 启动类：扫描 cn.cloudwalk.serial 与 cn.cloudwalk.cwos.client.resource，补 UUIDSerial 与 ApplicationService
- deploy：v1/v2 application 中 cloudwalk.serial.enabled、Kafka 指向 192.168.3.12:9092；deploy/.gitignore 忽略日志
- cloudwalk-common-serial：补充 META-INF/spring.factories（Boot 自动配置）
- 电梯：Session 配置、Davinci Bean、Feign 包、MQTT/Visitor/Zone Feign；部署脚本与 API parity 工具更新
- 文档与根脚本若干；未纳入大体积 jar/zip 与 v1 CFR 对比目录

Made-with: Cursor

Former-commit-id: b76d142d13ebb5c0898de2d9d11bc583876829c2

maven-cw-elevator-application/tools/elevator_api_parity/parity/catalog_loader.py

@@ -1,8 +1,42 @@
 import json
 from pathlib import Path
+from typing import Any, Dict, Optional
 
 _ROOT = Path(__file__).resolve().parent.parent
 
 
 def load() -> dict:
     return json.loads((_ROOT / "api_catalog.json").read_text(encoding="utf-8"))
+
+
+def endpoint_body(ep: Dict[str, Any]) -> Dict[str, Any]:
+    """Resolve request JSON: inline ``body`` wins, else load ``fixture`` file."""
+    if ep.get("body") is not None:
+        return dict(ep["body"])
+    fix = ep.get("fixture")
+    if fix:
+        p = _ROOT / "fixtures" / fix
+        return json.loads(p.read_text(encoding="utf-8"))
+    return {}
+
+
+def iter_endpoints(catalog: dict, *, tag: Optional[str] = None) -> list[dict]:
+    """If ``tag`` is set, only endpoints with ``tags`` containing it (or legacy entries with no tags = all)."""
+    out: list[dict] = []
+    for ep in catalog.get("endpoints", []):
+        tags = ep.get("tags") or []
+        if tag is None:
+            out.append(ep)
+        elif not tags:
+            out.append(ep)
+        elif tag in tags:
+            out.append(ep)
+    return out
+
+
+def include_in_parity(ep: dict) -> bool:
+    return bool(ep.get("include_in_parity", True))
+
+
+def include_in_smoke(ep: dict) -> bool:
+    return bool(ep.get("include_in_smoke", True))

maven-cw-elevator-application/tools/elevator_api_parity/parity/client.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import json
 import os
+import time
 from dataclasses import dataclass
 from typing import Any, Optional
 
@@ -124,6 +125,52 @@ def _safe_json(text: str) -> Any:
         return None
 
 
+def call_single(
+    name: str,
+    method: str,
+    path: str,
+    body: Any,
+    base_url: str,
+    session: requests.Session | None = None,
+) -> dict[str, Any]:
+    """One HTTP call for smoke coverage; returns a flat dict for reporting."""
+    s = session or requests.Session()
+    h = default_headers()
+    url = base_url.rstrip("/") + path
+    data = (
+        None
+        if body is None
+        else (body if isinstance(body, str) else json.dumps(body, ensure_ascii=False))
+    )
+    m = method.upper()
+    t0 = time.perf_counter()
+    if m == "GET":
+        r = s.get(url, headers=h, timeout=120)
+    else:
+        r = s.post(url, headers=h, data=data, timeout=120)
+    elapsed_ms = int((time.perf_counter() - t0) * 1000)
+    txt = r.text or ""
+    oj = _safe_json(txt)
+    bc = compare.business_code(oj) if oj is not None else None
+    head = txt[:400].replace("\n", " ")
+    return {
+        "name": name,
+        "method": m,
+        "path": path,
+        "http_status": r.status_code,
+        "elapsed_ms": elapsed_ms,
+        "business_code": bc,
+        "response_head": head,
+        "reachable": True,
+    }
+
+
+def can_reach_one(base_url: str, s: requests.Session | None = None) -> tuple[bool, str]:
+    s = s or requests.Session()
+    _, ok, _ = probe_healthy(base_url, s)
+    return ok, base_url
+
+
 def can_reach_both(
     base_old: str, base_new: str, s: requests.Session | None = None
 ) -> tuple[bool, str]: