Consulta y fusiona datos empresariales desde datos.gov.co y rues.org.co. Retorna una estructura consolidada y verificada.

consultar_empresa_por_nit.py

import requests

def consultar_empresa_por_nit(nit: str) -> dict:
    """
    Consulta y fusiona datos empresariales desde datos.gov.co y rues.org.co.
    Retorna una estructura consolidada y verificada.
    """
    fuentes = []
    resultado = {}
    datos_gov = {}
    datos_rues = {}

    # === Consulta a datos.gov.co ===
    try:
        url_gov = f"https://www.datos.gov.co/resource/c82u-588k.json?nit={nit}"
        res_gov = requests.get(url_gov, timeout=10)
        res_gov.raise_for_status()
        data_gov = res_gov.json()
        if data_gov:
            fuentes.append("datos.gov.co")
            item = data_gov[0]
            datos_gov = {
                "razon_social": item.get("razon_social"),
                "nit": item.get("nit"),
                "dv": item.get("digito_verificacion"),
                "camara_comercio": item.get("camara_comercio"),
                "matricula": item.get("matricula").zfill(10) if item.get("matricula") else None,
                "estado": item.get("estado_matricula"),
                "fecha_matricula": item.get("fecha_matricula"),
                "fecha_renovacion": item.get("fecha_renovacion"),
                "ultimo_ano_renovado": item.get("ultimo_ano_renovado"),
                "tipo_sociedad": item.get("tipo_sociedad"),
                "organizacion_juridica": item.get("organizacion_juridica"),
                "ciiu_principal": {
                    "codigo": item.get("cod_ciiu_act_econ_pri"),
                    "descripcion": ""  # datos.gov.co no da descripción
                }
            }
    except Exception as e:
        pass  # Se continúa con rues si falla datos.gov.co

    # === Consulta a datos.gov.co para código_camara y matrícula ===
    try:
        if not datos_gov:
            query = f"https://www.datos.gov.co/resource/c82u-588k.json?$select=codigo_camara,matricula&nit={nit}"
            meta = requests.get(query, timeout=10)
            meta.raise_for_status()
            fallback = meta.json()
        else:
            fallback = [{
                "codigo_camara": item.get("codigo_camara"),
                "matricula": item.get("matricula")
            }]
    except:
        fallback = []

    # === Si tenemos código_camara y matrícula, consultar rues ===
    try:
        if fallback and fallback[0].get("codigo_camara") and fallback[0].get("matricula"):
            camara = fallback[0]["codigo_camara"]
            matricula = fallback[0]["matricula"]
            union = f"{camara}{matricula}"
            relleno = "0" * (12 - len(union))
            codigo = f"{camara}{relleno}{matricula}"

            url_rues = f"https://ruesapi.rues.org.co/WEB2/api/Expediente/DetalleRM/{codigo}"
            res_rues = requests.get(url_rues, timeout=10)
            res_rues.raise_for_status()
            json_rues = res_rues.json()

            if json_rues.get("codigo_error") == "0000":
                fuentes.append("rues.org.co")
                r = json_rues.get("registros", {})
                datos_rues = {
                    "razon_social": r.get("razon_social"),
                    "nit": r.get("numero_identificacion"),
                    "dv": r.get("dv"),
                    "camara_comercio": r.get("camara"),
                    "matricula": r.get("matricula"),
                    "estado": r.get("estado"),
                    "fecha_matricula": r.get("fecha_matricula"),
                    "fecha_renovacion": r.get("fecha_renovacion"),
                    "ultimo_ano_renovado": r.get("ultimo_ano_renovado"),
                    "tipo_sociedad": r.get("tipo_sociedad"),
                    "organizacion_juridica": r.get("organizacion_juridica"),
                    "ciiu_principal": {
                        "codigo": r.get("cod_ciiu_act_econ_pri"),
                        "descripcion": r.get("desc_ciiu_act_econ_pri"),
                    }
                }
    except Exception as e:
        pass  # Rues no disponible

    # === Unir datos dando prioridad a datos.gov.co ===
    campos = [
        "razon_social", "nit", "dv", "camara_comercio", "matricula", "estado",
        "fecha_matricula", "fecha_renovacion", "ultimo_ano_renovado",
        "tipo_sociedad", "organizacion_juridica", "ciiu_principal"
    ]

    for campo in campos:
        valor = datos_gov.get(campo)
        if valor in [None, "", {}, []]:
            resultado[campo] = datos_rues.get(campo)
        else:
            resultado[campo] = valor

    resultado["fuentes"] = fuentes

    if not resultado.get("razon_social"):
        return {"error": "No se encontró información para el NIT"}

    return resultado

data = consultar_empresa_por_nit("900929107")
print(data)