131 lines
6.1 KiB
Python
131 lines
6.1 KiB
Python
"""
|
|
restaurar.py — Recupera los datos desde la copia cifrada de tu servidor git.
|
|
|
|
Para un USB nuevo (o una carpeta de datos perdida): descarga «cartera-datos»,
|
|
lo descifra con la llave de hardware y escribe los datos y los ajustes (`bitcoin.json`
|
|
y `extractos.json`). Nunca sobrescribe datos que ya existan. Deja la sincronización configurada, así que la siguiente
|
|
sincronización desde la app parte de este punto.
|
|
|
|
python3 -m cartera.restaurar ← a programa/datos
|
|
python3 -m cartera.restaurar --datos OTRA/ ← a otra carpeta
|
|
python3 -m cartera.restaurar --comparar-con programa/datos ← simulacro
|
|
|
|
Con --comparar-con no escribe nada: descifra y cuenta cuántos registros
|
|
difieren de esa carpeta. Solo se muestran recuentos, nunca importes.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from .datos import FICHEROS, Datos, escribir, leer
|
|
from .sincronizar import (AJUSTES, INDICE, ErrorSync, Sincronizador, _entorno, escribir_ajuste,
|
|
huella_total)
|
|
|
|
PROGRAMA = Path(__file__).resolve().parent.parent
|
|
|
|
|
|
def url_por_defecto() -> str | None:
|
|
"""El repositorio de datos vive junto al del programa: …/cartera-app → …/cartera-datos."""
|
|
try:
|
|
r = subprocess.run(["git", "-C", str(PROGRAMA.parent), "remote", "get-url", "origin"],
|
|
capture_output=True, text=True, timeout=10, env=_entorno())
|
|
except (OSError, subprocess.SubprocessError):
|
|
return None
|
|
url = r.stdout.strip()
|
|
if r.returncode != 0 or "cartera-app" not in url:
|
|
return None
|
|
return url.replace("cartera-app", "cartera-datos")
|
|
|
|
|
|
def elegir_clave(sync: Sincronizador, huella: str | None) -> dict:
|
|
claves = sync.claves()
|
|
if huella:
|
|
clave = next((c for c in claves if c["huella"] == huella.strip().upper()), None)
|
|
if clave is None:
|
|
raise ErrorSync("Esa clave GPG no está en este equipo (¿está puesta tu llave?).")
|
|
return clave
|
|
en_tarjeta = [c for c in claves if c["en_tarjeta"]]
|
|
candidatas = en_tarjeta or claves
|
|
if len(candidatas) != 1:
|
|
nombres = ", ".join(f"{c['nombre']} ({c['huella'][-16:]})" for c in candidatas) or "ninguna"
|
|
raise ErrorSync(f"No sé qué clave usar ({nombres}). Indícala con --huella.")
|
|
return candidatas[0]
|
|
|
|
|
|
def diferencias(a: Datos, b: Datos) -> int:
|
|
total = 0
|
|
for nombre in FICHEROS:
|
|
x = {json.dumps(r, sort_keys=True, ensure_ascii=False) for r in getattr(a, nombre)}
|
|
y = {json.dumps(r, sort_keys=True, ensure_ascii=False) for r in getattr(b, nombre)}
|
|
total += len(x ^ y)
|
|
return total
|
|
|
|
|
|
def restaurar(datos: Path, url: str, huella: str | None = None, sync_dir: Path | None = None,
|
|
comparar_con: Path | None = None) -> dict:
|
|
datos = Path(datos).resolve()
|
|
if comparar_con is None and (datos / "activos.jsonl").exists():
|
|
raise ErrorSync(f"Ya hay datos en {datos}: no se sobrescriben. Usa la sincronización de la app.")
|
|
sync = Sincronizador(Path(sync_dir) if sync_dir else datos.parent / "sync", ajustes=datos.parent)
|
|
clave = elegir_clave(sync, huella)
|
|
sync.configurar(url, clave["huella"])
|
|
blob = sync._preparar_repo()
|
|
if blob is None:
|
|
raise ErrorSync("En el servidor git aún no hay ninguna copia de los datos.")
|
|
remoto, ajustes = sync._descifrar()
|
|
registros = {n: len(getattr(remoto, n)) for n in FICHEROS}
|
|
if comparar_con is not None:
|
|
return {"registros": registros, "ajustes": sorted(ajustes),
|
|
"diferencias": diferencias(remoto, leer(Path(comparar_con)))}
|
|
escribir(datos, remoto)
|
|
for nombre in AJUSTES: # las copias antiguas no los traen
|
|
if nombre in ajustes:
|
|
escribir_ajuste(datos.parent / nombre, ajustes[nombre])
|
|
adjuntos = sync.traer_adjuntos(datos, remoto, ajustes.get(INDICE))
|
|
# La app arranca sabiendo que sus datos son justo los del servidor git
|
|
sync._guardar_json(sync._estado, {"blob": blob, "huella_local": huella_total(remoto, ajustes)})
|
|
return {"registros": registros, "ajustes": sorted(ajustes), "carpeta": str(datos), "adjuntos": adjuntos}
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
p = argparse.ArgumentParser(description="Recupera los datos desde la copia cifrada de tu servidor git.")
|
|
p.add_argument("--datos", type=Path, default=PROGRAMA / "datos")
|
|
p.add_argument("--url", help="repositorio de datos (por defecto, cartera-datos junto a cartera-app)")
|
|
p.add_argument("--huella", help="huella de la clave GPG (por defecto, la de tu llave de hardware)")
|
|
p.add_argument("--sync", type=Path, help="carpeta de la copia local del repositorio")
|
|
p.add_argument("--comparar-con", type=Path, help="no escribe: compara la copia de tu servidor git con esta carpeta")
|
|
a = p.parse_args(argv)
|
|
url = a.url or url_por_defecto()
|
|
if not url:
|
|
print("No sé la dirección del repositorio de datos: indícala con --url.")
|
|
return 1
|
|
print("Descargando la copia cifrada de tu servidor git… (si tu llave parpadea, tócala; puede pedir el PIN)")
|
|
try:
|
|
r = restaurar(a.datos, url, a.huella, a.sync, a.comparar_con)
|
|
except ErrorSync as e:
|
|
print(f"✗ {e}")
|
|
return 1
|
|
resumen = ", ".join(f"{n}: {c}" for n, c in r["registros"].items())
|
|
if a.comparar_con is not None:
|
|
if r["diferencias"]:
|
|
print(f"⚠ La copia de tu servidor git difiere en {r['diferencias']} registro(s) de {a.comparar_con} "
|
|
f"({resumen}). Normal si no has sincronizado desde el último cambio.")
|
|
else:
|
|
print(f"✓ La copia de tu servidor git es idéntica a {a.comparar_con} ({resumen}).")
|
|
return 0
|
|
print(f"✓ Datos recuperados en {r['carpeta']} ({resumen}).")
|
|
if r["ajustes"]:
|
|
print(f" Y los ajustes: {', '.join(r['ajustes'])}.")
|
|
if r["adjuntos"]["traidos"] or r["adjuntos"]["faltan"]:
|
|
print(f" Justificantes recuperados: {r['adjuntos']['traidos']}"
|
|
+ (f" (⚠ {r['adjuntos']['faltan']} no estaban en la copia)" if r["adjuntos"]["faltan"] else "") + ".")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|