#!/usr/bin/env python3
"""Vigila los discos donde vive el material, antes de que uno se muera.

En esta máquina los masters viven en discos mecánicos. Un disco no suele morirse
de golpe: avisa durante semanas —sectores que no puede leer, reasignaciones que
crecen, temperatura alta— y nadie está mirando. El trabajo de un sistema
operativo de cine no es explicar la pérdida después: es avisar antes.

Qué se mira y por qué
---------------------
- **Sectores pendientes** (197). Es LA alarma: el disco no ha podido leer un
  sector y todavía no lo ha reasignado. Es material que quizá ya no puede
  devolver. Uno solo basta para sacar ese disco de «único sitio donde está algo».
- **Sectores reasignados** (5). No es morirse: es que ya ha empezado a gastar
  repuesto. Se vigila su crecimiento.
- **Errores no corregibles** (187, 198).
- **Temperatura.** Por encima de 55 °C un disco mecánico dura bastante menos.
- En los SSD NVMe: el **aviso crítico**, el **desgaste** y el **repuesto**
  disponible, que es lo que se agota antes de que empiece a perder celdas.

Escribe el parte en /var/lib/filmos/discos.json para que lo pueda leer quien no
es root —`filmos-comprobar` y el botón del agente—, porque `smartctl` sí lo
necesita y ellos no deberían.
"""

from __future__ import annotations

import contextlib
import json
import os
import shutil
import subprocess
import sys
import time
from pathlib import Path

PARTE = Path("/var/lib/filmos/discos.json")

BIEN, VIGILAR, MAL = "bien", "vigilar", "mal"
GRAVEDAD = {BIEN: 0, VIGILAR: 1, MAL: 2}

#: Un disco mecánico por encima de esto dura bastante menos.
TEMPERATURA_ALTA = 55
TEMPERATURA_GRAVE = 60

#: Cinco años encendido. No es un fallo y no se cuenta como tal: es el momento
#: de dejar de tener material en un solo sitio sobre ese disco. Los fabricantes
#: dan garantías de tres a cinco años para uso continuo; a partir de ahí un
#: disco sano sigue funcionando, y también sigue siendo un disco de ocho años.
HORAS_VIEJO = 43_800


def correr(*orden: str) -> tuple[int, str]:
    try:
        hecho = subprocess.run(  # noqa: S603 - órdenes fijas, sin shell
            list(orden), capture_output=True, text=True, timeout=120, check=False
        )
    except (OSError, subprocess.SubprocessError):
        return 127, ""
    return hecho.returncode, hecho.stdout


def discos() -> list[dict[str, str]]:
    """Los discos físicos, con lo que tienen montado encima."""

    codigo, salida = correr(
        "lsblk", "-J", "-o", "NAME,TYPE,SIZE,MODEL,MOUNTPOINTS", "--tree"
    )
    if codigo != 0:
        return []
    try:
        arbol = json.loads(salida)
    except ValueError:
        return []

    encontrados = []
    for nodo in arbol.get("blockdevices", []):
        if nodo.get("type") != "disk":
            continue
        encontrados.append(
            {
                "nombre": nodo.get("name", ""),
                "tamano": nodo.get("size", ""),
                "modelo": (nodo.get("model") or "").strip(),
                "donde": ", ".join(sorted(puntos_de_montaje(nodo))) or "",
            }
        )
    return encontrados


def puntos_de_montaje(nodo: dict) -> set[str]:
    puntos = {p for p in (nodo.get("mountpoints") or []) if p}
    for hija in nodo.get("children", []) or []:
        puntos |= puntos_de_montaje(hija)
    return puntos


def leer_smart(nombre: str) -> dict:
    codigo, salida = correr("smartctl", "-j", "-H", "-A", "-i", f"/dev/{nombre}")
    if not salida.strip():
        return {}
    try:
        return json.loads(salida)
    except ValueError:
        return {}


def atributos_ata(datos: dict) -> dict[int, int]:
    tabla = (datos.get("ata_smart_attributes") or {}).get("table") or []
    valores = {}
    for atributo in tabla:
        try:
            valores[int(atributo["id"])] = int(atributo["raw"]["value"])
        except (KeyError, TypeError, ValueError):
            continue
    return valores


