307 lines
12 KiB
Python
307 lines
12 KiB
Python
import traceback
|
|
from typing import Optional, List, Tuple
|
|
from sqlalchemy.orm import Session
|
|
from core.exceptions import ErrorCollector, DuplicateResourceException
|
|
from .common.mappers import clean_dict
|
|
from .imports.temporary.validators.create import validate_create
|
|
from .imports.temporary.validators.update import validate_update
|
|
from .common.common_validators import invoice_exists
|
|
|
|
from . import models, schemas
|
|
|
|
|
|
class InvoiceService:
|
|
"""Service for Invoice Header operations"""
|
|
|
|
@staticmethod
|
|
def get_by_id(
|
|
db: Session, invoice_id: int, tenant_id: int, company_id: int
|
|
) -> Optional[models.InvoiceHeader]:
|
|
"""Get an invoice by ID with tenant/company validation"""
|
|
return (
|
|
db.query(models.InvoiceHeader)
|
|
.filter(
|
|
models.InvoiceHeader.id == invoice_id,
|
|
models.InvoiceHeader.tenant_id == tenant_id,
|
|
models.InvoiceHeader.company_id == company_id,
|
|
)
|
|
.first()
|
|
)
|
|
|
|
@staticmethod
|
|
def get_all(
|
|
db: Session,
|
|
tenant_id: int,
|
|
company_id: int,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
filters: Optional[dict] = None,
|
|
) -> Tuple[List[models.InvoiceHeader], int]:
|
|
"""Get all invoices for a tenant/company with pagination and optional filters"""
|
|
query = db.query(models.InvoiceHeader).filter(
|
|
models.InvoiceHeader.tenant_id == tenant_id,
|
|
models.InvoiceHeader.company_id == company_id,
|
|
)
|
|
|
|
# Apply filters if provided
|
|
if filters:
|
|
if filters.get("status"):
|
|
query = query.filter(models.InvoiceHeader.status == filters["status"])
|
|
if filters.get("operation_type"):
|
|
query = query.filter(
|
|
models.InvoiceHeader.operation_type == filters["operation_type"]
|
|
)
|
|
if filters.get("invoice_type"):
|
|
query = query.filter(
|
|
models.InvoiceHeader.invoice_type == filters["invoice_type"]
|
|
)
|
|
if filters.get("invoice_number"):
|
|
query = query.filter(
|
|
models.InvoiceHeader.invoice_number.ilike(
|
|
f"%{filters['invoice_number']}%"
|
|
)
|
|
)
|
|
if filters.get("pedimento"):
|
|
query = query.join(models.InvoiceComplianceMx).filter(
|
|
models.InvoiceComplianceMx.pedimento.ilike(
|
|
f"%{filters['pedimento']}%"
|
|
)
|
|
)
|
|
if (
|
|
not filters.get("invoice_type")
|
|
and filters.get("operation_type") == "exp"
|
|
):
|
|
query = query.filter(models.InvoiceHeader.operation_type != "REPAR")
|
|
|
|
total = query.count()
|
|
items = query.offset(skip).limit(limit).all()
|
|
return items, total
|
|
|
|
@staticmethod
|
|
def create(
|
|
db: Session,
|
|
invoice_data: schemas.InvoiceHeaderCreate,
|
|
tenant_id: int,
|
|
company_id: int,
|
|
) -> models.InvoiceHeader:
|
|
"""Create a new invoice with all related data"""
|
|
|
|
# Validaciones con ErrorCollector
|
|
errors = ErrorCollector()
|
|
|
|
# Validar si la factura ya existe
|
|
invoice_exists(db, invoice_data.invoice_number, tenant_id, company_id, errors)
|
|
validate_create(db, invoice_data, tenant_id, company_id, errors)
|
|
|
|
# Si hay errores, lanzar excepción ANTES de intentar crear
|
|
errors.raise_if_errors("Error al crear la factura")
|
|
|
|
# Extract nested data
|
|
compliance_data = invoice_data.compliance_mx
|
|
financials_data = invoice_data.financials
|
|
logistics_data = invoice_data.logistics
|
|
details_data = invoice_data.details or []
|
|
collections_data = invoice_data.collections or []
|
|
|
|
try:
|
|
|
|
# Create main invoice header
|
|
raw_invoice_dict = invoice_data.model_dump(
|
|
exclude={
|
|
"compliance_mx",
|
|
"financials",
|
|
"logistics",
|
|
"details",
|
|
"collections",
|
|
}
|
|
)
|
|
invoice_dict = clean_dict(raw_invoice_dict)
|
|
invoice_dict["tenant_id"] = tenant_id
|
|
invoice_dict["company_id"] = company_id
|
|
|
|
new_invoice = models.InvoiceHeader(**invoice_dict)
|
|
|
|
db.add(new_invoice)
|
|
db.flush() # Flush to get the invoice ID
|
|
|
|
# Create compliance_mx if provided
|
|
if compliance_data:
|
|
raw_comp_dict = compliance_data.model_dump()
|
|
compliance_dict = clean_dict(raw_comp_dict)
|
|
|
|
compliance_dict["invoice_id"] = new_invoice.id
|
|
compliance_dict["tenant_id"] = tenant_id
|
|
compliance_dict["company_id"] = company_id
|
|
|
|
new_compliance = models.InvoiceComplianceMx(**compliance_dict)
|
|
db.add(new_compliance)
|
|
|
|
# Create financials if provided
|
|
if financials_data:
|
|
raw_fin_dict = financials_data.model_dump()
|
|
financials_dict = clean_dict(raw_fin_dict)
|
|
|
|
financials_dict["invoice_id"] = new_invoice.id
|
|
financials_dict["tenant_id"] = tenant_id
|
|
financials_dict["company_id"] = company_id
|
|
|
|
new_financials = models.InvoiceFinancials(**financials_dict)
|
|
db.add(new_financials)
|
|
|
|
# Create logistics entries
|
|
if logistics_data:
|
|
raw_log_dict = logistics_data.model_dump()
|
|
logistics_dict = clean_dict(raw_log_dict)
|
|
|
|
logistics_dict["invoice_id"] = new_invoice.id
|
|
logistics_dict["tenant_id"] = tenant_id
|
|
logistics_dict["company_id"] = company_id
|
|
new_logistics = models.InvoiceLogistics(**logistics_dict)
|
|
db.add(new_logistics)
|
|
|
|
# Create sales details
|
|
for detail_item in details_data:
|
|
raw_det_dict = detail_item.model_dump()
|
|
detail_dict = clean_dict(raw_det_dict)
|
|
|
|
detail_dict["invoice_id"] = new_invoice.id
|
|
detail_dict["tenant_id"] = tenant_id
|
|
detail_dict["company_id"] = company_id
|
|
new_detail = models.InvoiceSalesDetails(**detail_dict)
|
|
db.add(new_detail)
|
|
|
|
# Create collections
|
|
for collection_item in collections_data:
|
|
raw_col_dict = collection_item.model_dump()
|
|
collection_dict = clean_dict(raw_col_dict)
|
|
|
|
collection_dict["invoice_id"] = new_invoice.id
|
|
collection_dict["tenant_id"] = tenant_id
|
|
collection_dict["company_id"] = company_id
|
|
new_collection = models.InvoiceCollections(**collection_dict)
|
|
db.add(new_collection)
|
|
|
|
db.commit()
|
|
db.refresh(new_invoice)
|
|
return new_invoice
|
|
|
|
except Exception as e:
|
|
db.rollback()
|
|
print("\n\n🔥 ERROR AL GUARDAR FACTURA 🔥")
|
|
print(f"Error: {str(e)}")
|
|
traceback.print_exc() # Esto imprime el error real en la consola
|
|
print("--------------------------------\n")
|
|
raise e
|
|
|
|
@staticmethod
|
|
def update(
|
|
db: Session,
|
|
invoice_id: int,
|
|
tenant_id: int,
|
|
invoice_data: schemas.InvoiceHeaderUpdate,
|
|
company_id: int,
|
|
) -> Optional[models.InvoiceHeader]:
|
|
"""Update an existing invoice with validation"""
|
|
|
|
# Validaciones con ErrorCollector
|
|
errors = ErrorCollector()
|
|
|
|
# Obtener la factura existente
|
|
invoice = InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
|
|
if not invoice:
|
|
return None
|
|
|
|
# Si se cambió el número de factura, validar que no exista otra con ese número
|
|
if invoice_data.invoice_number and invoice_data.invoice_number != invoice.invoice_number:
|
|
# Verificar que no exista otra factura con el nuevo número
|
|
existing_invoice = (
|
|
db.query(models.InvoiceHeader.id)
|
|
.filter(
|
|
models.InvoiceHeader.invoice_number == invoice_data.invoice_number,
|
|
models.InvoiceHeader.tenant_id == tenant_id,
|
|
models.InvoiceHeader.company_id == company_id,
|
|
models.InvoiceHeader.id != invoice_id, # Excluir la factura actual
|
|
)
|
|
.first()
|
|
)
|
|
|
|
if existing_invoice:
|
|
errors.add_duplicate_error(
|
|
"invoice_number",
|
|
invoice_data.invoice_number,
|
|
f"Ya existe otra factura con el número '{invoice_data.invoice_number}'",
|
|
)
|
|
validate_update(invoice_data, invoice, errors)
|
|
|
|
# Si hay errores, lanzar excepción ANTES de actualizar
|
|
errors.raise_if_errors("Error al actualizar la factura")
|
|
|
|
# Update main invoice header fields
|
|
update_dict = invoice_data.model_dump(
|
|
exclude={
|
|
"compliance_mx",
|
|
"financials",
|
|
"logistics",
|
|
"details",
|
|
"collections",
|
|
},
|
|
exclude_unset=True,
|
|
)
|
|
for key, value in update_dict.items():
|
|
setattr(invoice, key, value)
|
|
|
|
# Update compliance_mx if provided
|
|
if invoice_data.compliance_mx is not None:
|
|
if invoice.compliance_mx:
|
|
for key, value in invoice_data.compliance_mx.model_dump(
|
|
exclude_unset=True
|
|
).items():
|
|
# Parche rápido para update
|
|
if value == "":
|
|
value = None
|
|
setattr(invoice.compliance_mx, key, value)
|
|
else:
|
|
compliance_dict = invoice_data.compliance_mx.model_dump()
|
|
# Aplicar limpieza manual si es necesario
|
|
if "customs_agent" in compliance_dict:
|
|
compliance_dict["customs_broker_id"] = compliance_dict.pop(
|
|
"customs_agent"
|
|
)
|
|
|
|
compliance_dict["invoice_id"] = invoice.id
|
|
compliance_dict["tenant_id"] = tenant_id
|
|
compliance_dict["company_id"] = company_id
|
|
new_compliance = models.InvoiceComplianceMx(**compliance_dict)
|
|
db.add(new_compliance)
|
|
|
|
# Update financials if provided
|
|
if invoice_data.financials is not None:
|
|
if invoice.financials:
|
|
for key, value in invoice_data.financials.model_dump(
|
|
exclude_unset=True
|
|
).items():
|
|
if value == "":
|
|
value = None
|
|
setattr(invoice.financials, key, value)
|
|
else:
|
|
financials_dict = invoice_data.financials.model_dump()
|
|
financials_dict["invoice_id"] = invoice.id
|
|
financials_dict["tenant_id"] = tenant_id
|
|
financials_dict["company_id"] = company_id
|
|
new_financials = models.InvoiceFinancials(**financials_dict)
|
|
db.add(new_financials)
|
|
|
|
db.commit()
|
|
db.refresh(invoice)
|
|
return invoice
|
|
|
|
@staticmethod
|
|
def delete(db: Session, invoice_id: int, tenant_id: int, company_id: int) -> bool:
|
|
"""Delete an invoice and all related data (cascade delete)"""
|
|
invoice = InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
|
|
if invoice:
|
|
db.delete(invoice)
|
|
db.commit()
|
|
return True
|
|
return False
|