"""Pairing UI (port 8088): 6-digit code entry, submits to cloud via WebSocket; when claimed shows connection info.""" import asyncio import html import re from typing import Any, Callable from fastapi import FastAPI, Request, Response from fastapi.responses import HTMLResponse, JSONResponse from pydantic import BaseModel # IPv4 with optional :port (same as main app) HOST_IP_PATTERN = re.compile(r"^(?:[0-9]{1,3}\.){3}[0-9]{1,3}(?::\d+)?$") def _connected_html(tenant_name: str, is_online: bool) -> str: """HTML when bridge is claimed: status (Online/Offline) + tenant; persists on refresh until unpaired.""" safe_name = html.escape(tenant_name.strip() or "Standort") if is_online: badge_class = "badge badge-online" badge_label = "Online" badge_dot = '' else: badge_class = "badge badge-offline" badge_label = "Offline" badge_dot = '' return f""" OrderPy Bridge – Status
{badge_dot} {badge_label}

Bridge-Status

Verbunden mit {safe_name}

""" PAIRING_HTML = """ OrderPy Bridge – Verbinden
✓

Verbunden!

Das Gerät wurde erfolgreich mit dem Standort verbunden.

Sie können dieses Fenster schließen.

Bridge mit OrderPy verbinden

Geben Sie den 6-stelligen Code ein, der in der OrderPy-App angezeigt wird.

""" class PairRequest(BaseModel): code: str def create_pairing_app( pairing_queue: asyncio.Queue, get_connection_info: Callable[[], dict[str, Any]] | None = None, ) -> FastAPI: """Create FastAPI app for pairing UI. pairing_queue sends code to cloud client. If get_connection_info is set and returns claimed=True, index shows connection info (online + tenant name) instead of the code form.""" app = FastAPI(title="OrderPy Bridge Pairing", version="1.0") @app.middleware("http") async def host_header_check(request: Request, call_next): host = request.headers.get("host", "") if not HOST_IP_PATTERN.match(host): return Response("Forbidden: Direct IP access only", status_code=403) return await call_next(request) @app.get("/", response_class=HTMLResponse) async def index(request: Request) -> Response: # Always reflect current state; prevent cached pairing form when already connected no_cache = {"Cache-Control": "no-store, no-cache, must-revalidate"} if get_connection_info: info = get_connection_info() if info.get("claimed"): html_content = _connected_html( info.get("tenant_name") or "", info.get("is_online", False), ) return HTMLResponse(html_content, headers=no_cache) return HTMLResponse(PAIRING_HTML, headers=no_cache) @app.post("/api/pair") async def pair(body: PairRequest) -> Response: if get_connection_info: info = get_connection_info() if info.get("claimed"): return JSONResponse( status_code=409, content={"ok": False, "reason": "already_connected"}, headers={"Cache-Control": "no-store"}, ) code = "".join(c for c in (body.code or "").strip() if c.isdigit())[:6] if len(code) != 6 or not code.isdigit(): return JSONResponse({"ok": False, "reason": "invalid_code"}) loop = asyncio.get_event_loop() future: asyncio.Future = loop.create_future() try: pairing_queue.put_nowait((code, future)) result = await asyncio.wait_for(future, timeout=15.0) return JSONResponse({"ok": result.get("ok", False), "reason": result.get("reason")}) except asyncio.TimeoutError: return JSONResponse({"ok": False, "reason": "timeout"}) except asyncio.QueueFull: return JSONResponse({"ok": False, "reason": "busy"}) return app