This commit is contained in:
2025-04-18 14:08:09 +00:00
parent 6d2a50dcdf
commit b33bea2dbd

View File

@@ -1394,290 +1394,210 @@ class WikipediaScraper:
debug_print(f"Fehler beim Setzen der Wikipedia-Sprache: {e}")
def _get_full_domain(self, website):
"""Extrahiert Domain (ohne www, ohne Pfad) aus URL."""
if not website or not isinstance(website, str): return ""
# Nutze die normalisierte URL
normalized_url = simple_normalize_url(website)
if normalized_url == "k.A.": return ""
# Entferne 'www.' falls vorhanden
if normalized_url.startswith("www."):
return normalized_url[4:]
return normalized_url
# (Unverändert zu deiner Version)
if not website: return ""; website = website.lower().strip()
website = re.sub(r'^https?:\/\/', '', website); website = re.sub(r'^www\.', '', website)
return website.split('/')[0]
def _generate_search_terms(self, company_name, website):
"""Generiert Suchbegriffe für Wikipedia."""
terms = set() # Verwende Set, um Duplikate zu vermeiden
# 1. Domain (ohne www)
full_domain = self._get_full_domain(website)
if full_domain:
terms.add(full_domain.split('.')[0]) # Nur der Domain-Name selbst
# 2. Normalisierter Firmenname (verschiedene Längen)
# (Unverändert zu deiner Version - leicht optimiert mit Set)
terms = set(); full_domain = self._get_full_domain(website)
if full_domain: terms.add(full_domain) # Ganze Domain kann helfen
normalized_name = normalize_company_name(company_name)
if normalized_name:
name_parts = normalized_name.split()
if len(name_parts) > 0:
terms.add(name_parts[0]) # Erstes Wort
if len(name_parts) > 1:
terms.add(" ".join(name_parts[:2])) # Erste zwei Worte
terms.add(normalized_name) # Ganzer normalisierter Name
# 3. Original Firmenname (falls abweichend und nicht zu lang)
original_name_cleaned = clean_text(company_name).lower()
if original_name_cleaned != normalized_name and len(original_name_cleaned) < 50:
terms.add(original_name_cleaned)
# Filter leere Strings und konvertiere zu Liste
final_terms = [term for term in terms if term]
debug_print(f"Generierte Wikipedia-Suchbegriffe für '{company_name}': {final_terms}")
if len(name_parts) > 0: terms.add(name_parts[0])
if len(name_parts) > 1: terms.add(" ".join(name_parts[:2]))
terms.add(normalized_name)
if company_name and company_name.lower() not in terms: terms.add(company_name.lower()) # Original auch dazu
final_terms = [term for term in list(terms)[:5] if term] # Max 5 Begriffe
debug_print(f"Generierte Suchbegriffe: {final_terms}")
return final_terms
# retry_on_failure für Requests hinzufügen
@retry_on_failure
def _fetch_page_content(self, page_title):
"""Lädt eine Wikipedia-Seite sicher."""
def _get_page_soup(self, url):
"""Holt HTML und gibt BeautifulSoup-Objekt zurück."""
try:
# Nutze page() mit auto_suggest=False und preload=True für Effizienz
page = wikipedia.page(page_title, auto_suggest=False, preload=True)
return page
except wikipedia.exceptions.PageError:
debug_print(f"Wikipedia PageError: Seite '{page_title}' nicht gefunden.")
response = requests.get(url, timeout=10)
response.raise_for_status()
response.encoding = response.apparent_encoding # Encoding korrigieren
return BeautifulSoup(response.text, Config.HTML_PARSER)
except requests.exceptions.RequestException as e:
debug_print(f"Fehler beim Abrufen von HTML von {url}: {e}")
return None
except wikipedia.exceptions.DisambiguationError as e:
debug_print(f"Wikipedia DisambiguationError für '{page_title}': {e.options[:5]}")
# Optional: Versuche, die erste Option automatisch zu wählen?
# try:
# return wikipedia.page(e.options[0], auto_suggest=False, preload=True)
# except Exception as inner_e:
# debug_print(f"Fehler beim Laden der ersten Disambiguation-Option: {inner_e}")
# return None
return None # Vorerst keine automatische Auswahl
except Exception as e:
debug_print(f"Allgemeiner Fehler beim Laden der Wikipedia-Seite '{page_title}': {e}")
debug_print(f"Fehler beim Parsen von HTML von {url}: {e}")
return None
@retry_on_failure
def _fetch_page_html(self, page_url):
""" Lädt HTML einer Seite für manuelles Parsing. """
try:
response = requests.get(page_url, timeout=10)
response.raise_for_status()
return response.text
except requests.exceptions.RequestException as e:
debug_print(f"Fehler beim Abrufen von HTML von {page_url}: {e}")
return None
def _validate_article(self, page, company_name, website):
"""Prüft Ähnlichkeit Titel vs. Name und ob Domain im Artikel vorkommt."""
if not page: return False
page_title = page.title
normalized_title = normalize_company_name(page_title)
normalized_company = normalize_company_name(company_name)
# 1. Ähnlichkeitsprüfung der Namen
name_similarity = fuzzy_similarity(normalized_title, normalized_company)
debug_print(f"Namensähnlichkeit für '{page_title}': {name_similarity:.2f} ('{normalized_title}' vs '{normalized_company}')")
# 2. Domain-Prüfung
full_domain = self._get_full_domain(website)
domain_found = False
if full_domain:
# (Unverändert zu deiner Version, außer Nutzung von _get_page_soup)
full_domain = self._get_full_domain(website); domain_found = False
if full_domain and page: # Prüfe ob page existiert
try:
# Prüfe externe Links zuerst (effizienter)
if hasattr(page, 'externallinks'):
for ext_link in page.externallinks:
if full_domain in ext_link.lower():
debug_print(f"Domain '{full_domain}' in externem Link gefunden: {ext_link}")
domain_found = True
break
# Wenn nicht gefunden, prüfe Infobox (aufwändiger, erfordert HTML-Parsing)
if not domain_found:
html_content = self._fetch_page_html(page.url)
if html_content:
soup = BeautifulSoup(html_content, Config.HTML_PARSER)
infobox = soup.find('table', class_=lambda c: c and 'infobox' in c.lower())
if infobox:
links = infobox.find_all('a', href=True)
for link in links:
href = link.get('href', '').lower()
# Suche nach der Domain in externen Links innerhalb der Infobox
if full_domain in href and ('http://' in href or 'https://' in href):
debug_print(f"Domain '{full_domain}' in Infobox-Link gefunden: {href}")
domain_found = True
break
except Exception as e:
debug_print(f"Fehler bei der Domain-Validierung für '{page_title}': {e}")
# 3. Entscheidung
# Hohe Ähnlichkeit ODER moderate Ähnlichkeit UND Domain gefunden
threshold = Config.SIMILARITY_THRESHOLD
if name_similarity >= threshold + 0.1: # Bei sehr hoher Ähnlichkeit
debug_print(f"Validierung OK (Hohe Namensähnlichkeit): {page_title}")
return True
if name_similarity >= threshold - 0.1 and domain_found: # Bei moderater Ähnlichkeit, wenn Domain passt
debug_print(f"Validierung OK (Moderate Ähnlichkeit + Domain gefunden): {page_title}")
return True
soup = self._get_page_soup(page.url) # Nutze neue Hilfsfunktion
if soup:
infobox = soup.find('table', class_=lambda c: c and 'infobox' in c.lower())
if infobox:
# (Link-Extraktion wie gehabt) ...
links = infobox.find_all('a', href=True)
for link in links:
href = link.get('href','').lower()
if href.startswith(('/wiki/datei:', '#')) : continue # Skip Datei- und interne Links
if full_domain in href: debug_print(f"Link-Match Infobox: {href}"); domain_found = True; break
# Prüfe externe Links nur, wenn in Infobox nichts war UND page.externallinks existiert
if not domain_found and hasattr(page, 'externallinks'):
for ext_link in page.externallinks:
if full_domain in ext_link.lower(): debug_print(f"Link-Match ExtLinks: {ext_link}"); domain_found = True; break
except Exception as e: debug_print(f"Fehler Link-Extraktion: {e}")
debug_print(f"Validierung fehlgeschlagen für '{page_title}' (Ähnlichkeit: {name_similarity:.2f}, Domain gefunden: {domain_found})")
return False
# Ähnlichkeitsberechnung (wie gehabt)
normalized_title = normalize_company_name(page.title); normalized_company = normalize_company_name(company_name)
similarity = SequenceMatcher(None, normalized_title, normalized_company).ratio()
debug_print(f"Ähnlichkeit: {similarity:.2f} ('{normalized_title}' vs '{normalized_company}') für {page.title}")
# Schwellenwert-Logik (wie gehabt)
threshold = 0.60 if domain_found else Config.SIMILARITY_THRESHOLD
is_valid = similarity >= threshold
if is_valid: debug_print(f" => Validiert (Schwelle: {threshold:.2f})")
else: debug_print(f" => Nicht validiert (Schwelle: {threshold:.2f})")
return is_valid
# Diese separate Funktion wird nicht mehr benötigt, da wir den soup schon haben
# def extract_first_paragraph(self, page_url): ...
def extract_first_paragraph(self, page_content):
"""Extrahiert den ersten sinnvollen Absatz aus dem Seiteninhalt."""
if not page_content: return "k.A."
# Nutze page.summary, da dies oft der erste Absatz ist
summary = clean_text(page_content)
if len(summary) > 50:
# Begrenze Länge, um nicht zu viel Text zu haben
return summary[:1000] # Max 1000 Zeichen
def _extract_first_paragraph_from_soup(self, soup):
"""Extrahiert ersten Absatz aus vorhandenem Soup-Objekt."""
if not soup: return "k.A."
# Suche nach dem ersten <p>-Tag, der direkt unter dem Hauptinhalt (z.B. div.mw-parser-output) liegt,
# oder einfach den ersten <p> nach der Infobox? Sicherer ist oft der erste <p> generell.
paragraphs = soup.find_all('p', recursive=True) # Finde alle <p>
for p in paragraphs:
# Ignoriere leere <p> oder solche in Tabellen/Sidebars etc. (optional)
# if p.find_parent(['table', 'aside']): continue
text = clean_text(p.get_text())
if len(text) > 50: # Nimm den ersten Absatz mit signifikanter Länge
return text[:1000] # Begrenze Länge
return "k.A."
def _extract_infobox_data(self, page_url):
"""Extrahiert Branche, Umsatz, Mitarbeiter aus der Infobox (via HTML)."""
data = {'branche': 'k.A.', 'umsatz': 'k.A.', 'mitarbeiter': 'k.A.'}
html_content = self._fetch_page_html(page_url)
if not html_content: return data
try:
soup = BeautifulSoup(html_content, Config.HTML_PARSER)
# Finde Infobox (flexiblere Suche nach Klassen)
infobox = soup.find('table', class_=lambda c: c and any(kw in c.lower() for kw in ['infobox', 'vcard', 'unternehmen']))
if not infobox: return data
# Definiere Keywords für jede Information
keywords_map = {
'branche': ['branche', 'industrie', 'tätigkeit', 'geschäftsfeld', 'sektor', 'produkte', 'leistungen', 'wirtschaftszweig'],
'umsatz': ['umsatz', 'jahresumsatz', 'erlöse', 'umsatzerlöse', 'einnahmen', 'ergebnis'],
'mitarbeiter': ['mitarbeiter', 'beschäftigte', 'personal', 'mitarbeiterzahl', 'angestellte', 'belegschaft']
}
rows = infobox.find_all('tr')
for row in rows:
header = row.find('th')
value_cell = row.find('td')
if header and value_cell:
header_text = clean_text(header.get_text()).lower()
raw_value_text = value_cell.get_text(separator=' ', strip=True) # Text aus der Zelle holen
# Suche nach Keywords in der Kopfzeile
for key, keywords in keywords_map.items():
if any(kw in header_text for kw in keywords):
# Wenn ein Keyword passt, verarbeite den Wert
if key == 'branche':
# Für Branche: Bereinige Referenzen und Klammern, bevor clean_text
cleaned_branch = re.sub(r'\[.*?\]|\(.*?\)', '', raw_value_text)
data['branche'] = clean_text(cleaned_branch)
elif key == 'umsatz':
data['umsatz'] = extract_numeric_value(raw_value_text, is_umsatz=True)
elif key == 'mitarbeiter':
data['mitarbeiter'] = extract_numeric_value(raw_value_text, is_umsatz=False)
# Optional: break, wenn ein Wert für diese Zeile gefunden wurde?
# break # Verhindert, dass z.B. "Umsatz" auch als "Ergebnis" interpretiert wird, falls beide Keywords passen
# Fallback: Manchmal steht die Branche ohne explizites th da
if data['branche'] == 'k.A.':
possible_branches = infobox.select('tr > td[colspan="2"]') # Suche nach Zellen über 2 Spalten
for pb in possible_branches:
pb_text = clean_text(pb.get_text())
# Prüfe, ob Text nach Branche aussieht (keine Zahlen, nicht zu lang)
if pb_text and not any(char.isdigit() for char in pb_text) and len(pb_text) < 100:
is_likely_branch = True
for kw_list in keywords_map.values(): # Nicht mit anderen Keywords verwechseln
if any(kw in pb_text.lower() for kw in kw_list):
is_likely_branch = False
break
if is_likely_branch:
data['branche'] = pb_text
break
except Exception as e:
debug_print(f"Fehler beim Parsen der Infobox von {page_url}: {e}")
return data
def extract_categories(self, page_url):
"""Extrahiert Kategorien (via HTML)."""
html_content = self._fetch_page_html(page_url)
if not html_content: return "k.A."
try:
soup = BeautifulSoup(html_content, Config.HTML_PARSER)
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')]
# Filtere leere Kategorien und Standardkategorien
cats = [cat for cat in cats if cat and cat != "Kategorien:" and "Wikipedia:" not in cat]
return ", ".join(cats) if cats else "k.A."
except Exception as e:
debug_print(f"Fehler beim Extrahieren der Kategorien von {page_url}: {e}")
def extract_categories(self, soup):
# (Unverändert zu deiner Version)
if not soup: return "k.A."
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') if clean_text(li.get_text()) and "Kategorien:" not in clean_text(li.get_text())]
return ", ".join(cats) if cats else "k.A."
return "k.A."
def _extract_infobox_value(self, soup, target):
# (Unverändert zu deiner Version)
if not soup: return "k.A."
infobox = soup.find('table', class_=lambda c: c and any(kw in c.lower() for kw in ['infobox', 'vcard', 'unternehmen']))
if not infobox: return "k.A."
keywords_map = { # ... (wie gehabt)
'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'],
'mitarbeiter': ['mitarbeiter', 'beschäftigte', 'personal', 'mitarbeiterzahl', 'angestellte', 'belegschaft', 'personalstärke']
}
keywords = keywords_map.get(target, [])
for row in infobox.find_all('tr'):
header = row.find('th'); value_cell = row.find('td')
if header and value_cell:
header_text = clean_text(header.get_text()).lower()
if any(kw in header_text for kw in keywords):
raw_value = value_cell.get_text(separator=' ', strip=True) # Nimm Text inkl. Unterelementen
cleaned_raw_value = clean_text(raw_value) # Bereinige Whitespace etc.
if target == 'branche':
# Entferne Referenzen etc. NACH get_text
clean_val = re.sub(r'\[\d+\]', '', cleaned_raw_value) # Entferne [1], [2]
clean_val = re.sub(r'\([^)]*\)', '', clean_val) # Entferne (...)
return clean_val.strip() if clean_val else "k.A."
elif target == 'umsatz': return extract_numeric_value(cleaned_raw_value, is_umsatz=True)
elif target == 'mitarbeiter': return extract_numeric_value(cleaned_raw_value, is_umsatz=False)
return "k.A."
# Diese Funktionen sind jetzt überflüssig, da _extract_infobox_value die Arbeit macht
# def extract_full_infobox(self, soup): ...
# def extract_fields_from_infobox_text(self, infobox_text, field_names): ...
# --- NEUE, KOMBINIERTE extract_company_data ---
def extract_company_data(self, page_url):
"""Extrahiert alle relevanten Daten von einer Wikipedia-Seite."""
default_data = {
'url': page_url if page_url else 'k.A.',
'first_paragraph': 'k.A.', 'branche': 'k.A.', 'umsatz': 'k.A.',
'mitarbeiter': 'k.A.', 'categories': 'k.A.'
}
if not page_url or page_url == 'k.A.':
return default_data
"""
Extrahiert Firmendaten von einer Wikipedia-URL. Holt die Seite nur einmal.
Priorisiert das Parsen von th/td in der Infobox.
"""
default_result = {'url': page_url if page_url else 'k.A.', 'first_paragraph': 'k.A.', 'branche': 'k.A.', 'umsatz': 'k.A.', 'mitarbeiter': 'k.A.', 'categories': 'k.A.'}
if not page_url or not isinstance(page_url, str) or "wikipedia.org" not in page_url:
return default_result
# Lade Seiteninhalt über die wikipedia library (für summary)
page = self._fetch_page_content(page_url.split('/')[-1]) # Nutze Titel aus URL
if not page:
# Wenn Seite nicht geladen werden kann, HTML trotzdem versuchen zu parsen
debug_print(f"Konnte Seite '{page_url}' nicht über Wikipedia-Lib laden, versuche HTML-Parsing.")
# Extrahiere Daten, die HTML benötigen
infobox_data = self._extract_infobox_data(page_url)
categories_val = self.extract_categories(page_url)
# Extrahiere Absatz (nutze page.summary wenn verfügbar, sonst leer)
first_paragraph = self.extract_first_paragraph(page.summary) if page else "k.A."
debug_print(f"Extrahiere Daten für Wiki-URL: {page_url}")
soup = self._get_page_soup(page_url) # Hole und parse die Seite EINMAL
# Kombiniere Ergebnisse
company_data = {
if not soup:
debug_print(" -> Fehler: Konnte Seite nicht laden oder parsen.")
# Optional: Versuche, page object für summary zu holen? Eher nicht, wenn HTML fehlscglug.
return default_result
# --- Extrahiere Daten aus dem Soup-Objekt ---
first_paragraph = self._extract_first_paragraph_from_soup(soup)
categories_val = self.extract_categories(soup)
# Nutze _extract_infobox_value für robustere Extraktion
branche_val = self._extract_infobox_value(soup, 'branche')
umsatz_val = self._extract_infobox_value(soup, 'umsatz')
mitarbeiter_val = self._extract_infobox_value(soup, 'mitarbeiter')
# --- Ergebnis zusammenstellen ---
result = {
'url': page_url,
'first_paragraph': first_paragraph,
'branche': infobox_data['branche'],
'umsatz': infobox_data['umsatz'],
'mitarbeiter': infobox_data['mitarbeiter'],
'branche': branche_val,
'umsatz': umsatz_val,
'mitarbeiter': mitarbeiter_val,
'categories': categories_val
# 'full_infobox' wird nicht mehr extrahiert
}
# debug_print(f"Extrahierte Wiki-Daten für {page_url}: {company_data}")
return company_data
debug_print(f" -> Extrahierte Daten: P={first_paragraph[:30]}..., B='{branche_val}', U='{umsatz_val}', M='{mitarbeiter_val}', C={categories_val[:30]}...")
return result
# retry_on_failure ist hier schon drauf
# --- search_company_article bleibt fast gleich, nutzt aber intern _get_page_soup ---
@retry_on_failure
def search_company_article(self, company_name, website):
"""Sucht nach einem passenden Wikipedia-Artikel."""
"""Sucht einen passenden Wikipedia-Artikel und gibt das page-Objekt zurück."""
search_terms = self._generate_search_terms(company_name, website)
if not search_terms:
debug_print("Keine Suchbegriffe generiert, Wikipedia-Suche übersprungen.")
return None
if not search_terms: return None
for term in search_terms:
try:
# wikipedia.search gibt Titel zurück
results = wikipedia.search(term, results=Config.WIKIPEDIA_SEARCH_RESULTS)
debug_print(f"Wikipedia-Suchergebnisse für '{term}': {results}")
debug_print(f"Suchergebnisse für '{term}': {results}")
for title in results:
page = self._fetch_page_content(title)
if page and self._validate_article(page, company_name, website):
debug_print(f"Passenden Wikipedia-Artikel gefunden: {page.url}")
return page # Gib das Page-Objekt zurück
except Exception as e:
# Fehler bei der Suche selbst (Netzwerk etc.)
debug_print(f"Fehler während der Wikipedia-Suche für '{term}': {e}")
try:
# Versuche, das Page-Objekt zu laden (für Validierung mit page.externallinks etc.)
page = wikipedia.page(title, auto_suggest=False, preload=True) # Preload für Effizienz
# Validierung prüft jetzt intern mit _get_page_soup
if self._validate_article(page, company_name, website):
debug_print(f"Valider Artikel gefunden: {page.url}")
return page # Gib das Page-Objekt zurück
# else: Artikel gefunden, aber nicht validiert -> nächsten Titel prüfen
except wikipedia.exceptions.PageError:
debug_print(f" -> Seite '{title}' nicht gefunden (PageError).")
continue
except wikipedia.exceptions.DisambiguationError as e:
debug_print(f" -> Seite '{title}' ist Begriffsklärung: {e.options[:3]}...")
# Optional: Versuche ersten Link der Begriffsklärung? Vorerst nicht.
continue
except Exception as e_page:
# Andere Fehler beim Laden/Validieren einer einzelnen Seite
debug_print(f" -> Fehler bei Verarbeitung von Titel '{title}': {e_page}")
continue # Zum nächsten Titel
except Exception as e_search:
# Fehler bei der Suche selbst (z.B. Netzwerk zu Wikipedia)
debug_print(f"Fehler während Wikipedia-Suche für '{term}': {e_search}")
# Hier nicht abbrechen, sondern nächsten Suchbegriff versuchen
continue # Zum nächsten Suchbegriff
debug_print(f"Kein passender Wikipedia-Artikel für '{company_name}' gefunden.")
debug_print(f"Kein passender Wikipedia-Artikel für '{company_name}' gefunden nach Prüfung aller Begriffe.")
return None
# ==================== WEBSITE SCRAPING ====================