#!/usr/bin/env python3
"""¿Sigue funcionando la máquina? Se pregunta después de cada actualización.

Por qué existe: filmos se actualiza por `apt` desde un repositorio propio que se
publica varias veces al día. Una actualización que rompe algo no avisa —dpkg
termina contento y el fallo aparece tres horas después, en mitad de un montaje—,
y para entonces nadie recuerda qué se instaló.

Esto mira lo que tiene que estar en pie y, si algo no lo está, **dice con qué
orden exacta se deshace la última actualización**. Una comprobación que detecta
un problema y no dice qué hacer con él solo sirve para preocupar.

No hace falta root: todo lo que mira se lee sin permisos.
"""

from __future__ import annotations

import json
import os
import re
import shutil
import socket
import subprocess
import sys
from pathlib import Path

INSTANTANEAS = Path("/.snapshots")
PARTE_DE_LOS_DISCOS = Path("/var/lib/filmos/discos.json")
BOTON = Path("/usr/bin/filmos-agente-boton")
PARTE_DE_LAS_GPU = Path("/run/filmos/gpu.json")

#: Lo que tiene que estar vivo. Si una unidad no está instalada no se echa en
#: falta: una estación sin escritorio no tiene botón, y no por eso está rota.
DEL_SISTEMA = {
    "filmos-hermes.service": "HermesOS",
}
DE_LA_SESION = {
    "filmos-vault.service": "el vault",
    "filmos-hermesd.service": "el puente del agente",
}

VERDE = "\033[32m"
ROJO = "\033[31m"
AMARILLO = "\033[33m"
APAGADO = "\033[2m"
FIN = "\033[0m"


def color(texto: str, tono: str) -> str:
    return f"{tono}{texto}{FIN}" if sys.stdout.isatty() else texto


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=30, check=False
        )
    except (OSError, subprocess.SubprocessError):
        return 127, ""
    return hecho.returncode, hecho.stdout


class Informe:
    def __init__(self) -> None:
        self.problemas: list[str] = []
        self.lineas: list[str] = []

    def bien(self, texto: str) -> None:
        self.lineas.append(f"  {color('·', APAGADO)} {texto}")

    def mal(self, texto: str, detalle: str = "") -> None:
        self.problemas.append(texto)
        self.lineas.append(f"  {color('✗', ROJO)} {texto}")
        for fila in detalle.strip().splitlines()[:6]:
            self.lineas.append(f"      {color(fila.strip(), APAGADO)}")

    def aviso(self, texto: str) -> None:
        self.lineas.append(f"  {color('!', AMARILLO)} {texto}")


def mirar_dpkg(informe: Informe) -> None:
    """Paquetes a medias. Pasó de verdad: doce desempaquetados sin configurar."""

    codigo, salida = correr("dpkg", "--audit")
    if codigo == 127:
        return
    if salida.strip():
        informe.mal("hay paquetes a medio instalar (`sudo dpkg --configure -a`)", salida)
    else:
        informe.bien("ningún paquete a medias")


def mirar_unidades(informe: Informe) -> None:
    _, salida = correr("systemctl", "--failed", "--no-legend", "--plain")
    rotas = [l for l in salida.splitlines() if l.strip()]
    if rotas:
        informe.mal(f"{len(rotas)} unidad(es) del sistema en fallo", salida)
    else:
        informe.bien("ninguna unidad del sistema en fallo")

    codigo, salida = correr("systemctl", "--user", "--failed", "--no-legend", "--plain")
    if codigo not in (0, 1):
        return
    rotas = [l for l in salida.splitlines() if l.strip()]
    if rotas:
        informe.mal(f"{len(rotas)} unidad(es) de tu sesión en fallo", salida)
    else:
        informe.bien("ninguna unidad de tu sesión en fallo")


def instalada(unidad: str, *de_la_sesion: str) -> bool:
    """¿Existe esa unidad en esta máquina?

    `systemctl is-active` contesta «inactive» tanto para un servicio parado
    como para uno que NO ESTÁ INSTALADO. Sin distinguirlos, una estación sin
    escritorio —o sin HermesOS— saldría con fallos que no existen, y una
    comprobación que grita sin motivo se deja de mirar.
    """

    _, salida = correr("systemctl", *de_la_sesion, "show", "-p", "LoadState", "--value", unidad)
    return salida.strip() not in ("not-found", "")


