backend/integrations/ceidg_client.py

"""
CEIDG (Centralna Ewidencja i Informacja o Działalności Gospodarczej) client.

Priority:
  1. Official CEIDG API v2 (dane.biznes.gov.pl) when CEIDG_API_KEY set
  2. Soft-fail empty result — GUS BIR already covers CEIDG natural persons

Env:
  CEIDG_API_KEY / CEIDG_API_TOKEN
  CEIDG_API_BASE (default https://dane.biznes.gov.pl/api/ceidg/v2)
  CEIDG_DISABLED=true
"""

from __future__ import annotations

import logging
import os
import re
from typing import Any, Dict, List, Optional

import httpx

logger = logging.getLogger(__name__)

DEFAULT_BASE = "https://dane.biznes.gov.pl/api/ceidg/v2"


def fetch_ceidg_for_nip(nip: str) -> Dict[str, Any]:
    nip_clean = re.sub(r"\D", "", nip or "")
    if len(nip_clean) != 10:
        return _empty(nip_clean, reason="invalid_nip")

    if os.environ.get("CEIDG_DISABLED", "").lower() in ("1", "true", "yes"):
        return _empty(nip_clean, reason="disabled")

    api_key = (
        os.environ.get("CEIDG_API_KEY")
        or os.environ.get("CEIDG_API_TOKEN")
        or os.environ.get("BIZNES_GOV_API_KEY")
        or ""
    ).strip()
    if not api_key:
        return _empty(
            nip_clean,
            reason="no_api_key",
            message="CEIDG API key missing — dane JDG pochodzą z GUS BIR (raport CEIDG).",
        )

    base = (os.environ.get("CEIDG_API_BASE") or DEFAULT_BASE).rstrip("/")
    timeout = float(os.environ.get("CEIDG_TIMEOUT", "12"))
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Accept": "application/json",
    }

    # Official search endpoints vary; try NIP filter then fallback path
    candidates = [
        (f"{base}/firmy", {"nip": nip_clean}),
        (f"{base}/company", {"nip": nip_clean}),
    ]
    last_err = ""
    for url, params in candidates:
        try:
            with httpx.Client(timeout=timeout, follow_redirects=True) as client:
                resp = client.get(url, params=params, headers=headers)
            if resp.status_code in (401, 403):
                return _empty(nip_clean, reason="unauthorized", message="CEIDG API: brak uprawnień.")
            if resp.status_code == 404:
                continue
            if resp.status_code != 200:
                last_err = f"http_{resp.status_code}"
                continue
            data = resp.json()
            parsed = _normalize(nip_clean, data)
            if parsed.get("configured"):
                return parsed
        except Exception as e:
            last_err = type(e).__name__
            logger.debug("[CEIDG] %s: %s", url, e)

    return _empty(nip_clean, reason=last_err or "not_found")


def _normalize(nip: str, data: Any) -> Dict[str, Any]:
    firm: Optional[Dict[str, Any]] = None
    if isinstance(data, dict):
        if "firmy" in data and isinstance(data["firmy"], list) and data["firmy"]:
            firm = data["firmy"][0]
        elif "company" in data and isinstance(data["company"], dict):
            firm = data["company"]
        elif data.get("nip") or data.get("nazwa"):
            firm = data
    elif isinstance(data, list) and data:
        firm = data[0] if isinstance(data[0], dict) else None

    if not firm:
        return _empty(nip, reason="empty_payload")

    name = firm.get("nazwa") or firm.get("name") or firm.get("firma") or ""
    pkd = _extract_pkd(firm)
    address = _extract_address(firm)
    status = firm.get("status") or firm.get("statusDzialalnosci") or firm.get("status_dzialalnosci")
    return {
        "configured": bool(name or pkd or address),
        "nip": nip,
        "source": "ceidg_api",
        "name": name,
        "regon": firm.get("regon") or firm.get("REGON"),
        "pkd": pkd,
        "address": address,
        "status": status,
        "legal_form": "jednoosobowa działalność gospodarcza",
        "entity_type": "jdg",
        "start_date": firm.get("dataRozpoczecia") or firm.get("data_rozpoczecia"),
        "message": None,
    }


def _extract_pkd(firm: Dict[str, Any]) -> List[str]:
    out: List[str] = []
    raw = firm.get("pkd") or firm.get("pkdList") or firm.get("kodyPkd") or []
    if isinstance(raw, str):
        raw = [raw]
    if isinstance(raw, dict):
        raw = [raw]
    for item in raw or []:
        if isinstance(item, str):
            code = item.strip().upper()
        elif isinstance(item, dict):
            code = (item.get("kod") or item.get("code") or item.get("pkd") or "").strip().upper()
        else:
            continue
        if not code:
            continue
        code = re.sub(r"[^0-9A-Z.]", "", code)
        if len(code) == 5 and code.isdigit() is False:
            # 6201Z → 62.01.Z
            digits = re.sub(r"\D", "", code)
            letter = re.sub(r"\d", "", code)
            if len(digits) >= 4:
                code = f"{digits[:2]}.{digits[2:4]}.{letter or digits[4:]}"
        if code and code not in out:
            out.append(code)
    return out[:20]


def _extract_address(firm: Dict[str, Any]) -> str:
    addr = firm.get("adres") or firm.get("address") or firm.get("adresDzialalnosci") or {}
    if isinstance(addr, str):
        return addr
    if not isinstance(addr, dict):
        return ""
    parts = [
        addr.get("ulica") or addr.get("street"),
        addr.get("budynek") or addr.get("building"),
        addr.get("lokal") or addr.get("apartment"),
        addr.get("miasto") or addr.get("city") or addr.get("miejscowosc"),
        addr.get("kod") or addr.get("postalCode") or addr.get("kodPocztowy"),
    ]
    return ", ".join(str(p) for p in parts if p)


def _empty(nip: str, *, reason: str = "", message: Optional[str] = None) -> Dict[str, Any]:
    return {
        "configured": False,
        "nip": nip,
        "source": "ceidg",
        "name": None,
        "regon": None,
        "pkd": [],
        "address": None,
        "status": None,
        "legal_form": None,
        "entity_type": None,
        "message": message or f"CEIDG niedostępne ({reason}).",
        "reason": reason,
    }