539 lines
26 KiB
Python
539 lines
26 KiB
Python
"""
|
|
datos.py — Los datos de la cartera en texto plano.
|
|
|
|
Ficheros JSON Lines (un objeto JSON por línea) en la carpeta de datos:
|
|
|
|
activos.jsonl un activo por línea
|
|
movimientos.jsonl una operación por línea
|
|
precios.jsonl un precio por línea (activo + fecha)
|
|
recurrentes.jsonl operaciones automáticas (hosting mensual, aportación al plan...)
|
|
version.json versión del formato: {"version": N} (si falta, es la 1)
|
|
|
|
Por qué así:
|
|
• Se leen con cualquier editor y git muestra cambios línea a línea.
|
|
• Cada registro tiene clave propia (id, o activo+fecha en precios) y fecha
|
|
de "modificado", así que juntar dos copias (USB y ordenador) es unir
|
|
líneas y, si una clave está en las dos, quedarse con la más reciente.
|
|
Un borrado se guarda como "borrado": true para que la fusión no lo resucite.
|
|
• Los números se guardan como JSON (punto decimal, precisión completa):
|
|
no hay líos de coma/punto ni de redondeo.
|
|
|
|
Formato de cada registro (los campos opcionales pueden faltar o ser null):
|
|
|
|
activo: id, nombre, tipo, codigo, unidad, fecha_inicio, visible, orden,
|
|
valoracion ("unitaria" por defecto, o "total": solo se sigue el
|
|
valor de la posición, sin participaciones),
|
|
estado ("abierta" | "cerrada"), fecha_cierre, alias (otros nombres para el asistente),
|
|
prestamo {importe_final, vencimiento} (solo tipo Prestamo: dinero que prestas),
|
|
deuda {capital, tin, cuotas, primera_cuota} (solo tipo Deuda: préstamo
|
|
que te han concedido, cuota constante; ver deuda.py),
|
|
lugares [{id, nombre, clase}] (dónde lo tienes: exchanges,
|
|
monederos, nodos Lightning; clase monedero|exchange|lightning|otro),
|
|
modificado, borrado
|
|
movimiento: id, fecha, activo, tipo, cantidad, precio, importe, comisiones,
|
|
coste_traspaso, notas, orden, recurrente, estimado, pendiente,
|
|
importe_original, moneda, actividad, extracto, adjuntos, txid,
|
|
importacion, lugar, lugar_destino, traslado, modificado, borrado
|
|
· lugar: id de uno de los lugares del activo (dónde ocurrió).
|
|
· Traslado: mover cantidad entre dos lugares tuyos del MISMO
|
|
activo (de un exchange a tu monedero). Una sola línea: lugar = origen,
|
|
lugar_destino = destino ("" = aún sin saber), cantidad = lo que
|
|
llega. NO cambia ninguna cifra de la cartera ni es fiscal: el
|
|
cálculo no lo mira. Su comisión es un «Comision» aparte, en el
|
|
lugar de origen, con traslado = id del traslado.
|
|
· extracto: de dónde viene la operación («btc:<txid>» si la trajo
|
|
el nodo de bitcoin, o la línea del extracto importado).
|
|
· txid: transacción de bitcoin ENLAZADA a una operación que ya
|
|
estaba apuntada (al importar el historial; no cambia cifras).
|
|
· importacion: marca de la importación del historial de bitcoin
|
|
que la creó, para poder deshacerla entera (historial_btc.py).
|
|
· adjuntos: justificantes, [{id, nombre, tipo, bytes}]; el
|
|
fichero está en adjuntos/<id> (ver adjuntos.py).
|
|
· actividad: nombre libre («Minería») que agrupa operaciones de
|
|
cualquier activo para poder ver su rentabilidad por separado
|
|
(ver actividades.py). El cálculo general NO la mira: marcar
|
|
operaciones no cambia ninguna cifra de la cartera.
|
|
· Gasto: euros sin cantidad (p. ej. hosting de mineros); cuenta
|
|
como dinero invertido en el activo.
|
|
· amortizable: en un Gasto o Devolucion, que es la compra de un
|
|
aparato de la actividad (un minero) y no un gasto del
|
|
mes. Solo lo mira el cálculo fiscal de la actividad
|
|
(fiscal/actividad.py): se resta poco a poco.
|
|
· Devolucion: euros que te devuelven de un gasto ya apuntado;
|
|
RESTAN dinero invertido (bajan el coste del activo),
|
|
no son un cobro. Espejo de Gasto.
|
|
· Mineria: cantidad recibida sin pagarla (BTC minado).
|
|
· Recompensa: acciones/participaciones recibidas gratis (bonificaciones del bróker, regalos); valor_recibido informativo.
|
|
· Dividendo: dinero cobrado en cuenta sin vender (dividendos,
|
|
cupones, intereses); importe neto y retención opcional.
|
|
· Amortizacion: pago anticipado de una Deuda (importe) con
|
|
modalidad "plazo" o "cuota".
|
|
recurrente: id, activo, tipo (Gasto | Compra), importe, moneda (EUR | USD),
|
|
dia (1-28), desde, hasta, pausado, notas, actividad, modificado,
|
|
borrado (las operaciones que genera heredan su actividad)
|
|
precio: fecha, activo, precio | valor_total, fuente, modificado, borrado
|
|
(precio null = "sin precio ese día"; valor_total = valoración
|
|
total en €, se usa en cuentas de trading)
|
|
|
|
Cambios de formato (migraciones):
|
|
Cuando una versión de la app necesite guardar los datos de otra forma, sube
|
|
VERSION y añade a MIGRACIONES la función que pasa de la versión anterior a
|
|
la nueva. Al leer datos antiguos se migran en memoria, paso a paso, antes de
|
|
validarlos; la primera vez que se escriben se guarda antes una copia de los
|
|
ficheros tal como estaban (carpeta «copia_vN» dentro de la de datos).
|
|
Lo mismo vale para la copia cifrada que llega del servidor git: se migra al leerla,
|
|
antes de fusionarla. Datos de una versión MÁS NUEVA que la app no se abren:
|
|
se para con un mensaje claro en vez de estropearlos.
|
|
|
|
Reglas para escribir una migración:
|
|
• Recibe {"activos": [...], "movimientos": [...], ...} y lo cambia en el sitio.
|
|
• Debe poder aplicarse dos veces sin daño (idempotente): version.json se
|
|
escribe el último, y si algo corta la escritura a medias se vuelve a migrar.
|
|
• Nunca se cambia una migración ya publicada: se añade otra.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import tempfile
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
|
|
from . import almacen
|
|
|
|
TIPOS_ACTIVO = ("Fondo", "PlanPensiones", "Cripto", "Metal", "ETF", "Accion",
|
|
"CuentaTrading", "Prestamo", "Tierra", "Cuenta", "Deuda")
|
|
TIPOS_LIQUIDEZ = ("Cuenta",) # dinero parado: no entra en la rentabilidad
|
|
TIPOS_DEUDA = ("Deuda",) # lo que debes: resta del patrimonio, no es inversión
|
|
NO_INVERSION = TIPOS_LIQUIDEZ + TIPOS_DEUDA
|
|
TIPOS_MOVIMIENTO = ("Compra", "Venta", "Comision", "Traspaso_Entrada",
|
|
"Traspaso_Salida", "Reembolso", "Gasto", "Devolucion", "Mineria", "Dividendo", "Recompensa",
|
|
"Amortizacion", "Traslado")
|
|
CLASES_LUGAR = ("monedero", "exchange", "lightning", "otro")
|
|
TIPOS_RECURRENTE = ("Gasto", "Compra")
|
|
MONEDAS = ("EUR", "USD")
|
|
|
|
FICHEROS = ("activos", "movimientos", "precios", "recurrentes")
|
|
HUELLA_ADJUNTO = re.compile(r"^[0-9a-f]{64}$")
|
|
FICHERO_VERSION = "version.json"
|
|
|
|
# Versión del formato de los datos que escribe esta app, y las migraciones:
|
|
# MIGRACIONES[n] pasa los datos de la versión n-1 a la n (ver el principio).
|
|
VERSION = 2
|
|
|
|
|
|
def _v2_lugares(tablas: dict) -> None:
|
|
"""Versión 2: lugares en los activos y el tipo «Traslado». Todo es nuevo y
|
|
opcional, no hay nada que convertir; la versión sube para que una app
|
|
anterior no abra datos con traslados (no los entendería)."""
|
|
|
|
|
|
MIGRACIONES: dict = {2: _v2_lugares}
|
|
|
|
|
|
class ErrorDatos(ValueError):
|
|
"""Datos mal formados; el mensaje indica fichero y línea."""
|
|
|
|
|
|
@dataclass
|
|
class Datos:
|
|
activos: list[dict] = field(default_factory=list)
|
|
movimientos: list[dict] = field(default_factory=list)
|
|
precios: list[dict] = field(default_factory=list)
|
|
recurrentes: list[dict] = field(default_factory=list)
|
|
|
|
def vivos(self, nombre: str) -> list[dict]:
|
|
"""Registros no borrados."""
|
|
return [r for r in getattr(self, nombre) if not r.get("borrado")]
|
|
|
|
|
|
# ── Validación ────────────────────────────────────────────────────────────────
|
|
|
|
def _fecha(v, donde: str, obligatoria: bool = True) -> None:
|
|
if v is None and not obligatoria:
|
|
return
|
|
if not isinstance(v, str):
|
|
raise ErrorDatos(f"{donde}: fecha ausente o no es texto AAAA-MM-DD")
|
|
try:
|
|
dt.date.fromisoformat(v)
|
|
except ValueError:
|
|
raise ErrorDatos(f"{donde}: fecha inválida {v!r} (se espera AAAA-MM-DD)") from None
|
|
|
|
|
|
def _numero(v, donde: str, campo: str) -> None:
|
|
if v is not None and (isinstance(v, bool) or not isinstance(v, (int, float))):
|
|
raise ErrorDatos(f"{donde}: «{campo}» debe ser un número o null")
|
|
|
|
|
|
def _validar(nombre: str, r: dict, donde: str) -> None:
|
|
if not isinstance(r, dict):
|
|
raise ErrorDatos(f"{donde}: cada línea debe ser un objeto JSON")
|
|
if r.get("borrado"):
|
|
return
|
|
if nombre == "activos":
|
|
if not r.get("id") or not isinstance(r["id"], str):
|
|
raise ErrorDatos(f"{donde}: falta «id»")
|
|
if r.get("tipo") not in TIPOS_ACTIVO:
|
|
raise ErrorDatos(f"{donde}: tipo de activo desconocido {r.get('tipo')!r}")
|
|
_fecha(r.get("fecha_inicio"), donde, obligatoria=False)
|
|
if r.get("valoracion", "unitaria") not in ("unitaria", "total"):
|
|
raise ErrorDatos(f"{donde}: valoracion debe ser «unitaria» o «total»")
|
|
if r.get("estado", "abierta") not in ("abierta", "cerrada"):
|
|
raise ErrorDatos(f"{donde}: estado debe ser «abierta» o «cerrada»")
|
|
_fecha(r.get("fecha_cierre"), donde, obligatoria=r.get("estado") == "cerrada")
|
|
p = r.get("prestamo")
|
|
if p is not None:
|
|
_numero(p.get("importe_final"), donde, "prestamo.importe_final")
|
|
_fecha(p.get("vencimiento"), donde)
|
|
lugares = r.get("lugares")
|
|
if lugares is not None:
|
|
if not isinstance(lugares, list) or len(lugares) > 200:
|
|
raise ErrorDatos(f"{donde}: «lugares» debe ser una lista (máximo 200)")
|
|
ids = set()
|
|
for x in lugares:
|
|
if not isinstance(x, dict) or not isinstance(x.get("id"), str) or not x["id"] \
|
|
or not isinstance(x.get("nombre"), str) or x.get("clase") not in CLASES_LUGAR:
|
|
raise ErrorDatos(f"{donde}: lugar mal formado")
|
|
if x["id"] in ids:
|
|
raise ErrorDatos(f"{donde}: lugar repetido {x['id']!r}")
|
|
ids.add(x["id"])
|
|
d = r.get("deuda")
|
|
if r.get("tipo") == "Deuda" or d is not None:
|
|
if not isinstance(d, dict):
|
|
raise ErrorDatos(f"{donde}: una Deuda necesita «deuda»")
|
|
for c in ("capital", "tin"):
|
|
if not isinstance(d.get(c), (int, float)) or isinstance(d.get(c), bool) or d[c] < 0:
|
|
raise ErrorDatos(f"{donde}: «deuda.{c}» debe ser un número no negativo")
|
|
if not isinstance(d.get("cuotas"), int) or isinstance(d.get("cuotas"), bool) or d["cuotas"] < 1:
|
|
raise ErrorDatos(f"{donde}: «deuda.cuotas» debe ser un entero positivo")
|
|
_fecha(d.get("primera_cuota"), donde)
|
|
_fecha(r.get("fecha_inicio"), donde)
|
|
elif nombre == "movimientos":
|
|
if not r.get("id") or not r.get("activo"):
|
|
raise ErrorDatos(f"{donde}: faltan «id» o «activo»")
|
|
if r.get("tipo") not in TIPOS_MOVIMIENTO:
|
|
raise ErrorDatos(f"{donde}: tipo de movimiento desconocido {r.get('tipo')!r}")
|
|
_fecha(r.get("fecha"), donde)
|
|
for c in ("cantidad", "precio", "importe", "comisiones", "coste_traspaso", "orden",
|
|
"importe_original", "retencion", "valor_recibido"):
|
|
_numero(r.get(c), donde, c)
|
|
adj = r.get("adjuntos")
|
|
if adj is not None:
|
|
if not isinstance(adj, list) or len(adj) > 20:
|
|
raise ErrorDatos(f"{donde}: «adjuntos» debe ser una lista (máximo 20)")
|
|
for x in adj:
|
|
if not isinstance(x, dict) or not HUELLA_ADJUNTO.match(str(x.get("id"))) \
|
|
or not isinstance(x.get("nombre"), str) or not isinstance(x.get("bytes"), int):
|
|
raise ErrorDatos(f"{donde}: adjunto mal formado")
|
|
for c in ("lugar", "lugar_destino", "traslado"):
|
|
if r.get(c) is not None and not isinstance(r.get(c), str):
|
|
raise ErrorDatos(f"{donde}: «{c}» debe ser texto")
|
|
if r.get("tipo") == "Amortizacion" and r.get("modalidad") not in ("plazo", "cuota"):
|
|
raise ErrorDatos(f"{donde}: «modalidad» debe ser «plazo» o «cuota»")
|
|
elif nombre == "precios":
|
|
if not r.get("activo"):
|
|
raise ErrorDatos(f"{donde}: falta «activo»")
|
|
_fecha(r.get("fecha"), donde)
|
|
_numero(r.get("precio"), donde, "precio")
|
|
_numero(r.get("valor_total"), donde, "valor_total")
|
|
elif nombre == "recurrentes":
|
|
if not r.get("id") or not r.get("activo"):
|
|
raise ErrorDatos(f"{donde}: faltan «id» o «activo»")
|
|
if r.get("tipo") not in TIPOS_RECURRENTE:
|
|
raise ErrorDatos(f"{donde}: tipo de recurrente debe ser Gasto o Compra")
|
|
if r.get("moneda", "EUR") not in MONEDAS:
|
|
raise ErrorDatos(f"{donde}: moneda debe ser EUR o USD")
|
|
if not isinstance(r.get("importe"), (int, float)) or isinstance(r.get("importe"), bool) or r["importe"] <= 0:
|
|
raise ErrorDatos(f"{donde}: «importe» debe ser un número positivo")
|
|
if not isinstance(r.get("dia"), int) or not 1 <= r["dia"] <= 28:
|
|
raise ErrorDatos(f"{donde}: «dia» debe estar entre 1 y 28")
|
|
_fecha(r.get("desde"), donde)
|
|
_fecha(r.get("hasta"), donde, obligatoria=False)
|
|
|
|
|
|
# ── Versión y migraciones ─────────────────────────────────────────────────────
|
|
|
|
def version_de(carpeta: Path) -> int:
|
|
"""Versión del formato de una carpeta de datos (1 si no hay version.json)."""
|
|
ruta = Path(carpeta) / FICHERO_VERSION
|
|
if not ruta.exists():
|
|
return 1
|
|
try:
|
|
v = json.loads(almacen.leer_texto(ruta)).get("version")
|
|
except (json.JSONDecodeError, AttributeError, UnicodeDecodeError):
|
|
v = None
|
|
if isinstance(v, bool) or not isinstance(v, int) or v < 1:
|
|
raise ErrorDatos(f"{FICHERO_VERSION}: no se entiende la versión de los datos")
|
|
return v
|
|
|
|
|
|
def _comprobar_version(v: int) -> None:
|
|
if v > VERSION:
|
|
raise ErrorDatos(f"estos datos son de una versión más nueva de la app (formato {v}; esta app "
|
|
f"entiende hasta el {VERSION}). Actualiza la app antes de abrirlos.")
|
|
|
|
|
|
def migrar(tablas: dict[str, list[dict]], desde: int) -> None:
|
|
"""Aplica en orden las migraciones de «desde» a VERSION (en el sitio)."""
|
|
_comprobar_version(desde)
|
|
for n in range(desde + 1, VERSION + 1):
|
|
MIGRACIONES[n](tablas)
|
|
|
|
|
|
def pendiente_de_migrar(carpeta: Path) -> bool:
|
|
"""¿Hay datos en disco en un formato anterior al de esta app?"""
|
|
carpeta = Path(carpeta)
|
|
return any((carpeta / f"{n}.jsonl").exists() for n in FICHEROS) and version_de(carpeta) < VERSION
|
|
|
|
|
|
def _copia_antes_de_migrar(carpeta: Path) -> None:
|
|
"""Antes de escribir en formato nuevo sobre datos antiguos, se guardan
|
|
los ficheros tal como estaban. Solo la primera vez: no se pisa."""
|
|
if not pendiente_de_migrar(carpeta):
|
|
return # al día, o carpeta nueva sin nada que guardar
|
|
v = version_de(carpeta)
|
|
copia = carpeta / f"copia_v{v}"
|
|
if copia.exists():
|
|
return
|
|
tmp = carpeta / f".copia_v{v}.tmp"
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
tmp.mkdir()
|
|
for nombre in [f"{n}.jsonl" for n in FICHEROS] + [FICHERO_VERSION]:
|
|
if (carpeta / nombre).exists():
|
|
shutil.copyfile(carpeta / nombre, tmp / nombre)
|
|
tmp.rename(copia)
|
|
_permisos(copia, 0o700)
|
|
|
|
|
|
# ── Lectura / escritura ───────────────────────────────────────────────────────
|
|
|
|
def clave(nombre: str, r: dict) -> str:
|
|
"""Clave única de un registro (la usa la fusión de copias)."""
|
|
if nombre == "precios":
|
|
return f"{r['activo']}|{r['fecha']}"
|
|
return r["id"]
|
|
|
|
|
|
def leer(carpeta: Path) -> Datos:
|
|
"""Lee la carpeta de datos, migrándola en memoria si es de un formato
|
|
anterior (en disco no cambia nada hasta que se escribe)."""
|
|
carpeta = Path(carpeta)
|
|
version = version_de(carpeta)
|
|
_comprobar_version(version)
|
|
tablas: dict[str, list[dict]] = {nombre: [] for nombre in FICHEROS}
|
|
lineas: dict[int, str] = {} # id(registro) → «fichero:línea», para los errores
|
|
for nombre in FICHEROS:
|
|
ruta = carpeta / f"{nombre}.jsonl"
|
|
if not ruta.exists():
|
|
continue
|
|
for n, linea in enumerate(almacen.leer_texto(ruta).splitlines(), 1):
|
|
if not linea.strip():
|
|
continue
|
|
donde = f"{ruta.name}:{n}"
|
|
try:
|
|
r = json.loads(linea)
|
|
except json.JSONDecodeError as e:
|
|
raise ErrorDatos(f"{donde}: JSON inválido ({e.msg})") from None
|
|
tablas[nombre].append(r)
|
|
lineas[id(r)] = donde
|
|
if version < VERSION:
|
|
migrar(tablas, version)
|
|
datos = Datos()
|
|
for nombre in FICHEROS:
|
|
vistos: set[str] = set()
|
|
for r in tablas[nombre]:
|
|
donde = lineas.get(id(r), f"{nombre}.jsonl (migrado)")
|
|
_validar(nombre, r, donde)
|
|
k = clave(nombre, r)
|
|
if k in vistos:
|
|
raise ErrorDatos(f"{donde}: clave repetida {k!r}")
|
|
vistos.add(k)
|
|
getattr(datos, nombre).append(r)
|
|
return datos
|
|
|
|
|
|
def _orden_escritura(nombre: str, r: dict):
|
|
if nombre in ("activos", "recurrentes"):
|
|
return (r.get("orden") or 0, r["id"])
|
|
if nombre == "movimientos":
|
|
return (r["fecha"], r.get("orden") or 0, r["id"])
|
|
return (r["fecha"], r["activo"])
|
|
|
|
|
|
def _permisos(ruta: Path, modo: int) -> None:
|
|
"""Solo propietario. En exFAT/FAT (el USB) no hay permisos: se ignora."""
|
|
try:
|
|
os.chmod(ruta, modo)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def _escribir_atomico(destino: Path, lineas) -> None:
|
|
fd, tmp = tempfile.mkstemp(dir=destino.parent, prefix=f".{destino.stem}.", suffix=".tmp")
|
|
try:
|
|
with os.fdopen(fd, "wb") as f:
|
|
f.write(almacen.sellar("".join(linea + "\n" for linea in lineas).encode("utf-8")))
|
|
f.flush()
|
|
os.fsync(f.fileno())
|
|
_permisos(Path(tmp), 0o600)
|
|
os.replace(tmp, destino)
|
|
except BaseException:
|
|
Path(tmp).unlink(missing_ok=True)
|
|
raise
|
|
|
|
|
|
def escribir(carpeta: Path, datos: Datos) -> None:
|
|
"""Escribe los ficheros de forma atómica (nunca quedan a medias), con
|
|
permisos solo para el propietario y en el formato de esta app (VERSION).
|
|
Si lo que había era de un formato anterior, antes guarda una copia."""
|
|
carpeta = Path(carpeta)
|
|
carpeta.mkdir(parents=True, exist_ok=True)
|
|
_permisos(carpeta, 0o700)
|
|
for nombre in FICHEROS:
|
|
for i, r in enumerate(getattr(datos, nombre), 1):
|
|
_validar(nombre, r, f"{nombre} (registro {i})")
|
|
_comprobar_version(version_de(carpeta)) # no pisar datos de una app más nueva
|
|
_copia_antes_de_migrar(carpeta)
|
|
for nombre in FICHEROS:
|
|
registros = sorted(getattr(datos, nombre), key=lambda r: _orden_escritura(nombre, r))
|
|
_escribir_atomico(carpeta / f"{nombre}.jsonl",
|
|
(json.dumps(r, ensure_ascii=False, allow_nan=False) for r in registros))
|
|
# La versión va la última: si algo corta antes, al releer se vuelve a migrar
|
|
_escribir_atomico(carpeta / FICHERO_VERSION, [json.dumps({"version": VERSION})])
|
|
|
|
|
|
# ── Fusión de dos copias ──────────────────────────────────────────────────────
|
|
|
|
def _huella(r: dict) -> str:
|
|
return json.dumps(r, ensure_ascii=False, sort_keys=True)
|
|
|
|
|
|
def fusionar(local: Datos, remoto: Datos) -> tuple[Datos, dict]:
|
|
"""Junta dos copias de los datos (USB y la del servidor git, o dos ordenadores).
|
|
|
|
Cada registro se identifica por su clave; si está en las dos copias con
|
|
contenido distinto gana el de «modificado» más reciente (a igualdad, uno
|
|
fijo, para que las dos copias lleguen al mismo resultado). Los borrados son
|
|
registros con «borrado»: true, así que un borrado nunca resucita.
|
|
Devuelve (datos fusionados, recuento de lo que ha cambiado en cada lado)."""
|
|
resultado = Datos()
|
|
recuento = {"de_remoto": 0, "de_local": 0, "actualizados_desde_remoto": 0, "actualizados_desde_local": 0}
|
|
for nombre in FICHEROS:
|
|
loc = {clave(nombre, r): r for r in getattr(local, nombre)}
|
|
rem = {clave(nombre, r): r for r in getattr(remoto, nombre)}
|
|
salida = []
|
|
for k in sorted(set(loc) | set(rem)):
|
|
a, b = loc.get(k), rem.get(k)
|
|
if b is None:
|
|
salida.append(a)
|
|
recuento["de_local"] += 1
|
|
elif a is None:
|
|
salida.append(b)
|
|
recuento["de_remoto"] += 1
|
|
elif _huella(a) == _huella(b):
|
|
salida.append(a)
|
|
else:
|
|
gana_remoto = (b.get("modificado") or "", _huella(b)) > (a.get("modificado") or "", _huella(a))
|
|
gana, pierde = (b, a) if gana_remoto else (a, b)
|
|
if nombre == "activos":
|
|
gana = _con_todos_los_lugares(gana, pierde)
|
|
salida.append(gana)
|
|
recuento["actualizados_desde_remoto" if gana_remoto else "actualizados_desde_local"] += 1
|
|
setattr(resultado, nombre, salida)
|
|
return resultado, recuento
|
|
|
|
|
|
def _con_todos_los_lugares(gana: dict, pierde: dict) -> dict:
|
|
"""Los lugares de una inversión (exchanges, monederos) no se pierden al
|
|
fusionar: si en una copia se añadió uno y en la otra otro, quedan los dos.
|
|
Si no, las operaciones apuntadas en el que desaparece se quedarían sin sitio."""
|
|
suyos = {x.get("id") for x in gana.get("lugares") or []}
|
|
faltan = [x for x in pierde.get("lugares") or [] if x.get("id") not in suyos]
|
|
if not faltan:
|
|
return gana
|
|
return {**gana, "lugares": [*(gana.get("lugares") or []), *faltan]}
|
|
|
|
|
|
# ── Antes de fusionar dos copias que han vivido separadas ─────────────────────
|
|
#
|
|
# Fusionar va registro a registro y se fía de los identificadores. Hay dos
|
|
# casos en los que eso solo no basta, y los dos se arreglan ANTES o DESPUÉS de
|
|
# fusionar, sin tocar la regla:
|
|
#
|
|
# · Dos inversiones DISTINTAS con el mismo identificador. Los de las
|
|
# inversiones son correlativos (F1, F2…), así que si se da de alta una en
|
|
# cada copia sin sincronizar entre medias, chocan. Se distinguen por su
|
|
# «uid» (un número al azar que se pone al crearlas); si chocan, a la que
|
|
# llega se le da otro identificador y sus operaciones la siguen.
|
|
# · La misma operación apuntada en las dos copias (el mismo extracto
|
|
# importado en los dos sitios): tienen identificadores distintos y la
|
|
# fusión las dejaría duplicadas. Si coinciden en TODO lo que importa y
|
|
# vienen de la misma línea de extracto, se queda una.
|
|
|
|
def choques_de_activos(local: Datos, remoto: Datos, prefijos: dict[str, str]) -> dict[str, str]:
|
|
"""{identificador en «remoto»: el nuevo que hay que darle} para las
|
|
inversiones de «remoto» que comparten identificador con otra distinta de
|
|
«local». Sin «uid» en alguna de las dos, se entiende que son la misma."""
|
|
suyos = {a["id"]: a for a in local.activos}
|
|
usados = set(suyos) | {a["id"] for a in remoto.activos}
|
|
mapa = {}
|
|
for a in remoto.activos:
|
|
otro = suyos.get(a["id"])
|
|
if otro is None or not a.get("uid") or not otro.get("uid") or a["uid"] == otro["uid"]:
|
|
continue
|
|
prefijo = prefijos.get(a.get("tipo"), "X")
|
|
n = 1
|
|
while f"{prefijo}{n}" in usados:
|
|
n += 1
|
|
mapa[a["id"]] = f"{prefijo}{n}"
|
|
usados.add(mapa[a["id"]])
|
|
return mapa
|
|
|
|
|
|
def renombrar_activos(datos: Datos, mapa: dict[str, str]) -> Datos:
|
|
"""Una copia de los datos con esas inversiones bajo su identificador nuevo
|
|
y todo lo que las nombra (operaciones, precios, automáticas) siguiéndolas."""
|
|
if not mapa:
|
|
return datos
|
|
cambia = lambda r, campo: {**r, campo: mapa[r[campo]]} if r.get(campo) in mapa else r # noqa: E731
|
|
return Datos([cambia(a, "id") for a in datos.activos], [cambia(m, "activo") for m in datos.movimientos],
|
|
[cambia(x, "activo") for x in datos.precios], [cambia(r, "activo") for r in datos.recurrentes])
|
|
|
|
|
|
_NO_CUENTAN = ("id", "orden", "modificado", "adjuntos", "notas", "importacion")
|
|
|
|
|
|
def unir_duplicadas(datos: Datos, momento: str) -> list[dict]:
|
|
"""Deja una sola de las operaciones que son la misma línea de extracto
|
|
apuntada dos veces: mismo extracto y mismos datos. Las demás quedan como
|
|
borradas (no se pierden: un borrado es una marca). Devuelve las retiradas."""
|
|
grupos: dict[str, list[dict]] = {}
|
|
for m in datos.movimientos:
|
|
if m.get("borrado") or not m.get("extracto"):
|
|
continue
|
|
firma = json.dumps({k: v for k, v in m.items() if k not in _NO_CUENTAN}, ensure_ascii=False, sort_keys=True)
|
|
grupos.setdefault(firma, []).append(m)
|
|
retiradas = []
|
|
for iguales in grupos.values():
|
|
if len(iguales) < 2:
|
|
continue
|
|
# Se queda la que más tiene (justificantes, notas) y, a igualdad, la más antigua: siempre la misma en las dos copias
|
|
iguales.sort(key=lambda m: (-len(m.get("adjuntos") or []), -len(m.get("notas") or ""), m.get("modificado") or "", m["id"]))
|
|
for sobra in iguales[1:]:
|
|
sobra["borrado"], sobra["modificado"] = True, momento
|
|
retiradas.append(sobra)
|
|
return retiradas
|
|
|
|
|
|
def huella_datos(datos: Datos) -> str:
|
|
"""Resumen del contenido (para saber si algo ha cambiado desde la última vez)."""
|
|
import hashlib
|
|
h = hashlib.sha256()
|
|
for nombre in FICHEROS:
|
|
for r in sorted(getattr(datos, nombre), key=lambda r: clave(nombre, r)):
|
|
h.update(_huella(r).encode())
|
|
return h.hexdigest()
|