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

wikipedia_scraper.py aktualisiert

Browse Source
This commit is contained in:
Floke
2025-07-20 06:53:02 +00:00
parent 01d04c1b8f
commit aa8ee04c87
1 changed files with 101 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files

200
wikipedia_scraper.py
View File

@@ -70,50 +70,30 @@ class WikipediaScraper:
if not company_name:
return []
# Basis-Normalisierung
normalized = normalize_company_name(company_name)
# NEUE LOGIK: Speziell für Namen wie "11 88 0 Solutions"
# Fügt eine Version hinzu, bei der Leerzeichen zwischen Zahlen entfernt werden.
# 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)
# Führe eine erneute, aggressivere Normalisierung durch, um Reste zu entfernen
condensed_normalized = normalize_company_name(condensed_normalized)
else:
condensed_normalized = None
search_terms = []
# Füge die kondensierte Version mit höchster Priorität hinzu, falls sie existiert
if condensed_normalized and condensed_normalized not in search_terms:
search_terms.append(condensed_normalized)
# Füge den Originalnamen und die normalisierte Version hinzu
if company_name not in search_terms:
search_terms.append(company_name)
if normalized not in search_terms:
search_terms.append(normalized)
if condensed_normalized: search_terms.append(condensed_normalized)
search_terms.append(company_name)
search_terms.append(normalized)
# Füge Teile des Namens hinzu
parts = normalized.split()
if len(parts) > 1:
if parts[0] not in search_terms: search_terms.append(parts[0])
first_two = " ".join(parts[:2])
if first_two not in search_terms: search_terms.append(first_two)
search_terms.append(parts[0])
search_terms.append(" ".join(parts[:2]))
# Füge die Website-Domain als Suchbegriff hinzu
if website:
domain = simple_normalize_url(website)
if domain != "k.A." and domain not in search_terms:
if domain != "k.A.":
search_terms.append(domain)
# Entferne Duplikate und behalte die Reihenfolge bei
unique_terms = []
for term in search_terms:
if term and term not in unique_terms:
unique_terms.append(term)
# Limitiere auf maximal 5 Suchbegriffe, um API-Calls zu sparen
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
@@ -135,15 +115,31 @@ class WikipediaScraper:
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):
def _validate_article(self, page, company_name, website, parent_name=None):
"""
Validiert, ob ein Wikipedia-Artikel zum Unternehmen passt.
v2.0: Nutzt parent_name als primäres Kriterium. Ihre bestehenden
Regeln bleiben als Fallback erhalten.
"""
if not page or not company_name: return False
if not page or not company_name:
return False
self.logger.debug(f"Validiere Artikel '{page.title[:100]}...' fuer Firma '{company_name[:100]}'")
# --- Stufe 1: Parent-Validierung (höchste Priorität) ---
normalized_parent = normalize_company_name(parent_name) if parent_name else None
if normalized_parent:
# Überprüfe Titel und den ersten Absatz (Summary) auf den Parent-Namen
page_content_for_check = (page.title + " " + page.summary).lower()
if normalized_parent in page_content_for_check:
reason = f"Parent-Name '{parent_name}' im Artikel-Titel oder -Summary gefunden."
self.logger.info(f" => Artikel '{page.title[:100]}...' VALIDIERT (Grund: {reason})")
return True
# --- Stufe 2: Ihre bestehende, detaillierte Validierungslogik als Fallback ---
normalized_company = normalize_company_name(company_name)
normalized_title = normalize_company_name(page.title)
if not normalized_company or not normalized_title:
self.logger.warning("Validierung nicht moeglich, da Normalisierung eines Namens fehlschlug.")
return False
@@ -164,28 +160,15 @@ class WikipediaScraper:
full_domain = self._get_full_domain(website)
if full_domain != "k.A.":
try:
article_html = page.html()
if article_html:
soup = BeautifulSoup(article_html, getattr(Config, 'HTML_PARSER', 'html.parser'))
external_links = soup.select('a[href^="http"]')
for link_tag in external_links:
href = link_tag.get('href', '')
if href and isinstance(href, str) and full_domain in simple_normalize_url(href):
if not any(ex in href.lower() for ex in ['wikipedia.org', 'wikimedia.org', 'wikidata.org', 'archive.org', 'webcitation.org']):
domain_found = True
break
except KeyError as e_key:
if 'extlinks' in str(e_key).lower():
self.logger.warning(f"KeyError ('{e_key}') bei Domain-Check für Artikel '{page.title[:100]}...'. Domain-Validierung übersprungen.")
else:
self.logger.error(f"Unerwarteter KeyError bei Domain-Prüfung für '{page.title[:100]}...': {e_key}")
# page.html() kann fehleranfällig sein, wir prüfen den gerenderten Text (page.content)
if page.content and full_domain in page.content.lower():
domain_found = True
except Exception as e_link_check:
self.logger.error(f"Allgemeiner Fehler waehrend der Domain-Link-Pruefung fuer '{page.title[:100]}...': {e_link_check}")
self.logger.error(f"Allgemeiner Fehler waehrend der Domain-Pruefung fuer '{page.title[:100]}...': {e_link_check}")
is_valid = False
reason = ""
# NEU: Detaillierte Debug-Ausgaben für jeden Schritt
self.logger.debug(f" Validierungs-Check für '{page.title[:50]}...':")
self.logger.debug(f" Validierungs-Check (Fallback) für '{page.title[:50]}...':")
self.logger.debug(f" - Aehnlichkeit: {similarity:.2f} (Schwelle: {standard_threshold:.2f})")
self.logger.debug(f" - Domain '{full_domain}' im Artikel gefunden: {domain_found}")
self.logger.debug(f" - Erstes Wort identisch: {first_word_match}")
@@ -194,17 +177,17 @@ class WikipediaScraper:
if similarity >= standard_threshold:
is_valid, reason = True, f"Gesamt-Aehnlichkeit ({similarity:.2f}) >= Schwelle ({standard_threshold:.2f})"
elif domain_found and first_two_words_match:
is_valid, reason = True, f"Domain gefunden UND erste 2 Worte stimmen ueberein"
is_valid, reason = True, "Domain gefunden UND erste 2 Worte stimmen ueberein"
elif domain_found and first_word_match and similarity >= 0.40:
is_valid, reason = True, f"Domain gefunden UND erstes Wort stimmt ueberein UND Aehnlichkeit >= 0.40"
is_valid, reason = True, "Domain gefunden UND erstes Wort stimmt ueberein UND Aehnlichkeit >= 0.40"
elif first_two_words_match and similarity >= 0.45:
is_valid, reason = True, f"Erste zwei Worte stimmen ueberein UND Aehnlichkeit >= 0.45"
is_valid, reason = True, "Erste zwei Worte stimmen ueberein UND Aehnlichkeit >= 0.45"
elif domain_found and similarity >= 0.50:
is_valid, reason = True, f"Domain gefunden UND Aehnlichkeit >= 0.50"
is_valid, reason = True, "Domain gefunden UND Aehnlichkeit >= 0.50"
elif first_word_match and similarity >= 0.55:
is_valid, reason = True, f"Erstes Wort stimmt ueberein UND Aehnlichkeit >= 0.55"
is_valid, reason = True, "Erstes Wort stimmt ueberein UND Aehnlichkeit >= 0.55"
else:
reason = "Keine Validierungsregel traf zu"
reason = "Keine der Fallback-Validierungsregeln traf zu"
log_level = logging.INFO if is_valid else logging.DEBUG
self.logger.log(log_level, f" => Artikel '{page.title[:100]}...' {'VALIDIERT' if is_valid else 'NICHT validiert'} (Grund: {reason})")
@@ -212,57 +195,76 @@ class WikipediaScraper:
def search_company_article(self, company_name, website=None, parent_name=None, max_recursion_depth=1):
"""
Sucht einen passenden Wikipedia-Artikel für ein Unternehmen durch eine lineare
Überprüfung generierter Suchbegriffe. Diese Methode ist einfacher und robuster
als die alte, rekursive Implementierung.
Sucht einen passenden Wikipedia-Artikel. Behält die komplexe Logik bei und behebt den TypeError.
"""
if not company_name:
self.logger.warning("Wikipedia-Suche übersprungen: Kein Firmenname angegeben.")
if not company_name or str(company_name).strip() == "":
return None
search_terms = self._generate_search_terms(company_name, website)
self.logger.info(f"Starte Wikipedia-Suche für '{company_name[:50]}...' mit Begriffen: {search_terms}")
if not search_terms:
return None
for term in search_terms:
self.logger.debug(f" -> Prüfe Suchbegriff: '{term}'")
try:
# Nutze die Suchfunktion von Wikipedia, die besser mit uneindeutigen Begriffen umgeht
# und oft direkt die richtige Seite vorschlägt.
suggested_titles = wikipedia.search(term, results=3)
if not suggested_titles:
continue
for title in suggested_titles:
self.logger.debug(f" -> Potenzieller Artikel gefunden: '{title}'")
try:
page = wikipedia.page(title, auto_suggest=False, redirect=True)
# Validiere den gefundenen Artikel mit dem vollen Kontext
if self._validate_article(page, company_name, website, parent_name):
self.logger.info(f" -> ERFOLG: Artikel '{page.title}' wurde für '{company_name}' validiert.")
return page
else:
# Artikel ist nicht relevant, fahre mit dem nächsten Vorschlag fort
continue
except wikipedia.exceptions.DisambiguationError as e:
self.logger.debug(f" -> Begriffsklärungsseite '{title}' gefunden. Prüfe Optionen...")
# Prüfe die ersten 3 Optionen der Begriffsklärungsseite
for option in e.options[:3]:
try:
page = wikipedia.page(option, auto_suggest=False, redirect=True)
if self._validate_article(page, company_name, website, parent_name):
self.logger.info(f" -> ERFOLG: Artikel '{page.title}' aus Begriffsklärung validiert.")
return page
except Exception:
continue # Ignoriere Fehler bei einzelnen Optionen
except Exception:
# Ignoriere Fehler beim Laden einer einzelnen Seite und mache weiter
continue
self.logger.info(f"Starte Wikipedia-Suche fuer '{company_name[:100]}...' mit Begriffen: {search_terms}")
processed_titles = set()
original_search_name_norm = normalize_company_name(company_name)
# Die innere Funktion "erbt" `parent_name` aus dem Scope der äußeren Funktion.
def check_page_recursive(title_to_check, current_depth):
effective_max_depth = max_recursion_depth if max_recursion_depth is not None else 2
if title_to_check in processed_titles or current_depth > effective_max_depth:
return None
except Exception as e_search:
self.logger.error(f" -> Unerwarteter Fehler bei der Suche mit Begriff '{term}': {e}")
continue # Mache mit dem nächsten Suchbegriff weiter
processed_titles.add(title_to_check)
self.logger.debug(f" -> Pruefe potenziellen Artikel: '{title_to_check[:100]}...' (Tiefe: {current_depth})")
# Ihre bestehende Logik mit fuzzy_similarity
normalized_option_title_local = normalize_company_name(title_to_check)
title_similarity_to_original = fuzzy_similarity(normalized_option_title_local, original_search_name_norm)
if current_depth > 0 and title_similarity_to_original < 0.3:
self.logger.debug(f" -> Option '{title_to_check[:100]}' hat zu geringe Ähnlichkeit ({title_similarity_to_original:.2f}). Übersprungen.")
return None
page = None
try:
page = wikipedia.page(title_to_check, auto_suggest=False, preload=False, redirect=True)
# KORRIGIERTER AUFRUF: Übergibt `parent_name` aus dem äußeren Scope
if self._validate_article(page, company_name, website, parent_name):
self.logger.info(f" -> Titel '{page.title[:100]}...' erfolgreich validiert!")
return page
else:
return None
except wikipedia.exceptions.PageError:
self.logger.debug(f" -> Artikel '{title_to_check[:100]}' nicht gefunden (PageError).")
return None
except wikipedia.exceptions.DisambiguationError as e_disamb:
self.logger.info(f" -> Begriffsklaerung '{e_disamb.title}' gefunden (Tiefe {current_depth}). Pruefe Optionen...")
if current_depth >= effective_max_depth: return None
# Ihre bestehende Logik zur Filterung von Optionen
relevant_options = []
for option in e_disamb.options:
option_lower = option.lower()
if not any(ex in option_lower for ex in ["(person)", "(familienname)"]) and len(option) < 80:
if fuzzy_similarity(normalize_company_name(option), original_search_name_norm) > 0.3:
relevant_options.append(option)
for option_to_check in relevant_options[:3]:
validated_page = check_page_recursive(option_to_check, current_depth + 1)
if validated_page: return validated_page
return None
except Exception as e_page:
# Ihre bestehende Fehlerbehandlung
title_for_log = page.title[:100] if page and hasattr(page, 'title') and page.title else title_to_check[:100]
self.logger.error(f" -> Unerwarteter Fehler bei Verarbeitung von Seite '{title_for_log}': {e_page}")
return None
# Ihre bestehende Hauptlogik der Suche
for term in search_terms:
page_found = check_page_recursive(term, 0)
if page_found: return page_found
self.logger.warning(f"Kein passender & validierter Wikipedia-Artikel für '{company_name[:50]}...' gefunden.")
self.logger.warning(f"Kein passender & validierter Wikipedia-Artikel fuer '{company_name[:100]}...' gefunden.")
return None
def _extract_first_paragraph_from_soup(self, soup):