def juzgar(datos: dict) -> tuple[str, list[str], dict]:
    """Un veredicto y por qué. El «por qué» es lo que permite decidir."""

    if not datos:
        return VIGILAR, ["no he podido leer su SMART"], {}

    motivos: list[str] = []
    veredicto = BIEN
    medidas: dict = {}

    def peor(nuevo: str) -> None:
        nonlocal veredicto
        if GRAVEDAD[nuevo] > GRAVEDAD[veredicto]:
            veredicto = nuevo

    salud = (datos.get("smart_status") or {}).get("passed")
    if salud is False:
        peor(MAL)
        motivos.append("el propio disco se declara en fallo")

    temperatura = (datos.get("temperature") or {}).get("current")
    nvme = datos.get("nvme_smart_health_information_log") or {}
    if temperatura is None:
        temperatura = nvme.get("temperature")
    if isinstance(temperatura, int):
        medidas["temperatura"] = temperatura
        if temperatura >= TEMPERATURA_GRAVE:
            peor(MAL)
            motivos.append(f"{temperatura} °C: demasiado caliente")
        elif temperatura >= TEMPERATURA_ALTA:
            peor(VIGILAR)
            motivos.append(f"{temperatura} °C: caliente")

    horas = (datos.get("power_on_time") or {}).get("hours") or nvme.get("power_on_hours")
    if isinstance(horas, int):
        medidas["horas"] = horas
        # La edad solo se juzga en los mecánicos: en un SSD lo que envejece es
        # lo escrito, no el tiempo, y para eso está el desgaste.
        gira = (datos.get("rotation_rate") or 0) > 0
        if gira and horas >= HORAS_VIEJO:
            peor(VIGILAR)
            anos = horas / 24 / 365
            motivos.append(f"{anos:.1f} años encendido: no dejes nada solo aquí")

    if nvme:
        aviso = nvme.get("critical_warning")
        if isinstance(aviso, int) and aviso != 0:
            peor(MAL)
            motivos.append("aviso crítico del propio SSD")
        desgaste = nvme.get("percentage_used")
        if isinstance(desgaste, int):
            medidas["desgaste"] = desgaste
            if desgaste >= 90:
                peor(MAL)
                motivos.append(f"{desgaste} % de su vida escrita")
            elif desgaste >= 80:
                peor(VIGILAR)
                motivos.append(f"{desgaste} % de su vida escrita")
        repuesto = nvme.get("available_spare")
        umbral = nvme.get("available_spare_threshold")
        if isinstance(repuesto, int) and isinstance(umbral, int):
            medidas["repuesto"] = repuesto
            if repuesto <= umbral:
                peor(MAL)
                motivos.append(f"repuesto agotado ({repuesto} % ≤ {umbral} %)")
        errores = nvme.get("media_errors")
        if isinstance(errores, int) and errores > 0:
            peor(MAL)
            medidas["errores"] = errores
            motivos.append(f"{errores} error(es) de medio")
        return veredicto, motivos, medidas

    valores = atributos_ata(datos)
    pendientes = valores.get(197, 0)
    if pendientes:
        # LA alarma: no ha podido leer y aún no ha reasignado. Es material que
        # quizá ya no puede devolver.
        peor(MAL)
        medidas["pendientes"] = pendientes
        motivos.append(f"{pendientes} sector(es) que no ha podido leer")
    reasignados = valores.get(5, 0)
    if reasignados:
        peor(VIGILAR)
        medidas["reasignados"] = reasignados
        motivos.append(f"{reasignados} sector(es) ya reasignados")
    for numero, texto in ((187, "errores no corregidos"), (198, "sectores ilegibles fuera de línea")):
        cuantos = valores.get(numero, 0)
        if cuantos:
            peor(VIGILAR if numero == 187 else MAL)
            medidas[texto] = cuantos
            motivos.append(f"{cuantos} {texto}")
    return veredicto, motivos, medidas


def revisar() -> dict:
    partes = []
    for disco in discos():
        datos = leer_smart(disco["nombre"])
        veredicto, motivos, medidas = juzgar(datos)
        partes.append(
            {
                **disco,
                "modelo": disco["modelo"] or (datos.get("model_name") or ""),
                "veredicto": veredicto,
                "motivos": motivos,
                "medidas": medidas,
            }
        )
    peor = BIEN
    for parte in partes:
        if GRAVEDAD[parte["veredicto"]] > GRAVEDAD[peor]:
            peor = parte["veredicto"]
    return {"cuando": int(time.time()), "veredicto": peor, "discos": partes}


def guardar(parte: dict) -> None:
    PARTE.parent.mkdir(parents=True, exist_ok=True)
    provisional = PARTE.with_suffix(".nuevo")
    provisional.write_text(json.dumps(parte, ensure_ascii=False, indent=2), encoding="utf-8")
    # Lo lee quien no es root: `filmos-comprobar` y el botón del agente. Ellos
    # no deberían necesitar permisos para saber si un disco se está muriendo.
    os.chmod(provisional, 0o644)
    os.replace(provisional, PARTE)


