135 lines
4.4 KiB
Python
135 lines
4.4 KiB
Python
import asyncio
|
|
from typing import Optional, Tuple
|
|
|
|
from api.v1.modules.a76.general_catalogs.company.models import Company
|
|
from api.v1.modules.sitar.tlcs import TLCSService
|
|
from api.v1.modules.sitar.prosec import ProsecService
|
|
from api.v1.modules.sitar.fracciones import FraccionesService
|
|
|
|
|
|
def search_historical_fraction() -> Tuple[Optional[str], float]:
|
|
"""Search for historical fraction data
|
|
|
|
TODO: Implement historical fraction search logic
|
|
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
|
|
|
|
|
|
def search_fraction_preference(
|
|
country: str,
|
|
fraccion: str,
|
|
fraction_type: str,
|
|
company: Company,
|
|
sector: Optional[str] = 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:
|
|
search_historical_fraction()
|
|
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":
|
|
search_historical_fraction()
|
|
|
|
""" 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:
|
|
search_historical_fraction()
|
|
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:
|
|
search_historical_fraction()
|
|
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
|