"""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))