#!/usr/bin/env python3 from __future__ import annotations from pathlib import Path import sys from fastapi import FastAPI, HTTPException, Form from fastapi.responses import HTMLResponse, RedirectResponse from pydantic import BaseModel ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from lib.life_noc_inputs import ( # noqa: E402 list_items, get_item, set_item, complete_item, get_defined_service, compute_state_for_item, ) class SetInputRequest(BaseModel): value: str origin: str = "manual" class CompleteInputRequest(BaseModel): origin: str = "manual" app = FastAPI(title="Life-NOC Input API", version="1.0.0") def html_escape(value: str) -> str: return ( str(value) .replace("&", "&") .replace("<", "<") .replace(">", ">") .replace('"', """) ) def link_row(label: str, url: str) -> str: if not url: return "" u = html_escape(url) l = html_escape(label) return f'
  • {l}
    • ' def resolve_instructions_url(domain: str, item_key: str, service: dict) -> str: explicit = str(service.get("instructions_url", "")).strip() if explicit: return explicit return f"/life-noc/item/{domain}/{item_key}" @app.get("/health") def health() -> dict: return {"status": "ok"} @app.get("/inputs/{domain}") def api_list_inputs(domain: str) -> dict: return {"domain": domain, "items": list_items(domain)} @app.get("/inputs/{domain}/{item_key}") def api_get_input(domain: str, item_key: str) -> dict: try: item = get_item(domain, item_key) except KeyError: raise HTTPException(status_code=404, detail="item introuvable") except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) return {"domain": domain, "item_key": item_key, **item} @app.post("/inputs/{domain}/{item_key}") def api_set_input(domain: str, item_key: str, req: SetInputRequest) -> dict: item = set_item(domain, item_key, req.value, origin=req.origin) return {"domain": domain, "item_key": item_key, **item} @app.post("/inputs/{domain}/{item_key}/complete") def api_complete_input(domain: str, item_key: str, req: CompleteInputRequest | None = None) -> dict: origin = "manual" if req is None else req.origin item = complete_item(domain, item_key, origin=origin) return {"domain": domain, "item_key": item_key, **item} @app.get("/life-noc/item/{domain}/{item_key}", response_class=HTMLResponse) def page_item(domain: str, item_key: str) -> HTMLResponse: try: service = get_defined_service(domain, item_key) current_input = get_item(domain, item_key) state = compute_state_for_item(domain, item_key) except KeyError: raise HTTPException(status_code=404, detail="item introuvable") except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) title = service.get("title") or service.get("display_name") or item_key.replace("-", " ").capitalize() summary = service.get("summary", "") notes = service.get("notes", "") instructions = service.get("instructions", "") notes_url = service.get("notes_url", "") instructions_url = resolve_instructions_url(domain, item_key, service) action_url = service.get("action_url", "") probe_type = service.get("probe", {}).get("type", "unknown") ui = service.get("ui", {}) if isinstance(service.get("ui", {}), dict) else {} form_mode = ui.get("form_mode", "") allow_complete = bool(ui.get("allow_complete", probe_type == "elapsed_time")) allow_manual_edit = bool(ui.get("allow_manual_edit", True)) html = f""" Life-NOC — {html_escape(title)}

    {html_escape(title)}

    État actuel : {html_escape(state)}

    Type de sonde : {html_escape(probe_type)}

    Dernier intrant

    Valeur : {html_escape(current_input.get("value", ""))}

    Capturé le : {html_escape(current_input.get("captured_at", ""))}

    Origine : {html_escape(current_input.get("origin", ""))}

    Contexte

    Domaine : {html_escape(domain)}

    Item key : {html_escape(item_key)}

    Notes : {html_escape(notes)}

    Liens utiles

    Action

    """ return HTMLResponse(content=html) @app.post("/life-noc/item/{domain}/{item_key}/set") def page_set_item(domain: str, item_key: str, value: str = Form(...), origin: str = Form(default="manual")) -> RedirectResponse: try: set_item(domain, item_key, value=value, origin=origin) except KeyError: raise HTTPException(status_code=404, detail="item introuvable") except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) return RedirectResponse( url=f"/life-noc/item/{domain}/{item_key}", status_code=303, ) @app.post("/life-noc/item/{domain}/{item_key}/complete") def page_complete_item(domain: str, item_key: str, origin: str = Form(default="manual")) -> RedirectResponse: try: complete_item(domain, item_key, origin=origin) except KeyError: raise HTTPException(status_code=404, detail="item introuvable") except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) return RedirectResponse( url=f"/life-noc/item/{domain}/{item_key}", status_code=303, )