nordabiz/krs_api_service.py
Maciej Pienczyn c73e90bc70 feat: Add Biała Lista VAT integration for NIP→KRS lookup
- Use official Ministry of Finance API (wl-api.mf.gov.pl) to get KRS from NIP
- Add KRS field to membership application form
- Workflow: NIP → Biała Lista → KRS Open API → full company data
- Fallback to CEIDG for JDG (sole proprietorship)
- Remove rejestr.io dependency - only official government APIs

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-02-01 14:32:36 +01:00

603 lines
20 KiB
Python

#!/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
poczta: Optional[str] # NOWE
# Contact from KRS (NOWE)
email_krs: Optional[str]
www_krs: Optional[str]
adres_epuap: Optional[str]
# Capital
kapital_zakladowy: Optional[float]
kapital_waluta: str
# Dates
data_rejestracji: Optional[str]
data_ostatniego_wpisu: Optional[str]
numer_ostatniego_wpisu: Optional[int]
# Company agreement/statute (NOWE)
data_umowy_spolki: Optional[str]
czas_trwania_spolki: Optional[str]
informacja_o_udzialach: Optional[str]
# Management (anonymized in Open API)
zarzad: List[Dict[str, str]]
nazwa_organu: Optional[str] # NOWE
sposob_reprezentacji: Optional[str]
# Shareholders (anonymized in Open API)
wspolnicy: List[Dict[str, Any]]
# Other
przedmiot_dzialalnosci: List[Dict[str, str]] # ZMIANA: teraz słownik z pełnymi kodami
czy_opp: bool
# Financial statements (NOWE)
sprawozdania_finansowe: List[Dict[str, str]]
# Court/Registry info (NOWE)
sygnatura_akt: Optional[str]
sad_rejestrowy: Optional[str]
dzien_konczacy_rok_obrotowy: Optional[str]
# 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,
'poczta': self.poczta,
},
'kontakt_krs': {
'email': self.email_krs,
'www': self.www_krs,
'adres_epuap': self.adres_epuap,
},
'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,
},
'umowa_spolki': {
'data_umowy': self.data_umowy_spolki,
'czas_trwania': self.czas_trwania_spolki,
'informacja_o_udzialach': self.informacja_o_udzialach,
},
'zarzad': self.zarzad,
'nazwa_organu': self.nazwa_organu,
'sposob_reprezentacji': self.sposob_reprezentacji,
'wspolnicy': self.wspolnicy,
'przedmiot_dzialalnosci': self.przedmiot_dzialalnosci,
'czy_opp': self.czy_opp,
'sprawozdania_finansowe': self.sprawozdania_finansowe,
'rejestr': {
'sygnatura_akt': self.sygnatura_akt,
'sad_rejestrowy': self.sad_rejestrowy,
'dzien_konczacy_rok_obrotowy': self.dzien_konczacy_rok_obrotowy,
},
'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 and Contact
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', {})
# Company agreement/statute
umowa_statut = dzial1.get('umowaStatut', {})
pozostale = dzial1.get('pozostaleInformacje', {})
# Get first date of company agreement
data_umowy = None
umowy_list = umowa_statut.get('informacjaOZawarciuZmianieUmowyStatutu', [])
if umowy_list:
data_umowy = umowy_list[0].get('zawarcieZmianaUmowyStatutu')
# 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', ''),
'zawieszona': osoba.get('czyZawieszona', False),
})
# 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) - with full codes
przedmiot = []
przedmiot_dzial = dzial3.get('przedmiotDzialalnosci', {})
for pkd in przedmiot_dzial.get('przedmiotPrzewazajacejDzialalnosci', []):
kod_pelny = f"{pkd.get('kodDzial', '')}.{pkd.get('kodKlasa', '')}.{pkd.get('kodPodklasa', '')}"
przedmiot.append({
'kod': kod_pelny,
'opis': pkd.get('opis', ''),
'glowna': True,
})
for pkd in przedmiot_dzial.get('przedmiotPozostalejDzialalnosci', []):
kod_pelny = f"{pkd.get('kodDzial', '')}.{pkd.get('kodKlasa', '')}.{pkd.get('kodPodklasa', '')}"
przedmiot.append({
'kod': kod_pelny,
'opis': pkd.get('opis', ''),
'glowna': False,
})
# Financial statements
sprawozdania = []
wzmianki = dzial3.get('wzmiankiOZlozonychDokumentach', {})
for sf in wzmianki.get('wzmiankaOZlozeniuRocznegoSprawozdaniaFinansowego', []):
sprawozdania.append({
'data_zlozenia': sf.get('dataZlozenia', ''),
'za_okres': sf.get('zaOkresOdDo', ''),
})
# Fiscal year end
rok_obrotowy = dzial3.get('informacjaODniuKonczacymRokObrotowy', {})
dzien_konczacy = rok_obrotowy.get('dzienKonczacyPierwszyRokObrotowy')
# 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', ''),
# Address
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'),
poczta=adres.get('poczta'),
# Contact from KRS
email_krs=siedziba_adres.get('adresPocztyElektronicznej'),
www_krs=siedziba_adres.get('adresStronyInternetowej'),
adres_epuap=siedziba_adres.get('adresDoDoreczenElektronicznychWpisanyDoBAE'),
# Capital
kapital_zakladowy=kapital_value,
kapital_waluta=kapital_zakladowy.get('waluta', 'PLN'),
# Dates
data_rejestracji=naglowek.get('dataRejestracjiWKRS'),
data_ostatniego_wpisu=naglowek.get('dataOstatniegoWpisu'),
numer_ostatniego_wpisu=naglowek.get('numerOstatniegoWpisu'),
# Company agreement
data_umowy_spolki=data_umowy,
czas_trwania_spolki=pozostale.get('czasNaJakiUtworzonyZostalPodmiot'),
informacja_o_udzialach=pozostale.get('informacjaOLiczbieUdzialow'),
# Management
zarzad=zarzad,
nazwa_organu=reprezentacja.get('nazwaOrganu'),
sposob_reprezentacji=reprezentacja.get('sposobReprezentacji'),
# Shareholders
wspolnicy=wspolnicy,
# Business activities
przedmiot_dzialalnosci=przedmiot,
czy_opp=dane_podmiotu.get('czyPosiadaStatusOPP', False),
# Financial statements
sprawozdania_finansowe=sprawozdania,
# Court/Registry info
sygnatura_akt=naglowek.get('sygnaturaAktSprawyDotyczacejOstatniegoWpisu'),
sad_rejestrowy=naglowek.get('oznaczenieSaduDokonujacegoOstatniegoWpisu'),
dzien_konczacy_rok_obrotowy=dzien_konczacy,
# Metadata
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)
# ============================================================
# KRS API Service Class (for search_by_nip compatibility)
# ============================================================
# Biała Lista VAT API (Ministry of Finance) - returns KRS from NIP
BIALA_LISTA_API_URL = "https://wl-api.mf.gov.pl/api/search/nip"
BIALA_LISTA_TIMEOUT = 10
class KRSApiService:
"""
KRS API Service class providing unified interface for KRS lookups.
Uses official government APIs:
1. Biała Lista VAT (wl-api.mf.gov.pl) - to get KRS from NIP
2. KRS Open API (api-krs.ms.gov.pl) - to get full company data from KRS
"""
def search_by_nip(self, nip: str) -> Optional[Dict[str, Any]]:
"""
Search KRS by NIP number using official government APIs.
Workflow:
1. Query Biała Lista VAT API (Ministry of Finance) to get KRS from NIP
2. Fetch full data from KRS Open API using the KRS number
Args:
nip: NIP number (10 digits)
Returns:
Dictionary with company data or None if not found
"""
import logging
from datetime import date
logger = logging.getLogger(__name__)
# Clean NIP
nip = nip.strip().replace('-', '').replace(' ', '')
if not nip or len(nip) != 10 or not nip.isdigit():
return None
# Step 1: Get KRS from Biała Lista VAT API
krs_number = self._get_krs_from_biala_lista(nip)
if not krs_number:
# Fallback: check our database
krs_number = self._get_krs_from_database(nip)
if not krs_number:
logger.info(f"No KRS found for NIP {nip}")
return None
# Step 2: Fetch full data from official KRS Open API
logger.info(f"Found KRS {krs_number} for NIP {nip}, fetching from KRS Open API")
krs_data = get_company_from_krs(krs_number)
if not krs_data:
return None
# Return as dict with expected format
return krs_data.to_dict()
def _get_krs_from_biala_lista(self, nip: str) -> Optional[str]:
"""
Get KRS number from Biała Lista VAT API (Ministry of Finance).
Official API: https://wl-api.mf.gov.pl/
"""
import logging
from datetime import date
logger = logging.getLogger(__name__)
try:
today = date.today().strftime('%Y-%m-%d')
url = f"{BIALA_LISTA_API_URL}/{nip}?date={today}"
response = requests.get(url, timeout=BIALA_LISTA_TIMEOUT)
if response.status_code == 200:
data = response.json()
subject = data.get('result', {}).get('subject', {})
if subject:
krs = subject.get('krs')
if krs:
logger.info(f"Biała Lista: Found KRS {krs} for NIP {nip}")
return str(krs).zfill(10)
else:
logger.debug(f"Biała Lista: NIP {nip} found but no KRS (likely JDG)")
else:
logger.debug(f"Biała Lista: NIP {nip} not found")
return None
except Exception as e:
logger.warning(f"Biała Lista API error for NIP {nip}: {e}")
return None
def _get_krs_from_database(self, nip: str) -> Optional[str]:
"""Check our database for KRS number (fallback)."""
import logging
logger = logging.getLogger(__name__)
try:
from database import SessionLocal, Company
db = SessionLocal()
try:
company = db.query(Company).filter(Company.nip == nip).first()
if company and company.krs:
logger.debug(f"Found KRS {company.krs} in database for NIP {nip}")
return company.krs
finally:
db.close()
except Exception as e:
logger.debug(f"Database KRS lookup failed: {e}")
return None
# Singleton instance for import
krs_api_service = KRSApiService()
# 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}")