From 5137e5d22eb37ea2749d5124df6ef0a9de82001b Mon Sep 17 00:00:00 2001 From: Floke Date: Mon, 4 Aug 2025 18:39:16 +0000 Subject: [PATCH] v2.0.2: feat: Implement Google-First Wikipedia Search - 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. --- wikipedia_scraper.py | 248 +++++++++++++++++++++---------------------- 1 file changed, 119 insertions(+), 129 deletions(-) diff --git a/wikipedia_scraper.py b/wikipedia_scraper.py index 4cabe93e..1291b437 100644 --- a/wikipedia_scraper.py +++ b/wikipedia_scraper.py @@ -6,7 +6,7 @@ Klasse zur Kapselung der Interaktionen mit Wikipedia, inklusive Suche, Validierung und Extraktion von Unternehmensdaten. """ -__version__ = "v2.0.1" +__version__ = "v2.0.2" import logging import re @@ -58,6 +58,55 @@ class WikipediaScraper: 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) @@ -115,157 +164,98 @@ 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, parent_name=None): + def _validate_article(self, page, company_name, website, crm_city, 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. + Validiert faktenbasiert, ob ein Wikipedia-Artikel zum Unternehmen passt. + Priorisiert harte Fakten (Domain, Sitz) vor reiner Namensähnlichkeit. """ - if not page or not company_name: + 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 - - self.logger.debug(f"Validiere Artikel '{page.title[:100]}...' fuer Firma '{company_name[:100]}'") - # --- Stufe 1: Parent-Validierung (höchste Priorität) --- + # --- 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: - # Ü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})") + self.logger.info(f" => VALIDATION SUCCESS (Parent Match): Parent-Name '{parent_name}' im Artikel gefunden.") return True - - # --- Stufe 2: Ihre bestehende, detaillierte Validierungslogik als Fallback --- + + # --- Stufe 4: Namensähnlichkeit (Fallback mit strengeren Regeln) --- 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 - - standard_threshold = getattr(Config, 'SIMILARITY_THRESHOLD', 0.65) similarity = fuzzy_similarity(normalized_title, normalized_company) - company_tokens = normalized_company.split() - title_tokens = normalized_title.split() - first_word_match = False - first_two_words_match = False - if company_tokens and title_tokens and company_tokens[0] == title_tokens[0]: - first_word_match = True - if len(company_tokens) > 1 and len(title_tokens) > 1 and company_tokens[1] == title_tokens[1]: - first_two_words_match = True + if similarity > 0.85: # Strengere Schwelle + self.logger.info(f" => VALIDATION SUCCESS (High Similarity): Hohe Namensähnlichkeit ({similarity:.2f}).") + return True - domain_found = False - full_domain = self._get_full_domain(website) - if full_domain != "k.A.": - try: - # 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-Pruefung fuer '{page.title[:100]}...': {e_link_check}") - - is_valid = False - reason = "" - 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}") - self.logger.debug(f" - Erste 2 Worte identisch: {first_two_words_match}") - - 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, "Domain gefunden UND erste 2 Worte stimmen ueberein" - elif domain_found and first_word_match and similarity >= 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, "Erste zwei Worte stimmen ueberein UND Aehnlichkeit >= 0.45" - elif domain_found and similarity >= 0.50: - is_valid, reason = True, "Domain gefunden UND Aehnlichkeit >= 0.50" - elif first_word_match and similarity >= 0.55: - is_valid, reason = True, "Erstes Wort stimmt ueberein UND Aehnlichkeit >= 0.55" - else: - 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})") - return is_valid + 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, parent_name=None, max_recursion_depth=1): + def search_company_article(self, company_name, website=None, crm_city=None, parent_name=None): """ - Sucht einen passenden Wikipedia-Artikel. Behält die komplexe Logik bei und behebt den TypeError. + 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 or str(company_name).strip() == "": + if not company_name: return None - search_terms = self._generate_search_terms(company_name, website) - if not search_terms: + 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 - 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 + # 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) - 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.") + # 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 - - 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 fuer '{company_name[:100]}...' gefunden.") - 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): """