sergio/takana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
6f5cc2629a35d5fe14abbc76da04ad6926d8f9d1
takana/scripts/mudanza/planear.py
T
SergioandClaude Opus 5 6f5cc2629a mudanza: de dónde sale cada binario — y la herramienta corrigió un error mío 
«Instalá el paquete» es lo que uno ya sabía. La respuesta útil sale de cruzar dos hechos: quién posee
el fichero en el ORIGEN (el censo se lo pregunta al gestor de paquetes de esa máquina, en un lote) y
si el corpus de takana tiene una receta con ese nombre.

  receta-takana  → instalarlo del repo firmado, NO copiar el binario
  paquete-ajeno  → escribir receta, o qorpa (ADR 0015)
  suelto         → nadie lo provee: llevarlo con su entorno o escribirle receta

Medido sobre gioser (37 servicios vivos): 4 receta-takana · 4 paquete-ajeno · **11 SUELTOS** · 7 sin
binario legible. Los sueltos viven en `/usr/local/bin` o en un home (`qdrant`, `matilda`,
`pacha-secretos`, `act_runner`, `shuma-gateway`) y son los que se pierden al apagar el origen.
Sorpresa medida: `/usr/bin/caddy` tampoco tiene dueño — está puesto a mano.

**Y corrigió un error mío**: en el SDD 28 escribí que `gitea` no tenía receta y que «es la que más
peso tiene». Las dos cosas falsas — `recipes/gitea.toml` existe y está sellada. Lo afirmé de memoria;
el programa fue a mirar. Corregido allá con la nota.

**Y un fallo del clasificador, que vale como regla**: calculaba la raíz del repo con un `dirname` de
menos, no encontraba ninguna receta y contestaba `suelto` A TODO, incluido `caddy`. Un clasificador
que contesta siempre lo mismo no clasifica. Ahora falla ruidosamente si no encuentra el catálogo, en
vez de dar una respuesta falsa con forma de respuesta.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RomoxEGZUhaT4pob1QSX5x
2026-09-11 17:02:52 +00:00

