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}"'},
        )
