caudal/programa/cartera/datos.py

"""
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()