maciejpi/nordabiz
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix: reverse registry lookup order — CEIDG first, then KRS
Some checks are pending
NordaBiz Tests / Unit & Integration Tests (push) Waiting to run
Details
NordaBiz Tests / E2E Tests (Playwright) (push) Blocked by required conditions
Details
NordaBiz Tests / Smoke Tests (Production) (push) Blocked by required conditions
Details
NordaBiz Tests / Send Failure Notification (push) Blocked by required conditions
Details

Browse Source 
When fetching company data by NIP, check CEIDG first (JDG) before
trying Biała Lista/KRS. This prevents JDG companies from being
overwritten with spółka data. Also auto-clears KRS fields when
CEIDG confirms the company is a JDG.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Maciej Pienczyn 2026-04-08 17:51:12 +02:00
parent fb6edadbfe
commit 8c5d334b21
1 changed files with 176 additions and 157 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

333
blueprints/api/routes_company.py
View File

@ -415,172 +415,191 @@ def api_enrich_company_registry(company_id):
'error': f'Nie udało się pobrać danych z KRS dla numeru {company.krs}' 'error': f'Nie udało się pobrać danych z KRS dla numeru {company.krs}'
}), 404 }), 404
# Strategy 2: No KRS but has NIP — try Biała Lista to find KRS # Strategy 2: No KRS but has NIP — try CEIDG first (JDG), then Biała Lista/KRS
elif effective_nip: elif effective_nip:
krs_service = krs_api_service.KRSApiService() # Try CEIDG first — if the NIP belongs to a JDG, this will find it
biala_lista_result = krs_service.search_by_nip(effective_nip) ceidg_data = fetch_ceidg_by_nip(effective_nip)
if ceidg_data:
source = 'CEIDG'
updated_fields = []
if biala_lista_result and biala_lista_result.get('krs'): # --- CEIDG identifiers & metadata ---
# Found KRS via Biała Lista — save it and enrich if ceidg_data.get('ceidg_id'):
company.krs = biala_lista_result['krs'] company.ceidg_id = ceidg_data['ceidg_id']
db.flush() updated_fields.append('CEIDG ID')
if ceidg_data.get('status'):
company.ceidg_status = ceidg_data['status']
updated_fields.append('status CEIDG')
company.ceidg_raw_data = ceidg_data.get('raw')
company.ceidg_fetched_at = datetime.now()
company.data_source = 'CEIDG API'
company.last_verified_at = datetime.now()
from blueprints.admin.routes_membership import _enrich_company_from_krs # --- Owner ---
success = _enrich_company_from_krs(company, db) wlasciciel = ceidg_data.get('wlasciciel', {})
if success: if wlasciciel.get('imie'):
source = 'KRS (via Biała Lista VAT)' company.owner_first_name = wlasciciel['imie']
people_count = db.query(CompanyPerson).filter_by(company_id=company.id).count() updated_fields.append('właściciel')
pkd_count = db.query(CompanyPKD).filter_by(company_id=company.id).count() if wlasciciel.get('nazwisko'):
details = { company.owner_last_name = wlasciciel['nazwisko']
'krs_found': company.krs, if ceidg_data.get('obywatelstwa'):
'people_imported': people_count, company.owner_citizenships = ceidg_data['obywatelstwa']
'pkd_codes': pkd_count,
'legal_form': company.legal_form or '' # --- Legal name ---
} if ceidg_data.get('firma') and (not company.legal_name or company.legal_name == company.name):
message_parts.append(f'znaleziono KRS: {company.krs}') company.legal_name = ceidg_data['firma']
message_parts.append(f'zarząd ({people_count} osób)') updated_fields.append('nazwa pełna')
message_parts.append(f'kody PKD ({pkd_count})')
else: # --- REGON ---
source = 'Biała Lista VAT' if not company.regon:
details = {'krs_found': company.krs} regon = ceidg_data.get('regon') or wlasciciel.get('regon')
message_parts.append(f'znaleziono KRS: {company.krs} (dane KRS niedostępne)') if regon:
company.regon = regon
updated_fields.append('REGON')
# --- Business start date ---
if ceidg_data.get('dataRozpoczecia'):
try:
d = ceidg_data['dataRozpoczecia']
if isinstance(d, str):
company.business_start_date = date.fromisoformat(d)
updated_fields.append('data rozpoczęcia')
except (ValueError, TypeError):
pass
# --- Legal form (set to JDG since found in CEIDG) ---
if company.legal_form != 'JEDNOOSOBOWA DZIAŁALNOŚĆ GOSPODARCZA':
company.legal_form = 'JEDNOOSOBOWA DZIAŁALNOŚĆ GOSPODARCZA'
updated_fields.append('forma prawna → JDG')
# --- Clear KRS data (CEIDG = JDG, not a spółka) ---
if company.krs:
company.krs = None
company.krs_registration_date = None
company.krs_company_agreement_date = None
company.krs_representation_rules = None
company.krs_duration = None
company.krs_raw_data = None
company.krs_fetched_at = None
company.krs_representation = None
company.krs_activities = None
company.krs_last_audit_at = None
company.krs_pdf_path = None
company.capital_amount = None
company.capital_shares_count = None
company.capital_share_value = None
company.capital_currency = None
updated_fields.append('wyczyszczono dane KRS (firma jest JDG)')
# --- PKD (main) ---
pkd_gl = ceidg_data.get('pkdGlowny', {})
if pkd_gl and pkd_gl.get('kod'):
company.pkd_code = pkd_gl['kod']
company.pkd_description = pkd_gl.get('nazwa')
updated_fields.append(f'PKD główny ({pkd_gl["kod"]})')
# --- PKD (full list) ---
pkd_lista = ceidg_data.get('pkd', [])
if pkd_lista:
company.ceidg_pkd_list = pkd_lista
# Create CompanyPKD records (same pattern as KRS enrichment)
pkd_main_code = pkd_gl.get('kod', '') if pkd_gl else ''
for pkd_item in pkd_lista:
kod = pkd_item.get('kod', '')
if not kod:
continue
existing_pkd = db.query(CompanyPKD).filter(
CompanyPKD.company_id == company.id,
CompanyPKD.pkd_code == kod
).first()
if not existing_pkd:
db.add(CompanyPKD(
company_id=company.id,
pkd_code=kod,
pkd_description=pkd_item.get('nazwa', ''),
is_primary=(kod == pkd_main_code)
))
updated_fields.append(f'lista PKD ({len(pkd_lista)} kodów)')
# --- Business address ---
adres = ceidg_data.get('adresDzialalnosci', {})
ulica = adres.get('ulica', '')
budynek = adres.get('budynek', '')
lokal = adres.get('lokal', '')
if ulica or budynek:
street_parts = [ulica, budynek]
if lokal:
street_parts[-1] = (budynek + '/' + lokal) if budynek else lokal
company.address_street = ' '.join(p for p in street_parts if p)
updated_fields.append('adres')
if adres.get('kod') or adres.get('kodPocztowy'):
company.address_postal = adres.get('kod') or adres.get('kodPocztowy')
if adres.get('miasto') or adres.get('miejscowosc'):
company.address_city = adres.get('miasto') or adres.get('miejscowosc')
if company.address_street and company.address_postal and company.address_city:
company.address_full = f'{company.address_street}, {company.address_postal} {company.address_city}'
# --- Correspondence address ---
koresp = ceidg_data.get('adresKorespondencyjny', {})
if koresp and (koresp.get('ulica') or koresp.get('budynek')):
k_ulica = koresp.get('ulica', '')
k_budynek = koresp.get('budynek', '')
k_lokal = koresp.get('lokal', '')
k_parts = [k_ulica, k_budynek]
if k_lokal:
k_parts[-1] = (k_budynek + '/' + k_lokal) if k_budynek else k_lokal
company.correspondence_street = ' '.join(p for p in k_parts if p)
company.correspondence_postal = koresp.get('kod')
company.correspondence_city = koresp.get('miasto')
updated_fields.append('adres korespondencyjny')
# --- Contact (only if empty) ---
if ceidg_data.get('email') and not company.email:
company.email = ceidg_data['email']
updated_fields.append('email')
if ceidg_data.get('stronaWWW') and not company.website:
company.website = ceidg_data['stronaWWW']
updated_fields.append('strona www')
if ceidg_data.get('telefon') and not company.phone:
company.phone = ceidg_data['telefon']
updated_fields.append('telefon')
details = {'updated_fields': updated_fields}
message_parts.append(f'zaktualizowano {len(updated_fields)} pól')
if updated_fields:
message_parts.append(', '.join(updated_fields))
else: else:
# No KRS found — try CEIDG (likely JDG) # Not found in CEIDG — try Biała Lista to find KRS (spółka)
ceidg_data = fetch_ceidg_by_nip(effective_nip) krs_service = krs_api_service.KRSApiService()
if ceidg_data: biala_lista_result = krs_service.search_by_nip(effective_nip)
source = 'CEIDG'
updated_fields = []
# --- CEIDG identifiers & metadata --- if biala_lista_result and biala_lista_result.get('krs'):
if ceidg_data.get('ceidg_id'): company.krs = biala_lista_result['krs']
company.ceidg_id = ceidg_data['ceidg_id'] db.flush()
updated_fields.append('CEIDG ID')
if ceidg_data.get('status'):
company.ceidg_status = ceidg_data['status']
updated_fields.append('status CEIDG')
company.ceidg_raw_data = ceidg_data.get('raw')
company.ceidg_fetched_at = datetime.now()
company.data_source = 'CEIDG API'
company.last_verified_at = datetime.now()
# --- Owner --- from blueprints.admin.routes_membership import _enrich_company_from_krs
wlasciciel = ceidg_data.get('wlasciciel', {}) success = _enrich_company_from_krs(company, db)
if wlasciciel.get('imie'): if success:
company.owner_first_name = wlasciciel['imie'] source = 'KRS (via Biała Lista VAT)'
updated_fields.append('właściciel') people_count = db.query(CompanyPerson).filter_by(company_id=company.id).count()
if wlasciciel.get('nazwisko'): pkd_count = db.query(CompanyPKD).filter_by(company_id=company.id).count()
company.owner_last_name = wlasciciel['nazwisko'] details = {
if ceidg_data.get('obywatelstwa'): 'krs_found': company.krs,
company.owner_citizenships = ceidg_data['obywatelstwa'] 'people_imported': people_count,
'pkd_codes': pkd_count,
# --- Legal name --- 'legal_form': company.legal_form or ''
if ceidg_data.get('firma') and (not company.legal_name or company.legal_name == company.name): }
company.legal_name = ceidg_data['firma'] message_parts.append(f'znaleziono KRS: {company.krs}')
updated_fields.append('nazwa pełna') message_parts.append(f'zarząd ({people_count} osób)')
message_parts.append(f'kody PKD ({pkd_count})')
# --- REGON --- else:
if not company.regon: source = 'Biała Lista VAT'
regon = ceidg_data.get('regon') or wlasciciel.get('regon') details = {'krs_found': company.krs}
if regon: message_parts.append(f'znaleziono KRS: {company.krs} (dane KRS niedostępne)')
company.regon = regon
updated_fields.append('REGON')
# --- Business start date ---
if ceidg_data.get('dataRozpoczecia'):
try:
d = ceidg_data['dataRozpoczecia']
if isinstance(d, str):
company.business_start_date = date.fromisoformat(d)
updated_fields.append('data rozpoczęcia')
except (ValueError, TypeError):
pass
# --- Legal form ---
if not company.legal_form:
company.legal_form = 'JEDNOOSOBOWA DZIAŁALNOŚĆ GOSPODARCZA'
updated_fields.append('forma prawna')
# --- PKD (main) ---
pkd_gl = ceidg_data.get('pkdGlowny', {})
if pkd_gl and pkd_gl.get('kod'):
company.pkd_code = pkd_gl['kod']
company.pkd_description = pkd_gl.get('nazwa')
updated_fields.append(f'PKD główny ({pkd_gl["kod"]})')
# --- PKD (full list) ---
pkd_lista = ceidg_data.get('pkd', [])
if pkd_lista:
company.ceidg_pkd_list = pkd_lista
# Create CompanyPKD records (same pattern as KRS enrichment)
pkd_main_code = pkd_gl.get('kod', '') if pkd_gl else ''
for pkd_item in pkd_lista:
kod = pkd_item.get('kod', '')
if not kod:
continue
existing_pkd = db.query(CompanyPKD).filter(
CompanyPKD.company_id == company.id,
CompanyPKD.pkd_code == kod
).first()
if not existing_pkd:
db.add(CompanyPKD(
company_id=company.id,
pkd_code=kod,
pkd_description=pkd_item.get('nazwa', ''),
is_primary=(kod == pkd_main_code)
))
updated_fields.append(f'lista PKD ({len(pkd_lista)} kodów)')
# --- Business address ---
adres = ceidg_data.get('adresDzialalnosci', {})
ulica = adres.get('ulica', '')
budynek = adres.get('budynek', '')
lokal = adres.get('lokal', '')
if ulica or budynek:
street_parts = [ulica, budynek]
if lokal:
street_parts[-1] = (budynek + '/' + lokal) if budynek else lokal
company.address_street = ' '.join(p for p in street_parts if p)
updated_fields.append('adres')
if adres.get('kod') or adres.get('kodPocztowy'):
company.address_postal = adres.get('kod') or adres.get('kodPocztowy')
if adres.get('miasto') or adres.get('miejscowosc'):
company.address_city = adres.get('miasto') or adres.get('miejscowosc')
if company.address_street and company.address_postal and company.address_city:
company.address_full = f'{company.address_street}, {company.address_postal} {company.address_city}'
# --- Correspondence address ---
koresp = ceidg_data.get('adresKorespondencyjny', {})
if koresp and (koresp.get('ulica') or koresp.get('budynek')):
k_ulica = koresp.get('ulica', '')
k_budynek = koresp.get('budynek', '')
k_lokal = koresp.get('lokal', '')
k_parts = [k_ulica, k_budynek]
if k_lokal:
k_parts[-1] = (k_budynek + '/' + k_lokal) if k_budynek else k_lokal
company.correspondence_street = ' '.join(p for p in k_parts if p)
company.correspondence_postal = koresp.get('kod')
company.correspondence_city = koresp.get('miasto')
updated_fields.append('adres korespondencyjny')
# --- Contact (only if empty) ---
if ceidg_data.get('email') and not company.email:
company.email = ceidg_data['email']
updated_fields.append('email')
if ceidg_data.get('stronaWWW') and not company.website:
company.website = ceidg_data['stronaWWW']
updated_fields.append('strona www')
if ceidg_data.get('telefon') and not company.phone:
company.phone = ceidg_data['telefon']
updated_fields.append('telefon')
details = {'updated_fields': updated_fields}
message_parts.append(f'zaktualizowano {len(updated_fields)} pól')
if updated_fields:
message_parts.append(', '.join(updated_fields))
else: else:
return jsonify({ return jsonify({
'success': False, 'success': False,
'error': 'Nie znaleziono danych w rejestrach KRS ani CEIDG dla tego NIP' 'error': 'Nie znaleziono danych w rejestrach CEIDG ani KRS dla tego NIP'
}), 404 }), 404
else: else:
return jsonify({ return jsonify({