nordabiz/krs_api_service.py

#!/usr/bin/env python3
"""
KRS Open API Integration Service for NordaBiznes.
Fetches official company data from Krajowy Rejestr Sądowy (Ministry of Justice).

API Documentation: https://prs.ms.gov.pl/krs/openApi
"""
import requests
from datetime import datetime
from typing import Optional, Dict, Any, List
from dataclasses import dataclass


# API Configuration
KRS_API_BASE_URL = "https://api-krs.ms.gov.pl/api/krs"
KRS_API_TIMEOUT = 15  # seconds


@dataclass
class KRSCompanyData:
    """Parsed company data from KRS API."""
    krs: str
    nazwa: str
    nazwa_skrocona: Optional[str]
    nip: Optional[str]
    regon: Optional[str]
    forma_prawna: str

    # Address
    ulica: Optional[str]
    nr_domu: Optional[str]
    nr_lokalu: Optional[str]
    kod_pocztowy: Optional[str]
    miejscowosc: Optional[str]
    wojewodztwo: Optional[str]
    powiat: Optional[str]
    gmina: Optional[str]
    kraj: str

    # Capital
    kapital_zakladowy: Optional[float]
    kapital_waluta: str

    # Dates
    data_rejestracji: Optional[str]
    data_ostatniego_wpisu: Optional[str]
    numer_ostatniego_wpisu: Optional[int]

    # Management (anonymized in Open API)
    zarzad: List[Dict[str, str]]
    sposob_reprezentacji: Optional[str]

    # Shareholders (anonymized in Open API)
    wspolnicy: List[Dict[str, Any]]

    # Other
    przedmiot_dzialalnosci: List[str]
    czy_opp: bool

    # Metadata
    data_odpisu: str
    stan_z_dnia: str

    def to_dict(self) -> Dict[str, Any]:
        """Convert to dictionary for JSON serialization."""
        return {
            'krs': self.krs,
            'nazwa': self.nazwa,
            'nazwa_skrocona': self.nazwa_skrocona,
            'nip': self.nip,
            'regon': self.regon,
            'forma_prawna': self.forma_prawna,
            'adres': {
                'ulica': self.ulica,
                'nr_domu': self.nr_domu,
                'nr_lokalu': self.nr_lokalu,
                'kod_pocztowy': self.kod_pocztowy,
                'miejscowosc': self.miejscowosc,
                'wojewodztwo': self.wojewodztwo,
                'powiat': self.powiat,
                'gmina': self.gmina,
                'kraj': self.kraj,
            },
            'kapital': {
                'zakladowy': self.kapital_zakladowy,
                'waluta': self.kapital_waluta,
            },
            'daty': {
                'rejestracji': self.data_rejestracji,
                'ostatniego_wpisu': self.data_ostatniego_wpisu,
                'numer_ostatniego_wpisu': self.numer_ostatniego_wpisu,
            },
            'zarzad': self.zarzad,
            'sposob_reprezentacji': self.sposob_reprezentacji,
            'wspolnicy': self.wspolnicy,
            'przedmiot_dzialalnosci': self.przedmiot_dzialalnosci,
            'czy_opp': self.czy_opp,
            'metadata': {
                'data_odpisu': self.data_odpisu,
                'stan_z_dnia': self.stan_z_dnia,
                'zrodlo': 'KRS Open API (prs.ms.gov.pl)',
            }
        }


def fetch_krs_data(krs_number: str, rejestr: str = 'P') -> Optional[Dict[str, Any]]:
    """
    Fetch raw data from KRS Open API.

    Args:
        krs_number: KRS number (with or without leading zeros)
        rejestr: 'P' for przedsiębiorców (companies), 'S' for stowarzyszeń (associations)

    Returns:
        Raw JSON response or None if not found/error
    """
    # Normalize KRS number to 10 digits with leading zeros
    krs_normalized = krs_number.zfill(10)

    url = f"{KRS_API_BASE_URL}/OdpisAktualny/{krs_normalized}"
    params = {
        'rejestr': rejestr,
        'format': 'json'
    }

    try:
        response = requests.get(url, params=params, timeout=KRS_API_TIMEOUT)

        if response.status_code == 200:
            return response.json()
        elif response.status_code == 404:
            return None  # Company not found
        else:
            print(f"KRS API error: {response.status_code}")
            return None

    except requests.RequestException as e:
        print(f"KRS API request failed: {e}")
        return None


