esetheyers
4b158e662d
Link another Caudal (phone, USB) with a code or a QR instead of the password, choosing the server's root certificate as a file; read Bitcoin straight from Bitcoin Core or Knots, without an Electrum server; Android app (beta) and a Windows build script. Fixed: Save did nothing for expenses, refunds and dividends entered by hand.
458 lines
23 KiB
Python
458 lines
23 KiB
Python
"""
|
|
enlace.py — Sincronizar dos Caudal: el de un servidor y otro que se enlaza con
|
|
él (el del USB). Mezcla, no sustituye: lo que se haya apuntado en cada lado
|
|
acaba en los dos.
|
|
|
|
Quién es quién. El SERVIDOR es el Caudal que atiende en la red y pide
|
|
contraseña. El otro se ENLAZA una vez (hace falta la contraseña del servidor,
|
|
que no se guarda: se cambia por una credencial propia de ese equipo, que se
|
|
puede retirar) y a partir de ahí sincroniza cuando se le pide.
|
|
|
|
Cómo se sincroniza, en un solo viaje:
|
|
|
|
1. El enlazado envía TODO lo suyo: los registros (también los borrados, que
|
|
son marcas) y los ajustes.
|
|
2. El servidor lo comprueba entero antes de tocar nada, lo mezcla con lo
|
|
suyo y responde con el resultado, que es lo que queda en los dos.
|
|
3. El enlazado comprueba la respuesta y la pone en lugar de lo suyo.
|
|
4. Después se intercambian los justificantes que le falten a cada uno.
|
|
|
|
Para que mezclar no pueda romper nada:
|
|
|
|
· Nada se pierde: la mezcla es la unión de las dos copias; si un registro
|
|
está en las dos con contenido distinto gana el más reciente, y un borrado
|
|
es una marca, no una ausencia (datos.fusionar).
|
|
· Dos inversiones distintas con el mismo identificador no se pisan: a la
|
|
que llega se le da otro y sus operaciones la siguen (datos.choques_de_activos).
|
|
· La misma línea de extracto apuntada en los dos sitios no queda duplicada
|
|
(datos.unir_duplicadas).
|
|
· Antes de aplicar, cada lado guarda una copia local de lo que tenía.
|
|
· Si las dos copias son de versiones distintas del formato, no se mezclan.
|
|
· Se puede simular: dice qué cambiaría en cada lado y no cambia nada. Si las
|
|
dos copias no tienen nada en común, se avisa: serían dos carteras juntas.
|
|
· El enlazado no aplica una respuesta en la que falte algo de lo que envió.
|
|
|
|
Si el viaje se corta a medias no pasa nada: repetirlo lleva al mismo sitio.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import datetime as dt
|
|
import http.client
|
|
import json
|
|
import os
|
|
import socket
|
|
import ssl
|
|
import tempfile
|
|
from pathlib import Path
|
|
from urllib.parse import urlsplit
|
|
|
|
from . import adjuntos, almacen
|
|
from .datos import (FICHEROS, VERSION, Datos, ErrorDatos, choques_de_activos, clave, escribir, fusionar, huella_datos,
|
|
renombrar_activos, unir_duplicadas)
|
|
from .sincronizar import _canonico, elegir_ajustes, escribir_ajuste, leer_ajustes
|
|
|
|
AJUSTES = ("bitcoin.json", "extractos.json") # los que viajan; el índice de la copia en git no es de aquí
|
|
FICHERO = "enlace.json"
|
|
FICHERO_CERTIFICADO = "enlace-certificado.crt" # el certificado raíz del servidor, si se eligió como fichero
|
|
TIEMPO = 120
|
|
TIEMPO_CONECTAR = 8 # si en este rato no hay camino, se prueba el siguiente
|
|
TIEMPO_TOR = 60 # por Tor, abrir el camino tarda
|
|
|
|
|
|
class ErrorEnlace(Exception):
|
|
"""Lo que se le dice al usuario tal cual."""
|
|
|
|
|
|
class SinCamino(ErrorEnlace):
|
|
"""No se llega al servidor por esa dirección (quizá sí por la otra)."""
|
|
|
|
|
|
def es_onion(url: str) -> bool:
|
|
return (urlsplit(url).hostname or "").endswith(".onion")
|
|
|
|
|
|
class _Respuesta:
|
|
def __init__(self, estado: int, cuerpo: bytes):
|
|
self.status_code, self._cuerpo = estado, cuerpo
|
|
|
|
def json(self):
|
|
return json.loads(self._cuerpo)
|
|
|
|
|
|
class _SesionTor:
|
|
"""Lo justo de «requests.Session» para llegar a una dirección .onion por el
|
|
Tor de este equipo, sin instalar nada más (el SOCKS5 es el de electrum.py)."""
|
|
|
|
def __init__(self, proxy: tuple[str, int], certificado: str = ""):
|
|
self.proxy, self.certificado = proxy, certificado
|
|
self.cookies: dict[str, str] = {}
|
|
|
|
def post(self, url: str, json=None, headers=None, timeout=TIEMPO):
|
|
from .electrum import ErrorElectrum, _socks5
|
|
import json as _json
|
|
u = urlsplit(url)
|
|
puerto = u.port or (443 if u.scheme == "https" else 80)
|
|
try:
|
|
sock = _socks5(u.hostname, puerto, self.proxy[0], self.proxy[1], TIEMPO_TOR)
|
|
except ErrorElectrum as e:
|
|
raise SinCamino(f"No llego al servidor por Tor: {e}.") from None
|
|
try:
|
|
if u.scheme == "https":
|
|
# Por .onion es Tor quien garantiza con quién se habla; si además se ha dado el
|
|
# certificado raíz del servidor, se comprueba también la firma (no el nombre)
|
|
contexto = ssl.create_default_context(cafile=self.certificado) if self.certificado \
|
|
else ssl._create_unverified_context()
|
|
contexto.check_hostname = False
|
|
try:
|
|
# El nombre va en el saludo (SNI) aunque no se compruebe: un servidor que reparte
|
|
# varios servicios por el mismo puerto, como StartOS, lo necesita para saber cuál es
|
|
sock = contexto.wrap_socket(sock, server_hostname=u.hostname)
|
|
except ssl.SSLError:
|
|
raise ErrorEnlace("El certificado que presenta el servidor por Tor no es de la autoridad que indicaste.") from None
|
|
sock.settimeout(timeout if isinstance(timeout, (int, float)) else TIEMPO)
|
|
conexion = http.client.HTTPConnection(u.hostname, puerto)
|
|
conexion.sock = sock
|
|
cabeceras = {"Host": u.netloc, "Content-Type": "application/json", "Connection": "close", **(headers or {})}
|
|
if self.cookies:
|
|
cabeceras["Cookie"] = "; ".join(f"{k}={v}" for k, v in self.cookies.items())
|
|
conexion.request("POST", u.path or "/", _json.dumps(json or {}).encode(), cabeceras)
|
|
r = conexion.getresponse()
|
|
for galleta in r.headers.get_all("Set-Cookie") or []:
|
|
nombre, _, valor = galleta.split(";")[0].partition("=")
|
|
self.cookies[nombre.strip()] = valor.strip()
|
|
return _Respuesta(r.status, r.read())
|
|
except (OSError, http.client.HTTPException) as e:
|
|
raise SinCamino(f"Se ha cortado la conexión por Tor ({e}).") from None
|
|
finally:
|
|
sock.close()
|
|
|
|
|
|
NO_ENTIENDO = "No entiendo esa petición."
|
|
|
|
|
|
def _ahora() -> str:
|
|
return dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ")
|
|
|
|
|
|
# ── lo que viaja ──
|
|
|
|
def empaquetar(datos: Datos, ajustes: dict) -> dict:
|
|
return {"version": VERSION, "datos": {n: getattr(datos, n) for n in FICHEROS},
|
|
"ajustes": {n: ajustes[n] for n in AJUSTES if isinstance(ajustes.get(n), dict)}}
|
|
|
|
|
|
def desempaquetar(paquete: dict) -> tuple[Datos, dict]:
|
|
"""(datos, ajustes) de lo recibido, comprobado entero: si algo no vale, no vale nada."""
|
|
if not isinstance(paquete, dict) or not isinstance(paquete.get("datos"), dict):
|
|
raise ErrorEnlace("Lo recibido no es una copia de Caudal.")
|
|
if paquete.get("version") != VERSION:
|
|
raise ErrorEnlace("Los dos Caudal no son de la misma versión: actualiza el más antiguo y vuelve a sincronizar.")
|
|
tablas = {}
|
|
for nombre in FICHEROS:
|
|
registros = paquete["datos"].get(nombre, [])
|
|
if not isinstance(registros, list) or not all(isinstance(r, dict) for r in registros):
|
|
raise ErrorEnlace("Lo recibido no es una copia de Caudal.")
|
|
claves = [clave(nombre, r) for r in registros if (r.get("id") or r.get("activo"))]
|
|
if len(claves) != len(registros) or len(set(claves)) != len(claves):
|
|
raise ErrorEnlace("Lo recibido trae registros repetidos o sin identificar: no se mezcla.")
|
|
tablas[nombre] = registros
|
|
datos = Datos(**tablas)
|
|
with tempfile.TemporaryDirectory() as tmp: # la misma comprobación que al guardar
|
|
try:
|
|
escribir(Path(tmp), datos)
|
|
except ErrorDatos as e:
|
|
raise ErrorEnlace(f"Lo recibido tiene un dato que no vale ({e}): no se mezcla.") from None
|
|
ajustes = paquete.get("ajustes") if isinstance(paquete.get("ajustes"), dict) else {}
|
|
return datos, {n: ajustes[n] for n in AJUSTES if isinstance(ajustes.get(n), dict)}
|
|
|
|
|
|
def _ajustes_renombrados(ajustes: dict, mapa: dict[str, str]) -> dict:
|
|
"""Los ajustes que nombran inversiones (monederos, cuentas recordadas), con los identificadores nuevos."""
|
|
if not mapa:
|
|
return ajustes
|
|
salida = json.loads(json.dumps(ajustes))
|
|
for m in (salida.get("bitcoin.json") or {}).get("monederos") or []:
|
|
if m.get("activo") in mapa:
|
|
m["activo"] = mapa[m["activo"]]
|
|
ex = salida.get("extractos.json") or {}
|
|
for campo in ("cuenta_intereses", "destinos_exchange"):
|
|
if isinstance(ex.get(campo), dict):
|
|
ex[campo] = {k: mapa.get(v, v) for k, v in ex[campo].items()}
|
|
return salida
|
|
|
|
|
|
def mezclar(mios: Datos, mis_ajustes: dict, suyos: Datos, sus_ajustes: dict, prefijos: dict) -> dict:
|
|
"""La mezcla de dos copias, sin tocar ninguna. Devuelve los datos y ajustes
|
|
que deben quedar en las dos y el recuento de lo que cambia en cada lado."""
|
|
mapa = choques_de_activos(mios, suyos, prefijos)
|
|
suyos = renombrar_activos(suyos, mapa)
|
|
sus_ajustes = _ajustes_renombrados(sus_ajustes, mapa)
|
|
en_comun = sum(len({clave(n, r) for r in getattr(mios, n)} & {clave(n, r) for r in getattr(suyos, n)})
|
|
for n in ("activos", "movimientos", "recurrentes"))
|
|
fusionado, recuento = fusionar(mios, suyos)
|
|
fusionado = json.loads(json.dumps({n: getattr(fusionado, n) for n in FICHEROS})) # copia: no tocar los originales
|
|
fusionado = Datos(**fusionado)
|
|
retiradas = unir_duplicadas(fusionado, _ahora())
|
|
locales = {n: mis_ajustes.get(n) for n in AJUSTES if mis_ajustes.get(n) is not None}
|
|
ajustes = {n: v for n, v in elegir_ajustes(locales, sus_ajustes).items() if n in AJUSTES}
|
|
vivos = lambda d: len(d.vivos("activos")) + len(d.vivos("movimientos")) # noqa: E731
|
|
return {
|
|
"datos": fusionado, "ajustes": ajustes,
|
|
"resumen": {
|
|
# «aquí» es quien mezcla (el servidor); «alli», quien ha enviado lo suyo
|
|
"llegan_aqui": recuento["de_remoto"], "cambian_aqui": recuento["actualizados_desde_remoto"],
|
|
"llegan_alli": recuento["de_local"], "cambian_alli": recuento["actualizados_desde_local"],
|
|
"ajustes_aqui": [n for n in AJUSTES if _canonico(ajustes.get(n)) != _canonico(mis_ajustes.get(n))],
|
|
"ajustes_alli": [n for n in AJUSTES if _canonico(ajustes.get(n)) != _canonico(sus_ajustes.get(n))],
|
|
"renombradas": [{"de": k, "a": v, "nombre": next((a.get("nombre", "") for a in suyos.activos if a["id"] == v), "")}
|
|
for k, v in mapa.items()],
|
|
"duplicadas": len(retiradas), "en_comun": en_comun,
|
|
"carteras_distintas": en_comun == 0 and vivos(mios) > 0 and vivos(suyos) > 0,
|
|
},
|
|
}
|
|
|
|
|
|
def justificantes(carpeta: Path, datos: Datos) -> tuple[list[str], list[str]]:
|
|
"""(los que hay en este equipo, los que nombran los datos y faltan aquí)."""
|
|
nombrados = sorted(adjuntos.referenciados(datos.movimientos))
|
|
tengo = [h for h in nombrados if adjuntos.tiene(carpeta, h)]
|
|
return tengo, [h for h in nombrados if h not in tengo]
|
|
|
|
|
|
# ── el lado que se enlaza (el del USB) ──
|
|
|
|
class Enlace:
|
|
"""La dirección del servidor y la credencial de este equipo, junto a los datos."""
|
|
|
|
def __init__(self, carpeta_datos: Path):
|
|
self.ruta = Path(carpeta_datos).resolve().parent / FICHERO
|
|
|
|
def config(self) -> dict:
|
|
try:
|
|
return json.loads(almacen.leer_texto(self.ruta))
|
|
except (OSError, ValueError):
|
|
return {}
|
|
|
|
def guardar(self, config: dict) -> None:
|
|
almacen.escribir(self.ruta, json.dumps(config, ensure_ascii=False, indent=1).encode())
|
|
|
|
def olvidar(self) -> None:
|
|
self.ruta.unlink(missing_ok=True)
|
|
(self.ruta.parent / FICHERO_CERTIFICADO).unlink(missing_ok=True)
|
|
|
|
def guardar_certificado(self, pem: str) -> str:
|
|
"""Guarda junto a los datos el certificado raíz que se ha elegido como fichero (en el
|
|
teléfono no hay una ruta que escribir) y devuelve dónde ha quedado."""
|
|
pem = (pem or "").strip()
|
|
if len(pem) > 50_000 or "-----BEGIN CERTIFICATE-----" not in pem:
|
|
raise ErrorEnlace("Ese fichero no es un certificado (se espera un .crt o .pem, el que descargaste del servidor).")
|
|
try:
|
|
ssl.create_default_context(cadata=pem)
|
|
except (ssl.SSLError, ValueError):
|
|
raise ErrorEnlace("No puedo leer ese certificado: ¿es el certificado raíz que descargaste del servidor?") from None
|
|
destino = self.ruta.parent / FICHERO_CERTIFICADO
|
|
tmp = destino.with_name(destino.name + ".tmp")
|
|
tmp.write_text(pem + "\n", encoding="ascii", errors="replace")
|
|
os.chmod(tmp, 0o600)
|
|
os.replace(tmp, destino)
|
|
return str(destino)
|
|
|
|
def _proxy(self) -> tuple[str, int]:
|
|
"""El Tor de este equipo: el mismo que se usa para llegar al nodo de bitcoin."""
|
|
from . import btc
|
|
return btc.proxy_de(btc.cargar(self.ruta.parent / "datos"))
|
|
|
|
def _sesion(self, url: str):
|
|
c = self.config()
|
|
if es_onion(url):
|
|
return _SesionTor(self._proxy(), c.get("certificado") or "")
|
|
import requests
|
|
s = requests.Session()
|
|
s.verify = c.get("certificado") or True
|
|
return s
|
|
|
|
def caminos(self) -> list[str]:
|
|
"""Las direcciones del servidor, en el orden en que se prueban: la de
|
|
casa y, si la hay, la .onion para cuando no se está en casa."""
|
|
c = self.config()
|
|
return [u.rstrip("/") for u in (c.get("url"), c.get("url_tor")) if u]
|
|
|
|
def _pedir(self, sesion, url: str, cuerpo: dict, cabeceras: dict):
|
|
import requests
|
|
try:
|
|
r = sesion.post(url, json=cuerpo, headers=cabeceras, timeout=(TIEMPO_CONECTAR, TIEMPO))
|
|
except requests.exceptions.SSLError:
|
|
raise ErrorEnlace("No puedo comprobar el certificado del servidor. Si tu servidor usa su propia autoridad "
|
|
"(como StartOS), descarga su certificado raíz e indica aquí dónde está.") from None
|
|
except requests.exceptions.RequestException:
|
|
raise SinCamino("No llego al servidor: comprueba la dirección y que haya red hasta él.") from None
|
|
try:
|
|
datos = r.json()
|
|
except ValueError:
|
|
raise ErrorEnlace("Esa dirección contesta, pero no parece un Caudal.") from None
|
|
if r.status_code != 200:
|
|
raise ErrorEnlace(datos.get("error") or f"El servidor ha contestado con un error ({r.status_code}).")
|
|
return datos
|
|
|
|
def enlazar(self, url: str, clave_servidor: str, certificado: str = "", nombre: str = "", codigo: str = "",
|
|
certificado_pem: str = "") -> dict:
|
|
"""Entra una vez con la contraseña del servidor y la cambia por una credencial de este equipo.
|
|
Con un código de enlace (el que enseña el servidor) no hace falta la contraseña. El certificado
|
|
raíz llega como ruta a un fichero o, elegido en la página, con su contenido."""
|
|
url = (url or "").strip().rstrip("/")
|
|
if not url.startswith(("http://", "https://")):
|
|
raise ErrorEnlace("La dirección tiene que empezar por https:// (o http:// si es dentro de tu casa o una .onion).")
|
|
if certificado_pem:
|
|
certificado = self.guardar_certificado(certificado_pem)
|
|
if certificado and not Path(certificado).is_file():
|
|
raise ErrorEnlace("No encuentro ese fichero de certificado.")
|
|
self.guardar({"url": url, "certificado": certificado or ""})
|
|
try:
|
|
sesion, base = self._sesion(url), url
|
|
if codigo.strip():
|
|
credencial = self._pedir(sesion, f"{base}/api/enlace/canje", {"codigo": codigo, "nombre": nombre or "Otro Caudal"},
|
|
{"X-Caudal": "1"})["credencial"]
|
|
else:
|
|
entrada = self._pedir(sesion, f"{base}/api/acceso/entrar", {"clave": clave_servidor}, {"X-Caudal": "1"})
|
|
alta = self._pedir(sesion, f"{base}/api/enlace/alta", {"nombre": nombre or "Otro Caudal"}, {"X-Token": entrada["token"]})
|
|
self._pedir(sesion, f"{base}/api/acceso/salir", {}, {"X-Caudal": "1"})
|
|
credencial = alta["resultado"]["credencial"]
|
|
except ErrorEnlace:
|
|
self.olvidar()
|
|
raise
|
|
self.guardar({"url": url, "certificado": certificado or "", "credencial": credencial,
|
|
"enlazado": _ahora()})
|
|
return {"url": url}
|
|
|
|
def llamar(self, orden: str, cuerpo: dict) -> dict:
|
|
c = self.config()
|
|
if not c.get("credencial"):
|
|
raise ErrorEnlace("Este Caudal no está enlazado con ningún servidor.")
|
|
return self._por_algun_camino(self.caminos(), orden, cuerpo, c["credencial"])
|
|
|
|
def _por_algun_camino(self, caminos: list[str], orden: str, cuerpo: dict, credencial: str) -> dict:
|
|
fallo = None
|
|
for base in caminos:
|
|
try:
|
|
return self._pedir(self._sesion(base), f"{base}/api/enlace", {"orden": orden, **cuerpo},
|
|
{"X-Enlace": credencial})
|
|
except SinCamino as e:
|
|
fallo = e
|
|
if len(caminos) > 1:
|
|
raise SinCamino(f"No llego al servidor ni por la dirección de casa ni por la .onion. {fallo}")
|
|
raise fallo or ErrorEnlace("Este Caudal no está enlazado con ningún servidor.")
|
|
|
|
def poner_onion(self, url_tor: str) -> dict:
|
|
"""Añade (o quita, con «») la dirección .onion del servidor, para sincronizar
|
|
fuera de casa. Antes de guardarla comprueba que por ella se llega al mismo
|
|
servidor y que acepta la credencial de este equipo."""
|
|
c = self.config()
|
|
if not c.get("credencial"):
|
|
raise ErrorEnlace("Este Caudal no está enlazado con ningún servidor.")
|
|
url_tor = (url_tor or "").strip().rstrip("/")
|
|
if not url_tor:
|
|
self.guardar({k: v for k, v in c.items() if k != "url_tor"})
|
|
return {"url_tor": ""}
|
|
if "://" not in url_tor:
|
|
url_tor = "http://" + url_tor
|
|
if not url_tor.startswith(("http://", "https://")) or not es_onion(url_tor):
|
|
raise ErrorEnlace("Tiene que ser una dirección .onion (la que da tu servidor para Caudal).")
|
|
try:
|
|
self._por_algun_camino([url_tor], "hola", {}, c["credencial"])
|
|
except SinCamino:
|
|
raise
|
|
except ErrorEnlace as e:
|
|
# Un Caudal anterior a esta orden contesta que no la entiende: vale igual,
|
|
# porque para contestar eso ya ha dado por buena la credencial
|
|
if str(e) != NO_ENTIENDO:
|
|
raise
|
|
self.guardar({**c, "url_tor": url_tor})
|
|
return {"url_tor": url_tor}
|
|
|
|
|
|
def sincronizar(app, simular: bool) -> dict:
|
|
"""Desde el Caudal enlazado: envía lo suyo, recibe la mezcla y la aplica (salvo que sea un simulacro)."""
|
|
enlace = Enlace(app.carpeta)
|
|
carpeta = app.cartera.carpeta
|
|
with app.cartera._lock:
|
|
mios = app.cartera.datos
|
|
mis_ajustes = leer_ajustes(carpeta.parent)
|
|
tengo, _ = justificantes(carpeta, mios)
|
|
respuesta = enlace.llamar("simular" if simular else "mezclar", {**empaquetar(mios, mis_ajustes), "justificantes": tengo})
|
|
resumen = respuesta["resumen"]
|
|
if simular:
|
|
return {"simulacro": True, "resumen": resumen}
|
|
fusionado, ajustes = desempaquetar(respuesta)
|
|
# Lo que vuelve tiene que contener todo lo que se envió (salvo lo que el servidor renombró)
|
|
renombradas = {x["de"] for x in resumen["renombradas"]}
|
|
for nombre in FICHEROS:
|
|
vuelven = {clave(nombre, r) for r in getattr(fusionado, nombre)}
|
|
faltan = [k for k in (clave(nombre, r) for r in getattr(mios, nombre))
|
|
if k not in vuelven and k.split("|")[0] not in renombradas]
|
|
if nombre in ("movimientos", "recurrentes") and faltan:
|
|
raise ErrorEnlace("El servidor ha devuelto menos de lo que le envié: no aplico nada. Vuelve a intentarlo.")
|
|
cambia = huella_datos(fusionado) != huella_datos(mios)
|
|
cambian_ajustes = [n for n in AJUSTES if _canonico(ajustes.get(n)) != _canonico(mis_ajustes.get(n))]
|
|
if cambia or cambian_ajustes:
|
|
app._copia_datos() # lo que había, a la carpeta de copias locales
|
|
if cambia:
|
|
app.cartera.reemplazar(fusionado)
|
|
for nombre in cambian_ajustes:
|
|
escribir_ajuste(carpeta.parent / nombre, ajustes[nombre])
|
|
# Los justificantes, después y de uno en uno: si alguno falla, el siguiente sincronizar lo reintenta
|
|
subidos = bajados = 0
|
|
for h in respuesta.get("le_faltan") or []:
|
|
if adjuntos.tiene(carpeta, h):
|
|
enlace.llamar("subir", {"contenido": base64.b64encode(adjuntos.leer(carpeta, h)).decode()})
|
|
subidos += 1
|
|
for h in justificantes(carpeta, app.cartera.datos)[1]:
|
|
if h in (respuesta.get("tiene") or []):
|
|
contenido = base64.b64decode(enlace.llamar("bajar", {"huella": h})["contenido"])
|
|
if adjuntos.guardar(carpeta, contenido) == h:
|
|
bajados += 1
|
|
config = enlace.config()
|
|
enlace.guardar({**config, "ultima": _ahora()})
|
|
return {"simulacro": False, "resumen": resumen, "justificantes": {"subidos": subidos, "bajados": bajados}}
|
|
|
|
|
|
# ── el lado del servidor ──
|
|
|
|
def atender(app, orden: str, cuerpo: dict, prefijos: dict) -> dict:
|
|
"""Lo que pide un Caudal enlazado. Quien llama ya ha comprobado su credencial."""
|
|
carpeta = app.cartera.carpeta
|
|
if orden == "subir":
|
|
try:
|
|
return {"huella": adjuntos.guardar(carpeta, base64.b64decode(str(cuerpo.get("contenido") or ""), validate=True))}
|
|
except (ValueError, adjuntos.ErrorAdjunto) as e:
|
|
raise ErrorEnlace(f"Ese justificante no se puede guardar: {e}") from None
|
|
if orden == "bajar":
|
|
try:
|
|
return {"contenido": base64.b64encode(adjuntos.leer(carpeta, str(cuerpo.get("huella") or ""))).decode()}
|
|
except adjuntos.ErrorAdjunto as e:
|
|
raise ErrorEnlace(str(e)) from None
|
|
if orden == "hola": # para probar un camino sin mover nada
|
|
return {"hola": True}
|
|
if orden not in ("simular", "mezclar"):
|
|
raise ErrorEnlace(NO_ENTIENDO)
|
|
suyos, sus_ajustes = desempaquetar(cuerpo)
|
|
with app.cartera._lock:
|
|
mios = app.cartera.datos
|
|
mis_ajustes = leer_ajustes(carpeta.parent)
|
|
m = mezclar(mios, mis_ajustes, suyos, sus_ajustes, prefijos)
|
|
if orden == "simular":
|
|
return {"resumen": m["resumen"]}
|
|
r = m["resumen"]
|
|
# Se compara el resultado entero con lo que había, no los recuentos: una inversión puede «ganar»
|
|
# y aun así salir distinta de la mezcla (con los lugares que se añadieron en el otro lado)
|
|
cambia = huella_datos(m["datos"]) != huella_datos(mios)
|
|
if cambia or r["ajustes_aqui"]:
|
|
app._copia_datos()
|
|
if cambia:
|
|
app.cartera.reemplazar(m["datos"])
|
|
for nombre in r["ajustes_aqui"]:
|
|
escribir_ajuste(carpeta.parent / nombre, m["ajustes"][nombre])
|
|
tengo, me_faltan = justificantes(carpeta, m["datos"])
|
|
return {**empaquetar(m["datos"], m["ajustes"]), "resumen": r, "tiene": tengo,
|
|
"le_faltan": [h for h in me_faltan if h in (cuerpo.get("justificantes") or [])]}
|