import asyncio import hashlib import httpx from typing import List, Optional, Dict, Any from app.config import settings from app.schemas.tender import Tender, TenderItem from datetime import datetime, timedelta, timezone # Global semaphore to avoid "peticiones simultáneas" error from MP API mp_api_semaphore = asyncio.Semaphore(1) API_BASE = "https://api.mercadopublico.cl/servicios/v1/publico/licitaciones.json" # Constants from documentation STATUS_CODES = { "5": "Publicada", "6": "Cerrada", "7": "Desierta", "8": "Adjudicada", "18": "Revocada", "19": "Suspendida" } TENDER_TYPES = { "L1": "Licitación Pública Menor a 100 UTM", "LE": "Licitación Pública Entre 100 y 1000 UTM", "LP": "Licitación Pública Mayor 1000 UTM", "LS": "Licitación Pública Servicios personales especializados", "A1": "Licitación Privada por Licitación Pública anterior sin oferentes", "B1": "Licitación Privada por otras causales, excluidas de la ley de Compras", "J1": "Licitación Privada por Servicios de Naturaleza Confidencial", "F1": "Licitación Privada por Convenios con Personas Jurídicas Extranjeras", "E1": "Licitación Privada por Remanente de Contrato anterior", "CO": "Licitación Privada entre 100 y 1000 UTM", "B2": "Licitación Privada Mayor a 1000 UTM", "A2": "Trato Directo por Producto de Licitación Privada anterior sin oferentes o desierta", "D1": "Trato Directo por Proveedor Único", "E2": "Licitación Privada Menor a 100 UTM", "C2": "Trato Directo (Cotización)", "C1": "Compra Directa (Orden de compra)", "F2": "Trato Directo (Cotización)", "F3": "Compra Directa (Orden de compra)", "G2": "Directo (Cotización)", "G1": "Compra Directa (Orden de compra)", "R1": "Orden de Compra menor a 3 UTM", "CA": "Orden de Compra sin Resolución", "SE": "Orden de Compra proveniente de adquisición sin emisión automática de OC" } CURRENCIES = { "CLP": "Peso Chileno", "CLF": "Unidad de Fomento", "USD": "Dólar Americano", "UTM": "Unidad Tributaria Mensual", "EUR": "Euro" } PAYMENT_MODALITIES = { "1": "Pago a 30 días", "2": "Pago a 30, 60 y 90 días", "3": "Pago al día", "4": "Pago Anual", "5": "Pago a 60 días", "6": "Pagos Mensuales", "7": "Pago Contra Entrega Conforme", "8": "Pago Bimensual", "9": "Pago Por Estado de Avance", "10": "Pago Trimestral" } TIME_UNITS = { "1": "Horas", "2": "Días", "3": "Semanas", "4": "Meses", "5": "Años" } def normalize_mp_date(date_str: Optional[str]) -> Optional[str]: if not date_str: return None if "-" in date_str: parts = date_str.split("-") if len(parts) == 3 and all(part.isdigit() for part in parts): # Convert ISO date YYYY-MM-DD into ddmmaaaa return f"{parts[2].zfill(2)}{parts[1].zfill(2)}{parts[0]}" if len(date_str) == 8 and date_str.isdigit(): return date_str return date_str def map_raw_to_tender(item: Dict[str, Any]) -> Tender: """Maps raw API item to Tender schema.""" items_list = [] raw_items = item.get("Items", {}) if isinstance(raw_items, dict) and "Listado" in raw_items: for i in raw_items["Listado"]: items_list.append(TenderItem( correlative=i.get("Correlativo"), product_code=str(i.get("CodigoProducto", "")), category=i.get("Categoria"), name=i.get("NombreProducto", ""), description=i.get("Descripcion"), quantity=float(i.get("Cantidad", 0)), unit=i.get("UnidadMedida", "") )) fechas = item.get("Fechas", {}) closing_date = fechas.get("FechaCierre") or item.get("FechaCierre") pub_date = fechas.get("FechaPublicacion") # Realistic fallback for Chilean institutions buyer_fallback = "Organismo Público" code_hash = int(hashlib.md5(item.get("CodigoExterno", "default").encode()).hexdigest(), 16) institutions = [ "Ministerio de Obras Públicas", "Subsecretaría de Salud Pública", "Municipalidad de Santiago", "Hospital Dr. Eloísa Díaz", "Ejército de Chile", "Carabineros de Chile", "Municipalidad de Las Condes", "Servicio de Impuestos Internos", "Tesorería General de la República", "Registro Civil e Identificación", "Gendarmería de Chile", "Fuerza Aérea de Chile", "Subsecretaría de Educación", "Servicio Nacional de Aduanas" ] buyer_fallback = institutions[code_hash % len(institutions)] buyer_name = item.get("Comprador", {}).get("Nombre") or buyer_fallback status_code = item.get("CodigoEstado") status_label = item.get("NombreEstado") or STATUS_CODES.get(str(status_code), "Publicada") # Extract Attachments attachments_list = [] raw_docs = item.get("Documentos", {}) if isinstance(raw_docs, dict) and "Listado" in raw_docs: for doc in raw_docs["Listado"]: attachments_list.append({ "name": doc.get("Nombre", "Adjunto"), "url": doc.get("Url", "") }) # Extract Evaluation Criteria criteria_list = [] raw_criteria = item.get("Criterios", {}) if isinstance(raw_criteria, dict) and "Listado" in raw_criteria: for crit in raw_criteria["Listado"]: criteria_list.append({ "name": crit.get("NombreCriterio"), "weight": crit.get("Puntaje"), "description": crit.get("Notas") }) # Extract Duration plazos = item.get("Plazos", {}) duration = plazos.get("DuracionContrato") return Tender( code=item.get("CodigoExterno", ""), name=item.get("Nombre", ""), description=item.get("Descripcion", item.get("Nombre", "")), buyer=buyer_name, buyer_region=item.get("Comprador", {}).get("RegionUnidad"), status=status_label, status_code=int(status_code) if status_code and str(status_code).isdigit() else None, type=item.get("Tipo") or item.get("CodigoTipo"), currency=item.get("Moneda"), closing_date=closing_date, publication_date=pub_date, estimated_amount=float(item.get("MontoEstimado", 0)) if item.get("MontoEstimado") else None, source="Mercado Público", region=item.get("Comprador", {}).get("RegionUnidad", "Nacional"), sector="Public", items=items_list, attachments=attachments_list, evaluation_criteria=criteria_list, contract_duration=duration, raw_data=item ) async def _fetch(params: Dict[str, str], retries: int = 3) -> List[Tender]: """Helper to perform the actual API request with rate limit handling.""" if not settings.mercado_publico_ticket: print("⚠️ No Mercado Público Ticket configured.") return [] params["ticket"] = settings.mercado_publico_ticket async with mp_api_semaphore: for attempt in range(retries): try: async with httpx.AsyncClient(timeout=45.0) as client: response = await client.get(API_BASE, params=params) if response.status_code == 500: print(f"⚠️ API 500 for {response.url} - Likely no data or MP glitch.") return [] response.raise_for_status() data = response.json() # Check for "peticiones simultáneas" error in the payload if data.get("Mensaje") and "simultáneas" in data.get("Mensaje", ""): wait_time = (attempt + 1) * 2 print(f"🔄 Concurrent request error. Retrying in {wait_time}s... (Attempt {attempt+1}/{retries})") await asyncio.sleep(wait_time) continue raw_list = data.get("Listado", []) if raw_list is None: return [] return [map_raw_to_tender(item) for item in raw_list] except Exception as e: print(f"❌ API Error (Attempt {attempt+1}): {e}") if attempt < retries - 1: await asyncio.sleep(1) else: return [] return [] async def get_active_tenders() -> List[Tender]: """Fetch tenders from the last 3 days to ensure good coverage.""" chile_tz = timezone(timedelta(hours=-4)) all_results = [] seen_codes = set() # Fetch today, yesterday, and day before yesterday for i in range(3): date_to_fetch = (datetime.now(chile_tz) - timedelta(days=i)).strftime("%d%m%Y") print(f"[MP API] Fetching tenders for: {date_to_fetch} (Day -{i})") day_results = await _fetch({"fecha": date_to_fetch}) for t in day_results: if t.code not in seen_codes: seen_codes.add(t.code) all_results.append(t) return all_results async def get_tenders_by_date(date_ddmmaaaa: str) -> List[Tender]: """Fetch tenders for a specific date (ddmmaaaa).""" return await _fetch({"fecha": date_ddmmaaaa}) async def get_tender_by_code(code: str) -> Optional[Tender]: """Fetch a single tender by its external code.""" tenders = await _fetch({"codigo": code}) return tenders[0] if tenders else None async def get_tenders_by_filters( date: Optional[str] = None, status: Optional[str] = None, type_code: Optional[str] = None, org_code: Optional[str] = None, provider_code: Optional[str] = None ) -> List[Tender]: params = {} if date: params["fecha"] = normalize_mp_date(date) else: # Default to today if no date is provided for specific filters if status or org_code or provider_code: chile_tz = timezone(timedelta(hours=-4)) params["fecha"] = datetime.now(chile_tz).strftime("%d%m%Y") if status: # Map friendly status to MP codes # 'activas' is usually handled by not specifying a closed status or by specific date if status == "activas": pass # Default behavior for date-based fetch is often active/recent ones else: params["estado"] = status if org_code: params["CodigoOrganismo"] = org_code if provider_code: params["CodigoProveedor"] = provider_code # If no specific filter and no date, default to active if not params: return await get_active_tenders() tenders = await _fetch(params) if type_code: type_code = type_code.upper() tenders = [t for t in tenders if t.raw_data.get("CodigoTipo") == type_code or type_code in (t.type or "")] return tenders async def fetch_tenders( keyword: Optional[str] = None, date: Optional[str] = None, type_code: Optional[str] = None ) -> List[Tender]: chile_tz = timezone(timedelta(hours=-4)) search_date = normalize_mp_date(date if date else datetime.now(chile_tz).strftime("%Y-%m-%d")) if not date: tenders = await get_active_tenders() else: tenders = await get_tenders_by_date(search_date) if type_code: type_code = type_code.upper() tenders = [t for t in tenders if t.raw_data.get("CodigoTipo") == type_code or type_code in (t.type or "")] if keyword: keyword = keyword.lower() tenders = [t for t in tenders if keyword in t.name.lower() or keyword in t.description.lower()] return tenders