137 lines
4.9 KiB
Python
137 lines
4.9 KiB
Python
"""
|
|
CVR.dk API service for looking up Danish company information
|
|
Free public API - no authentication required
|
|
Adapted from OmniSync for BMC Hub
|
|
"""
|
|
import asyncio
|
|
import aiohttp
|
|
import logging
|
|
from typing import Optional, Dict
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class CVRService:
|
|
"""Service for CVR.dk API lookups"""
|
|
|
|
BASE_URL = "https://cvrapi.dk/api"
|
|
|
|
async def lookup_by_name(self, company_name: str) -> Optional[Dict]:
|
|
"""
|
|
Lookup company by name using CVR.dk API
|
|
|
|
Args:
|
|
company_name: Company name to search for
|
|
|
|
Returns:
|
|
Company data dict or None if not found
|
|
"""
|
|
if not company_name or len(company_name) < 3:
|
|
return None
|
|
|
|
# Clean company name
|
|
clean_name = company_name.strip()
|
|
|
|
try:
|
|
params = {
|
|
'search': clean_name,
|
|
'country': 'dk'
|
|
}
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(
|
|
f"{self.BASE_URL}",
|
|
params=params,
|
|
timeout=aiohttp.ClientTimeout(total=10)
|
|
) as response:
|
|
if response.status == 200:
|
|
data = await response.json()
|
|
|
|
if data and 'vat' in data:
|
|
logger.info(f"✅ Found CVR {data['vat']} for '{company_name}'")
|
|
return {
|
|
'cvr': data.get('vat'),
|
|
'name': data.get('name'),
|
|
'address': data.get('address'),
|
|
'city': data.get('city'),
|
|
'zipcode': data.get('zipcode'),
|
|
'country': data.get('country'),
|
|
'phone': data.get('phone'),
|
|
'email': data.get('email'),
|
|
'vat': data.get('vat'),
|
|
'status': data.get('status')
|
|
}
|
|
|
|
elif response.status == 404:
|
|
logger.warning(f"⚠️ No CVR found for '{company_name}'")
|
|
return None
|
|
|
|
else:
|
|
logger.error(f"❌ CVR API error {response.status} for '{company_name}'")
|
|
return None
|
|
|
|
except asyncio.TimeoutError:
|
|
logger.error(f"⏱️ CVR API timeout for '{company_name}'")
|
|
return None
|
|
|
|
except Exception as e:
|
|
logger.error(f"❌ CVR lookup error for '{company_name}': {e}")
|
|
return None
|
|
|
|
async def lookup_by_cvr(self, cvr_number: str) -> Optional[Dict]:
|
|
"""
|
|
Lookup company by CVR number
|
|
|
|
Args:
|
|
cvr_number: CVR number (8 digits)
|
|
|
|
Returns:
|
|
Company data dict or None if not found
|
|
"""
|
|
if not cvr_number:
|
|
return None
|
|
|
|
# Extract only digits
|
|
cvr_clean = ''.join(filter(str.isdigit, str(cvr_number)))
|
|
|
|
if len(cvr_clean) != 8:
|
|
logger.warning(f"⚠️ Invalid CVR number format: {cvr_number}")
|
|
return None
|
|
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(
|
|
f"{self.BASE_URL}",
|
|
params={'vat': cvr_clean, 'country': 'dk'},
|
|
timeout=aiohttp.ClientTimeout(total=10)
|
|
) as response:
|
|
if response.status == 200:
|
|
data = await response.json()
|
|
|
|
if data and 'vat' in data:
|
|
logger.info(f"✅ Validated CVR {cvr_clean}")
|
|
return {
|
|
'cvr': data.get('vat'),
|
|
'name': data.get('name'),
|
|
'address': data.get('address'),
|
|
'city': data.get('city'),
|
|
'zipcode': data.get('zipcode'),
|
|
'postal_code': data.get('zipcode'), # Alias for consistency
|
|
'country': data.get('country'),
|
|
'phone': data.get('phone'),
|
|
'email': data.get('email'),
|
|
'vat': data.get('vat'),
|
|
'status': data.get('status')
|
|
}
|
|
|
|
return None
|
|
|
|
except Exception as e:
|
|
logger.error(f"❌ CVR validation error for {cvr_number}: {e}")
|
|
return None
|
|
|
|
|
|
def get_cvr_service() -> CVRService:
|
|
"""Get CVR service instance"""
|
|
return CVRService()
|