Merge pull request 'Se arreglo los filtros de facturas' (#260) from fix/filtro_facturas into development

Reviewed-on: ADUANASOFT/anexo76#260
This commit is contained in:
2026-04-02 13:20:01 +00:00
2 changed files with 50 additions and 21 deletions

View File

@@ -95,6 +95,9 @@ def list_invoices(
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", regex="^(asc|desc)$", description="Sort order (asc or desc)"),
db: Session = Depends(get_core_db),
@@ -109,12 +112,14 @@ def list_invoices(
skip = (page - 1) * page_size
filters = {
"invoice_number": search,
"invoice_number": invoice_number or search,
"status": status,
"operation_type": operation_type,
"invoice_type": invoice_type,
"manifest_number": manifest_number,
"pedimento": pedimento,
"project_number": project_number,
"year": year,
}
# Remove None values

View File

@@ -258,46 +258,70 @@ class InvoiceService:
# Apply filters if provided
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 = models.InvoiceStatus.PROCESSED if filters["status"] == True else models.InvoiceStatus.PENDING
status = (
models.InvoiceStatus.PROCESSED
if filters["status"] == True
else models.InvoiceStatus.PENDING
)
query = query.filter(models.InvoiceHeader.status == 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
)
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("pedimento"):
query = query.join(models.InvoiceComplianceMx).filter(
models.InvoiceComplianceMx.pedimento.ilike(
f"%{filters['pedimento']}%"
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.operation_type != "REPAR")
if filters.get("manifest_number"):
# Avoid duplicate joins if pedimento filter was also applied (though rare in this context)
# For safety, we can just use the relationship attribute directly if mapped,
# but explicit join is clearer given the previous pattern.
# Assuming SQLAlchemy handles the join overlap or we just accept it for now.
# To be safe and consistent with previous 'pedimento' block:
query = query.join(models.InvoiceComplianceMx).filter(
models.InvoiceComplianceMx.manifest_number.ilike(f"%{filters['manifest_number']}%")
)
query = query.filter(models.InvoiceHeader.invoice_type != "REPAR")
# Apply sorting
if sort_by: