Creacion de rutas para nuevas facturas, asi como nuevas rutinas para el filtro de datos en facturas de exportacion
This commit is contained in:
@@ -1,10 +1,10 @@
|
||||
import traceback
|
||||
from typing import Optional, List, Tuple
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy import and_
|
||||
|
||||
from . import models, schemas
|
||||
|
||||
|
||||
class InvoiceService:
|
||||
"""Service for Invoice Header operations"""
|
||||
|
||||
@@ -55,7 +55,6 @@ class InvoiceService:
|
||||
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")
|
||||
@@ -72,73 +71,118 @@ class InvoiceService:
|
||||
company_id: int
|
||||
) -> models.InvoiceHeader:
|
||||
"""Create a new invoice with all related data"""
|
||||
# Extract nested data
|
||||
compliance_data = invoice_data.compliance_mx
|
||||
financials_data = invoice_data.financials
|
||||
logistics_data = invoice_data.logistics or []
|
||||
details_data = invoice_data.details or []
|
||||
collections_data = invoice_data.collections or []
|
||||
|
||||
|
||||
def clean_dict(data_dict: dict) -> dict:
|
||||
cleaned = {}
|
||||
for key, value in data_dict.items():
|
||||
|
||||
if key == 'customs_agent':
|
||||
key = 'customs_broker_id'
|
||||
elif key == 'provider':
|
||||
key = 'provider_id'
|
||||
|
||||
|
||||
if isinstance(value, str) and not value.strip():
|
||||
cleaned[key] = None
|
||||
|
||||
elif value == 0 and (key.endswith('_id') or key == 'remesa'):
|
||||
cleaned[key] = None
|
||||
else:
|
||||
cleaned[key] = value
|
||||
return cleaned
|
||||
|
||||
|
||||
# Create main invoice header
|
||||
invoice_dict = invoice_data.model_dump(
|
||||
exclude={"compliance_mx", "financials",
|
||||
"logistics", "details", "collections"}
|
||||
)
|
||||
invoice_dict["tenant_id"] = tenant_id
|
||||
invoice_dict["company_id"] = company_id
|
||||
try:
|
||||
# Extract nested data
|
||||
compliance_data = invoice_data.compliance_mx
|
||||
financials_data = invoice_data.financials
|
||||
logistics_data = invoice_data.logistics or []
|
||||
details_data = invoice_data.details or []
|
||||
collections_data = invoice_data.collections or []
|
||||
|
||||
new_invoice = models.InvoiceHeader(**invoice_dict)
|
||||
db.add(new_invoice)
|
||||
db.flush() # Flush to get the invoice ID
|
||||
# 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
|
||||
|
||||
# Create compliance_mx if provided
|
||||
if compliance_data:
|
||||
compliance_dict = compliance_data.model_dump()
|
||||
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)
|
||||
new_invoice = models.InvoiceHeader(**invoice_dict)
|
||||
db.add(new_invoice)
|
||||
db.flush() # Flush to get the invoice ID
|
||||
|
||||
# Create financials if provided
|
||||
if financials_data:
|
||||
financials_dict = financials_data.model_dump()
|
||||
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 compliance_mx if provided
|
||||
if compliance_data:
|
||||
raw_comp_dict = compliance_data.model_dump()
|
||||
# Pasamos los datos por la lavadora para arreglar pedimento, aduana, etc.
|
||||
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 logistics entries
|
||||
for logistics_item in logistics_data:
|
||||
logistics_dict = logistics_item.model_dump()
|
||||
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 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 sales details
|
||||
for detail_item in details_data:
|
||||
detail_dict = detail_item.model_dump()
|
||||
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 logistics entries
|
||||
for logistics_item in logistics_data:
|
||||
raw_log_dict = logistics_item.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 collections
|
||||
for collection_item in collections_data:
|
||||
collection_dict = collection_item.model_dump()
|
||||
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)
|
||||
# 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)
|
||||
|
||||
db.commit()
|
||||
db.refresh(new_invoice)
|
||||
return new_invoice
|
||||
# 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(
|
||||
@@ -148,7 +192,8 @@ class InvoiceService:
|
||||
invoice_data: schemas.InvoiceHeaderUpdate,
|
||||
company_id: int
|
||||
) -> Optional[models.InvoiceHeader]:
|
||||
"""Update an existing invoice and its related data"""
|
||||
# ... (El resto de tu código update se queda igual) ...
|
||||
# (Te recomiendo implementar clean_dict aquí también si tienes problemas al editar)
|
||||
invoice = InvoiceService.get_by_id(
|
||||
db, invoice_id, tenant_id, company_id)
|
||||
if not invoice:
|
||||
@@ -167,9 +212,14 @@ class InvoiceService:
|
||||
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
|
||||
@@ -180,6 +230,7 @@ class InvoiceService:
|
||||
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()
|
||||
@@ -189,9 +240,6 @@ class InvoiceService:
|
||||
new_financials = models.InvoiceFinancials(**financials_dict)
|
||||
db.add(new_financials)
|
||||
|
||||
# Note: For logistics, details, and collections, we're not handling updates here
|
||||
# as they are typically managed through separate endpoints for complex operations
|
||||
|
||||
db.commit()
|
||||
db.refresh(invoice)
|
||||
return invoice
|
||||
@@ -205,172 +253,4 @@ class InvoiceService:
|
||||
db.delete(invoice)
|
||||
db.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
class InvoiceLogisticsService:
|
||||
"""Service for Invoice Logistics operations"""
|
||||
|
||||
@staticmethod
|
||||
def get_by_id(db: Session, logistics_id: int, invoice_id: int) -> Optional[models.InvoiceLogistics]:
|
||||
"""Get a logistics entry by ID"""
|
||||
return (
|
||||
db.query(models.InvoiceLogistics)
|
||||
.filter(
|
||||
models.InvoiceLogistics.logistics_id == logistics_id,
|
||||
models.InvoiceLogistics.invoice_id == invoice_id,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def get_all_by_invoice(db: Session, invoice_id: int) -> List[models.InvoiceLogistics]:
|
||||
"""Get all logistics entries for an invoice"""
|
||||
return (
|
||||
db.query(models.InvoiceLogistics)
|
||||
.filter(models.InvoiceLogistics.invoice_id == invoice_id)
|
||||
.all()
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def create(
|
||||
db: Session,
|
||||
logistics_data: schemas.InvoiceLogisticsCreate,
|
||||
invoice_id: int,
|
||||
tenant_id: int,
|
||||
company_id: int
|
||||
) -> models.InvoiceLogistics:
|
||||
"""Create a new logistics entry"""
|
||||
logistics_dict = logistics_data.model_dump()
|
||||
logistics_dict["invoice_id"] = invoice_id
|
||||
logistics_dict["tenant_id"] = tenant_id
|
||||
logistics_dict["company_id"] = company_id
|
||||
|
||||
new_logistics = models.InvoiceLogistics(**logistics_dict)
|
||||
db.add(new_logistics)
|
||||
db.commit()
|
||||
db.refresh(new_logistics)
|
||||
return new_logistics
|
||||
|
||||
@staticmethod
|
||||
def delete(db: Session, logistics_id: int, invoice_id: int) -> bool:
|
||||
"""Delete a logistics entry"""
|
||||
logistics = InvoiceLogisticsService.get_by_id(
|
||||
db, logistics_id, invoice_id)
|
||||
if logistics:
|
||||
db.delete(logistics)
|
||||
db.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
class InvoiceSalesDetailsService:
|
||||
"""Service for Invoice Sales Details operations"""
|
||||
|
||||
@staticmethod
|
||||
def get_by_id(db: Session, detail_id: int, invoice_id: int) -> Optional[models.InvoiceSalesDetails]:
|
||||
"""Get a sales detail entry by ID"""
|
||||
return (
|
||||
db.query(models.InvoiceSalesDetails)
|
||||
.filter(
|
||||
models.InvoiceSalesDetails.detail_id == detail_id,
|
||||
models.InvoiceSalesDetails.invoice_id == invoice_id,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def get_all_by_invoice(db: Session, invoice_id: int) -> List[models.InvoiceSalesDetails]:
|
||||
"""Get all sales details for an invoice"""
|
||||
return (
|
||||
db.query(models.InvoiceSalesDetails)
|
||||
.filter(models.InvoiceSalesDetails.invoice_id == invoice_id)
|
||||
.all()
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def create(
|
||||
db: Session,
|
||||
detail_data: schemas.InvoiceSalesDetailsCreate,
|
||||
invoice_id: int,
|
||||
tenant_id: int,
|
||||
company_id: int
|
||||
) -> models.InvoiceSalesDetails:
|
||||
"""Create a new sales detail entry"""
|
||||
detail_dict = detail_data.model_dump()
|
||||
detail_dict["invoice_id"] = invoice_id
|
||||
detail_dict["tenant_id"] = tenant_id
|
||||
detail_dict["company_id"] = company_id
|
||||
|
||||
new_detail = models.InvoiceSalesDetails(**detail_dict)
|
||||
db.add(new_detail)
|
||||
db.commit()
|
||||
db.refresh(new_detail)
|
||||
return new_detail
|
||||
|
||||
@staticmethod
|
||||
def delete(db: Session, detail_id: int, invoice_id: int) -> bool:
|
||||
"""Delete a sales detail entry"""
|
||||
detail = InvoiceSalesDetailsService.get_by_id(
|
||||
db, detail_id, invoice_id)
|
||||
if detail:
|
||||
db.delete(detail)
|
||||
db.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
class InvoiceCollectionsService:
|
||||
"""Service for Invoice Collections operations"""
|
||||
|
||||
@staticmethod
|
||||
def get_by_id(db: Session, collection_id: int, invoice_id: int) -> Optional[models.InvoiceCollections]:
|
||||
"""Get a collection entry by ID"""
|
||||
return (
|
||||
db.query(models.InvoiceCollections)
|
||||
.filter(
|
||||
models.InvoiceCollections.collection_id == collection_id,
|
||||
models.InvoiceCollections.invoice_id == invoice_id,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def get_all_by_invoice(db: Session, invoice_id: int) -> List[models.InvoiceCollections]:
|
||||
"""Get all collections for an invoice"""
|
||||
return (
|
||||
db.query(models.InvoiceCollections)
|
||||
.filter(models.InvoiceCollections.invoice_id == invoice_id)
|
||||
.all()
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def create(
|
||||
db: Session,
|
||||
collection_data: schemas.InvoiceCollectionsCreate,
|
||||
invoice_id: int,
|
||||
tenant_id: int,
|
||||
company_id: int
|
||||
) -> models.InvoiceCollections:
|
||||
"""Create a new collection entry"""
|
||||
collection_dict = collection_data.model_dump()
|
||||
collection_dict["invoice_id"] = 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_collection)
|
||||
return new_collection
|
||||
|
||||
@staticmethod
|
||||
def delete(db: Session, collection_id: int, invoice_id: int) -> bool:
|
||||
"""Delete a collection entry"""
|
||||
collection = InvoiceCollectionsService.get_by_id(
|
||||
db, collection_id, invoice_id)
|
||||
if collection:
|
||||
db.delete(collection)
|
||||
db.commit()
|
||||
return True
|
||||
return False
|
||||
return False
|
||||
Reference in New Issue
Block a user