70 lines
1.8 KiB
Python
70 lines
1.8 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
import yaml
|
|
|
|
INPUTS_DIR = Path("data/inputs")
|
|
|
|
|
|
def utc_now_iso() -> str:
|
|
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
|
|
|
|
def utc_today_date() -> str:
|
|
return datetime.now(timezone.utc).strftime("%Y-%m-%d")
|
|
|
|
|
|
def domain_file(domain: str) -> Path:
|
|
return INPUTS_DIR / f"{domain}.yaml"
|
|
|
|
|
|
def load_domain(domain: str) -> dict:
|
|
path = domain_file(domain)
|
|
if not path.exists():
|
|
return {}
|
|
data = yaml.safe_load(path.read_text(encoding="utf-8"))
|
|
if data is None:
|
|
return {}
|
|
if not isinstance(data, dict):
|
|
raise ValueError(f"Fichier invalide: {path}")
|
|
return data
|
|
|
|
|
|
def save_domain(domain: str, data: dict) -> Path:
|
|
path = domain_file(domain)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(
|
|
yaml.safe_dump(data, sort_keys=False, allow_unicode=True),
|
|
encoding="utf-8",
|
|
)
|
|
return path
|
|
|
|
|
|
def list_items(domain: str) -> list[str]:
|
|
return list(load_domain(domain).keys())
|
|
|
|
|
|
def get_item(domain: str, item_key: str) -> dict:
|
|
data = load_domain(domain)
|
|
if item_key not in data:
|
|
raise KeyError(item_key)
|
|
item = data[item_key]
|
|
if not isinstance(item, dict):
|
|
raise ValueError(f"Entrée invalide pour {item_key}")
|
|
return item
|
|
|
|
|
|
def set_item(domain: str, item_key: str, value: str, origin: str = "manual") -> dict:
|
|
data = load_domain(domain)
|
|
data[item_key] = {
|
|
"value": value,
|
|
"captured_at": utc_now_iso(),
|
|
"origin": origin,
|
|
}
|
|
save_domain(domain, data)
|
|
return data[item_key]
|
|
|
|
|
|
def complete_item(domain: str, item_key: str, origin: str = "manual") -> dict:
|
|
return set_item(domain, item_key, utc_today_date(), origin=origin)
|