feat(office): align heartbeat office registrations

server-tools/heartbeat-office-building/host_vehicle_registry.py

@@ -0,0 +1,57 @@
+"""Deterministic registered-host vehicle and Asia/Shanghai daily-slot gate."""
+from __future__ import annotations
+import json
+from datetime import datetime
+from pathlib import Path
+from zoneinfo import ZoneInfo
+
+ROOT = Path(__file__).resolve().parents[2]
+REGISTRY = ROOT / "routing/heartbeat-office-host-vehicle-registry.json"
+
+def local_day() -> str:
+    return datetime.now(ZoneInfo("Asia/Shanghai")).date().isoformat()
+
+def load() -> dict:
+    return json.loads(REGISTRY.read_text(encoding="utf-8"))
+
+def vehicle_for(host: str, registry: dict | None = None) -> dict | None:
+    d = registry or load()
+    host = host.lower()
+    for item in d["registered_vehicles"]:
+        if host == item["host"] or host in [str(x).lower() for x in item.get("aliases", [])]:
+            return item
+    return None
+
+def slot(host: str, day: str | None = None, registry: dict | None = None) -> dict:
+    d = registry or load(); day = day or local_day()
+    vehicle = vehicle_for(host, d)
+    if not vehicle:
+        return {"state":"UNREGISTERED_HOST","host":host,"day":day}
+    return d.setdefault("daily_slots", {}).get(day, {}).get(vehicle["host"], {
+        "host": vehicle["host"], "vehicle_id": vehicle["vehicle_id"],
+        "day": day, "state": "AUTHORIZATION_REQUIRED"
+    })
+
+def verify_registered_vehicle(host: str, vehicle_id: str, day: str | None = None, require_authorized: bool = False) -> dict:
+    vehicle = vehicle_for(host)
+    if not vehicle:
+        return {"outcome":"REJECTED","code":"HOST_VEHICLE_NOT_REGISTERED","host":host}
+    if vehicle["vehicle_id"] != vehicle_id:
+        return {"outcome":"REJECTED","code":"VEHICLE_ID_MISMATCH","host":host,"expected":vehicle["vehicle_id"],"received":vehicle_id}
+    current = slot(host, day)
+    if require_authorized and current.get("state") != "AUTHORIZED":
+        return {"outcome":"REJECTED","code":"DAILY_SLOT_AUTHORIZATION_REQUIRED","slot":current}
+    return {"outcome":"PASS","code":"REGISTERED_HOST_VEHICLE","vehicle":vehicle,"slot":current}
+
+def reset_slots(day: str | None = None) -> dict:
+    d=load(); day=day or local_day(); slots={}
+    for item in d["registered_vehicles"]:
+        slots[item["host"]]={"host":item["host"],"vehicle_id":item["vehicle_id"],"day":day,"state":"AUTHORIZATION_REQUIRED"}
+    d.setdefault("daily_slots", {})[day]=slots
+    REGISTRY.write_text(json.dumps(d,ensure_ascii=False,indent=2)+"\n",encoding="utf-8")
+    return {"outcome":"RESET","day":day,"slots":slots,"state":"AUTHORIZATION_REQUIRED"}
+
+if __name__ == "__main__":
+    import argparse
+    p=argparse.ArgumentParser(); p.add_argument("command",choices=["verify","reset"]); p.add_argument("host",nargs="?"); p.add_argument("vehicle_id",nargs="?"); p.add_argument("--authorized",action="store_true"); p.add_argument("--day")
+    a=p.parse_args(); value=reset_slots(a.day) if a.command=="reset" else verify_registered_vehicle(a.host,a.vehicle_id,require_authorized=a.authorized); print(json.dumps(value,ensure_ascii=False,indent=2))