diff --git a/brancheneinstufung.py b/brancheneinstufung.py index 4d2d530a..d1961f93 100644 --- a/brancheneinstufung.py +++ b/brancheneinstufung.py @@ -241,17 +241,18 @@ def retry_on_failure(func): return None # Sollte nicht erreicht werden, aber zur Sicherheit return wrapper -@retry_on_failure -def serp_wikipedia_lookup(company_name, website=None, min_similarity=0.5): # Mindestähnlichkeit hinzugefügt +@retry_on_failure # Annahme: Decorator existiert +def serp_wikipedia_lookup(company_name, website=None, min_score=0.4): """ Sucht über SerpAPI (Google) nach dem wahrscheinlichsten Wikipedia-Artikel. - Sammelt Top-Kandidaten und wählt den mit der höchsten Titelähnlichkeit aus. + Verwendet flexible Query, sammelt Top-10-Kandidaten, bewertet nach Titelähnlichkeit + und Keywords, bevorzugt deutsche/englische Artikel. Args: company_name (str): Der Name des Unternehmens. 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. + min_score (float, optional): Mindest-Score (Kombination aus Ähnlichkeit + und Boni) für einen gültigen Treffer. Defaults to 0.4. Returns: str: Die URL des relevantesten Wikipedia-Artikels oder None. @@ -264,13 +265,16 @@ def serp_wikipedia_lookup(company_name, website=None, min_similarity=0.5): # Min logging.warning("serp_wikipedia_lookup: Kein Firmenname angegeben.") return None - query = f'"{company_name}" Wikipedia' + # --- Flexiblere Query Konstruktion --- + # Ohne Anführungszeichen für breitere Suche + query = f'{company_name} Wikipedia' logging.info(f"Starte SerpAPI Wikipedia-Suche für '{company_name}' mit Query: '{query}'") + # --- Ende Query --- params = { "engine": "google", "q": query, "api_key": serp_key, "hl": "de", "gl": "de", - "num": 10 # Frage mehr Ergebnisse an, um Auswahl zu haben + "num": 10 # Top 10 Ergebnisse } api_url = "https://serpapi.com/search" @@ -279,60 +283,89 @@ def serp_wikipedia_lookup(company_name, website=None, min_similarity=0.5): # Min response.raise_for_status() data = response.json() - candidates = [] + candidates = [] # Liste von Dictionaries: {'url': str, 'title': str} if "organic_results" in data: logging.debug(f" -> Prüfe {len(data['organic_results'])} organische Ergebnisse...") - for result in data["organic_results"][:5]: # Nur die Top 5 prüfen + for result in data["organic_results"]: # Prüfe alle 10 link = result.get("link") - # Prüfe, ob es ein gültiger Wiki-Artikel-Link ist - if link and "wikipedia.org" in link.lower() and "/wiki/" in link \ + # Filtere gültige Wiki-Artikel-Links (de oder en) + if link and "wikipedia.org/wiki/" in link.lower() \ + and (link.startswith("https://de.wikipedia.org") or link.startswith("https://en.wikipedia.org")) \ and not any(x in link for x in ['Datei:', 'Spezial:', 'Portal:', 'Hilfe:', 'Diskussion:']): - logging.debug(f" -> Kandidaten-URL gefunden: {link}") - candidates.append(link) + try: + # Extrahiere Titel aus URL + title_part = link.split('/wiki/', 1)[1] + # Handle evtl. Anchors (#) + title_part = title_part.split('#')[0] + title = unquote(title_part).replace('_', ' ') + candidates.append({'url': link, 'title': title}) + logging.debug(f" -> Kandidat gefunden: '{title}' ({link})") + except Exception as e_title_extract: + logging.warning(f" -> Fehler beim Extrahieren des Titels aus Link {link}: {e_title_extract}") + continue # Nächsten Kandidaten prüfen if not candidates: - logging.warning(f" -> SerpAPI: Keine Wikipedia-Kandidaten-URLs in Top-Ergebnissen für '{company_name}' gefunden.") + logging.warning(f" -> SerpAPI: Keine de/en Wikipedia-Kandidaten-URLs in Ergebnissen für '{company_name}' gefunden.") return None - # Bewerte Kandidaten basierend auf Titelähnlichkeit + # Bewerte Kandidaten best_match_url = None - highest_similarity = -1.0 - normalized_search_name = normalize_company_name(company_name) + highest_score = -1.0 + normalized_search_name = normalize_company_name(company_name) # Annahme: existiert - 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) + logging.debug(f" -> Bewerte {len(candidates)} Kandidaten...") + for cand in candidates: + url = cand['url'] + title = cand['title'] + try: # Füge Try-Except um die Normalisierung hinzu + normalized_title = normalize_company_name(title) + title_lower = title.lower() # Für Keyword-Suche + except Exception as e_norm: + logging.warning(f"Fehler beim Normalisieren des Titels '{title}': {e_norm}") + continue # Überspringe diesen Kandidaten - similarity = SequenceMatcher(None, normalized_title, normalized_search_name).ratio() - logging.debug(f" -> Prüfe Ähnlichkeit für '{title}' (Norm: '{normalized_title}'): {similarity:.2f}") + # 1. Basisscore: Titelähnlichkeit + similarity = SequenceMatcher(None, normalized_title, normalized_search_name).ratio() + score = similarity + logging.debug(f" -> Kandidat '{title}': Basis-Ähnlichkeit={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})") + # 2. Bonus für Keywords im Titel + bonus = 0.0 + if "(unternehmen)" in title_lower: + bonus += 0.2 # Starker Bonus + logging.debug(" -> Bonus +0.2 für '(unternehmen)'") + elif any(form in title_lower for form in [' gmbh', ' ag', ' kg', ' ltd', ' inc', ' corp', ' s.a.', ' se', ' group']): # 'group' hinzugefügt + bonus += 0.1 # Kleinerer Bonus + logging.debug(" -> Bonus +0.1 für Rechtsform/Gruppen-Keyword") - 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 + # 3. Bonus für Sprache (Deutsch bevorzugt) + if url.startswith("https://de.wikipedia.org"): + bonus += 0.05 + logging.debug(" -> Bonus +0.05 für de.wikipedia.org") + + # Gesamtscore + total_score = score + bonus + logging.debug(f" -> Gesamtscore für '{title}': {total_score:.3f} (Ähnlichkeit={similarity:.2f}, Bonus={bonus:.2f})") + + # Aktualisiere besten Treffer + if total_score > highest_score and total_score >= min_score: + highest_score = total_score + best_match_url = url + logging.debug(f" ====> Neuer bester Kandidat: {best_match_url} (Score: {highest_score:.3f}) ====") if best_match_url: - logging.info(f" -> SerpAPI: Bester relevanter Wikipedia-Link ausgewählt: {best_match_url} (Ähnlichkeit: {highest_similarity:.2f})") + logging.info(f" -> SerpAPI: Bester relevanter Wikipedia-Link ausgewählt: {best_match_url} (Score: {highest_score:.3f})") 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}'.") + logging.warning(f" -> SerpAPI: Keiner der {len(candidates)} Kandidaten erreichte den Mindestscore ({min_score}) 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 + raise e # Fehler weitergeben für Retry except Exception as e: logging.error(f"Allgemeiner Fehler bei der SerpAPI Wikipedia Suche für '{company_name}': {e}") - return None + return None # Bei unerwarteten Fehlern None zurückgeben # 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):