def parse_krs_response(data: Dict[str, Any]) -> Optional[KRSCompanyData]:
    """
    Parse raw KRS API response into structured KRSCompanyData.

    Args:
        data: Raw JSON response from KRS API

    Returns:
        Parsed KRSCompanyData or None if parsing fails
    """
    try:
        odpis = data.get('odpis', {})
        naglowek = odpis.get('naglowekA', {})
        dane = odpis.get('dane', {})
        dzial1 = dane.get('dzial1', {})
        dzial2 = dane.get('dzial2', {})
        dzial3 = dane.get('dzial3', {})

        # Basic data
        dane_podmiotu = dzial1.get('danePodmiotu', {})
        identyfikatory = dane_podmiotu.get('identyfikatory', {})

        # Address
        siedziba_adres = dzial1.get('siedzibaIAdres', {})
        siedziba = siedziba_adres.get('siedziba', {})
        adres = siedziba_adres.get('adres', {})

        # Capital
        kapital = dzial1.get('kapital', {})
        kapital_zakladowy = kapital.get('wysokoscKapitaluZakladowego', {})

        # Management
        reprezentacja = dzial2.get('reprezentacja', {})
        sklad_zarzadu = reprezentacja.get('sklad', [])

        zarzad = []
        for osoba in sklad_zarzadu:
            zarzad.append({
                'imie': osoba.get('imiona', {}).get('imie', ''),
                'imie_drugie': osoba.get('imiona', {}).get('imieDrugie', ''),
                'nazwisko': osoba.get('nazwisko', {}).get('nazwiskoICzlon', ''),
                'funkcja': osoba.get('funkcjaWOrganie', ''),
                # Note: Data is anonymized in Open API
            })

        # Shareholders
        wspolnicy_raw = dzial1.get('wspolnicySpzoo', [])
        wspolnicy = []
        for wspolnik in wspolnicy_raw:
            wspolnicy.append({
                'imie': wspolnik.get('imiona', {}).get('imie', ''),
                'nazwisko': wspolnik.get('nazwisko', {}).get('nazwiskoICzlon', ''),
                'udzialy': wspolnik.get('posiadaneUdzialy', ''),
                'calosc_udzialow': wspolnik.get('czyPosiadaCaloscUdzialow', False),
            })

        # Business activities (PKD)
        przedmiot = []
        przedmiot_dzial = dzial3.get('przedmiotDzialalnosci', {})
        for pkd in przedmiot_dzial.get('przedmiotPrzewazajacejDzialalnosci', []):
            przedmiot.append(f"{pkd.get('kodDzial', '')} - {pkd.get('opis', '')} (główna)")
        for pkd in przedmiot_dzial.get('przedmiotPozostalejDzialalnosci', []):
            przedmiot.append(f"{pkd.get('kodDzial', '')} - {pkd.get('opis', '')}")

        # Parse capital value
        kapital_value = None
        if kapital_zakladowy.get('wartosc'):
            try:
                kapital_value = float(kapital_zakladowy['wartosc'].replace(',', '.').replace(' ', ''))
            except ValueError:
                pass

        return KRSCompanyData(
            krs=naglowek.get('numerKRS', ''),
            nazwa=dane_podmiotu.get('nazwa', ''),
            nazwa_skrocona=dane_podmiotu.get('nazwaSkrocona'),
            nip=identyfikatory.get('nip'),
            regon=identyfikatory.get('regon'),
            forma_prawna=dane_podmiotu.get('formaPrawna', ''),

            ulica=adres.get('ulica'),
            nr_domu=adres.get('nrDomu'),
            nr_lokalu=adres.get('nrLokalu'),
            kod_pocztowy=adres.get('kodPocztowy'),
            miejscowosc=adres.get('miejscowosc'),
            wojewodztwo=siedziba.get('wojewodztwo'),
            powiat=siedziba.get('powiat'),
            gmina=siedziba.get('gmina'),
            kraj=adres.get('kraj', 'POLSKA'),

            kapital_zakladowy=kapital_value,
            kapital_waluta=kapital_zakladowy.get('waluta', 'PLN'),

            data_rejestracji=naglowek.get('dataRejestracjiWKRS'),
            data_ostatniego_wpisu=naglowek.get('dataOstatniegoWpisu'),
            numer_ostatniego_wpisu=naglowek.get('numerOstatniegoWpisu'),

            zarzad=zarzad,
            sposob_reprezentacji=reprezentacja.get('sposobReprezentacji'),
            wspolnicy=wspolnicy,
            przedmiot_dzialalnosci=przedmiot,
            czy_opp=dane_podmiotu.get('czyPosiadaStatusOPP', False),

            data_odpisu=naglowek.get('dataCzasOdpisu', ''),
            stan_z_dnia=naglowek.get('stanZDnia', ''),
        )

    except Exception as e:
        print(f"Error parsing KRS response: {e}")
        return None


