332 lines
12 KiB
Python
332 lines
12 KiB
Python
"""Export endpoints — PDF, Excel, CSV generation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import io
|
|
import json
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from fastapi.responses import PlainTextResponse, Response
|
|
from asyncpg import Connection
|
|
|
|
from app.dependencies import get_db
|
|
from app.models.export_ import ExportDraft
|
|
from app.models.common import ErrorResponse
|
|
from app.auth import require_auth
|
|
from app.services.calculators import bucefalo_precio, calcular_vigencia, sanitize_filename
|
|
from app.services.pdf_generator import generate_cotizacion_pdf
|
|
from app.services.excel_generator import generate_cotizacion_excel
|
|
|
|
router = APIRouter(prefix="/export", tags=["Export"])
|
|
|
|
|
|
async def _load_branding(db: Connection) -> dict[str, str]:
|
|
rows = await db.fetch('SELECT clave, valor FROM "Configuracion"')
|
|
return {r["clave"]: r["valor"] for r in rows}
|
|
|
|
|
|
def _build_pdf_data_from_draft(draft: ExportDraft, branding: dict[str, str]) -> dict[str, Any]:
|
|
fecha = datetime.now()
|
|
if draft.fecha:
|
|
try:
|
|
fecha = datetime.fromisoformat(draft.fecha.replace("Z", "+00:00"))
|
|
except ValueError:
|
|
try:
|
|
fecha = datetime.strptime(draft.fecha, "%Y-%m-%d")
|
|
except ValueError:
|
|
pass
|
|
|
|
vigencia = calcular_vigencia(fecha)
|
|
logo_raw = branding.get("logo_base64", "")
|
|
logo_mime = None
|
|
logo_b64 = None
|
|
if logo_raw and ":" in logo_raw:
|
|
parts = logo_raw.split(":", 1)
|
|
logo_mime = parts[0]
|
|
logo_b64 = parts[1]
|
|
elif logo_raw:
|
|
logo_b64 = logo_raw
|
|
|
|
return {
|
|
"numero": "BORRADOR",
|
|
"clienteNombre": draft.clienteNombre,
|
|
"clienteEmpresa": draft.clienteEmpresa,
|
|
"asesorNombre": draft.asesorNombre,
|
|
"fecha": fecha,
|
|
"vigencia": vigencia,
|
|
"moneda": draft.moneda,
|
|
"tipoCambio": draft.tipoCambio,
|
|
"proyecto": draft.proyecto,
|
|
"esquemaPago": draft.esquemaPago,
|
|
"servicios": [
|
|
{
|
|
"nombre": s.nombre,
|
|
"fase": s.fase,
|
|
"tipoPago": s.tipoPago,
|
|
"precio": s.precio,
|
|
"tiempoEntrega": s.tiempoEntrega,
|
|
"entregables": s.entregables,
|
|
}
|
|
for s in draft.servicios
|
|
],
|
|
"planBucefaloNivel": draft.planBucefaloNivel,
|
|
"planBucefaloPrecio": bucefalo_precio(draft.planBucefaloNivel) if draft.planBucefaloNivel else 0,
|
|
"incluirBonos": draft.incluirBonos,
|
|
"colorPrimario": branding.get("color_primario", "#2563eb"),
|
|
"colorSecundario": branding.get("color_secundario", "#1e293b"),
|
|
"logoBase64": logo_b64,
|
|
"logoMime": logo_mime,
|
|
"configBancaria": branding,
|
|
"razonSocial": branding.get("razon_social", "Cotizador E3"),
|
|
}
|
|
|
|
|
|
async def _build_pdf_data_from_db(db: Connection, cotizacion_id: str) -> dict[str, Any] | None:
|
|
cot = await db.fetchrow(
|
|
"""SELECT c.*, cl.nombre as cliente_nombre, cl.empresa as cliente_empresa,
|
|
u.name as asesor_nombre
|
|
FROM "Cotizacion" c
|
|
LEFT JOIN "Cliente" cl ON cl.id = c."clienteId"
|
|
LEFT JOIN "User" u ON u.id = c."asesorId"
|
|
WHERE c.id = $1""",
|
|
cotizacion_id,
|
|
)
|
|
if not cot:
|
|
return None
|
|
|
|
servicios = await db.fetch(
|
|
"""SELECT sc.fase, sc."tipoPago", sc.precio, sc."tiempoEntrega", sc.entregables,
|
|
s.nombre
|
|
FROM "ServicioCotizado" sc
|
|
JOIN "ServicioCatalogo" s ON s.id = sc."servicioCatalogoId"
|
|
WHERE sc."cotizacionId" = $1 AND sc.seleccionado = true
|
|
ORDER BY sc.fase, sc.id""",
|
|
cotizacion_id,
|
|
)
|
|
|
|
plan = await db.fetchrow(
|
|
'SELECT nivel, precio FROM "PlanBucefaloCotizacion" WHERE "cotizacionId" = $1',
|
|
cotizacion_id,
|
|
)
|
|
|
|
branding = await _load_branding(db)
|
|
|
|
logo_raw = branding.get("logo_base64", "")
|
|
logo_mime = None
|
|
logo_b64 = None
|
|
if logo_raw and ":" in logo_raw:
|
|
parts = logo_raw.split(":", 1)
|
|
logo_mime = parts[0]
|
|
logo_b64 = parts[1]
|
|
elif logo_raw:
|
|
logo_b64 = logo_raw
|
|
|
|
servicios_list = []
|
|
for s in servicios:
|
|
ent = s["entregables"]
|
|
if isinstance(ent, str):
|
|
try:
|
|
ent = json.loads(ent)
|
|
except (json.JSONDecodeError, TypeError):
|
|
ent = []
|
|
servicios_list.append({
|
|
"nombre": s["nombre"],
|
|
"fase": s["fase"],
|
|
"tipoPago": s["tipoPago"],
|
|
"precio": float(s["precio"]),
|
|
"tiempoEntrega": s["tiempoEntrega"],
|
|
"entregables": ent or [],
|
|
})
|
|
|
|
plan_nivel = plan["nivel"] if plan else None
|
|
|
|
return {
|
|
"numero": cot["numero"],
|
|
"clienteNombre": cot["cliente_nombre"] or "",
|
|
"clienteEmpresa": cot["cliente_empresa"] or "",
|
|
"asesorNombre": cot["asesor_nombre"] or "",
|
|
"fecha": cot["fecha"],
|
|
"vigencia": cot["vigencia"],
|
|
"moneda": cot["moneda"],
|
|
"tipoCambio": cot["tipoCambio"],
|
|
"proyecto": cot["proyecto"],
|
|
"esquemaPago": cot["esquemaPago"],
|
|
"servicios": servicios_list,
|
|
"planBucefaloNivel": plan_nivel,
|
|
"planBucefaloPrecio": float(plan["precio"]) if plan else 0,
|
|
"incluirBonos": cot["incluirBonos"],
|
|
"colorPrimario": branding.get("color_primario", "#2563eb"),
|
|
"colorSecundario": branding.get("color_secundario", "#1e293b"),
|
|
"logoBase64": logo_b64,
|
|
"logoMime": logo_mime,
|
|
"configBancaria": branding,
|
|
"razonSocial": branding.get("razon_social", "Cotizador E3"),
|
|
}
|
|
|
|
|
|
# ── PDF ENDPOINTS ──────────────────────────────────
|
|
|
|
@router.post("/pdf")
|
|
async def export_pdf_draft(
|
|
body: ExportDraft,
|
|
_auth: dict = Depends(require_auth),
|
|
db: Connection = Depends(get_db),
|
|
):
|
|
branding = await _load_branding(db)
|
|
pdf_data = _build_pdf_data_from_draft(body, branding)
|
|
pdf_bytes = generate_cotizacion_pdf(pdf_data)
|
|
|
|
empresa = body.clienteEmpresa or body.clienteNombre or "Cotizacion"
|
|
filename = f"{sanitize_filename(empresa)} - BORRADOR.pdf"
|
|
|
|
return Response(
|
|
content=pdf_bytes,
|
|
media_type="application/pdf",
|
|
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
|
|
)
|
|
|
|
|
|
@router.get("/pdf/{cotizacion_id}")
|
|
async def export_pdf_saved(
|
|
cotizacion_id: str,
|
|
_auth: dict = Depends(require_auth),
|
|
db: Connection = Depends(get_db),
|
|
):
|
|
pdf_data = await _build_pdf_data_from_db(db, cotizacion_id)
|
|
if not pdf_data:
|
|
raise HTTPException(status_code=404, detail="Cotización no encontrada")
|
|
|
|
pdf_bytes = generate_cotizacion_pdf(pdf_data)
|
|
|
|
empresa = pdf_data["clienteEmpresa"] or pdf_data["clienteNombre"]
|
|
filename = f"{sanitize_filename(empresa)} - {sanitize_filename(pdf_data['numero'])}.pdf"
|
|
|
|
return Response(
|
|
content=pdf_bytes,
|
|
media_type="application/pdf",
|
|
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
|
|
)
|
|
|
|
|
|
# ── EXCEL ENDPOINTS ────────────────────────────────
|
|
|
|
@router.post("/excel")
|
|
async def export_excel_draft(
|
|
body: ExportDraft,
|
|
_auth: dict = Depends(require_auth),
|
|
db: Connection = Depends(get_db),
|
|
):
|
|
branding = await _load_branding(db)
|
|
excel_data = _build_pdf_data_from_draft(body, branding)
|
|
excel_bytes = generate_cotizacion_excel(excel_data, saved=False)
|
|
|
|
empresa = body.clienteEmpresa or body.clienteNombre or "Cotizacion"
|
|
filename = f"{sanitize_filename(empresa)} - {sanitize_filename(body.clienteNombre)} - BORRADOR.xlsx"
|
|
|
|
return Response(
|
|
content=excel_bytes,
|
|
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
|
|
)
|
|
|
|
|
|
@router.get("/excel/{cotizacion_id}")
|
|
async def export_excel_saved(
|
|
cotizacion_id: str,
|
|
_auth: dict = Depends(require_auth),
|
|
db: Connection = Depends(get_db),
|
|
):
|
|
excel_data = await _build_pdf_data_from_db(db, cotizacion_id)
|
|
if not excel_data:
|
|
raise HTTPException(status_code=404, detail="Cotización no encontrada")
|
|
|
|
excel_bytes = generate_cotizacion_excel(excel_data, saved=True)
|
|
|
|
empresa = excel_data["clienteEmpresa"] or excel_data["clienteNombre"]
|
|
filename = f"{sanitize_filename(empresa)} - {sanitize_filename(excel_data['numero'])}.xlsx"
|
|
|
|
return Response(
|
|
content=excel_bytes,
|
|
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
|
|
)
|
|
|
|
|
|
# ── CSV ENDPOINTS ──────────────────────────────────
|
|
|
|
@router.get("/catalogo", response_class=PlainTextResponse)
|
|
async def export_catalogo_csv(
|
|
_auth: dict = Depends(require_auth),
|
|
db: Connection = Depends(get_db),
|
|
):
|
|
rows = await db.fetch(
|
|
"""SELECT s.id, s.nombre, s.descripcion, s.fase, s."tipoPago",
|
|
s."precioBase", s."tiempoEntrega", s."entregablesDefault",
|
|
s.variante, s.orden, c.nombre AS categoria
|
|
FROM "ServicioCatalogo" s
|
|
LEFT JOIN "Categoria" c ON c.id = s."categoriaId"
|
|
WHERE s.activo = true
|
|
ORDER BY s.fase, s.orden"""
|
|
)
|
|
|
|
buf = io.StringIO()
|
|
writer = csv.writer(buf)
|
|
writer.writerow([
|
|
"Nombre", "Categoria", "Fase", "Tipo de Pago", "Precio Base",
|
|
"Tiempo de Entrega", "Entregables", "Variante",
|
|
])
|
|
for r in rows:
|
|
entregables = r["entregablesDefault"] or []
|
|
if isinstance(entregables, list):
|
|
entregables = " | ".join(entregables)
|
|
elif isinstance(entregables, str):
|
|
try:
|
|
ent_list = json.loads(entregables)
|
|
entregables = " | ".join(ent_list) if isinstance(ent_list, list) else entregables
|
|
except (json.JSONDecodeError, TypeError):
|
|
pass
|
|
writer.writerow([
|
|
r["nombre"], r["categoria"], r["fase"], r["tipoPago"],
|
|
r["precioBase"], r["tiempoEntrega"], entregables, r["variante"],
|
|
])
|
|
|
|
return PlainTextResponse(
|
|
content=buf.getvalue(),
|
|
media_type="text/csv; charset=utf-8",
|
|
headers={"Content-Disposition": "attachment; filename=catalogo-servicios.csv"},
|
|
)
|
|
|
|
|
|
@router.get("/catalogo/plantilla", response_class=PlainTextResponse)
|
|
async def export_catalogo_template(db: Connection = Depends(get_db)):
|
|
cats = await db.fetch(
|
|
'SELECT nombre FROM "Categoria" WHERE activo = true ORDER BY orden'
|
|
)
|
|
cat_names = [r["nombre"] for r in cats]
|
|
|
|
buf = io.StringIO()
|
|
writer = csv.writer(buf)
|
|
writer.writerow([
|
|
"nombre", "descripcion", "fase", "tipoPago",
|
|
"precioBase", "tiempoEntrega", "entregablesDefault", "variante", "categoria",
|
|
])
|
|
writer.writerow([
|
|
"SEO On-Page", "Optimización on-page para buscadores", "Contenido y SEO", "mensual",
|
|
"2900", "7 - 14 dias", "Keyword research | On-page optimization", "", "SEO",
|
|
])
|
|
writer.writerow([
|
|
"", "", "", "", "", "", "", "", "",
|
|
])
|
|
writer.writerow([f"CATEGORIAS: {', '.join(cat_names)}"])
|
|
writer.writerow(["FASES: Auditoria, Setup e Infraestructura, Publicidad y Manejo, Contenido y SEO"])
|
|
writer.writerow(["TIPOS DE PAGO: unico, mensual"])
|
|
|
|
return PlainTextResponse(
|
|
content=buf.getvalue(),
|
|
media_type="text/csv; charset=utf-8",
|
|
headers={"Content-Disposition": "attachment; filename=plantilla-catalogo.csv"},
|
|
)
|