feat: implement historical fraction search logic and integrate with fraction preference search

This commit is contained in:
2026-02-11 12:49:17 -06:00
parent b2299e9efc
commit ef4f3cdeac

View File

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