Floke/Brancheneinstufung2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Optimierung der Wikipedia-Auswertung: Umsatz-Extraktion und Mitarbeiterzahl verbessert (v1.1.5)

Browse Source 
Zusammenfassung:

Spaltenreihenfolge aktualisiert:

G: Wikipedia URL

H: Erster Absatz des Wikipedia-Artikels

I: Branche (aus der Infobox)

J: Umsatz (als Zahl in Mio €, z. B. "159")

K: Anzahl Mitarbeiter (aus der Infobox)

N: Datum und aktuelle Zeit

Q: Version

Umsatz-Extraktion:

Regex erkennt jetzt "Mio"/"Millionen" und "Mrd"/"Milliarden".

Ist keine Einheit vorhanden, wird der Wert in Euro in Mio € umgerechnet (Division durch 1.000.000).

Mitarbeiterzahl-Extraktion:

Regex wurde erweitert, um Zahlen mit Leerzeichen zu erfassen.

Zusätzliche Schlüsselwörter wie "Mitarbeiterzahl" werden berücksichtigt.

Deprecation-Warnings:

Die Update-Aufrufe für Google Sheets wurden so angepasst, dass die Werte vor dem Range-Namen übergeben werden.
This commit is contained in:
Floke
2025-03-31 20:26:57 +00:00
parent 59d6aed51a
commit 2839cfe665
1 changed files with 32 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

80
brancheneinstufung.py
View File

