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

fix: use ceidg_api_service in arm_company.py instead of manual API call
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 
The script was calling /firma?nip=X (wrong endpoint) instead of using
fetch_ceidg_by_nip() which does two-phase /firmy?nip=X then /firma/{id}.
Now uses the same service and field mapping as the admin panel button.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Maciej Pienczyn 2026-02-20 17:08:26 +01:00
parent c86d9cf368
commit 44d8ab932b
1 changed files with 98 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

143
scripts/arm_company.py
View File

@ -74,55 +74,108 @@ def arm_company(company_id, force=False):
results['registry'] = 'FAIL (KRS)' results['registry'] = 'FAIL (KRS)'
print(" -> FAIL: Nie udało się pobrać z KRS") print(" -> FAIL: Nie udało się pobrać z KRS")
else: else:
# Próbuj CEIDG # Próbuj CEIDG — używamy tego samego serwisu co API endpoint
import requests as req from ceidg_api_service import fetch_ceidg_by_nip
ceidg_key = os.getenv("CEIDG_API_KEY", "") from datetime import datetime, date as date_type
nip_clean = company.nip.replace('-', '').replace(' ', '') ceidg_data = fetch_ceidg_by_nip(company.nip)
ceidg_url = "https://dane.biznes.gov.pl/api/ceidg/v3/firma?nip=%s" % nip_clean if ceidg_data:
headers = {"Authorization": "Bearer %s" % ceidg_key} # CEIDG identifiers & metadata
resp = req.get(ceidg_url, headers=headers, timeout=10) if ceidg_data.get('ceidg_id'):
if resp.status_code == 200: company.ceidg_id = ceidg_data['ceidg_id']
data = resp.json() if ceidg_data.get('status'):
firmy = data.get('firmy', []) company.ceidg_status = ceidg_data['status']
if firmy: company.ceidg_raw_data = ceidg_data.get('raw')
firma = firmy[0] company.ceidg_fetched_at = datetime.now()
# Zapisz surowe dane company.data_source = 'CEIDG API'
company.ceidg_raw_data = firma company.last_verified_at = datetime.now()
from datetime import datetime
company.ceidg_fetched_at = datetime.utcnow()
# Mapuj podstawowe pola # Owner
if firma.get('wlasciciel'): wlasciciel = ceidg_data.get('wlasciciel', {})
wl = firma['wlasciciel'] if wlasciciel.get('imie'):
company.legal_form = 'JDG' company.owner_first_name = wlasciciel['imie']
if not company.owner_name: if wlasciciel.get('nazwisko'):
company.owner_name = '%s %s' % (wl.get('imie', ''), wl.get('nazwisko', '')) company.owner_last_name = wlasciciel['nazwisko']
if ceidg_data.get('obywatelstwa'):
company.owner_citizenships = ceidg_data['obywatelstwa']
if firma.get('adresDzialalnosci'): # Legal name
addr = firma['adresDzialalnosci'] if ceidg_data.get('firma') and (not company.legal_name or company.legal_name == company.name):
if not company.address_street: company.legal_name = ceidg_data['firma']
ulica = addr.get('ulica', '')
nr = addr.get('budynek', '')
lokal = addr.get('lokal', '')
company.address_street = ('%s %s%s' % (ulica, nr, '/%s' % lokal if lokal else '')).strip()
if not company.address_city:
company.address_city = addr.get('miasto', '')
if not company.address_zip:
company.address_zip = addr.get('kodPocztowy', '')
if firma.get('email') and not company.email: # REGON
company.email = firma['email'] if not company.regon:
if firma.get('telefon') and not company.phone: regon = ceidg_data.get('regon') or wlasciciel.get('regon')
company.phone = firma['telefon'] if regon:
if firma.get('adresStronyInternetowej') and not company.website: company.regon = regon
company.website = firma['adresStronyInternetowej']
db.commit() # Business start date
results['registry'] = 'OK (CEIDG)' if ceidg_data.get('dataRozpoczecia'):
print(" -> OK: Dane z CEIDG pobrane") try:
else: d = ceidg_data['dataRozpoczecia']
results['registry'] = 'NOT FOUND' if isinstance(d, str):
print(" -> Nie znaleziono w CEIDG") company.business_start_date = date_type.fromisoformat(d)
except (ValueError, TypeError):
pass
# Legal form
if not company.legal_form:
company.legal_form = 'JEDNOOSOBOWA DZIAŁALNOŚĆ GOSPODARCZA'
# 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')
# PKD (full list)
pkd_lista = ceidg_data.get('pkd', [])
if pkd_lista:
company.ceidg_pkd_list = pkd_lista
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)
))
# 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)
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 getattr(company, 'address_postal', None) and company.address_city:
company.address_full = '%s, %s %s' % (company.address_street, company.address_postal, company.address_city)
# Contact (only if empty)
if ceidg_data.get('email') and not company.email:
company.email = ceidg_data['email']
if ceidg_data.get('stronaWWW') and not company.website:
company.website = ceidg_data['stronaWWW']
if ceidg_data.get('telefon') and not company.phone:
company.phone = ceidg_data['telefon']
db.commit()
results['registry'] = 'OK (CEIDG)'
print(" -> OK: Dane z CEIDG pobrane")
else: else:
results['registry'] = 'NOT FOUND' results['registry'] = 'NOT FOUND'
print(" -> Nie znaleziono w żadnym rejestrze") print(" -> Nie znaleziono w żadnym rejestrze")