def mirar_servicios(informe: Informe) -> None:
    for unidad, quien in DEL_SISTEMA.items():
        if not instalada(unidad):
            continue
        _, salida = correr("systemctl", "is-active", unidad)
        estado = salida.strip()
        if estado == "active":
            informe.bien(f"{quien} en marcha")
        elif estado in ("inactive", "failed"):
            informe.mal(f"{quien} no está en marcha ({unidad}: {estado})")

    for unidad, quien in DE_LA_SESION.items():
        if not instalada(unidad, "--user"):
            continue
        _, salida = correr("systemctl", "--user", "is-active", unidad)
        estado = salida.strip()
        if estado == "active":
            informe.bien(f"{quien} en marcha")
        elif estado in ("inactive", "failed"):
            informe.mal(f"{quien} no está en marcha ({unidad}: {estado})")


def mirar_boton(informe: Informe) -> None:
    """¿Está el botón del agente? Si no, `Meta+A` no abre nada.

    Solo se pregunta con el escritorio en marcha: por ssh no hay botón que
    valga, y echarlo en falta ahí sería gritar sin motivo.

    Se comprueba conectando a su socket y no mirando procesos: el socket es lo
    que usa la tecla, así que si el socket no contesta, la tecla tampoco.
    """

    if not BOTON.is_file():
        return
    _, salida = correr("systemctl", "--user", "is-active", "plasma-plasmashell.service")
    if salida.strip() != "active":
        return

    runtime = os.environ.get("XDG_RUNTIME_DIR")
    sitio = Path(runtime) / "filmos-agente-boton.sock" if runtime else None
    if sitio is not None and sitio.exists():
        cliente = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        cliente.settimeout(2)
        try:
            cliente.connect(str(sitio))
            informe.bien("el botón del agente, en su sitio")
            return
        except OSError:
            pass
        finally:
            cliente.close()
    informe.mal("el botón del agente no está: Meta+A no abriría nada")


def mirar_arranque(informe: Informe) -> None:
    """Que la máquina pueda volver a arrancar. Es lo único irrecuperable."""

    boot = Path("/boot")
    if not boot.is_dir():
        return
    nucleos = sorted(boot.glob("vmlinuz-*"))
    if not nucleos:
        informe.mal("no hay ningún núcleo en /boot: esta máquina no volvería a arrancar")
        return
    sueltos = [
        n.name
        for n in nucleos
        if not (boot / n.name.replace("vmlinuz-", "initrd.img-", 1)).exists()
    ]
    if sueltos:
        informe.mal(
            f"{len(sueltos)} núcleo(s) sin su initrd (`sudo update-initramfs -u -k all`)",
            "\n".join(sueltos),
        )
    else:
        informe.bien(f"{len(nucleos)} núcleo(s) en /boot, cada uno con su initrd")

    mirar_la_salida_de_emergencia(informe)


#: El trozo del menú de arranque que ofrece volver a una instantánea. Lo escribe
#: `/etc/grub.d/41_filmos_instantaneas` cada vez que se regenera el menú.
MENU = Path("/boot/grub/grub.cfg")
GENERADOR = Path("/etc/grub.d/41_filmos_instantaneas")


def mirar_la_salida_de_emergencia(informe: Informe, menu: Path = MENU,
                                  generador: Path = GENERADOR) -> None:
    """Que las instantáneas sigan estando en el menú de arranque.

    Es la única salida cuando la máquina ya no enciende, y es exactamente el
    tipo de cosa que desaparece sin hacer ruido: basta con que el generador
    pierda el permiso de ejecución para que `update-grub` lo salte en silencio.
    Nadie lo notaría hasta el día en que hiciera falta, que es el peor día para
    enterarse.
    """

    if not generador.exists():
        return
    if not os.access(generador, os.X_OK):
        informe.mal(
            "las instantáneas ya no salen en el menú de arranque: el generador "
            "no es ejecutable (`sudo chmod +x /etc/grub.d/41_filmos_instantaneas "
            "&& sudo update-grub`)"
        )
        return
    try:
        texto = menu.read_text(encoding="utf-8", errors="replace")
    except OSError:
        # /boot/grub/grub.cfg suele ser de root. No poder leerlo es lo normal
        # corriendo como una persona, y no es una noticia.
        return
    if "volver a una instantánea" in texto:
        informe.bien("las instantáneas están en el menú de arranque")
    else:
        informe.aviso(
            "las instantáneas aún no están en el menú de arranque (`sudo update-grub`)"
        )


