import os from typing import Any, Dict, List, Optional from fastapi import Depends, FastAPI, Header, HTTPException API_KEY = os.getenv("ONEOK_ADAPTER_API_KEY", "").strip() BASE_RATE_PER_M2 = float(os.getenv("ONEOK_BASE_RATE_PER_M2", "3200")) INSTALL_RATE_PER_M2 = float(os.getenv("ONEOK_INSTALL_RATE_PER_M2", "900")) CURRENCY = os.getenv("ONEOK_CURRENCY", "UAH") app = FastAPI(title="1OK Calc Adapter", version="1.0.0") def _auth(authorization: Optional[str] = Header(default=None)) -> None: if not API_KEY: return expected = f"Bearer {API_KEY}" if authorization != expected: raise HTTPException(status_code=401, detail="Unauthorized") def _normalize_units(payload: Dict[str, Any]) -> List[Dict[str, Any]]: units = payload.get("window_units") if isinstance(units, list) and units: return [u for u in units if isinstance(u, dict)] # Compatibility alias used by some LLM tool calls. units = payload.get("windows") if isinstance(units, list) and units: return [u for u in units if isinstance(u, dict)] # fallback to one unit in root payload if "width_mm" in payload and "height_mm" in payload: return [payload] return [] @app.get("/health") def health() -> Dict[str, Any]: return {"status": "ok", "service": "oneok-calc-adapter"} @app.post("/calc/window_quote") def calc_window_quote(input_payload: Dict[str, Any], _: None = Depends(_auth)) -> Dict[str, Any]: units = _normalize_units(input_payload) if not units: raise HTTPException(status_code=400, detail="window_units/windows or width_mm/height_mm required") line_items: List[Dict[str, Any]] = [] subtotal = 0.0 for idx, unit in enumerate(units, 1): width_mm = float(unit.get("width_mm") or 0) height_mm = float(unit.get("height_mm") or 0) if width_mm <= 0 or height_mm <= 0: continue area_m2 = (width_mm * height_mm) / 1_000_000.0 base = round(area_m2 * BASE_RATE_PER_M2, 2) install = round(area_m2 * INSTALL_RATE_PER_M2, 2) total = round(base + install, 2) subtotal += total line_items.append( { "item": unit.get("label") or f"window_{idx}", "width_mm": width_mm, "height_mm": height_mm, "area_m2": round(area_m2, 3), "base_price": base, "install_price": install, "total": total, } ) if not line_items: raise HTTPException(status_code=400, detail="No valid window units for calculation") assumptions = [ f"Базова ставка: {BASE_RATE_PER_M2} {CURRENCY}/м2", f"Монтаж: {INSTALL_RATE_PER_M2} {CURRENCY}/м2", "ОЦІНКА: без виїзного підтвердженого заміру ціна попередня.", ] lead_days = 10 if str(input_payload.get("urgency", "")).lower() in {"urgent", "терміново"}: lead_days = 7 return { "currency": CURRENCY, "line_items": line_items, "totals": { "subtotal": round(subtotal, 2), "discount": 0.0, "grand_total": round(subtotal, 2), }, "assumptions": assumptions, "lead_time_if_known": f"{lead_days} днів", }