673 lines
34 KiB
Python
673 lines
34 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
|
||
(salvo las operaciones automáticas: ver fusionar).
|
||
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, automatica, modificado, borrado
|
||
· automatica: {nivel, modificado}, la pone la app al crear,
|
||
completar o afinar una operación automática (nivel 1 pendiente,
|
||
2 estimada, 3 exacta). Solo vale mientras su «modificado» sea el
|
||
de la operación: en cuanto el usuario la toca, deja de valer.
|
||
· 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 | Dividendo), 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", "Inmueble", "RentaFija", "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", "Dividendo")
|
||
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 = 3
|
||
|
||
|
||
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)."""
|
||
|
||
|
||
def _v3_reglas_de_sincronizar(tablas: dict) -> None:
|
||
"""Versión 3: las operaciones que genera la app llevan la marca «automatica»
|
||
y la fusión las trata aparte (ver fusionar). No hay nada que convertir; la
|
||
versión sube para que un Caudal anterior, que fusionaría con las reglas
|
||
viejas, no pueda sincronizar con este hasta que se actualice."""
|
||
|
||
|
||
MIGRACIONES: dict = {2: _v2_lugares, 3: _v3_reglas_de_sincronizar}
|
||
|
||
|
||
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)
|
||
rf = r.get("renta_fija")
|
||
if rf is not None:
|
||
from .rentafija import CLASES as CLASES_RF, PAGOS
|
||
if not isinstance(rf, dict) or rf.get("clase") not in CLASES_RF or rf.get("pagos", 0) not in PAGOS:
|
||
raise ErrorDatos(f"{donde}: «renta_fija» necesita una clase y una frecuencia de pago válidas")
|
||
_fecha(rf.get("vencimiento"), donde)
|
||
for c in ("nominal", "interes"):
|
||
v = rf.get(c)
|
||
if v is not None and (not isinstance(v, (int, float)) or isinstance(v, bool) or v < 0):
|
||
raise ErrorDatos(f"{donde}: «renta_fija.{c}» debe ser un número no negativo")
|
||
f = r.get("inmueble")
|
||
if f is not None:
|
||
from .inmuebles import CLASES, USOS
|
||
if not isinstance(f, dict) or f.get("clase") not in CLASES or f.get("uso") not in USOS:
|
||
raise ErrorDatos(f"{donde}: «inmueble» necesita una clase y un uso válidos")
|
||
for c in ("precio_compra", "valor_catastral", "porcentaje_construccion", "precio_maximo", "superficie",
|
||
"intereses_hipoteca", "coef_vivienda", "coef_anejo"):
|
||
v = f.get(c)
|
||
if v is not None and (not isinstance(v, (int, float)) or isinstance(v, bool) or v < 0):
|
||
raise ErrorDatos(f"{donde}: «inmueble.{c}» debe ser un número no negativo")
|
||
cultivos = f.get("cultivos")
|
||
if cultivos is not None and (not isinstance(cultivos, list) or
|
||
not all(isinstance(x, dict) and isinstance(x.get("cultivo"), str) for x in cultivos)):
|
||
raise ErrorDatos(f"{donde}: «inmueble.cultivos» debe ser una lista de cultivos con su fecha")
|
||
piezas = f.get("piezas")
|
||
if piezas is not None and (not isinstance(piezas, list) or not all(
|
||
isinstance(x, dict) and isinstance(x.get("m2"), (int, float)) and not isinstance(x.get("m2"), bool)
|
||
and x.get("clase") in ("vivienda", "anejo") for x in piezas)):
|
||
raise ErrorDatos(f"{donde}: «inmueble.piezas» debe ser una lista de piezas con sus metros")
|
||
modulos = f.get("modulos")
|
||
if modulos is not None and (not isinstance(modulos, dict) or not all(
|
||
isinstance(k, str) and k.isdigit() and isinstance(v, (int, float)) and not isinstance(v, bool)
|
||
for k, v in modulos.items())):
|
||
raise ErrorDatos(f"{donde}: «inmueble.modulos» debe ir por años con su importe")
|
||
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, Compra o Dividendo")
|
||
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)
|
||
|
||
|
||
# Niveles de una operación automática tal como la dejó la app (movimiento["automatica"])
|
||
AUTO_PENDIENTE, AUTO_ESTIMADA, AUTO_EXACTA = 1, 2, 3
|
||
|
||
|
||
def nivel_automatica(m: dict) -> int | None:
|
||
"""Si esta versión de la operación la dejó así la app (y no el usuario),
|
||
su nivel; si no, None. Vale para las automáticas y para los cupones y
|
||
vencimientos de renta fija. Una pendiente de una recurrente siempre es de
|
||
la app, aunque venga de una versión anterior que aún no ponía la marca."""
|
||
if m.get("borrado"):
|
||
return None
|
||
marca = m.get("automatica")
|
||
if isinstance(marca, dict) and marca.get("modificado") and marca.get("modificado") == m.get("modificado"):
|
||
return int(marca.get("nivel") or AUTO_PENDIENTE)
|
||
return AUTO_PENDIENTE if m.get("recurrente") and m.get("pendiente") else None
|
||
|
||
|
||
def _prioridad(nombre: str, r: dict) -> tuple:
|
||
"""Orden para elegir entre dos versiones del mismo registro: primero lo
|
||
que hizo el usuario, después lo de la app; dentro de cada grupo, lo más
|
||
completo y lo más reciente. En precios, el puesto a mano manda siempre
|
||
sobre el descargado o el calculado."""
|
||
if nombre == "precios":
|
||
return (r.get("fuente") == "manual" and not r.get("borrado"), 0, r.get("modificado") or "", _huella(r))
|
||
nivel = nivel_automatica(r) if nombre == "movimientos" else None
|
||
return (nivel is None, nivel or 0, r.get("modificado") or "", _huella(r))
|
||
|
||
|
||
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.
|
||
|
||
Excepción, las operaciones automáticas: cada copia las genera al abrirse,
|
||
así que la que se abre la última tiene la versión «más reciente» aunque
|
||
la otra ya la tuviera completa o corregida a mano. Por eso una versión que
|
||
tocó el usuario gana siempre a una que solo escribió la app, y entre dos
|
||
de la app gana la más completa (exacta > estimada > pendiente). Y las
|
||
que una copia generó sin saber lo que había pasado en la otra (automática
|
||
finalizada, pausada o borrada; inversión cerrada; cupón apuntado a mano)
|
||
se retiran (ver sobrantes_de_automaticas). Un precio puesto a mano gana
|
||
siempre a uno descargado.
|
||
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 = _prioridad(nombre, b) > _prioridad(nombre, 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)
|
||
retirar = {m["id"]: m for m in sobrantes_de_automaticas(resultado)}
|
||
resultado.movimientos = [retirar.get(m["id"], m) for m in resultado.movimientos]
|
||
return resultado, recuento
|
||
|
||
|
||
def sobrantes_de_automaticas(datos: Datos, hoy: dt.date | None = None) -> list[dict]:
|
||
"""Operaciones que la app generó sola y que ya no tocan, devueltas como
|
||
copias marcadas de borradas (no toca las originales).
|
||
|
||
Solo las que la app escribió sola: si el usuario tocó una, se queda.
|
||
De una automática sobra:
|
||
· la de fecha fuera de «desde»–«hasta» (se finalizó o se acortó);
|
||
· la posterior al cierre de su inversión;
|
||
· la de fecha posterior al día en que se pausó o borró la regla: la
|
||
generó una copia que aún no lo sabía (el USB al abrirse, antes de
|
||
sincronizar). Las de antes de ese día se quedan: tocaban.
|
||
De renta fija sobra el cupón o la devolución que el calendario ya no
|
||
daría: se vendió, se cerró o el usuario apuntó el cobro de verdad.
|
||
La marca de borrado lleva un «modificado» que no depende de quién limpie:
|
||
así las dos copias llegan al mismo registro."""
|
||
from . import rentafija
|
||
hoy = hoy or dt.date.today()
|
||
reglas = {r["id"]: r for r in datos.recurrentes}
|
||
activos = {a["id"]: a for a in datos.activos}
|
||
vivos = [m for m in datos.movimientos if not m.get("borrado")]
|
||
calendarios: dict[str, set] = {}
|
||
sobran = []
|
||
for m in vivos:
|
||
if nivel_automatica(m) is None:
|
||
continue
|
||
a = activos.get(m.get("activo")) or {}
|
||
cierre = a.get("fecha_cierre") if a.get("estado") == "cerrada" else None
|
||
momento = m.get("modificado") or ""
|
||
r = reglas.get(m.get("recurrente"))
|
||
if r is not None:
|
||
parada = bool(r.get("borrado") or r.get("pausado")) and bool(r.get("modificado")) \
|
||
and m["fecha"] > r["modificado"][:10]
|
||
sobra = m["fecha"] < r.get("desde", "") or bool(r.get("hasta") and m["fecha"] > r["hasta"]) \
|
||
or bool(cierre and m["fecha"] > cierre) or parada
|
||
momento = max(momento, r.get("modificado") or "")
|
||
elif str(m.get("id", "")).startswith(f"rf-{m.get('activo')}-") and a.get("tipo") == "RentaFija":
|
||
if a["id"] not in calendarios:
|
||
suyos = [x for x in vivos if x["activo"] == a["id"]]
|
||
calendarios[a["id"]] = {x["id"] for x in rentafija.calendario(a, suyos, hoy)}
|
||
sobra = m["fecha"] <= hoy.isoformat() and m["id"] not in calendarios[a["id"]]
|
||
else:
|
||
continue
|
||
if sobra:
|
||
momento = max(momento, a.get("modificado") or "")
|
||
sobran.append({**m, "borrado": True, "modificado": momento})
|
||
return sobran
|
||
|
||
|
||
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()
|