#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ cvr_nyeste_udtraek.py ===================== Trækker de NYESTE aktive ENKELTMANDSVIRKSOMHEDER i udvalgte brancher (som standard Region Nordjylland + Region Midtjylland) fra Erhvervsstyrelsens officielle CVR-distributions-API (Elasticsearch) og gemmer dem som CSV. Bygget til: at finde nystartede virksomheder der sandsynligvis mangler en hjemmeside — et godt salgsmål for en billig hjemmeside. ------------------------------------------------------------------------------- VIGTIGT OM COMPLIANCE (GDPR + markedsføringsloven) — læs før brug ------------------------------------------------------------------------------- * Ejeren af en enkeltmandsvirksomhed er en FYSISK PERSON. Navn, adresse, telefon og e-mail er persondata. Du bliver dataansvarlig for din brug. * Uanmodet elektronisk markedsføring (e-mail / SMS) kræver som udgangspunkt forudgående samtykke (markedsføringsloven § 10). Uanmodet telefonopkald til erhverv er som udgangspunkt tilladt, men respektér Robinsonlisten. * Dette script SPRINGER AUTOMATISK reklamebeskyttede virksomheder over (reklamebeskyttet = true). Data om reklamebeskyttede må IKKE bruges til opsøgende salg/markedsføring. * Overvej at bruge telefon/fysisk post frem for uanmodet e-mail. ------------------------------------------------------------------------------- SÅDAN FÅR DU ADGANG (gratis) ------------------------------------------------------------------------------- 1) Skriv til cvrselvbetjening@erst.dk og bed om "system-til-system adgang til CVR-data (Elasticsearch)". Du underskriver en erklæring om vilkår og får tilsendt et BRUGERNAVN og PASSWORD. 2) Sæt dem som miljøvariabler (så de ikke står i koden): macOS/Linux: export CVR_USER="dit-brugernavn" export CVR_PASS="dit-password" 3) Installér afhængighed: pip install requests 4) Kør: python3 cvr_nyeste_udtraek.py Alternativ uden ventetid: cvr.dev har en 30-dages gratis prøve med samme branche/region-segmentering. Sig til, så tilpasser jeg scriptet til cvr.dev. ------------------------------------------------------------------------------- Feltnavne følger Erhvervsstyrelsens CVR-datamodel (indeks: cvr-permanent). Hvis et felt skulle være omdøbt, se dokumentationen: https://erhvervsstyrelsen.dk/kom-godt-igang-med-elasticSearch """ import os import sys import csv import json import time from datetime import datetime try: import requests from requests.auth import HTTPBasicAuth except ImportError: sys.exit("Mangler 'requests'. Kør: pip install requests") # ============================================================================= # KONFIGURATION — ret efter behov # ============================================================================= ES_URL = "http://distribution.virk.dk/cvr-permanent/_search" ANTAL_PR_BRANCHE = 10 # antal virksomheder pr. branche (efter frasortering) HENT_BUFFER = 90 # hent ekstra, så der er nok efter filtrering KUN_ENKELTMAND = True # True = kun virksomhedsform 10 (enkeltmandsvirksomhed) KUN_UDEN_HJEMMESIDE = False # True = tag kun dem UDEN registreret hjemmeside i CVR # (dit egentlige salgsmål — men CVR's hjemmeside- # felt er ofte tomt, så det er kun en indikator) # De anbefalede topbrancher (branchekoder fra DB07 / estatistik.dk) BRANCHER = { "toemrer_snedker": "433200", # Tømrer- og bygningssnedkervirksomhed "maler": "433410", # Malerforretninger "anlaegsgartner": "813000", # Landskabspleje (anlægsgartner / havemand) "rengoering": "812100", # Almindelig rengøring i bygninger "vinduespudser": "812210", # Vinduespolering } # Kommunekoder: Region Nordjylland (11) + Region Midtjylland (19). # Verificér evt. mod https://danmarksadresser.dk / DAWA hvis i tvivl. KOMMUNER_NORDJYLLAND = [ "773", # Morsø "787", # Thisted "810", # Brønderslev "813", # Frederikshavn "820", # Vesthimmerland "825", # Læsø "840", # Rebild "846", # Mariagerfjord "849", # Jammerbugt "851", # Aalborg "860", # Hjørring ] KOMMUNER_MIDTJYLLAND = [ "615", # Horsens "657", # Herning "661", # Holstebro "665", # Lemvig "671", # Struer "706", # Syddjurs "707", # Norddjurs "710", # Favrskov "727", # Odder "730", # Randers "740", # Silkeborg "741", # Samsø "746", # Skanderborg "751", # Aarhus "756", # Ikast-Brande "760", # Ringkøbing-Skjern "766", # Hedensted "779", # Skive "791", # Viborg ] KOMMUNER = KOMMUNER_NORDJYLLAND + KOMMUNER_MIDTJYLLAND OUTPUT_CSV = "cvr_nyeste_enkeltmand_nord_midtjylland.csv" # ============================================================================= # LOGIK # ============================================================================= BASE = "Vrvirksomhed" def byg_query(branchekode): """Elasticsearch-forespørgsel: nyeste aktive virksomheder i branche+kommuner.""" must = [ {"term": {f"{BASE}.virksomhedMetadata.nyesteHovedbranche.branchekode": branchekode}}, {"term": {f"{BASE}.virksomhedMetadata.sammensatStatus": "AKTIV"}}, {"terms": {f"{BASE}.virksomhedMetadata.nyesteBeliggenhedsadresse.kommune.kommuneKode": KOMMUNER}}, ] if KUN_ENKELTMAND: must.append( {"term": {f"{BASE}.virksomhedMetadata.nyesteVirksomhedsform.virksomhedsformkode": 10}} ) return { "size": ANTAL_PR_BRANCHE + HENT_BUFFER, "_source": [ f"{BASE}.cvrNummer", f"{BASE}.reklamebeskyttet", f"{BASE}.virksomhedMetadata", f"{BASE}.telefonNummer", f"{BASE}.elektroniskPost", f"{BASE}.hjemmeside", ], "query": {"bool": {"must": must}}, # Nyeste først. Hvis 'stiftelsesDato' skulle mangle sortering, prøv i stedet # feltet 'Vrvirksomhed.sidstOpdateret'. "sort": [{f"{BASE}.virksomhedMetadata.stiftelsesDato": {"order": "desc"}}], } def hent_kontakt(liste): """Saml kontaktværdier fra en CVR-liste (telefon/e-mail/hjemmeside). Springer hemmelige (hemmelig=true) over. Dedupliker, bevar rækkefølge.""" ud = [] for post in (liste or []): if isinstance(post, dict): if post.get("hemmelig"): continue v = post.get("kontaktoplysning") else: v = post if v: ud.append(str(v).strip()) return "; ".join(dict.fromkeys(ud)) def parse_virksomhed(kilde, branche_navn, branchekode): v = kilde.get(BASE, {}) or {} md = v.get("virksomhedMetadata", {}) or {} navn = (md.get("nyesteNavn") or {}).get("navn", "") adr = md.get("nyesteBeliggenhedsadresse") or {} vej = adr.get("vejnavn", "") or "" hus = adr.get("husnummerFra", "") or "" bogstav = adr.get("bogstavFra", "") or "" postnr = adr.get("postnummer", "") or "" by = adr.get("postdistrikt", "") or "" kommune = (adr.get("kommune") or {}).get("kommunenavn", "") or "" adresse = f"{vej} {hus}{bogstav}, {postnr} {by}".strip().strip(",").strip() tlf = hent_kontakt(v.get("telefonNummer")) mail = hent_kontakt(v.get("elektroniskPost")) web = hent_kontakt(v.get("hjemmeside")) return { "branche": branche_navn, "branchekode": branchekode, "cvr": v.get("cvrNummer", ""), "navn": navn, "adresse": adresse, "postnr": postnr, "by": by, "kommune": kommune, "startdato": md.get("stiftelsesDato", "") or "", "telefon": tlf, "email": mail, "hjemmeside": web, } def koer(): user = os.environ.get("CVR_USER") pw = os.environ.get("CVR_PASS") if not user or not pw: sys.exit("Sæt miljøvariablerne CVR_USER og CVR_PASS først (se toppen af filen).") auth = HTTPBasicAuth(user, pw) headers = {"Content-Type": "application/json"} alle = [] for branche_navn, kode in BRANCHER.items(): query = byg_query(kode) try: r = requests.post(ES_URL, auth=auth, headers=headers, data=json.dumps(query), timeout=60) r.raise_for_status() except requests.HTTPError as e: print(f" FEJL for {branche_navn} ({kode}): {e} — {r.text[:200]}") continue except requests.RequestException as e: print(f" NETVÆRKSFEJL for {branche_navn} ({kode}): {e}") continue hits = r.json().get("hits", {}).get("hits", []) antal = 0 for h in hits: v = h.get("_source", {}).get(BASE, {}) or {} if v.get("reklamebeskyttet"): # COMPLIANCE: spring reklamebeskyttede over continue row = parse_virksomhed(h.get("_source", {}), branche_navn, kode) if KUN_UDEN_HJEMMESIDE and row["hjemmeside"]: continue alle.append(row) antal += 1 if antal >= ANTAL_PR_BRANCHE: break print(f" {branche_navn} ({kode}): {antal} virksomheder fundet") time.sleep(1) # vær venlig mod API'et if not alle: print("Ingen resultater. Tjek adgang, feltnavne og filtre.") return felter = ["branche", "branchekode", "cvr", "navn", "adresse", "postnr", "by", "kommune", "startdato", "telefon", "email", "hjemmeside"] with open(OUTPUT_CSV, "w", newline="", encoding="utf-8-sig") as f: w = csv.DictWriter(f, fieldnames=felter) w.writeheader() w.writerows(alle) print(f"\nFærdig: {len(alle)} virksomheder skrevet til {OUTPUT_CSV}") print(f"Kørt: {datetime.now():%Y-%m-%d %H:%M}") if __name__ == "__main__": koer()