461 lines
26 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env python3
"""planear.py — convierte un CENSO decidido en un PLAN ejecutable y exportable (SDD 29 etapa 2).
El plan es el producto: un documento que se revisa, se versiona, se lleva a otro proveedor y se
vuelve a correr. **Cada paso lleva su comando literal**, así que el plan se puede ejecutar con la
herramienta o a mano, línea por línea, sin la herramienta y sin este repo.
── LAS TRES REGLAS, Y POR QUÉ ──────────────────────────────────────────────────────────────────
1. **Nada sin decidir se ejecuta.** Una entrada con `decision = ""` ABORTA el plan, nombrándola. El
silencio no es consentimiento: si nadie dijo qué hacer con un servicio, la respuesta no es
«entonces no lo muevo» —eso lo mata en silencio— ni «entonces lo muevo» —eso muda fósiles—.
2. **Lo que muere se dice por su nombre y con su tamaño, ANTES de borrar nada.** En esta mudanza
`terapeuta.ec` eran 279 M sin DNS: la diferencia entre «lo decidimos» y «se nos pasó» es una
línea escrita.
3. **Cada copia se verifica EN DESTINO.** «Lo copié» no prueba nada: el `rsync` que llenó el disco
de la caja nueva devolvió 0 y dejó 1367 artefactos VACÍOS, y sólo se vio contando del otro lado.
── LO QUE EL PLAN SABE QUE UN GUIÓN A MANO OLVIDA ──────────────────────────────────────────────
· **Dimensiona con el tamaño de COPIA, no con `du`.** Un árbol con hardlinks mide 60 G y ocupa 85 G
al copiarlo sin `-H` (medido). El preflight compara contra el número correcto y falla ANTES.
· **`rsync -aH` y excluyendo cachés.** Sin `-H` los hardlinks se expanden; `.dmerge` son 11 G de
caché regenerable que llenó un disco.
· **Los servicios `no-declarado` no se copian: se DECLARAN.** Copiar un binario no lo levanta al
arrancar — el caddy de la caja nueva murió en el primer reinicio por exactamente eso.
Uso:
scripts/mudanza/planear.py --censo censo.toml --target root@1.2.3.4 [--decide] --out plan.toml
"""
import argparse, os, sys, shlex, time, tomllib
# ── tamaños ─────────────────────────────────────────────────────────────────────────────────────
_MUL = {"K": 1 / 1024, "M": 1, "G": 1024, "T": 1024 * 1024}
def a_mib(s):
"""'61.5G' → MiB. Devuelve None si no se entiende: un tamaño ilegible NO se inventa como 0,
porque un 0 inventado apaga el preflight justo cuando más hace falta."""
s = (s or "").strip()
if not s or not s[0].isdigit():
return None
try:
return float(s[:-1]) * _MUL.get(s[-1].upper(), 1) if s[-1].isalpha() else float(s)
except ValueError:
return None
def humano(mib):
if mib is None:
return "?"
return f"{mib/1024:.1f}G" if mib >= 1024 else f"{mib:.0f}M"
# ── decisión ────────────────────────────────────────────────────────────────────────────────────
DECISIONES = ("muda", "muere")
def entradas(censo):
"""Todo lo que necesita una decisión, con una etiqueta legible."""
out = []
for s in censo.get("servicio", []):
# `declarado-muerto` no necesita decisión: no corre. Se informa y se deja morir.
if s.get("class") != "declarado-muerto":
out.append(("servicio", s["name"], s))
for d in censo.get("dominio", []):
if d.get("class") != "fosil-sin-dns":
out.append(("dominio", d["domain"], d))
for d in censo.get("datos", []):
out.append(("datos", d["path"], d))
return out
# ── servicios que NO APLICAN en una caja remota ─────────────────────────────────────────────────
# Correr en el origen no significa que tenga sentido en el destino. Un servidor en un datacenter no
# tiene bluetooth, ni módem, ni batería, ni teclado — y arrastrarlos es peor que inútil: algunos
# PELEAN con el destino. Cada familia con su motivo, porque una recomendación sin motivo no se puede
# discutir y entonces se acepta a ciegas o se ignora entera.
NO_APLICA = {
"hardware local (una caja remota no tiene ese dispositivo)": {
"bluetoothd", "bluetooth", "modemmanager", "upowerd", "upower", "acpid", "adb",
"thermald", "iio-sensor-proxy", "fwupd", "alsactl", "pcscd", "cups", "cupsd",
},
"escritorio o pantalla (un servidor no los tiene)": {
"waypipe", "xorg", "x", "wayland", "gdm", "sddm", "lightdm", "pipewire",
"wireplumber", "pulseaudio", "carmen-dm",
},
"la red del destino ya la configura su propio init — pelearían": {
"networkmanager", "dhcpcd", "wpa-supplicant", "connman", "systemd-networkd", "dhclient",
},
"lo provee el init del destino, no se muda": {
"agetty", "getty", "console-getty", "udevd", "udevadm", "systemd-udevd", "eudev",
"elogind-daemon", "elogind", "polkitd", "dbus-daemon", "dbus",
},
}
def _no_aplica(nombre):
n = nombre.lower().replace("_", "-")
for motivo, conjunto in NO_APLICA.items():
if n in conjunto:
return motivo
return None
def origen_binario(e, repo=None):
"""De dónde sale el binario, y por lo tanto QUÉ HAY QUE HACER para tenerlo en el destino.
«Instalá el paquete» es lo que uno ya sabía. Esto responde otra cosa, cruzando dos hechos: quién
posee el fichero en el ORIGEN (se lo preguntó el censo a su gestor de paquetes) y si el corpus de
takana tiene una receta con ese nombre. Los tres casos piden trabajo distinto:
· `receta-takana` — lo mejor: takana ya sabe construirlo. Se instala del repo firmado.
· `paquete-ajeno` — lo trae el gestor del origen. En el destino hace falta una receta, o
`qorpa` (ADR 0015) si se acepta una jaula con la distro ajena.
· `suelto` — NADIE lo provee: un binario puesto a mano. Hay que LLEVARLO con su árbol
o escribirle una receta. Es el que se pierde al apagar el origen.
Medido en gioser: 13 servicios con paquete y **13 binarios sueltos** — `caddy`, `qdrant`,
`matilda`, `act_runner`, `pacha-secretos`… en `/usr/local/bin` o en un home. La mitad."""
nombre, exe, pkg = e.get("name", ""), e.get("exec", ""), e.get("paquete", "")
# scripts/mudanza/planear.py → scripts/mudanza → scripts → RAÍZ. Con un `dirname` de menos
# apuntaba a `scripts/`, no encontraba ninguna receta, y TODO salía `suelto` — incluido `caddy`,
# que sí tiene `recipes/caddy.toml`. Un clasificador que contesta siempre lo mismo no clasifica.
repo = repo or os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
if not os.path.isdir(os.path.join(repo, "recipes")):
# Falla ruidosamente: sin el catálogo no se puede distinguir `receta-takana` de `suelto`, y
# contestar `suelto` a todo sería una respuesta falsa con forma de respuesta.
raise SystemExit(f"!! no encuentro el catálogo de recetas en {repo}/recipes — "
f"pasá --repo o corré desde el repo")
for cand in (nombre, nombre.replace("_", "-"), os.path.basename(exe or "")):
if cand and os.path.exists(os.path.join(repo, "recipes", f"{cand}.toml")):
import glob
sellado = bool(glob.glob(os.path.join(repo, "store", f"*-{cand}")))
return ("receta-takana",
f"takana YA lo construye: `recipes/{cand}.toml`"
+ (" (sellado en el store)" if sellado else " (sin sellar: hay que construirlo)")
+ f" ⇒ instalarlo del repo firmado, no copiar el binario de {exe or '?'}")
if pkg:
return ("paquete-ajeno",
f"lo provee el gestor del origen: paquete `{pkg}`. En el destino hace falta una "
f"receta takana, o `qorpa` si se acepta una jaula con la distro ajena (ADR 0015)")
if exe:
return ("suelto",
f"NADIE lo provee: binario puesto a mano en {exe}. Hay que LLEVARLO (con lo que "
f"necesite alrededor) o escribirle una receta — es de los que se pierden al apagar "
f"el origen")
return ("desconocido", "no se pudo leer su binario: hace falta averiguarlo a mano")
def recomendar(tipo, e, datos_que_mueren=()):
"""Devuelve `(decision, motivo)`. **Recomendación, nunca decisión**: se imprime con su razón para
que aceptarla sea barato y equivocarse sea caro de no notar.
Un motivo no es cortesía: una recomendación sin razón no se puede discutir, así que o se acepta a
ciegas o se ignora entera. Las dos son malas."""
c = e.get("class", "")
if tipo == "servicio":
m = _no_aplica(e.get("name", ""))
if m:
return "muere", m
# DERIVADA, y es la más fuerte: si el binario vive en un árbol que NO se muda, el servicio no
# va a poder arrancar allá. Se detecta cruzando su `cmdline` contra las decisiones de datos.
exe = (e.get("cmdline") or "").split(" ")[0]
for raiz in datos_que_mueren:
if exe.startswith(raiz.rstrip("/") + "/"):
return "muere", f"su binario vive en {raiz}, que NO se muda ⇒ no podría arrancar allá"
if c == "no-declarado":
p = e.get("ports") or []
return "muda", ("corre y sirve " + ", ".join(str(x) for x in p) + ": alguien lo quiere, "
"y NADIE lo declara" if p else
"corre sin estar declarado: alguien lo quiso, y nadie sabe levantarlo")
if c == "vivo":
return "muda", "declarado y corriendo"
return "", ""
if tipo == "dominio":
if c == "vivo":
return "muda", "resuelve a esta máquina y contesta"
if c == "ya-mudado":
return "muere", f"su DNS ya apunta a {e.get('dns','otra máquina')}: el bloque de acá es fósil"
if c == "backend-caido":
return "muere", f"resuelve acá pero devuelve {e.get('http','5xx')}: no hay backend"
if c == "sin-sondear":
return "", "sin sondear (censado en local): hace falta mirarlo desde afuera"
return "", ""
if tipo == "datos":
# A propósito SIN recomendación: qué datos valen no se deduce de la máquina. `terapeuta.ec`
# eran 279 M sin DNS y el usuario decidió que mueran; nada en el sistema podía decidir eso.
return "", "los datos no se recomiendan solos: sólo vos sabés si valen"
return "", ""
def sugerir(tipo, e, datos_que_mueren=()):
return recomendar(tipo, e, datos_que_mueren)[0]
def _etiqueta(tipo, nombre, e):
if tipo == "datos":
extra = e.get("size", "?")
if e.get("size_copy") and e["size_copy"] != e.get("size"):
extra += f" (copiar {e['size_copy']})"
return extra
if tipo == "servicio":
p = e.get("ports") or []
return ("puertos " + ",".join(str(x) for x in p)) if p else e.get("class", "")
return f"{e.get('dns','')} {e.get('http','')}".strip() or e.get("class", "")
def revisar(censo, interactivo=True):
"""Muestra TODO con su recomendación y su motivo, y deja elegir.
── EL ORDEN NO ES CAPRICHOSO ────────────────────────────────────────────────────────────────
Primero los DATOS, después servicios y dominios. Porque la recomendación más fuerte sobre un
servicio se DERIVA de los datos: si su binario vive en un árbol que no se muda, no va a poder
arrancar allá. Al revés no se puede calcular.
"""
todos = entradas(censo)
datos = [(t, n, e) for t, n, e in todos if t == "datos"]
resto = [(t, n, e) for t, n, e in todos if t != "datos"]
# ── 1. datos ────────────────────────────────────────────────────────────────────────────────
if datos:
print("\n══ 1/2 · DATOS — esto no se recomienda solo ═══════════════════════════════════")
print(" Qué datos valen no se deduce de la máquina: sólo vos sabés. Y de esto se deriva")
print(" la recomendación de los servicios, así que va primero.\n")
for t, n, e in datos:
marca = e.get("decision") or "—"
print(f" {n:34} {_etiqueta(t,n,e):>22} decisión actual: {marca}")
if interactivo:
print()
for t, n, e in datos:
if e.get("decision") in DECISIONES:
continue
r = input(f" {n:34} [{_etiqueta(t,n,e)}] [m]uda/[x]muere > ").strip().lower()
e["decision"] = {"m": "muda", "x": "muere"}.get(r, "")
mueren = [e["path"] for t, n, e in datos if e.get("decision") == "muere"]
# ── 2. lo demás, con recomendación y motivo ─────────────────────────────────────────────────
grupos = {"muda": [], "muere": [], "": []}
for t, n, e in resto:
d, motivo = recomendar(t, e, mueren)
e["_rec"], e["_motivo"] = d, motivo
grupos[d].append((t, n, e))
print("\n══ 2/2 · SERVICIOS Y DOMINIOS ════════════════════════════════════════════════════")
titulos = {"muda": "se recomienda MUDAR", "muere": "se recomienda NO mudar",
"": "SIN recomendación — hay que decidir"}
for g in ("muda", "muere", ""):
if not grupos[g]:
continue
print(f"\n── {titulos[g]} ({len(grupos[g])}) " + "─" * 30)
for t, n, e in sorted(grupos[g], key=lambda x: (x[0], x[1])):
print(f" {t:8} {n:30} {_etiqueta(t,n,e):>18}")
if e.get("_motivo"):
print(f" └─ {e['_motivo']}")
if t == "servicio" and e.get("decision") != "muere" and g != "muere":
oc, od = origen_binario(e)
print(f" └─ binario: [{oc}] {od[:96]}")
if not interactivo:
return censo
print("\n [a] aceptar todas las recomendaciones · [e] revisar una por una · [q] salir")
r = input(" > ").strip().lower()
if r == "q":
return censo
for t, n, e in resto:
if e.get("decision") in DECISIONES:
continue
rec = e.get("_rec", "")
if r == "a" and rec:
e["decision"] = rec
continue
sug = f"[{rec}]" if rec else "[sin recomendación]"
resp = input(f" {t:8} {n:30} {sug} {e.get('_motivo','')[:60]}\n"
f" [m]uda/[x]muere/Enter={rec or 'nada'} > ").strip().lower()
e["decision"] = {"m": "muda", "x": "muere", "": rec}.get(resp, "")
return censo
def validar(censo):
"""Regla 1: nada sin decidir se ejecuta."""
faltan = [(t, n) for t, n, e in entradas(censo) if e.get("decision") not in DECISIONES]
if faltan:
print("\n!! el plan NO se genera: hay decisiones sin tomar.", file=sys.stderr)
print(" El silencio no es consentimiento — sin decisión, ni mudar ni matar es correcto.\n",
file=sys.stderr)
for t, n in faltan[:40]:
print(f" {t:9} {n}", file=sys.stderr)
if len(faltan) > 40:
print(f" … y {len(faltan)-40} más", file=sys.stderr)
print("\n Corré con --decide, o poné `decision` a mano en el censo.", file=sys.stderr)
sys.exit(2)
# ── plan ────────────────────────────────────────────────────────────────────────────────────────
def pasos(censo, target, key):
ssh_opts = "-o StrictHostKeyChecking=accept-new -o ConnectTimeout=20"
if key:
ssh_opts = f"-i {key} " + ssh_opts
ssh = f"ssh {ssh_opts} {target}"
P = []
def paso(clase, titulo, cmd, verifica=None, nota=None, verifica_tipo="cmd"):
# `verifica_tipo` no es decoración: el aplicador NO puede tratar igual «corré esto» que «un
# humano tiene que mirar esto». Dar por buena una verificación humana porque «no dio error»
# es exactamente cómo un aplicador miente.
P.append({"clase": clase, "titulo": titulo, "cmd": cmd,
"verifica": verifica or "", "verifica_tipo": verifica_tipo, "nota": nota or ""})
datos_muda = [d for d in censo.get("datos", []) if d.get("decision") == "muda"]
svc_muda = [s for s in censo.get("servicio", []) if s.get("decision") == "muda"]
doms_muda = [d for d in censo.get("dominio", []) if d.get("decision") == "muda"]
muere = [(t, n, e) for t, n, e in entradas(censo) if e.get("decision") == "muere"]
# ── 0. preflight ────────────────────────────────────────────────────────────────────────────
need = sum(x for x in (a_mib(d.get("size_copy") or d.get("size")) for d in datos_muda)
if x is not None)
ilegibles = [d["path"] for d in datos_muda
if a_mib(d.get("size_copy") or d.get("size")) is None]
paso("preflight", f"el destino responde y tiene sitio (hacen falta {humano(need)})",
f"{ssh} 'df -Pm / | tail -1'",
verifica=f"la columna 'Available' debe ser > {int(need)} MiB",
verifica_tipo="humano",
nota=("⚠ tamaños ilegibles, NO contados en el total: " + ", ".join(ilegibles)) if ilegibles else
"Dimensionado con el tamaño de COPIA (hardlinks expandidos), no con `du`.")
# ── 1. datos ────────────────────────────────────────────────────────────────────────────────
for d in datos_muda:
p = d["path"]
paso("datos", f"copiar {p} ({d.get('size','?')})",
f"rsync -aH --partial --info=stats2 --exclude='.dmerge' "
f"-e {shlex.quote('ssh ' + ssh_opts)} {shlex.quote(p + '/')} {target}:{shlex.quote(p + '/')}",
verifica=(f"src=$(find {shlex.quote(p)} -type f | wc -l); "
f"dst=$({ssh} 'find {shlex.quote(p)} -type f | wc -l'); "
f'echo "ficheros: origen=$src destino=$dst"; [ "$src" = "$dst" ]'),
nota="`-H` obligatorio: sin él los hardlinks se expanden (60 G medidos → 85 G). "
"Y la verificación COMPARA, no informa: cuenta los ficheros de los dos lados y "
"FALLA si no coinciden. La versión anterior imprimía el número de directorios "
"vacíos y devolvía 0 igual — un guardián que siempre pasa no es un guardián. "
"Hace falta porque un rsync que se corta deja el destino a medias y devuelve 0: "
"así se llenó un disco y quedaron 1367 artefactos vacíos.")
# ── 2. servicios ────────────────────────────────────────────────────────────────────────────
for s in svc_muda:
nd = s.get("class") == "no-declarado"
paso("servicio", f"declarar y levantar {s['name']}" + (" ⚠ NO-DECLARADO" if nd else ""),
(f"# DE DÓNDE SALE EL BINARIO: {origen_binario(s)[0]}\n"
f"# {origen_binario(s)[1]}\n"
f"#\n"
f"# CÓMO CORRE HOY EN EL ORIGEN (leído de /proc, es lo que hay que reproducir):\n"
f"# cmdline: {s.get('cmdline') or '(no se pudo leer)'}\n"
+ (f"# cwd : {s['cwd']}\n" if s.get("cwd") else "")
+ (f"# puertos: {', '.join(str(p) for p in s['ports'])}\n" if s.get("ports") else "")
+ f"#\n# 1) conseguir el binario según lo de arriba\n"
f"# 2) DECLARARLO en el init del destino con ESA línea de comando\n"
f"# (arje: una tarjeta en /ente/seed.card.json · systemd: una unit)\n"
f"# 3) arrancarlo"),
verifica=f"{ssh} 'ps -eo ppid,comm | awk \"\\$1==1\" | grep -w {shlex.quote(s['name'])}'"
+ (f" && {ssh} 'ss -lnt | grep -E \":({'|'.join(str(p) for p in s.get('ports', []))}) \"'"
if s.get("ports") else ""),
nota="⚠ ESTE SERVICIO CORRE SIN ESTAR DECLARADO en el origen: copiar el binario no lo "
"levanta al arrancar. Hay que escribir su declaración — es el trabajo real de la "
"mudanza, y el que se olvida (el caddy de la caja nueva murió así)." if nd else
"Declarado en el origen: traducir su unidad al init del destino.")
# ── 3. dominios ─────────────────────────────────────────────────────────────────────────────
for d in doms_muda:
paso("dominio", f"apuntar {d['domain']} a la máquina nueva",
f"# cambiar el registro A de {d['domain']}: {d.get('dns','?')} → <IP NUEVA>",
verifica=f"getent ahostsv4 {d['domain']} | head -1 && "
f"curl -sk -o /dev/null -w '%{{http_code}}' https://{d['domain']}/",
nota="La mudanza NO está hecha hasta que el dominio contesta 2xx contra la IP NUEVA. "
"Comprobarlo desde AFUERA, nunca desde la máquina: pedirle a un servidor sus "
"propios dominios puede disparar su fail2ban (pasó).")
# ── 4. lo que muere ─────────────────────────────────────────────────────────────────────────
if muere:
det = []
for t, n, e in muere:
tam = f" ({e.get('size')})" if e.get("size") else ""
det.append(f"# {t:9} {n}{tam}")
paso("muere", f"{len(muere)} cosa(s) NO se mudan y mueren con la máquina vieja",
"# NO SE BORRA NADA ACÁ. Esta lista existe para leerse antes de apagar el origen:\n"
+ "\n".join(det),
verifica="revisión humana: leer la lista completa antes de destruir el origen",
verifica_tipo="humano",
nota="El día que se apague la máquina vieja esto no se puede recuperar. Que esté "
"escrito ES el paso.")
return P
def emitir(censo, P, target):
L = ["# PLAN DE MUDANZA — generado por scripts/mudanza/planear.py (SDD 29 etapa 2).",
"#",
"# Cada paso lleva su COMANDO LITERAL: este fichero se puede ejecutar a mano, línea por",
"# línea, sin la herramienta y sin el repo. Ésa es la idea — es el registro exportable.",
'schema = "takana-plan/1"',
f'fecha = "{time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())}"',
f'origen = "{censo.get("maquina", {}).get("hostname", "?")}"',
f'destino = "{target}"',
""]
for i, p in enumerate(P, 1):
L.append(f"[[paso]]")
L.append(f'n = {i}')
L.append(f'clase = "{p["clase"]}"')
L.append(f'titulo = """{p["titulo"]}"""')
L.append(f'cmd = """\n{p["cmd"]}\n"""')
if p["verifica"]:
L.append(f'verifica = """\n{p["verifica"]}\n"""')
L.append(f'verifica_tipo = "{p["verifica_tipo"]}"')
if p["nota"]:
L.append(f'nota = """{p["nota"]}"""')
L.append("")
return "\n".join(L) + "\n"
def main():
ap = argparse.ArgumentParser(description="Convierte un censo decidido en un plan ejecutable.")
ap.add_argument("--censo", required=True)
ap.add_argument("--target", required=True, help="usuario@host destino")
ap.add_argument("--key", help="clave SSH para el destino")
ap.add_argument("--decide", action="store_true",
help="mostrar TODO con su recomendación y su motivo, y elegir")
ap.add_argument("--revisar", action="store_true",
help="sólo mostrar la revisión con recomendaciones, sin preguntar")
ap.add_argument("--out", help="fichero del plan (default: stdout)")
a = ap.parse_args()
with open(a.censo, "rb") as f:
censo = tomllib.load(f)
if a.revisar:
revisar(censo, interactivo=False)
return
if a.decide:
censo = revisar(censo, interactivo=True)
validar(censo)
P = pasos(censo, a.target, a.key)
por_clase = {}
for p in P:
por_clase[p["clase"]] = por_clase.get(p["clase"], 0) + 1
print(f"\n══ PLAN: {censo.get('maquina',{}).get('hostname','?')} → {a.target} ══")
for c, n in por_clase.items():
print(f" {c:10} {n:3} paso(s)")
txt = emitir(censo, P, a.target)
if a.out:
open(a.out, "w").write(txt)
print(f"\n plan → {a.out} ({len(P)} pasos, con su comando literal cada uno)")
else:
print(txt)
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink