nordabiz/ceidg_api_service.py

#!/usr/bin/env python3
"""
CEIDG API Service
==================

Service module for fetching company data from CEIDG (Centralna Ewidencja
i Informacja o Działalności Gospodarczej) using the official API at
dane.biznes.gov.pl.

Provides fetch_ceidg_by_nip function for membership application workflow.
"""

import os
import logging
import requests
from typing import Optional, Dict, Any
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

logger = logging.getLogger(__name__)

# API Configuration
CEIDG_API_V3_URL = "https://dane.biznes.gov.pl/api/ceidg/v3/firmy"
CEIDG_API_KEY = os.getenv("CEIDG_API_KEY")
CEIDG_TIMEOUT = 15  # seconds


def fetch_ceidg_by_nip(nip: str) -> Optional[Dict[str, Any]]:
    """
    Fetch company data from CEIDG API by NIP.

    Args:
        nip: NIP number (10 digits, no dashes)

    Returns:
        Dictionary with company data or None if not found
    """
    if not CEIDG_API_KEY:
        logger.warning("CEIDG_API_KEY not configured - CEIDG lookup disabled")
        return None

    # Clean NIP
    nip = nip.strip().replace('-', '').replace(' ', '')
    if not nip or len(nip) != 10 or not nip.isdigit():
        logger.warning(f"Invalid NIP format: {nip}")
        return None

    headers = {
        "Authorization": f"Bearer {CEIDG_API_KEY}",
        "Accept": "application/json"
    }

    try:
        logger.info(f"Fetching CEIDG data for NIP {nip}")

        response = requests.get(
            CEIDG_API_V3_URL,
            params={"nip": nip},
            headers=headers,
            timeout=CEIDG_TIMEOUT
        )

        if response.status_code == 401:
            logger.error("CEIDG API authentication failed - check CEIDG_API_KEY")
            return None

        if response.status_code == 404:
            logger.info(f"NIP {nip} not found in CEIDG")
            return None

        if response.status_code != 200:
            logger.error(f"CEIDG API error: {response.status_code} - {response.text[:200]}")
            return None

        data = response.json()

        # Handle response format - can be list or dict
        if isinstance(data, list):
            if not data:
                logger.info(f"NIP {nip} not found in CEIDG (empty list)")
                return None
            firma = data[0]
        elif isinstance(data, dict):
            if 'firmy' in data:
                firmy = data.get('firmy', [])
                if not firmy:
                    logger.info(f"NIP {nip} not found in CEIDG")
                    return None
                firma = firmy[0]
            else:
                firma = data
        else:
            logger.error(f"Unexpected CEIDG response format: {type(data)}")
            return None

        # Extract address
        adres = firma.get('adresDzialalnosci', {}) or firma.get('adres', {}) or {}
        if isinstance(adres, str):
            adres = {'full': adres}

        # Build normalized result
        result = {
            'firma': firma.get('nazwa') or firma.get('nazwaSkrocona'),
            'nip': firma.get('nip'),
            'regon': firma.get('regon'),
            'adresDzialalnosci': {
                'kodPocztowy': adres.get('kodPocztowy') or adres.get('kod'),
                'miejscowosc': adres.get('miejscowosc') or adres.get('miasto'),
                'ulica': adres.get('ulica'),
                'budynek': adres.get('budynek') or adres.get('nrDomu') or adres.get('nrBudynku'),
                'lokal': adres.get('lokal') or adres.get('nrLokalu'),
            },
            'email': firma.get('email') or firma.get('adresEmail'),
            'stronaWWW': firma.get('stronaWWW') or firma.get('www') or firma.get('strona'),
            'telefon': firma.get('telefon'),
            'dataRozpoczeciaDzialalnosci': firma.get('dataRozpoczeciaDzialalnosci') or firma.get('dataWpisuDoCeidg'),
            'status': firma.get('status'),
            'raw': firma
        }

        logger.info(f"CEIDG data found for NIP {nip}: {result['firma']}")
        return result

    except requests.exceptions.Timeout:
        logger.error(f"CEIDG API timeout for NIP {nip}")
        return None
    except requests.exceptions.RequestException as e:
        logger.error(f"CEIDG API request error for NIP {nip}: {e}")
        return None
    except Exception as e:
        logger.error(f"Error fetching CEIDG data for NIP {nip}: {e}")
        return None


# For testing
if __name__ == '__main__':
    import sys
    import json

    if len(sys.argv) < 2:
        print("Usage: python ceidg_api_service.py <NIP>")
        print("Example: python ceidg_api_service.py 5881571773")
        sys.exit(1)

    nip = sys.argv[1]
    print(f"Pobieranie danych z CEIDG API dla NIP: {nip}")
    print("=" * 60)

    data = fetch_ceidg_by_nip(nip)

    if data:
        print(json.dumps(data, indent=2, ensure_ascii=False, default=str))
    else:
        print(f"Nie znaleziono firmy o NIP {nip} w CEIDG")