SIGNO = {BIEN: "·", VIGILAR: "!", MAL: "✗"}


def contar(parte: dict) -> int:
    print()
    for disco in parte["discos"]:
        donde = f"  →  {disco['donde']}" if disco["donde"] else ""
        print(f"  {SIGNO[disco['veredicto']]} /dev/{disco['nombre']}  {disco['tamano']:>7}"
              f"  {disco['modelo']}{donde}")
        for motivo in disco["motivos"]:
            print(f"        {motivo}")
        medidas = disco["medidas"]
        if medidas.get("temperatura") is not None and not disco["motivos"]:
            resumen = f"{medidas['temperatura']} °C"
            if medidas.get("horas"):
                resumen += f" · {medidas['horas']} horas encendido"
            if medidas.get("desgaste") is not None:
                resumen += f" · {medidas['desgaste']} % de vida escrita"
            print(f"        {resumen}")
    print()
    if parte["veredicto"] == BIEN:
        print("Todos los discos bien.")
        return 0
    if parte["veredicto"] == VIGILAR:
        print("Hay algo que vigilar. Ninguno para cambiar hoy, pero tampoco para")
        print("dejar material en un solo sitio.")
        return 0
    print("Un disco está fallando. Saca de ahí lo que solo esté ahí.")
    return 1


def sesion_del_escritorio() -> tuple[str, str] | None:
    """Quién tiene sesión abierta, para poder hablarle.

    Esto corre como root desde un temporizador, y root no tiene escritorio: un
    aviso enviado desde aquí no lo ve nadie. Hay que cruzar a la sesión de la
    persona, igual que hace el enganche de apt.
    """

    for orden, columna in ((("loginctl", "list-sessions", "--no-legend"), 2),
                           (("loginctl", "list-users", "--no-legend"), 1)):
        codigo, salida = correr(*orden)
        if codigo != 0:
            continue
        for fila in salida.splitlines():
            partes = fila.split()
            if len(partes) <= columna:
                continue
            usuario = partes[columna]
            hecho = subprocess.run(  # noqa: S603 - orden fija, sin shell
                ["id", "-u", usuario], capture_output=True, text=True, check=False
            )
            uid = hecho.stdout.strip()
            if hecho.returncode == 0 and uid.isdigit() and int(uid) >= 1000:
                if Path(f"/run/user/{uid}").is_dir():
                    return usuario, uid
    return None


def avisar_al_escritorio(parte: dict) -> None:
    """Un disco muriéndose merece interrumpir, no esperar a que alguien mire."""

    if parte["veredicto"] == BIEN or shutil.which("notify-send") is None:
        return
    malos = [d for d in parte["discos"] if d["veredicto"] != BIEN]
    cuerpo = "\n".join(
        f"/dev/{d['nombre']} {d['modelo']}: {'; '.join(d['motivos'])}" for d in malos[:3]
    )
    titulo = (
        "Un disco está fallando" if parte["veredicto"] == MAL else "Un disco a vigilar"
    )
    aviso = [
        "notify-send",
        "--app-name=filmos",
        "--icon=drive-harddisk",
        "--expire-time=0",
        f"--urgency={'critical' if parte['veredicto'] == MAL else 'normal'}",
        titulo,
        f"{cuerpo}\n\nMira `sudo filmos-discos` para el detalle.",
    ]

    quien = sesion_del_escritorio()
    if quien is not None and os.geteuid() == 0:
        usuario, uid = quien
        aviso = [
            "runuser", "-u", usuario, "--",
            "env",
            f"XDG_RUNTIME_DIR=/run/user/{uid}",
            f"DBUS_SESSION_BUS_ADDRESS=unix:path=/run/user/{uid}/bus",
            *aviso,
        ]
    with contextlib.suppress(OSError, subprocess.SubprocessError):
        subprocess.run(aviso, capture_output=True, timeout=15, check=False)  # noqa: S603


def main(argumentos: list[str]) -> int:
    if shutil.which("smartctl") is None:
        print("falta smartctl (paquete smartmontools)", file=sys.stderr)
        return 1
    if os.geteuid() != 0:
        print("leer el SMART de un disco necesita root: usa sudo.", file=sys.stderr)
        return 1

    parte = revisar()
    if not parte["discos"]:
        print("no he encontrado ningún disco", file=sys.stderr)
        return 1
    guardar(parte)

    if "--callado" in argumentos:
        # Lo llama un temporizador: se habla en el escritorio, no en un registro
        # que nadie lee.
        avisar_al_escritorio(parte)
        return 1 if parte["veredicto"] == MAL else 0
    return contar(parte)


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))
