# ──────────────────────────────────────────────────────────────────────────────
# apps/wp_invoices/api/blueprint.py
# MVP mínimo: formulario de carga → extracción con OpenAI → revisión → guardado
# ──────────────────────────────────────────────────────────────────────────────
from __future__ import annotations
import base64
from pdf2image import convert_from_bytes
import io
import os, uuid, mimetypes, shutil
from datetime import datetime
import json
import logging
from dataclasses import dataclass
from typing import Dict, Any, Tuple
import re
from flask import Blueprint, request, render_template_string, redirect, url_for, flash, session, render_template, current_app
from werkzeug.utils import secure_filename
from decimal import Decimal, InvalidOperation
from sqlalchemy.orm import Session
from apps.wp_invoices.db.base import Base
from apps.wp_invoices.models import Invoice, InvoiceItem
from config.db import get_session # Asumimos que ya existe un helper que retorna Session()
# ── Blueprint ────────────────────────────────────────────────────────────────
wp_invoices_bp = Blueprint("wp_invoices", __name__, url_prefix="/wp_invoices")
log = logging.getLogger("wp_invoices")
ALLOWED_EXTS = {"pdf", "png", "jpg", "jpeg"}
# ── Prompt base (mejorado levemente) ─────────────────────────────────────────
PROMPT_BASE = (
"Eres un verificador de facturas. Reconstruye la lógica contable, valida "
"que las cifras cuadren y NO inventes datos. Si un dato no aparece impreso, "
"deja su valor.verbatim = null y explica el porqué en notes.\n\n"
"Formato y restricciones ESTRICTAS:\n"
"- DEVUELVE ÚNICAMENTE UN JSON VÁLIDO Y MINIFICADO (una sola línea, sin markdown, sin comentarios).\n"
"- NUNCA agregues texto fuera del JSON.\n"
"- NO uses saltos de línea dentro de strings; reemplaza \\n por espacios en valores de texto.\n"
"- En arrays 'evidence' usa como máximo 3 tokens cortos (<=40 caracteres) y evita frases largas.\n"
"- Evita comillas tipográficas; usa solo comillas ASCII.\n\n"
"Distinciones: verbatim (impreso), computed (cálculo), evidence (IDs/bboxes), confidence (0..1).\n"
"Detección tipo: 'Corporate' vs 'POS' según reglas ya especificadas.\n"
"Reglas de cálculo y GST como se definieron.\n"
"POS 2 líneas: fusiona adecuadamente header_line/pricing_line y valida qty*unit≈line_total (0.01).\n"
"Marca inconsistencias aritméticas vs extraction_error vs human_error.\n"
"Normaliza decimales con punto.\n\n"
"SALIDA EXACTA en este esquema (sin comentarios):"
"{'invoice_type':{'verbatim':'POS|Corporate','confidence':0},"
"'header':{'supplier':{'name':{'verbatim':null,'computed':null,'confidence':0,'evidence':[],'notes':''},"
"'abn':{'verbatim':null,'computed':null,'confidence':0,'evidence':[],'notes':''}},"
"'invoice':{'number':{'verbatim':null,'computed':null,'confidence':0,'evidence':[]},"
"'date':{'verbatim':null,'computed':null,'confidence':0,'evidence':[]},"
"'time':{'verbatim':null,'computed':null,'confidence':0,'evidence':[]},"
"'type':{'verbatim':null,'computed':null,'confidence':0,'evidence':[]}},"
"'payment':{'method':{'verbatim':null,'computed':null,'confidence':0,'evidence':[]},"
"'tendered_total':{'verbatim':null,'computed':null,'confidence':0,'evidence':[]}}},"
"'items':[{'sku':{'verbatim':null,'computed':null,'confidence':0,'evidence':[],'notes':''},"
"'description':{'verbatim':null,'computed':null,'confidence':0,'evidence':[]},"
"'qty':{'verbatim':null,'computed':null,'confidence':0,'evidence':[]},"
"'unit_price':{'verbatim':null,'computed':null,'confidence':0,'evidence':[]},"
"'line_total':{'verbatim':null,'computed':null,'confidence':0,'evidence':[]},"
"'checks':{'qty_x_unit_eq_total':false,'tolerance':0.01}}],"
"'extras':[{'type':'surcharge|tip|donation|shipping|discount_global|rounding|fee',"
"'label':{'verbatim':null,'computed':null,'confidence':0,'evidence':[]},"
"'amount':{'verbatim':null,'computed':null,'confidence':0,'evidence':[]},"
"'tax_type':'taxable|gst_free|unknown'}],"
"'totals':{'items_sum':{'verbatim':null,'computed':null,'confidence':0,'evidence':[]},"
"'gst':{'verbatim':null,'computed':null,'confidence':0,'evidence':[],'notes':''},"
"'grand_total':{'verbatim':null,'computed':null,'confidence':0,'evidence':[]},"
"'checks':{'items_sum_matches_subtotal_or_total':false,"
"'subtotal_plus_gst_plus_extras_equals_grand_total':false,"
"'tendered_equals_grand_total':false}},"
"'status':'consistent|extraction_error|human_error','notes':[]}"
)
# ── Modelo recomendado de OpenAI (calidad/precio) ───────────────────────────
# "gpt-4.1-mini" (texto+visión), muy buen costo/beneficio para invoices.
OPENAI_MODEL = "gpt-4.1-mini"
# ── Pequeño wrapper para invocar OpenAI sin acoplar fuerte ──────────────────
@dataclass
class ExtractResult:
ok: bool
data: Dict[str, Any] | None
error: str | None = None
def _b64(data: bytes) -> str:
return base64.b64encode(data).decode("utf-8")
# Activa Files API para PDF (y opcionalmente para imágenes)
USE_FILES_API_FOR_PDF = True
FILES_PURPOSE = "vision" # ← válido para vision/Responses en tu versión
import re
def _detect_gst_inclusive(raw: dict) -> bool:
"""Heurística simple: true si el JSON extraído sugiere 'includes GST'."""
try:
# 1) Revisar notes del bloque totals.gst
gst_block = (raw or {}).get("totals", {}).get("gst", {}) or {}
notes = (gst_block.get("notes") or "") if isinstance(gst_block.get("notes"), str) else ""
evid = gst_block.get("evidence") or []
hay_notes = "includes gst" in notes.lower() or "incl gst" in notes.lower() or "gst included" in notes.lower()
hay_evid = any(("includes gst" in str(x).lower() or "incl gst" in str(x).lower() or "gst included" in str(x).lower()) for x in evid)
# 2) También probar en raw["notes"] si existe
raw_notes = raw.get("notes") or []
if isinstance(raw_notes, list):
hay_raw = any(("includes gst" in str(x).lower() or "incl gst" in str(x).lower() or "gst included" in str(x).lower()) for x in raw_notes)
else:
hay_raw = "includes gst" in str(raw_notes).lower()
return bool(hay_notes or hay_evid or hay_raw)
except Exception:
return False
def _build_viewmodel_from_form(form):
"""Arma un viewmodel con la misma estructura que usa el template
(header/items/totals con subcampos .verbatim) a partir de los valores POST."""
header_vm = {
'supplier': {
'name': {'verbatim': form.get('supplier_name') or ''},
'abn': {'verbatim': form.get('supplier_abn') or ''},
},
'invoice': {
'number': {'verbatim': form.get('invoice_number') or ''},
'date': {'verbatim': form.get('invoice_date') or ''},
'time': {'verbatim': ''},
'type': {'verbatim': ''},
},
'payment': {
'method': {'verbatim': form.get('payment_method') or ''},
'tendered_total': {'verbatim': form.get('tendered_total') or ''},
},
}
# Descubrir índices de items en el POST
idxs = set()
for k in form.keys():
m = re.match(r"items-(\d+)-", k)
if m:
idxs.add(int(m.group(1)))
items_vm = []
for i in sorted(idxs):
items_vm.append({
'sku': {'verbatim': form.get(f'items-{i}-sku') or ''},
'description': {'verbatim': form.get(f'items-{i}-description') or ''},
'qty': {'verbatim': form.get(f'items-{i}-qty') or ''},
'unit_price': {'verbatim': form.get(f'items-{i}-unit_price') or ''},
'line_total': {'verbatim': form.get(f'items-{i}-line_total') or ''},
})
totals_vm = {
'items_sum': {'verbatim': form.get('items_sum') or ''},
'gst': {'verbatim': form.get('gst') or ''},
'grand_total': {'verbatim': form.get('grand_total') or ''},
}
return header_vm, items_vm, totals_vm
def coerce_json(raw: str) -> dict:
"""
Convierte 'raw' en JSON estricto:
- quita ```json ... ``` si existe
- recorta texto extraño
- extrae el bloque { ... } hasta el último cierre balanceado (ignorando llaves dentro de strings)
- intenta parsear; si hay comas colgantes, las limpia.
"""
s = raw.strip()
# 1) quitar fences de markdown
if s.startswith("```"):
s = re.sub(r"^```(?:json)?\s*|\s*```$", "", s.strip(), flags=re.IGNORECASE)
# 2) si ya parsea, devolver
try:
return json.loads(s)
except json.JSONDecodeError:
pass
# 3) buscar el primer '{'
start = s.find("{")
if start == -1:
raise json.JSONDecodeError("No JSON object detected", s, 0)
# 4) escanear y cortar cuando el nivel de llaves vuelva a 0 (fuera de comillas)
in_str = False
esc = False
level = 0
cut = None
for i, ch in enumerate(s[start:], start=start):
if in_str:
if esc:
esc = False
elif ch == "\\":
esc = True
elif ch == '"':
in_str = False
else:
if ch == '"':
in_str = True
elif ch == "{":
level += 1
elif ch == "}":
level -= 1
if level == 0:
cut = i + 1
break
if cut is None:
# no encontró cierre balanceado, intenta al menos hasta el último '}' de la cadena
end = s.rfind("}")
if end != -1 and end > start:
cut = end + 1
else:
raise json.JSONDecodeError("Unbalanced JSON braces", s, start)
snippet = s[start:cut]
# 5) primer intento de parseo del bloque balanceado
try:
return json.loads(snippet)
except json.JSONDecodeError:
# 6) quitar comas colgantes antes de ] o }
snippet2 = re.sub(r",(\s*[}\]])", r"\1", snippet)
return json.loads(snippet2)
def _pdf_first_page_to_png(pdf_bytes: bytes) -> bytes:
pages = convert_from_bytes(pdf_bytes, dpi=200, fmt="png") # primera página por defecto
buf = io.BytesIO()
pages[0].save(buf, format="PNG")
return buf.getvalue()
# Ajustes globales
USE_FILES_API_FOR_PDF = True
FILES_PURPOSE = "vision" # ← ¡IMPORTANTE! no uses "responses"
def extract_with_openai(file_bytes: bytes, filename: str, mimetype: str) -> ExtractResult:
"""
- Imágenes (png/jpg/jpeg): input_image + image_url (string con data URL base64)
- PDFs: Files API (purpose="vision") y luego input_file con file_id
- Sin response_format: coercion robusto a JSON con coerce_json()
"""
try:
from openai import OpenAI
client = OpenAI() # Requiere OPENAI_API_KEY
# === Construir el mensaje ===
if mimetype == "application/pdf":
try:
file_bytes = _pdf_first_page_to_png(file_bytes)
mimetype = "image/png"
except Exception as e:
return ExtractResult(ok=False, data=None,
error=f"No pude convertir PDF a imagen: {e}")
# --- Imágenes (o PDF convertido) → image_url data:... ---
b64 = _b64(file_bytes)
user_message = {
"role": "user",
"content": [
{"type": "input_text", "text": PROMPT_BASE},
{"type": "input_image", "image_url": f"data:{mimetype};base64,{b64}"},
],
}
else:
# Imágenes → data URL base64 (estable)
b64 = _b64(file_bytes)
user_message = {
"role": "user",
"content": [
{"type": "input_text", "text": PROMPT_BASE},
{"type": "input_image", "image_url": f"data:{mimetype};base64,{b64}"},
],
}
# === Llamada al modelo (sin response_format) ===
resp = client.responses.create(
model=OPENAI_MODEL, # "gpt-4.1-mini" recomendado
input=[user_message],
max_output_tokens=8192,
temperature=0.0,
)
# Preferimos output_text si existe
raw = getattr(resp, "output_text", None)
if not raw:
raw = json.dumps(resp.dict())
print("DEBUG raw(800):", raw[:800])
data = coerce_json(raw) # ← robusto contra texto extra / comas
return ExtractResult(ok=True, data=data)
except Exception as e:
log.exception("OpenAI extraction failed")
try:
print("DEBUG last raw(800):", raw[:800])
except Exception:
pass
return ExtractResult(ok=False, data=None, error=str(e))
# ── Templates embebidos (para evitar muchos archivos) ───────────────────────
TPL_UPLOAD = """
WP Invoices – Subir factura
Subir factura (pdf/png/jpg/jpeg)
{% with msgs = get_flashed_messages(with_categories=true) %}
{% if msgs %}
{% endif %}
"""
# ── Helpers de template ─────────────────────────────────────────────────────
from markupsafe import Markup, escape as _escape
def _safe(val: Any) -> str:
if val is None:
return ""
return str(val)
wp_invoices_bp.add_app_template_global(lambda v: Markup(_escape(_safe(v))), name="safe")
wp_invoices_bp.add_app_template_global(lambda seq: enumerate(seq), name="enumerate")
#helpers de revalidacion aritmetica
#----------------------------------
def _approx_equal(a: Decimal | None, b: Decimal | None, tol: Decimal = Decimal('0.01')) -> bool:
if a is None or b is None:
return False
try:
return abs(a - b) <= tol
except Exception:
return False
def _val_get(it, field):
"""Obtiene un campo desde dict o desde un objeto (modelo) por getattr."""
if isinstance(it, dict):
return it.get(field)
return getattr(it, field, None)
def _to_dec_or_none(v):
if v is None:
return None
if isinstance(v, Decimal):
return v
try:
return Decimal(str(v))
except Exception:
return None
def _validate_items_and_totals(items, items_sum, gst, grand_total, tendered_total, tol: Decimal = Decimal("0.01"), gst_mode: str = "auto"): # 'auto' | 'inclusive' | 'exclusive':
"""
Acepta items como lista de dicts Ó de modelos con attrs: qty, unit_price, line_total.
Retorna (ok: bool, errors: dict, computed: dict).
"""
errors = {"items": {}, "totals": {}}
computed = {"items_sum_calc": Decimal("0")}
for i, it in enumerate(items):
qty = _to_dec_or_none(_val_get(it, "qty"))
unit = _to_dec_or_none(_val_get(it, "unit_price"))
line = _to_dec_or_none(_val_get(it, "line_total"))
item_err = {}
missing = []
if qty is None: missing.append("qty")
if unit is None: missing.append("unit_price")
expected = None
if qty is not None and unit is not None:
try:
expected = (qty * unit).quantize(Decimal("0.01"))
except Exception:
expected = None
if line is not None:
computed["items_sum_calc"] += line
elif expected is not None:
computed["items_sum_calc"] += expected
if expected is not None and line is not None:
try:
if abs(expected - line) > tol:
item_err["qty_unit_total_mismatch"] = str(expected - line)
except Exception:
item_err["qty_unit_total_mismatch"] = "mismatch"
if missing:
item_err["missing_fields"] = missing
if item_err:
errors["items"][i] = item_err
mode = gst_mode # por defecto lo que venga
if mode == "auto":
# auto = por defecto 'exclusive'; si más arriba detectas 'includes GST', pásalo desde save() (ver paso 3)
mode = "exclusive"
# Totales
if items_sum is not None:
items_sum = _to_dec_or_none(items_sum)
print("item_sum")
print(items_sum)
print("Citem_sum")
print(computed["items_sum_calc"])
print("diffs")
print(abs(items_sum - computed["items_sum_calc"]))
print("tol")
print(tol)
if items_sum is None or abs(items_sum - computed["items_sum_calc"]) > tol:
try:
diff = (items_sum - computed["items_sum_calc"]) if items_sum is not None else "mismatch"
except Exception:
diff = "mismatch"
errors["totals"]["items_sum_mismatch"] = str(diff)
if mode == "exclusive":
if gst is not None and grand_total is not None:
gst = _to_dec_or_none(gst)
grand_total = _to_dec_or_none(grand_total)
if gst is not None and grand_total is not None:
try:
if abs((computed["items_sum_calc"] + gst) - grand_total) > tol:
errors["totals"]["sum_plus_gst_mismatch"] = str((computed["items_sum_calc"] + gst) - grand_total)
except Exception:
errors["totals"]["sum_plus_gst_mismatch"] = "mismatch"
elif mode == "inclusive":
# Total ya incluye GST → esperamos items_sum_calc ≈ grand_total (sin sumar GST otra vez)
if grand_total is not None:
grand_total = _to_dec_or_none(grand_total)
if grand_total is not None:
try:
if abs(computed["items_sum_calc"] - grand_total) > tol:
errors["totals"]["items_sum_vs_total_mismatch"] = str(computed["items_sum_calc"] - grand_total)
except Exception:
errors["totals"]["items_sum_vs_total_mismatch"] = "mismatch"
# Verificación de coherencia del GST impreso: total - total/1.1
if grand_total is not None and gst is not None:
try:
implied = (grand_total - (grand_total / Decimal("1.1"))).quantize(Decimal("0.01"))
print("implied")
print(implied)
if abs(implied - gst) > tol:
errors["totals"]["gst_inclusive_mismatch"] = str(implied - gst)
except Exception:
errors["totals"]["gst_inclusive_mismatch"] = "mismatch"
if tendered_total is not None and grand_total is not None:
tendered_total = _to_dec_or_none(tendered_total)
grand_total = _to_dec_or_none(grand_total)
if tendered_total is not None and grand_total is not None:
try:
if abs(tendered_total - grand_total) > tol:
errors["totals"]["tendered_vs_grand_mismatch"] = str(tendered_total - grand_total)
except Exception:
errors["totals"]["tendered_vs_grand_mismatch"] = "mismatch"
ok = (not errors["items"]) and (not errors["totals"])
return ok, errors, computed
#helpers de guardado de archivos
def _ensure_dir(path: str):
os.makedirs(path, exist_ok=True)
def _slugify_name(name: str) -> str:
s = name.strip().replace(' ', '_')
# elimina caracteres problemáticos
return ''.join(ch for ch in s if ch.isalnum() or ch in ('_', '-', '.'))[:120]
def _ext_from_mime_or_name(mimetype: str | None, filename: str | None) -> str:
ext = None
if filename and '.' in filename:
ext = filename.rsplit('.', 1)[-1].lower()
if not ext and mimetype:
ext = (mimetypes.guess_extension(mimetype) or '').lstrip('.')
return ('.' + ext) if ext else ''
# ── Rutas ───────────────────────────────────────────────────────────────────
@wp_invoices_bp.get("/")
def upload_form():
return render_template_string(TPL_UPLOAD)
@wp_invoices_bp.post("/process")
def process():
f = request.files.get("file")
if not f:
flash("Selecciona un archivo de factura", "error")
return redirect(url_for("wp_invoices.upload_form"))
data = f.read()
filename = secure_filename(f.filename or "invoice")
mimetype = f.mimetype or {
"pdf": "application/pdf",
"png": "image/png",
"jpg": "image/jpeg",
"jpeg": "image/jpeg",
}.get(ext, "application/octet-stream")
staging_dir = current_app.config.get('INVOICE_STAGING_DIR')
_ensure_dir(staging_dir)
tmpname = uuid.uuid4().hex + _ext_from_mime_or_name(mimetype, filename)
tmppath = os.path.join(staging_dir, tmpname)
with open(tmppath, 'wb') as out:
out.write(data)
# guarda metadatos en sesión para que /save lo recoja
session['wpinv_staging_file'] = {
'path': tmppath,
'orig_name': filename,
'mimetype': mimetype,
'uploaded_at': datetime.utcnow().isoformat() + 'Z',
}
ext = (filename.rsplit(".", 1)[-1].lower() if "." in filename else "").strip()
if ext not in ALLOWED_EXTS:
flash("Formato no permitido", "error")
return redirect(url_for("wp_invoices.upload_form"))
# 1) Llamar a OpenAI
extracted = extract_with_openai(data, filename, mimetype)
# Log/print para trazabilidad
if extracted.ok:
log.info("Extraction OK: %s", json.dumps(extracted.data, ensure_ascii=False)[:4000])
print("WP_INVOICES_EXTRACT:", json.dumps(extracted.data, ensure_ascii=False))
else:
log.error("Extraction ERROR: %s", extracted.error)
flash(f"Error en extracción: {extracted.error}", "error")
return redirect(url_for("wp_invoices.upload_form"))
# 2) Guardar JSON en sesión y redirigir a review
session["wpinv_extract"] = extracted.data
session['wp_inv_raw_min'] = json.dumps(extracted.data, separators=(",", ":"), ensure_ascii=False)
session['wp_inv_raw_pretty'] = json.dumps(extracted.data, indent=2, ensure_ascii=False)
return redirect(url_for("wp_invoices.review"))
@wp_invoices_bp.get("/review")
def review():
data = session.get("wpinv_extract")
if not data:
flash("No hay datos para revisar. Sube una factura.", "error")
return redirect(url_for("wp_invoices.upload_form"))
header = data.get("header", {})
items = data.get("items", [])
totals = data.get("totals", {})
raw_pretty = session.get('wp_inv_raw_pretty')
#return render_template_string(
# TPL_REVIEW,
# header=header,
# items=items,
# totals=totals,
# raw_json=data,
#)
return render_template('wp_invoices/review.html', header=header, items=items, totals=totals, raw_json=raw_pretty)
@wp_invoices_bp.post("/save")
def save():
# 1) Parseo seguro de formulario (con defaults)
def g(name, default=""):
return (request.form.get(name) or default).strip()
supplier_name = g("supplier_name")
supplier_abn = g("supplier_abn")
invoice_number = g("invoice_number")
invoice_date = g("invoice_date")
payment_method = g("payment_method")
tendered_total = g("tendered_total") or None
items_sum = g("items_sum") or None
gst = g("gst") or None
grand_total = g("grand_total") or None
# 2) Items dinámicos (según lo renderizado)
# Encontrar índices por prefijo items--field
idxs = set()
for k in request.form.keys():
if k.startswith("items-"):
try:
idx = int(k.split("-", 2)[1])
idxs.add(idx)
except Exception:
pass
items: list[InvoiceItem] = []
for i in sorted(idxs):
it = InvoiceItem(
sku=g(f"items-{i}-sku"),
description=g(f"items-{i}-description"),
qty=_to_decimal(g(f"items-{i}-qty")),
unit_price=_to_decimal(g(f"items-{i}-unit_price")),
line_total=_to_decimal(g(f"items-{i}-line_total")),
)
items.append(it)
# 3) Guardar en DB
sess: Session = get_session()
inv = Invoice(
supplier_name=supplier_name or None,
supplier_abn=supplier_abn or None,
invoice_number=invoice_number or None,
invoice_date=invoice_date or None,
payment_method=payment_method or None,
tendered_total=_to_decimal(tendered_total),
items_sum=_to_decimal(items_sum),
gst=_to_decimal(gst),
grand_total=_to_decimal(grand_total),
)
for it in items:
inv.items.append(it)
raw = session.get("wpinv_extract") or {}
gst_inclusive = _detect_gst_inclusive(raw)
#hacemos validaciones
ok, errors, computed = _validate_items_and_totals(
items=items,
items_sum=items_sum,
gst=gst,
grand_total=grand_total,
tendered_total=tendered_total,
tol=Decimal('0.01'),
gst_mode=("inclusive" if gst_inclusive else "exclusive"),
)
if not ok:
# Muestra errores y retorna a la review sin guardar
flash("Hay inconsistencias aritméticas. Revisa los campos resaltados.", "error")
data = session.get("wpinv_extract") or {}
header_vm, items_vm, totals_vm = _build_viewmodel_from_form(request.form)
raw_pretty = session.get('wp_inv_raw_pretty') or json.dumps(data, indent=2, ensure_ascii=False)
header = data.get("header", {})
items_data = data.get("items", [])
totals_data = data.get("totals", {})
return render_template(
'wp_invoices/review.html',
header=header_vm,
items=items_vm,
totals=totals_vm,
raw_json=raw_pretty,
safe=_safe,
errors=errors,
), 400
# return render_template(
# 'wp_invoices/review.html',
# header=header,
# items=items_data,
# totals=totals_data,
# raw_json=raw_pretty,
# safe=_safe,
# errors=errors,
# ), 400
# Si pasa validación, opcionalmente puedes fijar items_sum al calculado si venía vacío
if items_sum is None:
items_sum = computed['items_sum_calc']
sess.add(inv)
sess.commit()
#guardado del archivo
staging = session.pop('wpinv_staging_file', None)
if staging and os.path.isfile(staging.get('path', '')):
store_dir = current_app.config.get('INVOICE_STORE_DIR')
_ensure_dir(store_dir)
# crea subcarpeta por ID para evitar colisiones y organizar
inv_dir = os.path.join(store_dir, f"{inv.id:08d}")
_ensure_dir(inv_dir)
# nombre destino: _original o fallback al original slugificado
ext = _ext_from_mime_or_name(staging.get('mimetype'), staging.get('orig_name'))
base = _slugify_name(inv.invoice_number or staging.get('orig_name') or f'invoice_{inv.id}')
dest_path = os.path.join(inv_dir, f"{base}_original{ext}")
inv.file_path = dest_path####
# evita sobreescritura si ya existe (agrega sufijo incremental)
i = 1
final_path = dest_path
while os.path.exists(final_path):
final_path = os.path.join(inv_dir, f"{base}_original_{i}{ext}")
i += 1
shutil.move(staging['path'], final_path)
# Opcional: guarda un .json con metadatos junto al archivo
meta_path = os.path.join(inv_dir, f"{base}_original.json")
try:
with open(meta_path, 'w', encoding='utf-8') as mf:
json.dump({
'invoice_id': inv.id,
'original_name': staging.get('orig_name'),
'mimetype': staging.get('mimetype'),
'stored_path': final_path,
'uploaded_at': staging.get('uploaded_at'),
}, mf, ensure_ascii=False, indent=2)
except Exception:
pass
# INFO al log y flash de confirmación
sess.add(inv)
sess.commit()
logging.info("Archivo de factura %s almacenado en %s", staging.get('orig_name'), final_path)
flash("Archivo original guardado en disco.", "success")
else:
logging.warning("No staging file found in session for invoice %s", inv.id)
flash(f"Factura guardada (id={inv.id}).", "success")
return redirect(url_for("wp_invoices.upload_form"))
# ── Utils numéricos ─────────────────────────────────────────────────────────
def _to_decimal(s: str | None) -> Decimal | None:
if not s:
return None
try:
# normaliza comas → puntos por si el modelo entrega con coma
s2 = s.replace(",", ".")
return Decimal(s2)
except (InvalidOperation, AttributeError):
return None