769 lines
29 KiB
Python
769 lines
29 KiB
Python
import traceback
|
|
from typing import Optional, List, Tuple
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import and_, func, or_
|
|
from core.exceptions import ErrorCollector, DuplicateResourceException
|
|
from core.context import get_user_context
|
|
from api.v1.modules.a76.audit_log.models import AuditLog
|
|
from api.v1.modules.a76.audit_log.services.service import AuditService
|
|
from .common.mappers import clean_dict
|
|
from .imports.validators.create import validate_create as validate_create_import
|
|
from .imports.validators.update import validate_update as validate_update_import
|
|
from .exports.validators.create import validate_create as validate_create_export
|
|
from .exports.validators.update import validate_update as validate_update_export
|
|
from .common.common_validators import invoice_exists
|
|
from api.v1.modules.a76.items.models import LineItem
|
|
from . import models, schemas
|
|
|
|
|
|
def _autofill_transport_int_ids(
|
|
db: Session,
|
|
logistics_target,
|
|
tenant_id: int,
|
|
company_id: int,
|
|
) -> None:
|
|
"""
|
|
Ensures invoice logistics internal int IDs are populated when catalog rows exist.
|
|
|
|
String keys (carrier_id, transport_id, trailer_num) are matched to existing catalog
|
|
rows for the tenant/company; no placeholder rows are created. Invalid references must
|
|
be rejected by validate_common before persist.
|
|
"""
|
|
|
|
# Local imports to avoid circular dependencies.
|
|
from api.v1.modules.a76.transportation.transporters.models import Transporter
|
|
from api.v1.modules.a76.transportation.vehicles.models import Vehicle
|
|
from api.v1.modules.a76.transportation.trailers.models import Trailer
|
|
|
|
def _get(field: str):
|
|
if isinstance(logistics_target, dict):
|
|
return logistics_target.get(field)
|
|
return getattr(logistics_target, field, None)
|
|
|
|
def _set(field: str, value):
|
|
if isinstance(logistics_target, dict):
|
|
logistics_target[field] = value
|
|
else:
|
|
setattr(logistics_target, field, value)
|
|
|
|
carrier_code = _get("carrier_id")
|
|
carrier_int_id = _get("carrier_int_id")
|
|
if carrier_code and carrier_int_id is None:
|
|
transporter_obj = (
|
|
db.query(Transporter)
|
|
.filter(
|
|
Transporter.transporter_key == carrier_code,
|
|
Transporter.tenant_id == tenant_id,
|
|
Transporter.company_id == company_id,
|
|
)
|
|
.first()
|
|
)
|
|
if transporter_obj:
|
|
_set("carrier_int_id", transporter_obj.transporter_id)
|
|
|
|
transport_code = _get("transport_id")
|
|
transport_int_id = _get("transport_int_id")
|
|
if transport_code and transport_int_id is None:
|
|
vehicle_obj = (
|
|
db.query(Vehicle)
|
|
.filter(
|
|
func.upper(Vehicle.vehicle_key) == str(transport_code).strip().upper(),
|
|
Vehicle.tenant_id == tenant_id,
|
|
Vehicle.company_id == company_id,
|
|
)
|
|
.first()
|
|
)
|
|
if vehicle_obj:
|
|
_set("transport_int_id", vehicle_obj.vehicle_id)
|
|
|
|
trailer_code = _get("trailer_num")
|
|
trailer_int_id = _get("trailer_int_id")
|
|
if trailer_code and trailer_int_id is None:
|
|
trailer_obj = (
|
|
db.query(Trailer)
|
|
.filter(
|
|
func.upper(Trailer.trailer_number)
|
|
== str(trailer_code).strip().upper(),
|
|
Trailer.tenant_id == tenant_id,
|
|
Trailer.company_id == company_id,
|
|
)
|
|
.first()
|
|
)
|
|
if trailer_obj:
|
|
_set("trailer_int_id", trailer_obj.trailer_id)
|
|
|
|
# Placa: si no vino valor, tomar del vehículo y si no del remolque (catálogo)
|
|
lp = _get("license_plate")
|
|
if lp is None or str(lp).strip() == "":
|
|
veh = None
|
|
tv_id = _get("transport_int_id")
|
|
tv_code = _get("transport_id")
|
|
if tv_id is not None:
|
|
veh = (
|
|
db.query(Vehicle)
|
|
.filter(
|
|
Vehicle.vehicle_id == tv_id,
|
|
Vehicle.tenant_id == tenant_id,
|
|
Vehicle.company_id == company_id,
|
|
)
|
|
.first()
|
|
)
|
|
elif tv_code:
|
|
veh = (
|
|
db.query(Vehicle)
|
|
.filter(
|
|
func.upper(Vehicle.vehicle_key) == str(tv_code).strip().upper(),
|
|
Vehicle.tenant_id == tenant_id,
|
|
Vehicle.company_id == company_id,
|
|
)
|
|
.first()
|
|
)
|
|
if veh and getattr(veh, "plate_number", None):
|
|
_set("license_plate", veh.plate_number)
|
|
else:
|
|
tr = None
|
|
tr_id = _get("trailer_int_id")
|
|
tr_code = _get("trailer_num")
|
|
if tr_id is not None:
|
|
tr = (
|
|
db.query(Trailer)
|
|
.filter(
|
|
Trailer.trailer_id == tr_id,
|
|
Trailer.tenant_id == tenant_id,
|
|
Trailer.company_id == company_id,
|
|
)
|
|
.first()
|
|
)
|
|
elif tr_code:
|
|
tr = (
|
|
db.query(Trailer)
|
|
.filter(
|
|
func.upper(Trailer.trailer_number)
|
|
== str(tr_code).strip().upper(),
|
|
Trailer.tenant_id == tenant_id,
|
|
Trailer.company_id == company_id,
|
|
)
|
|
.first()
|
|
)
|
|
if tr and getattr(tr, "plate_number", None):
|
|
_set("license_plate", tr.plate_number)
|
|
|
|
|
|
def _get_current_username() -> str:
|
|
"""Helper to get current username from context or fallback to System"""
|
|
try:
|
|
context = get_user_context()
|
|
if context:
|
|
# Token usually has 'preferred_username' or 'name' or 'sub'
|
|
username = (
|
|
context.get("preferred_username")
|
|
or context.get("email")
|
|
or context.get("sub")
|
|
or "System"
|
|
)
|
|
print(f"DEBUG: _get_current_username found context: {username}")
|
|
return username
|
|
except Exception:
|
|
pass
|
|
print("DEBUG: _get_current_username NO context found, using System")
|
|
return "System"
|
|
|
|
|
|
def _autofill_remesa_if_needed(db: Session, invoice_data, tenant_id: int, company_id: int) -> None:
|
|
"""
|
|
Autocalcula remesa cuando hay pedimento consolidado y remesa viene vacía.
|
|
Se hace ANTES de validar para que cumpla reglas de required en validators.
|
|
"""
|
|
try:
|
|
compliance = getattr(invoice_data, "compliance_mx", None)
|
|
if not compliance:
|
|
return
|
|
|
|
pedimento_id = getattr(compliance, "pedimento_id", None)
|
|
remesa = getattr(compliance, "remesa", None)
|
|
|
|
if not pedimento_id or remesa:
|
|
return
|
|
|
|
# No aplicar a MEX (por consistencia con CSV import donde remesa es None para MEX)
|
|
invoice_type = getattr(invoice_data, "invoice_type", None)
|
|
if invoice_type == "MEX":
|
|
return
|
|
|
|
from api.v1.modules.a76.pedmientos.models.pedimentos import Pedimentos, PedimentoType
|
|
|
|
ped = (
|
|
db.query(Pedimentos)
|
|
.filter(
|
|
Pedimentos.id == pedimento_id,
|
|
Pedimentos.tenant_id == tenant_id,
|
|
Pedimentos.company_id == company_id,
|
|
)
|
|
.first()
|
|
)
|
|
if not ped:
|
|
return
|
|
|
|
if getattr(ped, "pedimento_type", None) != PedimentoType.CONSOLIDATED:
|
|
return
|
|
|
|
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()
|
|
)
|
|
next_rem = (max_rem or 0) + 1
|
|
compliance.remesa = next_rem
|
|
except Exception:
|
|
# No bloquear guardado por fallo de autocalculo; validación normal aplicará.
|
|
return
|
|
|
|
|
|
def _ensure_create_audit_log(
|
|
db: Session,
|
|
*,
|
|
table_name: str,
|
|
record_id: str,
|
|
record_data: dict,
|
|
username: str,
|
|
tenant_id: int,
|
|
company_id: int,
|
|
) -> bool:
|
|
"""
|
|
Backup audit writer: writes CREATE only when missing.
|
|
Returns True when a backup log row was added.
|
|
"""
|
|
exists = (
|
|
db.query(AuditLog.spec_id)
|
|
.filter(
|
|
AuditLog.table_name == table_name,
|
|
AuditLog.operation_type == "CREATE",
|
|
AuditLog.record_id == record_id,
|
|
AuditLog.tenant_id == tenant_id,
|
|
AuditLog.company_id == company_id,
|
|
)
|
|
.first()
|
|
)
|
|
if exists:
|
|
return False
|
|
|
|
AuditService.log_crud_operation(
|
|
db=db,
|
|
table_name=table_name,
|
|
operation_type="CREATE",
|
|
record_data=record_data,
|
|
username=username,
|
|
record_id=record_id,
|
|
company_id=company_id,
|
|
tenant_id=tenant_id,
|
|
)
|
|
return True
|
|
|
|
|
|
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,
|
|
sort_by: Optional[str] = None,
|
|
sort_order: Optional[str] = "asc",
|
|
) -> 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
|
|
print(f"DEBUG: Invoice Query - Company: {company_id}, Filters: {filters}, Skip: {skip}, Limit: {limit}")
|
|
if filters:
|
|
# Join compliance_mx if needed for filters
|
|
needs_compliance_join = any(k in filters for k in ["pedimento", "manifest_number"])
|
|
if needs_compliance_join:
|
|
query = query.join(models.InvoiceComplianceMx)
|
|
|
|
if filters.get("status") is not None:
|
|
status_val = filters["status"]
|
|
if status_val in [True, "processed", models.InvoiceStatus.PROCESSED]:
|
|
target_status = models.InvoiceStatus.PROCESSED
|
|
elif status_val in [False, "pending", models.InvoiceStatus.PENDING]:
|
|
target_status = models.InvoiceStatus.PENDING
|
|
else:
|
|
target_status = status_val
|
|
query = query.filter(models.InvoiceHeader.status == target_status)
|
|
|
|
if filters.get("operation_type"):
|
|
ot = filters["operation_type"]
|
|
ot_val = ot.value if hasattr(ot, "value") else ot
|
|
query = query.filter(models.InvoiceHeader.operation_type == ot_val)
|
|
|
|
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("project_number"):
|
|
query = query.filter(
|
|
models.InvoiceHeader.project_number.ilike(
|
|
f"%{filters['project_number']}%"
|
|
)
|
|
)
|
|
|
|
if filters.get("year"):
|
|
try:
|
|
year_val = int(filters["year"])
|
|
query = query.filter(
|
|
func.extract("year", models.InvoiceHeader.invoice_date) == year_val
|
|
)
|
|
except (ValueError, TypeError):
|
|
pass
|
|
|
|
if filters.get("pedimento"):
|
|
from api.v1.modules.a76.pedmientos.models.pedimentos import Pedimentos
|
|
query = query.join(models.InvoiceComplianceMx.pedimento).filter(
|
|
Pedimentos.pedimento_number.ilike(f"%{filters['pedimento']}%")
|
|
)
|
|
|
|
if filters.get("manifest_number"):
|
|
query = query.filter(
|
|
models.InvoiceComplianceMx.manifest_number.ilike(
|
|
f"%{filters['manifest_number']}%"
|
|
)
|
|
)
|
|
|
|
# Special case for exports: exclude REPAR if no invoice_type specified
|
|
ot_exp = filters.get("operation_type")
|
|
ot_exp_val = ot_exp.value if hasattr(ot_exp, "value") else ot_exp
|
|
if not filters.get("invoice_type") and ot_exp_val == "exp":
|
|
query = query.filter(models.InvoiceHeader.invoice_type != "REPAR")
|
|
|
|
# Filtro por permisos granulares (allowed_types)
|
|
if "allowed_types" in filters:
|
|
allowed = filters["allowed_types"]
|
|
if allowed is None:
|
|
# Acceso global (admin o view_all) - no filtramos por tipos
|
|
pass
|
|
elif not allowed:
|
|
# Seguridad: Si el usuario NO tiene permisos para ningún tipo específico
|
|
query = query.filter(models.InvoiceHeader.id == -1)
|
|
else:
|
|
conditions = []
|
|
for op, inv in allowed:
|
|
# Aseguramos comparación insensible a mayúsculas para mayor robustez con la DB
|
|
op_str = str(op).lower()
|
|
inv_str = str(inv).lower()
|
|
conditions.append(
|
|
and_(
|
|
func.lower(models.InvoiceHeader.operation_type) == op_str,
|
|
func.lower(models.InvoiceHeader.invoice_type) == inv_str
|
|
)
|
|
)
|
|
if conditions:
|
|
query = query.filter(or_(*conditions))
|
|
else:
|
|
# Seguridad: Si tiene allowed_types pero no generamos condiciones, no debe ver nada
|
|
query = query.filter(models.InvoiceHeader.id == -1)
|
|
|
|
# Apply sorting
|
|
if sort_by:
|
|
# Simple column mapping
|
|
# This can be improved to handle joins if needed
|
|
column = getattr(models.InvoiceHeader, sort_by, None)
|
|
if column:
|
|
if sort_order == "desc":
|
|
query = query.order_by(column.desc())
|
|
else:
|
|
query = query.order_by(column.asc())
|
|
else:
|
|
# Default sorting if none provided
|
|
query = query.order_by(models.InvoiceHeader.id.desc())
|
|
|
|
total = query.count()
|
|
items = query.offset(skip).limit(limit).all()
|
|
|
|
# Keep party_count aligned with the real number of line items.
|
|
# This avoids stale values stored in invoice_header.party_count.
|
|
if items:
|
|
invoice_ids = [inv.id for inv in items]
|
|
counts = (
|
|
db.query(LineItem.invoice_id, func.count(LineItem.id))
|
|
.filter(
|
|
LineItem.invoice_id.in_(invoice_ids),
|
|
LineItem.tenant_id == tenant_id,
|
|
LineItem.company_id == company_id,
|
|
)
|
|
.group_by(LineItem.invoice_id)
|
|
.all()
|
|
)
|
|
count_map = {invoice_id: int(count) for invoice_id, count in counts}
|
|
for inv in items:
|
|
inv.party_count = count_map.get(inv.id, 0)
|
|
|
|
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()
|
|
|
|
# Autocalculo remesa (si aplica) ANTES de validar
|
|
_autofill_remesa_if_needed(db, invoice_data, tenant_id, company_id)
|
|
|
|
# DEBUG: Log payload for analysis
|
|
print(f"DEBUG: Creating invoice {invoice_data.invoice_number} of type {invoice_data.invoice_type}")
|
|
print(f"DEBUG: Payload: {invoice_data.model_dump()}")
|
|
|
|
# Validar si la factura ya existe
|
|
invoice_exists(db, invoice_data.invoice_number, tenant_id, company_id, errors)
|
|
if invoice_data.operation_type == "exp":
|
|
validate_create_export(db, invoice_data, tenant_id, company_id, errors)
|
|
else:
|
|
validate_create_import(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
|
|
|
|
# Automatic status and audit fields
|
|
username = _get_current_username()
|
|
invoice_dict["capture_user"] = username
|
|
invoice_dict["who_processed"] = username
|
|
|
|
# Ensure document_type respects DB constraints for MEX/AME invoices (bypass clean_dict)
|
|
if invoice_dict.get("invoice_type") in ["MEX", "AME"] and not invoice_dict.get("document_type"):
|
|
invoice_dict["document_type"] = None
|
|
|
|
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
|
|
_autofill_transport_int_ids(
|
|
db,
|
|
logistics_dict,
|
|
tenant_id=tenant_id,
|
|
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)
|
|
backup_written = _ensure_create_audit_log(
|
|
db,
|
|
table_name=new_invoice.__tablename__,
|
|
record_id=str(new_invoice.id),
|
|
record_data={
|
|
c.name: getattr(new_invoice, c.name)
|
|
for c in models.InvoiceHeader.__table__.columns
|
|
},
|
|
username=username,
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
)
|
|
for detail in new_invoice.details:
|
|
backup_written = (
|
|
_ensure_create_audit_log(
|
|
db,
|
|
table_name=detail.__tablename__,
|
|
record_id=str(detail.id),
|
|
record_data={
|
|
c.name: getattr(detail, c.name)
|
|
for c in models.InvoiceSalesDetails.__table__.columns
|
|
},
|
|
username=username,
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
)
|
|
or backup_written
|
|
)
|
|
if backup_written:
|
|
db.commit()
|
|
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"""
|
|
|
|
# DEBUG: Log payload for analysis
|
|
print(f"DEBUG: Updating invoice ID {invoice_id} of type {invoice_data.invoice_type}")
|
|
print(f"DEBUG: Payload: {invoice_data.model_dump(exclude_unset=True)}")
|
|
|
|
# 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}'",
|
|
)
|
|
|
|
if invoice_data.operation_type == "exp":
|
|
validate_update_export(db, invoice_data, invoice, tenant_id, company_id, errors)
|
|
else:
|
|
# Autocalculo remesa (si aplica) ANTES de validar
|
|
_autofill_remesa_if_needed(db, invoice_data, tenant_id, company_id)
|
|
validate_update_import(db, invoice_data, invoice, tenant_id, company_id, 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={
|
|
"id",
|
|
"compliance_mx",
|
|
"financials",
|
|
"logistics",
|
|
"details",
|
|
"collections",
|
|
},
|
|
exclude_unset=True,
|
|
)
|
|
for key, value in update_dict.items():
|
|
setattr(invoice, key, value)
|
|
|
|
# Audit update fields
|
|
username = _get_current_username()
|
|
invoice.who_processed = username
|
|
invoice.updated_date = func.now()
|
|
|
|
# Backfill capture_user if missing or previous generic 'System'
|
|
if not invoice.capture_user or invoice.capture_user == "System":
|
|
if username != "System":
|
|
invoice.capture_user = username
|
|
|
|
# Update compliance_mx if provided
|
|
if invoice_data.compliance_mx is not None:
|
|
print(f"DEBUG: 更新 compliance_mx para factura {invoice.id}: {invoice_data.compliance_mx}")
|
|
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)
|
|
|
|
# Update logistics if provided
|
|
if invoice_data.logistics is not None:
|
|
if invoice.logistics:
|
|
for key, value in invoice_data.logistics.model_dump(
|
|
exclude_unset=True
|
|
).items():
|
|
if value == "":
|
|
value = None
|
|
setattr(invoice.logistics, key, value)
|
|
_autofill_transport_int_ids(
|
|
db,
|
|
invoice.logistics,
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
)
|
|
else:
|
|
logistics_dict = invoice_data.logistics.model_dump()
|
|
logistics_dict["invoice_id"] = invoice.id
|
|
logistics_dict["tenant_id"] = tenant_id
|
|
logistics_dict["company_id"] = company_id
|
|
_autofill_transport_int_ids(
|
|
db,
|
|
logistics_dict,
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
)
|
|
new_logistics = models.InvoiceLogistics(**logistics_dict)
|
|
db.add(new_logistics)
|
|
|
|
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
|