Floke
5137e5d22e
- Erstellung einer robusten `serp_wikipedia_lookup`-Funktion im WikipediaScraper. - Verbesserung der `_validate_article`-Logik um harte Fakten (Domain, Sitz). - Anpassung von `search_company_article` zur Nutzung der neuen "Google-First"-Strategie.
481 lines
23 KiB
Python
481 lines
23 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
wikipedia_scraper.py
|
|
|
|
Klasse zur Kapselung der Interaktionen mit Wikipedia, inklusive Suche,
|
|
Validierung und Extraktion von Unternehmensdaten.
|
|
"""
|
|
|
|
__version__ = "v2.0.2"
|
|
|
|
import logging
|
|
import re
|
|
import time
|
|
import traceback
|
|
from urllib.parse import unquote
|
|
|
|
import requests
|
|
import wikipedia
|
|
from bs4 import BeautifulSoup
|
|
|
|
# Import der abhängigen Module
|
|
from config import Config
|
|
from helpers import (retry_on_failure, simple_normalize_url,
|
|
normalize_company_name, extract_numeric_value,
|
|
clean_text, fuzzy_similarity)
|
|
|
|
class WikipediaScraper:
|
|
"""
|
|
Handhabt das Suchen von Wikipedia-Artikeln und das Extrahieren relevanter
|
|
Unternehmensdaten. Beinhaltet Validierungslogik fuer Artikel.
|
|
Nutzt die wikipedia-Bibliothek und Requests fuer direktes HTML-Scraping.
|
|
"""
|
|
def __init__(self, user_agent=None):
|
|
"""
|
|
Initialisiert den Scraper mit einer Requests-Session und konfigurierter
|
|
Wikipedia-Bibliothek.
|
|
"""
|
|
self.logger = logging.getLogger(__name__ + ".WikipediaScraper")
|
|
self.logger.debug("WikipediaScraper initialisiert.")
|
|
|
|
self.user_agent = user_agent or getattr(Config, 'USER_AGENT', 'Mozilla/5.0 (compatible; UnternehmenSkript/1.0; +http://www.example.com/bot)')
|
|
self.session = requests.Session()
|
|
self.session.headers.update({'User-Agent': self.user_agent})
|
|
self.logger.debug(f"Requests Session mit User-Agent '{self.user_agent}' initialisiert.")
|
|
|
|
self.keywords_map = {
|
|
'branche': ['branche', 'wirtschaftszweig', 'industry', 'taetigkeit', 'sektor', 'produkte', 'leistungen'],
|
|
'umsatz': ['umsatz', 'erloes', 'revenue', 'jahresumsatz', 'konzernumsatz', 'ergebnis'],
|
|
'mitarbeiter': ['mitarbeiter', 'mitarbeiterzahl', 'beschaeftigte', 'employees', 'number of employees', 'personal', 'belegschaft'],
|
|
'sitz': ['sitz', 'hauptsitz', 'unternehmenssitz', 'firmensitz', 'headquarters', 'standort', 'sitz des unternehmens', 'anschrift', 'adresse']
|
|
}
|
|
|
|
try:
|
|
wiki_lang = getattr(Config, 'LANG', 'de')
|
|
wikipedia.set_lang(wiki_lang)
|
|
wikipedia.set_rate_limiting(False)
|
|
self.logger.info(f"Wikipedia library language set to '{wiki_lang}'. Rate limiting DISABLED.")
|
|
except Exception as e:
|
|
self.logger.warning(f"Fehler beim Setzen der Wikipedia-Sprache oder Rate Limiting: {e}")
|
|
|
|
@retry_on_failure
|
|
def serp_wikipedia_lookup(self, company_name, lang='de'):
|
|
"""
|
|
Sucht die beste Wikipedia-URL für ein Unternehmen über eine Google-Suche (via SerpAPI).
|
|
Priorisiert Treffer aus dem Knowledge Graph und organische Ergebnisse.
|
|
|
|
Args:
|
|
company_name (str): Der Name des zu suchenden Unternehmens.
|
|
lang (str): Der Sprachcode für die Wikipedia-Suche (z.B. 'de').
|
|
|
|
Returns:
|
|
str: Die URL des besten Treffers oder None, wenn nichts Passendes gefunden wurde.
|
|
"""
|
|
self.logger.info(f"Starte SerpAPI Wikipedia-Suche für '{company_name}'...")
|
|
serp_key = Config.API_KEYS.get('serpapi')
|
|
if not serp_key:
|
|
self.logger.warning("SerpAPI Key nicht konfiguriert. Suche wird übersprungen.")
|
|
return None
|
|
|
|
query = f'site:{lang}.wikipedia.org "{company_name}"'
|
|
params = {"engine": "google", "q": query, "api_key": serp_key, "hl": lang}
|
|
|
|
try:
|
|
response = requests.get("https://serpapi.com/search", params=params, timeout=Config.REQUEST_TIMEOUT)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
# 1. Knowledge Graph prüfen (höchste Priorität)
|
|
if "knowledge_graph" in data and "source" in data["knowledge_graph"]:
|
|
source = data["knowledge_graph"]["source"]
|
|
if "link" in source and f"{lang}.wikipedia.org" in source["link"]:
|
|
url = source["link"]
|
|
self.logger.info(f" -> Treffer aus Knowledge Graph gefunden: {url}")
|
|
return url
|
|
|
|
# 2. Organische Ergebnisse prüfen
|
|
if "organic_results" in data:
|
|
for result in data.get("organic_results", []):
|
|
link = result.get("link")
|
|
if link and f"{lang}.wikipedia.org/wiki/" in link:
|
|
self.logger.info(f" -> Bester organischer Treffer gefunden: {link}")
|
|
return link
|
|
|
|
self.logger.warning(f" -> Keine passende Wikipedia-URL für '{company_name}' in den SerpAPI-Ergebnissen gefunden.")
|
|
return None
|
|
except Exception as e:
|
|
self.logger.error(f"Fehler bei der SerpAPI-Anfrage für '{company_name}': {e}")
|
|
return None
|
|
|
|
def _get_full_domain(self, website):
|
|
"""Extrahiert die normalisierte Domain (ohne www, ohne Pfad) aus einer URL."""
|
|
return simple_normalize_url(website)
|
|
|
|
def _generate_search_terms(self, company_name, website=None):
|
|
"""
|
|
Generiert eine Liste von potenziellen Wikipedia-Artikeltiteln.
|
|
v2.0: Mit verbesserter Logik für Namen, die Zahlen enthalten.
|
|
"""
|
|
if not company_name:
|
|
return []
|
|
|
|
normalized = normalize_company_name(company_name)
|
|
|
|
# Verbesserte Logik für Namen wie "11 88 0 Solutions"
|
|
condensed_normalized = None
|
|
if re.search(r'\d[\s\d]+\d', normalized):
|
|
condensed_normalized = re.sub(r'(\d)\s+(\d)', r'\1\2', normalized)
|
|
condensed_normalized = normalize_company_name(condensed_normalized)
|
|
|
|
search_terms = []
|
|
if condensed_normalized: search_terms.append(condensed_normalized)
|
|
search_terms.append(company_name)
|
|
search_terms.append(normalized)
|
|
|
|
parts = normalized.split()
|
|
if len(parts) > 1:
|
|
search_terms.append(parts[0])
|
|
search_terms.append(" ".join(parts[:2]))
|
|
|
|
if website:
|
|
domain = simple_normalize_url(website)
|
|
if domain != "k.A.":
|
|
search_terms.append(domain)
|
|
|
|
unique_terms = list(dict.fromkeys([term for term in search_terms if term])) # Entfernt Duplikate, behält Reihenfolge
|
|
return unique_terms[:5]
|
|
|
|
@retry_on_failure
|
|
def _get_page_soup(self, url):
|
|
"""
|
|
Holt HTML von einer URL und gibt ein BeautifulSoup-Objekt zurueck.
|
|
"""
|
|
if not url or not isinstance(url, str) or not url.lower().startswith(("http://", "https://")):
|
|
self.logger.warning(f"_get_page_soup: Ungueltige URL '{url[:100]}...'.")
|
|
return None
|
|
try:
|
|
self.logger.debug(f"_get_page_soup: Rufe URL ab: {url[:100]}...")
|
|
response = self.session.get(url, timeout=getattr(Config, 'REQUEST_TIMEOUT', 15))
|
|
response.raise_for_status()
|
|
response.encoding = response.apparent_encoding
|
|
soup = BeautifulSoup(response.text, getattr(Config, 'HTML_PARSER', 'html.parser'))
|
|
return soup
|
|
except Exception as e:
|
|
self.logger.error(f"_get_page_soup: Fehler beim Abrufen oder Parsen von HTML von {url[:100]}...: {e}")
|
|
raise e
|
|
|
|
def _validate_article(self, page, company_name, website, crm_city, parent_name=None):
|
|
"""
|
|
Validiert faktenbasiert, ob ein Wikipedia-Artikel zum Unternehmen passt.
|
|
Priorisiert harte Fakten (Domain, Sitz) vor reiner Namensähnlichkeit.
|
|
"""
|
|
if not page or not hasattr(page, 'html'):
|
|
return False
|
|
|
|
self.logger.debug(f"Validiere Artikel '{page.title}' für Firma '{company_name}'...")
|
|
|
|
try:
|
|
page_html = page.html()
|
|
soup = BeautifulSoup(page_html, Config.HTML_PARSER)
|
|
except Exception as e:
|
|
self.logger.error(f"Konnte HTML für Artikel '{page.title}' nicht parsen: {e}")
|
|
return False
|
|
|
|
# --- Stufe 1: Website-Domain-Validierung (sehr starkes Signal) ---
|
|
normalized_domain = simple_normalize_url(website)
|
|
if normalized_domain != "k.A.":
|
|
# Suche nach der Domain im "Weblinks"-Abschnitt oder in der Infobox
|
|
external_links = soup.select('.external, .infobox a[href*="."]')
|
|
for link in external_links:
|
|
href = link.get('href', '')
|
|
if normalized_domain in href:
|
|
self.logger.info(f" => VALIDATION SUCCESS (Domain Match): Domain '{normalized_domain}' in Weblinks gefunden.")
|
|
return True
|
|
|
|
# --- Stufe 2: Sitz-Validierung (starkes Signal) ---
|
|
if crm_city and crm_city.lower() != 'k.a.':
|
|
infobox_sitz_raw = self._extract_infobox_value(soup, 'sitz')
|
|
if infobox_sitz_raw and infobox_sitz_raw.lower() != 'k.a.':
|
|
if crm_city.lower() in infobox_sitz_raw.lower():
|
|
self.logger.info(f" => VALIDATION SUCCESS (City Match): CRM-Ort '{crm_city}' in Infobox-Sitz '{infobox_sitz_raw}' gefunden.")
|
|
return True
|
|
|
|
# --- Stufe 3: Parent-Validierung ---
|
|
normalized_parent = normalize_company_name(parent_name) if parent_name else None
|
|
if normalized_parent:
|
|
page_content_for_check = (page.title + " " + page.summary).lower()
|
|
if normalized_parent in page_content_for_check:
|
|
self.logger.info(f" => VALIDATION SUCCESS (Parent Match): Parent-Name '{parent_name}' im Artikel gefunden.")
|
|
return True
|
|
|
|
# --- Stufe 4: Namensähnlichkeit (Fallback mit strengeren Regeln) ---
|
|
normalized_company = normalize_company_name(company_name)
|
|
normalized_title = normalize_company_name(page.title)
|
|
similarity = fuzzy_similarity(normalized_title, normalized_company)
|
|
|
|
if similarity > 0.85: # Strengere Schwelle
|
|
self.logger.info(f" => VALIDATION SUCCESS (High Similarity): Hohe Namensähnlichkeit ({similarity:.2f}).")
|
|
return True
|
|
|
|
self.logger.debug(f" => VALIDATION FAILED: Kein harter Fakt (Domain, Sitz, Parent) und Ähnlichkeit ({similarity:.2f}) zu gering.")
|
|
return False
|
|
|
|
def search_company_article(self, company_name, website=None, crm_city=None, parent_name=None):
|
|
"""
|
|
Sucht und validiert einen passenden Wikipedia-Artikel nach der "Google-First"-Strategie.
|
|
1. Sucht die beste URL via SerpAPI.
|
|
2. Validiert den gefundenen Artikel mit harten Fakten.
|
|
"""
|
|
if not company_name:
|
|
return None
|
|
|
|
self.logger.info(f"Starte 'Google-First' Wikipedia-Suche für '{company_name}'...")
|
|
|
|
# 1. Finde den besten URL-Kandidaten via Google-Suche
|
|
url_candidate = self.serp_wikipedia_lookup(company_name)
|
|
|
|
if not url_candidate:
|
|
self.logger.warning(f" -> Keine URL via SerpAPI gefunden. Suche abgebrochen.")
|
|
return None
|
|
|
|
# 2. Lade und validiere den gefundenen Artikel
|
|
try:
|
|
page_title = unquote(url_candidate.split('/wiki/')[-1].replace('_', ' '))
|
|
page = wikipedia.page(title=page_title, auto_suggest=False, redirect=True)
|
|
|
|
# Nutze die neue, faktenbasierte Validierung
|
|
if self._validate_article(page, company_name, website, crm_city, parent_name):
|
|
self.logger.info(f" -> Artikel '{page.title}' erfolgreich validiert.")
|
|
return page
|
|
else:
|
|
self.logger.warning(f" -> Artikel '{page.title}' konnte nicht validiert werden.")
|
|
return None
|
|
except wikipedia.exceptions.PageError:
|
|
self.logger.error(f" -> Fehler: Gefundene URL '{url_candidate}' führte zu keiner gültigen Wikipedia-Seite.")
|
|
return None
|
|
except Exception as e:
|
|
self.logger.error(f" -> Unerwarteter Fehler bei der Verarbeitung der Seite '{url_candidate}': {e}")
|
|
return None
|
|
|
|
def _extract_first_paragraph_from_soup(self, soup):
|
|
"""
|
|
Extrahiert den ersten aussagekraeftigen Absatz aus dem Soup-Objekt eines Wikipedia-Artikels.
|
|
"""
|
|
if not soup: return "k.A."
|
|
paragraph_text = "k.A."
|
|
try:
|
|
content_div = soup.find('div', class_='mw-parser-output')
|
|
search_area = content_div if content_div else soup
|
|
paragraphs = search_area.find_all('p', recursive=False)
|
|
if not paragraphs: paragraphs = search_area.find_all('p')
|
|
|
|
for p in paragraphs:
|
|
for sup in p.find_all('sup', class_='reference'): sup.decompose()
|
|
for span in p.find_all('span', style=lambda v: v and 'display:none' in v): span.decompose()
|
|
for span in p.find_all('span', id='coordinates'): span.decompose()
|
|
text = clean_text(p.get_text(separator=' ', strip=True))
|
|
if text != "k.A." and len(text) > 50 and not re.match(r'^(Datei:|Abbildung:|Siehe auch:|Einzelnachweise|Siehe auch|Literatur)', text, re.IGNORECASE):
|
|
paragraph_text = text[:1500]
|
|
break
|
|
except Exception as e:
|
|
self.logger.error(f"Fehler beim Extrahieren des ersten Absatzes: {e}")
|
|
return paragraph_text
|
|
|
|
def extract_categories(self, soup):
|
|
"""
|
|
Extrahiert Wikipedia-Kategorien aus dem Soup-Objekt.
|
|
"""
|
|
if not soup: return "k.A."
|
|
cats_filtered = []
|
|
try:
|
|
cat_div = soup.find('div', id="mw-normal-catlinks")
|
|
if cat_div:
|
|
ul = cat_div.find('ul')
|
|
if ul:
|
|
cats = [clean_text(li.get_text()) for li in ul.find_all('li')]
|
|
cats_filtered = [c for c in cats if c and isinstance(c, str) and c.strip() and "kategorien:" not in c.lower()]
|
|
except Exception as e:
|
|
self.logger.error(f"Fehler beim Extrahieren der Kategorien: {e}")
|
|
return ", ".join(cats_filtered) if cats_filtered else "k.A."
|
|
|
|
def _extract_infobox_value(self, soup, target):
|
|
"""
|
|
Extrahiert gezielt Werte (Branche, Umsatz, etc.) aus der Infobox.
|
|
"""
|
|
if not soup or target not in self.keywords_map:
|
|
return "k.A."
|
|
keywords = self.keywords_map[target]
|
|
infobox = soup.select_one('table[class*="infobox"]')
|
|
if not infobox: return "k.A."
|
|
|
|
value_found = "k.A."
|
|
try:
|
|
rows = infobox.find_all('tr')
|
|
for row in rows:
|
|
cells = row.find_all(['th', 'td'], recursive=False)
|
|
header_text, value_cell = None, None
|
|
|
|
if len(cells) >= 2:
|
|
if cells[0].name == 'th':
|
|
header_text, value_cell = cells[0].get_text(strip=True), cells[1]
|
|
elif cells[0].name == 'td' and cells[1].name == 'td':
|
|
style = cells[0].get('style', '').lower()
|
|
is_header_like = 'font-weight' in style and ('bold' in style or '700' in style) or cells[0].find(['b', 'strong'], recursive=False)
|
|
if is_header_like:
|
|
header_text, value_cell = cells[0].get_text(strip=True), cells[1]
|
|
|
|
if header_text and value_cell:
|
|
if any(kw in header_text.lower() for kw in keywords):
|
|
for sup in value_cell.find_all(['sup', 'span']):
|
|
sup.decompose()
|
|
|
|
raw_value_text = value_cell.get_text(separator=' ', strip=True)
|
|
|
|
if target == 'branche' or target == 'sitz':
|
|
value_found = clean_text(raw_value_text).split('\n')[0].strip()
|
|
elif target == 'umsatz':
|
|
value_found = extract_numeric_value(raw_value_text, is_umsatz=True)
|
|
elif target == 'mitarbeiter':
|
|
value_found = extract_numeric_value(raw_value_text, is_umsatz=False)
|
|
|
|
value_found = value_found if value_found else "k.A."
|
|
self.logger.info(f" --> Infobox '{target}' gefunden: '{value_found}'")
|
|
break
|
|
except Exception as e:
|
|
self.logger.exception(f"Fehler beim Durchlaufen der Infobox-Zeilen fuer '{target}': {e}")
|
|
return "k.A. (Fehler Extraktion)"
|
|
|
|
return value_found
|
|
|
|
def _parse_sitz_string_detailed(self, raw_sitz_string_input):
|
|
"""
|
|
Versucht, aus einem rohen Sitz-String Stadt und Land detailliert zu extrahieren.
|
|
"""
|
|
sitz_stadt_val, sitz_land_val = "k.A.", "k.A."
|
|
if not raw_sitz_string_input or not isinstance(raw_sitz_string_input, str):
|
|
return {'sitz_stadt': sitz_stadt_val, 'sitz_land': sitz_land_val}
|
|
|
|
temp_sitz = raw_sitz_string_input.strip()
|
|
if not temp_sitz or temp_sitz.lower() == "k.a.":
|
|
return {'sitz_stadt': sitz_stadt_val, 'sitz_land': sitz_land_val}
|
|
|
|
# Diese Mappings könnten in die Config ausgelagert werden
|
|
known_countries_detailed = {
|
|
"deutschland": "Deutschland", "germany": "Deutschland", "de": "Deutschland",
|
|
"österreich": "Österreich", "austria": "Österreich", "at": "Österreich",
|
|
"schweiz": "Schweiz", "switzerland": "Schweiz", "ch": "Schweiz", "suisse": "Schweiz",
|
|
"usa": "USA", "u.s.": "USA", "united states": "USA", "vereinigte staaten": "USA",
|
|
"vereinigtes königreich": "Vereinigtes Königreich", "united kingdom": "Vereinigtes Königreich", "uk": "Vereinigtes Königreich",
|
|
}
|
|
region_to_country = {
|
|
"nrw": "Deutschland", "nordrhein-westfalen": "Deutschland", "bayern": "Deutschland", "hessen": "Deutschland",
|
|
"zg": "Schweiz", "zug": "Schweiz", "zh": "Schweiz", "zürich": "Schweiz",
|
|
"ca": "USA", "california": "USA", "ny": "USA", "new york": "USA",
|
|
}
|
|
|
|
extracted_country = ""
|
|
original_temp_sitz = temp_sitz
|
|
|
|
klammer_match = re.search(r'\(([^)]+)\)$', temp_sitz)
|
|
if klammer_match:
|
|
suffix_in_klammer = klammer_match.group(1).strip().lower()
|
|
if suffix_in_klammer in known_countries_detailed:
|
|
extracted_country = known_countries_detailed[suffix_in_klammer]
|
|
temp_sitz = temp_sitz[:klammer_match.start()].strip(" ,")
|
|
elif suffix_in_klammer in region_to_country:
|
|
extracted_country = region_to_country[suffix_in_klammer]
|
|
temp_sitz = temp_sitz[:klammer_match.start()].strip(" ,")
|
|
|
|
if not extracted_country and ',' in temp_sitz:
|
|
parts = [p.strip() for p in temp_sitz.split(',')]
|
|
if len(parts) > 1:
|
|
last_part_lower = parts[-1].lower()
|
|
if last_part_lower in known_countries_detailed:
|
|
extracted_country = known_countries_detailed[last_part_lower]
|
|
temp_sitz = ", ".join(parts[:-1]).strip(" ,")
|
|
elif last_part_lower in region_to_country:
|
|
extracted_country = region_to_country[last_part_lower]
|
|
temp_sitz = ", ".join(parts[:-1]).strip(" ,")
|
|
|
|
sitz_land_val = extracted_country if extracted_country else "k.A."
|
|
sitz_stadt_val = re.sub(r'^\d{4,8}\s*', '', temp_sitz).strip(" ,")
|
|
|
|
if not sitz_stadt_val:
|
|
sitz_stadt_val = "k.A." if sitz_land_val != "k.A." else re.sub(r'^\d{4,8}\s*', '', original_temp_sitz).strip(" ,") or "k.A."
|
|
|
|
return {'sitz_stadt': sitz_stadt_val, 'sitz_land': sitz_land_val}
|
|
|
|
@retry_on_failure
|
|
def extract_company_data(self, url_or_page):
|
|
"""
|
|
Extrahiert strukturierte Unternehmensdaten aus einem Wikipedia-Artikel (URL oder page-Objekt).
|
|
Gibt nun auch den gesamten Rohtext des Artikels ('full_text') und den Titel zurück.
|
|
"""
|
|
default_result = {
|
|
'url': 'k.A.', 'title': 'k.A.', 'sitz_stadt': 'k.A.', 'sitz_land': 'k.A.',
|
|
'first_paragraph': 'k.A.', 'branche': 'k.A.', 'umsatz': 'k.A.',
|
|
'mitarbeiter': 'k.A.', 'categories': 'k.A.', 'full_text': ''
|
|
}
|
|
page = None
|
|
|
|
try:
|
|
if isinstance(url_or_page, str) and "wikipedia.org" in url_or_page:
|
|
page_title = unquote(url_or_page.split('/wiki/')[-1].replace('_', ' '))
|
|
page = wikipedia.page(title=page_title, auto_suggest=False, redirect=True)
|
|
elif not isinstance(url_or_page, str): # Annahme: es ist ein page-Objekt
|
|
page = url_or_page
|
|
else:
|
|
self.logger.warning(f"extract_company_data: Ungültiger Input '{str(url_or_page)[:100]}...'.")
|
|
return default_result
|
|
|
|
self.logger.info(f"Extrahiere Daten für Wiki-Artikel: {page.title[:100]}...")
|
|
|
|
# Grundlegende Daten direkt aus dem page-Objekt extrahieren
|
|
first_paragraph = page.summary.split('\n')[0] if page.summary else 'k.A.'
|
|
categories = ", ".join(page.categories)
|
|
full_text = page.content
|
|
|
|
# Für Infobox-Daten benötigen wir weiterhin BeautifulSoup, da die 'wikipedia'-Bibliothek
|
|
# keinen strukturierten Zugriff darauf bietet.
|
|
soup = self._get_page_soup(page.url)
|
|
if not soup:
|
|
self.logger.warning(f" -> Konnte Seite für Soup-Parsing nicht laden. Extrahiere nur Basis-Daten.")
|
|
# Fallback, wenn Soup fehlschlägt
|
|
return {
|
|
'url': page.url, 'title': page.title, 'sitz_stadt': 'k.A.', 'sitz_land': 'k.A.',
|
|
'first_paragraph': first_paragraph, 'branche': 'k.A.', 'umsatz': 'k.A.',
|
|
'mitarbeiter': 'k.A.', 'categories': categories, 'full_text': full_text
|
|
}
|
|
|
|
# Extraktion der Infobox-Daten mit den bestehenden Helper-Funktionen
|
|
branche_val = self._extract_infobox_value(soup, 'branche')
|
|
umsatz_val = self._extract_infobox_value(soup, 'umsatz')
|
|
mitarbeiter_val = self._extract_infobox_value(soup, 'mitarbeiter')
|
|
raw_sitz_string = self._extract_infobox_value(soup, 'sitz')
|
|
parsed_sitz = self._parse_sitz_string_detailed(raw_sitz_string)
|
|
sitz_stadt_val = parsed_sitz['sitz_stadt']
|
|
sitz_land_val = parsed_sitz['sitz_land']
|
|
|
|
# Sammle die finalen Daten
|
|
result = {
|
|
'url': page.url,
|
|
'title': page.title,
|
|
'sitz_stadt': sitz_stadt_val,
|
|
'sitz_land': sitz_land_val,
|
|
'first_paragraph': first_paragraph,
|
|
'branche': branche_val,
|
|
'umsatz': umsatz_val,
|
|
'mitarbeiter': mitarbeiter_val,
|
|
'categories': categories,
|
|
'full_text': full_text
|
|
}
|
|
|
|
self.logger.info(f" -> Extrahierte Daten: Stadt='{sitz_stadt_val}', Land='{sitz_land_val}', U='{umsatz_val}', M='{mitarbeiter_val}'")
|
|
return result
|
|
|
|
except wikipedia.exceptions.PageError:
|
|
self.logger.error(f" -> Fehler: Wikipedia-Artikel für '{str(url_or_page)[:100]}' konnte nicht gefunden werden (PageError).")
|
|
return {**default_result, 'url': str(url_or_page) if isinstance(url_or_page, str) else 'k.A.'}
|
|
except Exception as e:
|
|
self.logger.error(f" -> Unerwarteter Fehler bei der Extraktion von '{str(url_or_page)[:100]}': {e}")
|
|
return {**default_result, 'url': str(url_or_page) if isinstance(url_or_page, str) else 'k.A.'} |