diff --git a/brancheneinstufung.py b/brancheneinstufung.py index fbfd7407..4d2d530a 100644 --- a/brancheneinstufung.py +++ b/brancheneinstufung.py @@ -241,17 +241,20 @@ def retry_on_failure(func): return None # Sollte nicht erreicht werden, aber zur Sicherheit return wrapper -@retry_on_failure # Annahme: Decorator existiert -def serp_wikipedia_lookup(company_name, website=None): +@retry_on_failure +def serp_wikipedia_lookup(company_name, website=None, min_similarity=0.5): # Mindestähnlichkeit hinzugefügt """ - Sucht über SerpAPI (Google) nach dem wahrscheinlichsten Wikipedia-Artikel für ein Unternehmen. + Sucht über SerpAPI (Google) nach dem wahrscheinlichsten Wikipedia-Artikel. + Sammelt Top-Kandidaten und wählt den mit der höchsten Titelähnlichkeit aus. Args: company_name (str): Der Name des Unternehmens. - website (str, optional): Die Website des Unternehmens zur Kontextverbesserung. Defaults to None. + website (str, optional): Die Website des Unternehmens. Defaults to None. + min_similarity (float, optional): Mindestähnlichkeit zwischen Firmenname + und Wiki-Artikeltitel. Defaults to 0.5. Returns: - str: Die URL des ersten gefundenen Wikipedia-Artikels oder None. + str: Die URL des relevantesten Wikipedia-Artikels oder None. """ serp_key = Config.API_KEYS.get('serpapi') if not serp_key: @@ -261,58 +264,75 @@ def serp_wikipedia_lookup(company_name, website=None): logging.warning("serp_wikipedia_lookup: Kein Firmenname angegeben.") return None - # Query Konstruktion: Name + "Wikipedia" sollte meistens gut funktionieren query = f'"{company_name}" Wikipedia' - # Optional: Website hinzufügen, wenn vorhanden und sinnvoll? Kann aber auch einschränken. - # if website and website != "k.A.": - # query = f'"{company_name}" "{website}" Wikipedia' - logging.info(f"Starte SerpAPI Wikipedia-Suche für '{company_name}' mit Query: '{query}'") params = { - "engine": "google", - "q": query, - "api_key": serp_key, - "hl": "de", # Sprache Deutsch bevorzugen - "gl": "de" # Ergebnisse aus Deutschland bevorzugen + "engine": "google", "q": query, "api_key": serp_key, + "hl": "de", "gl": "de", + "num": 10 # Frage mehr Ergebnisse an, um Auswahl zu haben } api_url = "https://serpapi.com/search" try: - response = requests.get(api_url, params=params, timeout=15) # Etwas längerer Timeout + response = requests.get(api_url, params=params, timeout=15) response.raise_for_status() data = response.json() - # Durchsuche organische Ergebnisse nach dem ersten Wikipedia-Link + candidates = [] if "organic_results" in data: - for result in data["organic_results"]: + logging.debug(f" -> Prüfe {len(data['organic_results'])} organische Ergebnisse...") + for result in data["organic_results"][:5]: # Nur die Top 5 prüfen link = result.get("link") - displayed_link = result.get("displayed_link", "").lower() - # Prüfe, ob es ein Wikipedia-Link ist (flexibler Check) - if link and "wikipedia.org" in link.lower(): - # Zusätzliche Prüfung: Ist es wahrscheinlich die Hauptseite des Artikels? - # '/wiki/' ist ein starker Indikator. - if "/wiki/" in link: - logging.info(f" -> SerpAPI: Wikipedia-Link gefunden: {link}") - # Keine weitere Normalisierung hier, gib die gefundene URL zurück - return link - else: - logging.debug(f" -> SerpAPI: Überspringe Link ohne '/wiki/': {link}") - # Manchmal ist der Link selbst keine Wiki-URL, aber der angezeigte Link schon - elif displayed_link and "wikipedia.org" in displayed_link and link: - if "/wiki/" in link: # Stelle sicher, dass der *echte* Link auch auf einen Artikel zeigt - logging.info(f" -> SerpAPI: Wikipedia-Link über Displayed Link gefunden: {link}") - return link + # Prüfe, ob es ein gültiger Wiki-Artikel-Link ist + if link and "wikipedia.org" in link.lower() and "/wiki/" in link \ + and not any(x in link for x in ['Datei:', 'Spezial:', 'Portal:', 'Hilfe:', 'Diskussion:']): + logging.debug(f" -> Kandidaten-URL gefunden: {link}") + candidates.append(link) - logging.warning(f" -> SerpAPI: Kein passender Wikipedia-Link in organischen Ergebnissen für '{company_name}' gefunden.") - return None + if not candidates: + logging.warning(f" -> SerpAPI: Keine Wikipedia-Kandidaten-URLs in Top-Ergebnissen für '{company_name}' gefunden.") + return None + + # Bewerte Kandidaten basierend auf Titelähnlichkeit + best_match_url = None + highest_similarity = -1.0 + normalized_search_name = normalize_company_name(company_name) + + for url in candidates: + try: + # Extrahiere Titel aus URL (vereinfacht, ohne vollen API-Call für Performance) + # Annahme: Titel ist der Teil nach /wiki/ mit Underscores statt Leerzeichen + title_part = url.split('/wiki/', 1)[1] + title = unquote(title_part).replace('_', ' ') + normalized_title = normalize_company_name(title) + + similarity = SequenceMatcher(None, normalized_title, normalized_search_name).ratio() + logging.debug(f" -> Prüfe Ähnlichkeit für '{title}' (Norm: '{normalized_title}'): {similarity:.2f}") + + # Aktualisiere besten Treffer, wenn Ähnlichkeit höher UND über Schwelle + if similarity > highest_similarity and similarity >= min_similarity: + highest_similarity = similarity + best_match_url = url + logging.debug(f" -> Neuer bester Kandidat: {best_match_url} (Ähnlichkeit: {highest_similarity:.2f})") + + except Exception as e_title: + logging.warning(f" -> Fehler beim Extrahieren/Vergleichen des Titels für URL {url}: {e_title}") + continue # Nächsten Kandidaten prüfen + + if best_match_url: + logging.info(f" -> SerpAPI: Bester relevanter Wikipedia-Link ausgewählt: {best_match_url} (Ähnlichkeit: {highest_similarity:.2f})") + return best_match_url + else: + logging.warning(f" -> SerpAPI: Keiner der gefundenen Wikipedia-Links ({candidates}) erreichte die Mindestähnlichkeit ({min_similarity}) für '{company_name}'.") + return None except requests.exceptions.RequestException as e: logging.error(f"Fehler bei der SerpAPI Wikipedia Suche für '{company_name}': {e}") - raise e # Fehler weitergeben für Retry + raise e except Exception as e: logging.error(f"Allgemeiner Fehler bei der SerpAPI Wikipedia Suche für '{company_name}': {e}") - return None # Bei unerwarteten Fehlern None zurückgeben + return None # Kann als eigenständige Funktion oder Methode in DataProcessor implementiert werden def process_find_wiki_with_serp(sheet_handler, row_limit=None, min_employees=500):