227 lines
6.9 KiB
Python
227 lines
6.9 KiB
Python
#!/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'<li><a href="{u}" target="_blank" rel="noopener">{l}</a></li>'
|
|
|
|
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"""<!DOCTYPE html>
|
|
<html lang="fr">
|
|
<head>
|
|
<meta charset="utf-8">
|
|
<meta name="viewport" content="width=device-width, initial-scale=1">
|
|
<title>Life-NOC — {html_escape(title)}</title>
|
|
<style>
|
|
body {{
|
|
font-family: Arial, sans-serif;
|
|
max-width: 760px;
|
|
margin: 0 auto;
|
|
padding: 1rem;
|
|
line-height: 1.45;
|
|
}}
|
|
.card {{
|
|
border: 1px solid #ccc;
|
|
border-radius: 10px;
|
|
padding: 1rem;
|
|
margin-bottom: 1rem;
|
|
}}
|
|
.state {{
|
|
font-size: 1.2rem;
|
|
font-weight: bold;
|
|
}}
|
|
button {{
|
|
font-size: 1rem;
|
|
padding: 0.9rem 1.2rem;
|
|
border-radius: 8px;
|
|
border: 1px solid #444;
|
|
cursor: pointer;
|
|
}}
|
|
ul {{
|
|
padding-left: 1.25rem;
|
|
}}
|
|
code {{
|
|
background: #f3f3f3;
|
|
padding: 0.1rem 0.3rem;
|
|
border-radius: 4px;
|
|
}}
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<main>
|
|
<section class="card">
|
|
<h1>{html_escape(title)}</h1>
|
|
<p class="state">État actuel : {html_escape(state)}</p>
|
|
<p><strong>Type de sonde :</strong> {html_escape(probe_type)}</p>
|
|
</section>
|
|
|
|
<section class="card">
|
|
<h2>Dernier intrant</h2>
|
|
<p><strong>Valeur :</strong> {html_escape(current_input.get("value", ""))}</p>
|
|
<p><strong>Capturé le :</strong> {html_escape(current_input.get("captured_at", ""))}</p>
|
|
<p><strong>Origine :</strong> {html_escape(current_input.get("origin", ""))}</p>
|
|
</section>
|
|
|
|
<section class="card">
|
|
<h2>Contexte</h2>
|
|
<p><strong>Domaine :</strong> {html_escape(domain)}</p>
|
|
<p><strong>Item key :</strong> <code>{html_escape(item_key)}</code></p>
|
|
<p><strong>Notes :</strong> {html_escape(notes)}</p>
|
|
</section>
|
|
|
|
<section class="card">
|
|
<h2>Liens utiles</h2>
|
|
<ul>
|
|
{link_row("Notes", notes_url)}
|
|
{link_row("Instructions", instructions_url)}
|
|
{link_row("Action", action_url)}
|
|
</ul>
|
|
</section>
|
|
|
|
<section class="card">
|
|
<h2>Action</h2>
|
|
<form method="post" action="/life-noc/item/{html_escape(domain)}/{html_escape(item_key)}/complete">
|
|
<button type="submit">Compléter</button>
|
|
</form>
|
|
</section>
|
|
</main>
|
|
</body>
|
|
</html>"""
|
|
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,
|
|
)
|