118 lines
4.4 KiB
Python
118 lines
4.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""config.py — Carga y valida la configuracion del monitor de estado.
|
|
|
|
La configuracion vive en `INFRAESTRUCTURA/estado/monitor/config.json` (runtime, no
|
|
versionada). Si no existe, se crea a partir de la plantilla `config.example.json`
|
|
que va junto a este modulo, de modo que el monitor arranca siempre con valores validos.
|
|
|
|
Las 5 fuentes de Telegram y sus identificadores se resolvieron desde la sesion real
|
|
de Telegram (ver plan 005).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# El modulo vive en INFRAESTRUCTURA/scripts/monitor_estado/; su padre es scripts/,
|
|
# donde esta infra_paths.py.
|
|
_SCRIPTS_DIR = Path(__file__).resolve().parent.parent
|
|
if str(_SCRIPTS_DIR) not in sys.path:
|
|
sys.path.insert(0, str(_SCRIPTS_DIR))
|
|
|
|
import infra_paths # noqa: E402
|
|
|
|
# Directorio raiz de datos runtime del monitor (gitignored).
|
|
MONITOR_DIR = infra_paths.ESTADO_DIR / "monitor"
|
|
CONFIG_PATH = MONITOR_DIR / "config.json"
|
|
EXAMPLE_PATH = Path(__file__).resolve().parent / "config.example.json"
|
|
|
|
|
|
def default_config() -> dict:
|
|
"""Configuracion por defecto (equivale a la plantilla de ejemplo)."""
|
|
return {
|
|
"intervalo_minutos": 30,
|
|
"retencion_dias": 30,
|
|
"telegram": {
|
|
# Sesion Telethon DEDICADA (no reutiliza la del watcher, que esta bloqueada).
|
|
"sesion": "~/.telegram-agent/sessions/monitor",
|
|
"telefono": "+34657866417",
|
|
"fuentes": [
|
|
{"nombre": "oscar_rodriguez", "id": 129020043},
|
|
{"nombre": "isa", "id": 16182959},
|
|
{"nombre": "daniel_garcia", "id": 8249301783},
|
|
{"nombre": "antonio_romero", "id": 880864397},
|
|
{"nombre": "desarrollo_software", "id": -1002693936231},
|
|
],
|
|
},
|
|
"correo": {"cuenta": "jminguez@prolongo.es", "horas": 24},
|
|
"jira": {"proyecto": "DS"},
|
|
"ia": {"base_url": "http://localhost:20128/v1", "modelo": "antigravity/gemini-2.5-flash-lite"},
|
|
"aviso": {"voz": True},
|
|
}
|
|
|
|
|
|
def _merge(base: dict, override: dict) -> dict:
|
|
"""Mezcla recursiva de diccionarios (override gana; no borra claves base)."""
|
|
result = copy.deepcopy(base)
|
|
for clave, valor in override.items():
|
|
if isinstance(valor, dict) and isinstance(result.get(clave), dict):
|
|
result[clave] = _merge(result[clave], valor)
|
|
else:
|
|
result[clave] = valor
|
|
return result
|
|
|
|
|
|
def _escribir_plantilla(destino: Path) -> None:
|
|
"""Escribe la config de ejemplo en `destino` (creando el directorio)."""
|
|
destino.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(destino, "w", encoding="utf-8") as f:
|
|
json.dump(default_config(), f, indent=2, ensure_ascii=False)
|
|
|
|
|
|
def cargar(crear_si_falta: bool = True) -> dict:
|
|
"""Carga la configuracion, creando `config.json` desde la plantilla si no existe.
|
|
|
|
Devuelve siempre un dict completo (las claves que falten se toman por defecto).
|
|
"""
|
|
cfg = default_config()
|
|
if not CONFIG_PATH.exists():
|
|
if not crear_si_falta:
|
|
return cfg
|
|
_escribir_plantilla(CONFIG_PATH)
|
|
try:
|
|
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
|
|
usuario = json.load(f)
|
|
if isinstance(usuario, dict):
|
|
cfg = _merge(cfg, usuario)
|
|
except Exception as e: # config corrupta -> seguimos con los valores por defecto
|
|
print(f"[monitor_estado] Aviso: no se pudo leer {CONFIG_PATH}: {e}", file=sys.stderr)
|
|
return cfg
|
|
|
|
|
|
def fuentes_telegram(cfg: dict | None = None) -> list[dict]:
|
|
"""Devuelve la lista de fuentes de Telegram configuradas."""
|
|
cfg = cfg or cargar()
|
|
return list(cfg.get("telegram", {}).get("fuentes", []))
|
|
|
|
|
|
def resumen_config(cfg: dict | None = None) -> str:
|
|
"""Texto legible con las fuentes configuradas (para el CLI/estado)."""
|
|
cfg = cfg or cargar()
|
|
lineas = [
|
|
f"Intervalo : {cfg.get('intervalo_minutos')} min",
|
|
f"Retencion : {cfg.get('retencion_dias')} dias",
|
|
f"Correo : {cfg.get('correo', {}).get('cuenta')} (ultimas {cfg.get('correo', {}).get('horas')} h)",
|
|
f"Jira : proyecto {cfg.get('jira', {}).get('proyecto')}",
|
|
"Telegram:",
|
|
]
|
|
for f in fuentes_telegram(cfg):
|
|
lineas.append(f" - {f.get('nombre')} (id {f.get('id')})")
|
|
return "\n".join(lineas)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print(resumen_config())
|