156 lines
6.5 KiB
Python
156 lines
6.5 KiB
Python
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(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
|
|
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,
|
|
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
|
|
|
|
Args:
|
|
country: Country code
|
|
fraction_type: Type of fraction (e.g., 'TLCS', 'PROSEC')
|
|
company: Company object with configuration
|
|
fraccion: Tariff fraction code to search
|
|
sector: Sector code (required for PROSEC searches)
|
|
|
|
Returns:
|
|
Tuple of (rate_im, adv_impo) where:
|
|
- rate_im: Tax rate as string (e.g., "EXE", "5.0%")
|
|
- adv_impo: Numeric ad valorem rate
|
|
"""
|
|
rate_im = None
|
|
adv_impo = 0.0
|
|
fraccion_8 = fraccion[:8]
|
|
|
|
country_group = "USA" if country == "MEX" else country
|
|
|
|
"""" TLCS Search """
|
|
if fraction_type.upper() == "TLCS":
|
|
try:
|
|
# Get TLCS service instance
|
|
tlcs_service = TLCSService.get_instance()
|
|
|
|
# Fetch TLCS data using the service
|
|
tlcs_data = asyncio.run(
|
|
tlcs_service.search(fraccion=fraccion_8, pais=country_group, limit=100)
|
|
)
|
|
|
|
if not tlcs_data:
|
|
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
|
|
adv_impo = float(first_record.TASA1NUM or 0.0)
|
|
|
|
except Exception as e:
|
|
print(f"Error fetching Fracciones data from SITAR API: {e}")
|
|
|
|
""" ALADI Search """
|
|
if fraction_type.upper() == "ALADI":
|
|
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":
|
|
try:
|
|
# Búsqueda en API de PROSEC (TARIFA_AS..sProsec)
|
|
prosec_service = ProsecService.get_instance()
|
|
|
|
# Buscar primero con ARTICULO = '4to'
|
|
prosec_data = asyncio.run(
|
|
prosec_service.search(
|
|
fraccion=fraccion_8, sector=sector, articulo="4to", limit=100
|
|
)
|
|
)
|
|
|
|
# Si no se encuentra con '4to', buscar con '5to'
|
|
if not prosec_data:
|
|
prosec_data = asyncio.run(
|
|
prosec_service.search(
|
|
fraccion=fraccion_8, sector=sector, articulo="5to", limit=100
|
|
)
|
|
)
|
|
|
|
if not prosec_data:
|
|
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
|
|
adv_impo = float(first_record.TASANUM or 0.0)
|
|
|
|
except Exception as e:
|
|
print(f"Error fetching Fracciones data from SITAR API: {e}")
|
|
|
|
""" General search fallback """
|
|
if fraction_type.upper() == "GENERAL":
|
|
# Búsqueda en API de Fracciones (sFracciones)
|
|
try:
|
|
fracciones_service = FraccionesService.get_instance()
|
|
|
|
# Extraer fracción (primeros 8 caracteres) e nico (caracteres 9-10)
|
|
nico = fraccion[8:10] if len(fraccion) >= 10 else None
|
|
|
|
# Buscar por fracción e histórico
|
|
fracciones_data = asyncio.run(
|
|
fracciones_service.search(
|
|
fraccion=fraccion_8, nico=nico, limit=100
|
|
)
|
|
)
|
|
|
|
if not fracciones_data:
|
|
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]
|
|
rate_im = first_record.ADVIMPOTXT # AdvImpoTxt
|
|
adv_impo = float(first_record.ADVIMPONUM or 0.0)
|
|
|
|
|
|
except Exception as e:
|
|
print(f"Error fetching Fracciones data from SITAR API: {e}")
|
|
|
|
return rate_im, adv_impo
|