diff --git a/backend/api/v1/modules/a76/items/common/fractions.py b/backend/api/v1/modules/a76/items/common/fractions.py index aad8eedc..8f4fc501 100644 --- a/backend/api/v1/modules/a76/items/common/fractions.py +++ b/backend/api/v1/modules/a76/items/common/fractions.py @@ -1,31 +1,52 @@ import asyncio +from datetime import datetime from typing import Optional, Tuple from api.v1.modules.a76.general_catalogs.company.models import Company +from api.v1.modules.a76.general_catalogs.fractions.historical_tariff_fractions.models import HistoricalTariffFraction from api.v1.modules.sitar.tlcs import TLCSService from api.v1.modules.sitar.prosec import ProsecService from api.v1.modules.sitar.fracciones import FraccionesService +from sqlalchemy.orm import Session +from sqlalchemy import exists +from core.exceptions import ErrorCollector - -def search_historical_fraction() -> Tuple[Optional[str], float]: - """Search for historical fraction data - - TODO: Implement historical fraction search logic +def search_historical_fraction(db: Session, fraction: str, country: str, fraction_type: str, sector: str, invoice_date: Optional[datetime], errors: Optional[ErrorCollector]) -> Tuple[Optional[str], float]: + """Search for historical fraction data This corresponds to BUSCA_FRACCION_HISTORICA in original Clarion code Returns: Tuple of (rate_im, adv_impo) """ # Placeholder for historical search - return None, 0.0 + historical_exists = db.query(exists().where(HistoricalTariffFraction.historical_fraction == fraction)).scalar() + if not historical_exists: + errors.add_error( + field=f"fraction, country, fraction_type, sector", + message=f"La fraccion {fraction}, con pais {country}, con preferencia {fraction_type} y sector {sector} no existe.", + solution=["Revisar que si exista la preferencia para esta fracción en caso de ser historico, registrarlo en el catálogo de fracciones historicas."], + code="HISTORICAL_FRACTION_NOT_FOUND", + ) + else: + historical_exists = db.query(HistoricalTariffFraction).filter( + HistoricalTariffFraction.historical_fraction == fraction, + HistoricalTariffFraction.country == country, + HistoricalTariffFraction.fraction_type == fraction_type, + HistoricalTariffFraction.sector == sector, + HistoricalTariffFraction.publication_date <= invoice_date, + ).first() + + return historical_exists.import_tax_rate, historical_exists.import_tax_rate def search_fraction_preference( + db: Session, country: str, fraccion: str, - fraction_type: str, - company: Company, + fraction_type: str, sector: Optional[str] = None, + invoice_date: Optional[datetime] = None, + errors: Optional[ErrorCollector] = None ) -> Tuple[Optional[str], float]: """Search fraction preference and return rate_im and adv_impo @@ -59,7 +80,7 @@ def search_fraction_preference( ) if not tlcs_data: - search_historical_fraction() + rate_im, adv_impo = search_historical_fraction(db=db, fraction=fraccion_8, country=country, fraction_type=fraction_type, sector=sector, invoice_date=invoice_date, errors=errors) else: first_record = tlcs_data[0] rate_im = first_record.TASATXT @@ -70,7 +91,7 @@ def search_fraction_preference( """ ALADI Search """ if fraction_type.upper() == "ALADI": - search_historical_fraction() + rate_im, adv_impo = search_historical_fraction(db=db, fraction=fraccion_8, country=country, fraction_type=fraction_type, sector=sector, invoice_date=invoice_date, errors=errors) """ PROSEC Search """ if fraction_type.upper() == "PROSEC": @@ -94,7 +115,7 @@ def search_fraction_preference( ) if not prosec_data: - search_historical_fraction() + rate_im, adv_impo = search_historical_fraction(db=db, fraction=fraccion_8, country=country, fraction_type=fraction_type, sector=sector, invoice_date=invoice_date, errors=errors) else: first_record = prosec_data[0] rate_im = first_record.TASATXT @@ -120,7 +141,7 @@ def search_fraction_preference( ) if not fracciones_data: - search_historical_fraction() + rate_im, adv_impo = search_historical_fraction(db=db, fraction=fraccion_8, country=country, fraction_type=fraction_type, sector=sector, invoice_date=invoice_date, errors=errors) else: # Se encontraron registros, tomar el primero first_record = fracciones_data[0]