#!/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 ") 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}")