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.
This commit is contained in:
@@ -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,156 +164,97 @@ 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[:100]}...' fuer Firma '{company_name[:100]}'")
|
||||
self.logger.debug(f"Validiere Artikel '{page.title}' für Firma '{company_name}'...")
|
||||
|
||||
# --- 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})")
|
||||
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: Ihre bestehende, detaillierte Validierungslogik als Fallback ---
|
||||
# --- 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)
|
||||
|
||||
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}")
|
||||
self.logger.debug(f" => VALIDATION FAILED: Kein harter Fakt (Domain, Sitz, Parent) und Ähnlichkeit ({similarity:.2f}) zu gering.")
|
||||
return False
|
||||
|
||||
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
|
||||
|
||||
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
|
||||
|
||||
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
|
||||
# 2. Lade und validiere den gefundenen Artikel
|
||||
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!")
|
||||
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.debug(f" -> Artikel '{title_to_check[:100]}' nicht gefunden (PageError).")
|
||||
self.logger.error(f" -> Fehler: Gefundene URL '{url_candidate}' führte zu keiner gültigen Wikipedia-Seite.")
|
||||
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.")
|
||||
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):
|
||||
|
||||
Reference in New Issue
Block a user