486 lines
24 KiB
Python
486 lines
24 KiB
Python
"""
|
|
sincronizar.py — Copia cifrada de los datos en el servidor git y sincronización entre copias.
|
|
|
|
Los datos (los .jsonl) y los ajustes (`bitcoin.json` y `extractos.json`) se
|
|
empaquetan, se cifran con GPG para la clave del usuario (la que tenga, si acaso en una llave de hardware) y se
|
|
guardan como un único fichero en un repositorio git privado («cartera-datos»).
|
|
el servidor git solo ve un fichero cifrado.
|
|
|
|
Sincronizar =
|
|
1. traer el repositorio;
|
|
2. si el fichero cifrado ha cambiado desde la última vez, descifrarlo
|
|
(pide PIN o toque de la llave de hardware) y fusionarlo con los datos locales;
|
|
3. si el resultado difiere de lo que hay en el servidor git, cifrarlo y subirlo
|
|
(cifrar solo necesita la clave pública: no pide la llave de hardware).
|
|
|
|
La copia lleva version.json: si viene de una app más antigua se migra al leerla
|
|
(antes de fusionar); si viene de una más nueva, se para sin tocar nada.
|
|
|
|
Los justificantes (adjuntos) NO van en ese fichero, que se sube entero cada vez:
|
|
van en paquetes aparte, `adjuntos/<nombre>.tar.gpg`, cifrados para la misma
|
|
clave. Cada sincronización que encuentra justificantes nuevos sube un paquete
|
|
solo con esos, y un paquete no se vuelve a tocar: cada fichero se sube una vez.
|
|
El índice (qué hay en cada paquete) viaja dentro de la copia cifrada, como un
|
|
ajuste más («adjuntos.json»), así que el servidor git no ve ni los nombres.
|
|
|
|
La copia local del repositorio vive en programa/sync/ (fuera del repositorio
|
|
del programa). Nada de esto sale de la máquina sin cifrar.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
import io
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import tarfile
|
|
import tempfile
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
from . import adjuntos as adj
|
|
|
|
from .btc import MAX_VISTAS
|
|
from .datos import FICHERO_VERSION, FICHEROS, Datos, ErrorDatos, escribir, fusionar, huella_datos, leer
|
|
from . import almacen
|
|
|
|
CIFRADO = "datos.tar.gz.gpg"
|
|
|
|
# Ajustes que viajan con la copia: no son datos de la cartera, pero si se
|
|
# pierden hay que volver a escribirlos a mano, y las claves públicas de los
|
|
# monederos no son cómodas de teclear.
|
|
AJUSTES = ("bitcoin.json", "extractos.json", "adjuntos.json")
|
|
INDICE = "adjuntos.json" # {"paquetes": {nombre: [huellas]}, "modificado": …}
|
|
PAQUETES = "adjuntos" # carpeta del repositorio con los paquetes cifrados
|
|
PERMITIDOS = {f"{nombre}.jsonl" for nombre in FICHEROS} | set(AJUSTES) | {FICHERO_VERSION}
|
|
|
|
|
|
class ErrorSync(RuntimeError):
|
|
"""Mensaje pensado para mostrárselo al usuario."""
|
|
|
|
|
|
def _entorno() -> dict:
|
|
env = dict(os.environ)
|
|
# Con doble clic se hereda el agente SSH del escritorio, que no tiene la
|
|
# clave de la llave de hardware: si el agente SSH de gpg-agent tiene claves, se usa ese.
|
|
try:
|
|
socket = subprocess.run(["gpgconf", "--list-dirs", "agent-ssh-socket"], capture_output=True,
|
|
text=True, timeout=10).stdout.strip()
|
|
if socket and subprocess.run(["ssh-add", "-L"], env={**env, "SSH_AUTH_SOCK": socket},
|
|
capture_output=True, timeout=10).returncode == 0:
|
|
env["SSH_AUTH_SOCK"] = socket
|
|
except (OSError, subprocess.SubprocessError):
|
|
pass
|
|
env["GIT_TERMINAL_PROMPT"] = "0"
|
|
env.setdefault("GIT_SSH_COMMAND", "ssh -o BatchMode=yes -o ConnectTimeout=15")
|
|
return env
|
|
|
|
|
|
def _ultima_linea(texto: str) -> str:
|
|
lineas = [l.strip() for l in (texto or "").splitlines() if l.strip()]
|
|
return lineas[-1] if lineas else ""
|
|
|
|
|
|
# ── Ajustes (bitcoin.json, extractos.json) ───────────────────────────────────
|
|
#
|
|
# A diferencia de los datos, los ajustes NO se fusionan campo a campo: se queda
|
|
# entero el fichero que se tocó más tarde. Es lo que hay que hacer cuando solo
|
|
# hay un equipo, que es el caso normal: si borras un monedero y se fusionara por
|
|
# partes, el monedero volvería desde la copia de tu servidor git en la siguiente
|
|
# sincronización.
|
|
#
|
|
# La única excepción son las transacciones ya apuntadas de cada monedero, que se
|
|
# suman: perderlas haría que la aplicación volviese a proponer operaciones que
|
|
# ya están guardadas. Sumarlas no puede resucitar un monedero borrado, porque
|
|
# solo se tocan los monederos que estén en el fichero que gana.
|
|
|
|
def _marca(ajuste: dict | None) -> str:
|
|
return str((ajuste or {}).get("modificado") or "")
|
|
|
|
|
|
def _canonico(ajuste: dict | None) -> str:
|
|
return json.dumps(ajuste or {}, sort_keys=True, ensure_ascii=False)
|
|
|
|
|
|
def leer_ajustes(carpeta: Path) -> dict[str, dict]:
|
|
"""Los ficheros de ajustes que existan, por nombre."""
|
|
salida = {}
|
|
for nombre in AJUSTES:
|
|
try:
|
|
contenido = json.loads(almacen.leer_texto(Path(carpeta) / nombre))
|
|
except (OSError, ValueError):
|
|
continue
|
|
if isinstance(contenido, dict):
|
|
salida[nombre] = contenido
|
|
return salida
|
|
|
|
|
|
def escribir_ajuste(ruta: Path, contenido: dict) -> None:
|
|
"""Lo escribe entero o no lo escribe; y solo lo lee su dueño."""
|
|
ruta = Path(ruta)
|
|
tmp = ruta.with_name(f".{ruta.name}.tmp")
|
|
try:
|
|
tmp.write_bytes(almacen.sellar(json.dumps(contenido, ensure_ascii=False, indent=1).encode("utf-8")))
|
|
os.chmod(tmp, 0o600)
|
|
os.replace(tmp, ruta)
|
|
except OSError:
|
|
tmp.unlink(missing_ok=True)
|
|
raise
|
|
|
|
|
|
def _sumar_vistas(gana: dict, otro: dict) -> dict:
|
|
"""Suma al fichero ganador las transacciones ya apuntadas del otro."""
|
|
if not isinstance(otro.get("monederos"), list):
|
|
return gana
|
|
otros = {m.get("id"): m for m in otro["monederos"] if isinstance(m, dict)}
|
|
for m in gana.get("monederos") or []:
|
|
suyas = otros.get(m.get("id"), {}).get("vistas") or []
|
|
if suyas:
|
|
m["vistas"] = sorted(set(m.get("vistas") or []) | set(suyas))[-MAX_VISTAS:]
|
|
return gana
|
|
|
|
|
|
def huella_total(datos: Datos, ajustes: dict[str, dict]) -> str:
|
|
"""Resumen de todo lo que viaja en la copia: datos y ajustes."""
|
|
import hashlib
|
|
h = hashlib.sha256(huella_datos(datos).encode())
|
|
for nombre in AJUSTES:
|
|
h.update(f"{nombre}:{_canonico(ajustes.get(nombre))}".encode())
|
|
return h.hexdigest()
|
|
|
|
|
|
def elegir_ajustes(locales: dict[str, dict], remotos: dict[str, dict]) -> dict[str, dict]:
|
|
"""Para cada ajuste, el más reciente; si no hay marca de fecha, el local."""
|
|
salida = {}
|
|
for nombre in AJUSTES:
|
|
local, remoto = locales.get(nombre), remotos.get(nombre)
|
|
if remoto is None:
|
|
elegido = local
|
|
elif local is None:
|
|
elegido = remoto
|
|
else:
|
|
elegido = remoto if _marca(remoto) > _marca(local) else local
|
|
otro = local if elegido is remoto else remoto
|
|
if nombre == "bitcoin.json":
|
|
elegido = _sumar_vistas(json.loads(json.dumps(elegido)), otro)
|
|
if nombre == INDICE: # los paquetes nunca se quitan: se suman
|
|
elegido = json.loads(json.dumps(elegido))
|
|
elegido["paquetes"] = {**(otro.get("paquetes") or {}), **(elegido.get("paquetes") or {})}
|
|
if elegido is not None:
|
|
salida[nombre] = elegido
|
|
return salida
|
|
|
|
|
|
class Sincronizador:
|
|
def __init__(self, carpeta: Path, ajustes: Path | None = None, git: str = "git", gpg: str = "gpg"):
|
|
self.carpeta = Path(carpeta).resolve() # rutas relativas: git trabaja en otra carpeta
|
|
self.ajustes = Path(ajustes).resolve() if ajustes else self.carpeta.parent
|
|
self.repo = self.carpeta / "cartera-datos"
|
|
self._config = self.carpeta / "config.json"
|
|
self._estado = self.carpeta / "estado.json"
|
|
self.git, self.gpg = git, gpg
|
|
|
|
# ── configuración y estado ──
|
|
def _leer_json(self, ruta: Path) -> dict:
|
|
try:
|
|
return json.loads(ruta.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError):
|
|
return {}
|
|
|
|
def _guardar_json(self, ruta: Path, datos: dict) -> None:
|
|
self.carpeta.mkdir(parents=True, exist_ok=True)
|
|
tmp = ruta.with_suffix(".tmp")
|
|
tmp.write_text(json.dumps(datos, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
os.replace(tmp, ruta)
|
|
|
|
def config(self) -> dict:
|
|
return self._leer_json(self._config)
|
|
|
|
def configurar(self, url: str, huella: str) -> dict:
|
|
url, huella = (url or "").strip(), (huella or "").strip().upper()
|
|
if not url:
|
|
raise ErrorSync("Indica la dirección del repositorio de tu servidor git.")
|
|
clave = next((c for c in self.claves() if c["huella"] == huella), None)
|
|
if clave is None:
|
|
raise ErrorSync("Esa clave GPG no está en este equipo.")
|
|
anterior = self.config()
|
|
if anterior.get("url") and anterior["url"] != url and self.repo.exists():
|
|
shutil.rmtree(self.repo) # la copia local era de otro repositorio
|
|
self._estado.unlink(missing_ok=True)
|
|
cfg = {"url": url, "huella": huella, "nombre": clave["nombre"]}
|
|
self._guardar_json(self._config, cfg)
|
|
return cfg
|
|
|
|
def estado(self, datos: Datos) -> dict:
|
|
cfg, est = self.config(), self._leer_json(self._estado)
|
|
return {
|
|
"configurado": bool(cfg.get("url") and cfg.get("huella")),
|
|
"url": cfg.get("url"), "clave": cfg.get("nombre"), "huella": cfg.get("huella"),
|
|
"ultima": est.get("fecha"),
|
|
"pendiente": bool(cfg.get("url")) and est.get("huella_local") != huella_total(
|
|
datos, leer_ajustes(self.ajustes)),
|
|
}
|
|
|
|
def claves(self) -> list[dict]:
|
|
"""Claves GPG con subclave de cifrado de las que hay parte privada
|
|
(en una llave de hardware o en el equipo)."""
|
|
try:
|
|
r = subprocess.run([self.gpg, "--list-secret-keys", "--with-colons"], capture_output=True,
|
|
text=True, timeout=20, env=_entorno())
|
|
except (OSError, subprocess.SubprocessError):
|
|
return []
|
|
claves, actual, ultima = [], None, None
|
|
for linea in r.stdout.splitlines():
|
|
c = linea.split(":")
|
|
if c[0] == "sec":
|
|
actual = {"huella": "", "nombre": "", "cifra": False, "en_tarjeta": False,
|
|
"valida": c[1] not in ("e", "r", "d")}
|
|
claves.append(actual)
|
|
ultima = "sec"
|
|
elif c[0] == "ssb" and actual is not None:
|
|
ultima = "ssb"
|
|
if "e" in c[11] and c[1] not in ("e", "r", "d"):
|
|
actual["cifra"] = True
|
|
if len(c) > 14 and c[14] not in ("", "#", "+"): # + = en el equipo; nº de serie = tarjeta
|
|
actual["en_tarjeta"] = True
|
|
elif c[0] == "fpr" and actual is not None and ultima == "sec" and not actual["huella"]:
|
|
actual["huella"] = c[9]
|
|
elif c[0] == "uid" and actual is not None and not actual["nombre"]:
|
|
actual["nombre"] = c[9]
|
|
return [c for c in claves if c["cifra"] and c["valida"] and c["huella"]]
|
|
|
|
# ── git ──
|
|
def _git(self, *args, cwd: Path | None = None, timeout: int = 180) -> str:
|
|
try:
|
|
r = subprocess.run([self.git, *args], cwd=cwd or self.repo, capture_output=True, text=True,
|
|
timeout=timeout, env=_entorno(), stdin=subprocess.DEVNULL)
|
|
except subprocess.TimeoutExpired:
|
|
raise ErrorSync("El servidor git no responde (¿hay red hasta él? ¿has atendido tu llave?).") from None
|
|
except OSError:
|
|
raise ErrorSync("Falta git en este equipo (sudo apt install git).") from None
|
|
if r.returncode != 0:
|
|
raise ErrorSync(f"git {args[0]}: {_ultima_linea(r.stderr) or 'error'}")
|
|
return r.stdout.strip()
|
|
|
|
def _preparar_repo(self) -> str | None:
|
|
"""Deja la copia local igual que el servidor git. Devuelve el id del fichero
|
|
cifrado en el servidor git o None si aún no hay copia."""
|
|
cfg = self.config()
|
|
if not cfg.get("url"):
|
|
raise ErrorSync("La sincronización no está configurada.")
|
|
self.carpeta.mkdir(parents=True, exist_ok=True)
|
|
if not (self.repo / ".git").exists():
|
|
if self.repo.exists():
|
|
shutil.rmtree(self.repo)
|
|
self._git("clone", "--quiet", cfg["url"], str(self.repo), cwd=self.carpeta)
|
|
self._git("config", "user.name", "cartera")
|
|
self._git("config", "user.email", "cartera@local")
|
|
self._git("config", "core.fileMode", "false")
|
|
else:
|
|
self._git("fetch", "--quiet", "origin")
|
|
try:
|
|
self._git("rev-parse", "--verify", "--quiet", "refs/remotes/origin/main", timeout=30)
|
|
except ErrorSync:
|
|
self._git("symbolic-ref", "HEAD", "refs/heads/main", timeout=30) # repositorio vacío
|
|
return None
|
|
self._git("checkout", "--quiet", "-B", "main", "origin/main", timeout=60)
|
|
self._git("reset", "--hard", "--quiet", "origin/main", timeout=60)
|
|
try:
|
|
return self._git("rev-parse", f"HEAD:{CIFRADO}", timeout=30)
|
|
except ErrorSync:
|
|
return None
|
|
|
|
# ── cifrado ──
|
|
def _empaquetar(self, datos: Datos, ajustes: dict[str, dict]) -> bytes:
|
|
with tempfile.TemporaryDirectory() as t:
|
|
escribir(Path(t), datos)
|
|
for nombre, contenido in ajustes.items():
|
|
(Path(t) / nombre).write_text(_canonico(contenido), encoding="utf-8")
|
|
buf = io.BytesIO()
|
|
with tarfile.open(fileobj=buf, mode="w:gz", format=tarfile.PAX_FORMAT) as tar:
|
|
for nombre in sorted(PERMITIDOS):
|
|
ruta = Path(t) / nombre
|
|
if not ruta.exists():
|
|
continue # los ajustes son opcionales
|
|
info = tarfile.TarInfo(nombre)
|
|
contenido = ruta.read_bytes()
|
|
info.size, info.mtime, info.mode = len(contenido), 0, 0o600
|
|
tar.addfile(info, io.BytesIO(contenido))
|
|
return buf.getvalue()
|
|
|
|
def _cifrar(self, datos: Datos, ajustes: dict[str, dict], destino: Path) -> None:
|
|
self._cifrar_bytes(self._empaquetar(datos, ajustes), destino)
|
|
|
|
def _cifrar_bytes(self, contenido: bytes, destino: Path) -> None:
|
|
huella = self.config().get("huella")
|
|
r = subprocess.run([self.gpg, "--batch", "--yes", "--trust-model", "always", "--encrypt",
|
|
"--recipient", huella, "--output", str(destino)],
|
|
input=contenido, capture_output=True, timeout=300, env=_entorno())
|
|
if r.returncode != 0:
|
|
raise ErrorSync(f"No se pudo cifrar: {_ultima_linea(r.stderr.decode(errors='replace'))}")
|
|
|
|
def _descifrar_bytes(self, origen: Path) -> bytes:
|
|
try:
|
|
r = subprocess.run([self.gpg, "--quiet", "--decrypt", "--output", "-", str(origen)],
|
|
capture_output=True, timeout=300, env=_entorno(), stdin=subprocess.DEVNULL)
|
|
except subprocess.TimeoutExpired:
|
|
raise ErrorSync("Se agotó el tiempo esperando la llave (PIN o toque).") from None
|
|
if r.returncode != 0:
|
|
raise ErrorSync("No se pudo descifrar la copia de tu servidor git (¿llave puesta y PIN correcto?): "
|
|
+ _ultima_linea(r.stderr.decode(errors="replace")))
|
|
return r.stdout
|
|
|
|
def _descifrar(self) -> tuple[Datos, dict[str, dict]]:
|
|
claro = self._descifrar_bytes(self.repo / CIFRADO)
|
|
with tempfile.TemporaryDirectory() as t:
|
|
with tarfile.open(fileobj=io.BytesIO(claro), mode="r:gz") as tar:
|
|
for miembro in tar.getmembers():
|
|
if miembro.name not in PERMITIDOS or not miembro.isfile():
|
|
raise ErrorSync(f"La copia de tu servidor git contiene algo inesperado: {miembro.name}")
|
|
(Path(t) / miembro.name).write_bytes(tar.extractfile(miembro).read())
|
|
# Las copias anteriores a septiembre de 2026 no llevan ajustes, y las
|
|
# anteriores a octubre, versión. leer() las migra antes de fusionar.
|
|
try:
|
|
return leer(Path(t)), leer_ajustes(Path(t))
|
|
except ErrorDatos as e:
|
|
raise ErrorSync(f"No se puede usar la copia de tu servidor git: {e}") from None
|
|
|
|
# ── justificantes ──
|
|
def _paquetes_en_repo(self, indice: dict | None) -> dict[str, list[str]]:
|
|
"""Los paquetes del índice cuyo fichero está de verdad en el repositorio."""
|
|
return {n: h for n, h in ((indice or {}).get("paquetes") or {}).items()
|
|
if (self.repo / PAQUETES / f"{n}.tar.gpg").is_file()}
|
|
|
|
def traer_adjuntos(self, carpeta_datos: Path, datos: Datos, indice: dict | None) -> dict:
|
|
"""Descifra los paquetes que tienen justificantes que aquí faltan."""
|
|
faltan = {h for h in adj.referenciados(datos.movimientos) if not adj.tiene(carpeta_datos, h)}
|
|
traidos = 0
|
|
for nombre, huellas in self._paquetes_en_repo(indice).items():
|
|
if not faltan & set(huellas):
|
|
continue
|
|
claro = self._descifrar_bytes(self.repo / PAQUETES / f"{nombre}.tar.gpg")
|
|
with tarfile.open(fileobj=io.BytesIO(claro), mode="r:gz") as tar:
|
|
for miembro in tar.getmembers():
|
|
if miembro.name not in faltan or not miembro.isfile():
|
|
continue
|
|
contenido = tar.extractfile(miembro).read()
|
|
if adj.guardar(carpeta_datos, contenido) != miembro.name:
|
|
raise ErrorSync(f"Un justificante del paquete {nombre} no coincide con su huella.")
|
|
faltan.discard(miembro.name)
|
|
traidos += 1
|
|
return {"traidos": traidos, "faltan": len(faltan)}
|
|
|
|
def _subir_adjuntos(self, carpeta_datos: Path, datos: Datos, ajustes: dict[str, dict]) -> Path | None:
|
|
"""Empaqueta y cifra los justificantes que aún no están en el servidor git. Actualiza
|
|
el índice en «ajustes» (y en disco). Devuelve el paquete nuevo, o None."""
|
|
indice = ajustes.get(INDICE) or {}
|
|
en_repo = self._paquetes_en_repo(indice)
|
|
# Un paquete del índice que no está en el repositorio (p. ej. una subida
|
|
# que falló) se olvida: sus justificantes se vuelven a subir.
|
|
subidos = {h for huellas in en_repo.values() for h in huellas}
|
|
nuevos = sorted(h for h in adj.referenciados(datos.movimientos)
|
|
if h not in subidos and adj.tiene(carpeta_datos, h))
|
|
paquete = None
|
|
if nuevos:
|
|
nombre = f"{dt.datetime.now().strftime('%Y%m%d-%H%M%S')}-{uuid.uuid4().hex[:8]}"
|
|
buf = io.BytesIO()
|
|
with tarfile.open(fileobj=buf, mode="w:gz", format=tarfile.PAX_FORMAT) as tar:
|
|
for h in nuevos:
|
|
contenido = adj.leer(carpeta_datos, h)
|
|
info = tarfile.TarInfo(h)
|
|
info.size, info.mtime, info.mode = len(contenido), 0, 0o600
|
|
tar.addfile(info, io.BytesIO(contenido))
|
|
paquete = self.repo / PAQUETES / f"{nombre}.tar.gpg"
|
|
paquete.parent.mkdir(parents=True, exist_ok=True)
|
|
self._cifrar_bytes(buf.getvalue(), paquete)
|
|
en_repo[nombre] = nuevos
|
|
if en_repo != (indice.get("paquetes") or {}):
|
|
ajustes[INDICE] = {"paquetes": en_repo, "modificado": dt.datetime.now().isoformat(timespec="seconds")}
|
|
escribir_ajuste(self.ajustes / INDICE, ajustes[INDICE])
|
|
return paquete
|
|
|
|
# ── acciones ──
|
|
def sincronizar(self, cartera, copia_seguridad=None) -> dict:
|
|
if not self.config().get("huella"):
|
|
raise ErrorSync("La sincronización no está configurada.")
|
|
estado = self._leer_json(self._estado)
|
|
blob = self._preparar_repo()
|
|
remoto, ajustes_remotos, huella_remota = None, {}, None
|
|
if blob is not None:
|
|
if blob == estado.get("blob"):
|
|
huella_remota = estado.get("huella_local") # el servidor git sigue como lo dejamos
|
|
else:
|
|
remoto, ajustes_remotos = self._descifrar()
|
|
huella_remota = huella_total(remoto, ajustes_remotos)
|
|
|
|
with cartera._lock:
|
|
local = cartera.datos
|
|
ajustes_locales = leer_ajustes(self.ajustes)
|
|
if remoto is not None:
|
|
fusionado, recuento = fusionar(local, remoto)
|
|
else:
|
|
fusionado, recuento = local, {}
|
|
ajustes = elegir_ajustes(ajustes_locales, ajustes_remotos)
|
|
cambia_local = huella_datos(fusionado) != huella_datos(local)
|
|
cambian_ajustes = [n for n in AJUSTES
|
|
if _canonico(ajustes.get(n)) != _canonico(ajustes_locales.get(n))]
|
|
if cambia_local or cambian_ajustes:
|
|
if copia_seguridad:
|
|
copia_seguridad()
|
|
if cambia_local:
|
|
cartera.datos = fusionado
|
|
cartera._guardar()
|
|
for nombre in cambian_ajustes:
|
|
escribir_ajuste(self.ajustes / nombre, ajustes[nombre])
|
|
adjuntos = self.traer_adjuntos(cartera.carpeta, fusionado, ajustes.get(INDICE))
|
|
paquete = self._subir_adjuntos(cartera.carpeta, fusionado, ajustes)
|
|
huella_final = huella_total(fusionado, ajustes)
|
|
|
|
subido = False
|
|
if huella_final != huella_remota:
|
|
self._cifrar(fusionado, ajustes, self.repo / CIFRADO)
|
|
self._git("add", CIFRADO)
|
|
if paquete is not None:
|
|
self._git("add", str(paquete.relative_to(self.repo)))
|
|
momento = dt.datetime.now().strftime("%Y-%m-%d %H:%M")
|
|
self._git("commit", "--quiet", "-m", f"Copia cifrada de la cartera ({momento})")
|
|
try:
|
|
self._git("push", "--quiet", "origin", "HEAD:main")
|
|
except ErrorSync as e:
|
|
raise ErrorSync(f"No se pudo subir a tu servidor git ({e}). Vuelve a sincronizar.") from None
|
|
blob = self._git("rev-parse", f"HEAD:{CIFRADO}", timeout=30)
|
|
subido = True
|
|
|
|
self._guardar_json(self._estado, {"blob": blob, "huella_local": huella_final,
|
|
"fecha": dt.datetime.now().isoformat(timespec="seconds")})
|
|
return {"descifrado": remoto is not None, "subido": subido,
|
|
"cambios_locales": cambia_local or bool(cambian_ajustes),
|
|
"recuento": recuento, "ajustes": cambian_ajustes,
|
|
"adjuntos": {**adjuntos, "subidos": len(ajustes.get(INDICE, {}).get("paquetes", {})
|
|
.get(paquete.name.removesuffix(".tar.gpg"), []))
|
|
if paquete else 0}}
|
|
|
|
def comprobar(self, datos: Datos) -> dict:
|
|
"""Simulacro: descifra la copia de tu servidor git y la compara con lo que hay aquí."""
|
|
if not self.config().get("huella"):
|
|
raise ErrorSync("La sincronización no está configurada.")
|
|
if self._preparar_repo() is None:
|
|
return {"hay_copia": False}
|
|
remoto, ajustes_remotos = self._descifrar()
|
|
diferencias = 0
|
|
for nombre in FICHEROS:
|
|
a = {json.dumps(r, sort_keys=True, ensure_ascii=False) for r in getattr(datos, nombre)}
|
|
b = {json.dumps(r, sort_keys=True, ensure_ascii=False) for r in getattr(remoto, nombre)}
|
|
diferencias += len(a ^ b)
|
|
locales = leer_ajustes(self.ajustes)
|
|
ajustes = {n: {"en_la_copia": n in ajustes_remotos,
|
|
"igual": _canonico(ajustes_remotos.get(n)) == _canonico(locales.get(n))}
|
|
for n in AJUSTES if n in locales or n in ajustes_remotos}
|
|
en_copia = {h for hs in self._paquetes_en_repo(ajustes_remotos.get(INDICE)).values() for h in hs}
|
|
sin_copia = len(adj.referenciados(datos.movimientos) - en_copia)
|
|
return {"hay_copia": True, "diferencias": diferencias, "adjuntos_sin_copia": sin_copia,
|
|
"identica": diferencias == 0 and all(a["igual"] for a in ajustes.values()),
|
|
"registros": sum(len(getattr(remoto, n)) for n in FICHEROS),
|
|
"ajustes": ajustes}
|