123 lines
3.9 KiB
Python
123 lines
3.9 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""archivo.py — Archivado diario por fuente y resumen (RF-5, RF-7, RF-9, RF-10, RF-18).
|
|
|
|
Escribe un fichero Markdown por fuente y dia (`estado/monitor/<ruta>/<YYYY-MM-DD>.md`).
|
|
En cada ciclo el fichero del dia se **reescribe** con el estado actual completo, de modo
|
|
que siempre contiene todo lo de hoy y nada duplicado. Los ficheros de mas de
|
|
`retencion_dias` se eliminan. Tambien persiste el ultimo resumen.
|
|
|
|
El formateo de cada elemento se delega en `render_fn`, para no acoplar este modulo a
|
|
la forma concreta de cada fuente.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from datetime import date, datetime, timedelta
|
|
from pathlib import Path
|
|
from typing import Callable, Iterable
|
|
|
|
from . import config
|
|
|
|
MONITOR_DIR: Path = config.MONITOR_DIR
|
|
|
|
_PATRON_FECHA = re.compile(r"^(\d{4}-\d{2}-\d{2})\.md$")
|
|
|
|
|
|
def _fecha_str(fecha) -> str:
|
|
"""Normaliza una fecha (date o texto ISO) a 'YYYY-MM-DD'."""
|
|
if isinstance(fecha, datetime):
|
|
fecha = fecha.date()
|
|
if isinstance(fecha, date):
|
|
return fecha.isoformat()
|
|
return str(fecha)
|
|
|
|
|
|
def escribe_dia(
|
|
ruta_logica: str,
|
|
fecha,
|
|
elementos: Iterable,
|
|
id_fn: Callable = lambda e: e,
|
|
render_fn: Callable = str,
|
|
titulo: str | None = None,
|
|
) -> Path:
|
|
"""Reescribe el fichero del dia de una fuente, sin duplicados por id.
|
|
|
|
`ruta_logica` es la ruta relativa bajo el directorio del monitor (p. ej.
|
|
`telegram/oscar_rodriguez`, `correo`, `jira`).
|
|
"""
|
|
fecha_str = _fecha_str(fecha)
|
|
vistos: set[str] = set()
|
|
bloques: list[str] = []
|
|
for e in elementos:
|
|
clave = str(id_fn(e))
|
|
if clave in vistos:
|
|
continue
|
|
vistos.add(clave)
|
|
bloques.append(render_fn(e))
|
|
|
|
titulo = titulo or f"{ruta_logica} — {fecha_str}"
|
|
cuerpo = "\n\n".join(bloques)
|
|
contenido = f"# {titulo}\n\n{cuerpo}\n" if cuerpo else f"# {titulo}\n"
|
|
|
|
destino = MONITOR_DIR / ruta_logica / f"{fecha_str}.md"
|
|
destino.parent.mkdir(parents=True, exist_ok=True)
|
|
destino.write_text(contenido, encoding="utf-8")
|
|
return destino
|
|
|
|
|
|
def poda(retencion_dias: int = 30, hoy: date | None = None) -> list[Path]:
|
|
"""Elimina los ficheros diarios con fecha anterior a `hoy - retencion_dias`.
|
|
|
|
Devuelve la lista de ficheros borrados.
|
|
"""
|
|
hoy = hoy or date.today()
|
|
limite = hoy - timedelta(days=retencion_dias)
|
|
borrados: list[Path] = []
|
|
if not MONITOR_DIR.exists():
|
|
return borrados
|
|
for f in MONITOR_DIR.rglob("*.md"):
|
|
m = _PATRON_FECHA.match(f.name)
|
|
if not m:
|
|
continue
|
|
try:
|
|
fd = date.fromisoformat(m.group(1))
|
|
except ValueError:
|
|
continue
|
|
if fd < limite:
|
|
try:
|
|
f.unlink()
|
|
borrados.append(f)
|
|
except OSError:
|
|
pass
|
|
return borrados
|
|
|
|
|
|
def escribe_resumen(resumen: dict) -> None:
|
|
"""Persiste el ultimo resumen en `resumen_actual.md` y `resumen_actual.json`."""
|
|
MONITOR_DIR.mkdir(parents=True, exist_ok=True)
|
|
md = resumen.get("resumen_md", "") or ""
|
|
(MONITOR_DIR / "resumen_actual.md").write_text(md, encoding="utf-8")
|
|
with open(MONITOR_DIR / "resumen_actual.json", "w", encoding="utf-8") as f:
|
|
json.dump(resumen, f, indent=2, ensure_ascii=False)
|
|
|
|
|
|
def escribe_ultimo_ciclo(data: dict) -> None:
|
|
"""Persiste la marca del ultimo ciclo en `ultimo_ciclo.json`."""
|
|
MONITOR_DIR.mkdir(parents=True, exist_ok=True)
|
|
with open(MONITOR_DIR / "ultimo_ciclo.json", "w", encoding="utf-8") as f:
|
|
json.dump(data, f, indent=2, ensure_ascii=False)
|
|
|
|
|
|
def leer_ultimo_ciclo() -> dict:
|
|
"""Lee `ultimo_ciclo.json` (o {} si no existe)."""
|
|
ruta = MONITOR_DIR / "ultimo_ciclo.json"
|
|
if ruta.exists():
|
|
try:
|
|
with open(ruta, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except Exception:
|
|
pass
|
|
return {}
|