feat: Add NIP lookup services for membership application
- Add ceidg_api_service.py with fetch_ceidg_by_nip() function - Add KRSApiService class with search_by_nip() method - KRS lookup uses rejestr.io API (unofficial) or database fallback - CEIDG lookup uses official dane.biznes.gov.pl API Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
parent
3e23bd3e4e
commit
28affce99f
156
ceidg_api_service.py
Normal file
156
ceidg_api_service.py
Normal file
@ -0,0 +1,156 @@
|
||||
#!/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")
|
||||
@ -440,6 +440,121 @@ def format_address(krs_data: KRSCompanyData) -> str:
|
||||
return ', '.join(parts)
|
||||
|
||||
|
||||
# ============================================================
|
||||
# KRS API Service Class (for search_by_nip compatibility)
|
||||
# ============================================================
|
||||
|
||||
class KRSApiService:
|
||||
"""
|
||||
KRS API Service class providing unified interface for KRS lookups.
|
||||
|
||||
Note: KRS Open API doesn't support direct NIP lookup.
|
||||
This class uses rejestr.io unofficial API as a fallback.
|
||||
"""
|
||||
|
||||
REJESTR_IO_URL = "https://rejestr.io/api/v2/org"
|
||||
REJESTR_IO_TIMEOUT = 10
|
||||
|
||||
def search_by_nip(self, nip: str) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Search KRS by NIP number.
|
||||
|
||||
Since KRS Open API doesn't support NIP lookup, this method:
|
||||
1. First tries rejestr.io API (unofficial but reliable)
|
||||
2. Falls back to checking our database for KRS number
|
||||
3. Then fetches full data from KRS Open API
|
||||
|
||||
Args:
|
||||
nip: NIP number (10 digits)
|
||||
|
||||
Returns:
|
||||
Dictionary with company data or None if not found
|
||||
"""
|
||||
import logging
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Clean NIP
|
||||
nip = nip.strip().replace('-', '').replace(' ', '')
|
||||
if not nip or len(nip) != 10 or not nip.isdigit():
|
||||
return None
|
||||
|
||||
# Try rejestr.io first (supports NIP lookup)
|
||||
krs_number = self._get_krs_from_rejestr_io(nip)
|
||||
|
||||
if not krs_number:
|
||||
# Try 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
|
||||
|
||||
# Fetch full data from KRS Open API
|
||||
logger.info(f"Found KRS {krs_number} for NIP {nip}, fetching details")
|
||||
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_rejestr_io(self, nip: str) -> Optional[str]:
|
||||
"""Try to get KRS number from rejestr.io API."""
|
||||
import logging
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
try:
|
||||
# rejestr.io API endpoint for NIP lookup
|
||||
url = f"{self.REJESTR_IO_URL}"
|
||||
params = {"nip": nip}
|
||||
|
||||
response = requests.get(url, params=params, timeout=self.REJESTR_IO_TIMEOUT)
|
||||
|
||||
if response.status_code == 200:
|
||||
data = response.json()
|
||||
# Handle different response formats
|
||||
if isinstance(data, list) and data:
|
||||
krs = data[0].get('krs')
|
||||
if krs:
|
||||
return str(krs).zfill(10)
|
||||
elif isinstance(data, dict):
|
||||
krs = data.get('krs')
|
||||
if krs:
|
||||
return str(krs).zfill(10)
|
||||
|
||||
logger.debug(f"rejestr.io: No KRS found for NIP {nip}")
|
||||
return None
|
||||
|
||||
except Exception as e:
|
||||
logger.debug(f"rejestr.io lookup failed: {e}")
|
||||
return None
|
||||
|
||||
def _get_krs_from_database(self, nip: str) -> Optional[str]:
|
||||
"""Check our database for KRS number."""
|
||||
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
|
||||
|
||||
Loading…
Reference in New Issue
Block a user