20-sep: infra global - OmniRoute watchdog + plugin OpenCode + changelog sobremesa/portatil
This commit is contained in:
juanminsanz
256
scripts/infra_changelog.py
Normal file
256
scripts/infra_changelog.py
Normal file
@@ -0,0 +1,256 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Changelog global de infraestructura (sobremesa <-> portatil <-> VPS).
|
||||
|
||||
Fichero unico para saber que cambios de infraestructura se han hecho en cada equipo
|
||||
y cuales hay que replicar en el otro. Fuente de verdad: `changelog_infra.json`
|
||||
(en este repo). Se genera `CHANGELOG_INFRA.md` para lectura humana.
|
||||
|
||||
Uso:
|
||||
python scripts/infra_changelog.py add --titulo "..." --categoria omniroute \
|
||||
--transferible si --detalle "..." [--equipo sobremesa] [--id slug]
|
||||
python scripts/infra_changelog.py pendientes [--json]
|
||||
python scripts/infra_changelog.py aplicado <id> [--equipo sobremesa]
|
||||
python scripts/infra_changelog.py listar [--equipo X]
|
||||
python scripts/infra_changelog.py estado
|
||||
python scripts/infra_changelog.py render
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import platform
|
||||
import re
|
||||
import socket
|
||||
import subprocess
|
||||
import sys
|
||||
from datetime import date
|
||||
from pathlib import Path
|
||||
|
||||
REPO = Path(__file__).resolve().parent.parent
|
||||
JSON_FILE = REPO / "changelog_infra.json"
|
||||
MD_FILE = REPO / "CHANGELOG_INFRA.md"
|
||||
|
||||
CATEGORIAS = ["omniroute", "opencode", "red", "servicios", "drivers", "memoria", "otros"]
|
||||
|
||||
|
||||
# --- Deteccion de equipo ---------------------------------------------------
|
||||
|
||||
def _powershell(query: str) -> str:
|
||||
"""Ejecuta una consulta CIM via PowerShell y devuelve la salida."""
|
||||
try:
|
||||
out = subprocess.run(
|
||||
["powershell", "-NoProfile", "-NonInteractive", "-Command", query],
|
||||
capture_output=True, text=True, timeout=15,
|
||||
).stdout
|
||||
return out.strip()
|
||||
except Exception:
|
||||
return ""
|
||||
|
||||
|
||||
def _huella_hardware() -> str:
|
||||
"""CPU + modelo del equipo (Windows). wmic ya no existe en Windows 11 nuevo."""
|
||||
cpu = _powershell("(Get-CimInstance Win32_Processor).Name")
|
||||
modelo = _powershell("(Get-CimInstance Win32_ComputerSystem).Model")
|
||||
if not cpu and not modelo:
|
||||
cpu = platform.processor() or ""
|
||||
return f"{cpu} | {modelo}".lower()
|
||||
|
||||
|
||||
def detectar_equipo() -> str:
|
||||
"""Devuelve 'sobremesa', 'portatil' o el hostname si no se reconoce."""
|
||||
huella = _huella_hardware()
|
||||
if "7900x" in huella or "b650" in huella:
|
||||
return "sobremesa"
|
||||
if "255hx" in huella or "vector" in huella or "ultra 7" in huella:
|
||||
return "portatil"
|
||||
return socket.gethostname().lower()
|
||||
|
||||
|
||||
# --- Persistencia ----------------------------------------------------------
|
||||
|
||||
def cargar() -> dict:
|
||||
if JSON_FILE.exists():
|
||||
return json.loads(JSON_FILE.read_text(encoding="utf-8"))
|
||||
return {"version": 1, "entries": []}
|
||||
|
||||
|
||||
def guardar(data: dict) -> None:
|
||||
JSON_FILE.write_text(
|
||||
json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
|
||||
)
|
||||
render(data)
|
||||
|
||||
|
||||
def _slug(texto: str) -> str:
|
||||
s = re.sub(r"[^a-z0-9]+", "-", texto.lower()).strip("-")
|
||||
return s[:60] or "cambio"
|
||||
|
||||
|
||||
# --- Render markdown -------------------------------------------------------
|
||||
|
||||
def render(data: dict | None = None) -> None:
|
||||
data = data or cargar()
|
||||
entries = data.get("entries", [])
|
||||
lineas: list[str] = []
|
||||
lineas.append("# Changelog de infraestructura — global (sobremesa <-> portatil <-> VPS)")
|
||||
lineas.append("")
|
||||
lineas.append("> Fichero generado por `scripts/infra_changelog.py`. NO editar a mano.")
|
||||
lineas.append("> Fuente de verdad: `changelog_infra.json`.")
|
||||
lineas.append("")
|
||||
lineas.append("## Pendientes de replicar")
|
||||
lineas.append("")
|
||||
pend = [e for e in entries if e.get("transferible") and len(e.get("aplicado_en", [])) < 2]
|
||||
if pend:
|
||||
for e in pend:
|
||||
equipos = "|".join(e.get("aplicado_en", [])) or "-"
|
||||
lineas.append(f"- **{e['id']}** [{e.get('categoria','otros')}] {e['titulo']} "
|
||||
f"(aplicado en: {equipos})")
|
||||
else:
|
||||
lineas.append("_(nada pendiente)_")
|
||||
lineas.append("")
|
||||
lineas.append("## Historial")
|
||||
lineas.append("")
|
||||
for e in sorted(entries, key=lambda x: x.get("fecha", ""), reverse=True):
|
||||
lineas.append(f"### {e.get('fecha','?')} · {e.get('titulo','')}")
|
||||
lineas.append(f"- **id**: `{e['id']}`")
|
||||
lineas.append(f"- **Equipo**: {e.get('equipo','?')}")
|
||||
lineas.append(f"- **Categoria**: {e.get('categoria','otros')}")
|
||||
lineas.append(f"- **Transferible**: {'si' if e.get('transferible') else 'no'}")
|
||||
lineas.append(f"- **Aplicado en**: {', '.join(e.get('aplicado_en', [])) or '-'}")
|
||||
if e.get("detalle"):
|
||||
lineas.append(f"- **Detalle**: {e['detalle']}")
|
||||
lineas.append("")
|
||||
MD_FILE.write_text("\n".join(lineas), encoding="utf-8")
|
||||
|
||||
|
||||
# --- Comandos --------------------------------------------------------------
|
||||
|
||||
def cmd_add(args: argparse.Namespace) -> int:
|
||||
data = cargar()
|
||||
equipo = args.equipo or detectar_equipo()
|
||||
entrada_id = args.id or f"{date.today().isoformat()}-{_slug(args.titulo)}"
|
||||
if any(e["id"] == entrada_id for e in data["entries"]):
|
||||
print(f"Ya existe una entrada con id '{entrada_id}'.", file=sys.stderr)
|
||||
return 1
|
||||
data["entries"].append({
|
||||
"id": entrada_id,
|
||||
"fecha": args.fecha or date.today().isoformat(),
|
||||
"equipo": equipo,
|
||||
"categoria": args.categoria,
|
||||
"titulo": args.titulo,
|
||||
"transferible": args.transferible == "si",
|
||||
"aplicado_en": [equipo],
|
||||
"detalle": args.detalle or "",
|
||||
})
|
||||
guardar(data)
|
||||
print(f"Anadido '{entrada_id}' (equipo={equipo}, transferible={args.transferible}).")
|
||||
return 0
|
||||
|
||||
|
||||
def _cargar_para_lectura() -> dict:
|
||||
return cargar()
|
||||
|
||||
|
||||
def cmd_pendientes(args: argparse.Namespace) -> int:
|
||||
data = _cargar_para_lectura()
|
||||
equipo = args.equipo or detectar_equipo()
|
||||
pend = [
|
||||
e for e in data.get("entries", [])
|
||||
if e.get("transferible") and equipo not in e.get("aplicado_en", [])
|
||||
]
|
||||
if args.json:
|
||||
print(json.dumps(pend, ensure_ascii=False, indent=2))
|
||||
return 0
|
||||
if not pend:
|
||||
print(f"Sin pendientes de replicar en '{equipo}'.")
|
||||
return 0
|
||||
print(f"Pendientes de replicar en '{equipo}':")
|
||||
for e in pend:
|
||||
print(f" - {e['id']} [{e.get('categoria')}] {e['titulo']} "
|
||||
f"(origen: {e.get('equipo')}, aplicado en: {', '.join(e.get('aplicado_en', []))})")
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_aplicado(args: argparse.Namespace) -> int:
|
||||
data = cargar()
|
||||
equipo = args.equipo or detectar_equipo()
|
||||
for e in data.get("entries", []):
|
||||
if e["id"] == args.id:
|
||||
if equipo not in e.get("aplicado_en", []):
|
||||
e.setdefault("aplicado_en", []).append(equipo)
|
||||
guardar(data)
|
||||
print(f"Marcado '{args.id}' como aplicado en '{equipo}'.")
|
||||
else:
|
||||
print(f"'{args.id}' ya estaba aplicado en '{equipo}'.")
|
||||
return 0
|
||||
print(f"No existe la entrada '{args.id}'.", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
|
||||
def cmd_listar(args: argparse.Namespace) -> int:
|
||||
data = cargar()
|
||||
entries = data.get("entries", [])
|
||||
if args.equipo:
|
||||
entries = [e for e in entries if e.get("equipo") == args.equipo]
|
||||
for e in sorted(entries, key=lambda x: x.get("fecha", ""), reverse=True):
|
||||
print(f"{e.get('fecha','?')} [{e.get('categoria','otros')}] {e['titulo']} "
|
||||
f"(equipo={e.get('equipo')}, transf={'si' if e.get('transferible') else 'no'}, "
|
||||
f"aplicado en: {', '.join(e.get('aplicado_en', [])) or '-'})")
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_estado(args: argparse.Namespace) -> int:
|
||||
equipo = detectar_equipo()
|
||||
data = cargar()
|
||||
total = len(data.get("entries", []))
|
||||
pend = [
|
||||
e for e in data.get("entries", [])
|
||||
if e.get("transferible") and equipo not in e.get("aplicado_en", [])
|
||||
]
|
||||
print(f"Equipo actual: {equipo}")
|
||||
print(f"Entradas totales: {total}")
|
||||
print(f"Pendientes de replicar aqui: {len(pend)}")
|
||||
return 0
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
p = argparse.ArgumentParser(description="Changelog global de infraestructura.")
|
||||
sub = p.add_subparsers(dest="cmd", required=True)
|
||||
|
||||
a = sub.add_parser("add", help="Anadir una entrada")
|
||||
a.add_argument("--titulo", required=True)
|
||||
a.add_argument("--categoria", default="otros", choices=CATEGORIAS)
|
||||
a.add_argument("--transferible", default="si", choices=["si", "no"])
|
||||
a.add_argument("--detalle", default="")
|
||||
a.add_argument("--equipo", default=None)
|
||||
a.add_argument("--fecha", default=None)
|
||||
a.add_argument("--id", default=None)
|
||||
a.set_defaults(func=cmd_add)
|
||||
|
||||
pe = sub.add_parser("pendientes", help="Ver pendientes de replicar en este equipo")
|
||||
pe.add_argument("--equipo", default=None)
|
||||
pe.add_argument("--json", action="store_true")
|
||||
pe.set_defaults(func=cmd_pendientes)
|
||||
|
||||
ap = sub.add_parser("aplicado", help="Marcar una entrada como aplicada en este equipo")
|
||||
ap.add_argument("id")
|
||||
ap.add_argument("--equipo", default=None)
|
||||
ap.set_defaults(func=cmd_aplicado)
|
||||
|
||||
li = sub.add_parser("listar", help="Listar entradas")
|
||||
li.add_argument("--equipo", default=None)
|
||||
li.set_defaults(func=cmd_listar)
|
||||
|
||||
es = sub.add_parser("estado", help="Resumen")
|
||||
es.set_defaults(func=cmd_estado)
|
||||
|
||||
re_ = sub.add_parser("render", help="Regenerar CHANGELOG_INFRA.md")
|
||||
re_.set_defaults(func=lambda a: (render(), 0)[1])
|
||||
|
||||
args = p.parse_args(argv)
|
||||
return args.func(args)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user