355 lines
12 KiB
Python
355 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
from copy import deepcopy
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
import os
|
|
import tempfile
|
|
import yaml
|
|
|
|
DOMAINS_FILE = Path(os.environ.get("LIFE_NOC_DOMAINS_FILE", "domains.yaml"))
|
|
|
|
ALLOWED_PROBE_TYPES = [
|
|
"",
|
|
"elapsed_time",
|
|
"days_until_due",
|
|
"elapsed_distance",
|
|
"current_value",
|
|
"remaining_quantity",
|
|
]
|
|
ALLOWED_SOURCE_TYPES = [
|
|
"",
|
|
"manual_date",
|
|
"manual_counter",
|
|
"manual_value",
|
|
"mqtt",
|
|
"file_json",
|
|
"file_text",
|
|
"csv",
|
|
"api",
|
|
"command",
|
|
"derived",
|
|
]
|
|
ALLOWED_METRIC_UNITS = [
|
|
"",
|
|
"days",
|
|
"km",
|
|
"units",
|
|
"%",
|
|
"volts",
|
|
"litres",
|
|
"watts",
|
|
]
|
|
ALLOWED_ON_ERROR = ["", "critical", "warning", "unknown"]
|
|
ALLOWED_FORM_MODES = [
|
|
"",
|
|
"complete_date",
|
|
"due_date",
|
|
"counter_pair",
|
|
"numeric_value",
|
|
"remaining_quantity",
|
|
]
|
|
THRESHOLD_FIELDS = [
|
|
"unknown_lt",
|
|
"ok_gte",
|
|
"warning_gte",
|
|
"critical_gte",
|
|
"unknown_gte",
|
|
"ok_lt",
|
|
"warning_lt",
|
|
"critical_lt",
|
|
]
|
|
|
|
|
|
class DomainsError(ValueError):
|
|
pass
|
|
|
|
|
|
class DomainsConflictError(DomainsError):
|
|
pass
|
|
|
|
|
|
def _now_suffix() -> str:
|
|
return datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')
|
|
|
|
|
|
def load_domains_document() -> dict[str, Any]:
|
|
if not DOMAINS_FILE.exists():
|
|
raise DomainsError(f'Fichier introuvable: {DOMAINS_FILE}')
|
|
data = yaml.safe_load(DOMAINS_FILE.read_text(encoding='utf-8'))
|
|
if not isinstance(data, dict):
|
|
raise DomainsError('domains.yaml invalide')
|
|
domains = data.get('domains')
|
|
if not isinstance(domains, dict):
|
|
raise DomainsError("domains.yaml invalide: clé 'domains' absente ou invalide")
|
|
return data
|
|
|
|
|
|
def save_domains_document(doc: dict[str, Any], create_backup: bool = True) -> Path:
|
|
if not isinstance(doc, dict) or not isinstance(doc.get('domains'), dict):
|
|
raise DomainsError("document domains invalide")
|
|
|
|
target = DOMAINS_FILE
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
rendered = yaml.safe_dump(doc, sort_keys=False, allow_unicode=True, width=1000)
|
|
|
|
if create_backup and target.exists():
|
|
backup = target.with_suffix(target.suffix + f'.bak.{_now_suffix()}')
|
|
backup.write_text(target.read_text(encoding='utf-8'), encoding='utf-8')
|
|
|
|
fd, tmp_name = tempfile.mkstemp(prefix=target.name + '.', dir=str(target.parent))
|
|
try:
|
|
with os.fdopen(fd, 'w', encoding='utf-8') as handle:
|
|
handle.write(rendered)
|
|
os.replace(tmp_name, target)
|
|
finally:
|
|
if os.path.exists(tmp_name):
|
|
os.unlink(tmp_name)
|
|
return target
|
|
|
|
|
|
def list_domains() -> list[str]:
|
|
doc = load_domains_document()
|
|
return list(doc['domains'].keys())
|
|
|
|
|
|
def list_domain_items(domain: str) -> list[dict[str, Any]]:
|
|
doc = load_domains_document()
|
|
items = doc['domains'].get(domain)
|
|
if not isinstance(items, list):
|
|
raise KeyError(domain)
|
|
return items
|
|
|
|
|
|
def get_domain_item(domain: str, item_key: str) -> dict[str, Any]:
|
|
items = list_domain_items(domain)
|
|
for item in items:
|
|
if isinstance(item, dict) and str(item.get('name', '')).strip() == item_key:
|
|
return deepcopy(item)
|
|
raise KeyError(item_key)
|
|
|
|
|
|
def _find_item_index(items: list[dict[str, Any]], item_key: str) -> int:
|
|
for index, item in enumerate(items):
|
|
if isinstance(item, dict) and str(item.get('name', '')).strip() == item_key:
|
|
return index
|
|
raise KeyError(item_key)
|
|
|
|
|
|
def _clean_value(value: Any) -> Any:
|
|
if isinstance(value, dict):
|
|
cleaned = {}
|
|
for key, subvalue in value.items():
|
|
candidate = _clean_value(subvalue)
|
|
if candidate in ('', None, {}, []):
|
|
continue
|
|
cleaned[key] = candidate
|
|
return cleaned
|
|
if isinstance(value, list):
|
|
cleaned_list = [_clean_value(v) for v in value]
|
|
return [v for v in cleaned_list if v not in ('', None, {}, [])]
|
|
return value
|
|
|
|
|
|
def _parse_number(value: Any) -> Any:
|
|
if value is None:
|
|
return None
|
|
text = str(value).strip()
|
|
if text == '':
|
|
return None
|
|
try:
|
|
if any(ch in text for ch in ('.', 'e', 'E')):
|
|
number = float(text)
|
|
return int(number) if number.is_integer() else number
|
|
return int(text)
|
|
except ValueError:
|
|
raise DomainsError(f'Valeur numérique invalide: {text}')
|
|
|
|
|
|
def _coerce_bool(value: Any) -> bool:
|
|
return str(value).strip().lower() in {'1', 'true', 'yes', 'on'}
|
|
|
|
|
|
def _validate_choice(label: str, value: str, allowed: list[str]) -> str:
|
|
if value not in allowed:
|
|
raise DomainsError(f'{label} invalide: {value}')
|
|
return value
|
|
|
|
|
|
def normalize_item_payload(form: dict[str, Any], existing_item: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
base = deepcopy(existing_item) if existing_item else {}
|
|
|
|
name = str(form.get('name', base.get('name', ''))).strip()
|
|
if not name:
|
|
raise DomainsError("Le champ 'name' est obligatoire.")
|
|
item: dict[str, Any] = {'name': name}
|
|
|
|
date_value = str(form.get('date', base.get('date', ''))).strip()
|
|
if date_value:
|
|
item['date'] = date_value
|
|
|
|
for field in ['title', 'summary', 'notes', 'instructions', 'notes_url', 'instructions_url', 'action_url']:
|
|
raw = form.get(field, base.get(field, ''))
|
|
value = str(raw).strip() if raw is not None else ''
|
|
if value:
|
|
item[field] = value
|
|
|
|
probe_type = str(form.get('probe_type', base.get('probe', {}).get('type', ''))).strip()
|
|
source_type = str(form.get('probe_source_type', base.get('probe', {}).get('source', {}).get('type', ''))).strip()
|
|
inputs_file = str(form.get('probe_inputs_file', base.get('probe', {}).get('source', {}).get('inputs_file', ''))).strip()
|
|
source_item_key = str(form.get('probe_item_key', base.get('probe', {}).get('source', {}).get('item_key', ''))).strip()
|
|
metric_unit = str(form.get('metric_unit', base.get('probe', {}).get('metric', {}).get('unit', ''))).strip()
|
|
on_error = str(form.get('policy_on_error', base.get('probe', {}).get('policy', {}).get('on_error', ''))).strip()
|
|
|
|
_validate_choice('probe.type', probe_type, ALLOWED_PROBE_TYPES)
|
|
_validate_choice('probe.source.type', source_type, ALLOWED_SOURCE_TYPES)
|
|
_validate_choice('metric.unit', metric_unit, ALLOWED_METRIC_UNITS)
|
|
_validate_choice('policy.on_error', on_error, ALLOWED_ON_ERROR)
|
|
|
|
thresholds: dict[str, Any] = {}
|
|
base_thresholds = base.get('probe', {}).get('thresholds', {}) if isinstance(base.get('probe'), dict) else {}
|
|
for field in THRESHOLD_FIELDS:
|
|
raw = form.get(field, base_thresholds.get(field, ''))
|
|
parsed = _parse_number(raw)
|
|
if parsed is not None:
|
|
thresholds[field] = parsed
|
|
|
|
if probe_type or source_type or inputs_file or source_item_key or metric_unit or thresholds or on_error:
|
|
probe: dict[str, Any] = {}
|
|
if probe_type:
|
|
probe['type'] = probe_type
|
|
source: dict[str, Any] = {}
|
|
if source_type:
|
|
source['type'] = source_type
|
|
if inputs_file:
|
|
source['inputs_file'] = inputs_file
|
|
if source_item_key:
|
|
source['item_key'] = source_item_key
|
|
if source:
|
|
probe['source'] = source
|
|
if metric_unit:
|
|
probe['metric'] = {'unit': metric_unit}
|
|
if thresholds:
|
|
probe['thresholds'] = thresholds
|
|
if on_error:
|
|
probe['policy'] = {'on_error': on_error}
|
|
item['probe'] = probe
|
|
|
|
form_mode = str(form.get('ui_form_mode', base.get('ui', {}).get('form_mode', ''))).strip()
|
|
_validate_choice('ui.form_mode', form_mode, ALLOWED_FORM_MODES)
|
|
allow_complete = _coerce_bool(form.get('ui_allow_complete', base.get('ui', {}).get('allow_complete', False)))
|
|
allow_manual_edit = _coerce_bool(form.get('ui_allow_manual_edit', base.get('ui', {}).get('allow_manual_edit', False)))
|
|
if form_mode or allow_complete or allow_manual_edit:
|
|
ui: dict[str, Any] = {}
|
|
if form_mode:
|
|
ui['form_mode'] = form_mode
|
|
if allow_complete:
|
|
ui['allow_complete'] = True
|
|
if allow_manual_edit:
|
|
ui['allow_manual_edit'] = True
|
|
item['ui'] = ui
|
|
|
|
normalized = _clean_value(item)
|
|
_validate_item_structure(normalized)
|
|
return normalized
|
|
|
|
|
|
def _validate_item_structure(item: dict[str, Any]) -> None:
|
|
if 'name' not in item or not str(item['name']).strip():
|
|
raise DomainsError("Le champ 'name' est obligatoire.")
|
|
probe = item.get('probe')
|
|
if probe is None:
|
|
return
|
|
if not isinstance(probe, dict):
|
|
raise DomainsError('probe doit être un objet')
|
|
if 'type' not in probe:
|
|
raise DomainsError("probe.type est requis dès qu'un probe est défini")
|
|
source = probe.get('source')
|
|
if source is not None and not isinstance(source, dict):
|
|
raise DomainsError('probe.source doit être un objet')
|
|
metric = probe.get('metric')
|
|
if metric is not None and not isinstance(metric, dict):
|
|
raise DomainsError('probe.metric doit être un objet')
|
|
thresholds = probe.get('thresholds')
|
|
if thresholds is not None and not isinstance(thresholds, dict):
|
|
raise DomainsError('probe.thresholds doit être un objet')
|
|
|
|
|
|
def upsert_item(domain: str, item: dict[str, Any], original_item_key: str | None = None) -> Path:
|
|
doc = load_domains_document()
|
|
items = doc['domains'].get(domain)
|
|
if not isinstance(items, list):
|
|
raise KeyError(domain)
|
|
|
|
existing_names = [str(entry.get('name', '')).strip() for entry in items if isinstance(entry, dict)]
|
|
new_name = str(item['name']).strip()
|
|
if new_name in existing_names and original_item_key != new_name:
|
|
raise DomainsConflictError(f"Un item nommé '{new_name}' existe déjà dans le domaine '{domain}'.")
|
|
|
|
if original_item_key:
|
|
index = _find_item_index(items, original_item_key)
|
|
items[index] = item
|
|
else:
|
|
items.append(item)
|
|
return save_domains_document(doc)
|
|
|
|
|
|
def delete_item(domain: str, item_key: str) -> Path:
|
|
doc = load_domains_document()
|
|
items = doc['domains'].get(domain)
|
|
if not isinstance(items, list):
|
|
raise KeyError(domain)
|
|
index = _find_item_index(items, item_key)
|
|
items.pop(index)
|
|
return save_domains_document(doc)
|
|
|
|
|
|
def reorder_item(domain: str, item_key: str, direction: str) -> Path:
|
|
if direction not in {'up', 'down'}:
|
|
raise DomainsError('direction invalide')
|
|
doc = load_domains_document()
|
|
items = doc['domains'].get(domain)
|
|
if not isinstance(items, list):
|
|
raise KeyError(domain)
|
|
index = _find_item_index(items, item_key)
|
|
if direction == 'up' and index > 0:
|
|
items[index - 1], items[index] = items[index], items[index - 1]
|
|
elif direction == 'down' and index < len(items) - 1:
|
|
items[index + 1], items[index] = items[index], items[index + 1]
|
|
return save_domains_document(doc)
|
|
|
|
|
|
def item_to_form_values(item: dict[str, Any] | None, domain: str = '') -> dict[str, Any]:
|
|
item = deepcopy(item) if item else {}
|
|
probe = item.get('probe', {}) if isinstance(item.get('probe'), dict) else {}
|
|
source = probe.get('source', {}) if isinstance(probe.get('source'), dict) else {}
|
|
metric = probe.get('metric', {}) if isinstance(probe.get('metric'), dict) else {}
|
|
thresholds = probe.get('thresholds', {}) if isinstance(probe.get('thresholds'), dict) else {}
|
|
policy = probe.get('policy', {}) if isinstance(probe.get('policy'), dict) else {}
|
|
ui = item.get('ui', {}) if isinstance(item.get('ui'), dict) else {}
|
|
|
|
values = {
|
|
'domain': domain,
|
|
'name': item.get('name', ''),
|
|
'date': item.get('date', ''),
|
|
'title': item.get('title', ''),
|
|
'summary': item.get('summary', ''),
|
|
'notes': item.get('notes', ''),
|
|
'instructions': item.get('instructions', ''),
|
|
'notes_url': item.get('notes_url', ''),
|
|
'instructions_url': item.get('instructions_url', ''),
|
|
'action_url': item.get('action_url', ''),
|
|
'probe_type': probe.get('type', ''),
|
|
'probe_source_type': source.get('type', ''),
|
|
'probe_inputs_file': source.get('inputs_file', ''),
|
|
'probe_item_key': source.get('item_key', ''),
|
|
'metric_unit': metric.get('unit', ''),
|
|
'policy_on_error': policy.get('on_error', ''),
|
|
'ui_form_mode': ui.get('form_mode', ''),
|
|
'ui_allow_complete': bool(ui.get('allow_complete', False)),
|
|
'ui_allow_manual_edit': bool(ui.get('allow_manual_edit', False)),
|
|
}
|
|
for field in THRESHOLD_FIELDS:
|
|
values[field] = thresholds.get(field, '')
|
|
return values
|