fix: correct service API calls in arm_company.py
Some checks are pending
NordaBiz Tests / Unit & Integration Tests (push) Waiting to run
NordaBiz Tests / E2E Tests (Playwright) (push) Blocked by required conditions
NordaBiz Tests / Smoke Tests (Production) (push) Blocked by required conditions
NordaBiz Tests / Send Failure Notification (push) Blocked by required conditions
Some checks are pending
NordaBiz Tests / Unit & Integration Tests (push) Waiting to run
NordaBiz Tests / E2E Tests (Playwright) (push) Blocked by required conditions
NordaBiz Tests / Smoke Tests (Production) (push) Blocked by required conditions
NordaBiz Tests / Send Failure Notification (push) Blocked by required conditions
- KRS: search_by_nip() returns dict, not just KRS number - SEO: SEOAuditor(database_url) + audit_company(company_dict) - Social: SocialMediaAuditor() + audit_company(company_dict) - GBP: GBPAuditService(db) + audit_company(company_id) - Support multiple company IDs in one invocation Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
1967fe8695
commit
0e04a17785
@ -5,6 +5,7 @@ Odpowiednik przycisku "Uzbrój firmę" w panelu admina.
|
||||
|
||||
Użycie:
|
||||
python3 scripts/arm_company.py <company_id> [--force]
|
||||
python3 scripts/arm_company.py 120 121 122 --force # wiele firm naraz
|
||||
|
||||
Opcje:
|
||||
--force Wymusza ponowne wykonanie wszystkich kroków (jak "Zaktualizuj dane")
|
||||
@ -14,7 +15,12 @@ import os
|
||||
import logging
|
||||
|
||||
# Setup path
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
sys.path.insert(0, BASE_DIR)
|
||||
|
||||
scripts_dir = os.path.join(BASE_DIR, 'scripts')
|
||||
if scripts_dir not in sys.path:
|
||||
sys.path.insert(0, scripts_dir)
|
||||
|
||||
from database import SessionLocal, Company, CompanyWebsiteAnalysis, CompanySocialMedia, CompanyPKD, CompanyPerson
|
||||
from database import GBPAudit
|
||||
@ -50,13 +56,13 @@ def arm_company(company_id, force=False):
|
||||
|
||||
krs_service = KRSApiService()
|
||||
|
||||
# Sprawdź KRS przez Białą Listę
|
||||
# Sprawdź KRS przez Białą Listę (search_by_nip zwraca dict lub None)
|
||||
if not company.krs:
|
||||
krs_number = krs_service.get_krs_by_nip(company.nip)
|
||||
if krs_number:
|
||||
company.krs = krs_number
|
||||
krs_data = krs_service.search_by_nip(company.nip)
|
||||
if krs_data and krs_data.get('krs'):
|
||||
company.krs = krs_data['krs']
|
||||
db.flush()
|
||||
logger.info("Znaleziono KRS %s dla NIP %s" % (krs_number, company.nip))
|
||||
logger.info("Znaleziono KRS %s dla NIP %s" % (company.krs, company.nip))
|
||||
|
||||
if company.krs:
|
||||
success = _enrich_company_from_krs(company, db)
|
||||
@ -69,28 +75,57 @@ def arm_company(company_id, force=False):
|
||||
print(" -> FAIL: Nie udało się pobrać z KRS")
|
||||
else:
|
||||
# Próbuj CEIDG
|
||||
try:
|
||||
from blueprints.api.routes_company import _enrich_from_ceidg
|
||||
ceidg_ok = _enrich_from_ceidg(company, db)
|
||||
if ceidg_ok:
|
||||
import requests as req
|
||||
ceidg_key = os.getenv("CEIDG_API_KEY", "")
|
||||
nip_clean = company.nip.replace('-', '').replace(' ', '')
|
||||
ceidg_url = "https://dane.biznes.gov.pl/api/ceidg/v3/firma?nip=%s" % nip_clean
|
||||
headers = {"Authorization": "Bearer %s" % ceidg_key}
|
||||
resp = req.get(ceidg_url, headers=headers, timeout=10)
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
firmy = data.get('firmy', [])
|
||||
if firmy:
|
||||
firma = firmy[0]
|
||||
# Zapisz surowe dane
|
||||
company.ceidg_raw_data = firma
|
||||
from datetime import datetime
|
||||
company.ceidg_fetched_at = datetime.utcnow()
|
||||
|
||||
# Mapuj podstawowe pola
|
||||
if firma.get('wlasciciel'):
|
||||
wl = firma['wlasciciel']
|
||||
company.legal_form = 'JDG'
|
||||
if not company.owner_name:
|
||||
company.owner_name = '%s %s' % (wl.get('imie', ''), wl.get('nazwisko', ''))
|
||||
|
||||
if firma.get('adresDzialalnosci'):
|
||||
addr = firma['adresDzialalnosci']
|
||||
if not company.address_street:
|
||||
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:
|
||||
company.email = firma['email']
|
||||
if firma.get('telefon') and not company.phone:
|
||||
company.phone = firma['telefon']
|
||||
if firma.get('adresStronyInternetowej') and not company.website:
|
||||
company.website = firma['adresStronyInternetowej']
|
||||
|
||||
db.commit()
|
||||
results['registry'] = 'OK (CEIDG)'
|
||||
print(" -> OK: Dane z CEIDG pobrane")
|
||||
else:
|
||||
results['registry'] = 'FAIL (CEIDG)'
|
||||
print(" -> FAIL: Nie znaleziono w CEIDG")
|
||||
except ImportError:
|
||||
# Fallback: bezpośrednie wywołanie CEIDG API
|
||||
import requests as req
|
||||
ceidg_url = "https://dane.biznes.gov.pl/api/ceidg/v3/firma?nip=%s" % company.nip
|
||||
headers = {"Authorization": "Bearer %s" % os.getenv("CEIDG_API_KEY", "")}
|
||||
resp = req.get(ceidg_url, headers=headers, timeout=10)
|
||||
if resp.status_code == 200 and resp.json().get('firmy'):
|
||||
results['registry'] = 'OK (CEIDG-raw)'
|
||||
print(" -> OK: Dane z CEIDG (raw)")
|
||||
else:
|
||||
results['registry'] = 'NOT FOUND'
|
||||
print(" -> Nie znaleziono w żadnym rejestrze")
|
||||
print(" -> Nie znaleziono w CEIDG")
|
||||
else:
|
||||
results['registry'] = 'NOT FOUND'
|
||||
print(" -> Nie znaleziono w żadnym rejestrze")
|
||||
except Exception as e:
|
||||
results['registry'] = 'ERROR: %s' % str(e)[:80]
|
||||
print(" -> ERROR: %s" % str(e)[:80])
|
||||
@ -110,18 +145,20 @@ def arm_company(company_id, force=False):
|
||||
if company.website:
|
||||
print("\n[2/5] Audyt SEO...")
|
||||
try:
|
||||
scripts_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'scripts')
|
||||
if scripts_dir not in sys.path:
|
||||
sys.path.insert(0, scripts_dir)
|
||||
from seo_audit import SEOAuditService
|
||||
seo_service = SEOAuditService()
|
||||
result = seo_service.audit_company(company.id, db)
|
||||
if result and result.get('success'):
|
||||
results['seo'] = 'OK (score: %s)' % result.get('seo_score', '?')
|
||||
print(" -> OK: SEO=%s, Perf=%s" % (result.get('seo_score', '?'), result.get('performance_score', '?')))
|
||||
else:
|
||||
results['seo'] = 'FAIL'
|
||||
print(" -> FAIL: %s" % (result.get('error', 'unknown') if result else 'no result'))
|
||||
from seo_audit import SEOAuditor
|
||||
seo_service = SEOAuditor()
|
||||
company_dict = {
|
||||
'id': company.id,
|
||||
'name': company.name,
|
||||
'slug': company.slug,
|
||||
'website': company.website,
|
||||
'address_city': company.address_city or '',
|
||||
}
|
||||
audit_result = seo_service.audit_company(company_dict)
|
||||
seo_score = audit_result.get('scores', {}).get('pagespeed_seo', '?')
|
||||
perf_score = audit_result.get('scores', {}).get('pagespeed_performance', '?')
|
||||
results['seo'] = 'OK (SEO: %s, Perf: %s)' % (seo_score, perf_score)
|
||||
print(" -> OK: SEO=%s, Perf=%s" % (seo_score, perf_score))
|
||||
except Exception as e:
|
||||
results['seo'] = 'ERROR: %s' % str(e)[:80]
|
||||
print(" -> ERROR: %s" % str(e)[:80])
|
||||
@ -137,13 +174,17 @@ def arm_company(company_id, force=False):
|
||||
if force or not social_done:
|
||||
print("\n[3/5] Audyt Social Media...")
|
||||
try:
|
||||
scripts_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'scripts')
|
||||
if scripts_dir not in sys.path:
|
||||
sys.path.insert(0, scripts_dir)
|
||||
from social_media_audit import SocialMediaAuditor
|
||||
auditor = SocialMediaAuditor(db)
|
||||
audit_result = auditor.audit_company(company.id)
|
||||
profiles = audit_result.get('profiles_found', 0) if audit_result else 0
|
||||
auditor = SocialMediaAuditor() # uses DATABASE_URL from env
|
||||
company_dict = {
|
||||
'id': company.id,
|
||||
'name': company.name,
|
||||
'slug': company.slug,
|
||||
'website': company.website,
|
||||
'address_city': company.address_city or '',
|
||||
}
|
||||
audit_result = auditor.audit_company(company_dict)
|
||||
profiles = len(audit_result.get('social_media', {}).get('profiles', [])) if audit_result else 0
|
||||
results['social'] = 'OK (%d profili)' % profiles
|
||||
print(" -> OK: %d profili znalezionych" % profiles)
|
||||
except Exception as e:
|
||||
@ -159,15 +200,15 @@ def arm_company(company_id, force=False):
|
||||
print("\n[4/5] Audyt GBP...")
|
||||
try:
|
||||
from gbp_audit_service import GBPAuditService
|
||||
gbp_service = GBPAuditService()
|
||||
gbp_result = gbp_service.audit_company(company.id, db)
|
||||
if gbp_result and gbp_result.get('success'):
|
||||
score = gbp_result.get('completeness_score', '?')
|
||||
gbp_service = GBPAuditService(db)
|
||||
gbp_result = gbp_service.audit_company(company.id)
|
||||
if gbp_result:
|
||||
score = gbp_result.get('completeness_score', gbp_result.get('score', '?'))
|
||||
results['gbp'] = 'OK (score: %s)' % score
|
||||
print(" -> OK: Score=%s" % score)
|
||||
else:
|
||||
results['gbp'] = 'FAIL'
|
||||
print(" -> FAIL: %s" % (gbp_result.get('error', 'unknown') if gbp_result else 'no result'))
|
||||
print(" -> FAIL: brak wyniku")
|
||||
except Exception as e:
|
||||
results['gbp'] = 'ERROR: %s' % str(e)[:80]
|
||||
print(" -> ERROR: %s" % str(e)[:80])
|
||||
@ -235,11 +276,14 @@ def arm_company(company_id, force=False):
|
||||
|
||||
if __name__ == '__main__':
|
||||
if len(sys.argv) < 2:
|
||||
print("Użycie: python3 scripts/arm_company.py <company_id> [--force]")
|
||||
print("Użycie: python3 scripts/arm_company.py <company_id> [<id2> ...] [--force]")
|
||||
print(" --force Wymusza ponowne wykonanie wszystkich kroków")
|
||||
sys.exit(1)
|
||||
|
||||
cid = int(sys.argv[1])
|
||||
force = '--force' in sys.argv
|
||||
ids = [int(a) for a in sys.argv[1:] if a != '--force' and a.isdigit()]
|
||||
|
||||
arm_company(cid, force=force)
|
||||
for cid in ids:
|
||||
arm_company(cid, force=force)
|
||||
if len(ids) > 1:
|
||||
print("\n")
|
||||
|
||||
Loading…
Reference in New Issue
Block a user