def get_company_from_krs(krs_number: str) -> Optional[KRSCompanyData]:
    """
    Fetch and parse company data from KRS Open API.

    Args:
        krs_number: KRS number

    Returns:
        Parsed KRSCompanyData or None if not found/error
    """
    raw_data = fetch_krs_data(krs_number)
    if raw_data:
        return parse_krs_response(raw_data)
    return None


def verify_company_data(company_krs: str, company_nip: str = None, company_regon: str = None) -> Dict[str, Any]:
    """
    Verify company data against KRS Open API.

    Args:
        company_krs: KRS number to verify
        company_nip: Expected NIP (optional)
        company_regon: Expected REGON (optional)

    Returns:
        Dictionary with verification results
    """
    result = {
        'verified': False,
        'krs_found': False,
        'nip_match': None,
        'regon_match': None,
        'krs_data': None,
        'errors': [],
        'timestamp': datetime.now().isoformat(),
    }

    krs_data = get_company_from_krs(company_krs)

    if krs_data is None:
        result['errors'].append(f"Nie znaleziono podmiotu o KRS {company_krs}")
        return result

    result['krs_found'] = True
    result['krs_data'] = krs_data.to_dict()

    # Verify NIP if provided
    if company_nip:
        krs_nip = krs_data.nip.replace('-', '').replace(' ', '') if krs_data.nip else ''
        expected_nip = company_nip.replace('-', '').replace(' ', '')
        result['nip_match'] = krs_nip == expected_nip
        if not result['nip_match']:
            result['errors'].append(f"NIP niezgodny: oczekiwano {expected_nip}, w KRS: {krs_nip}")

    # Verify REGON if provided
    if company_regon:
        krs_regon = krs_data.regon[:9] if krs_data.regon else ''  # Use first 9 digits
        expected_regon = company_regon[:9].replace('-', '').replace(' ', '')
        result['regon_match'] = krs_regon == expected_regon
        if not result['regon_match']:
            result['errors'].append(f"REGON niezgodny: oczekiwano {expected_regon}, w KRS: {krs_regon}")

    # Overall verification
    result['verified'] = result['krs_found'] and len(result['errors']) == 0

    return result


def format_address(krs_data: KRSCompanyData) -> str:
    """Format address from KRS data."""
    parts = []

    if krs_data.ulica:
        addr = krs_data.ulica
        if krs_data.nr_domu:
            addr += f" {krs_data.nr_domu}"
        if krs_data.nr_lokalu:
            addr += f"/{krs_data.nr_lokalu}"
        parts.append(addr)

    if krs_data.kod_pocztowy and krs_data.miejscowosc:
        parts.append(f"{krs_data.kod_pocztowy} {krs_data.miejscowosc}")
    elif krs_data.miejscowosc:
        parts.append(krs_data.miejscowosc)

    return ', '.join(parts)


# CLI for testing
if __name__ == '__main__':
    import sys
    import json

    if len(sys.argv) < 2:
        print("Usage: python krs_api_service.py <KRS_NUMBER>")
        print("Example: python krs_api_service.py 0000817317")
        sys.exit(1)

    krs = sys.argv[1]
    print(f"Pobieranie danych z KRS Open API dla: {krs}")
    print("=" * 60)

    data = get_company_from_krs(krs)

    if data:
        print(f"Nazwa: {data.nazwa}")
        print(f"NIP: {data.nip}")
        print(f"REGON: {data.regon}")
        print(f"Forma prawna: {data.forma_prawna}")
        print(f"Adres: {format_address(data)}")
        print(f"Kapitał zakładowy: {data.kapital_zakladowy} {data.kapital_waluta}")
        print(f"Data rejestracji: {data.data_rejestracji}")
        print(f"Ostatni wpis: {data.data_ostatniego_wpisu} (nr {data.numer_ostatniego_wpisu})")
        print()
        print("Zarząd (dane zanonimizowane w Open API):")
        for osoba in data.zarzad:
            print(f"  - {osoba['imie']} {osoba['nazwisko']} - {osoba['funkcja']}")
        print()
        print("Wspólnicy (dane zanonimizowane w Open API):")
        for w in data.wspolnicy:
            print(f"  - {w['imie']} {w['nazwisko']}: {w['udzialy']}")
        print()
        print(f"Stan z dnia: {data.stan_z_dnia}")
        print(f"Data odpisu: {data.data_odpisu}")
    else:
        print(f"Nie znaleziono podmiotu o KRS {krs}")