#!/usr/bin/env python3
from __future__ import annotations
from pathlib import Path
from urllib.parse import quote
import os
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_domains import ( # noqa: E402
ALLOWED_FORM_MODES,
ALLOWED_METRIC_UNITS,
ALLOWED_ON_ERROR,
ALLOWED_PROBE_TYPES,
ALLOWED_SOURCE_TYPES,
THRESHOLD_FIELDS,
DomainsConflictError,
DomainsError,
delete_item,
get_domain_item,
item_to_form_values,
list_domain_items,
list_domains,
normalize_item_payload,
reorder_item,
upsert_item,
)
from lib.life_noc_inputs import ( # noqa: E402
list_items,
get_item,
set_item,
set_counter_pair,
complete_item,
get_defined_service,
compute_state_for_item,
)
class SetInputRequest(BaseModel):
value: str
origin: str = "manual"
class CompleteInputRequest(BaseModel):
origin: str = "manual"
class CounterPairInputRequest(BaseModel):
current_value: str
reference_value: str
origin: str = "manual"
app = FastAPI(title="Life-NOC Input API", version="1.1.0")
ADMIN_UI_ENABLED = os.environ.get("LIFE_NOC_ADMIN_UI_ENABLED", "0").strip().lower() in {
"1",
"true",
"yes",
"on",
}
BASE_CSS = """
:root {
color-scheme: light;
--bg: #f5f7fb;
--card: #ffffff;
--border: #d9e0ea;
--text: #112033;
--muted: #5a697b;
--accent: #164c96;
--accent-soft: #eaf2ff;
--danger: #8b1e1e;
--ok: #0f6b3d;
}
* { box-sizing: border-box; }
body {
margin: 0;
background: var(--bg);
color: var(--text);
font-family: Arial, sans-serif;
line-height: 1.45;
}
main {
max-width: 920px;
margin: 0 auto;
padding: 1rem;
}
.stack { display: grid; gap: 1rem; }
.card {
background: var(--card);
border: 1px solid var(--border);
border-radius: 16px;
padding: 1rem;
box-shadow: 0 1px 2px rgba(0,0,0,0.03);
}
.hero { padding: 1rem; }
.muted { color: var(--muted); }
.state { font-size: 1.15rem; font-weight: bold; }
h1, h2, h3 { margin-top: 0; }
ul { padding-left: 1.2rem; }
code {
background: #eef2f7;
padding: 0.15rem 0.35rem;
border-radius: 6px;
}
a { color: var(--accent); text-decoration: none; }
a:hover { text-decoration: underline; }
.toolbar, .actions {
display: flex;
flex-wrap: wrap;
gap: 0.6rem;
align-items: center;
}
.actions form, .inline-form { display: inline-flex; margin: 0; }
.pill {
display: inline-block;
padding: 0.25rem 0.55rem;
border-radius: 999px;
background: var(--accent-soft);
color: var(--accent);
font-size: 0.9rem;
}
.button, button {
display: inline-flex;
align-items: center;
justify-content: center;
gap: 0.35rem;
min-height: 44px;
padding: 0.75rem 1rem;
border-radius: 12px;
border: 1px solid #335b94;
background: #ffffff;
color: #163454;
font-size: 1rem;
cursor: pointer;
text-decoration: none;
}
.button.primary, button.primary {
background: var(--accent);
color: #fff;
border-color: var(--accent);
}
.button.danger, button.danger {
border-color: #b64b4b;
color: var(--danger);
}
.grid-cards {
display: grid;
gap: 0.9rem;
}
.row-between {
display: flex;
justify-content: space-between;
align-items: center;
gap: 1rem;
}
.item-title {
font-size: 1.05rem;
font-weight: bold;
margin: 0 0 0.35rem 0;
}
.form-grid {
display: grid;
gap: 0.85rem;
}
.form-grid.two { grid-template-columns: 1fr; }
label {
display: grid;
gap: 0.35rem;
font-size: 0.95rem;
font-weight: bold;
}
input[type="text"], input[type="date"], textarea, select {
width: 100%;
min-height: 44px;
padding: 0.75rem 0.8rem;
border-radius: 12px;
border: 1px solid #aebacc;
font: inherit;
background: #fff;
color: var(--text);
}
textarea { min-height: 100px; resize: vertical; }
.checks {
display: grid;
gap: 0.55rem;
}
.check {
display: flex;
align-items: center;
gap: 0.6rem;
font-weight: normal;
}
.check input { width: 20px; height: 20px; }
.section-title {
margin-bottom: 0.5rem;
font-size: 1.02rem;
}
details {
border: 1px solid var(--border);
border-radius: 14px;
background: #fcfdff;
padding: 0.15rem 0.85rem 0.85rem 0.85rem;
}
details + details { margin-top: 0.8rem; }
summary {
cursor: pointer;
font-weight: bold;
list-style: none;
padding: 0.9rem 0 0.6rem 0;
}
summary::-webkit-details-marker { display: none; }
.alert {
border: 1px solid #f2c5c5;
background: #fff4f4;
color: #7a1a1a;
border-radius: 12px;
padding: 0.85rem 1rem;
}
pre.yaml-preview {
white-space: pre-wrap;
word-break: break-word;
background: #0f1720;
color: #e8eef6;
padding: 1rem;
border-radius: 14px;
overflow-x: auto;
}
.mobile-sticky {
position: sticky;
bottom: 0;
background: rgba(245,247,251,0.96);
padding-top: 0.7rem;
}
@media (min-width: 760px) {
.form-grid.two { grid-template-columns: 1fr 1fr; }
.grid-cards { grid-template-columns: repeat(auto-fit, minmax(260px, 1fr)); }
}
"""
def html_escape(value: object) -> 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} {html_escape(title)}
{body}
"""
return HTMLResponse(content=html)
def require_admin_enabled() -> None:
if not ADMIN_UI_ENABLED:
raise HTTPException(status_code=404, detail="interface admin désactivée")
def bool_attr(value: bool) -> str:
return " checked" if value else ""
def option_tags(options: list[str], selected: str) -> str:
tags = []
for value in options:
label = value if value else "—"
sel = " selected" if value == selected else ""
tags.append(f'{html_escape(label)} ')
return "".join(tags)
def render_alert(message: str) -> str:
if not message:
return ""
return f'{html_escape(message)}
'
def render_domain_card(domain: str, items: list[dict]) -> str:
probed = sum(1 for item in items if isinstance(item, dict) and isinstance(item.get("probe"), dict))
return f"""
{html_escape(domain)}
{len(items)} item(s) · {probed} avec sonde réelle
Ouvrir
"""
def render_item_card(domain: str, item: dict, index: int, total: int) -> str:
name = str(item.get("name", "")).strip()
title = str(item.get("title") or name or "Item").strip()
summary = str(item.get("summary") or item.get("notes") or "").strip()
probe_type = str(item.get("probe", {}).get("type", "mock") if isinstance(item.get("probe"), dict) else "mock")
probe_badge = f'{html_escape(probe_type)} '
move_up = f"""
""" if total > 1 else ""
move_down = f"""
""" if total > 1 else ""
return f"""
{html_escape(title)}
{html_escape(name)}
{html_escape(summary)}
{probe_badge}
"""
def render_item_editor(domain: str, values: dict, action_url: str, heading: str, error_message: str = "") -> str:
thresholds_html = "".join(
f'''{html_escape(field)} '''
for field in THRESHOLD_FIELDS
)
yaml_preview = render_yaml_preview(values)
return f"""
{html_escape(heading)}
Éditeur mobile-first optionnel de domains.yaml. Sauvegarde explicite, sans changer le flux runtime existant.
{render_alert(error_message)}
"""
def render_yaml_preview(values: dict) -> str:
try:
payload = normalize_item_payload(values)
except Exception:
payload = {k: v for k, v in values.items() if k not in {"domain"}}
import yaml
return yaml.safe_dump(payload, sort_keys=False, allow_unicode=True, width=1000)
@app.get("/health")
def health() -> dict:
return {"status": "ok", "admin_ui_enabled": ADMIN_UI_ENABLED}
@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.post("/inputs/{domain}/{item_key}/counter")
def api_set_counter_pair(domain: str, item_key: str, req: CounterPairInputRequest) -> dict:
item = set_counter_pair(
domain,
item_key,
current_value=req.current_value,
reference_value=req.reference_value,
origin=req.origin or "manual",
)
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)
except KeyError:
raise HTTPException(status_code=404, detail="item introuvable")
try:
current_input = get_item(domain, item_key)
except KeyError:
current_input = {}
try:
state = compute_state_for_item(domain, item_key)
except (KeyError, ValueError):
state = "UNKNOWN"
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))
links_html = "".join(
fragment for fragment in [
link_row("Notes", notes_url),
link_row("Instructions", instructions_url),
link_row("Action", action_url),
] if fragment
) or "Aucun lien défini. "
summary_html = f'Résumé : {html_escape(summary)}
' if summary else ""
instructions_html = ""
if instructions:
instructions_html = f"""
Instructions
{html_escape(instructions)}
"""
manual_form_html = ""
if allow_manual_edit and form_mode in {"complete_date", "due_date"}:
form_title = "Mettre à jour la date" if form_mode == "complete_date" else "Définir l’échéance"
field_label = "Date" if form_mode == "complete_date" else "Date d’échéance"
form_hint = ""
if form_mode == "due_date":
form_hint = 'Saisir la date cible de cette échéance.
'
manual_form_html = f"""
{html_escape(form_title)}
{form_hint}
"""
elif allow_manual_edit and form_mode == "counter_pair":
manual_form_html = f"""
Mettre à jour les compteurs
Saisir le compteur courant et le compteur au dernier entretien.
"""
complete_form_html = ""
if allow_complete:
complete_form_html = f"""
"""
admin_link = ""
if ADMIN_UI_ENABLED:
admin_link = f'Admin '
body = f"""
{html_escape(title)}
État actuel : {html_escape(state)}
Type de sonde : {html_escape(probe_type)}
{summary_html}
Dernier intrant
Valeur : {html_escape(current_input.get('value', ''))}
{f'Valeur de référence : {html_escape(current_input.get("reference_value", ""))}
' if 'reference_value' in current_input else ''}
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)}
{instructions_html}
{manual_form_html}
{complete_form_html}
"""
return page_shell(f"Life-NOC — {title}", body)
@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}/counter")
def page_set_counter(domain: str, item_key: str, current_value: str = Form(...), reference_value: str = Form(...), origin: str = Form(default="manual")) -> RedirectResponse:
try:
set_counter_pair(domain, item_key, current_value=current_value, reference_value=reference_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)
@app.get("/life-noc/admin", response_class=HTMLResponse)
def admin_home() -> HTMLResponse:
require_admin_enabled()
domain_cards = []
for domain in list_domains():
items = list_domain_items(domain)
domain_cards.append(render_domain_card(domain, items))
body = f"""
Life-NOC Admin
Interface web optionnelle pour administrer domains.yaml sans casser le pipeline existant.
"""
return page_shell("Life-NOC Admin", body)
@app.get("/life-noc/admin/domain/{domain}", response_class=HTMLResponse)
def admin_domain(domain: str, error: str = "") -> HTMLResponse:
require_admin_enabled()
try:
items = list_domain_items(domain)
except KeyError:
raise HTTPException(status_code=404, detail="domaine introuvable")
cards = [render_item_card(domain, item, idx, len(items)) for idx, item in enumerate(items)]
body = f"""
Domaine : {html_escape(domain)}
{len(items)} item(s). Les boutons ↑ et ↓ réordonnent l’ordre des items dans domains.yaml.
{render_alert(error)}
"""
return page_shell(f"Life-NOC Admin — {domain}", body)
@app.post("/life-noc/admin/domain/{domain}/{item_key}/move")
def admin_move_item(domain: str, item_key: str, direction: str = Form(...)) -> RedirectResponse:
require_admin_enabled()
try:
reorder_item(domain, item_key, direction)
except (DomainsError, KeyError) as exc:
target = f"/life-noc/admin/domain/{domain}?error={quote(str(exc))}"
return RedirectResponse(url=target, status_code=303)
return RedirectResponse(url=f"/life-noc/admin/domain/{domain}", status_code=303)
@app.get("/life-noc/admin/item/{domain}/{item_key}", response_class=HTMLResponse)
def admin_item(domain: str, item_key: str) -> HTMLResponse:
require_admin_enabled()
try:
item = get_domain_item(domain, item_key)
except KeyError:
raise HTTPException(status_code=404, detail="item introuvable")
values = item_to_form_values(item, domain)
yaml_preview = render_yaml_preview(values)
body = f"""
{html_escape(values.get('title') or item_key)}
{html_escape(item_key)}
Résumé
{html_escape(values.get('summary') or values.get('notes') or '')}
probe.type = {html_escape(values.get('probe_type') or 'mock')}
Aperçu YAML
{html_escape(yaml_preview)}
"""
return page_shell(f"Life-NOC Admin — {item_key}", body)
@app.get("/life-noc/admin/domain/{domain}/new", response_class=HTMLResponse)
def admin_new_item_form(domain: str, error: str = "") -> HTMLResponse:
require_admin_enabled()
try:
list_domain_items(domain)
except KeyError:
raise HTTPException(status_code=404, detail="domaine introuvable")
values = item_to_form_values(None, domain)
body = render_item_editor(domain, values, f"/life-noc/admin/domain/{domain}/new", f"Créer un item dans {domain}", error)
return page_shell(f"Life-NOC Admin — nouveau item {domain}", body)
@app.post("/life-noc/admin/domain/{domain}/new")
def admin_new_item_submit(
domain: str,
name: str = Form(...),
date: str = Form(default=""),
title: str = Form(default=""),
summary: str = Form(default=""),
notes: str = Form(default=""),
instructions: str = Form(default=""),
notes_url: str = Form(default=""),
instructions_url: str = Form(default=""),
action_url: str = Form(default=""),
probe_type: str = Form(default=""),
probe_source_type: str = Form(default=""),
probe_inputs_file: str = Form(default=""),
probe_item_key: str = Form(default=""),
metric_unit: str = Form(default=""),
policy_on_error: str = Form(default=""),
unknown_lt: str = Form(default=""),
ok_gte: str = Form(default=""),
warning_gte: str = Form(default=""),
critical_gte: str = Form(default=""),
unknown_gte: str = Form(default=""),
ok_lt: str = Form(default=""),
warning_lt: str = Form(default=""),
critical_lt: str = Form(default=""),
ui_form_mode: str = Form(default=""),
ui_allow_complete: str = Form(default=""),
ui_allow_manual_edit: str = Form(default=""),
) -> RedirectResponse:
require_admin_enabled()
form_values = {
key: value
for key, value in locals().items()
if key not in {"domain"}
}
try:
item = normalize_item_payload(form_values)
upsert_item(domain, item, original_item_key=None)
except (DomainsError, DomainsConflictError, KeyError) as exc:
return RedirectResponse(url=f"/life-noc/admin/domain/{domain}/new?error={quote(str(exc))}", status_code=303)
return RedirectResponse(url=f"/life-noc/admin/item/{domain}/{item['name']}", status_code=303)
@app.get("/life-noc/admin/item/{domain}/{item_key}/edit", response_class=HTMLResponse)
def admin_edit_item_form(domain: str, item_key: str, error: str = "") -> HTMLResponse:
require_admin_enabled()
try:
item = get_domain_item(domain, item_key)
except KeyError:
raise HTTPException(status_code=404, detail="item introuvable")
values = item_to_form_values(item, domain)
body = render_item_editor(domain, values, f"/life-noc/admin/item/{domain}/{item_key}/edit", f"Modifier {item_key}", error)
return page_shell(f"Life-NOC Admin — modifier {item_key}", body)
@app.post("/life-noc/admin/item/{domain}/{item_key}/edit")
def admin_edit_item_submit(
domain: str,
item_key: str,
name: str = Form(...),
date: str = Form(default=""),
title: str = Form(default=""),
summary: str = Form(default=""),
notes: str = Form(default=""),
instructions: str = Form(default=""),
notes_url: str = Form(default=""),
instructions_url: str = Form(default=""),
action_url: str = Form(default=""),
probe_type: str = Form(default=""),
probe_source_type: str = Form(default=""),
probe_inputs_file: str = Form(default=""),
probe_item_key: str = Form(default=""),
metric_unit: str = Form(default=""),
policy_on_error: str = Form(default=""),
unknown_lt: str = Form(default=""),
ok_gte: str = Form(default=""),
warning_gte: str = Form(default=""),
critical_gte: str = Form(default=""),
unknown_gte: str = Form(default=""),
ok_lt: str = Form(default=""),
warning_lt: str = Form(default=""),
critical_lt: str = Form(default=""),
ui_form_mode: str = Form(default=""),
ui_allow_complete: str = Form(default=""),
ui_allow_manual_edit: str = Form(default=""),
) -> RedirectResponse:
require_admin_enabled()
try:
existing = get_domain_item(domain, item_key)
except KeyError:
raise HTTPException(status_code=404, detail="item introuvable")
form_values = {
key: value
for key, value in locals().items()
if key not in {"domain", "item_key", "existing"}
}
try:
item = normalize_item_payload(form_values, existing_item=existing)
upsert_item(domain, item, original_item_key=item_key)
except (DomainsError, DomainsConflictError, KeyError) as exc:
return RedirectResponse(url=f"/life-noc/admin/item/{domain}/{item_key}/edit?error={quote(str(exc))}", status_code=303)
return RedirectResponse(url=f"/life-noc/admin/item/{domain}/{item['name']}", status_code=303)
@app.post("/life-noc/admin/item/{domain}/{item_key}/delete")
def admin_delete_item(domain: str, item_key: str) -> RedirectResponse:
require_admin_enabled()
try:
delete_item(domain, item_key)
except (DomainsError, KeyError) as exc:
return RedirectResponse(url=f"/life-noc/admin/domain/{domain}?error={quote(str(exc))}", status_code=303)
return RedirectResponse(url=f"/life-noc/admin/domain/{domain}", status_code=303)