#: Sistemas de ficheros que guardan cosas. Lo demás -tmpfs, overlay, proc- no
#: se mira: llenar un tmpfs no es un problema de disco.
DE_VERDAD = {"btrfs", "ext2", "ext3", "ext4", "xfs", "zfs", "vfat", "ntfs3", "fuseblk", "exfat"}

#: Por debajo de esto es una tarjeta, un pendrive o una partición de servicio, y
#: avisar de que están llenos es ruido.
MINIMO_GIGAS = 20


def montados() -> list[tuple[str, bool]]:
    """Dónde hay algo montado que merezca mirarse, y si es de solo lectura."""

    try:
        crudo = Path("/proc/self/mounts").read_text(encoding="utf-8", errors="replace")
    except OSError:
        return [("/", False)]
    sitios: list[tuple[str, bool]] = []
    vistos: set[str] = set()
    for linea in crudo.splitlines():
        partes = linea.split()
        if len(partes) < 4 or partes[2] not in DE_VERDAD:
            continue
        # El punto de montaje viene con los espacios escapados en octal.
        punto = partes[1].replace("\\040", " ")
        if punto in vistos:
            continue
        vistos.add(punto)
        sitios.append((punto, "ro" in partes[3].split(",")))
    return sitios or [("/", False)]


def mirar_espacio(informe: Informe) -> None:
    """Todos los discos que guardan algo, no solo el del sistema.

    Los de material son los que se llenan: una raíz de 900 GB aguanta meses y
    un disco de masters se llena en un rodaje. Mirar solo `/` es mirar donde no
    pasa nada.
    """

    for punto, solo_lectura in montados():
        try:
            uso = shutil.disk_usage(punto)
        except OSError:
            continue
        if uso.total < MINIMO_GIGAS * 1024**3:
            continue
        libre = uso.free / uso.total * 100 if uso.total else 100
        gigas = uso.free / 1024**3
        if libre < 3 and not solo_lectura:
            informe.mal(f"{punto} casi lleno: quedan {gigas:.1f} GB ({libre:.0f} %)")
        elif libre < 10:
            # En uno de solo lectura no se puede hacer sitio, pero saber que
            # está lleno es lo que decide si cabe el próximo rodaje.
            coletilla = " (solo lectura)" if solo_lectura else ""
            informe.aviso(
                f"{punto} apretado: quedan {gigas:.1f} GB ({libre:.0f} %){coletilla}"
            )
        elif punto in ("/", "/boot"):
            informe.bien(f"{punto} con {gigas:.1f} GB libres")


