126 lines
4.3 KiB
Python
126 lines
4.3 KiB
Python
"""
|
|
adjuntos.py — Justificantes de las operaciones (extracto del banco, factura,
|
|
contrato...), guardados junto a los datos.
|
|
|
|
Cada fichero se guarda en `datos/adjuntos/<sha256>`, con su huella como nombre:
|
|
• el mismo fichero adjuntado dos veces ocupa una sola vez;
|
|
• un fichero nunca cambia, así que juntar copias es solo añadir los que falten;
|
|
• al leerlo se comprueba que no se ha estropeado.
|
|
|
|
La operación guarda la referencia: {"id": huella, "nombre", "tipo", "bytes"}.
|
|
Quitar un adjunto de una operación NO borra el fichero (puede estar en otra, y
|
|
así una fusión o un deshacer nunca se queda sin él).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import os
|
|
import re
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
from . import almacen
|
|
|
|
CARPETA = "adjuntos"
|
|
MAX_BYTES = 5 * 1024 * 1024
|
|
MAX_POR_OPERACION = 10
|
|
HUELLA = re.compile(r"^[0-9a-f]{64}$")
|
|
|
|
# Solo formatos de documento e imagen corrientes. Se descargan siempre como
|
|
# fichero, nunca se abren dentro de la aplicación.
|
|
TIPOS = {
|
|
".pdf": "application/pdf", ".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg",
|
|
".webp": "image/webp", ".heic": "image/heic", ".txt": "text/plain", ".csv": "text/csv",
|
|
".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
".xls": "application/vnd.ms-excel", ".ods": "application/vnd.oasis.opendocument.spreadsheet",
|
|
".eml": "message/rfc822",
|
|
}
|
|
|
|
|
|
class ErrorAdjunto(ValueError):
|
|
"""Mensaje pensado para el usuario."""
|
|
|
|
|
|
def carpeta_de(datos: Path) -> Path:
|
|
return Path(datos) / CARPETA
|
|
|
|
|
|
def ruta(datos: Path, huella: str) -> Path:
|
|
if not HUELLA.match(huella or ""):
|
|
raise ErrorAdjunto("Adjunto no válido.")
|
|
return carpeta_de(datos) / huella
|
|
|
|
|
|
def limpiar_nombre(nombre: str) -> str:
|
|
"""Solo el nombre (sin carpetas) y sin caracteres de control."""
|
|
nombre = re.sub(r"[\x00-\x1f\x7f/\\]", "_", Path(str(nombre or "")).name).strip()
|
|
return nombre[-120:] or "adjunto"
|
|
|
|
|
|
def tipo_de(nombre: str) -> str:
|
|
ext = Path(nombre).suffix.lower()
|
|
if ext not in TIPOS:
|
|
raise ErrorAdjunto(f"Ese tipo de fichero no se admite ({ext or 'sin extensión'}). "
|
|
f"Vale: {', '.join(sorted(TIPOS))}.")
|
|
return TIPOS[ext]
|
|
|
|
|
|
def guardar(datos: Path, contenido: bytes) -> str:
|
|
"""Guarda el contenido (si no estaba ya) y devuelve su huella."""
|
|
if not contenido:
|
|
raise ErrorAdjunto("El fichero está vacío.")
|
|
if len(contenido) > MAX_BYTES:
|
|
raise ErrorAdjunto(f"El fichero pasa de {MAX_BYTES // (1024 * 1024)} MB.")
|
|
huella = hashlib.sha256(contenido).hexdigest()
|
|
destino = ruta(datos, huella)
|
|
if destino.exists() and _sano(destino, huella):
|
|
return huella
|
|
destino.parent.mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
os.chmod(destino.parent, 0o700)
|
|
except OSError:
|
|
pass # exFAT (el USB): sin permisos
|
|
fd, tmp = tempfile.mkstemp(dir=destino.parent, prefix=".", suffix=".tmp")
|
|
try:
|
|
with os.fdopen(fd, "wb") as f:
|
|
f.write(almacen.sellar(contenido))
|
|
f.flush()
|
|
os.fsync(f.fileno())
|
|
try:
|
|
os.chmod(tmp, 0o600)
|
|
except OSError:
|
|
pass
|
|
os.replace(tmp, destino)
|
|
except BaseException:
|
|
Path(tmp).unlink(missing_ok=True)
|
|
raise
|
|
return huella
|
|
|
|
|
|
def _sano(camino: Path, huella: str) -> bool:
|
|
try:
|
|
return hashlib.sha256(almacen.leer(camino)).hexdigest() == huella
|
|
except OSError:
|
|
return False
|
|
|
|
|
|
def leer(datos: Path, huella: str) -> bytes:
|
|
camino = ruta(datos, huella)
|
|
try:
|
|
contenido = almacen.leer(camino)
|
|
except FileNotFoundError:
|
|
raise ErrorAdjunto("Ese adjunto no está en este equipo (¿falta sincronizar?).") from None
|
|
if hashlib.sha256(contenido).hexdigest() != huella:
|
|
raise ErrorAdjunto("El fichero adjunto está dañado: recupéralo de la copia.")
|
|
return contenido
|
|
|
|
|
|
def tiene(datos: Path, huella: str) -> bool:
|
|
return ruta(datos, huella).exists()
|
|
|
|
|
|
def referenciados(movimientos: list[dict]) -> set[str]:
|
|
"""Huellas de los adjuntos de las operaciones no borradas."""
|
|
return {a["id"] for m in movimientos if not m.get("borrado")
|
|
for a in m.get("adjuntos") or [] if isinstance(a, dict) and HUELLA.match(str(a.get("id")))}
|