From aa8ee04c878040fed7d173a5cb02e49e25d4d5c3 Mon Sep 17 00:00:00 2001 From: Floke Date: Sun, 20 Jul 2025 06:53:02 +0000 Subject: [PATCH] wikipedia_scraper.py aktualisiert --- wikipedia_scraper.py | 200 ++++++++++++++++++++++--------------------- 1 file changed, 101 insertions(+), 99 deletions(-) diff --git a/wikipedia_scraper.py b/wikipedia_scraper.py index 0102bd2a..8c6be4df 100644 --- a/wikipedia_scraper.py +++ b/wikipedia_scraper.py @@ -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):