def mirar_discos(informe: Informe, parte: Path = PARTE_DE_LOS_DISCOS) -> None:
    """Lo que el vigilante de discos dejó apuntado.

    Se lee el parte en vez de preguntar al disco porque `smartctl` necesita root
    y esto no: quien usa la máquina no debería necesitar permisos para saber que
    uno de sus discos se está muriendo.
    """

    if not parte.is_file():
        return
    try:
        datos = json.loads(parte.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return

    malos = [d for d in datos.get("discos", []) if d.get("veredicto") == "mal"]
    vigilar = [d for d in datos.get("discos", []) if d.get("veredicto") == "vigilar"]
    if malos:
        informe.mal(
            f"{len(malos)} disco(s) fallando (`sudo filmos-discos`)",
            "\n".join(f"/dev/{d['nombre']}: {'; '.join(d.get('motivos', []))}" for d in malos),
        )
    elif vigilar:
        informe.aviso(f"{len(vigilar)} disco(s) a vigilar (`sudo filmos-discos`)")
    else:
        informe.bien(f"{len(datos.get('discos', []))} discos, todos bien")


def mirar_gpus(informe: Informe, parte: Path = PARTE_DE_LAS_GPU) -> None:
    """Lo que el gestor de GPU dejó escrito. Sin ejecutar nada.

    Leer un fichero cuesta microsegundos; preguntarle a `nvidia-smi` cuesta
    cientos de milisegundos y abre el dispositivo. Esto corre después de cada
    `apt`, así que lee.
    """

    if not parte.is_file():
        return
    try:
        datos = json.loads(parte.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return

    gpus = datos.get("gpus", [])
    if not gpus:
        return
    creativa = next((g for g in gpus if "creative" in (g.get("role") or [])), None)
    if creativa is None:
        informe.mal("ninguna tarjeta lleva el escritorio (`filmos-gpu doctor`)")
    else:
        nombre = creativa.get("model") or creativa.get("pci")
        informe.bien(f"{nombre} lleva la pantalla")

    if datos.get("salud") == "mal":
        calientes = [
            g.get("model") or g.get("pci")
            for g in gpus
            if (g.get("measures") or {}).get("temperatura") is not None
        ]
        informe.mal(
            "alguna GPU está por encima de su límite (`filmos-gpu temperatures`)",
            ", ".join(str(c) for c in calientes),
        )
    elif datos.get("salud") == "vigilar":
        informe.aviso("alguna GPU está caliente (`filmos-gpu temperatures`)")


def ultima_de_apt(donde: Path = INSTANTANEAS) -> tuple[str, str] | None:
    """La instantánea de antes de la última actualización, si la hay."""

    if not donde.is_dir():
        return None
    mejor: tuple[int, str] | None = None
    try:
        carpetas = list(donde.iterdir())
    except OSError:
        # `/.snapshots` es de root. Esto corre como la persona que usa la
        # máquina, así que lo normal es NO poder leerlo: se dice la orden sin el
        # número en vez de reventar justo cuando algo ya está roto.
        return None
    for carpeta in carpetas:
        if not carpeta.name.isdigit():
            continue
        info = carpeta / "info.xml"
        if not info.is_file():
            continue
        try:
            texto = info.read_text(encoding="utf-8", errors="replace")
        except OSError:
            continue
        if "<type>pre</type>" not in texto:
            continue
        numero = int(carpeta.name)
        if mejor is None or numero > mejor[0]:
            fecha = re.search(r"<date>(.*?)</date>", texto)
            mejor = (numero, fecha.group(1) if fecha else "")
    return (str(mejor[0]), mejor[1]) if mejor else None


def me_dicen_la_instantanea(argumentos: list[str]) -> tuple[str, str] | None:
    """La instantánea que nos pasan hecha, para no tener que leerla.

    El enganche de apt corre como root y SÍ puede leer `/.snapshots`; la
    comprobación corre como la persona, que no. Así el número exacto llega
    igualmente en vez de quedarse en «mira la lista».
    """

    if "--instantanea" not in argumentos:
        return None
    donde = argumentos.index("--instantanea") + 1
    if donde >= len(argumentos):
        return None
    numero, _, fecha = argumentos[donde].partition("|")
    return (numero, fecha) if numero.isdigit() else None


def main(argumentos: list[str]) -> int:
    # En modo callado solo se habla cuando hay algo que decir: un mensaje
    # después de cada `apt` que siempre dice «todo bien» se deja de leer, y
    # entonces tampoco se lee el día que dice otra cosa.
    callado = "--callado" in argumentos

    if "--decir-instantanea" in argumentos:
        anterior = ultima_de_apt()
        if anterior:
            print(f"{anterior[0]}|{anterior[1]}")
            return 0
        return 1

    informe = Informe()
    mirar_dpkg(informe)
    mirar_unidades(informe)
    mirar_servicios(informe)
    mirar_boton(informe)
    mirar_arranque(informe)
    mirar_espacio(informe)
    mirar_discos(informe)
    mirar_gpus(informe)

    if not informe.problemas:
        if not callado:
            for linea in informe.lineas:
                print(linea)
            print()
            print(color("Todo en pie.", VERDE))
        return 0

    print()
    for linea in informe.lineas:
        print(linea)
    print()
    print(color(f"{len(informe.problemas)} cosa(s) rotas.", ROJO))

    print()
    print("Si esto lo ha roto la última actualización, se deshace con:")
    anterior = me_dicen_la_instantanea(argumentos) or ultima_de_apt()
    if anterior:
        numero, fecha = anterior
        cuando = f"   {color(f'(de antes de {fecha})', APAGADO)}" if fecha else ""
        print(color(f"    sudo filmos-volver-atras {numero}", AMARILLO) + cuando)
    else:
        # Sin poder leer las instantáneas -lo normal sin root- se manda a la
        # orden, que las lista. Peor sería callarse.
        print(color("    sudo filmos-volver-atras", AMARILLO) + "   (lista a dónde volver)")
    print("y reiniciar. La raíz de ahora se guarda, no se pierde.")
    return 1


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