1864 lines
103 KiB
Python
1864 lines
103 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
servidor.py — La aplicación de la cartera en el navegador.
|
|
|
|
python3 -m cartera.servidor [--datos datos/] [--puerto 0] [--no-navegador]
|
|
|
|
Seguridad:
|
|
• Escucha solo en 127.0.0.1 (no es accesible desde la red).
|
|
• Cada arranque genera una clave aleatoria que va en la dirección que se abre;
|
|
sin ella la API no responde (otro programa o usuario del equipo no puede
|
|
leer ni tocar la cartera).
|
|
• Comprueba la cabecera Host (evita ataques de «DNS rebinding» desde webs).
|
|
• Nada se envía a ningún servidor salvo las consultas públicas de precios.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import binascii
|
|
import datetime as dt
|
|
import gzip
|
|
import hashlib
|
|
import hmac
|
|
import io
|
|
import json
|
|
import mimetypes
|
|
import os
|
|
import secrets
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import traceback
|
|
import webbrowser
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from urllib.parse import parse_qs, quote, urlparse
|
|
|
|
from . import actividades
|
|
from . import almacen
|
|
from . import marcas
|
|
from .acceso import (DURACION as DURACION_SESION, MINIMO as MINIMO_CLAVE, RECOMENDADO as RECOMENDADO_CLAVE,
|
|
Acceso, ErrorAcceso, Espera)
|
|
from .sobres import ErrorCifrado
|
|
from . import adjuntos
|
|
from . import fuentes
|
|
from . import extractos
|
|
from . import fiscal
|
|
from . import inmuebles
|
|
from . import vpo
|
|
from . import btc
|
|
from . import electrum
|
|
from . import cache_btc
|
|
from . import copia
|
|
from . import instalacion
|
|
from . import qr
|
|
from . import enlace
|
|
from . import deshacer_btc
|
|
from . import historial_btc
|
|
from . import lugares
|
|
from . import prima
|
|
from . import spot_btc
|
|
from . import traza_btc
|
|
from .calculo import Precios
|
|
from .datos import FICHERO_VERSION, TIPOS_ACTIVO, ErrorDatos
|
|
from .operaciones import PREFIJOS, Cartera, CarteraOcupada, ErrorOperacion
|
|
from .btc import ErrorBitcoin
|
|
from .electrum import ErrorElectrum
|
|
from .sincronizar import AJUSTES, ErrorSync, Sincronizador, escribir_ajuste
|
|
from .vista import cantidad_en, evolucion_activo, resumen
|
|
|
|
WEB = Path(__file__).resolve().parent / "web"
|
|
PROYECTO = Path(__file__).resolve().parent.parent
|
|
# La cartera de mentira que se enseña a quien quiere mirar antes de empezar: inversiones
|
|
# corrientes e inventadas (la fabrica herramientas/generar_ejemplo.py)
|
|
EJEMPLO = Path(__file__).resolve().parent / "ejemplo"
|
|
COOKIE = "caudal_sesion"
|
|
COMPRIMIR_DESDE = 1024 # bytes
|
|
COMPRIMIBLES = {"application/json", "application/javascript", "text/javascript", "text/css", "text/html", "image/svg+xml", "text/plain"}
|
|
# Lo pone quien empaqueta la aplicación para un sistema que no cifra el disco
|
|
CIFRAR_DE_ENTRADA = os.environ.get("CAUDAL_CIFRADO", "").strip().lower() in ("1", "si", "sí", "true")
|
|
# Lo pone quien empaqueta para un sistema que YA cifra el disco (StartOS): se avisa de que cifrar aquí no hace falta
|
|
DISCO_YA_CIFRADO = os.environ.get("CAUDAL_DISCO_CIFRADO", "").strip().lower() in ("1", "si", "sí", "true")
|
|
MAX_CUERPO = 8_000_000
|
|
MAX_COPIA = 210_000_000 # una copia de seguridad que se sube para restaurarla (va en base64) # los extractos pueden ser grandes
|
|
SESION = ".ultima_sesion.json"
|
|
|
|
|
|
# Las fechas que se mueven al abrir la cartera de ejemplo, también dentro de las
|
|
# fichas (vencimientos, cuotas, cultivos, alquileres)
|
|
_FECHAS_EJEMPLO = {"fecha", "fecha_inicio", "fecha_cierre", "desde", "hasta", "vencimiento", "primera_cuota", "renta_desde"}
|
|
|
|
|
|
def _desplazar(r, salto: dt.timedelta):
|
|
if isinstance(r, list):
|
|
return [_desplazar(x, salto) for x in r]
|
|
if not isinstance(r, dict):
|
|
return r
|
|
salida = {}
|
|
for k, v in r.items():
|
|
if k in _FECHAS_EJEMPLO and isinstance(v, str) and len(v) == 10:
|
|
try:
|
|
v = (dt.date.fromisoformat(v) + salto).isoformat()
|
|
except ValueError:
|
|
pass
|
|
salida[k] = _desplazar(v, salto)
|
|
return salida
|
|
|
|
|
|
class App:
|
|
def __init__(self, cartera: Cartera | None, acceso: Acceso | None = None, carpeta: Path | None = None):
|
|
# Sin cartera = BLOQUEADA: los datos están cifrados y no se abren hasta que alguien entre
|
|
self.cartera = cartera
|
|
self.carpeta = Path(carpeta) if carpeta else cartera.carpeta
|
|
# En un servidor se entra con contraseña (ver acceso.py); en tu equipo, con la dirección secreta
|
|
self.acceso = acceso
|
|
self.token = secrets.token_urlsafe(24)
|
|
self.puerto = 0
|
|
self.servidor: ThreadingHTTPServer | None = None
|
|
self._precios_lock = threading.Lock()
|
|
self._sync_lock = threading.Lock()
|
|
self._btc_lock = threading.Lock()
|
|
self.sync = Sincronizador(self.carpeta.parent / "sync", ajustes=self.carpeta.parent)
|
|
self.ultima_sesion = self._leer_sesion() if cartera else None
|
|
self._real = None # (cartera, sync, sesión) de verdad mientras se enseña el ejemplo
|
|
self._copia_lista = None # (id, nombre, fichero) de la copia recién hecha, hasta que se descarga
|
|
self.llave: bytes | None = None # la llave de la instalación, en memoria desde que se entra (ver acceso.py)
|
|
|
|
# ── copia de seguridad cifrada (ver copia.py) ──
|
|
def copia_crear(self, cuerpo: dict) -> dict:
|
|
if self._real is not None:
|
|
raise ErrorOperacion("Sal de la cartera de ejemplo para hacer una copia de la tuya.")
|
|
con_justificantes = cuerpo.get("justificantes") is not False
|
|
if self.acceso is not None and self.llave:
|
|
# En un servidor va con la llave de la instalación: se abrirá con la contraseña de Caudal de hoy
|
|
fichero, resumen = copia.crear(self.cartera.carpeta, None, con_justificantes, self.llave, self.acceso.sobres())
|
|
self.acceso.anotar_copia()
|
|
else:
|
|
fichero, resumen = copia.crear(self.cartera.carpeta, str(cuerpo.get("clave") or ""), con_justificantes)
|
|
nombre = f"caudal-{dt.date.today().isoformat()}{copia.EXTENSION}"
|
|
self._copia_lista = (secrets.token_urlsafe(12), nombre, fichero)
|
|
return {"id": self._copia_lista[0], "nombre": nombre, "bytes": resumen["bytes"],
|
|
"justificantes": resumen["justificantes"]}
|
|
|
|
def copia_descargar(self, ident: str) -> tuple[str, bytes]:
|
|
lista = self._copia_lista
|
|
if not lista or not hmac.compare_digest(lista[0], ident or ""):
|
|
raise ErrorOperacion("Esa copia ya no está: vuelve a crearla.")
|
|
self._copia_lista = None # se entrega una vez y no se queda en memoria
|
|
ajustes = self._ajustes() # para la lista de primeros pasos: ya hay una copia
|
|
ajustes["primer_uso"] = {**(ajustes.get("primer_uso") or {}), "copia": dt.date.today().isoformat()}
|
|
self._guardar_ajustes(ajustes)
|
|
return lista[1], lista[2]
|
|
|
|
def copia_restaurar(self, cuerpo: dict) -> dict:
|
|
"""Sin «confirmar» solo abre la copia y cuenta lo que trae; con él, sustituye lo que hay."""
|
|
if self._real is not None:
|
|
raise ErrorOperacion("Sal de la cartera de ejemplo para restaurar una copia.")
|
|
try:
|
|
fichero = base64.b64decode(str(cuerpo.get("fichero") or ""), validate=True)
|
|
except (binascii.Error, ValueError):
|
|
raise ErrorOperacion("El fichero no ha llegado bien. Vuelve a elegirlo.") from None
|
|
abierta = copia.abrir(fichero, str(cuerpo.get("clave") or ""))
|
|
datos = abierta["datos"]
|
|
trae = {"creada": abierta["resumen"].get("creada"),
|
|
"inversiones": len(datos.vivos("activos")), "operaciones": len(datos.vivos("movimientos")),
|
|
"justificantes": len(abierta["adjuntos"]), "ajustes": sorted(abierta["ajustes"])}
|
|
ahora = self.cartera.datos
|
|
trae["ahora"] = {"inversiones": len(ahora.vivos("activos")), "operaciones": len(ahora.vivos("movimientos"))}
|
|
if not cuerpo.get("confirmar"):
|
|
return {"restaurada": False, **trae}
|
|
if trae["ahora"]["inversiones"] or trae["ahora"]["operaciones"]:
|
|
self._copia_datos() # lo que había queda en la carpeta de copias locales
|
|
self.cartera.reemplazar(datos)
|
|
for nombre, contenido in abierta["ajustes"].items():
|
|
escribir_ajuste(self.cartera.carpeta.parent / nombre, json.loads(contenido))
|
|
for contenido in abierta["adjuntos"].values():
|
|
adjuntos.guardar(self.cartera.carpeta, contenido)
|
|
return {"restaurada": True, **trae}
|
|
|
|
# ── entrar, y el cifrado de los datos en el disco (ver acceso.py y almacen.py) ──
|
|
def entrar(self, llave: bytes) -> None:
|
|
"""Ya se sabe la contraseña: la llave queda en memoria y, si los datos
|
|
estaban cerrados con ella, se abren."""
|
|
self.llave = llave
|
|
almacen.poner_llave(llave, self.acceso.cifrado())
|
|
if self.cartera is None:
|
|
self.cartera = Cartera(self.carpeta)
|
|
self.ultima_sesion = self._leer_sesion()
|
|
|
|
def _ficheros_de_datos(self) -> list[Path]:
|
|
"""Todo lo que Caudal guarda en el disco y hay que cifrar o descifrar."""
|
|
raiz = self.carpeta.parent
|
|
ficheros = [f for f in self.carpeta.rglob("*") if f.is_file() and f.name != ".cerrojo" and not f.name.endswith(".tmp")]
|
|
ficheros += [raiz / nombre for nombre in (*AJUSTES, cache_btc.FICHERO, enlace.FICHERO_DIRECCIONES)]
|
|
ficheros += sorted((raiz / "backups").glob("*.tar.gz"))
|
|
return [f for f in ficheros if f.is_file()]
|
|
|
|
def cifrado_poner(self, cuerpo: dict) -> dict:
|
|
"""Activa o desactiva el cifrado de los datos. Pide la contraseña otra vez."""
|
|
if self.acceso is None:
|
|
raise ErrorOperacion("El cifrado de los datos es para cuando Caudal va en un servidor.")
|
|
if self._real is not None:
|
|
raise ErrorOperacion("Sal de la cartera de ejemplo para cambiar el cifrado.")
|
|
clave = str(cuerpo.get("clave") or "")
|
|
if cuerpo.get("activar"):
|
|
return self._cifrar(clave)
|
|
llave = self.acceso.comprobar(clave)
|
|
# Primero se deja todo en claro y solo entonces se apunta que ya no va cifrado:
|
|
# si se corta a medias, al arrancar se sigue pidiendo la contraseña y todo se lee
|
|
almacen.poner_llave(llave, False)
|
|
cambiados = almacen.reescribir(self._ficheros_de_datos())
|
|
self.acceso.poner_cifrado(False)
|
|
return {"cifrado": False, "ficheros": cambiados}
|
|
|
|
def _cifrar(self, clave: str) -> dict:
|
|
llave, recuperacion = self.acceso.recuperacion_nueva(clave)
|
|
almacen.poner_llave(llave, False)
|
|
for f in self._ficheros_de_datos():
|
|
almacen.leer(f) # si algo no se pudiera leer, mejor saberlo antes de empezar
|
|
# Primero se apunta que va cifrado y luego se cifra: si se corta a medias,
|
|
# al arrancar se pide la contraseña y lo que quedó en claro se lee igual
|
|
self.acceso.poner_cifrado(True)
|
|
almacen.poner_llave(llave, True)
|
|
cambiados = almacen.reescribir(self._ficheros_de_datos())
|
|
return {"cifrado": True, "recuperacion": recuperacion, "ficheros": cambiados}
|
|
|
|
def recuperacion_nueva(self, cuerpo: dict) -> dict:
|
|
"""Otra clave de recuperación (la anterior deja de valer). Pide la contraseña."""
|
|
if self.acceso is None:
|
|
raise ErrorOperacion("En este equipo Caudal no usa contraseña.")
|
|
return {"recuperacion": self.acceso.recuperacion_nueva(str(cuerpo.get("clave") or ""))[1]}
|
|
|
|
# ── sincronizar con otro Caudal (ver enlace.py) ──
|
|
def _sin_ejemplo(self) -> None:
|
|
if self._real is not None:
|
|
raise ErrorOperacion("Sal de la cartera de ejemplo para sincronizar la tuya.")
|
|
|
|
def enlace_alta(self, cuerpo: dict) -> dict:
|
|
"""En el servidor: da una credencial a otro Caudal que acaba de entrar con la contraseña."""
|
|
if self.acceso is None or not self.llave:
|
|
raise ErrorOperacion("Solo un Caudal que va en un servidor puede enlazar a otros.")
|
|
return {"credencial": self.acceso.enlace_nuevo(self.llave, str(cuerpo.get("nombre") or ""))}
|
|
|
|
def enlace_codigo(self) -> dict:
|
|
"""En el servidor: un código de pocos minutos para enlazar otro equipo sin teclear la contraseña."""
|
|
if self.acceso is None or not self.llave:
|
|
raise ErrorOperacion("Solo un Caudal que va en un servidor puede enlazar a otros.")
|
|
return {**self.acceso.codigo_de_enlace(), "direcciones": enlace.direcciones_de_enlace(self.carpeta)}
|
|
|
|
def enlace_direcciones(self, cuerpo: dict) -> dict:
|
|
"""En el servidor: las direcciones que lleva el QR (lo escrito a mano manda sobre lo del sistema)."""
|
|
if self.acceso is None or not self.llave:
|
|
raise ErrorOperacion("Solo un Caudal que va en un servidor puede enlazar a otros.")
|
|
if "casa" in cuerpo or "tor" in cuerpo:
|
|
return enlace.poner_direcciones_de_enlace(self.carpeta, str(cuerpo.get("casa") or ""), str(cuerpo.get("tor") or ""))
|
|
return enlace.direcciones_de_enlace(self.carpeta)
|
|
|
|
def enlace_canje(self, cuerpo: dict) -> dict:
|
|
"""En el servidor: otro Caudal trae el código (sin sesión) y se lleva su credencial."""
|
|
return {"credencial": self.acceso.canjear_codigo(str(cuerpo.get("codigo") or ""), self.llave,
|
|
str(cuerpo.get("nombre") or ""))}
|
|
|
|
def dibujar_qr(self, cuerpo: dict) -> dict:
|
|
try:
|
|
return {"filas": qr.filas(str(cuerpo.get("texto") or ""))}
|
|
except qr.TextoDemasiadoLargo as e:
|
|
raise ErrorOperacion(str(e)) from None
|
|
|
|
def enlace_olvidar_equipos(self, cuerpo: dict) -> dict:
|
|
if self.acceso is None:
|
|
raise ErrorOperacion("Solo un Caudal que va en un servidor tiene equipos enlazados.")
|
|
return {"olvidados": self.acceso.olvidar_enlaces()}
|
|
|
|
def enlace_configurar(self, cuerpo: dict) -> dict:
|
|
"""En el otro Caudal: se enlaza con un servidor (o deja de estarlo)."""
|
|
e = enlace.Enlace(self.carpeta)
|
|
if cuerpo.get("desenlazar"):
|
|
e.olvidar()
|
|
return {"enlazado": False}
|
|
if "url_tor" in cuerpo and not cuerpo.get("url"): # ya enlazado: la dirección .onion para fuera de casa
|
|
return {"enlazado": True, **e.poner_onion(str(cuerpo.get("url_tor") or ""))}
|
|
r = e.enlazar(str(cuerpo.get("url") or ""), str(cuerpo.get("clave") or ""), str(cuerpo.get("certificado") or ""),
|
|
str(cuerpo.get("nombre") or ""), str(cuerpo.get("codigo") or ""), str(cuerpo.get("certificado_pem") or ""),
|
|
str(cuerpo.get("url_tor") or ""))
|
|
return {"enlazado": True, "url_tor": r["url_tor"]}
|
|
|
|
def enlace_sincronizar(self, cuerpo: dict) -> dict:
|
|
self._sin_ejemplo()
|
|
if not self._sync_lock.acquire(blocking=False):
|
|
raise ErrorOperacion("Ya hay una sincronización en marcha.")
|
|
try:
|
|
return enlace.sincronizar(self, bool(cuerpo.get("simular")))
|
|
finally:
|
|
self._sync_lock.release()
|
|
|
|
def enlace_atender(self, credencial: str, cuerpo: dict) -> dict:
|
|
"""En el servidor: lo que pide un Caudal enlazado, que se identifica con su credencial."""
|
|
llave = self.acceso.llave_de_enlace(credencial)
|
|
self.entrar(llave) # si los datos estaban cerrados, su credencial los abre
|
|
self._sin_ejemplo()
|
|
return enlace.atender(self, str(cuerpo.get("orden") or ""), cuerpo, PREFIJOS)
|
|
|
|
def cambiar_clave(self, cuerpo: dict) -> dict:
|
|
if self.acceso is None:
|
|
raise ErrorOperacion("En este equipo Caudal no usa contraseña.")
|
|
# «_sesion» lo pone el manejador: quien la cambia sigue dentro; las demás sesiones se cierran
|
|
self.acceso.cambiar(str(cuerpo.get("actual") or ""), str(cuerpo.get("nueva") or ""), str(cuerpo.get("_sesion") or ""))
|
|
return {"cambiada": True}
|
|
|
|
def ejemplo(self, entrar: bool) -> dict:
|
|
"""Entra o sale de la cartera de ejemplo. El ejemplo es una copia en una
|
|
carpeta temporal: la cartera de verdad no se toca ni se mezcla, y lo
|
|
que se haga en el ejemplo se tira al salir."""
|
|
if entrar and self._real is None:
|
|
if not EJEMPLO.is_dir():
|
|
raise ErrorOperacion("Esta instalación no trae la cartera de ejemplo.")
|
|
carpeta = Path(tempfile.mkdtemp(prefix="caudal-ejemplo-")) / "datos"
|
|
shutil.copytree(EJEMPLO, carpeta)
|
|
# El ejemplo está escrito respecto a una fecha base: se desplaza entero para que acabe hoy
|
|
# y siempre parezca una cartera al día, se abra cuando se abra
|
|
try:
|
|
base = dt.date.fromisoformat(json.loads((carpeta / "base.json").read_text(encoding="utf-8"))["base"])
|
|
except (OSError, ValueError, KeyError):
|
|
base = dt.date.today()
|
|
salto = dt.timedelta(days=max((dt.date.today() - base).days, 0))
|
|
for fichero in carpeta.glob("*.jsonl"):
|
|
registros = [_desplazar(json.loads(x), salto) for x in fichero.read_text(encoding="utf-8").splitlines() if x.strip()]
|
|
for r in registros:
|
|
if fichero.stem == "precios":
|
|
r["fuente"] = "ejemplo"
|
|
fichero.write_text("".join(json.dumps(r, ensure_ascii=False) + "\n" for r in registros), encoding="utf-8")
|
|
(carpeta / "base.json").unlink(missing_ok=True)
|
|
self._real = (self.cartera, self.sync, self.ultima_sesion)
|
|
self.cartera = Cartera(carpeta)
|
|
self.cartera._guardar() # el valor de inmuebles y Letras, la cuota de la hipoteca… ya a las fechas nuevas
|
|
self.sync = Sincronizador(carpeta.parent / "sync", ajustes=carpeta.parent)
|
|
self.ultima_sesion = None
|
|
elif not entrar and self._real is not None:
|
|
temporal = self.cartera.carpeta.parent
|
|
self.cartera.cerrar()
|
|
self.cartera, self.sync, self.ultima_sesion = self._real
|
|
self._real = None
|
|
shutil.rmtree(temporal, ignore_errors=True)
|
|
return {"ejemplo": self._real is not None}
|
|
|
|
# ── dónde viven los datos del instalable de escritorio (ver instalacion.py) ──
|
|
def instalacion(self) -> dict | None:
|
|
"""None salvo en el instalable. «elegible»: aún no hay nada apuntado y se puede cambiar de sitio."""
|
|
if not instalacion.empaquetado() or self.acceso is not None:
|
|
return None
|
|
base, modo = instalacion.base_actual()
|
|
vacia = self._real is None and self.cartera is not None and not (
|
|
self.cartera.datos.activos or self.cartera.datos.movimientos)
|
|
return {"modo": modo, "carpeta": str(base), "elegible": vacia and self.cartera.carpeta.parent == base,
|
|
"portatil": str(instalacion.base_portatil()), "personal": str(instalacion.base_personal()),
|
|
"portatil_posible": instalacion.se_puede_escribir_junto_al_programa(),
|
|
"lanzador": instalacion.lanzador()}
|
|
|
|
def instalacion_lanzador(self, poner: bool) -> dict:
|
|
"""Añade Caudal al menú de aplicaciones del equipo, o lo quita."""
|
|
if not self.instalacion() or instalacion.lanzador() is None:
|
|
raise ErrorOperacion("Aquí no sé añadir Caudal al menú de aplicaciones.")
|
|
try:
|
|
if poner:
|
|
instalacion.poner_lanzador(WEB / "icono.svg")
|
|
else:
|
|
instalacion.quitar_lanzador()
|
|
except (OSError, ValueError) as e:
|
|
raise ErrorOperacion(f"No he podido cambiar el menú de aplicaciones: {e}") from None
|
|
return self.instalacion()
|
|
|
|
def instalacion_mover(self, modo: str) -> dict:
|
|
estado = self.instalacion()
|
|
if not estado or not estado["elegible"]:
|
|
raise ErrorOperacion("Los datos ya no se pueden cambiar de sitio desde aquí: haz una copia de seguridad "
|
|
"y restáurala donde los quieras.")
|
|
self.cartera.cerrar()
|
|
try:
|
|
instalacion.mover(modo)
|
|
except (OSError, ValueError) as e:
|
|
raise ErrorOperacion(f"No he podido cambiar los datos de sitio: {e}") from None
|
|
finally:
|
|
# Haya ido bien o mal, se vuelve a abrir la cartera donde haya quedado
|
|
base = instalacion.base_actual()[0]
|
|
self.carpeta = base / "datos"
|
|
self.cartera = Cartera(self.carpeta)
|
|
self.sync = Sincronizador(base / "sync", ajustes=base)
|
|
return self.instalacion()
|
|
|
|
def _leer_sesion(self) -> dict | None:
|
|
try:
|
|
return json.loads(almacen.leer_texto(self.cartera.carpeta / SESION))
|
|
except (OSError, ValueError):
|
|
return None
|
|
|
|
def guardar_sesion(self, valor: float, neto: float = 0.0) -> None:
|
|
ruta = self.cartera.carpeta / SESION
|
|
try:
|
|
almacen.escribir(ruta, json.dumps({"fecha": dt.date.today().isoformat(), "valor": valor, "neto": neto}).encode())
|
|
except OSError:
|
|
pass
|
|
|
|
# ── API ──
|
|
def estado(self) -> dict:
|
|
datos = self.cartera.datos
|
|
hoy = dt.date.today()
|
|
r = resumen(datos, hoy, vpo.serie(self.carpeta.parent))
|
|
precios = datos.vivos("precios")
|
|
auto = [p for p in precios if p.get("fuente") not in ("manual",) and not str(p.get("fuente", "")).startswith("xlsx")]
|
|
ajustes = self._ajustes()
|
|
return {
|
|
"resumen": r,
|
|
"inmuebles": self._listas_inmuebles(),
|
|
"activos": datos.vivos("activos"),
|
|
"movimientos": sorted(datos.vivos("movimientos"),
|
|
key=lambda m: (m["fecha"], m.get("orden") or 0), reverse=True),
|
|
"recurrentes": datos.vivos("recurrentes"),
|
|
"ultima_descarga_precios": max((p.get("modificado", "")[:10] for p in auto), default=None),
|
|
"ultima_sesion": self.ultima_sesion,
|
|
"sync": self.sync.estado(datos),
|
|
"btc": self.btc_resumen(),
|
|
# Cuánto hay en cada lugar de los activos que los tienen (ver lugares.py)
|
|
"lugares": {a["id"]: lugares.saldos(datos.vivos("movimientos"), a["id"], hoy.isoformat())
|
|
for a in datos.vivos("activos") if a.get("lugares")},
|
|
# Compras y ventas de bitcoin frente al mercado de ese día (ver prima.py)
|
|
# (en el ejemplo no: sus precios son inventados y no casan con los de ningún día de verdad)
|
|
"primas": {} if self._real is not None else prima.de_movimientos(datos.vivos("activos"), datos.vivos("movimientos"), precios),
|
|
"carpeta": str(self.cartera.carpeta),
|
|
"ejemplo": self._real is not None,
|
|
# Los formatos de extracto que esta instalación sabe leer (ver extractos.registrar)
|
|
"extractos": sorted(extractos.NOMBRES.values()),
|
|
"servidor": self.acceso is not None,
|
|
# Cambió la contraseña y no hay copia de seguridad con la nueva (ver acceso.py)
|
|
"copia_pendiente": self.acceso is not None and self.acceso.copia_pendiente(),
|
|
"cifrado": self.acceso is not None and self.acceso.cifrado(),
|
|
# Con qué servidor sincroniza este Caudal, y qué equipos sincronizan con él si el servidor es este
|
|
"enlace": {k: v for k, v in enlace.Enlace(self.carpeta).config().items() if k in ("url", "url_tor", "ultima")},
|
|
"equipos": self.acceso.enlaces() if self.acceso is not None else [],
|
|
"disco_cifrado": DISCO_YA_CIFRADO,
|
|
"recuperacion": self.acceso is not None and self.acceso.tiene_recuperacion(),
|
|
"hay_ejemplo": EJEMPLO.is_dir(),
|
|
"instalacion": self.instalacion(),
|
|
# El primer uso: qué dijo que tiene, si cerró la lista de primeros pasos y cuándo bajó una copia
|
|
"primer_uso": ajustes.get("primer_uso") or {},
|
|
"territorio": fiscal.opciones_validas(ajustes.get("fiscal"))["territorio"],
|
|
"territorio_elegido": "territorio" in (ajustes.get("fiscal") or {}),
|
|
"territorios": fiscal.cuota.TERRITORIOS,
|
|
# Justificantes apuntados cuyo fichero no está en este equipo
|
|
"adjuntos_faltan": sorted(h for h in adjuntos.referenciados(datos.movimientos)
|
|
if not adjuntos.tiene(self.cartera.carpeta, h)),
|
|
}
|
|
|
|
def _guardar_activo(self, cuerpo: dict) -> dict:
|
|
"""Guarda un activo; si es un inmueble con renta mensual, sus cobros ya
|
|
vencidos se apuntan en el momento, sin esperar a la siguiente apertura."""
|
|
a = self.cartera.guardar_activo(cuerpo)
|
|
if a["tipo"] == "Inmueble" and (a.get("inmueble") or {}).get("renta_mensual"):
|
|
self.cartera.generar_recurrentes(dt.date.today(), fuentes.cambio)
|
|
return a
|
|
|
|
def _listas_inmuebles(self) -> dict:
|
|
"""Lo que la ficha de un inmueble ofrece para elegir, con las modalidades
|
|
de alquiler del territorio donde declara el usuario."""
|
|
from .fiscal.alquiler import CLASES_GASTO
|
|
territorio = fiscal.opciones_validas(self._ajustes().get("fiscal"))["territorio"]
|
|
reglas = fiscal.cuota.reglas(territorio).get("alquiler") or {}
|
|
actuales = fiscal.alquiler.reglas_del_año(reglas, dt.date.today().year) if reglas else {}
|
|
return {"clases": inmuebles.CLASES, "usos": inmuebles.USOS,
|
|
"comunidades": inmuebles.comunidades(inmuebles.serie(self.carpeta.parent)),
|
|
"territorio": territorio, "nombre_territorio": fiscal.cuota.TERRITORIOS[territorio],
|
|
"modalidades": {k: v["texto"] for k, v in (actuales.get("modalidades") or {}).items()},
|
|
"reducciones": reglas.get("reducciones") or [], "imputa": bool(reglas.get("imputacion")),
|
|
"clases_gasto": CLASES_GASTO, "tierra": self._listas_tierra(), "vpo": self._listas_vpo()}
|
|
|
|
def _listas_vpo(self) -> dict:
|
|
"""Para la ficha de una vivienda protegida: regímenes, piezas de partida
|
|
y el módulo de este año en cada comunidad que lo trae (oficial o estimado)."""
|
|
s = vpo.serie(self.carpeta.parent)
|
|
año = dt.date.today().year
|
|
modulos = {}
|
|
for comunidad in s["modulos"]:
|
|
m = vpo.modulo({"comunidad": comunidad}, año, s)
|
|
if m:
|
|
modulos[comunidad] = {"valor": m[0], "origen": m[1]}
|
|
return {"regimenes": {k: {"nombre": v["nombre"], "vivienda": v["vivienda"], "anejo": v["anejo"]} for k, v in s["regimenes"].items()},
|
|
"clases_pieza": vpo.CLASES_PIEZA, "piezas_iniciales": vpo.PIEZAS_INICIALES, "año": año, "modulos": modulos}
|
|
|
|
@staticmethod
|
|
def _listas_tierra() -> dict:
|
|
"""Para la ficha de una finca rústica: los cultivos y el precio medio
|
|
del último año publicado por comunidad (la web avisa si sale el nacional)."""
|
|
from . import tierra
|
|
s, año = tierra.serie(), tierra.ultimo_año()
|
|
return {"cultivos": tierra.cultivos(), "origenes": tierra.ORIGENES, "año": año, "fuente": s["fuente"],
|
|
"precios": {ccaa: {c: tierra.precio(ccaa, c, año)[0] for c in datos}
|
|
for ccaa, datos in s["precios"].items()}}
|
|
|
|
def fiscal(self, ejercicio: str = "") -> dict:
|
|
"""Lo que hay que mirar para declarar un ejercicio (ver cartera/fiscal).
|
|
Sin ejercicio: el último año ya cerrado que tenga operaciones."""
|
|
datos = self.cartera.datos
|
|
movimientos = datos.vivos("movimientos")
|
|
años = fiscal.ejercicios(movimientos)
|
|
try:
|
|
año = int(ejercicio)
|
|
except (TypeError, ValueError):
|
|
pasado = dt.date.today().year - 1
|
|
año = pasado if pasado in años or not años else años[0]
|
|
if not 1990 <= año <= dt.date.today().year:
|
|
raise ErrorOperacion("Ese ejercicio no es válido.")
|
|
ajustes = self._ajustes()
|
|
elegidos = ajustes.get("fiscal") or {}
|
|
r = fiscal.completo(datos.vivos("activos"), movimientos, datos.vivos("precios"), año, elegidos,
|
|
trading=ajustes.get("trading"))
|
|
# Mientras nadie haya dicho dónde declara, la pantalla lo pregunta: las cifras dependen de ello
|
|
r["territorio_elegido"] = "territorio" in elegidos
|
|
return r
|
|
|
|
# Lo que se vende y cómo se apunta, para «si la vendieras hoy» (lo demás no tributa al venderlo:
|
|
# una cuenta, un préstamo o un plan de pensiones, que se cobra como sueldo)
|
|
VENDIBLES = {"Cripto": "Venta", "Accion": "Venta", "ETF": "Venta", "Fondo": "Reembolso", "Metal": "Venta",
|
|
"Tierra": "Venta", "Inmueble": "Venta", "RentaFija": "Reembolso"}
|
|
|
|
def si_vendo(self, aid: str) -> dict:
|
|
"""Si se vendiera hoy toda una inversión al precio de hoy: la ganancia
|
|
según Hacienda y lo que subiría (o bajaría) la cuota del ahorro de este
|
|
año, calculado con todo lo demás del año y las pérdidas pendientes."""
|
|
datos = self.cartera.datos
|
|
a = self.cartera.activo(aid)
|
|
tipo = self.VENDIBLES.get(a["tipo"])
|
|
hoy = dt.date.today()
|
|
fila = next((f for f in resumen(datos, hoy)["abiertos"] if f["id"] == aid), None)
|
|
if not tipo or fila is None or not fila.get("valor") or (inmuebles.es_propio(a)):
|
|
return {"posible": False}
|
|
movs = datos.vivos("movimientos")
|
|
venta = {"id": "si-vendo", "activo": aid, "tipo": tipo, "fecha": hoy.isoformat(), "importe": round(fila["valor"], 2),
|
|
"orden": 10 ** 9}
|
|
if a["tipo"] == "RentaFija":
|
|
from . import rentafija
|
|
venta["cantidad"] = round(rentafija.coste_vivo([m for m in movs if m["activo"] == aid], (hoy + dt.timedelta(days=1)).isoformat(),
|
|
float((a.get("renta_fija") or {}).get("nominal") or 0))[1], 2)
|
|
elif a.get("valoracion") != "total" and a["tipo"] != "Inmueble":
|
|
venta.update({"cantidad": fila["cantidad"], "precio": fila.get("precio")})
|
|
ajustes = self._ajustes()
|
|
elegidos = ajustes.get("fiscal") or {}
|
|
args = (datos.vivos("activos"),)
|
|
sin = fiscal.completo(*args, movs, datos.vivos("precios"), hoy.year, elegidos, trading=ajustes.get("trading"))
|
|
con = fiscal.completo(*args, movs + [venta], datos.vivos("precios"), hoy.year, elegidos, trading=ajustes.get("trading"))
|
|
ganancia = round(con["cuota"]["ganancias"] + con["cuota"]["rendimientos"] - sin["cuota"]["ganancias"] - sin["cuota"]["rendimientos"], 2)
|
|
return {"posible": True, "valor": round(fila["valor"], 2), "ganancia": ganancia,
|
|
"impuesto": round(con["cuota"]["cuota"] - sin["cuota"]["cuota"], 2),
|
|
"tipo_medio": round((con["cuota"]["cuota"] - sin["cuota"]["cuota"]) / ganancia, 4) if ganancia > 0 else None,
|
|
"territorio": con["cuota"]["nombre_territorio"], "ejercicio": hoy.year,
|
|
"pendientes_antes": round(sum(p["importe"] for p in sin["cuota"]["pendientes"]), 2)}
|
|
|
|
def fiscal_xlsx(self, ejercicio: str = "") -> tuple[int, bytes]:
|
|
"""(ejercicio, libro) con el informe fiscal de ese año en una hoja de cálculo."""
|
|
from .fiscal.exportar import exportar
|
|
r = self.fiscal(ejercicio)
|
|
with tempfile.TemporaryDirectory() as t:
|
|
destino = Path(t) / "fiscal.xlsx"
|
|
exportar(r, destino)
|
|
return r["ejercicio"], destino.read_bytes()
|
|
|
|
def fiscal_regularizacion(self) -> dict:
|
|
"""Lo que costaría declarar ahora lo no declarado (ver fiscal/regularizar.py)."""
|
|
datos, ajustes = self.cartera.datos, self._ajustes()
|
|
return fiscal.regularizacion(datos.vivos("activos"), datos.vivos("movimientos"), datos.vivos("precios"),
|
|
dt.date.today(), ajustes.get("fiscal"), ajustes.get("regularizacion"),
|
|
trading=ajustes.get("trading"))
|
|
|
|
def fiscal_trading(self, cuerpo: dict) -> dict:
|
|
"""Guarda el resultado de un año de una cuenta de bróker, copiado de su
|
|
informe fiscal (ganancia, o pérdida en negativo). Vacío = quitarlo."""
|
|
año, aid, valor = str(cuerpo.get("ejercicio") or ""), str(cuerpo.get("activo") or ""), cuerpo.get("resultado")
|
|
activo = next((a for a in self.cartera.datos.vivos("activos") if a["id"] == aid), None)
|
|
if not año.isdigit() or activo is None or activo["tipo"] != "CuentaTrading":
|
|
raise ErrorOperacion("Petición no válida.")
|
|
ajustes = self._ajustes()
|
|
todo = {k: dict(v) for k, v in (ajustes.get("trading") or {}).items()}
|
|
fila = todo.setdefault(año, {})
|
|
if valor in (None, ""):
|
|
fila.pop(aid, None)
|
|
else:
|
|
try:
|
|
importe = round(float(str(valor).replace(",", ".")), 2)
|
|
except ValueError:
|
|
raise ErrorOperacion("Pon el resultado en euros (con un menos delante si fue pérdida).") from None
|
|
if abs(importe) > 1e9:
|
|
raise ErrorOperacion("Ese importe no es válido.")
|
|
fila[aid] = importe
|
|
if not fila:
|
|
todo.pop(año)
|
|
ajustes["trading"] = todo
|
|
self._guardar_ajustes(ajustes)
|
|
return {"ok": True}
|
|
|
|
def fiscal_declarado(self, cuerpo: dict) -> dict:
|
|
"""Guarda lo que el usuario dice de un ejercicio: lo que ya pagó por el
|
|
ahorro y su tipo marginal. Vacío = quitarlo."""
|
|
año = str(cuerpo.get("ejercicio") or "")
|
|
if not año.isdigit():
|
|
raise ErrorOperacion("Petición no válida.")
|
|
ajustes = self._ajustes()
|
|
todo = dict(ajustes.get("regularizacion") or {})
|
|
fila = {k: cuerpo.get(k) for k in ("pagado", "marginal") if cuerpo.get(k) not in (None, "")}
|
|
nuevo = fiscal.regularizar.limpiar({año: fila})
|
|
if len(nuevo.get(año, {})) != len(fila):
|
|
raise ErrorOperacion("Pon un importe en euros y un tipo entre 0 y 60 %.")
|
|
todo.pop(año, None)
|
|
todo.update(nuevo)
|
|
ajustes["regularizacion"] = todo
|
|
self._guardar_ajustes(ajustes)
|
|
return self.fiscal_regularizacion()
|
|
|
|
def fiscal_criterios(self, cuerpo: dict) -> dict:
|
|
"""Guarda los criterios elegidos en la pestaña Fiscal (viajan con los ajustes)."""
|
|
if not isinstance(cuerpo.get("opciones"), dict):
|
|
raise ErrorOperacion("Petición no válida.")
|
|
ajustes = self._ajustes()
|
|
ajustes["fiscal"] = fiscal.opciones_validas({**(ajustes.get("fiscal") or {}), **cuerpo["opciones"]})
|
|
self._guardar_ajustes(ajustes)
|
|
return ajustes["fiscal"]
|
|
|
|
PRIMER_USO_TIENE = ("bitcoin", "fondos", "acciones", "pension", "rentafija", "inmuebles", "cuentas", "broker")
|
|
|
|
def primer_uso(self, cuerpo: dict) -> dict:
|
|
"""Las respuestas de la bienvenida: qué tiene (para su lista de primeros
|
|
pasos) y dónde declara; también cerrar esa lista. Viajan con los ajustes."""
|
|
ajustes = self._ajustes()
|
|
p = dict(ajustes.get("primer_uso") or {})
|
|
if "tiene" in cuerpo:
|
|
tiene = cuerpo.get("tiene")
|
|
if not isinstance(tiene, list):
|
|
raise ErrorOperacion("Petición no válida.")
|
|
p["tiene"] = [t for t in self.PRIMER_USO_TIENE if t in tiene]
|
|
if "cerrado" in cuerpo:
|
|
p["cerrado"] = bool(cuerpo.get("cerrado"))
|
|
if cuerpo.get("territorio"):
|
|
if cuerpo["territorio"] not in fiscal.cuota.TERRITORIOS:
|
|
raise ErrorOperacion("Ese territorio no lo conozco.")
|
|
ajustes["fiscal"] = fiscal.opciones_validas({**(ajustes.get("fiscal") or {}), "territorio": cuerpo["territorio"]})
|
|
ajustes["primer_uso"] = p
|
|
self._guardar_ajustes(ajustes)
|
|
return p
|
|
|
|
def unir_lugares(self, cuerpo: dict) -> dict:
|
|
"""Junta dos lugares; los monederos y los exchanges recordados que
|
|
apuntaban al que desaparece pasan al que se queda."""
|
|
aid, desde, hacia = (str(cuerpo.get(k) or "") for k in ("activo", "desde", "hacia"))
|
|
with self._btc_lock:
|
|
r = self.cartera.unir_lugares(aid, desde, hacia)
|
|
config = self._btc_config()
|
|
monederos = [m for m in config.get("monederos") or [] if m.get("activo") == aid and m.get("lugar") == desde]
|
|
for m in monederos:
|
|
m["lugar"] = hacia
|
|
if monederos:
|
|
self._btc_guardar(config)
|
|
ajustes = self._ajustes()
|
|
destinos = ajustes.get("destinos_exchange")
|
|
if isinstance(destinos, dict) and desde in destinos.values():
|
|
ajustes["destinos_exchange"] = {k: (hacia if v == desde else v) for k, v in destinos.items()}
|
|
self._guardar_ajustes(ajustes)
|
|
return {**r, "monederos": len(monederos)}
|
|
|
|
def convertir_en_lugar(self, cuerpo: dict) -> dict:
|
|
"""Un activo que es un sitio del mismo bitcoin pasa a ser un lugar de
|
|
otro; sus monederos del nodo pasan también al activo de destino."""
|
|
origen, destino = str(cuerpo.get("id") or ""), str(cuerpo.get("destino") or "")
|
|
with self._btc_lock:
|
|
lugar = self.cartera.convertir_en_lugar(origen, destino, cuerpo.get("clase") or "otro")
|
|
config = self._btc_config()
|
|
movidos = 0
|
|
for m in config.get("monederos") or []:
|
|
if m.get("activo") == origen:
|
|
m.update({"activo": destino, "lugar": lugar["id"]})
|
|
movidos += 1
|
|
if movidos:
|
|
self._btc_guardar(config)
|
|
return {"lugar": lugar, "monederos": movidos}
|
|
|
|
def adjuntar(self, cuerpo: dict) -> dict:
|
|
try:
|
|
contenido = base64.b64decode(str(cuerpo.get("contenido") or ""), validate=True)
|
|
except (binascii.Error, ValueError):
|
|
raise ErrorOperacion("El fichero no ha llegado bien; vuelve a elegirlo.") from None
|
|
return self.cartera.adjuntar(cuerpo.get("id"), cuerpo.get("nombre"), contenido)
|
|
|
|
def actualizar_precios(self) -> dict:
|
|
if self._real is not None:
|
|
raise ErrorOperacion("En la cartera de ejemplo los precios son inventados y no se actualizan.")
|
|
if not self._precios_lock.acquire(blocking=False):
|
|
raise ErrorOperacion("Ya se están actualizando los precios.")
|
|
try:
|
|
informe = fuentes.actualizar(self.cartera.datos.vivos("activos"))
|
|
# Los inmuebles con valor estimado: el índice del INE más reciente (se guarda junto a los ajustes)
|
|
if any(a["tipo"] == "Inmueble" and (a.get("inmueble") or {}).get("auto") for a in self.cartera.datos.vivos("activos")):
|
|
try:
|
|
nueva = inmuebles.descargar(fuentes._sesion())
|
|
(self.carpeta.parent / inmuebles.FICHERO_ACTUALIZADO).write_text(
|
|
json.dumps(nueva, ensure_ascii=False, separators=(",", ":")), encoding="utf-8")
|
|
informe.append({"activo": "", "nombre": "Índice de precios de vivienda (INE)", "ok": True, "resultado": None,
|
|
"mensaje": f"hasta {nueva['series']['nacional']['segunda_mano'][-1][0]}"})
|
|
except Exception as e: # sin INE se sigue con la serie que haya
|
|
informe.append({"activo": "", "nombre": "Índice de precios de vivienda (INE)", "ok": False,
|
|
"resultado": None, "mensaje": f"no se pudo descargar ({e})"})
|
|
# Las viviendas protegidas por metros: el IPC, para estimar el módulo de los años aún sin publicar
|
|
if any(vpo.es_por_metros(a.get("inmueble") or {}) for a in self.cartera.datos.vivos("activos") if a["tipo"] == "Inmueble"):
|
|
try:
|
|
nuevo = vpo.descargar_ipc(fuentes._sesion())
|
|
(self.carpeta.parent / vpo.FICHERO_IPC).write_text(
|
|
json.dumps(nuevo, ensure_ascii=False, separators=(",", ":")), encoding="utf-8")
|
|
informe.append({"activo": "", "nombre": "IPC (INE), para el módulo de vivienda protegida", "ok": True,
|
|
"resultado": None, "mensaje": f"hasta {nuevo['datos'][-1][0]}"})
|
|
except Exception as e: # sin INE se sigue con el que haya
|
|
informe.append({"activo": "", "nombre": "IPC (INE), para el módulo de vivienda protegida", "ok": False,
|
|
"resultado": None, "mensaje": f"no se pudo descargar ({e})"})
|
|
self.cartera.aplicar_precios([i["resultado"] for i in informe if i["ok"] and i["resultado"]])
|
|
informe += self._precios_renta_fija()
|
|
gen = self.cartera.generar_recurrentes(dt.date.today(), fuentes.cambio)
|
|
return {"informe": informe, "automaticas": gen}
|
|
finally:
|
|
self._precios_lock.release()
|
|
|
|
def _precios_renta_fija(self) -> list[dict]:
|
|
"""Bonos, obligaciones y Letras con ISIN: su precio en BME (AIAF o SENAF).
|
|
Sin cotización, su valor sigue calculándose solo (devengo)."""
|
|
from . import bme, rentafija
|
|
datos = self.cartera.datos
|
|
suyos = [a for a in datos.vivos("activos") if a["tipo"] == "RentaFija" and a.get("estado") != "cerrada"
|
|
and bme.ISIN.match((a.get("codigo") or "").strip().upper())]
|
|
if not suyos:
|
|
return []
|
|
try:
|
|
precios = bme.consultar([a["codigo"] for a in suyos], fuentes._sesion())
|
|
except Exception as e: # BME caído o cambiado: se sigue con el valor calculado
|
|
return [{"activo": a["id"], "nombre": a["nombre"], "ok": False, "resultado": None,
|
|
"mensaje": f"BME no contestó ({e})"} for a in suyos]
|
|
informe, valores = [], []
|
|
for a in suyos:
|
|
p = precios.get(a["codigo"].strip().upper())
|
|
movs = [m for m in datos.vivos("movimientos") if m["activo"] == a["id"]]
|
|
v = rentafija.valor_de_mercado(a, movs, dt.date.fromisoformat(p["fecha"]), p["precio"]) if p else None
|
|
if v is None:
|
|
informe.append({"activo": a["id"], "nombre": a["nombre"], "ok": False, "resultado": None,
|
|
"mensaje": "no cotiza hoy en BME: sigue el valor calculado"})
|
|
continue
|
|
valores.append({"activo": a["id"], "fecha": p["fecha"], "valor_total": v, "fuente": rentafija.FUENTE_MERCADO})
|
|
informe.append({"activo": a["id"], "nombre": a["nombre"], "ok": True, "resultado": None,
|
|
"mensaje": f"{p['precio']} % ({p['origen']}, {p['mercado']})"})
|
|
if valores:
|
|
self.cartera.aplicar_valores(valores)
|
|
return informe
|
|
|
|
def comprobar_fuentes(self) -> dict:
|
|
"""Prueba una por una las webs de las que salen los precios.
|
|
|
|
Sale a internet, pero solo con los identificadores públicos de ejemplo
|
|
que lleva escritos `precios.py`: no consulta ningún activo de la cartera.
|
|
"""
|
|
if not self._precios_lock.acquire(blocking=False):
|
|
raise ErrorOperacion("Espera a que terminen de actualizarse los precios.")
|
|
try:
|
|
return fuentes.comprobar()
|
|
finally:
|
|
self._precios_lock.release()
|
|
|
|
def interpretar(self, cuerpo: dict) -> list[dict]:
|
|
from .interprete import interpretar
|
|
texto = str(cuerpo.get("texto") or "")[:1000]
|
|
if not texto.strip():
|
|
raise ErrorOperacion("Escribe qué has hecho.")
|
|
cache: dict = {}
|
|
|
|
def cambio(fecha, moneda): # solo consulta internet si hay dólares
|
|
if (fecha, moneda) not in cache:
|
|
cache[(fecha, moneda)] = fuentes.cambio(fecha, moneda)
|
|
return cache[(fecha, moneda)]
|
|
|
|
nuevo = cuerpo.get("activo_nuevo")
|
|
if nuevo is not None:
|
|
if not isinstance(nuevo, dict) or nuevo.get("tipo") not in TIPOS_ACTIVO or not nuevo.get("nombre"):
|
|
raise ErrorOperacion("Activo nuevo no válido.")
|
|
nuevo = {k: nuevo.get(k) for k in ("nombre", "codigo", "tipo", "unidad", "precio", "fecha_precio", "fuente", "alias")}
|
|
if not nuevo.get("precio"):
|
|
encontrado, _ = fuentes.precio({"id": "nuevo", **nuevo})
|
|
if encontrado:
|
|
nuevo.update({"precio": encontrado["precio"], "fecha_precio": encontrado["fecha"],
|
|
"fuente": encontrado["fuente"]})
|
|
propuestas = interpretar(texto, self.cartera.datos, dt.date.today(), cambio,
|
|
cuerpo.get("activo") or None, nuevo)
|
|
for p in propuestas:
|
|
if p.get("busqueda") and nuevo is None:
|
|
try:
|
|
encontrados = fuentes.buscar(p["busqueda"])
|
|
p["nuevos"] = encontrados["resultados"][:6]
|
|
if not p["nuevos"] and encontrados["errores"]:
|
|
p["avisos"].append("No he podido buscar activos nuevos en internet.")
|
|
except Exception:
|
|
p["avisos"].append("No he podido buscar activos nuevos en internet.")
|
|
return propuestas
|
|
|
|
# ── sincronización cifrada ──
|
|
def sync_claves(self) -> dict:
|
|
sugerencia = ""
|
|
try:
|
|
import subprocess
|
|
url = subprocess.run(["git", "-C", str(PROYECTO.parent), "remote", "get-url", "origin"],
|
|
capture_output=True, text=True, timeout=10).stdout.strip()
|
|
if "cartera-app" in url:
|
|
sugerencia = url.replace("cartera-app", "cartera-datos")
|
|
except OSError:
|
|
pass
|
|
return {"claves": self.sync.claves(), "sugerencia": sugerencia, "config": self.sync.config()}
|
|
|
|
def _copia_datos(self) -> None:
|
|
import tarfile
|
|
copias = self.cartera.carpeta.parent / "backups"
|
|
copias.mkdir(parents=True, exist_ok=True)
|
|
nombre = copias / f"datos-{dt.datetime.now().strftime('%Y%m%d-%H%M%S')}-sync.tar.gz"
|
|
# Sincronizar también puede cambiar los ajustes: entran en la copia
|
|
ficheros = sorted(self.cartera.carpeta.glob("*.jsonl")) + [self.cartera.carpeta / FICHERO_VERSION] \
|
|
+ [self.cartera.carpeta.parent / ajuste for ajuste in AJUSTES]
|
|
bruto = io.BytesIO()
|
|
with tarfile.open(fileobj=bruto, mode="w:gz") as tar:
|
|
for f in ficheros:
|
|
if f.exists():
|
|
contenido = almacen.leer(f) # dentro va en claro; lo que se cifra, si toca, es la copia entera
|
|
info = tarfile.TarInfo(f.name)
|
|
info.size, info.mtime = len(contenido), int(f.stat().st_mtime)
|
|
tar.addfile(info, io.BytesIO(contenido))
|
|
almacen.escribir(nombre, bruto.getvalue())
|
|
|
|
def sync_ejecutar(self, accion: str) -> dict:
|
|
if not self._sync_lock.acquire(blocking=False):
|
|
raise ErrorOperacion("Ya hay una sincronización en marcha.")
|
|
try:
|
|
if accion == "comprobar":
|
|
return self.sync.comprobar(self.cartera.datos)
|
|
return self.sync.sincronizar(self.cartera, self._copia_datos)
|
|
finally:
|
|
self._sync_lock.release()
|
|
|
|
# ── extractos ──
|
|
def _ajustes(self) -> dict:
|
|
ruta = self.cartera.carpeta.parent / "extractos.json"
|
|
try:
|
|
return json.loads(almacen.leer_texto(ruta))
|
|
except (OSError, ValueError):
|
|
return {}
|
|
|
|
def _guardar_ajustes(self, ajustes: dict) -> None:
|
|
# Las marcas dicen qué cambió el usuario y cuándo: deciden al sincronizar (ver marcas.py)
|
|
ajustes = marcas.marcar(self._ajustes() or None, ajustes, "extractos.json")
|
|
try:
|
|
escribir_ajuste(self.cartera.carpeta.parent / "extractos.json", ajustes)
|
|
except OSError:
|
|
pass
|
|
|
|
def extracto_analizar(self, cuerpo: dict) -> dict:
|
|
texto = cuerpo.get("texto") or ""
|
|
if not texto.strip():
|
|
raise ErrorOperacion("El fichero está vacío.")
|
|
# La cuenta de los intereses se recuerda por entidad: los de un banco
|
|
# no van a la cuenta del bróker
|
|
ajustes = self._ajustes()
|
|
por_entidad = ajustes.get("cuenta_intereses")
|
|
mapa = cuerpo.get("mapa") if isinstance(cuerpo.get("mapa"), dict) else None
|
|
lector = extractos.detectar(texto) or (extractos.GENERICO if mapa else None)
|
|
if not isinstance(por_entidad, dict):
|
|
# Versión anterior: una sola cuenta para todo. Se le queda a la entidad del primer extracto que llegue
|
|
por_entidad = {lector: por_entidad} if por_entidad and lector else {}
|
|
if por_entidad: # y se apunta ya, para que no pase a la siguiente
|
|
ajustes["cuenta_intereses"] = por_entidad
|
|
self._guardar_ajustes(ajustes)
|
|
cuenta = cuerpo.get("cuenta") or por_entidad.get(lector)
|
|
if cuerpo.get("cuenta") and lector:
|
|
por_entidad[lector] = cuerpo["cuenta"]
|
|
ajustes["cuenta_intereses"] = por_entidad
|
|
self._guardar_ajustes(ajustes)
|
|
# A qué activo va cada exchange del extracto (o «no importar»), también recordado
|
|
destinos = ajustes.get("destinos_exchange")
|
|
destinos = dict(destinos) if isinstance(destinos, dict) else {}
|
|
if isinstance(cuerpo.get("destinos"), dict):
|
|
destinos.update({str(k)[:80]: str(v)[:40] for k, v in cuerpo["destinos"].items()})
|
|
ajustes["destinos_exchange"] = destinos
|
|
self._guardar_ajustes(ajustes)
|
|
try:
|
|
return extractos.analizar(texto, self.cartera.datos, cuenta, destinos, mapa)
|
|
except ValueError as e:
|
|
raise ErrorOperacion(str(e)) from None
|
|
|
|
def extracto_columnas(self, cuerpo: dict) -> dict:
|
|
"""Si el extracto es de un formato conocido, lo dice; si no, da lo
|
|
necesario para que el usuario reparta sus columnas (ver extractos.columnas)."""
|
|
texto = cuerpo.get("texto") or ""
|
|
if not texto.strip():
|
|
raise ErrorOperacion("El fichero está vacío.")
|
|
lector = extractos.detectar(texto)
|
|
if lector:
|
|
return {"reconocido": extractos.NOMBRES[lector]}
|
|
try:
|
|
return {"reconocido": None, **extractos.columnas(texto)}
|
|
except ValueError as e:
|
|
raise ErrorOperacion(str(e)) from None
|
|
|
|
def extracto_aplicar(self, cuerpo: dict) -> dict:
|
|
propuestas = cuerpo.get("propuestas") or []
|
|
if not isinstance(propuestas, list):
|
|
raise ErrorOperacion("Petición no válida.")
|
|
creados, guardados, actualizados, fallos = {}, 0, 0, []
|
|
propuestas = sorted(propuestas, key=lambda p: ((p or {}).get("movimiento") or {}).get("fecha") or "")
|
|
for p in propuestas[:2000]:
|
|
if (p or {}).get("estado") in ("en_historial", "sin_precio", "incompleta", "falta_cuenta", "ya_estaba"):
|
|
continue # no se guardan desde aquí
|
|
movimiento = dict((p or {}).get("movimiento") or {})
|
|
lugar_nuevo = movimiento.pop("lugar_nuevo", None)
|
|
nuevo = (p or {}).get("activo_nuevo")
|
|
if nuevo:
|
|
isin = (nuevo.get("isin") or "").upper()
|
|
nombre = (nuevo.get("nombre") or "").strip()
|
|
clave = isin or extractos.sin_tildes(nombre)
|
|
if clave and clave not in creados:
|
|
# Los extractos de fondos y planes no siempre traen ISIN: se
|
|
# busca por código y, si no hay, por nombre
|
|
ya = next((a for a in self.cartera.datos.vivos("activos")
|
|
if (isin and (a.get("codigo") or "").upper() == isin)
|
|
or (not isin and extractos.sin_tildes(a.get("nombre", "")) == clave)), None)
|
|
if ya is None:
|
|
tipo = nuevo.get("tipo") or "Accion"
|
|
codigo = isin or str(nuevo.get("codigo") or "")[:40]
|
|
ya = self.cartera.guardar_activo({
|
|
"nombre": nombre or isin, "tipo": tipo, "codigo": codigo,
|
|
"unidad": str(nuevo.get("unidad") or "")[:12]
|
|
or ("acciones" if tipo == "Accion" else "participaciones"),
|
|
"fecha_inicio": movimiento.get("fecha")})
|
|
if codigo:
|
|
r, _ = fuentes.precio(ya)
|
|
if r:
|
|
self.cartera.aplicar_precios([r])
|
|
creados[clave] = ya["id"]
|
|
movimiento["activo"] = creados.get(clave)
|
|
if not movimiento.get("activo"):
|
|
continue
|
|
try:
|
|
if isinstance(lugar_nuevo, dict) and lugar_nuevo.get("nombre"):
|
|
lugar = self.cartera.asegurar_lugar(movimiento["activo"], str(lugar_nuevo["nombre"])[:60],
|
|
lugar_nuevo.get("clase") or "exchange")
|
|
lado = lugar_nuevo.get("en") if lugar_nuevo.get("en") in ("lugar", "lugar_destino") else "lugar"
|
|
movimiento[lado] = lugar["id"]
|
|
if p.get("estado") == "poner_lugar":
|
|
# Ya estaba apuntada: solo se le pone el lugar (y la línea, si no tenía)
|
|
self.cartera.poner_lugar([p.get("coincide_con")], movimiento.get("lugar") or "")
|
|
previo = next((m for m in self.cartera.datos.vivos("movimientos")
|
|
if m["id"] == p.get("coincide_con")), {})
|
|
if not previo.get("extracto") and movimiento.get("extracto"):
|
|
self.cartera.anotar_movimientos([previo["id"]], {"extracto": movimiento["extracto"]})
|
|
actualizados += 1
|
|
continue
|
|
self.cartera.guardar_movimiento(movimiento)
|
|
except ErrorOperacion as e:
|
|
# Una línea rara (p. ej. una venta cuyas compras no están) no
|
|
# debe impedir el resto de la importación
|
|
fallos.append({"fecha": movimiento.get("fecha"), "tipo": movimiento.get("tipo"),
|
|
"motivo": str(e)})
|
|
continue
|
|
if movimiento.get("id"):
|
|
actualizados += 1
|
|
else:
|
|
guardados += 1
|
|
return {"guardados": guardados, "actualizados": actualizados, "activos_nuevos": len(creados),
|
|
"fallos": fallos}
|
|
|
|
# ── bitcoin desde el nodo propio ──
|
|
def _btc_config(self) -> dict:
|
|
return btc.cargar(self.cartera.carpeta)
|
|
|
|
def _btc_guardar(self, config: dict) -> None:
|
|
try:
|
|
btc.guardar(self.cartera.carpeta, config)
|
|
except OSError as e:
|
|
raise ErrorOperacion(f"No he podido guardar la configuración de Bitcoin: {e}") from None
|
|
|
|
def btc_resumen(self) -> dict:
|
|
"""Lo justo para que la aplicación sepa si hay algo que mirar."""
|
|
config = self._btc_config()
|
|
return {"monederos": len(config.get("monederos") or []),
|
|
"servidores": len(config.get("servidores") or []),
|
|
"comprobar_al_abrir": bool(config.get("comprobar_al_abrir", True)),
|
|
"ultima": config.get("ultima")}
|
|
|
|
def btc_estado(self) -> dict:
|
|
config = self._btc_config()
|
|
monederos = [{k: v for k, v in m.items() if k not in ("clave", "vistas")}
|
|
| {"clave_corta": _recortar(m.get("clave")) if m.get("clave")
|
|
else f"{len(m.get('direcciones') or [])} direcciones sueltas",
|
|
"vistas": len(m.get("vistas") or [])}
|
|
for m in config.get("monederos") or []]
|
|
cripto = [{"id": a["id"], "nombre": a["nombre"], "lugares": lugares.de(a)}
|
|
for a in self.cartera.datos.vivos("activos") if a["tipo"] == "Cripto"]
|
|
return {"servidores": btc.sin_claves(config.get("servidores") or []), "monederos": monederos,
|
|
"proxy": config.get("proxy"), "comprobar_al_abrir": bool(config.get("comprobar_al_abrir", True)),
|
|
"ultima": config.get("ultima"), "activos": cripto,
|
|
"tipos": btc.bitcoin.NOMBRES_TIPO, "fichero": str(btc.ruta_config(self.cartera.carpeta)),
|
|
# El servidor Electrum del propio aparato, si quien empaqueta lo ha dicho (CAUDAL_ELECTRUM=host:puerto)
|
|
"sugerencia": None if config.get("servidores") else _electrum_del_aparato()}
|
|
|
|
def btc_servidores(self, cuerpo: dict) -> dict:
|
|
lista = cuerpo.get("servidores")
|
|
if not isinstance(lista, list):
|
|
raise ErrorOperacion("Petición no válida.")
|
|
config = self._btc_config()
|
|
guardados = config.get("servidores") or []
|
|
config["servidores"] = [btc.con_su_clave(btc.normalizar_servidor(s), guardados) for s in lista[:10]]
|
|
if cuerpo.get("proxy"):
|
|
proxy = cuerpo["proxy"]
|
|
config["proxy"] = {"host": str(proxy.get("host") or "127.0.0.1")[:60],
|
|
"puerto": int(proxy.get("puerto") or 9050)}
|
|
if "comprobar_al_abrir" in cuerpo:
|
|
config["comprobar_al_abrir"] = bool(cuerpo["comprobar_al_abrir"])
|
|
self._btc_guardar(config)
|
|
return self.btc_estado()
|
|
|
|
def btc_monedero(self, cuerpo: dict) -> dict:
|
|
config = self._btc_config()
|
|
monederos = config.get("monederos") or []
|
|
entrada = dict(cuerpo.get("monedero") or {})
|
|
if entrada.get("id") and not entrada.get("clave"):
|
|
anterior = next((m for m in monederos if m.get("id") == entrada["id"]), None)
|
|
if anterior is None:
|
|
raise ErrorOperacion("Ese monedero ya no está.")
|
|
if anterior.get("direcciones"):
|
|
# Al editar llega la lista entera; si no llega, se conserva
|
|
if entrada.get("direcciones") is None:
|
|
entrada["direcciones"] = anterior["direcciones"]
|
|
else:
|
|
entrada["clave"] = anterior.get("clave")
|
|
if entrada.get("clave") and not entrada.get("tipo"):
|
|
# Sin tipo elegido no se supone por el prefijo de la clave: se le pregunta al nodo qué direcciones usa
|
|
entrada["tipo"] = self._tipo_de_clave(str(entrada["clave"]).strip().replace(" ", ""), config)
|
|
nuevo = btc.normalizar_monedero(entrada, monederos)
|
|
config["monederos"] = [m for m in monederos if m.get("id") != nuevo["id"]] + [nuevo]
|
|
self._btc_guardar(config)
|
|
return self.btc_estado()
|
|
|
|
def _tipo_de_clave(self, clave: str, config: dict) -> str:
|
|
try:
|
|
extendida = btc.bitcoin.ClaveExtendida(clave)
|
|
except ValueError:
|
|
return "" # la clave no vale: que lo diga normalizar_monedero
|
|
try:
|
|
cliente, _, _ = btc.conectar([btc.normalizar_servidor(s) for s in config.get("servidores") or []],
|
|
btc.proxy_de(config))
|
|
except (ErrorBitcoin, ErrorElectrum, OSError):
|
|
return btc.tipo_de_partida(extendida) # sin nodo a mano: el más probable; se puede corregir después
|
|
try:
|
|
return btc.detectar(cliente, clave)[0]
|
|
except (ErrorBitcoin, ErrorElectrum, OSError):
|
|
return btc.tipo_de_partida(extendida)
|
|
finally:
|
|
cliente.cerrar()
|
|
|
|
def btc_borrar_monedero(self, mid: str) -> dict:
|
|
config = self._btc_config()
|
|
config["monederos"] = [m for m in (config.get("monederos") or []) if m.get("id") != mid]
|
|
self._btc_guardar(config)
|
|
return self.btc_estado()
|
|
|
|
def btc_probar(self, cuerpo: dict) -> dict:
|
|
return btc.probar(self._btc_config(), cuerpo.get("servidor") or {})
|
|
|
|
def btc_detectar(self, cuerpo: dict) -> dict:
|
|
config = self._btc_config()
|
|
clave = str(cuerpo.get("clave") or "").strip().replace(" ", "")
|
|
btc.bitcoin.ClaveExtendida(clave) # avisa pronto si la clave está mal
|
|
cliente, _, _ = btc.conectar([btc.normalizar_servidor(s) for s in config.get("servidores") or []],
|
|
btc.proxy_de(config))
|
|
try:
|
|
tipo, visto = btc.detectar(cliente, clave)
|
|
return {"tipo": tipo, "visto": visto}
|
|
finally:
|
|
cliente.cerrar()
|
|
|
|
def btc_revisar_monederos(self) -> dict:
|
|
"""Mira en el nodo qué tiene cada monedero (ver btc.revisar) y lo guarda."""
|
|
if not self._btc_lock.acquire(blocking=False):
|
|
raise ErrorOperacion("Ya estoy consultando el nodo.")
|
|
try:
|
|
config = self._btc_config()
|
|
monederos = [m for m in config.get("monederos") or [] if btc.vigilable(m)]
|
|
if not monederos:
|
|
raise ErrorOperacion("Todavía no has añadido ningún monedero.")
|
|
cliente, _, _ = btc.conectar([btc.normalizar_servidor(s) for s in config.get("servidores") or []],
|
|
btc.proxy_de(config))
|
|
try:
|
|
for m in monederos:
|
|
try:
|
|
m["revision"] = btc.revisar(cliente, m)
|
|
except (ValueError, btc.ErrorElectrum) as e:
|
|
m["revision"] = {"error": str(e)[:200]}
|
|
finally:
|
|
cliente.cerrar()
|
|
self._btc_guardar(config)
|
|
finally:
|
|
self._btc_lock.release()
|
|
return self.btc_estado()
|
|
|
|
def _lugares_de_monederos(self, config: dict) -> bool:
|
|
"""Cada monedero es un lugar de su activo (el del ordenador, el del móvil…): a los que no
|
|
tienen, se les crea uno con su nombre. Devuelve si ha cambiado algo."""
|
|
cambiado = False
|
|
activos = {a["id"]: a for a in self.cartera.datos.vivos("activos")}
|
|
for m in config.get("monederos") or []:
|
|
a = activos.get(m.get("activo"))
|
|
if a is None or (m.get("lugar") and m["lugar"] in lugares.por_id(a)):
|
|
continue
|
|
m["lugar"] = self.cartera.asegurar_lugar(a["id"], m.get("nombre") or "Monedero", "monedero")["id"]
|
|
activos[a["id"]] = self.cartera.activo(a["id"])
|
|
cambiado = True
|
|
if cambiado:
|
|
self._btc_guardar(config)
|
|
return cambiado
|
|
|
|
def _apuntado_en_monederos(self, config: dict, activo: str, hoy: dt.date) -> float:
|
|
"""Lo apuntado en los lugares que son monederos del nodo (lo de los
|
|
exchanges o los nodos Lightning no está en el nodo). Lo que no tiene
|
|
lugar (lo apuntado antes de haber lugares) cuenta como de los monederos."""
|
|
suyos = {m.get("lugar") for m in config.get("monederos") or [] if m.get("activo") == activo}
|
|
suyos.add(lugares.SIN_LUGAR)
|
|
saldos = lugares.saldos(self.cartera.datos.vivos("movimientos"), activo, hoy.isoformat())
|
|
return sum(q for lid, q in saldos.items() if lid in suyos)
|
|
|
|
def _btc_cuadres_por_activo(self, config: dict, r: dict, hoy: dt.date) -> list[dict]:
|
|
"""Lo que ve el nodo frente a lo apuntado, activo a activo (para el
|
|
aviso del resumen; el detalle por lugar va en la revisión)."""
|
|
en_nodo: dict[str, float] = {}
|
|
for c in r["cuadres"]:
|
|
en_nodo[c["activo"]] = en_nodo.get(c["activo"], 0.0) + c["en_nodo"]
|
|
salida = []
|
|
for aid, q in sorted(en_nodo.items()):
|
|
apuntado = self._apuntado_en_monederos(config, aid, hoy)
|
|
suyos = {m.get("nombre") or m.get("id") for m in config.get("monederos") or []
|
|
if m.get("activo") == aid}
|
|
salida.append({"activo": aid, "nombre_activo": self._nombre(aid),
|
|
"en_nodo": round(q, 9), "apuntado": round(apuntado, 9),
|
|
"diferencia": round(q - apuntado, 9),
|
|
"cortados": [n for n in r.get("cortados") or [] if n in suyos]})
|
|
return salida
|
|
|
|
def _btc_refrescar_ultima(self, config: dict) -> None:
|
|
"""Tras guardar o deshacer decisiones: lo que queda por revisar de la
|
|
última comprobación y lo apuntado, sin volver a preguntar al nodo."""
|
|
ultima = config.get("ultima")
|
|
if not ultima:
|
|
return
|
|
if "nuevas" in ultima:
|
|
resueltas = config.get("resueltas") or {}
|
|
ultima["nuevas"] = [i for i in ultima["nuevas"] if i not in resueltas]
|
|
ultima["pendientes"] = len(ultima["nuevas"])
|
|
hoy = dt.date.today()
|
|
for c in ultima.get("cuadres") or []:
|
|
c["apuntado"] = round(self._apuntado_en_monederos(config, c["activo"], hoy), 9)
|
|
c["diferencia"] = round(c["en_nodo"] - c["apuntado"], 9)
|
|
|
|
def btc_comprobar(self) -> dict:
|
|
"""La comprobación diaria: lo nuevo que ve el nodo, con el mismo motor
|
|
y la misma revisión que el historial completo."""
|
|
if not self._btc_lock.acquire(blocking=False):
|
|
raise ErrorOperacion("Ya estoy consultando el nodo.")
|
|
try:
|
|
config = self._btc_config()
|
|
self._lugares_de_monederos(config)
|
|
precios = Precios(self.cartera.datos.vivos("precios"))
|
|
hoy = dt.date.today()
|
|
try:
|
|
r = historial_btc.diaria(config, self.cartera.datos, hoy, lambda a: precios.de(a),
|
|
self.cartera.carpeta)
|
|
except OSError as e:
|
|
raise ErrorOperacion(f"No he podido guardar lo leído del nodo: {e}") from None
|
|
self._btc_presentar(config, r, hoy)
|
|
config["ultima"] = r["ultima"] = {
|
|
"fecha": dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds"),
|
|
"servidor": r["servidor"], "pendientes": len(r["partes"]),
|
|
"nuevas": [p["id"] for p in r["partes"]],
|
|
"cuadres": self._btc_cuadres_por_activo(config, r, hoy),
|
|
}
|
|
self._btc_guardar(config) # saldos, monederos estrenados y fecha
|
|
return r
|
|
finally:
|
|
self._btc_lock.release()
|
|
|
|
def btc_historial(self, cuerpo: dict | None = None) -> dict:
|
|
"""Todo el historial de los monederos, con lo que falta por apuntar.
|
|
Con «leer»: false usa lo último leído en este equipo, sin preguntar al nodo."""
|
|
if not self._btc_lock.acquire(blocking=False):
|
|
raise ErrorOperacion("Ya estoy consultando el nodo.")
|
|
try:
|
|
config = self._btc_config()
|
|
self._lugares_de_monederos(config)
|
|
datos = self.cartera.datos
|
|
precios = Precios(datos.vivos("precios"))
|
|
hoy = dt.date.today()
|
|
leer = bool((cuerpo or {}).get("leer", True))
|
|
try:
|
|
r = historial_btc.historial(config, datos, hoy, lambda a: precios.de(a),
|
|
self.cartera.carpeta, leer=leer)
|
|
except OSError as e:
|
|
raise ErrorOperacion(f"No he podido guardar lo leído del nodo: {e}") from None
|
|
return self._btc_presentar(config, r, hoy)
|
|
finally:
|
|
self._btc_lock.release()
|
|
|
|
def _btc_presentar(self, config: dict, r: dict, hoy: dt.date) -> dict:
|
|
"""Completa lo que devuelve el motor con lo que necesita la pantalla:
|
|
nombres, cuadre de cada lugar y actividad propuesta."""
|
|
datos = self.cartera.datos
|
|
movimientos = datos.vivos("movimientos")
|
|
activos = {a["id"]: a for a in datos.vivos("activos")}
|
|
r["cuadres"] = []
|
|
for k, q in sorted(r.pop("en_nodo").items()):
|
|
aid, lid = historial_btc.separar(k)
|
|
a = activos.get(aid) or {"id": aid, "nombre": aid}
|
|
apuntado = lugares.saldos(movimientos, aid, hoy.isoformat()).get(lid, 0.0) if lid \
|
|
else cantidad_en(movimientos, aid, hoy)
|
|
# De un monedero compartido, «en el nodo» es solo tu parte
|
|
partes = {btc.parte_de(m) for m in config.get("monederos") or []
|
|
if historial_btc.cuenta(m.get("activo"), m.get("lugar")) == k}
|
|
c = {"clave": k, "activo": aid, "lugar": lid,
|
|
"nombre_activo": a["nombre"] + (f" · {lugares.nombre(a, lid)}" if lid else ""),
|
|
"en_nodo": round(q, 9), "apuntado": round(apuntado, 9)}
|
|
if partes and min(partes) < 1:
|
|
c["parte"] = min(partes)
|
|
r["cuadres"].append(c)
|
|
r["activos"] = [{"id": a["id"], "nombre": a["nombre"]}
|
|
for a in datos.vivos("activos") if a["tipo"] == "Cripto"]
|
|
r["lugares"] = {a["id"]: lugares.de(a) for a in datos.vivos("activos") if a["tipo"] == "Cripto"}
|
|
r["sin_lugar"] = {aid: lugares.saldos(movimientos, aid, hoy.isoformat()).get("", 0.0)
|
|
for aid in {c["activo"] for c in r["cuadres"]}}
|
|
actividades_de = {m.get("id"): m.get("actividad") or "" for m in config.get("monederos") or []}
|
|
por_nombre = {m.get("nombre") or m.get("id"): m.get("id") for m in config.get("monederos") or []}
|
|
for p in r["partes"]:
|
|
a = activos.get(p["activo"]) or {"nombre": p["activo"]}
|
|
p["nombre_activo"] = a["nombre"] + (f" · {lugares.nombre(a, p['lugar'])}" if p.get("lugar") else "")
|
|
if p.get("destino_lugar"):
|
|
p["nombre_destino"] = lugares.nombre(activos.get(p["destino"]) or {}, p["destino_lugar"])
|
|
for c in p["candidatas"]:
|
|
if c["tipo"] == "Traslado":
|
|
c["desde"] = lugares.nombre(a, c.get("lugar")) if c.get("lugar") else "?"
|
|
c["hacia"] = lugares.nombre(a, c.get("lugar_destino")) if c.get("lugar_destino") else "?"
|
|
# Minería y entradas en monederos de una actividad: ya marcadas
|
|
suyas = {actividades_de.get(por_nombre.get(n)) or "" for n in p["monederos"]}
|
|
p["actividad"] = suyas.pop() if len(suyas) == 1 and p["clase"] in ("mineria", "entrada") else ""
|
|
return r
|
|
|
|
def btc_historial_aplicar(self, cuerpo: dict) -> dict:
|
|
decisiones = cuerpo.get("decisiones")
|
|
if not isinstance(decisiones, list):
|
|
raise ErrorOperacion("Petición no válida.")
|
|
with self._btc_lock:
|
|
config = self._btc_config()
|
|
r = historial_btc.aplicar(self.cartera, config, decisiones, dt.date.today())
|
|
self._btc_refrescar_ultima(config)
|
|
self._btc_guardar(config)
|
|
return r
|
|
|
|
def btc_borradores(self, cuerpo: dict) -> dict:
|
|
cambios = cuerpo.get("borradores")
|
|
if not isinstance(cambios, dict):
|
|
raise ErrorOperacion("Petición no válida.")
|
|
with self._btc_lock:
|
|
config = self._btc_config()
|
|
n = historial_btc.poner_borradores(config, cambios)
|
|
self._btc_guardar(config)
|
|
return {"borradores": n}
|
|
|
|
def btc_deshacer_partes(self, ids) -> dict:
|
|
if not isinstance(ids, list):
|
|
raise ErrorOperacion("Petición no válida.")
|
|
with self._btc_lock:
|
|
config = self._btc_config()
|
|
r = historial_btc.deshacer_partes(self.cartera, config, ids)
|
|
self._btc_refrescar_ultima(config)
|
|
self._btc_guardar(config)
|
|
return r
|
|
|
|
def btc_deshacer_parte(self, pid: str) -> dict:
|
|
with self._btc_lock:
|
|
config = self._btc_config()
|
|
r = historial_btc.deshacer_parte(self.cartera, config, str(pid or ""))
|
|
self._btc_refrescar_ultima(config)
|
|
self._btc_guardar(config)
|
|
return r
|
|
|
|
def _btc_guardado(self):
|
|
"""Lo último leído del nodo en este equipo: (txs, propios, nombres)."""
|
|
config = self._btc_config()
|
|
monederos = [m for m in config.get("monederos") or [] if btc.vigilable(m)]
|
|
cache = cache_btc.cargar(self.cartera.carpeta)
|
|
esc = cache_btc.escaneo_guardado(cache, monederos)
|
|
txs = {t: cache["txs"][t] for t in esc["alturas"] if t in cache["txs"]}
|
|
return config, txs, esc
|
|
|
|
def btc_monedas(self) -> dict:
|
|
"""Tus monedas (UTXO) de hoy y, de cada una, de dónde viene sobre todo."""
|
|
config, txs, esc = self._btc_guardado()
|
|
trazador = traza_btc.Trazador(txs, esc["propios"])
|
|
decididas = traza_btc.decisiones(config, self.cartera.datos.vivos("movimientos"))
|
|
descritos: dict[str, dict] = {}
|
|
salida = []
|
|
for m in traza_btc.monedas(txs, esc["propios"]):
|
|
origenes = sorted(trazador.origenes(m["txid"], m["indice"]).items(), key=lambda x: -x[1])
|
|
for clave, _ in origenes:
|
|
if clave not in descritos:
|
|
descritos[clave] = self._describir_origen(clave, txs, esc, decididas)
|
|
salida.append({**m, "cantidad": m["sats"] / btc.SATOSHI,
|
|
"monedero": esc["nombres"].get(m["monedero"], m["monedero"]),
|
|
"nombre_activo": self._nombre(m["activo"]),
|
|
"origenes": [{"clave": c, "fraccion": round(f, 6)} for c, f in origenes]})
|
|
return {"monedas": salida, "origenes": descritos, "leido": esc["fecha"]}
|
|
|
|
def btc_traza(self, cuerpo: dict) -> dict:
|
|
config, txs, esc = self._btc_guardado()
|
|
txid, indice = str(cuerpo.get("txid") or ""), int(cuerpo.get("indice") or 0)
|
|
tx = txs.get(txid)
|
|
s = next((x for x in (tx or {}).get("salidas", []) if x["indice"] == indice), None)
|
|
if s is None:
|
|
raise ErrorOperacion("No encuentro esa moneda en lo leído del nodo.")
|
|
trazador = traza_btc.Trazador(txs, esc["propios"])
|
|
decididas = traza_btc.decisiones(config, self.cartera.datos.vivos("movimientos"))
|
|
origenes = sorted(trazador.origenes(txid, indice).items(), key=lambda x: -x[1])
|
|
arbol = trazador.arbol(txid, indice, s["satoshis"])
|
|
|
|
def nombres(n):
|
|
n["monedero"] = esc["nombres"].get(n["monedero"], n["monedero"])
|
|
n["nombre_activo"] = self._nombre(n["activo"]) if n["activo"] else None
|
|
for hijo in n["hijos"]:
|
|
nombres(hijo)
|
|
nombres(arbol)
|
|
return {"arbol": arbol, "sats": s["satoshis"],
|
|
"origenes": [{**self._describir_origen(c, txs, esc, decididas), "fraccion": round(f, 6),
|
|
"sats": round(f * s["satoshis"])} for c, f in origenes]}
|
|
|
|
def _describir_origen(self, clave, txs, esc, decididas) -> dict:
|
|
d = traza_btc.describir(clave, txs, esc["propios"], decididas)
|
|
d["valor_dia"] = spot_btc.precio(d["fecha"])[0] if d.get("fecha") else None
|
|
if d.get("desde"):
|
|
d["desde"] = self._nombre(d["desde"])
|
|
m = (decididas.get(clave) or {}).get("mov")
|
|
if d.get("desde_lugar") and m:
|
|
# Vino de un exchange (u otro lugar): las compras de allí hasta ese día
|
|
a = next((x for x in self.cartera.datos.vivos("activos") if x["id"] == m["activo"]), {})
|
|
d["desde"] = lugares.nombre(a, d["desde_lugar"])
|
|
d["compras_alli"] = [{"fecha": x["fecha"], "cantidad": x.get("cantidad"), "importe": x.get("importe")}
|
|
for x in sorted(self.cartera.datos.vivos("movimientos"), key=lambda x: x["fecha"],
|
|
reverse=True)
|
|
if x["activo"] == m["activo"] and x.get("lugar") == d["desde_lugar"]
|
|
and x["tipo"] == "Compra" and x["fecha"] <= m["fecha"]][:5]
|
|
d["tipo"], d["importe"], d["cantidad"] = None, None, None
|
|
return d
|
|
|
|
def btc_importaciones(self) -> list[dict]:
|
|
config = self._btc_config()
|
|
return [{"id": i["id"], "fecha": i.get("fecha"), "creados": len(i.get("creados") or []),
|
|
"enlazados": len(i.get("enlazados") or []), "partes": len(i.get("partes") or [])}
|
|
for i in reversed(config.get("importaciones") or [])]
|
|
|
|
def btc_empezar_de_cero(self, cuerpo: dict) -> dict:
|
|
"""Quita lo del CSV de exchanges y del historial (ver deshacer_btc.py).
|
|
Sin «confirmar» solo dice lo que haría."""
|
|
with self._btc_lock:
|
|
config = self._btc_config()
|
|
if not cuerpo.get("confirmar"):
|
|
return deshacer_btc.plan(self.cartera, config)
|
|
ajustes = self._ajustes()
|
|
r = deshacer_btc.aplicar(self.cartera, config, ajustes)
|
|
self._btc_refrescar_ultima(config)
|
|
self._btc_guardar(config)
|
|
self._guardar_ajustes(ajustes)
|
|
return r
|
|
|
|
def btc_deshacer(self, imp_id: str) -> dict:
|
|
with self._btc_lock:
|
|
config = self._btc_config()
|
|
r = historial_btc.deshacer(self.cartera, config, str(imp_id or ""))
|
|
self._btc_refrescar_ultima(config)
|
|
self._btc_guardar(config)
|
|
return r
|
|
|
|
def _nombre(self, aid: str) -> str:
|
|
a = next((x for x in self.cartera.datos.vivos("activos") if x["id"] == aid), None)
|
|
return a["nombre"] if a else aid
|
|
|
|
def exportar_xlsx(self) -> bytes:
|
|
from .exportar_xlsx import exportar
|
|
with tempfile.TemporaryDirectory() as t:
|
|
destino = Path(t) / "cartera.xlsx"
|
|
exportar(self.cartera.datos, destino, dt.date.today())
|
|
return destino.read_bytes()
|
|
|
|
def rutas_post(self) -> dict:
|
|
c = self.cartera
|
|
return {
|
|
"/api/activo": lambda b: self._guardar_activo(b),
|
|
"/api/activo/cerrar": lambda b: c.cerrar_activo(b.get("id"), b.get("fecha"), b.get("importe")),
|
|
"/api/activo/evolucion": lambda b: evolucion_activo(c.datos, str(b.get("id") or ""), dt.date.today()),
|
|
"/api/activo/si_vendo": lambda b: self.si_vendo(str(b.get("id") or "")),
|
|
"/api/activo/reabrir": lambda b: c.reabrir_activo(b.get("id")),
|
|
"/api/activo/borrar": lambda b: c.borrar_activo(b.get("id")),
|
|
"/api/activo/comprobar": lambda b: dict(zip(("resultado", "mensaje"), fuentes.precio(
|
|
{"id": "prueba", "tipo": b.get("tipo"), "codigo": b.get("codigo"), "nombre": b.get("nombre")}))),
|
|
"/api/movimiento": lambda b: c.guardar_movimiento(b),
|
|
"/api/movimiento/borrar": lambda b: c.borrar_movimiento(b.get("id")),
|
|
"/api/movimiento/confirmar": lambda b: c.confirmar_movimiento(b.get("id")),
|
|
"/api/movimiento/adjuntar": lambda b: self.adjuntar(b),
|
|
"/api/movimiento/adjunto/quitar": lambda b: c.quitar_adjunto(b.get("id"), b.get("adjunto")),
|
|
"/api/movimientos/borrar": lambda b: c.borrar_movimientos(b.get("ids")),
|
|
"/api/movimientos/restaurar": lambda b: c.restaurar_movimientos(b.get("ids")),
|
|
"/api/movimientos/confirmar": lambda b: c.confirmar_movimientos(b.get("ids")),
|
|
"/api/ejemplo": lambda b: self.ejemplo(bool(b.get("entrar"))),
|
|
"/api/instalacion": lambda b: self.instalacion_mover(str(b.get("modo") or "")),
|
|
"/api/instalacion/lanzador": lambda b: self.instalacion_lanzador(bool(b.get("poner"))),
|
|
"/api/extracto/columnas": lambda b: self.extracto_columnas(b),
|
|
"/api/copia/crear": lambda b: self.copia_crear(b),
|
|
"/api/copia/restaurar": lambda b: self.copia_restaurar(b),
|
|
"/api/fiscal/criterios": lambda b: self.fiscal_criterios(b),
|
|
"/api/primer_uso": lambda b: self.primer_uso(b),
|
|
"/api/fiscal/declarado": lambda b: self.fiscal_declarado(b),
|
|
"/api/fiscal/trading": lambda b: self.fiscal_trading(b),
|
|
"/api/fiscal/amortizable": lambda b: c.marcar_amortizable(b.get("ids") or [], bool(b.get("si"))),
|
|
"/api/movimiento/importe": lambda b: c.poner_importe(b.get("id"), b.get("importe")),
|
|
"/api/movimientos/prima": lambda b: c.revisar_prima(b.get("ids"), bool(b.get("ok", True))),
|
|
"/api/movimientos/mover": lambda b: c.mover_movimientos(b.get("ids"), b.get("activo")),
|
|
"/api/traspaso": lambda b: c.traspaso(b),
|
|
"/api/lugar": lambda b: c.guardar_lugar(b.get("activo"), b),
|
|
"/api/lugar/borrar": lambda b: c.borrar_lugar(b.get("activo"), b.get("id")),
|
|
"/api/lugar/unir": lambda b: self.unir_lugares(b),
|
|
"/api/movimientos/lugar": lambda b: c.poner_lugar(b.get("ids"), b.get("lugar") or ""),
|
|
"/api/activo/convertir_en_lugar": lambda b: self.convertir_en_lugar(b),
|
|
"/api/deuda/cuadro": lambda b: c.estado_deuda(b.get("id")),
|
|
"/api/deuda/simular": lambda b: c.simular_amortizacion(b.get("id"), b.get("fecha"), b.get("importe")),
|
|
"/api/precio": lambda b: c.precio_manual(b.get("activo"), b.get("fecha"), b.get("valor"),
|
|
bool(b.get("valor_total"))),
|
|
"/api/precios/actualizar": lambda b: self.actualizar_precios(),
|
|
"/api/fuentes/comprobar": lambda b: self.comprobar_fuentes(),
|
|
"/api/interpretar": lambda b: self.interpretar(b),
|
|
"/api/alta_y_operacion": lambda b: c.alta_con_operacion(
|
|
b.get("activo") or {}, b.get("movimiento") or {},
|
|
{"precio": (b.get("activo") or {}).get("precio"), "fecha": (b.get("activo") or {}).get("fecha_precio"),
|
|
"fuente": (b.get("activo") or {}).get("fuente")}),
|
|
"/api/actividad/etiquetar": lambda b: c.etiquetar(b.get("ids") or [], b.get("actividad") or ""),
|
|
"/api/actividad/renombrar": lambda b: c.renombrar_actividad(b.get("viejo") or "", b.get("nuevo") or ""),
|
|
"/api/actividad/candidatas": lambda b: actividades.candidatas(
|
|
c.datos, b.get("actividad") or "", dt.date.today()),
|
|
"/api/recurrente": lambda b: c.guardar_recurrente(b),
|
|
"/api/recurrente/borrar": lambda b: c.borrar_recurrente(b.get("id")),
|
|
"/api/sesion": lambda b: self.guardar_sesion(float(b.get("valor") or 0), float(b.get("neto") or 0)),
|
|
"/api/sync/configurar": lambda b: self.sync.configurar(b.get("url"), b.get("huella")),
|
|
"/api/sync/ejecutar": lambda b: self.sync_ejecutar("sincronizar"),
|
|
"/api/sync/comprobar": lambda b: self.sync_ejecutar("comprobar"),
|
|
"/api/btc/estado": lambda b: self.btc_estado(),
|
|
"/api/btc/servidores": lambda b: self.btc_servidores(b),
|
|
"/api/btc/monedero": lambda b: self.btc_monedero(b),
|
|
"/api/btc/monedero/borrar": lambda b: self.btc_borrar_monedero(b.get("id")),
|
|
"/api/btc/probar": lambda b: self.btc_probar(b),
|
|
"/api/btc/detectar": lambda b: self.btc_detectar(b),
|
|
"/api/btc/comprobar": lambda b: self.btc_comprobar(),
|
|
"/api/btc/historial": lambda b: self.btc_historial(b),
|
|
"/api/btc/historial/borradores": lambda b: self.btc_borradores(b),
|
|
"/api/btc/historial/deshacer_parte": lambda b: self.btc_deshacer_parte(b.get("id")),
|
|
"/api/btc/historial/deshacer_partes": lambda b: self.btc_deshacer_partes(b.get("ids")),
|
|
"/api/btc/monedas": lambda b: self.btc_monedas(),
|
|
"/api/btc/traza": lambda b: self.btc_traza(b),
|
|
"/api/btc/historial/aplicar": lambda b: self.btc_historial_aplicar(b),
|
|
"/api/btc/importaciones": lambda b: self.btc_importaciones(),
|
|
"/api/btc/empezar_de_cero": lambda b: self.btc_empezar_de_cero(b),
|
|
"/api/btc/revisar_monederos": lambda b: self.btc_revisar_monederos(),
|
|
"/api/btc/importacion/deshacer": lambda b: self.btc_deshacer(b.get("id")),
|
|
"/api/extracto/analizar": lambda b: self.extracto_analizar(b),
|
|
"/api/extracto/aplicar": lambda b: self.extracto_aplicar(b),
|
|
# En un servidor «salir» es cerrar la sesión (ruta /api/acceso/salir): Caudal sigue en marcha
|
|
"/api/salir": lambda b: None if self.acceso is not None
|
|
else threading.Thread(target=self.servidor.shutdown, daemon=True).start(),
|
|
"/api/acceso/cambiar": lambda b: self.cambiar_clave(b),
|
|
"/api/cifrado": lambda b: self.cifrado_poner(b),
|
|
"/api/enlace/alta": lambda b: self.enlace_alta(b),
|
|
"/api/enlace/codigo": lambda b: self.enlace_codigo(),
|
|
"/api/enlace/direcciones": lambda b: self.enlace_direcciones(b),
|
|
"/api/qr": lambda b: self.dibujar_qr(b),
|
|
"/api/enlace/olvidar": lambda b: self.enlace_olvidar_equipos(b),
|
|
"/api/enlace/configurar": lambda b: self.enlace_configurar(b),
|
|
"/api/enlace/sincronizar": lambda b: self.enlace_sincronizar(b),
|
|
"/api/cifrado/recuperacion": lambda b: self.recuperacion_nueva(b),
|
|
}
|
|
|
|
|
|
def _recortar(clave: str | None) -> str:
|
|
"""La clave extendida abreviada, para enseñarla sin soltarla entera."""
|
|
clave = clave or ""
|
|
return clave if len(clave) <= 24 else f"{clave[:12]}…{clave[-6:]}"
|
|
|
|
|
|
def _electrum_del_aparato() -> dict | None:
|
|
"""El servidor Electrum que vive en el mismo aparato que Caudal: donde diga
|
|
quien empaqueta (CAUDAL_ELECTRUM=host:puerto, o varios separados por comas;
|
|
en Umbrel y StartOS, el electrs o Fulcrum de al lado) y, si no, el del propio
|
|
equipo en sus puertos de siempre. Solo se propone el primero que de verdad contesta."""
|
|
lista = [*btc.lista_de_direcciones(os.environ.get("CAUDAL_ELECTRUM", "")),
|
|
("127.0.0.1", 50001), ("127.0.0.1", 50002)]
|
|
for host, puerto in dict.fromkeys(lista):
|
|
candidato = {"host": host, "puerto": puerto, "tor": host.endswith(".onion"), "tls": False}
|
|
try:
|
|
btc.conectar([candidato], btc.proxy_de({}), espera=2)[0].cerrar()
|
|
except (ErrorElectrum, OSError):
|
|
continue
|
|
return candidato
|
|
return None
|
|
|
|
|
|
def crear_manejador(app: App):
|
|
class Manejador(BaseHTTPRequestHandler):
|
|
server_version = "Cartera"
|
|
sys_version = ""
|
|
|
|
def log_message(self, *args): # sin registro de peticiones en la terminal
|
|
pass
|
|
|
|
def _host_valido(self) -> bool:
|
|
if app.acceso is not None:
|
|
return True # en un servidor el nombre lo pone quien lo publica; protege la sesión
|
|
host = self.headers.get("Host", "")
|
|
return host in (f"127.0.0.1:{app.puerto}", f"localhost:{app.puerto}")
|
|
|
|
def _id_sesion(self) -> str:
|
|
for trozo in self.headers.get("Cookie", "").split(";"):
|
|
nombre, _, valor = trozo.strip().partition("=")
|
|
if nombre == COOKIE:
|
|
return valor
|
|
return ""
|
|
|
|
def _token_valido(self) -> bool:
|
|
dado = self.headers.get("X-Token", "")
|
|
if app.acceso is None:
|
|
return hmac.compare_digest(dado, app.token)
|
|
# La cookie dice quién eres; el token en la cabecera, que la petición sale de la propia aplicación
|
|
sesion = app.acceso.sesion(self._id_sesion())
|
|
return bool(sesion) and hmac.compare_digest(dado, sesion["token"])
|
|
|
|
def _cookie(self, sid: str, vida: int) -> dict:
|
|
seguro = "; Secure" if self.headers.get("X-Forwarded-Proto", "") == "https" else ""
|
|
return {"Set-Cookie": f"{COOKIE}={sid}; Path=/; HttpOnly; SameSite=Strict; Max-Age={vida}{seguro}"}
|
|
|
|
def _acceso(self, ruta: str, cuerpo: dict | None):
|
|
"""Las rutas de entrar y salir, que son las únicas que se atienden sin sesión."""
|
|
acceso = app.acceso
|
|
if ruta == "/api/acceso" and cuerpo is None:
|
|
sesion = acceso.sesion(self._id_sesion())
|
|
self._json(200, {"configurado": acceso.configurado(), "dentro": bool(sesion),
|
|
"cifrado": acceso.cifrado(), "recuperable": acceso.tiene_recuperacion(),
|
|
"token": sesion["token"] if sesion else None, "minimo": MINIMO_CLAVE,
|
|
"recomendado": list(RECOMENDADO_CLAVE)})
|
|
return
|
|
# Solo desde la propia aplicación: una cabecera que otra web no puede poner
|
|
if cuerpo is None or self.headers.get("X-Caudal") != "1":
|
|
self._json(403, {"error": "Prohibido"})
|
|
return
|
|
try:
|
|
extra = {}
|
|
if ruta == "/api/acceso/crear":
|
|
clave = str(cuerpo.get("clave") or "")
|
|
app.entrar(acceso.crear(clave))
|
|
if CIFRAR_DE_ENTRADA: # donde el disco no va cifrado (Umbrel), se cifra desde el principio
|
|
extra["recuperacion"] = app._cifrar(clave)["recuperacion"]
|
|
elif ruta == "/api/acceso/entrar":
|
|
app.entrar(acceso.comprobar(str(cuerpo.get("clave") or "")))
|
|
elif ruta == "/api/acceso/recuperar":
|
|
app.entrar(acceso.recuperar(str(cuerpo.get("recuperacion") or ""), str(cuerpo.get("nueva") or "")))
|
|
elif ruta == "/api/acceso/salir":
|
|
acceso.cerrar(self._id_sesion())
|
|
self._enviar(200, b'{"ok": true}', "application/json; charset=utf-8", self._cookie("", 0))
|
|
return
|
|
else:
|
|
self._json(404, {"error": "No encontrado"})
|
|
return
|
|
except Espera as e:
|
|
self._json(429, {"error": str(e)})
|
|
return
|
|
except (ErrorAcceso, ErrorCifrado, ErrorDatos, almacen.Bloqueado, CarteraOcupada) as e:
|
|
self._json(400, {"error": str(e)})
|
|
return
|
|
sid, token = acceso.abrir()
|
|
self._enviar(200, json.dumps({"ok": True, "token": token, **extra}).encode(), "application/json; charset=utf-8",
|
|
self._cookie(sid, DURACION_SESION))
|
|
|
|
def _enviar(self, codigo: int, cuerpo: bytes, tipo: str, extra: dict | None = None, guardable: bool = False):
|
|
extra = dict(extra or {})
|
|
# Texto grande, comprimido: la interfaz baja a la cuarta parte, que se nota por Tor o con poca cobertura
|
|
if len(cuerpo) >= COMPRIMIR_DESDE and tipo.split(";")[0] in COMPRIMIBLES \
|
|
and "gzip" in self.headers.get("Accept-Encoding", "") and "Content-Encoding" not in extra:
|
|
cuerpo = gzip.compress(cuerpo, 5)
|
|
extra["Content-Encoding"] = "gzip"
|
|
self.send_response(codigo)
|
|
self.send_header("Content-Type", tipo)
|
|
self.send_header("Content-Length", str(len(cuerpo)))
|
|
# Los datos nunca se guardan en el navegador. Los ficheros de la interfaz sí, pero preguntando
|
|
# siempre si han cambiado («no-cache»): tras actualizar Caudal se ve la versión nueva al momento
|
|
self.send_header("Cache-Control", "no-cache" if guardable else "no-store")
|
|
self.send_header("Vary", "Accept-Encoding")
|
|
self.send_header("X-Content-Type-Options", "nosniff")
|
|
self.send_header("Referrer-Policy", "no-referrer")
|
|
self.send_header("Content-Security-Policy",
|
|
"default-src 'self'; img-src 'self' data:; style-src 'self'; "
|
|
"script-src 'self'; connect-src 'self'; frame-ancestors 'none'")
|
|
for k, v in (extra or {}).items():
|
|
self.send_header(k, v)
|
|
self.end_headers()
|
|
self.wfile.write(cuerpo)
|
|
|
|
def _json(self, codigo: int, obj):
|
|
self._enviar(codigo, json.dumps(obj, ensure_ascii=False).encode(), "application/json; charset=utf-8")
|
|
|
|
def _rechazo(self):
|
|
if not self._host_valido():
|
|
self._enviar(403, b"Host no permitido", "text/plain")
|
|
return True
|
|
return False
|
|
|
|
def do_GET(self):
|
|
if self._rechazo():
|
|
return
|
|
url = urlparse(self.path)
|
|
if url.path == "/" and app.acceso is not None:
|
|
# La página no lleva nada privado: quien no tenga sesión solo verá la pantalla de entrada
|
|
html = (WEB / "index.html").read_text(encoding="utf-8").replace("__TOKEN__", "")
|
|
self._enviar(200, html.encode(), "text/html; charset=utf-8")
|
|
return
|
|
if url.path == "/":
|
|
if not hmac.compare_digest(parse_qs(url.query).get("t", [""])[0], app.token):
|
|
self._enviar(403, "Abre Caudal con la dirección completa que se muestra al arrancarlo.".encode(),
|
|
"text/plain; charset=utf-8")
|
|
return
|
|
html = (WEB / "index.html").read_text(encoding="utf-8").replace("__TOKEN__", app.token)
|
|
self._enviar(200, html.encode(), "text/html; charset=utf-8")
|
|
return
|
|
if url.path.startswith("/web/"):
|
|
ruta = (WEB / url.path[5:]).resolve()
|
|
if WEB not in ruta.parents or not ruta.is_file():
|
|
self._enviar(404, b"No encontrado", "text/plain")
|
|
return
|
|
tipo = mimetypes.guess_type(ruta.name)[0] or "application/octet-stream"
|
|
if tipo.startswith("text/") or tipo in ("application/javascript",):
|
|
tipo += "; charset=utf-8"
|
|
contenido = ruta.read_bytes()
|
|
marca = '"' + hashlib.sha256(contenido).hexdigest()[:20] + '"'
|
|
if self.headers.get("If-None-Match") == marca: # el navegador ya tiene esta versión
|
|
self.send_response(304)
|
|
self.send_header("ETag", marca)
|
|
self.send_header("Cache-Control", "no-cache")
|
|
self.end_headers()
|
|
return
|
|
self._enviar(200, contenido, tipo, {"ETag": marca}, guardable=True)
|
|
return
|
|
if app.acceso is not None and url.path == "/api/salud":
|
|
# Para que el sistema que aloja Caudal sepa que está en marcha; no cuenta nada privado
|
|
self._json(200, {"ok": True, "bloqueada": app.cartera is None})
|
|
return
|
|
if app.acceso is not None and url.path == "/api/acceso":
|
|
self._acceso(url.path, None)
|
|
return
|
|
if not url.path.startswith("/api/") or not self._token_valido():
|
|
self._sin_permiso()
|
|
return
|
|
try:
|
|
if url.path == "/api/estado":
|
|
self._json(200, app.estado())
|
|
elif url.path == "/api/precio":
|
|
from .calculo import Precios
|
|
q = parse_qs(url.query)
|
|
reg = Precios(app.cartera.datos.vivos("precios")).ultimo(
|
|
q.get("activo", [""])[0], dt.date.fromisoformat(q.get("fecha", [""])[0]))
|
|
self._json(200, {"precio": reg})
|
|
elif url.path == "/api/fiscal":
|
|
self._json(200, app.fiscal(parse_qs(url.query).get("ejercicio", [""])[0]))
|
|
elif url.path == "/api/sync/claves":
|
|
self._json(200, app.sync_claves())
|
|
elif url.path == "/api/buscar":
|
|
self._json(200, fuentes.buscar(parse_qs(url.query).get("q", [""])[0]))
|
|
elif url.path == "/api/adjunto":
|
|
q = parse_qs(url.query)
|
|
ref, contenido = app.cartera.leer_adjunto(q.get("m", [""])[0], q.get("a", [""])[0])
|
|
# Siempre como descarga y sin tipo propio: un adjunto nunca se
|
|
# interpreta dentro de la aplicación (un .html podría ejecutar código).
|
|
self._enviar(200, contenido, "application/octet-stream",
|
|
{"Content-Disposition": "attachment; filename*=UTF-8''" + quote(ref["nombre"])})
|
|
elif url.path == "/api/copia":
|
|
nombre, fichero = app.copia_descargar(parse_qs(url.query).get("id", [""])[0])
|
|
self._enviar(200, fichero, "application/octet-stream",
|
|
{"Content-Disposition": f'attachment; filename="{nombre}"'})
|
|
elif url.path == "/api/fiscal/regularizacion":
|
|
self._json(200, app.fiscal_regularizacion())
|
|
elif url.path == "/api/fiscal/exportar":
|
|
año, libro = app.fiscal_xlsx(parse_qs(url.query).get("ejercicio", [""])[0])
|
|
self._enviar(200, libro, "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
{"Content-Disposition": f'attachment; filename="fiscal_{año}.xlsx"'})
|
|
elif url.path == "/api/exportar":
|
|
nombre = f"cartera_{dt.date.today().isoformat()}.xlsx"
|
|
self._enviar(200, app.exportar_xlsx(),
|
|
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
{"Content-Disposition": f'attachment; filename="{nombre}"'})
|
|
else:
|
|
self._json(404, {"error": "No encontrado"})
|
|
except (ErrorOperacion, ErrorDatos, ErrorSync, ErrorBitcoin, ErrorElectrum, copia.ErrorCopia) as e:
|
|
self._json(400, {"error": str(e)})
|
|
except Exception:
|
|
traceback.print_exc()
|
|
self._json(500, {"error": "Error interno. Si se repite, el detalle está en el registro de la aplicación."})
|
|
|
|
def _sin_permiso(self):
|
|
if app.acceso is not None:
|
|
self._json(401, {"error": "Tu sesión ha terminado. Vuelve a entrar.", "sin_sesion": True})
|
|
else:
|
|
self._enviar(403, b"Prohibido", "text/plain")
|
|
|
|
def _cuerpo_json(self) -> dict:
|
|
largo = int(self.headers.get("Content-Length") or 0)
|
|
if largo > (MAX_COPIA if urlparse(self.path).path in ("/api/copia/restaurar", "/api/enlace") else MAX_CUERPO):
|
|
raise ErrorOperacion("Petición demasiado grande.")
|
|
if largo and "application/json" not in self.headers.get("Content-Type", ""):
|
|
raise ErrorOperacion("Formato de petición no válido.")
|
|
cuerpo = json.loads(self.rfile.read(largo) or b"{}") if largo else {}
|
|
if not isinstance(cuerpo, dict):
|
|
raise ErrorOperacion("Formato de petición no válido.")
|
|
return cuerpo
|
|
|
|
def do_POST(self):
|
|
if self._rechazo():
|
|
return
|
|
ruta = urlparse(self.path).path
|
|
if app.acceso is not None and ruta in ("/api/acceso/crear", "/api/acceso/entrar", "/api/acceso/salir",
|
|
"/api/acceso/recuperar"):
|
|
try:
|
|
self._acceso(ruta, self._cuerpo_json())
|
|
except (ErrorOperacion, json.JSONDecodeError, ValueError):
|
|
self._json(400, {"error": "Formato de petición no válido."})
|
|
return
|
|
if app.acceso is not None and ruta in ("/api/enlace", "/api/enlace/canje"):
|
|
# Otro Caudal que sincroniza con este: no trae sesión, trae su credencial (o el código para conseguirla)
|
|
try:
|
|
if ruta == "/api/enlace/canje":
|
|
if self.headers.get("X-Caudal") != "1": # como al entrar: que no lo pueda lanzar otra página
|
|
raise ValueError("falta la cabecera")
|
|
self._json(200, app.enlace_canje(self._cuerpo_json()))
|
|
else:
|
|
self._json(200, app.enlace_atender(self.headers.get("X-Enlace", ""), self._cuerpo_json()))
|
|
except Espera as e:
|
|
self._json(429, {"error": str(e)})
|
|
except ErrorAcceso as e:
|
|
self._json(403, {"error": str(e)})
|
|
except (enlace.ErrorEnlace, ErrorOperacion, ErrorDatos, almacen.Bloqueado, CarteraOcupada, ErrorCifrado) as e:
|
|
self._json(400, {"error": str(e)})
|
|
except (json.JSONDecodeError, ValueError):
|
|
self._json(400, {"error": "Formato de petición no válido."})
|
|
except Exception:
|
|
traceback.print_exc()
|
|
self._json(500, {"error": "Error interno. Si se repite, el detalle está en el registro de la aplicación."})
|
|
return
|
|
if not self._token_valido():
|
|
self._sin_permiso()
|
|
return
|
|
accion = app.rutas_post().get(ruta)
|
|
if accion is None:
|
|
self._json(404, {"error": "No encontrado"})
|
|
return
|
|
try:
|
|
cuerpo = self._cuerpo_json()
|
|
cuerpo["_sesion"] = self._id_sesion() # nunca viene del navegador: se pisa siempre
|
|
self._json(200, {"ok": True, "resultado": accion(cuerpo)})
|
|
except (ErrorOperacion, ErrorDatos, ErrorSync, ErrorBitcoin, ErrorElectrum, ErrorAcceso,
|
|
copia.ErrorCopia, adjuntos.ErrorAdjunto, ErrorCifrado, almacen.Bloqueado, enlace.ErrorEnlace) as e:
|
|
self._json(400, {"error": str(e)})
|
|
except json.JSONDecodeError:
|
|
self._json(400, {"error": "Formato de petición no válido."})
|
|
except Exception:
|
|
traceback.print_exc()
|
|
self._json(500, {"error": "Error interno. Si se repite, el detalle está en el registro de la aplicación."})
|
|
|
|
return Manejador
|
|
|
|
|
|
def arrancar(carpeta: Path, puerto: int = 0, escucha: str | None = None) -> App:
|
|
"""«escucha» es la dirección de red en la que atender cuando Caudal va en
|
|
un servidor (entonces se entra con contraseña). Sin ella, solo este equipo."""
|
|
acceso = Acceso(Path(carpeta).resolve().parent) if escucha else None
|
|
# Con los datos cifrados no se pueden abrir todavía: se abrirán cuando alguien entre con la contraseña
|
|
bloqueada = acceso is not None and acceso.cifrado()
|
|
app = App(None if bloqueada else Cartera(carpeta), acceso, Path(carpeta))
|
|
electrum.EN_SERVIDOR = acceso is not None # para explicar bien los fallos de conexión con el nodo
|
|
app.servidor = ThreadingHTTPServer((escucha or "127.0.0.1", puerto), crear_manejador(app))
|
|
app.servidor.daemon_threads = True
|
|
app.puerto = app.servidor.server_address[1]
|
|
return app
|
|
|
|
|
|
def main() -> int:
|
|
p = argparse.ArgumentParser(description="Cartera de inversiones en el navegador")
|
|
p.add_argument("--datos", default=None, help="carpeta de datos (por defecto, «datos» junto al programa)")
|
|
p.add_argument("--puerto", type=int, default=0, help="0 = uno libre al azar")
|
|
p.add_argument("--no-navegador", action="store_true")
|
|
p.add_argument("--servidor", nargs="?", const="0.0.0.0", default=os.environ.get("CAUDAL_SERVIDOR") or None,
|
|
metavar="DIRECCIÓN", help="atender en la red (por defecto 0.0.0.0) y pedir contraseña para entrar")
|
|
args = p.parse_args()
|
|
if args.datos is None and instalacion.empaquetado() and not args.servidor:
|
|
# El instalable: los datos, junto al programa si ahí hay una carpeta suya; si no, en la del usuario
|
|
base = instalacion.base_actual()[0]
|
|
args.datos = str(base / "datos")
|
|
try:
|
|
instalacion.copia_al_abrir(base)
|
|
except OSError as e:
|
|
print(f"⚠ No se pudo hacer la copia de seguridad al abrir: {e}")
|
|
elif args.datos is None:
|
|
args.datos = str(PROYECTO / "datos")
|
|
de_escritorio = instalacion.empaquetado() and not args.servidor
|
|
try:
|
|
app = arrancar(Path(args.datos), args.puerto, args.servidor)
|
|
except CarteraOcupada as e:
|
|
# El programa de escritorio no tiene ventana propia: si ya estaba abierto, se vuelve a enseñar
|
|
ya = instalacion.en_marcha(Path(args.datos).parent) if de_escritorio else None
|
|
if ya and not args.no_navegador:
|
|
webbrowser.open(ya)
|
|
return 0
|
|
print(f"✗ {e}")
|
|
return 1
|
|
except ErrorDatos as e:
|
|
print(f"✗ Los datos tienen un error y no se abren para no estropearlos:\n {e}")
|
|
return 1
|
|
url = f"http://127.0.0.1:{app.puerto}/?t={app.token}"
|
|
if args.servidor:
|
|
print(f"Caudal atiende en {args.servidor}:{app.puerto}. Se entra con contraseña"
|
|
+ ("." if app.acceso.configurado() else ": la crea quien entre primero."), flush=True)
|
|
else:
|
|
print(f"Caudal abierto: {url}")
|
|
print("Deja esta ventana abierta mientras lo uses. Para salir: botón «Salir» o Ctrl+C.")
|
|
if not args.no_navegador and not args.servidor:
|
|
threading.Timer(0.5, lambda: webbrowser.open(url)).start()
|
|
if de_escritorio:
|
|
instalacion.anotar_en_marcha(Path(args.datos).parent, url)
|
|
instalacion.repasar_lanzador(WEB / "icono.svg")
|
|
try:
|
|
app.servidor.serve_forever()
|
|
except KeyboardInterrupt:
|
|
pass
|
|
finally:
|
|
if de_escritorio:
|
|
instalacion.anotar_en_marcha(app.carpeta.parent, None) # donde estén ahora, por si se cambiaron de sitio
|
|
app.servidor.server_close()
|
|
if app.cartera is not None:
|
|
app.cartera.cerrar()
|
|
print("Caudal cerrado.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|