feature/reportes

backend/api/v1/modules/a76/invoices/common/process/review_equivalence.py

@@ -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)

backend/api/v1/modules/a76/reports/exportacion/partes_descargadas/__init__.py

@@ -0,0 +1 @@
+"""Downloaded parts report module."""

backend/api/v1/modules/a76/reports/exportacion/partes_descargadas/routes.py

@@ -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)

backend/api/v1/modules/a76/reports/exportacion/partes_descargadas/schemas.py

@@ -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]

backend/api/v1/modules/a76/reports/exportacion/partes_descargadas/service.py

@@ -0,0 +1,954 @@
+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 _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 = [
+                export_ped_values[0],
+                export_ped_values[1],
+                self._format_date(row.expo_payment_date, req.julian_date),
+                row.expo_clave or '',
+                self._format_date(row.discharge_date, req.julian_date),
+                row.expo_invoice_number or '',
+                self._format_date(row.expo_invoice_date, req.julian_date),
+                class_code,
+                description_value,
+                export_fraction,
+                self._decimal_str(quantity, 4),
+                row.unit_of_measure or '',
+                import_ped_values[0],
+                import_ped_values[1],
+                self._format_date(row.impo_payment_date, req.julian_date),
+                row.import_invoice_number or row.import_invoice_str or '',
+                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 '',
+                row.part_number_str or '',
+                part_description,
+                row.material_key or '',
+                class_fraction,
+                import_ped_18,
+                export_ped_18,
+                row.tariff_uom or '',
+            ]
+
+            if req.include_american_fraction_and_country:
+                csv_row.extend([
+                    row.american_fraction or '',
+                    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([
+                        series.row or '',
+                        series.serial_numbers or '',
+                        series.model or '',
+                        series.sub_model or '',
+                        row.part_number_str or '',
+                        series.number_id or '',
+                        row.export_brand or '',
+                        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}"'},
+        )

backend/api/v1/modules/a76/router.py

@@ -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
@@ -118,6 +119,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,