Merge branch 'development' into feature/digitalizacion-api
This commit is contained in:
@@ -1,6 +1,7 @@
|
||||
from decimal import Decimal
|
||||
from sqlalchemy.orm import Session
|
||||
from api.v1.modules.a76.general_catalogs.unit_conversions.models import UnitConversion
|
||||
from api.v1.modules.a76.general_catalogs.equivalencies.models import EquivalencyItem
|
||||
|
||||
def _get_unit_equivalence(
|
||||
db: Session,
|
||||
@@ -13,11 +14,15 @@ def _get_unit_equivalence(
|
||||
Busca una conversión entre dos unidades de medida.
|
||||
Paridad: REVEQUIVALENCIA (Clarion SCAII).
|
||||
|
||||
Busca primero en el catálogo de Conversiones (unit_conversions) y,
|
||||
si no encuentra, en el catálogo de Equivalencias (equivalency_items).
|
||||
|
||||
Retorna (multi_divide, factor_conv):
|
||||
- ('M', factor) → multiplicar cantidad por factor
|
||||
- ('D', factor) → dividir cantidad por factor
|
||||
- ('', 0) → no existe equivalencia
|
||||
"""
|
||||
# ── 1. Catálogo de Conversiones ──────────────────────────────────────────
|
||||
conv = (
|
||||
db.query(UnitConversion)
|
||||
.filter(
|
||||
@@ -44,4 +49,33 @@ def _get_unit_equivalence(
|
||||
if conv_inv and conv_inv.conversion_factor:
|
||||
return "D", conv_inv.conversion_factor
|
||||
|
||||
# ── 2. Catálogo de Equivalencias (fallback) ──────────────────────────────
|
||||
eq = (
|
||||
db.query(EquivalencyItem)
|
||||
.filter(
|
||||
EquivalencyItem.tenant_id == tenant_id,
|
||||
EquivalencyItem.company_id == company_id,
|
||||
EquivalencyItem.original_field == from_unit,
|
||||
EquivalencyItem.external_field == to_unit,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if eq:
|
||||
factor = eq.conversion_factor if eq.conversion_factor else Decimal(1)
|
||||
return "M", factor
|
||||
|
||||
eq_inv = (
|
||||
db.query(EquivalencyItem)
|
||||
.filter(
|
||||
EquivalencyItem.tenant_id == tenant_id,
|
||||
EquivalencyItem.company_id == company_id,
|
||||
EquivalencyItem.original_field == to_unit,
|
||||
EquivalencyItem.external_field == from_unit,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if eq_inv:
|
||||
factor = eq_inv.conversion_factor if eq_inv.conversion_factor else Decimal(1)
|
||||
return "D", factor
|
||||
|
||||
return "", Decimal(0)
|
||||
|
||||
@@ -14,6 +14,9 @@ from api.v1.modules.a76.invoices.models import InvoiceHeader
|
||||
from api.v1.modules.a76.classes.models import Class
|
||||
from api.v1.modules.a76.general_catalogs.units_of_measure.models import UnitOfMeasure
|
||||
from api.v1.modules.a76.general_catalogs.packages.models import Package
|
||||
from api.v1.modules.a76.general_catalogs.fractions.us_tariff_fractions.models import (
|
||||
USTariffFraction,
|
||||
)
|
||||
from api.v1.modules.a24.fa.fa_item_lines.models import FaLineItem
|
||||
from api.v1.modules.public.reference_data.countries.models import Country
|
||||
from api.v1.modules.a76.general_catalogs.sectors.models import Sector
|
||||
@@ -360,18 +363,56 @@ def validate_common(
|
||||
)
|
||||
|
||||
if line.customs.american_fraction:
|
||||
american_fraction_exists = db.query(
|
||||
exists().where(
|
||||
LineCustom.american_fraction == line.customs.american_fraction
|
||||
def _normalize_american_fraction_code(raw_code: str) -> list[str]:
|
||||
normalized_raw = (raw_code or "").strip()
|
||||
if not normalized_raw:
|
||||
return []
|
||||
|
||||
digits_only = normalized_raw.replace(".", "").replace(" ", "").replace("-", "")
|
||||
candidates = [normalized_raw]
|
||||
|
||||
if len(digits_only) == 10:
|
||||
candidates.append(
|
||||
f"{digits_only[:4]}.{digits_only[4:6]}.{digits_only[6:8]}.{digits_only[8:10]}"
|
||||
)
|
||||
elif len(digits_only) == 8:
|
||||
candidates.append(f"{digits_only[:4]}.{digits_only[4:6]}.{digits_only[6:8]}")
|
||||
|
||||
candidates.append(digits_only)
|
||||
|
||||
seen: set[str] = set()
|
||||
deduped: list[str] = []
|
||||
for candidate in candidates:
|
||||
if not candidate or candidate in seen:
|
||||
continue
|
||||
seen.add(candidate)
|
||||
deduped.append(candidate)
|
||||
return deduped
|
||||
|
||||
candidates = _normalize_american_fraction_code(line.customs.american_fraction)
|
||||
us_fraction: USTariffFraction | None = None
|
||||
for candidate in candidates:
|
||||
us_fraction = (
|
||||
db.query(USTariffFraction)
|
||||
.filter(
|
||||
USTariffFraction.code == candidate,
|
||||
USTariffFraction.tenant_id == tenant_id,
|
||||
USTariffFraction.company_id == company_id,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
).scalar()
|
||||
if not american_fraction_exists:
|
||||
if us_fraction:
|
||||
break
|
||||
|
||||
if not us_fraction:
|
||||
errors.add_error(
|
||||
field=f"line[{line_number}].customs.american_fraction",
|
||||
message="La fracción americana especificada no existe.",
|
||||
solution=["Proporciona una fracción americana valida."],
|
||||
code="AMERICAN_FRACTION_NOT_FOUND",
|
||||
)
|
||||
else:
|
||||
line.customs.american_fraction = us_fraction.code
|
||||
|
||||
if line.order:
|
||||
if len(line.order) > 20:
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
"""Downloaded parts report module."""
|
||||
@@ -0,0 +1,43 @@
|
||||
from typing import Any
|
||||
|
||||
from fastapi import APIRouter, Body, Depends, Query
|
||||
from fastapi.responses import StreamingResponse
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from core.database import get_core_db
|
||||
from core.security import get_current_user, validate_access_to_resource
|
||||
|
||||
from .schemas import DownloadedPartsReportBootstrap, DownloadedPartsReportRequest
|
||||
from .service import DownloadedPartsReportService
|
||||
|
||||
router = APIRouter(tags=["Reports - Downloaded Parts"])
|
||||
|
||||
|
||||
@router.get(
|
||||
"/bootstrap",
|
||||
summary="Get downloaded parts report bootstrap",
|
||||
description="Returns the base metadata required to render the downloaded parts report screen.",
|
||||
)
|
||||
def get_downloaded_parts_report_bootstrap(
|
||||
company_id: int = Query(..., description="Company ID"),
|
||||
db: Session = Depends(get_core_db),
|
||||
current_user: Any = Depends(get_current_user),
|
||||
):
|
||||
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
||||
service = DownloadedPartsReportService()
|
||||
return service.build_bootstrap(company_id=company_id, tenant_id=tenant_id)
|
||||
|
||||
|
||||
@router.post(
|
||||
"/generate",
|
||||
summary="Generate downloaded parts report CSV",
|
||||
)
|
||||
def generate_downloaded_parts_report(
|
||||
company_id: int = Query(..., description="Company ID"),
|
||||
request: DownloadedPartsReportRequest = Body(...),
|
||||
db: Session = Depends(get_core_db),
|
||||
current_user: Any = Depends(get_current_user),
|
||||
) -> StreamingResponse:
|
||||
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
||||
service = DownloadedPartsReportService()
|
||||
return service.generate_csv(db=db, req=request, company_id=company_id, tenant_id=tenant_id)
|
||||
@@ -0,0 +1,56 @@
|
||||
from datetime import date
|
||||
from typing import Literal, Optional
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
|
||||
class DownloadedPartsReportRequest(BaseModel):
|
||||
date_from: date
|
||||
date_to: date
|
||||
class_from: Optional[str] = None
|
||||
class_to: Optional[str] = None
|
||||
print_class_mode: Literal['exported', 'downloaded'] = 'downloaded'
|
||||
exchange_rate_mode: Literal['invoice', 'pedimento_payment'] = 'invoice'
|
||||
currency_mode: Literal['dollars', 'pesos', 'both'] = 'both'
|
||||
temporality_mode: Literal['temporales', 'definitivos', 'ambos'] = 'temporales'
|
||||
weight_type_mode: Literal['kilos', 'libras', 'ambos'] = 'kilos'
|
||||
operation_mode: Literal['importacion', 'exportacion'] = 'importacion'
|
||||
# Optional filters
|
||||
material_type: Optional[str] = None
|
||||
invoice_type: Optional[str] = None
|
||||
parts: Optional[list[str]] = None
|
||||
pedimento_key: Optional[str] = None
|
||||
provider_id: Optional[int] = None
|
||||
sold_to_id: Optional[int] = None
|
||||
shipped_to_id: Optional[int] = None
|
||||
destination_customs: Optional[str] = None
|
||||
# Option flags
|
||||
include_series: bool = False
|
||||
print_class_total: bool = False
|
||||
include_totals_by_fraction: bool = False
|
||||
julian_date: bool = False
|
||||
show_item_description: bool = True
|
||||
include_exempt_fraction: bool = False
|
||||
show_export_fraction: bool = False
|
||||
include_rule_octava: bool = False
|
||||
include_american_fraction_and_country: bool = False
|
||||
respect_import_invoice_value_in_pesos: bool = False
|
||||
show_all_temporary_balances: bool = False
|
||||
|
||||
|
||||
class DownloadedPartsReportSection(BaseModel):
|
||||
id: str
|
||||
title: str
|
||||
description: str
|
||||
|
||||
|
||||
class DownloadedPartsReportBootstrap(BaseModel):
|
||||
report_key: str = Field(default="downloaded_parts")
|
||||
title: str = Field(default="Partes descargadas")
|
||||
description: str
|
||||
company_id: int
|
||||
tenant_id: int
|
||||
status: str = Field(default="draft")
|
||||
available_filters: list[str]
|
||||
next_steps: list[str]
|
||||
sections: list[DownloadedPartsReportSection]
|
||||
@@ -0,0 +1,959 @@
|
||||
import csv
|
||||
import io
|
||||
from datetime import date, datetime, timedelta
|
||||
from decimal import Decimal
|
||||
from typing import Optional
|
||||
|
||||
from fastapi import HTTPException
|
||||
from fastapi.responses import StreamingResponse
|
||||
from sqlalchemy import and_, func, or_, select
|
||||
from sqlalchemy.orm import Session, aliased, selectinload
|
||||
|
||||
from api.v1.modules.a24.discharges.models import DischargeDetail, DischargeHeader, DischargeType
|
||||
from api.v1.modules.a24.fa.fa_item_lines.models import FaLineItem
|
||||
from api.v1.modules.a76.app_settings.service import AppSettingsService
|
||||
from api.v1.modules.a76.classes.models import Class
|
||||
from api.v1.modules.a76.general_catalogs.company.models import Company
|
||||
from api.v1.modules.a76.general_catalogs.exchange_rate.models import ExchangeRate
|
||||
from api.v1.modules.a76.general_catalogs.fractions.tariff_fractions.models import TariffFraction
|
||||
from api.v1.modules.a76.invoices.models import InvoiceComplianceMx, InvoiceHeader, InvoiceStatus
|
||||
from api.v1.modules.a76.items.line_customs.models import LineCustom
|
||||
from api.v1.modules.a76.items.line_descriptions.models import LineDescription
|
||||
from api.v1.modules.a76.items.line_financials.models import LineFinancial
|
||||
from api.v1.modules.a76.items.line_quantities.models import LineQuantity
|
||||
from api.v1.modules.a76.items.models import LineItem
|
||||
from api.v1.modules.a76.items.series.models import Serie
|
||||
from api.v1.modules.a76.parts.models import Part
|
||||
from api.v1.modules.a76.pedmientos.models.pedimento_dates import PedimentoDates
|
||||
from api.v1.modules.a76.pedmientos.models.pedimentos import Pedimentos
|
||||
|
||||
from .schemas import DownloadedPartsReportBootstrap, DownloadedPartsReportRequest, DownloadedPartsReportSection
|
||||
|
||||
LBS_PER_KG = Decimal('2.20462')
|
||||
|
||||
|
||||
class DownloadedPartsReportService:
|
||||
# ------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _build_pedimento_str(
|
||||
self,
|
||||
year: Optional[str],
|
||||
customs: Optional[str],
|
||||
license_: Optional[str],
|
||||
number: Optional[str],
|
||||
) -> str:
|
||||
if not all([year, customs, license_, number]):
|
||||
return ''
|
||||
return f"{year}/{customs}/{license_}/{number}"
|
||||
|
||||
def _build_pedimento_18(
|
||||
self,
|
||||
year: Optional[str],
|
||||
customs: Optional[str],
|
||||
license_: Optional[str],
|
||||
number: Optional[str],
|
||||
code: Optional[str],
|
||||
) -> str:
|
||||
if not all([year, customs, license_, number, code]):
|
||||
return ''
|
||||
return f"{year}{customs}{license_}{number}{code}"
|
||||
|
||||
def _format_date(self, d: Optional[date | datetime], julian: bool = False) -> str:
|
||||
if d is None:
|
||||
return ''
|
||||
if isinstance(d, datetime):
|
||||
d = d.date()
|
||||
if julian:
|
||||
return str(d.timetuple().tm_yday).zfill(3)
|
||||
return d.strftime('%d/%m/%Y')
|
||||
|
||||
def _decimal_str(self, val, decimals: int = 2) -> str:
|
||||
if val is None:
|
||||
return ''
|
||||
return f"{Decimal(str(val)):.{decimals}f}"
|
||||
|
||||
def _to_decimal(self, value) -> Decimal:
|
||||
if value is None:
|
||||
return Decimal('0')
|
||||
if isinstance(value, Decimal):
|
||||
return value
|
||||
return Decimal(str(value))
|
||||
|
||||
def _clean_text(self, value: Optional[str]) -> str:
|
||||
if not value:
|
||||
return ''
|
||||
return value.replace(',', ' ').replace('\r', ' ').replace('\n', ' ').strip()
|
||||
|
||||
def _excel_text(self, value) -> str:
|
||||
if value is None or value == '':
|
||||
return ''
|
||||
return f"'{value}"
|
||||
|
||||
def _as_date(self, value: Optional[date | datetime]) -> Optional[date]:
|
||||
if value is None:
|
||||
return None
|
||||
if isinstance(value, datetime):
|
||||
return value.date()
|
||||
return value
|
||||
|
||||
def _adjust_payment_date(
|
||||
self,
|
||||
value: Optional[date | datetime],
|
||||
use_previous_day: bool,
|
||||
) -> Optional[date]:
|
||||
resolved = self._as_date(value)
|
||||
if resolved is None:
|
||||
return None
|
||||
if use_previous_day:
|
||||
return resolved - timedelta(days=1)
|
||||
return resolved
|
||||
|
||||
def _pedimento_headers(self, company: Optional[Company], ped_type: str) -> tuple[str, str]:
|
||||
if company and company.rfc == 'MWE220512359':
|
||||
if ped_type == 'export':
|
||||
return ('PEDIMENTO DE EXPORTACIÓN', 'PEDIMENTO DE EXPORTACIÓN ORIGINAL RECTIFICADO')
|
||||
return ('PEDIMENTO DE IMPORTACIÓN', 'PEDIMENTO DE IMPORTACIÓN ORIGINAL RECTIFICADO')
|
||||
|
||||
if ped_type == 'export':
|
||||
return ('PEDIMENTO EXPORTACIÓN', 'PED. EXPO R1')
|
||||
return ('PEDIMENTO IMPORTACIÓN', 'PED. IMPO R1')
|
||||
|
||||
def _pedimento_values(
|
||||
self,
|
||||
company: Optional[Company],
|
||||
pedimento: str,
|
||||
pedimento_r1: str,
|
||||
) -> tuple[str, str]:
|
||||
if company and company.rfc == 'MWE220512359':
|
||||
if pedimento_r1:
|
||||
return pedimento_r1, pedimento
|
||||
return pedimento, ''
|
||||
return pedimento, pedimento_r1
|
||||
|
||||
def _build_headers(self, req: DownloadedPartsReportRequest, company: Optional[Company]) -> list[str]:
|
||||
export_headers = self._pedimento_headers(company, 'export')
|
||||
import_headers = self._pedimento_headers(company, 'import')
|
||||
|
||||
headers = [
|
||||
export_headers[0],
|
||||
export_headers[1],
|
||||
'FECHA PAGO PED EXPO',
|
||||
'CLAVE',
|
||||
'FECHA DE DESCARGA',
|
||||
'FACTURA EXPO',
|
||||
'FECHA EMISION',
|
||||
'CLASE',
|
||||
'DESCRIPCION',
|
||||
'FRACCION ARANCELARIA',
|
||||
'CANTIDAD',
|
||||
'U.M.',
|
||||
import_headers[0],
|
||||
import_headers[1],
|
||||
'FECHA PAGO PED IMPO',
|
||||
'FACTURA IMPORTACION',
|
||||
'FECHA EMISION IMPO',
|
||||
'PESO NETO',
|
||||
'VALOR TOTAL (DOLARES)',
|
||||
'VALOR TOTAL (MONEDA NACIONAL)',
|
||||
'TIPO DE CAMBIO',
|
||||
'NO. DE PARTE',
|
||||
'DESCRIPCION PARTE',
|
||||
'TIPOEXPO',
|
||||
'FRACCION CLASE',
|
||||
'PEDIMENTO IMPO 18',
|
||||
'PEDIMENTO EXPO 18',
|
||||
'U.M. TARIFA',
|
||||
]
|
||||
|
||||
if req.include_american_fraction_and_country:
|
||||
headers.extend(['FRACCION AMERICANA', 'PAIS DE ORIGEN'])
|
||||
|
||||
return headers
|
||||
|
||||
def _write_company_header(
|
||||
self,
|
||||
writer: csv.writer,
|
||||
company: Optional[Company],
|
||||
settings: dict,
|
||||
) -> None:
|
||||
writer.writerow(['REPORTE DE CLASES EXPORTADAS/DESCARGADAS'])
|
||||
|
||||
if not company:
|
||||
writer.writerow([
|
||||
f"Fecha Generación: {datetime.now().strftime('%d/%m/%Y')} Hora Generación: {datetime.now().strftime('%H:%M:%S')}"
|
||||
])
|
||||
writer.writerow(['PROVEEDOR DE SOFTWARE: ADUANASOFT'])
|
||||
writer.writerow([])
|
||||
return
|
||||
|
||||
if company.name:
|
||||
writer.writerow([company.name])
|
||||
|
||||
main_address = next((addr for addr in company.addresses if addr.address_type == 'main'), None)
|
||||
if main_address:
|
||||
fiscal_line = 'Domicilio Fiscal: ' + (main_address.street or '')
|
||||
if main_address.exterior_number:
|
||||
fiscal_line += f" Ext. Num: {main_address.exterior_number}"
|
||||
if main_address.interior_number:
|
||||
fiscal_line += f" Int. Num: {main_address.interior_number}"
|
||||
writer.writerow([fiscal_line.strip()])
|
||||
|
||||
colony_line = (main_address.neighborhood or '').strip()
|
||||
if main_address.postal_code:
|
||||
colony_line = (colony_line + f" Código Postal: {main_address.postal_code}").strip()
|
||||
if colony_line:
|
||||
writer.writerow([colony_line])
|
||||
|
||||
city_line = ' '.join(filter(None, [main_address.city, main_address.state]))
|
||||
if city_line:
|
||||
writer.writerow([city_line])
|
||||
|
||||
industrial_address = next(
|
||||
(addr for addr in company.addresses if addr.address_type == 'industrial'),
|
||||
None,
|
||||
)
|
||||
if industrial_address:
|
||||
industrial_line = 'Domicilio Industrial: ' + (industrial_address.street or '')
|
||||
if industrial_address.exterior_number:
|
||||
industrial_line += f" Ext. Num: {industrial_address.exterior_number}"
|
||||
if industrial_address.interior_number:
|
||||
industrial_line += f" Int. Num: {industrial_address.interior_number}"
|
||||
writer.writerow([industrial_line.strip()])
|
||||
|
||||
industrial_colony = (industrial_address.neighborhood or '').strip()
|
||||
if industrial_address.postal_code:
|
||||
industrial_colony = (
|
||||
industrial_colony + f" Código Postal: {industrial_address.postal_code}"
|
||||
).strip()
|
||||
if industrial_colony:
|
||||
writer.writerow([industrial_colony])
|
||||
|
||||
industrial_city = ' '.join(filter(None, [industrial_address.city, industrial_address.state]))
|
||||
if industrial_city:
|
||||
writer.writerow([industrial_city])
|
||||
|
||||
if company.rfc:
|
||||
writer.writerow([f"R.F.C: {company.rfc}"])
|
||||
|
||||
if settings.get('mostrarprogramaimmexprosec'):
|
||||
if company.program_number:
|
||||
if company.program == 'Maquila':
|
||||
writer.writerow([f"SICEX: {company.program_number}"])
|
||||
else:
|
||||
writer.writerow([f"{company.program or 'Programa'}: {company.program_number}"])
|
||||
if company.prosec_authorization:
|
||||
writer.writerow([f"Autorización PROSEC: {company.prosec_authorization}"])
|
||||
|
||||
writer.writerow([
|
||||
f"Fecha Generación: {datetime.now().strftime('%d/%m/%Y')} Hora Generación: {datetime.now().strftime('%H:%M:%S')}"
|
||||
])
|
||||
writer.writerow(['PROVEEDOR DE SOFTWARE: ADUANASOFT'])
|
||||
writer.writerow([])
|
||||
|
||||
def _get_settings(self, db: Session, tenant_id: int, company_id: int) -> dict:
|
||||
return AppSettingsService.get_resolved_settings(db, tenant_id, company_id) or {}
|
||||
|
||||
def _collect_rate_dates(
|
||||
self,
|
||||
rows: list,
|
||||
req: DownloadedPartsReportRequest,
|
||||
use_previous_payment_day: bool,
|
||||
) -> set[date]:
|
||||
dates: set[date] = set()
|
||||
|
||||
for row in rows:
|
||||
if req.exchange_rate_mode == 'invoice':
|
||||
export_invoice_date = self._as_date(row.expo_invoice_date)
|
||||
import_invoice_date = self._as_date(row.impo_invoice_date)
|
||||
if export_invoice_date:
|
||||
dates.add(export_invoice_date)
|
||||
if req.operation_mode == 'importacion' and import_invoice_date:
|
||||
dates.add(import_invoice_date)
|
||||
continue
|
||||
|
||||
target_date = row.impo_payment_date if req.operation_mode == 'importacion' else row.expo_payment_date
|
||||
adjusted = self._adjust_payment_date(target_date, use_previous_payment_day)
|
||||
if adjusted:
|
||||
dates.add(adjusted)
|
||||
|
||||
return dates
|
||||
|
||||
def _load_exchange_rates(
|
||||
self,
|
||||
db: Session,
|
||||
dates: set[date],
|
||||
company_id: int,
|
||||
tenant_id: int,
|
||||
) -> dict[date, Decimal]:
|
||||
if not dates:
|
||||
return {}
|
||||
|
||||
rows = db.execute(
|
||||
select(func.date(ExchangeRate.date), ExchangeRate.value).where(
|
||||
ExchangeRate.tenant_id == tenant_id,
|
||||
ExchangeRate.company_id == company_id,
|
||||
func.date(ExchangeRate.date).in_(sorted(dates)),
|
||||
)
|
||||
).fetchall()
|
||||
|
||||
return {
|
||||
self._as_date(rate_date): self._to_decimal(rate_value)
|
||||
for rate_date, rate_value in rows
|
||||
if rate_date is not None
|
||||
}
|
||||
|
||||
def _resolve_selected_rate(
|
||||
self,
|
||||
row,
|
||||
req: DownloadedPartsReportRequest,
|
||||
rate_lookup: dict[date, Decimal],
|
||||
use_previous_payment_day: bool,
|
||||
) -> Optional[Decimal]:
|
||||
export_invoice_date = self._as_date(row.expo_invoice_date)
|
||||
import_invoice_date = self._as_date(row.impo_invoice_date)
|
||||
export_payment_date = self._adjust_payment_date(row.expo_payment_date, use_previous_payment_day)
|
||||
import_payment_date = self._adjust_payment_date(row.impo_payment_date, use_previous_payment_day)
|
||||
|
||||
if req.exchange_rate_mode == 'invoice':
|
||||
if req.operation_mode == 'importacion':
|
||||
return rate_lookup.get(import_invoice_date) or rate_lookup.get(export_invoice_date)
|
||||
return rate_lookup.get(export_invoice_date)
|
||||
|
||||
if req.operation_mode == 'importacion':
|
||||
return rate_lookup.get(import_payment_date)
|
||||
return rate_lookup.get(export_payment_date)
|
||||
|
||||
def _find_missing_rate_dates(
|
||||
self,
|
||||
rows: list,
|
||||
req: DownloadedPartsReportRequest,
|
||||
rate_lookup: dict[date, Decimal],
|
||||
use_previous_payment_day: bool,
|
||||
) -> list[str]:
|
||||
missing: set[date] = set()
|
||||
|
||||
for row in rows:
|
||||
if req.exchange_rate_mode == 'invoice':
|
||||
export_invoice_date = self._as_date(row.expo_invoice_date)
|
||||
import_invoice_date = self._as_date(row.impo_invoice_date)
|
||||
if export_invoice_date and export_invoice_date not in rate_lookup:
|
||||
missing.add(export_invoice_date)
|
||||
if req.operation_mode == 'importacion' and import_invoice_date and import_invoice_date not in rate_lookup:
|
||||
missing.add(import_invoice_date)
|
||||
continue
|
||||
|
||||
target_date = row.impo_payment_date if req.operation_mode == 'importacion' else row.expo_payment_date
|
||||
adjusted = self._adjust_payment_date(target_date, use_previous_payment_day)
|
||||
if adjusted and adjusted not in rate_lookup:
|
||||
missing.add(adjusted)
|
||||
|
||||
return [d.strftime('%d/%m/%Y') for d in sorted(missing)]
|
||||
|
||||
def _resolve_export_fraction(self, row, req: DownloadedPartsReportRequest) -> str:
|
||||
if req.include_rule_octava and row.export_octave_fraction:
|
||||
return row.export_octave_fraction
|
||||
return row.export_fraction or ''
|
||||
|
||||
def _resolve_class_fraction(self, row, req: DownloadedPartsReportRequest) -> str:
|
||||
export_fraction = self._resolve_export_fraction(row, req)
|
||||
import_fraction = row.import_fraction or ''
|
||||
if req.include_rule_octava and row.import_octave_fraction:
|
||||
import_fraction = row.import_octave_fraction
|
||||
|
||||
if req.show_export_fraction:
|
||||
return row.class_fraction or ''
|
||||
|
||||
if req.print_class_mode == 'downloaded':
|
||||
return import_fraction or row.class_fraction or ''
|
||||
|
||||
return export_fraction or row.class_fraction or ''
|
||||
|
||||
def _resolve_description(self, row, req: DownloadedPartsReportRequest) -> str:
|
||||
line_description = row.export_line_description or row.export_line_part_description
|
||||
class_description = row.export_line_class_description or row.class_description
|
||||
part_description = row.part_description or row.export_line_part_description
|
||||
|
||||
if req.show_item_description:
|
||||
return self._clean_text(line_description or part_description or class_description)
|
||||
|
||||
if part_description and class_description:
|
||||
return self._clean_text(f"{part_description} / {class_description}")
|
||||
|
||||
return self._clean_text(part_description or class_description or line_description)
|
||||
|
||||
def _resolve_part_description(self, row) -> str:
|
||||
return self._clean_text(row.export_line_part_description or row.part_description)
|
||||
|
||||
def _resolve_base_values(self, row) -> tuple[Decimal, Decimal]:
|
||||
qty = self._to_decimal(row.quantity)
|
||||
detail_mn = self._to_decimal(row.value_mn)
|
||||
detail_usd = self._to_decimal(row.value_me)
|
||||
import_qty = self._to_decimal(row.import_quantity_total)
|
||||
|
||||
if row.import_is_subitem:
|
||||
return detail_mn, detail_usd
|
||||
|
||||
if row.import_unit_cost_mxn is not None or row.import_unit_cost_usd is not None:
|
||||
return (
|
||||
qty * self._to_decimal(row.import_unit_cost_mxn),
|
||||
qty * self._to_decimal(row.import_unit_cost_usd),
|
||||
)
|
||||
|
||||
if import_qty > 0:
|
||||
ratio = qty / import_qty
|
||||
customs_mxn = self._to_decimal(row.import_customs_value_mxn)
|
||||
customs_usd = self._to_decimal(row.import_customs_value_usd)
|
||||
if customs_mxn > 0 or customs_usd > 0:
|
||||
return customs_mxn * ratio, customs_usd * ratio
|
||||
|
||||
return detail_mn, detail_usd
|
||||
|
||||
def _resolve_values(
|
||||
self,
|
||||
row,
|
||||
req: DownloadedPartsReportRequest,
|
||||
rate_lookup: dict[date, Decimal],
|
||||
use_previous_payment_day: bool,
|
||||
) -> tuple[Decimal, Decimal, Optional[Decimal]]:
|
||||
detail_mn = self._to_decimal(row.value_mn)
|
||||
detail_usd = self._to_decimal(row.value_me)
|
||||
base_mn, base_usd = self._resolve_base_values(row)
|
||||
selected_rate = self._resolve_selected_rate(row, req, rate_lookup, use_previous_payment_day)
|
||||
|
||||
if req.print_class_mode == 'downloaded':
|
||||
source_mn = base_mn if base_mn > 0 else detail_mn
|
||||
source_usd = base_usd if base_usd > 0 else detail_usd
|
||||
else:
|
||||
source_mn = detail_mn if detail_mn > 0 else base_mn
|
||||
source_usd = detail_usd if detail_usd > 0 else base_usd
|
||||
|
||||
if req.operation_mode == 'importacion' and req.respect_import_invoice_value_in_pesos:
|
||||
value_mn = source_mn
|
||||
value_usd = source_mn / selected_rate if selected_rate and source_mn > 0 else source_usd
|
||||
return value_mn, value_usd, selected_rate
|
||||
|
||||
if source_usd > 0 and selected_rate:
|
||||
return source_usd * selected_rate, source_usd, selected_rate
|
||||
|
||||
if source_mn > 0 and selected_rate:
|
||||
return source_mn, source_mn / selected_rate, selected_rate
|
||||
|
||||
return source_mn, source_usd, selected_rate
|
||||
|
||||
def _build_series_map(self, db: Session, line_ids: list[int], tenant_id: int) -> dict[int, list[Serie]]:
|
||||
if not line_ids:
|
||||
return {}
|
||||
|
||||
series_rows = (
|
||||
db.query(Serie)
|
||||
.filter(
|
||||
Serie.tenant_id == tenant_id,
|
||||
Serie.line_item_id.in_(line_ids),
|
||||
)
|
||||
.order_by(Serie.line_item_id, Serie.row, Serie.id)
|
||||
.all()
|
||||
)
|
||||
|
||||
series_map: dict[int, list[Serie]] = {}
|
||||
for series in series_rows:
|
||||
series_map.setdefault(series.line_item_id, []).append(series)
|
||||
return series_map
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Bootstrap
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def build_bootstrap(self, company_id: int, tenant_id: int) -> DownloadedPartsReportBootstrap:
|
||||
return DownloadedPartsReportBootstrap(
|
||||
description=(
|
||||
"Base inicial para construir el reporte de partes descargadas desde exportacion. "
|
||||
"Incluye metadatos, filtros sugeridos y bloques base para la vista."
|
||||
),
|
||||
company_id=company_id,
|
||||
tenant_id=tenant_id,
|
||||
available_filters=[
|
||||
"fecha_inicio",
|
||||
"fecha_fin",
|
||||
"parte",
|
||||
"pedimento",
|
||||
"factura_exportacion",
|
||||
"cliente",
|
||||
],
|
||||
next_steps=[
|
||||
"Definir origen exacto de datos para descargas por parte.",
|
||||
"Agregar filtros funcionales y tabla de resultados.",
|
||||
"Conectar exportacion a Excel o CSV cuando el layout quede definido.",
|
||||
],
|
||||
sections=[
|
||||
DownloadedPartsReportSection(
|
||||
id="filters",
|
||||
title="Filtros",
|
||||
description="Contenedor para criterios de busqueda del reporte.",
|
||||
),
|
||||
DownloadedPartsReportSection(
|
||||
id="results",
|
||||
title="Resultados",
|
||||
description="Espacio reservado para tabla o listado de partes descargadas.",
|
||||
),
|
||||
DownloadedPartsReportSection(
|
||||
id="exports",
|
||||
title="Exportacion",
|
||||
description="Zona para acciones futuras de descarga y generacion de archivos.",
|
||||
),
|
||||
],
|
||||
)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Exchange rate validation
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def validate_exchange_rates(
|
||||
self,
|
||||
db: Session,
|
||||
req: DownloadedPartsReportRequest,
|
||||
company_id: int,
|
||||
tenant_id: int,
|
||||
) -> list[str]:
|
||||
settings = self._get_settings(db, tenant_id, company_id)
|
||||
use_previous_payment_day = bool(settings.get('utilizarfechapagopeddeundiaanterior'))
|
||||
rows = self.query_discharge_data(db, req, company_id, tenant_id)
|
||||
rate_dates = self._collect_rate_dates(rows, req, use_previous_payment_day)
|
||||
rate_lookup = self._load_exchange_rates(db, rate_dates, company_id, tenant_id)
|
||||
return self._find_missing_rate_dates(rows, req, rate_lookup, use_previous_payment_day)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Main data query
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def query_discharge_data(
|
||||
self,
|
||||
db: Session,
|
||||
req: DownloadedPartsReportRequest,
|
||||
company_id: int,
|
||||
tenant_id: int,
|
||||
) -> list:
|
||||
ExportLine = aliased(LineItem, name='export_line')
|
||||
ImportLine = aliased(LineItem, name='import_line')
|
||||
ExportInvoice = aliased(InvoiceHeader, name='export_invoice')
|
||||
ImportInvoice = aliased(InvoiceHeader, name='import_invoice')
|
||||
ExportCompliance = aliased(InvoiceComplianceMx, name='export_compliance')
|
||||
ImportCompliance = aliased(InvoiceComplianceMx, name='import_compliance')
|
||||
ExportPedimento = aliased(Pedimentos, name='export_pedimento')
|
||||
ImportPedimento = aliased(Pedimentos, name='import_pedimento')
|
||||
ExportPedR1 = aliased(Pedimentos, name='export_ped_r1')
|
||||
ImportPedR1 = aliased(Pedimentos, name='import_ped_r1')
|
||||
ExportPedDates = aliased(PedimentoDates, name='export_ped_dates')
|
||||
ImportPedDates = aliased(PedimentoDates, name='import_ped_dates')
|
||||
ExportPart = aliased(Part, name='export_part')
|
||||
ExportCustom = aliased(LineCustom, name='export_custom')
|
||||
ImportCustom = aliased(LineCustom, name='import_custom')
|
||||
ExportDescription = aliased(LineDescription, name='export_description')
|
||||
ImportFinancial = aliased(LineFinancial, name='import_financial')
|
||||
ImportQuantity = aliased(LineQuantity, name='import_quantity')
|
||||
ImportFa = aliased(FaLineItem, name='import_fa')
|
||||
ExportTariffFraction = aliased(TariffFraction, name='export_tariff_fraction')
|
||||
|
||||
stmt = (
|
||||
select(
|
||||
ExportLine.id.label('export_line_id'),
|
||||
ExportPedimento.year.label('expo_ped_year'),
|
||||
ExportPedimento.customs_office.label('expo_ped_customs'),
|
||||
ExportPedimento.license.label('expo_ped_license'),
|
||||
ExportPedimento.pedimento_number.label('expo_ped_number'),
|
||||
ExportPedimento.pedimento_code.label('expo_ped_code'),
|
||||
ExportPedR1.year.label('expo_r1_year'),
|
||||
ExportPedR1.customs_office.label('expo_r1_customs'),
|
||||
ExportPedR1.license.label('expo_r1_license'),
|
||||
ExportPedR1.pedimento_number.label('expo_r1_number'),
|
||||
ExportPedDates.payment_date.label('expo_payment_date'),
|
||||
ExportPedimento.pedimento_code.label('expo_clave'),
|
||||
DischargeHeader.discharge_date.label('discharge_date'),
|
||||
ExportInvoice.invoice_number.label('expo_invoice_number'),
|
||||
ExportInvoice.invoice_date.label('expo_invoice_date'),
|
||||
Class.class_code.label('class_code'),
|
||||
Class.description_es.label('class_description'),
|
||||
Class.material_key.label('material_key'),
|
||||
Class.fraction.label('class_fraction'),
|
||||
ExportCustom.fraction.label('export_fraction'),
|
||||
ExportCustom.american_fraction.label('american_fraction'),
|
||||
ExportCustom.octave_fraction.label('export_octave_fraction'),
|
||||
ImportCustom.fraction.label('import_fraction'),
|
||||
ImportCustom.octave_fraction.label('import_octave_fraction'),
|
||||
DischargeDetail.quantity_discharged.label('quantity'),
|
||||
DischargeDetail.unit_of_measure.label('unit_of_measure'),
|
||||
DischargeDetail.value_me.label('value_me'),
|
||||
DischargeDetail.value_mn.label('value_mn'),
|
||||
DischargeDetail.net_weight.label('net_weight'),
|
||||
DischargeDetail.origin_import_invoice.label('import_invoice_str'),
|
||||
DischargeDetail.part_number.label('part_number_str'),
|
||||
DischargeDetail.country_of_origin.label('country_of_origin'),
|
||||
ImportPedimento.year.label('impo_ped_year'),
|
||||
ImportPedimento.customs_office.label('impo_ped_customs'),
|
||||
ImportPedimento.license.label('impo_ped_license'),
|
||||
ImportPedimento.pedimento_number.label('impo_ped_number'),
|
||||
ImportPedimento.pedimento_code.label('impo_ped_code'),
|
||||
ImportPedR1.year.label('impo_r1_year'),
|
||||
ImportPedR1.customs_office.label('impo_r1_customs'),
|
||||
ImportPedR1.license.label('impo_r1_license'),
|
||||
ImportPedR1.pedimento_number.label('impo_r1_number'),
|
||||
ImportPedDates.payment_date.label('impo_payment_date'),
|
||||
ImportInvoice.invoice_number.label('import_invoice_number'),
|
||||
ImportInvoice.invoice_date.label('impo_invoice_date'),
|
||||
ImportInvoice.invoice_type.label('import_invoice_type'),
|
||||
ExportPart.description_spanish.label('part_description'),
|
||||
ExportDescription.description_spanish.label('export_line_description'),
|
||||
ExportDescription.part_description.label('export_line_part_description'),
|
||||
ExportDescription.class_description.label('export_line_class_description'),
|
||||
ExportDescription.brand.label('export_brand'),
|
||||
ExportDescription.model.label('export_model'),
|
||||
ImportFinancial.unit_cost_mxn.label('import_unit_cost_mxn'),
|
||||
ImportFinancial.unit_cost_usd.label('import_unit_cost_usd'),
|
||||
ImportFinancial.customs_value_mxn.label('import_customs_value_mxn'),
|
||||
ImportFinancial.customs_value_usd.label('import_customs_value_usd'),
|
||||
ImportQuantity.quantity.label('import_quantity_total'),
|
||||
ImportLine.payment_method.label('import_payment_method'),
|
||||
ImportFa.is_subitem.label('import_is_subitem'),
|
||||
ExportTariffFraction.umt.label('tariff_uom'),
|
||||
)
|
||||
.select_from(DischargeDetail)
|
||||
.join(DischargeHeader, DischargeDetail.discharge_header_id == DischargeHeader.id)
|
||||
.join(ExportLine, DischargeDetail.export_item_line_id == ExportLine.id)
|
||||
.join(ImportLine, DischargeDetail.import_item_line_id == ImportLine.id)
|
||||
.join(ExportInvoice, ExportLine.invoice_id == ExportInvoice.id)
|
||||
.join(ImportInvoice, ImportLine.invoice_id == ImportInvoice.id)
|
||||
.outerjoin(ExportCompliance, ExportCompliance.invoice_id == ExportInvoice.id)
|
||||
.outerjoin(ImportCompliance, ImportCompliance.invoice_id == ImportInvoice.id)
|
||||
.outerjoin(ExportPedimento, ExportPedimento.id == ExportCompliance.pedimento_id)
|
||||
.outerjoin(ImportPedimento, ImportPedimento.id == ImportCompliance.pedimento_id)
|
||||
.outerjoin(ExportPedR1, ExportPedR1.id == ExportCompliance.pedimento_r1)
|
||||
.outerjoin(ImportPedR1, ImportPedR1.id == ImportCompliance.pedimento_r1)
|
||||
.outerjoin(ExportPedDates, ExportPedDates.pedimento_id == ExportPedimento.id)
|
||||
.outerjoin(ImportPedDates, ImportPedDates.pedimento_id == ImportPedimento.id)
|
||||
.outerjoin(Class, Class.id == ExportLine.class_id)
|
||||
.outerjoin(ExportPart, ExportPart.id == ExportLine.part_number_id)
|
||||
.outerjoin(ExportCustom, ExportCustom.item_line_id == ExportLine.id)
|
||||
.outerjoin(ImportCustom, ImportCustom.item_line_id == ImportLine.id)
|
||||
.outerjoin(ExportDescription, ExportDescription.item_line_id == ExportLine.id)
|
||||
.outerjoin(ImportFinancial, ImportFinancial.item_line_id == ImportLine.id)
|
||||
.outerjoin(ImportQuantity, ImportQuantity.item_line_id == ImportLine.id)
|
||||
.outerjoin(ImportFa, ImportFa.id == ImportLine.id)
|
||||
.outerjoin(
|
||||
ExportTariffFraction,
|
||||
ExportTariffFraction.code == func.substr(
|
||||
func.replace(func.coalesce(ExportCustom.fraction, ''), '.', ''),
|
||||
1,
|
||||
8,
|
||||
),
|
||||
)
|
||||
.where(
|
||||
DischargeHeader.tenant_id == tenant_id,
|
||||
DischargeHeader.company_id == company_id,
|
||||
ExportInvoice.status == InvoiceStatus.PROCESSED,
|
||||
)
|
||||
)
|
||||
|
||||
# Date filter
|
||||
if req.print_class_mode == 'exported':
|
||||
stmt = stmt.where(
|
||||
ExportInvoice.invoice_date.between(req.date_from, req.date_to)
|
||||
)
|
||||
else:
|
||||
stmt = stmt.where(
|
||||
ExportInvoice.invoice_type != 'NODES',
|
||||
or_(
|
||||
ExportPedDates.payment_date.between(req.date_from, req.date_to),
|
||||
and_(
|
||||
ExportPedDates.payment_date.is_(None),
|
||||
ExportInvoice.invoice_date.between(req.date_from, req.date_to),
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
# Temporality
|
||||
if req.temporality_mode == 'temporales':
|
||||
stmt = stmt.where(DischargeHeader.discharge_type == DischargeType.TEMPORARY)
|
||||
elif req.temporality_mode == 'definitivos':
|
||||
stmt = stmt.where(DischargeHeader.discharge_type == DischargeType.DEFINITIVE)
|
||||
|
||||
# Class range
|
||||
if req.class_from:
|
||||
stmt = stmt.where(Class.class_code >= req.class_from)
|
||||
if req.class_to:
|
||||
stmt = stmt.where(Class.class_code <= req.class_to)
|
||||
|
||||
if req.material_type:
|
||||
stmt = stmt.where(Class.material_key == req.material_type)
|
||||
if req.invoice_type:
|
||||
stmt = stmt.where(ExportInvoice.invoice_type == req.invoice_type)
|
||||
if req.parts:
|
||||
stmt = stmt.where(DischargeDetail.part_number.in_(req.parts))
|
||||
if req.pedimento_key:
|
||||
stmt = stmt.where(ExportPedimento.pedimento_code == req.pedimento_key)
|
||||
|
||||
if req.provider_id:
|
||||
stmt = stmt.where(ExportCompliance.provider_id == req.provider_id)
|
||||
|
||||
if req.sold_to_id:
|
||||
stmt = stmt.where(ExportCompliance.sold_to_id == req.sold_to_id)
|
||||
|
||||
if getattr(req, 'shipped_to_id', None):
|
||||
stmt = stmt.where(ExportCompliance.shipped_to_id == req.shipped_to_id)
|
||||
|
||||
if req.destination_customs:
|
||||
stmt = stmt.where(ExportCompliance.aduana == req.destination_customs)
|
||||
|
||||
if not req.include_exempt_fraction:
|
||||
stmt = stmt.where(
|
||||
or_(
|
||||
Class.iva_exempt_fraction.is_(None),
|
||||
Class.iva_exempt_fraction != 'Si',
|
||||
)
|
||||
)
|
||||
|
||||
if req.print_class_mode == 'downloaded' and not req.show_all_temporary_balances:
|
||||
stmt = stmt.where(
|
||||
or_(
|
||||
ImportLine.payment_method.is_(None),
|
||||
ImportLine.payment_method != '2',
|
||||
)
|
||||
)
|
||||
|
||||
stmt = stmt.order_by(
|
||||
Class.class_code.nullslast(),
|
||||
ExportInvoice.invoice_date,
|
||||
DischargeDetail.id,
|
||||
)
|
||||
|
||||
return db.execute(stmt).fetchall()
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# CSV generation
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def generate_csv(
|
||||
self,
|
||||
db: Session,
|
||||
req: DownloadedPartsReportRequest,
|
||||
company_id: int,
|
||||
tenant_id: int,
|
||||
) -> StreamingResponse:
|
||||
settings = self._get_settings(db, tenant_id, company_id)
|
||||
use_previous_payment_day = bool(settings.get('utilizarfechapagopeddeundiaanterior'))
|
||||
company = (
|
||||
db.query(Company)
|
||||
.options(selectinload(Company.addresses))
|
||||
.filter(Company.id == company_id, Company.tenant_id == tenant_id)
|
||||
.first()
|
||||
)
|
||||
|
||||
rows = self.query_discharge_data(db, req, company_id, tenant_id)
|
||||
rate_dates = self._collect_rate_dates(rows, req, use_previous_payment_day)
|
||||
rate_lookup = self._load_exchange_rates(db, rate_dates, company_id, tenant_id)
|
||||
missing = self._find_missing_rate_dates(rows, req, rate_lookup, use_previous_payment_day)
|
||||
if missing:
|
||||
raise HTTPException(status_code=422, detail={'missing_dates': missing})
|
||||
|
||||
headers = self._build_headers(req, company)
|
||||
output = io.StringIO()
|
||||
writer = csv.writer(output)
|
||||
|
||||
self._write_company_header(writer, company, settings)
|
||||
writer.writerow(headers)
|
||||
|
||||
current_class: Optional[str] = None
|
||||
class_qty = Decimal('0')
|
||||
class_weight_kgs = Decimal('0')
|
||||
class_weight_lbs = Decimal('0')
|
||||
class_mn = Decimal('0')
|
||||
class_me = Decimal('0')
|
||||
fraction_totals: dict[str, dict[str, Decimal]] = {}
|
||||
series_map = self._build_series_map(
|
||||
db,
|
||||
[row.export_line_id for row in rows if row.export_line_id is not None],
|
||||
tenant_id,
|
||||
)
|
||||
|
||||
def flush_class_total() -> None:
|
||||
if req.print_class_total and current_class is not None:
|
||||
writer.writerow(['TOTAL DE LA CLASE'])
|
||||
writer.writerow([
|
||||
self._decimal_str(class_qty, 4),
|
||||
self._decimal_str(class_weight_kgs, 4),
|
||||
self._decimal_str(class_weight_lbs, 4),
|
||||
self._decimal_str(class_mn),
|
||||
self._decimal_str(class_me),
|
||||
])
|
||||
|
||||
for row in rows:
|
||||
class_code = row.class_code or ''
|
||||
if req.print_class_total and class_code != current_class:
|
||||
flush_class_total()
|
||||
current_class = class_code
|
||||
class_qty = Decimal('0')
|
||||
class_weight_kgs = Decimal('0')
|
||||
class_weight_lbs = Decimal('0')
|
||||
class_mn = Decimal('0')
|
||||
class_me = Decimal('0')
|
||||
|
||||
export_fraction = self._resolve_export_fraction(row, req)
|
||||
class_fraction = self._resolve_class_fraction(row, req)
|
||||
description_value = self._resolve_description(row, req)
|
||||
part_description = self._resolve_part_description(row)
|
||||
value_mn, value_me, selected_rate = self._resolve_values(
|
||||
row,
|
||||
req,
|
||||
rate_lookup,
|
||||
use_previous_payment_day,
|
||||
)
|
||||
|
||||
quantity = self._to_decimal(row.quantity)
|
||||
weight_kgs = self._to_decimal(row.net_weight)
|
||||
weight_lbs = weight_kgs * LBS_PER_KG
|
||||
|
||||
if req.print_class_total:
|
||||
class_qty += quantity
|
||||
class_weight_kgs += weight_kgs
|
||||
class_weight_lbs += weight_lbs
|
||||
class_mn += value_mn
|
||||
class_me += value_me
|
||||
|
||||
if req.include_totals_by_fraction:
|
||||
fraction_key = export_fraction or ''
|
||||
if fraction_key not in fraction_totals:
|
||||
fraction_totals[fraction_key] = {'mn': Decimal('0'), 'me': Decimal('0')}
|
||||
fraction_totals[fraction_key]['mn'] += value_mn
|
||||
fraction_totals[fraction_key]['me'] += value_me
|
||||
|
||||
export_ped = self._build_pedimento_str(
|
||||
row.expo_ped_year,
|
||||
row.expo_ped_customs,
|
||||
row.expo_ped_license,
|
||||
row.expo_ped_number,
|
||||
)
|
||||
export_r1 = self._build_pedimento_str(
|
||||
row.expo_r1_year,
|
||||
row.expo_r1_customs,
|
||||
row.expo_r1_license,
|
||||
row.expo_r1_number,
|
||||
)
|
||||
import_ped = self._build_pedimento_str(
|
||||
row.impo_ped_year,
|
||||
row.impo_ped_customs,
|
||||
row.impo_ped_license,
|
||||
row.impo_ped_number,
|
||||
)
|
||||
import_r1 = self._build_pedimento_str(
|
||||
row.impo_r1_year,
|
||||
row.impo_r1_customs,
|
||||
row.impo_r1_license,
|
||||
row.impo_r1_number,
|
||||
)
|
||||
export_ped_18 = self._build_pedimento_18(
|
||||
row.expo_ped_year,
|
||||
row.expo_ped_customs,
|
||||
row.expo_ped_license,
|
||||
row.expo_ped_number,
|
||||
row.expo_ped_code,
|
||||
)
|
||||
import_ped_18 = self._build_pedimento_18(
|
||||
row.impo_ped_year,
|
||||
row.impo_ped_customs,
|
||||
row.impo_ped_license,
|
||||
row.impo_ped_number,
|
||||
row.impo_ped_code,
|
||||
)
|
||||
|
||||
export_ped_values = self._pedimento_values(company, export_ped, export_r1)
|
||||
import_ped_values = self._pedimento_values(company, import_ped, import_r1)
|
||||
|
||||
csv_row = [
|
||||
self._excel_text(export_ped_values[0]),
|
||||
self._excel_text(export_ped_values[1]),
|
||||
self._excel_text(self._format_date(row.expo_payment_date, req.julian_date)),
|
||||
self._excel_text(row.expo_clave or ''),
|
||||
self._excel_text(self._format_date(row.discharge_date, req.julian_date)),
|
||||
self._excel_text(row.expo_invoice_number or ''),
|
||||
self._excel_text(self._format_date(row.expo_invoice_date, req.julian_date)),
|
||||
self._excel_text(class_code),
|
||||
self._excel_text(description_value),
|
||||
self._excel_text(export_fraction),
|
||||
self._decimal_str(quantity, 4),
|
||||
self._excel_text(row.unit_of_measure or ''),
|
||||
self._excel_text(import_ped_values[0]),
|
||||
self._excel_text(import_ped_values[1]),
|
||||
self._excel_text(self._format_date(row.impo_payment_date, req.julian_date)),
|
||||
self._excel_text(row.import_invoice_number or row.import_invoice_str or ''),
|
||||
self._excel_text(self._format_date(row.impo_invoice_date, req.julian_date)),
|
||||
self._decimal_str(weight_kgs, 4),
|
||||
self._decimal_str(value_me),
|
||||
self._decimal_str(value_mn),
|
||||
self._decimal_str(selected_rate, 6) if selected_rate else '',
|
||||
self._excel_text(row.part_number_str or ''),
|
||||
self._excel_text(part_description),
|
||||
self._excel_text(row.material_key or ''),
|
||||
self._excel_text(class_fraction),
|
||||
self._excel_text(import_ped_18),
|
||||
self._excel_text(export_ped_18),
|
||||
self._excel_text(row.tariff_uom or ''),
|
||||
]
|
||||
|
||||
if req.include_american_fraction_and_country:
|
||||
csv_row.extend([
|
||||
self._excel_text(row.american_fraction or ''),
|
||||
self._excel_text(row.country_of_origin or ''),
|
||||
])
|
||||
|
||||
writer.writerow(csv_row)
|
||||
|
||||
if req.include_series and row.export_line_id in series_map:
|
||||
writer.writerow([
|
||||
'RENGLON',
|
||||
'SERIE',
|
||||
'MODELO',
|
||||
'SUBMODELO',
|
||||
'PARTE',
|
||||
'NUM ID EXPO',
|
||||
'MARCA PARTIDA',
|
||||
'MODELO PARTIDA',
|
||||
])
|
||||
for series in series_map[row.export_line_id]:
|
||||
writer.writerow([
|
||||
self._excel_text(series.row or ''),
|
||||
self._excel_text(series.serial_numbers or ''),
|
||||
self._excel_text(series.model or ''),
|
||||
self._excel_text(series.sub_model or ''),
|
||||
self._excel_text(row.part_number_str or ''),
|
||||
self._excel_text(series.number_id or ''),
|
||||
self._excel_text(row.export_brand or ''),
|
||||
self._excel_text(row.export_model or ''),
|
||||
])
|
||||
|
||||
flush_class_total()
|
||||
|
||||
if req.include_totals_by_fraction and fraction_totals:
|
||||
writer.writerow([])
|
||||
writer.writerow(['TOTALES POR FRACCION'])
|
||||
writer.writerow(['FRACCION', 'VALOR MN', 'VALOR ME'])
|
||||
total_mn = Decimal('0')
|
||||
total_me = Decimal('0')
|
||||
for fraction, totals in sorted(fraction_totals.items()):
|
||||
writer.writerow([
|
||||
fraction,
|
||||
self._decimal_str(totals['mn']),
|
||||
self._decimal_str(totals['me']),
|
||||
])
|
||||
total_mn += totals['mn']
|
||||
total_me += totals['me']
|
||||
|
||||
writer.writerow(['TOTAL'])
|
||||
writer.writerow(['', self._decimal_str(total_mn), self._decimal_str(total_me)])
|
||||
|
||||
csv_content = output.getvalue()
|
||||
filename = f"partes_descargadas_{req.date_from}_{req.date_to}.csv"
|
||||
return StreamingResponse(
|
||||
iter([csv_content.encode('utf-8-sig')]),
|
||||
media_type='text/csv',
|
||||
headers={'Content-Disposition': f'attachment; filename="{filename}"'},
|
||||
)
|
||||
@@ -41,6 +41,7 @@ from .reports.importacion.facturas.routes import router as invoices_reports_rout
|
||||
from .reports.importacion.consolidados.routes import router as consolidated_reports_router
|
||||
from .reports.importacion.packing_list.routes import router as packing_list_router
|
||||
from .reports.exportacion.aviso_consolidado.routes import router as aviso_consolidado_export_router
|
||||
from .reports.exportacion.partes_descargadas.routes import router as downloaded_parts_reports_router
|
||||
from .reports.movements.invoices.routes import router as movement_invoices_router
|
||||
from .reports.movements.saldos.routes import router as movement_saldos_router
|
||||
from .reports.exportacion.descargo.routes import router as discharge_reports_router
|
||||
@@ -120,6 +121,12 @@ router.include_router(
|
||||
tags=["a76 / reports"]
|
||||
)
|
||||
|
||||
router.include_router(
|
||||
downloaded_parts_reports_router,
|
||||
prefix="/a76/reports/exportacion/partes-descargadas",
|
||||
tags=["a76 / reports"]
|
||||
)
|
||||
|
||||
router.include_router(
|
||||
|
||||
movement_invoices_router,
|
||||
|
||||
@@ -1,10 +1,13 @@
|
||||
from typing import Optional
|
||||
|
||||
from pydantic import BaseModel
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
|
||||
TRANSPORTER_KEY_MAX_LENGTH = 30
|
||||
|
||||
|
||||
class DriverBaseDTO(BaseModel):
|
||||
transporter_key: str
|
||||
transporter_key: str = Field(..., max_length=TRANSPORTER_KEY_MAX_LENGTH)
|
||||
driver_id: Optional[int] = None
|
||||
line: int
|
||||
driver_name: Optional[str] = None
|
||||
|
||||
@@ -10,7 +10,7 @@ class Driver(Base, TenantScopedMixin, TimestampMixin):
|
||||
)
|
||||
|
||||
transporter_key = Column(
|
||||
String(5),
|
||||
String(30),
|
||||
ForeignKey("a76.transporter.transporter_key", ondelete="CASCADE"),
|
||||
primary_key=True,
|
||||
nullable=False,
|
||||
|
||||
Reference in New Issue
Block a user