@@ -12,7 +12,7 @@ import csv
# ==================== KONFIGURATION ==================== # ==================== KONFIGURATION ====================
class Config: class Config:
VERSION = "1.1.4" VERSION = "1.1.5"
LANG = "de" LANG = "de"
CREDENTIALS_FILE = "service_account.json" CREDENTIALS_FILE = "service_account.json"
SHEET_URL = "https://docs.google.com/spreadsheets/d/1u_gHr9JUfmV1-iviRzbSe3575QEp7KLhK5jFV_gJcgo" SHEET_URL = "https://docs.google.com/spreadsheets/d/1u_gHr9JUfmV1-iviRzbSe3575QEp7KLhK5jFV_gJcgo"
@@ -47,7 +47,7 @@ def clean_text(text):
if not text: if not text:
return "k.A." return "k.A."
text = str(text) text = str(text)
text = re.sub(r'\[\d+\]', '', text) # Entferne Referenznummern text = re.sub(r'\[\d+\]', '', text)
text = re.sub(r'\s+', ' ', text).strip() text = re.sub(r'\s+', ' ', text).strip()
return text if text else "k.A." return text if text else "k.A."
@@ -55,55 +55,48 @@ def normalize_company_name(name):
"""Entfernt gängige Firmierungsformen und normalisiert den Namen.""" """Entfernt gängige Firmierungsformen und normalisiert den Namen."""
if not name: if not name:
return "" return ""
# Liste gängiger Firmierungsformen
forms = [ forms = [
r'gmbh', r'g\.m\.b\.h\.', r'ug', r'u\.g\.', r'ug \(haftungsbeschränkt\)', r'gmbh', r'g\.m\.b\.h\.', r'ug', r'u\.g\.', r'ug \(haftungsbeschränkt\)',
r'u\.g\. \(haftungsbeschränkt\)', r'ag', r'a\.g\.', r'ohg', r'o\.h\.g\.', r'u\.g\. \(haftungsbeschränkt\)', r'ag', r'a\.g\.', r'ohg', r'o\.h\.g\.',
r'kg', r'k\.g\.', r'gmbh & co\. kg', r'g\.m\.b\.h\. & co\. k\.g\.', r'kg', r'k\.g\.', r'gmbh & co\.?\s*kg', r'g\.m\.b\.h\. & co\.?\s*k\.g\.',
r'ag & co\. kg', r'a\.g\. & co\. k\.g\.', r'e\.k\.', r'e\.kfm\.', r'e\.kfr\.', r'ag & co\.?\s*kg', r'a\.g\. & co\.?\s*k\.g\.', r'e\.k\.', r'e\.kfm\.',
r'ltd\.', r'ltd & co\. kg', r's\.a r\.l\.', r'stiftung', r'genossenschaft', r'e\.kfr\.', r'ltd\.', r'ltd & co\.?\s*kg', r's\.a r\.l\.', r'stiftung',
r'ggmbh', r'gug', r'partg', r'partgmbb', r'kgaa', r'se', r'og', r'o\.g\.', r'genossenschaft', r'ggmbh', r'gug', r'partg', r'partgmbb', r'kgaa', r'se',
r'e\.u\.', r'ges\.n\.b\.r\.', r'genmbh', r'verein', r'kollektivgesellschaft', r'og', r'o\.g\.', r'e\.u\.', r'ges\.n\.b\.r\.', r'genmbh', r'verein',
r'kommanditgesellschaft', r'einzelfirma', r'sàrl', r'sa', r'sagl', r'kollektivgesellschaft', r'kommanditgesellschaft', r'einzelfirma', r'sàrl',
r'gmbh & co\. ohg', r'ag & co\. ohg', r'gmbh & co\. kgaa', r'ag & co\. kgaa', r'sa', r'sagl', r'gmbh & co\.?\s*ohg', r'ag & co\.?\s*ohg', r'gmbh & co\.?\s*kgaa',
r's\.a\.', r's\.p\.a\.', r'b\.v\.', r'n\.v\.' r'ag & co\.?\s*kgaa', r's\.a\.', r's\.p\.a\.', r'b\.v\.', r'n\.v\.'
] ]
pattern = r'\b(' + '|'.join(forms) + r')\b' pattern = r'\b(' + '|'.join(forms) + r')\b'
normalized = re.sub(pattern, '', name, flags=re.IGNORECASE) normalized = re.sub(pattern, '', name, flags=re.IGNORECASE)
normalized = re.sub(r'[\-–]', ' ', normalized) # Ersetze Bindestriche durch Leerzeichen normalized = re.sub(r'[\-–]', ' ', normalized)
normalized = re.sub(r'\s+', ' ', normalized).strip() normalized = re.sub(r'\s+', ' ', normalized).strip()
return normalized.lower() return normalized.lower()
# ==================== GOOGLE SHEET HANDLER ==================== # ==================== GOOGLE SHEET HANDLER ====================
class GoogleSheetHandler: class GoogleSheetHandler:
"""Handhabung der Google Sheets Interaktion""" """Handhabung der Google Sheets Interaktion"""
def __init__(self): def __init__(self):
self.sheet = None self.sheet = None
self.sheet_values = [] self.sheet_values = []
self._connect() self._connect()
def _connect(self): def _connect(self):
"""Stellt Verbindung zum Google Sheet her""" """Stellt Verbindung zum Google Sheet her"""
scope = ["https://www.googleapis.com/auth/spreadsheets"] scope = ["https://www.googleapis.com/auth/spreadsheets"]
creds = ServiceAccountCredentials.from_json_keyfile_name(Config.CREDENTIALS_FILE, scope) creds = ServiceAccountCredentials.from_json_keyfile_name(Config.CREDENTIALS_FILE, scope)
self.sheet = gspread.authorize(creds).open_by_url(Config.SHEET_URL).sheet1 self.sheet = gspread.authorize(creds).open_by_url(Config.SHEET_URL).sheet1
self.sheet_values = self.sheet.get_all_values() self.sheet_values = self.sheet.get_all_values()
def get_start_index(self): def get_start_index(self):
"""Ermittelt die erste leere Zeile in Spalte N (Index 14)""" """Ermittelt die erste leere Zeile in Spalte N (Index 14)"""
filled_n = [row[13] if len(row) > 13 else '' for row in self.sheet_values[1:]] filled_n = [row[13] if len(row) > 13 else '' for row in self.sheet_values[1:]]
return next((i + 1 for i, v in enumerate(filled_n, start=1) if not str(v).strip()), len(filled_n) + 1) return next((i + 1 for i, v in enumerate(filled_n, start=1) if not str(v).strip()), len(filled_n) + 1)
# Update-Aufrufe erfolgen separat für verschiedene Spalten.
# Für die neuen Updates nutzen wir separate Update-Aufrufe
# ==================== WIKIPEDIA SCRAPER ==================== # ==================== WIKIPEDIA SCRAPER ====================
class WikipediaScraper: class WikipediaScraper:
"""Klasse zur Handhabung der Wikipedia-Suche und Datenextraktion""" """Klasse zur Handhabung der Wikipedia-Suche und Datenextraktion"""
def __init__(self): def __init__(self):
wikipedia.set_lang(Config.LANG) wikipedia.set_lang(Config.LANG)
def _get_full_domain(self, website): def _get_full_domain(self, website):
"""Extrahiert den vollständigen Domainnamen (inklusive Topleveldomain) aus der URL.""" """Extrahiert den vollständigen Domainnamen (inklusive Topleveldomain) aus der URL."""
if not website: if not website:
@@ -113,7 +106,6 @@ class WikipediaScraper:
website = re.sub(r'^www\.', '', website) website = re.sub(r'^www\.', '', website)
website = website.split('/')[0] website = website.split('/')[0]
return website return website
def _generate_search_terms(self, company_name, website): def _generate_search_terms(self, company_name, website):
""" """
Generiert Suchbegriffe in folgender Reihenfolge: Generiert Suchbegriffe in folgender Reihenfolge:
@@ -133,11 +125,10 @@ class WikipediaScraper:
terms.append(normalized_name) terms.append(normalized_name)
debug_print(f"Generierte Suchbegriffe: {terms}") debug_print(f"Generierte Suchbegriffe: {terms}")
return terms return terms
def _validate_article(self, page, company_name, website): def _validate_article(self, page, company_name, website):
""" """
Validiert den Artikel: Validiert den Artikel:
- Sucht in der Infobox und in den externen Links nach Links, die den vollständigen Domainnamen enthalten. - Sucht in der Infobox und in externen Links nach Links, die den vollständigen Domainnamen enthalten.
Wird ein solcher Link gefunden, wird ein niedrigerer Schwellenwert (0.60) angewendet. Wird ein solcher Link gefunden, wird ein niedrigerer Schwellenwert (0.60) angewendet.
- Andernfalls werden der normalisierte Wikipedia-Titel und der normalisierte Firmenname verglichen. - Andernfalls werden der normalisierte Wikipedia-Titel und der normalisierte Firmenname verglichen.
""" """
@@ -147,7 +138,6 @@ class WikipediaScraper:
try: try:
html_raw = requests.get(page.url).text html_raw = requests.get(page.url).text
soup = BeautifulSoup(html_raw, Config.HTML_PARSER) soup = BeautifulSoup(html_raw, Config.HTML_PARSER)
# Suche in der Infobox
infobox = soup.find('table', class_=lambda c: c and 'infobox' in c.lower()) infobox = soup.find('table', class_=lambda c: c and 'infobox' in c.lower())
if infobox: if infobox:
links = infobox.find_all('a', href=True) links = infobox.find_all('a', href=True)
@@ -159,7 +149,6 @@ class WikipediaScraper:
debug_print(f"Definitiver Link-Match in Infobox gefunden: {href}") debug_print(f"Definitiver Link-Match in Infobox gefunden: {href}")
domain_found = True domain_found = True
break break
# Suche in externen Links, falls vorhanden
if not domain_found and hasattr(page, 'externallinks'): if not domain_found and hasattr(page, 'externallinks'):
for ext_link in page.externallinks: for ext_link in page.externallinks:
if full_domain in ext_link.lower(): if full_domain in ext_link.lower():
@@ -168,15 +157,12 @@ class WikipediaScraper:
break break
except Exception as e: except Exception as e:
debug_print(f"Fehler beim Extrahieren von Links: {str(e)}") debug_print(f"Fehler beim Extrahieren von Links: {str(e)}")
normalized_title = normalize_company_name(page.title) normalized_title = normalize_company_name(page.title)
normalized_company = normalize_company_name(company_name) normalized_company = normalize_company_name(company_name)
similarity = SequenceMatcher(None, normalized_title, normalized_company).ratio() similarity = SequenceMatcher(None, normalized_title, normalized_company).ratio()
debug_print(f"Ähnlichkeit (normalisiert): {similarity:.2f} ({normalized_title} vs {normalized_company})") debug_print(f"Ähnlichkeit (normalisiert): {similarity:.2f} ({normalized_title} vs {normalized_company})")
threshold = 0.60 if domain_found else Config.SIMILARITY_THRESHOLD threshold = 0.60 if domain_found else Config.SIMILARITY_THRESHOLD
return similarity >= threshold return similarity >= threshold
def extract_first_paragraph(self, page_url): def extract_first_paragraph(self, page_url):
"""Extrahiert den ersten sinnvollen Absatz aus dem Wikipedia-Artikel.""" """Extrahiert den ersten sinnvollen Absatz aus dem Wikipedia-Artikel."""
try: try:
@@ -191,7 +177,6 @@ class WikipediaScraper:
except Exception as e: except Exception as e:
debug_print(f"Fehler beim Extrahieren des ersten Absatzes: {e}") debug_print(f"Fehler beim Extrahieren des ersten Absatzes: {e}")
return "k.A." return "k.A."
def _extract_infobox_value(self, soup, target): def _extract_infobox_value(self, soup, target):
"""Extrahiert Werte aus der Infobox (Fallback-Methode) für 'branche', 'umsatz' und 'mitarbeiter'.""" """Extrahiert Werte aus der Infobox (Fallback-Methode) für 'branche', 'umsatz' und 'mitarbeiter'."""
infobox = soup.find('table', class_=lambda c: c and any(kw in c.lower() for kw in ['infobox', 'vcard', 'unternehmen'])) infobox = soup.find('table', class_=lambda c: c and any(kw in c.lower() for kw in ['infobox', 'vcard', 'unternehmen']))
@@ -200,7 +185,7 @@ class WikipediaScraper:
keywords_map = { keywords_map = {
'branche': ['branche', 'industrie', 'tätigkeit', 'geschäftsfeld', 'sektor', 'produkte', 'leistungen', 'aktivitäten', 'wirtschaftszweig'], 'branche': ['branche', 'industrie', 'tätigkeit', 'geschäftsfeld', 'sektor', 'produkte', 'leistungen', 'aktivitäten', 'wirtschaftszweig'],
'umsatz': ['umsatz', 'jahresumsatz', 'konzernumsatz', 'gesamtumsatz', 'erlöse', 'umsatzerlöse', 'einnahmen', 'ergebnis', 'jahresergebnis'], 'umsatz': ['umsatz', 'jahresumsatz', 'konzernumsatz', 'gesamtumsatz', 'erlöse', 'umsatzerlöse', 'einnahmen', 'ergebnis', 'jahresergebnis'],
'mitarbeiter': ['mitarbeiter', 'beschäftigte', 'personal', 'anzahl mitarbeiter'] 'mitarbeiter': ['mitarbeiter', 'beschäftigte', 'personal', 'mitarbeiterzahl']
} }
keywords = keywords_map.get(target, []) keywords = keywords_map.get(target, [])
for row in infobox.find_all('tr'): for row in infobox.find_all('tr'):
@@ -215,25 +200,32 @@ class WikipediaScraper:
clean_val = re.sub(r'\[.*?\]|\(.*?\)', '', raw_value) clean_val = re.sub(r'\[.*?\]|\(.*?\)', '', raw_value)
return ' '.join(clean_val.split()).strip() return ' '.join(clean_val.split()).strip()
if target == 'umsatz': if target == 'umsatz':
match = re.search(r'(\d{1,3}(?:[.,]\d{3})*|\d+)', raw_value.replace('.', '').replace(',', '.')) raw = raw_value.lower()
match = re.search(r'(\d{1,3}(?:[.,]\d{3})*|\d+)', raw)
if match: if match:
num = float(match.group(1)) num = float(match.group(1).replace(',', '.'))
if 'mrd' in raw or 'milliarden' in raw:
num *= 1000
elif 'mio' in raw or 'millionen' in raw:
num = num
else:
num /= 1e6
return str(int(round(num))) return str(int(round(num)))
return raw_value.strip() return raw_value.strip()
if target == 'mitarbeiter': if target == 'mitarbeiter':
match = re.search(r'(\d{1,3}(?:[.,]\d{3})*|\d+)', raw_value.replace('.', '').replace(',', '.')) raw = raw_value.lower()
match = re.search(r'(\d[\d\s,\.]*)', raw)
if match: if match:
return match.group(1) num_str = re.sub(r'\s+', '', match.group(1))
return num_str
return raw_value.strip() return raw_value.strip()
return "k.A." return "k.A."
def extract_full_infobox(self, soup): def extract_full_infobox(self, soup):
"""Extrahiert die komplette Infobox als Text (nicht mehr ausgegeben)""" """Extrahiert die komplette Infobox als Text (nicht mehr ausgegeben)"""
infobox = soup.find('table', class_=lambda c: c and any(kw in c.lower() for kw in ['infobox', 'vcard', 'unternehmen'])) infobox = soup.find('table', class_=lambda c: c and any(kw in c.lower() for kw in ['infobox', 'vcard', 'unternehmen']))
if not infobox: if not infobox:
return "k.A." return "k.A."
return clean_text(infobox.get_text(separator=' | ')) return clean_text(infobox.get_text(separator=' | '))
def extract_fields_from_infobox_text(self, infobox_text, field_names): def extract_fields_from_infobox_text(self, infobox_text, field_names):
"""Extrahiert die gewünschten Felder aus dem Infobox-Text (getrennt durch ' | ')""" """Extrahiert die gewünschten Felder aus dem Infobox-Text (getrennt durch ' | ')"""
result = {} result = {}
@@ -246,7 +238,6 @@ class WikipediaScraper:
j += 1 j += 1
result[field] = tokens[j] if j < len(tokens) else "k.A." result[field] = tokens[j] if j < len(tokens) else "k.A."
return result return result
def extract_company_data(self, page_url): def extract_company_data(self, page_url):
"""Extrahiert Daten aus dem Wikipedia-Artikel (Erster Absatz, Branche, Umsatz, Mitarbeiter)""" """Extrahiert Daten aus dem Wikipedia-Artikel (Erster Absatz, Branche, Umsatz, Mitarbeiter)"""
if not page_url: if not page_url:
@@ -270,7 +261,6 @@ class WikipediaScraper:
except Exception as e: except Exception as e:
debug_print(f"Extraktionsfehler: {str(e)}") debug_print(f"Extraktionsfehler: {str(e)}")
return {'url': 'k.A.', 'first_paragraph': 'k.A.', 'branche': 'k.A.', 'umsatz': 'k.A.', 'mitarbeiter': 'k.A.'} return {'url': 'k.A.', 'first_paragraph': 'k.A.', 'branche': 'k.A.', 'umsatz': 'k.A.', 'mitarbeiter': 'k.A.'}
@retry_on_failure @retry_on_failure
def search_company_article(self, company_name, website): def search_company_article(self, company_name, website):
"""Sucht mit optimierten Suchbegriffen (vollständiger Domainname, Candidate, normalisierter Name) nach dem Wikipedia-Artikel.""" """Sucht mit optimierten Suchbegriffen (vollständiger Domainname, Candidate, normalisierter Name) nach dem Wikipedia-Artikel."""
@@ -295,11 +285,9 @@ class WikipediaScraper:
# ==================== DATA PROCESSOR ==================== # ==================== DATA PROCESSOR ====================
class DataProcessor: class DataProcessor:
"""Steuerung des Gesamtprozesses""" """Steuerung des Gesamtprozesses"""
def __init__(self): def __init__(self):
self.sheet_handler = GoogleSheetHandler() self.sheet_handler = GoogleSheetHandler()
self.wiki_scraper = WikipediaScraper() self.wiki_scraper = WikipediaScraper()
def process_rows(self, num_rows): def process_rows(self, num_rows):
"""Verarbeitet die angegebene Anzahl an Zeilen""" """Verarbeitet die angegebene Anzahl an Zeilen"""
start_index = self.sheet_handler.get_start_index() start_index = self.sheet_handler.get_start_index()
@@ -307,7 +295,6 @@ class DataProcessor:
for i in range(start_index, min(start_index + num_rows, len(self.sheet_handler.sheet_values))): for i in range(start_index, min(start_index + num_rows, len(self.sheet_handler.sheet_values))):
row = self.sheet_handler.sheet_values[i] row = self.sheet_handler.sheet_values[i]
self._process_single_row(i+1, row) self._process_single_row(i+1, row)
def _process_single_row(self, row_num, row_data): def _process_single_row(self, row_num, row_data):
"""Verarbeitung einer einzelnen Zeile""" """Verarbeitung einer einzelnen Zeile"""
company_name = row_data[0] if len(row_data) > 0 else "" company_name = row_data[0] if len(row_data) > 0 else ""
@@ -318,22 +305,19 @@ class DataProcessor:
company_data = self.wiki_scraper.extract_company_data(article.url) company_data = self.wiki_scraper.extract_company_data(article.url)
else: else:
company_data = {'url': 'k.A.', 'first_paragraph': 'k.A.', 'branche': 'k.A.', 'umsatz': 'k.A.', 'mitarbeiter': 'k.A.'} company_data = {'url': 'k.A.', 'first_paragraph': 'k.A.', 'branche': 'k.A.', 'umsatz': 'k.A.', 'mitarbeiter': 'k.A.'}
# Update der Spalten G bis K: URL, erster Absatz, Branche, Umsatz, Mitarbeiter
# Update der Spalten: self.sheet_handler.sheet.update(values=[[
# G: URL, H: erster Absatz, I: Branche, J: Umsatz (als Zahl), K: Mitarbeiter
self.sheet_handler.sheet.update(f"G{row_num}:K{row_num}", [[
company_data.get('url', 'k.A.'), company_data.get('url', 'k.A.'),
company_data.get('first_paragraph', 'k.A.'), company_data.get('first_paragraph', 'k.A.'),
company_data.get('branche', 'k.A.'), company_data.get('branche', 'k.A.'),
company_data.get('umsatz', 'k.A.'), company_data.get('umsatz', 'k.A.'),
company_data.get('mitarbeiter', 'k.A.') company_data.get('mitarbeiter', 'k.A.')
]]) ]], range_name=f"G{row_num}:K{row_num}")
# Spalte N: Datum und aktuelle Zeit # Spalte N: Datum und aktuelle Zeit
current_dt = datetime.now().strftime("%Y-%m-%d %H:%M:%S") current_dt = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
self.sheet_handler.sheet.update(f"N{row_num}", [[current_dt]]) self.sheet_handler.sheet.update(values=[[current_dt]], range_name=f"N{row_num}")
# Spalte Q: Version # Spalte Q: Version
self.sheet_handler.sheet.update(f"Q{row_num}", [[Config.VERSION]]) self.sheet_handler.sheet.update(values=[[Config.VERSION]], range_name=f"Q{row_num}")
print(f"✅ Aktualisiert: URL: {company_data.get('url', 'k.A.')}, Erster Absatz: {company_data.get('first_paragraph', 'k.A.')[:30]}..., Branche: {company_data.get('branche', 'k.A.')}, Umsatz: {company_data.get('umsatz', 'k.A.')}, Mitarbeiter: {company_data.get('mitarbeiter', 'k.A.')}") print(f"✅ Aktualisiert: URL: {company_data.get('url', 'k.A.')}, Erster Absatz: {company_data.get('first_paragraph', 'k.A.')[:30]}..., Branche: {company_data.get('branche', 'k.A.')}, Umsatz: {company_data.get('umsatz', 'k.A.')}, Mitarbeiter: {company_data.get('mitarbeiter', 'k.A.')}")
time.sleep(Config.RETRY_DELAY) time.sleep(Config.RETRY_DELAY)