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