781 lines
35 KiB
Python
781 lines
35 KiB
Python
import logging
|
|
from typing import Dict, Any, Literal, Optional
|
|
|
|
from core.config import settings
|
|
from core.database import get_core_db
|
|
from core.exceptions import BaseAPIException
|
|
from core.security import (
|
|
_has_local_super_admin_role,
|
|
get_active_system,
|
|
get_current_user,
|
|
is_hub_admin,
|
|
validate_access_to_resource,
|
|
)
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, Path, Request
|
|
from sqlalchemy import func, or_, and_
|
|
from sqlalchemy.orm import Session
|
|
|
|
from . import schemas, services, models
|
|
from .catalog_service import InvoiceCatalogService
|
|
from .permission_map import (
|
|
build_allowed_types_from_view_permissions,
|
|
get_invoice_permission_base,
|
|
)
|
|
|
|
# Create main router
|
|
router = APIRouter()
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# --- RUTAS DE UTILIDAD ---
|
|
|
|
@router.get("/invoices/creation-data", response_model=schemas.InvoiceCreationResponse)
|
|
def get_creation_data(
|
|
company_id: int = Query(..., description="Company ID"),
|
|
operation_type: schemas.OperationType = Query(..., description="Operation type for the new invoice"),
|
|
invoice_type: Optional[str] = Query(
|
|
None,
|
|
description="Invoice type key (required for import; optional for export — uses invoice.exp.create)",
|
|
),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
"""Get consolidated data for creating a new invoice"""
|
|
try:
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
ot = operation_type.value if hasattr(operation_type, "value") else operation_type
|
|
if ot == "imp":
|
|
if not invoice_type:
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail="invoice_type is required for import invoices",
|
|
)
|
|
perm_base = get_invoice_permission_base(operation_type, invoice_type)
|
|
validate_access_to_resource(
|
|
db,
|
|
company_id,
|
|
current_user,
|
|
required_permissions=[f"{perm_base}.create"],
|
|
)
|
|
else:
|
|
if invoice_type:
|
|
perm_base = get_invoice_permission_base(operation_type, invoice_type)
|
|
validate_access_to_resource(
|
|
db,
|
|
company_id,
|
|
current_user,
|
|
required_permissions=[f"{perm_base}.create"],
|
|
)
|
|
else:
|
|
validate_access_to_resource(
|
|
db,
|
|
company_id,
|
|
current_user,
|
|
required_permissions=["invoice.exp.create"],
|
|
)
|
|
return InvoiceCatalogService.get_creation_data(db, tenant_id, company_id)
|
|
except HTTPException:
|
|
raise
|
|
except BaseAPIException:
|
|
raise
|
|
except Exception as e:
|
|
logger.exception("get_creation_data failed: %s", e)
|
|
raise HTTPException(status_code=500, detail=f"Error al cargar datos de creación: {str(e)}")
|
|
|
|
@router.get("/invoices/{invoice_id}/edition-data", response_model=schemas.InvoiceEditionResponse)
|
|
def get_edition_data(
|
|
invoice_id: int,
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
"""Get consolidated data for editing an existing invoice"""
|
|
try:
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
|
|
# 1. Traer la factura para saber su tipo
|
|
invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
|
|
if not invoice:
|
|
raise HTTPException(status_code=404, detail="Invoice not found")
|
|
|
|
# 2. Validar permiso dinámico de Edición
|
|
perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
|
|
validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.edit"])
|
|
|
|
data = InvoiceCatalogService.get_edition_data(db, invoice_id, tenant_id, company_id)
|
|
return data
|
|
except HTTPException:
|
|
raise
|
|
except BaseAPIException:
|
|
raise
|
|
except Exception as e:
|
|
logger.exception("get_edition_data failed: %s", e)
|
|
raise HTTPException(status_code=500, detail=f"Error al cargar datos de edición: {str(e)}")
|
|
|
|
@router.get("/invoices/remesa-suggestion", response_model=Dict[str, int])
|
|
def get_remesa_suggestion(
|
|
pedimento_id: int = Query(..., description="Pedimento ID"),
|
|
company_id: int = Query(..., description="Company ID"),
|
|
operation_type: Optional[schemas.OperationType] = Query(
|
|
None,
|
|
description="Required when no invoice exists yet for this pedimento (context for permission)",
|
|
),
|
|
invoice_type: Optional[str] = Query(
|
|
None,
|
|
description="Required when no invoice exists yet for this pedimento (context for permission)",
|
|
),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Sugiere siguiente remesa. Si ya hay facturas ligadas al pedimento, el permiso se infiere de ellas.
|
|
Si no hay facturas en compliance para ese pedimento, pasar operation_type e invoice_type del alta en curso.
|
|
"""
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
|
|
invoice_for_perm = (
|
|
db.query(models.InvoiceHeader)
|
|
.join(
|
|
models.InvoiceComplianceMx,
|
|
models.InvoiceComplianceMx.invoice_id == models.InvoiceHeader.id,
|
|
)
|
|
.filter(
|
|
models.InvoiceComplianceMx.pedimento_id == pedimento_id,
|
|
models.InvoiceHeader.tenant_id == tenant_id,
|
|
models.InvoiceHeader.company_id == company_id,
|
|
)
|
|
.order_by(models.InvoiceHeader.id.desc())
|
|
.first()
|
|
)
|
|
if invoice_for_perm:
|
|
perm_base = get_invoice_permission_base(
|
|
invoice_for_perm.operation_type, invoice_for_perm.invoice_type
|
|
)
|
|
validate_access_to_resource(
|
|
db,
|
|
company_id,
|
|
current_user,
|
|
required_permissions=[f"{perm_base}.edit"],
|
|
)
|
|
elif operation_type is not None and invoice_type:
|
|
perm_base = get_invoice_permission_base(operation_type, invoice_type)
|
|
validate_access_to_resource(
|
|
db,
|
|
company_id,
|
|
current_user,
|
|
required_permissions=[f"{perm_base}.edit"],
|
|
)
|
|
else:
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail="operation_type and invoice_type are required when no invoice is linked to this pedimento yet",
|
|
)
|
|
|
|
max_rem = (
|
|
db.query(func.max(models.InvoiceComplianceMx.remesa))
|
|
.filter(
|
|
models.InvoiceComplianceMx.pedimento_id == pedimento_id,
|
|
models.InvoiceComplianceMx.tenant_id == tenant_id,
|
|
models.InvoiceComplianceMx.company_id == company_id,
|
|
)
|
|
.scalar()
|
|
)
|
|
return {"next_remesa": int((max_rem or 0) + 1)}
|
|
|
|
|
|
# --- RUTAS CRUD MANUALES (Sustituyen al TenantCRUDRoutes por seguridad) ---
|
|
|
|
@router.post("/invoices/", response_model=schemas.InvoiceHeaderResponse)
|
|
def create_invoice(
|
|
request: Request,
|
|
data: schemas.InvoiceHeaderCreate,
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
try:
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
|
|
# Sistema autoritativo desde el contexto activo (header/cookie)
|
|
active_system = get_active_system(request)
|
|
if active_system:
|
|
data = data.model_copy(update={"system": active_system})
|
|
|
|
# Validamos usando los datos que vienen en el body (payload)
|
|
perm_base = get_invoice_permission_base(data.operation_type, data.invoice_type)
|
|
validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.create"])
|
|
|
|
return services.InvoiceService.create(db, data, tenant_id, company_id)
|
|
except HTTPException:
|
|
raise
|
|
except BaseAPIException:
|
|
raise
|
|
except Exception as e:
|
|
logger.exception("create_invoice failed: %s", e)
|
|
raise HTTPException(status_code=500, detail=f"Error al guardar factura: {str(e)}")
|
|
|
|
@router.get("/invoices/{invoice_id}", response_model=schemas.InvoiceHeaderResponse)
|
|
def get_invoice(
|
|
invoice_id: int,
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
|
|
if not invoice:
|
|
raise HTTPException(status_code=404, detail="Invoice not found")
|
|
|
|
perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
|
|
validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.view"])
|
|
|
|
return invoice
|
|
|
|
@router.put("/invoices/{invoice_id}", response_model=schemas.InvoiceHeaderResponse)
|
|
def update_invoice(
|
|
invoice_id: int,
|
|
data: schemas.InvoiceHeaderUpdate,
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
|
|
if not invoice:
|
|
raise HTTPException(status_code=404, detail="Invoice not found")
|
|
|
|
perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
|
|
validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.edit"])
|
|
|
|
return services.InvoiceService.update(db, invoice_id, tenant_id, data, company_id)
|
|
|
|
@router.delete("/invoices/{invoice_id}")
|
|
def delete_invoice(
|
|
invoice_id: int,
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
|
|
if not invoice:
|
|
raise HTTPException(status_code=404, detail="Invoice not found")
|
|
|
|
perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
|
|
validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.delete"])
|
|
|
|
success = services.InvoiceService.delete(db, invoice_id, tenant_id, company_id)
|
|
return {"success": success}
|
|
|
|
@router.post("/invoices/{invoice_id}/copy", response_model=schemas.InvoiceHeaderResponse)
|
|
def copy_invoice(
|
|
invoice_id: int = Path(..., description="ID de la factura a copiar"),
|
|
company_id: int = Query(..., description="Company ID"),
|
|
new_invoice_number: Optional[str] = Query(None, description="Número de factura para la copia"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
original = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
|
|
if not original:
|
|
raise HTTPException(status_code=404, detail="Factura no encontrada")
|
|
|
|
perm_base = get_invoice_permission_base(original.operation_type, original.invoice_type)
|
|
validate_access_to_resource(
|
|
db, company_id, current_user, required_permissions=[f"{perm_base}.create"]
|
|
)
|
|
|
|
try:
|
|
return services.InvoiceService.copy(db, invoice_id, tenant_id, company_id, new_invoice_number)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.exception("copy_invoice failed: %s", e)
|
|
raise HTTPException(status_code=500, detail=f"Error al copiar factura: {str(e)}")
|
|
|
|
|
|
@router.post("/invoices/{invoice_id}/copy-header", response_model=schemas.InvoiceHeaderResponse)
|
|
def copy_invoice_header(
|
|
invoice_id: int = Path(..., description="ID de la factura"),
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
original = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
|
|
if not original:
|
|
raise HTTPException(status_code=404, detail="Factura no encontrada")
|
|
perm_base = get_invoice_permission_base(original.operation_type, original.invoice_type)
|
|
validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.create"])
|
|
try:
|
|
return services.InvoiceService.copy_header_only(db, invoice_id, tenant_id, company_id)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.exception("copy_invoice_header failed: %s", e)
|
|
raise HTTPException(status_code=500, detail=f"Error al copiar encabezado: {str(e)}")
|
|
|
|
|
|
@router.get("/invoices/{invoice_id}/export")
|
|
def export_invoice(
|
|
invoice_id: int = Path(..., description="ID de la factura"),
|
|
format: Literal['csv', 'xlsx', 'txt'] = Query(..., description="Formato: csv | xlsx | txt"),
|
|
company_id: int = Query(...),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
|
|
if not invoice:
|
|
raise HTTPException(status_code=404, detail="Factura no encontrada")
|
|
perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
|
|
validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.view"])
|
|
try:
|
|
return services.InvoiceService.export_items(db, invoice, format)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.exception("export_invoice failed: %s", e)
|
|
raise HTTPException(status_code=500, detail=f"Error al exportar factura: {str(e)}")
|
|
|
|
|
|
@router.get("/invoices/{invoice_id}/interfaces/gm-transport")
|
|
def interface_gm_transport(
|
|
invoice_id: int = Path(..., description="ID de la factura"),
|
|
company_id: int = Query(...),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
|
|
if not invoice:
|
|
raise HTTPException(status_code=404, detail="Factura no encontrada")
|
|
perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
|
|
validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.view"])
|
|
try:
|
|
return services.InvoiceService.export_gm_transport(db, invoice)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.exception("interface_gm_transport failed: %s", e)
|
|
raise HTTPException(status_code=500, detail=f"Error al generar interfaz GM Transport: {str(e)}")
|
|
|
|
|
|
@router.get("/invoices/{invoice_id}/interfaces/carta-porte")
|
|
def interface_carta_porte(
|
|
invoice_id: int = Path(..., description="ID de la factura"),
|
|
company_id: int = Query(...),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
|
|
if not invoice:
|
|
raise HTTPException(status_code=404, detail="Factura no encontrada")
|
|
perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
|
|
validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.view"])
|
|
try:
|
|
return services.InvoiceService.export_carta_porte(db, invoice)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.exception("interface_carta_porte failed: %s", e)
|
|
raise HTTPException(status_code=500, detail=f"Error al generar interfaz Carta Porte: {str(e)}")
|
|
|
|
|
|
@router.get("/invoices/{invoice_id}/interfaces/tfc")
|
|
def interface_tfc(
|
|
invoice_id: int = Path(..., description="ID de la factura"),
|
|
company_id: int = Query(...),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
|
|
if not invoice:
|
|
raise HTTPException(status_code=404, detail="Factura no encontrada")
|
|
perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
|
|
validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.view"])
|
|
try:
|
|
return services.InvoiceService.export_tfc(db, invoice)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.exception("interface_tfc failed: %s", e)
|
|
raise HTTPException(status_code=500, detail=f"Error al generar interfaz TFC: {str(e)}")
|
|
|
|
|
|
@router.get("/invoices/{invoice_id}/interfaces/carta-porte-consolidada")
|
|
def interface_carta_porte_consolidada(
|
|
invoice_id: int = Path(..., description="ID de la factura"),
|
|
company_id: int = Query(...),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
|
|
if not invoice:
|
|
raise HTTPException(status_code=404, detail="Factura no encontrada")
|
|
perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
|
|
validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.view"])
|
|
try:
|
|
return services.InvoiceService.export_carta_porte_consolidada(db, invoice)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.exception("interface_carta_porte_consolidada failed: %s", e)
|
|
raise HTTPException(status_code=500, detail=f"Error al generar interfaz Carta Porte Consolidada: {str(e)}")
|
|
|
|
|
|
@router.get("/invoices/{invoice_id}/interfaces/aviso-cruce")
|
|
def interface_aviso_cruce(
|
|
invoice_id: int = Path(..., description="ID de la factura"),
|
|
company_id: int = Query(...),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
|
|
if not invoice:
|
|
raise HTTPException(status_code=404, detail="Factura no encontrada")
|
|
perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
|
|
validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.view"])
|
|
try:
|
|
return services.InvoiceService.export_aviso_cruce(db, invoice)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.exception("interface_aviso_cruce failed: %s", e)
|
|
raise HTTPException(status_code=500, detail=f"Error al generar interfaz Aviso Cruce: {str(e)}")
|
|
|
|
|
|
@router.get("/invoices/{invoice_id}/interfaces/aaduanal-rs")
|
|
def interface_aaduanal_rs(
|
|
invoice_id: int = Path(..., description="ID de la factura"),
|
|
company_id: int = Query(...),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
|
|
if not invoice:
|
|
raise HTTPException(status_code=404, detail="Factura no encontrada")
|
|
perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
|
|
validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.view"])
|
|
try:
|
|
return services.InvoiceService.export_aaduanal_rs(db, invoice)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.exception("interface_aaduanal_rs failed: %s", e)
|
|
raise HTTPException(status_code=500, detail=f"Error al generar interfaz AAduanal_RS: {str(e)}")
|
|
|
|
|
|
@router.get("/invoices/{invoice_id}/interfaces/caaarem")
|
|
def interface_caaarem(
|
|
invoice_id: int = Path(...),
|
|
company_id: int = Query(...),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
|
|
if not invoice:
|
|
raise HTTPException(status_code=404, detail="Factura no encontrada")
|
|
perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
|
|
validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.view"])
|
|
try:
|
|
return services.InvoiceService.export_caaarem(db, invoice)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.exception("interface_caaarem failed: %s", e)
|
|
raise HTTPException(status_code=500, detail=f"Error al generar interfaz CAAAREM: {str(e)}")
|
|
|
|
|
|
@router.get("/invoices/{invoice_id}/interfaces/cp-genesis")
|
|
def interface_cp_genesis(
|
|
invoice_id: int = Path(...),
|
|
company_id: int = Query(...),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
|
|
if not invoice:
|
|
raise HTTPException(status_code=404, detail="Factura no encontrada")
|
|
perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
|
|
validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.view"])
|
|
try:
|
|
return services.InvoiceService.export_cp_genesis(db, invoice)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.exception("interface_cp_genesis failed: %s", e)
|
|
raise HTTPException(status_code=500, detail=f"Error al generar interfaz Carta Porte Genesis: {str(e)}")
|
|
|
|
|
|
# --- RUTA DE LISTADO (FILTROS) ---
|
|
|
|
@router.get("/invoices/", response_model=schemas.InvoiceHeaderListResponse)
|
|
def list_invoices(
|
|
request: Request,
|
|
company_id: int = Query(..., description="Company ID"),
|
|
page: int = Query(1, ge=1, description="Page number"),
|
|
page_size: int = Query(50, ge=1, le=200, description="Items per page"),
|
|
search: str = Query(None, description="Search by invoice number"),
|
|
status: Optional[Any] = Query(None, description="Filter by status"),
|
|
operation_type: schemas.OperationType = Query(None, description="Filter by operation type"),
|
|
invoice_type: str = Query(None, description="Filter by invoice type"),
|
|
manifest_number: str = Query(None, description="Filter by manifest number"),
|
|
pedimento: str = Query(None, description="Filter by pedimento"),
|
|
invoice_number: str = Query(None, description="Filter by invoice number"),
|
|
project_number: str = Query(None, description="Filter by project number"),
|
|
year: str = Query(None, description="Filter by year"),
|
|
sort_by: Optional[str] = Query(None, description="Column to sort by"),
|
|
sort_order: Optional[str] = Query("asc", pattern="^(asc|desc)$", description="Sort order (asc or desc)"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
"""
|
|
List invoices with optional filters and granular permission enforcement.
|
|
"""
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
|
|
from api.v1.modules.core.permissions.service import PermissionService
|
|
|
|
user_id = current_user.get("sub") or current_user.get("id")
|
|
# Bypass para "ver todo": hub_admin atestado por el Hub o super_admin local
|
|
# en la compañía. Se evita depender de claims del JWT crudo.
|
|
is_global_admin = is_hub_admin(current_user) or _has_local_super_admin_role(
|
|
db, user_id, company_id
|
|
)
|
|
perm_service = PermissionService(db)
|
|
perm_codes = perm_service.get_user_permissions(user_id, company_id)
|
|
|
|
if settings.ENVIRONMENT == "development":
|
|
user_name = current_user.get("preferred_username") or current_user.get(
|
|
"email", "Desconocido"
|
|
)
|
|
logger.debug(
|
|
"[invoices.list] user=%s id=%s company=%s perms=%s",
|
|
user_name,
|
|
user_id,
|
|
company_id,
|
|
perm_codes,
|
|
)
|
|
|
|
allowed_filters = build_allowed_types_from_view_permissions(perm_codes)
|
|
|
|
if not allowed_filters:
|
|
if is_global_admin:
|
|
logger.debug(
|
|
"[invoices.list] global admin without invoice view perms — unfiltered types fallback"
|
|
)
|
|
allowed_filters = None
|
|
else:
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="No tienes permisos para ver facturas en esta empresa",
|
|
)
|
|
|
|
try:
|
|
skip = (page - 1) * page_size
|
|
|
|
# Combinar filtros de búsqueda con filtros granulares de permisos
|
|
filters = {
|
|
"invoice_number": invoice_number,
|
|
"status": status,
|
|
"operation_type": operation_type.value if operation_type else None,
|
|
"invoice_type": invoice_type,
|
|
"manifest_number": manifest_number,
|
|
"pedimento": pedimento,
|
|
"project_number": project_number,
|
|
"year": year,
|
|
"allowed_types": allowed_filters,
|
|
"system": get_active_system(request),
|
|
}
|
|
|
|
filters = {k: v for k, v in filters.items() if v is not None}
|
|
|
|
items, total = services.InvoiceService.get_all(
|
|
db, tenant_id, company_id, skip=skip, limit=page_size, filters=filters, sort_by=sort_by, sort_order=sort_order
|
|
)
|
|
|
|
return {
|
|
"items": items,
|
|
"total": total,
|
|
"page": page,
|
|
"page_size": page_size
|
|
}
|
|
except BaseAPIException:
|
|
raise
|
|
except Exception as e:
|
|
logger.exception("list_invoices failed: %s", e)
|
|
raise HTTPException(status_code=500, detail=f"Internal server error in invoices list: {str(e)}")
|
|
|
|
|
|
# --- RUTAS DE LOGÍSTICA ---
|
|
|
|
@router.get("/invoices/{invoice_id}/logistics", response_model=list[schemas.InvoiceLogisticsResponse])
|
|
def get_invoice_logistics(
|
|
invoice_id: int = Path(..., description="Invoice ID"),
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
|
|
if not invoice: raise HTTPException(status_code=404, detail="Invoice not found")
|
|
|
|
perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
|
|
validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.view"])
|
|
|
|
return services.InvoiceLogisticsService.get_all_by_invoice(db, invoice_id)
|
|
|
|
@router.post("/invoices/{invoice_id}/logistics", response_model=schemas.InvoiceLogisticsResponse, status_code=201)
|
|
def create_invoice_logistics(
|
|
logistics_data: schemas.InvoiceLogisticsCreate,
|
|
invoice_id: int = Path(..., description="Invoice ID"),
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
|
|
if not invoice: raise HTTPException(status_code=404, detail="Invoice not found")
|
|
|
|
# Editar los hijos cuenta como editar la factura padre
|
|
perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
|
|
validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.edit"])
|
|
|
|
return services.InvoiceLogisticsService.create(db, logistics_data, invoice_id, tenant_id, company_id)
|
|
|
|
@router.delete("/invoices/{invoice_id}/logistics/{logistics_id}", status_code=204)
|
|
def delete_invoice_logistics(
|
|
invoice_id: int = Path(..., description="Invoice ID"),
|
|
logistics_id: int = Path(..., description="Logistics ID"),
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
|
|
if not invoice: raise HTTPException(status_code=404, detail="Invoice not found")
|
|
|
|
perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
|
|
validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.edit"])
|
|
|
|
if not services.InvoiceLogisticsService.delete(db, logistics_id, invoice_id):
|
|
raise HTTPException(status_code=404, detail="Logistics entry not found")
|
|
return None
|
|
|
|
|
|
# --- RUTAS DE DETALLES DE VENTA (PARTIDAS) ---
|
|
|
|
@router.get("/invoices/{invoice_id}/details", response_model=list[schemas.InvoiceSalesDetailsResponse])
|
|
def get_invoice_details(
|
|
invoice_id: int = Path(..., description="Invoice ID"),
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
|
|
if not invoice: raise HTTPException(status_code=404, detail="Invoice not found")
|
|
|
|
perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
|
|
validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.view"])
|
|
|
|
return services.InvoiceSalesDetailsService.get_all_by_invoice(db, invoice_id)
|
|
|
|
@router.post("/invoices/{invoice_id}/details", response_model=schemas.InvoiceSalesDetailsResponse, status_code=201)
|
|
def create_invoice_detail(
|
|
detail_data: schemas.InvoiceSalesDetailsCreate,
|
|
invoice_id: int = Path(..., description="Invoice ID"),
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
|
|
if not invoice: raise HTTPException(status_code=404, detail="Invoice not found")
|
|
|
|
perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
|
|
validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.edit"])
|
|
|
|
return services.InvoiceSalesDetailsService.create(db, detail_data, invoice_id, tenant_id, company_id)
|
|
|
|
@router.delete("/invoices/{invoice_id}/details/{detail_id}", status_code=204)
|
|
def delete_invoice_detail(
|
|
invoice_id: int = Path(..., description="Invoice ID"),
|
|
detail_id: int = Path(..., description="Detail ID"),
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
|
|
if not invoice: raise HTTPException(status_code=404, detail="Invoice not found")
|
|
|
|
perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
|
|
validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.edit"])
|
|
|
|
if not services.InvoiceSalesDetailsService.delete(db, detail_id, invoice_id):
|
|
raise HTTPException(status_code=404, detail="Sales detail not found")
|
|
return None
|
|
|
|
|
|
# --- RUTAS DE COBRANZA (COLLECTIONS) ---
|
|
|
|
@router.get("/invoices/{invoice_id}/collections", response_model=list[schemas.InvoiceCollectionsResponse])
|
|
def get_invoice_collections(
|
|
invoice_id: int = Path(..., description="Invoice ID"),
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
|
|
if not invoice: raise HTTPException(status_code=404, detail="Invoice not found")
|
|
|
|
perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
|
|
validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.view"])
|
|
|
|
return services.InvoiceCollectionsService.get_all_by_invoice(db, invoice_id)
|
|
|
|
@router.post("/invoices/{invoice_id}/collections", response_model=schemas.InvoiceCollectionsResponse, status_code=201)
|
|
def create_invoice_collection(
|
|
collection_data: schemas.InvoiceCollectionsCreate,
|
|
invoice_id: int = Path(..., description="Invoice ID"),
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
|
|
if not invoice: raise HTTPException(status_code=404, detail="Invoice not found")
|
|
|
|
perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
|
|
validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.edit"])
|
|
|
|
return services.InvoiceCollectionsService.create(db, collection_data, invoice_id, tenant_id, company_id)
|
|
|
|
@router.delete("/invoices/{invoice_id}/collections/{collection_id}", status_code=204)
|
|
def delete_invoice_collection(
|
|
invoice_id: int = Path(..., description="Invoice ID"),
|
|
collection_id: int = Path(..., description="Collection ID"),
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
|
|
if not invoice: raise HTTPException(status_code=404, detail="Invoice not found")
|
|
|
|
perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
|
|
validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.edit"])
|
|
|
|
if not services.InvoiceCollectionsService.delete(db, collection_id, invoice_id):
|
|
raise HTTPException(status_code=404, detail="Collection not found")
|
|
return None |