diff --git a/brancheneinstufung.py b/brancheneinstufung.py index bd4e2611..9d4a9c10 100644 --- a/brancheneinstufung.py +++ b/brancheneinstufung.py @@ -1505,13 +1505,24 @@ class GoogleSheetHandler: class WikipediaScraper: """ Handles searching Wikipedia articles and extracting relevant company data. - Version: 1.6.5 - Improved infobox parsing reliability. + Version: 1.6.5 logic - Improved infobox parsing reliability and detailed logging. """ - def __init__(self, user_agent=None): # Optional User-Agent übergeben - # Verwende User-Agent aus Config, falls nicht anders angegeben - self.user_agent = user_agent or getattr(Config, 'USER_AGENT', 'Mozilla/5.0 (compatible; YourBotName/1.0; +http://yourwebsite.com/botinfo)') + def __init__(self, user_agent=None): + """ + Initialisiert den Scraper mit einer Requests-Session und Logger. + """ + # --- Logging explizit für diese Klasse holen und Level setzen --- + # Holt Logger basierend auf dem Modulnamen (__name__ wird hier zu 'WikipediaScraper' oder dem Dateinamen) + self.logger = logging.getLogger(__name__) + # Das Level wird global in main() gesetzt, hier zur Sicherheit nochmal debug loggen + self.logger.debug(f"Logger für WikipediaScraper ('{__name__}') initialisiert.") + # --- Ende Logging-Anpassung --- + + # Verwende User-Agent aus Config oder einen Standardwert + self.user_agent = user_agent or getattr(Config, 'USER_AGENT', 'Mozilla/5.0 (compatible; Datenanreicherungsskript/1.0)') self.session = requests.Session() self.session.headers.update({'User-Agent': self.user_agent}) + self.logger.debug(f"Requests Session mit User-Agent '{self.user_agent}' initialisiert.") # Erweiterte Keyword Map für robustere Infobox-Extraktion self.keywords_map = { @@ -1520,348 +1531,351 @@ class WikipediaScraper: 'mitarbeiter': ['mitarbeiter', 'mitarbeiterzahl', 'beschäftigte', 'employees', 'number of employees', 'personal', 'belegschaft'] } try: - wikipedia.set_lang(Config.LANG) + wiki_lang = getattr(Config, 'LANG', 'de') # Sprache aus Config holen + wikipedia.set_lang(wiki_lang) wikipedia.set_rate_limiting(True) # Respektiere Wikipedia API Limits - debug_print(f"Wikipedia library language set to '{Config.LANG}'.") + self.logger.info(f"Wikipedia library language set to '{wiki_lang}'. Rate limiting enabled.") except Exception as e: - debug_print(f"WARNUNG: Fehler beim Setzen der Wikipedia-Sprache oder Rate Limiting: {e}") + self.logger.warning(f"Fehler beim Setzen der Wikipedia-Sprache oder Rate Limiting: {e}") - # Beibehalten: Hilfsfunktion für Domain def _get_full_domain(self, website): + """Extrahiert die normalisierte Domain (ohne www, ohne Pfad) aus einer URL.""" if not website or not isinstance(website, str): return "" - website = website.lower().strip() - website = re.sub(r'^https?:\/\/', '', website) - website = re.sub(r'^www\.', '', website) - return website.split('/')[0] + website_lower = website.lower().strip() + if not website_lower or website_lower == 'k.a.': return "" + # Schema entfernen + website_lower = re.sub(r'^https?:\/\/', '', website_lower) + # User:Pass entfernen + if '@' in website_lower: website_lower = website_lower.split('@', 1)[1] + # www. entfernen + if website_lower.startswith('www.'): website_lower = website_lower[4:] + # Pfad und Port entfernen + domain = website_lower.split('/')[0].split(':')[0] + return domain - # Beibehalten: Suchbegriffe generieren def _generate_search_terms(self, company_name, website): - terms = set(); full_domain = self._get_full_domain(website) - if full_domain: terms.add(full_domain) - normalized_name = normalize_company_name(company_name) # Annahme: normalize_company_name existiert + """Generiert eine Liste von Suchbegriffen für die Wikipedia-Suche.""" + if not company_name: return [] + terms = set() + full_domain = self._get_full_domain(website) + if full_domain: terms.add(full_domain) # Domain als Suchbegriff + + # Normalisierten Namen hinzufügen (Annahme: Funktion existiert) + normalized_name = normalize_company_name(company_name) if normalized_name: name_parts = normalized_name.split() - if len(name_parts) > 0: terms.add(name_parts[0]) - if len(name_parts) > 1: terms.add(" ".join(name_parts[:2])) - terms.add(normalized_name) - # Füge Originalnamen hinzu, falls er signifikant anders ist - if company_name and company_name.lower() != normalized_name and company_name.lower() not in terms: - terms.add(company_name.lower()) + if len(name_parts) > 0: terms.add(name_parts[0]) # Erstes Wort + if len(name_parts) > 1: terms.add(" ".join(name_parts[:2])) # Erste zwei Worte + terms.add(normalized_name) # Ganzer normalisierter Name - # Nimm nur die ersten paar sinnvollen Begriffe + # Originalnamen hinzufügen, falls signifikant anders + company_name_lower = company_name.lower() + if company_name_lower != normalized_name and company_name_lower not in terms: + terms.add(company_name_lower) + + # Nimm nur sinnvolle (nicht leere) und begrenzte Anzahl Begriffe final_terms = [term for term in list(terms) if term][:5] # Max 5 Begriffe - debug_print(f"Generierte Suchbegriffe für '{company_name}': {final_terms}") + self.logger.debug(f"Generierte Suchbegriffe für '{company_name}': {final_terms}") return final_terms - # retry_on_failure Decorator hier hinzufügen! - @retry_on_failure + @retry_on_failure # Annahme: Decorator existiert und behandelt Exceptions def _get_page_soup(self, url): """Holt HTML von einer URL und gibt ein BeautifulSoup-Objekt zurück.""" if not url or not isinstance(url, str) or not url.startswith("http"): - logger.warning(f"_get_page_soup: Ungültige URL '{url}'.") + self.logger.warning(f"_get_page_soup: Ungültige URL '{url}'.") return None try: - logger.debug(f"_get_page_soup: Rufe URL ab: {url}") - response = self.session.get(url, timeout=15) # Timeout erhöhen + self.logger.debug(f"_get_page_soup: Rufe URL ab: {url}") + response = self.session.get(url, timeout=20) # Timeout etwas höher response.raise_for_status() # Fehler für 4xx/5xx auslösen - # Encoding explizit auf UTF-8 setzen, da Wikipedia dies verwendet + # Encoding explizit auf UTF-8 setzen (Standard für Wikipedia) response.encoding = 'utf-8' - soup = BeautifulSoup(response.text, Config.HTML_PARSER) # Nutze Parser aus Config - logger.debug(f"_get_page_soup: Parsen von {url} erfolgreich.") + soup = BeautifulSoup(response.text, getattr(Config, 'HTML_PARSER', 'html.parser')) # Nutze Parser aus Config + self.logger.debug(f"_get_page_soup: Parsen von {url} erfolgreich.") return soup except requests.exceptions.Timeout: - logger.error(f"_get_page_soup: Timeout beim Abrufen von {url}") - return None + self.logger.error(f"_get_page_soup: Timeout beim Abrufen von {url}") + raise # Fehler weitergeben für Retry except requests.exceptions.RequestException as e: - logger.error(f"_get_page_soup: Fehler beim Abrufen von HTML von {url}: {e}") - # Hier keinen Fehler weitergeben, damit der Prozess weiterläuft, aber None zurückgeben - return None + self.logger.error(f"_get_page_soup: Netzwerkfehler beim Abrufen von HTML von {url}: {e}") + raise e # Fehler weitergeben für Retry except Exception as e: - logger.error(f"_get_page_soup: Fehler beim Parsen von HTML von {url}: {e}") - return None + self.logger.error(f"_get_page_soup: Fehler beim Parsen von HTML von {url}: {e}") + # Parsing-Fehler nicht unbedingt wiederholen, None zurückgeben? Oder doch für Retry? + # Fürs Erste weitergeben, falls temporäres Problem + raise e - # Überarbeitete Validierung def _validate_article(self, page, company_name, website): """ Validiert, ob ein Wikipedia-Artikel zum Unternehmen passt. Prüft Titelähnlichkeit und ob die Firmenwebsite verlinkt ist. - Nutzt jetzt _get_page_soup für die Link-Prüfung. """ if not page or not company_name: return False - debug_print(f"Validiere Artikel '{page.title}' für Firma '{company_name}' (Website: {website})...") + self.logger.debug(f"Validiere Artikel '{page.title}' für Firma '{company_name}' (Website: {website})...") full_domain = self._get_full_domain(website) - normalized_company = normalize_company_name(company_name) # Annahme: normalize_company_name existiert + normalized_company = normalize_company_name(company_name) normalized_title = normalize_company_name(page.title) - # 1. Titelähnlichkeit prüfen + # 1. Titelähnlichkeit similarity = SequenceMatcher(None, normalized_title, normalized_company).ratio() - debug_print(f" -> Titelähnlichkeit: {similarity:.2f} ('{normalized_title}' vs '{normalized_company}')") + self.logger.debug(f" -> Titelähnlichkeit: {similarity:.2f} ('{normalized_title}' vs '{normalized_company}')") # 2. Link-Prüfung (nur wenn Domain vorhanden) domain_found = False if full_domain: - debug_print(f" -> Suche nach Domain '{full_domain}' in Links...") + self.logger.debug(f" -> Suche nach Domain '{full_domain}' in Links von {page.url}...") soup = self._get_page_soup(page.url) # Hole Soup für Link-Prüfung if soup: - # Suche zuerst in der Infobox (häufigster Ort für offizielle Links) - infobox = soup.select_one('table[class*="infobox"]') # Flexibler Selektor + # Suche zuerst in der Infobox + infobox = soup.select_one('table[class*="infobox"]') if infobox: website_links = infobox.find_all('a', href=True) for link in website_links: href = link.get('href', '') - # Prüfe, ob der Link auf eine externe Seite geht und die Domain enthält - if href.startswith('http') and full_domain in href.lower(): - # Zusätzliche Prüfung: Ist der Link-Text relevant? (Optional) + if href.startswith('http') and full_domain in self._get_full_domain(href): # Vergleiche Domains link_text = link.get_text(strip=True).lower() - if any(kw in link_text for kw in ['website', 'webseite', 'offizielle']): - debug_print(f" -> Domain '{full_domain}' in Infobox-Link gefunden (Text: '{link_text}', URL: {href})") + # Prüfe, ob Keyword im Link-Text ODER in der Tabellenüberschrift der Zeile + th = link.find_previous('th') + th_text = th.get_text(strip=True).lower() if th else "" + if any(kw in link_text for kw in ['website', 'webseite', 'offizielle']) or \ + any(kw in th_text for kw in ['website', 'webseite', 'webauftritt']): + self.logger.debug(f" -> Domain '{full_domain}' in Infobox-Link gefunden (TH: '{th_text}', Text: '{link_text}', URL: {href})") domain_found = True break - else: # Auch akzeptieren, wenn der Text nicht passt, aber die URL klar ist - debug_print(f" -> Domain '{full_domain}' in Infobox-Link gefunden (URL: {href})") + else: # Akzeptiere auch ohne Keyword, wenn URL eindeutig scheint + self.logger.debug(f" -> Domain '{full_domain}' in Infobox-Link gefunden (URL: {href}, kein Keyword-Match im Text/TH)") domain_found = True break - # Wenn nicht in Infobox, suche in allen externen Links des Artikels + # Wenn nicht in Infobox, suche in allen externen Links if not domain_found: - # Die wikipedia library liefert externe Links manchmal nicht zuverlässig, daher besser alle Links parsen - all_links = soup.find_all('a', href=True) + self.logger.debug(" -> Domain nicht in Infobox-Links gefunden, suche in allen externen Links...") + all_links = soup.find_all('a', href=True, class_=re.compile(r'.*\bexternal\b.*')) # Suche explizit externe Links + if not all_links: # Fallback: alle Links prüfen + all_links = soup.find_all('a', href=True) + for link in all_links: href = link.get('href', '') - # Klasse 'external text' ist ein guter Indikator, aber nicht immer vorhanden - is_external = link.has_attr('class') and any(c in ['external', 'text'] for c in link['class']) - # Prüfe, ob es ein externer Link ist und die Domain enthält - if href.startswith('http') and full_domain in href.lower(): + if href.startswith('http') and full_domain in self._get_full_domain(href): # Ignoriere Links zu anderen Wikimedia-Projekten etc. - if not any(site in href for site in ['wikipedia.org', 'wikimedia.org', 'wikidata.org']): - debug_print(f" -> Domain '{full_domain}' in externem Link gefunden (URL: {href}, External Class? {is_external})") + if not any(site in href for site in ['wikipedia.org', 'wikimedia.org', 'wikidata.org', 'archive.org', 'webcitation.org']): + self.logger.debug(f" -> Domain '{full_domain}' in externem Link gefunden (URL: {href})") domain_found = True break # Erster Treffer reicht else: - debug_print(f" -> Konnte HTML für Link-Prüfung von {page.url} nicht laden.") + self.logger.warning(f" -> Konnte HTML für Link-Prüfung von {page.url} nicht laden.") else: - debug_print(" -> Keine Website-Domain für Link-Prüfung vorhanden.") + self.logger.debug(" -> Keine Website-Domain für Link-Prüfung vorhanden.") - # 3. Entscheidung basierend auf Ähnlichkeit und Link - threshold = Config.SIMILARITY_THRESHOLD # Standard-Schwelle + # 3. Entscheidung + threshold = getattr(Config, 'SIMILARITY_THRESHOLD', 0.65) # Standard-Schwelle aus Config if domain_found: - threshold = max(0.4, threshold - 0.2) # Schwelle lockern, wenn Domain passt (aber nicht zu sehr) - debug_print(f" -> Domain gefunden, Schwelle angepasst auf {threshold:.2f}") + # Schwelle signifikant lockern, wenn Domain passt + threshold = max(0.35, threshold - 0.3) # Beispiel: 0.65 -> 0.35 + self.logger.debug(f" -> Domain gefunden, Ähnlichkeitsschwelle angepasst auf {threshold:.2f}") is_valid = similarity >= threshold - if is_valid: - debug_print(f" => Artikel '{page.title}' VALIDiert (Ähnlichkeit >= {threshold:.2f}, Domain gefunden? {domain_found})") - else: - debug_print(f" => Artikel '{page.title}' NICHT validiert (Ähnlichkeit < {threshold:.2f}, Domain gefunden? {domain_found})") + log_level = logging.INFO if is_valid else logging.DEBUG # Logge Erfolg als INFO + self.logger.log(log_level, f" => Artikel '{page.title}' {'VALIDIERT' if is_valid else 'NICHT validiert'} (Ähnlichkeit={similarity:.2f}, Schwelle={threshold:.2f}, Domain gefunden? {domain_found})") return is_valid - # Beibehalten: Extrahiere Kategorien def extract_categories(self, soup): + """Extrahiert Wikipedia-Kategorien aus dem Soup-Objekt.""" if not soup: return "k.A." + cats_filtered = [] try: cat_div = soup.find('div', id="mw-normal-catlinks") if cat_div: ul = cat_div.find('ul') if ul: - # Extrahiere Text, bereinige ihn und filtere den "Kategorien:" Link selbst cats = [clean_text(li.get_text()) for li in ul.find_all('li')] cats_filtered = [c for c in cats if c and "kategorien:" not in c.lower()] - return ", ".join(cats_filtered) if cats_filtered else "k.A." + self.logger.debug(f"Kategorien gefunden: {cats_filtered}") + else: self.logger.debug("Kein 'ul' in 'mw-normal-catlinks' gefunden.") + else: self.logger.debug("Kein 'div#mw-normal-catlinks' gefunden.") except Exception as e: - logger.error(f"Fehler beim Extrahieren der Kategorien: {e}") - return "k.A." + self.logger.error(f"Fehler beim Extrahieren der Kategorien: {e}") + + return ", ".join(cats_filtered) if cats_filtered else "k.A." - # Beibehalten: Extrahiere ersten Absatz def _extract_first_paragraph_from_soup(self, soup): + """Extrahiert den ersten aussagekräftigen Absatz aus dem Soup-Objekt.""" if not soup: return "k.A." + paragraph_text = "k.A." try: - # Finde das Haupt-Inhaltsdiv (üblich bei MediaWiki) + # Suche bevorzugt im Hauptinhalt content_div = soup.find('div', class_='mw-parser-output') - if not content_div: - content_div = soup.find('div', id='bodyContent') # Fallback - if not content_div: - content_div = soup # Fallback auf ganzen Soup + search_area = content_div if content_div else soup # Fallback auf ganzen Soup - # Suche nach dem ersten

-Tag, der direkten Text enthält (keine leeren

) - paragraphs = content_div.find_all('p', recursive=False) # Nur direkte Kinder suchen? Oft besser - if not paragraphs: - paragraphs = content_div.find_all('p', recursive=True) # Fallback: alle

+ # Finde alle

-Tags direkt unter dem Suchbereich (oft relevanter) + paragraphs = search_area.find_all('p', recursive=False) + if not paragraphs: # Fallback: Alle

-Tags suchen + paragraphs = search_area.find_all('p', recursive=True) + self.logger.debug(f"Suche ersten Absatz in {len(paragraphs)} gefundenen

-Tags...") - for p in paragraphs: - # Ignoriere

-Tags, die nur ein Bild oder eine Tabelle enthalten (heuristisch) - if p.find(['img', 'table', 'figure'], recursive=False): continue - # Entferne Referenz-Links [1], [2] etc. VOR der Textextraktion + for idx, p in enumerate(paragraphs): + # Überspringe leere oder zu kurze Absätze + if not p.get_text(strip=True): + self.logger.debug(f" -> Überspringe leeres

Tag (Index {idx})") + continue + # Überspringe Absätze, die nur Bilder/Tabellen etc. enthalten könnten + if p.find(['img', 'table', 'figure', 'div'], recursive=False): + self.logger.debug(f" -> Überspringe

Tag (Index {idx}), da er Blockelemente enthält.") + continue + + # Entferne Referenz-Links `[1]`, `[2]` etc. for sup in p.find_all('sup', class_='reference'): sup.decompose() - # Extrahiere und bereinige Text - text = p.get_text(separator=' ', strip=True) - text = clean_text(text) # Annahme: clean_text entfernt u.a. doppelte Leerzeichen - # Nimm den ersten Absatz mit signifikanter Länge - if text and len(text) > 30: # Mindestlänge, um leere oder irrelevante

zu überspringen - logger.debug(f"Ersten Absatz gefunden: {text[:100]}...") - return text[:1000] # Begrenze Länge - except Exception as e: - logger.error(f"Fehler beim Extrahieren des ersten Absatzes: {e}") - return "k.A." + # Entferne Koordinaten (oft am Anfang in ) + for span in p.find_all('span', id='coordinates'): + span.decompose() + + text = clean_text(p.get_text(separator=' ', strip=True)) + + # Prüfe Mindestlänge + if text and len(text) > 40: # Mindestlänge etwas erhöht + self.logger.debug(f" -> Ersten gültigen Absatz (Index {idx}) gefunden: {text[:100]}...") + paragraph_text = text[:1000] # Begrenze Länge + break # Nimm den ersten passenden + else: + self.logger.debug(f" -> Überspringe

Tag (Index {idx}), Text zu kurz oder leer nach clean_text: '{text[:50]}...'") + + except Exception as e: + self.logger.error(f"Fehler beim Extrahieren des ersten Absatzes: {e}") + + if paragraph_text == "k.A.": + self.logger.warning("Kein passender erster Absatz gefunden.") + return paragraph_text - # --- NEUE, VERBESSERTE Infobox-Extraktion --- def _extract_infobox_value(self, soup, target): """ - Extrahiert gezielt Branche, Umsatz oder Mitarbeiter aus der Infobox. - Robuster gegen Variationen in HTML und Keyword-Bezeichnungen. - - Args: - soup (BeautifulSoup): Das geparste HTML der Seite. - target (str): Welcher Wert gesucht wird ('branche', 'umsatz', 'mitarbeiter'). - - Returns: - str: Der gefundene und bereinigte Wert oder "k.A.". + Extrahiert gezielt Branche, Umsatz oder Mitarbeiter aus der Infobox. (v1.6.5 Logik) """ + # --- DEBUG LOG: Funktion betreten --- + self.logger.debug(f"--- Entering _extract_infobox_value for target '{target}' ---") + if not soup or target not in self.keywords_map: - logger.debug(f"_extract_infobox_value: Ungültiger Input (Soup: {soup is not None}, Target: {target})") + self.logger.debug(f"_extract_infobox_value: Ungültiger Input (Soup: {soup is not None}, Target: {target})") return "k.A." keywords = self.keywords_map[target] - logger.debug(f"_extract_infobox_value: Suche nach '{target}' mit Keywords: {keywords}") + self.logger.debug(f"_extract_infobox_value: Suche nach '{target}' mit Keywords: {keywords}") # Flexiblere Suche nach der Infobox infobox = soup.select_one('table[class*="infobox"]') - if not infobox: - logger.debug(" -> Keine Infobox mit Klasse '*infobox*' gefunden.") - # Fallback: Suche nach Tabellen mit typischen Attributen, falls Klasse fehlt - infobox = soup.find('table', {'summary': re.compile(r'Infobox', re.I)}) - if not infobox: - logger.debug(" -> Keine Infobox über summary='Infobox...' gefunden.") - return "k.A." - else: - logger.debug(" -> Infobox über summary='Infobox...' gefunden.") + # --- DEBUG LOG: Infobox gefunden? --- + if infobox: + self.logger.debug(f" -> Infobox gefunden (via select_one 'table[class*=\"infobox\"]')") else: - logger.debug(f" -> Infobox gefunden (Selektor 'table[class*=\"infobox\"]'). Erste Zeilen: {str(infobox.find('tr'))[:100]}...") + self.logger.debug(" -> KEINE Infobox via select_one 'table[class*=\"infobox\"]' gefunden.") + # Optional: Fallback-Suche (hier ausgelassen für Klarheit) + return "k.A." # Frühzeitiger Ausstieg - - value_found = "k.A." # Standardwert - - # Iteriere durch alle Zeilen (tr) der Infobox + value_found = "k.A." try: rows = infobox.find_all('tr') - logger.debug(f" -> Analysiere {len(rows)} Zeilen in der Infobox.") + self.logger.debug(f" -> Analysiere {len(rows)} Zeilen in der Infobox.") for idx, row in enumerate(rows): - header = row.find('th') # Finde die Header-Zelle - value_cell = row.find('td') # Finde die Daten-Zelle + header_cells = row.find_all('th', recursive=False) # Nur direkte th Kinder + value_cells = row.find_all('td', recursive=False) # Nur direkte td Kinder + + # Wir erwarten meistens ein th und ein td pro Zeile für Key-Value Paare + if len(header_cells) == 1 and len(value_cells) == 1: + header = header_cells[0] + value_cell = value_cells[0] - if header and value_cell: - # Normalisiere den Header-Text für den Keyword-Vergleich header_text = header.get_text(strip=True) header_text_lower = header_text.lower() - # Entferne Doppelpunkte etc. für besseren Match + # Normalisierte Version für flexibleren Match header_text_normalized = re.sub(r'[\s:]', '', header_text_lower) - # Detailliertes Log für jede geprüfte Zeile (Header+Value) - # Zeige den Wert nur gekürzt an, um Logs nicht zu fluten raw_value_preview = value_cell.get_text(strip=True)[:50] - logger.debug(f" -> Prüfe Zeile {idx}: TH='{header_text}' (Norm: '{header_text_normalized}') | TD Preview='{raw_value_preview}...'") + self.logger.debug(f" -> Prüfe Zeile {idx}: TH='{header_text}' (Norm: '{header_text_normalized}') | TD Preview='{raw_value_preview}...'") - # Prüfe, ob *irgendein* Keyword im normalisierten Header enthalten ist matched_keyword = None for kw in keywords: - # Suche nach dem Keyword als ganzes Wort oder am Anfang/Ende? - # Einfacher: `in`-Operator prüft auf Substring - if kw in header_text_lower: # Prüfe im weniger aggressiv normalisierten Text + # Prüfe ob Keyword im (nicht-normalisierten) Lowercase-Header vorkommt + if kw in header_text_lower: matched_keyword = kw break if matched_keyword: - logger.debug(f" --> Keyword '{matched_keyword}' gefunden in TH '{header_text}'!") + self.logger.debug(f" --> Keyword '{matched_keyword}' gefunden in TH '{header_text}'!") - # --- Wert extrahieren und bereinigen --- - # 1. Störende Elemente entfernen (Referenzen etc.) - for sup in value_cell.find_all(['sup', 'span']): # Entferne und ggf. überflüssige - # Spezifischer: Nur Referenzen entfernen? - if sup.has_attr('class') and 'reference' in sup['class']: - logger.debug(f" -> Entferne Referenz: {sup.get_text(strip=True)}") - sup.decompose() - # Entferne unsichtbare Elemente (manchmal in ) - elif sup.name == 'span' and sup.get('style') and 'display:none' in sup['style']: - logger.debug(f" -> Entferne unsichtbares Span: {sup.get_text(strip=True)}") - sup.decompose() + # Störende Elemente (Referenzen, unsichtbare Spans) im Value entfernen + for sup in value_cell.find_all(['sup', 'span']): + if (sup.name == 'sup' and sup.has_attr('class') and 'reference' in sup['class']) or \ + (sup.name == 'span' and sup.get('style') and 'display:none' in sup['style']): + self.logger.debug(f" -> Entferne störendes Element: {sup.get_text(strip=True)}") + sup.decompose() - - # 2. Text extrahieren (mit Leerzeichen für
) + # Text extrahieren (mit Leerzeichen für
) und bereinigen raw_value_text = value_cell.get_text(separator=' ', strip=True) - logger.debug(f" -> Roher TD-Text nach Decompose: '{raw_value_text}'") + self.logger.debug(f" -> Roher TD-Text nach Decompose: '{raw_value_text}'") + cleaned_raw_value = clean_text(raw_value_text) - # 3. Generelle Bereinigung (clean_text) - cleaned_raw_value = clean_text(raw_value_text) # Annahme: clean_text ist definiert - - # 4. Spezifische Verarbeitung je nach Zieltyp + # Spezifische Verarbeitung if target == 'branche': - # Entferne oft vorkommende Zusätze wie (Stand: ...), [1], etc. - clean_val = re.sub(r'\s*\[\d+\]', '', cleaned_raw_value) # Referenzen (falls clean_text sie nicht erwischt) - clean_val = re.sub(r'\s*\([^)]*\)', '', clean_val).strip() # Klammerzusätze - value_found = clean_val if clean_val else "k.A." - logger.debug(f" --> Branche extrahiert: '{value_found}'") - + # Einfache Bereinigung für Branche + clean_val = re.sub(r'\s*\([^)]*\)', '', cleaned_raw_value).strip() # Klammerzusätze entfernen + # Nimm nur den Teil vor einem möglichen Zeilenumbruch (manchmal gibt es Listen) + clean_val = clean_val.split('\n')[0].strip() + value_found = clean_val if clean_val else "k.A." + self.logger.debug(f" --> Branche extrahiert: '{value_found}'") elif target == 'umsatz': - # extract_numeric_value sollte robust sein - numeric_val = extract_numeric_value(cleaned_raw_value, is_umsatz=True) # Annahme: existiert - value_found = numeric_val - logger.debug(f" --> Umsatz extrahiert (aus '{cleaned_raw_value}'): '{value_found}'") - + numeric_val = extract_numeric_value(cleaned_raw_value, is_umsatz=True) + value_found = numeric_val + self.logger.debug(f" --> Umsatz extrahiert (aus '{cleaned_raw_value}'): '{value_found}'") elif target == 'mitarbeiter': - # extract_numeric_value sollte robust sein - numeric_val = extract_numeric_value(cleaned_raw_value, is_umsatz=False) # Annahme: existiert - value_found = numeric_val - logger.debug(f" --> Mitarbeiter extrahiert (aus '{cleaned_raw_value}'): '{value_found}'") + numeric_val = extract_numeric_value(cleaned_raw_value, is_umsatz=False) + value_found = numeric_val + self.logger.debug(f" --> Mitarbeiter extrahiert (aus '{cleaned_raw_value}'): '{value_found}'") - # Wenn ein Wert für das gesuchte Ziel gefunden wurde, beende die Schleife für dieses Ziel - break # WICHTIG: Verhindert Überschreiben durch spätere Zeilen - - # else: # Optional: Loggen, wenn kein Keyword passte - # logger.debug(f" --- Kein passendes Keyword in TH '{header_text}' gefunden.") + break # WICHTIG: Ersten Treffer für das gesuchte Ziel nehmen und Zeilenschleife verlassen + else: + # Logge Zeilen, die nicht dem Key-Value-Muster entsprechen (optional auf DEBUG) + # self.logger.debug(f" -> Überspringe Zeile {idx}: Struktur passt nicht (TH:{len(header_cells)}, TD:{len(value_cells)})") + pass # Ende der Zeilenschleife if value_found != "k.A.": - logger.debug(f" -> Finaler Wert für '{target}' gefunden: '{value_found}'") + self.logger.debug(f" -> Finaler Wert für '{target}' gefunden: '{value_found}'") else: - logger.debug(f" -> Kein passender Eintrag für '{target}' in der gesamten Infobox gefunden.") + self.logger.debug(f" -> Kein passender Eintrag für '{target}' in der gesamten Infobox gefunden.") except Exception as e: - logger.error(f"Fehler beim Durchlaufen der Infobox-Zeilen für '{target}': {e}") - # Gib "k.A." zurück, wenn Fehler beim Parsen auftreten - return "k.A." + self.logger.exception(f"Fehler beim Durchlaufen der Infobox-Zeilen für '{target}': {e}") + return "k.A." # Fehler beim Parsen return value_found - # --- Kombinierte Datenextraktion (ruft Helfer auf) --- def extract_company_data(self, page_url): """ - Extrahiert Firmendaten von einer Wikipedia-URL. Holt die Seite nur einmal - und verwendet die verbesserte Infobox-Extraktion. + Extrahiert Firmendaten von einer Wikipedia-URL (v1.6.5 Logik). """ default_result = {'url': page_url if page_url else 'k.A.', 'first_paragraph': 'k.A.', 'branche': 'k.A.', 'umsatz': 'k.A.', 'mitarbeiter': 'k.A.', 'categories': 'k.A.'} - if not page_url or not isinstance(page_url, str) or "wikipedia.org" not in page_url: - logger.warning(f"extract_company_data: Ungültige URL '{page_url}'.") + if not page_url or not isinstance(page_url, str) or "wikipedia.org" not in page_url.lower(): + self.logger.warning(f"extract_company_data: Ungültige URL '{page_url}'.") return default_result - logger.info(f"Extrahiere Daten für Wiki-URL: {page_url}") - soup = self._get_page_soup(page_url) # Hole und parse die Seite EINMAL + self.logger.info(f"Extrahiere Daten für Wiki-URL: {page_url}") + soup = self._get_page_soup(page_url) if not soup: - logger.error(f" -> Fehler: Konnte Seite {page_url} nicht laden oder parsen.") - # Gib Default zurück, aber behalte die URL - default_result['url'] = page_url + self.logger.error(f" -> Fehler: Konnte Seite {page_url} nicht laden oder parsen.") + default_result['url'] = page_url # Behalte URL im Ergebnis return default_result # --- Extrahiere Daten aus dem Soup-Objekt --- - logger.debug(" -> Extrahiere ersten Absatz...") + self.logger.debug(" -> Extrahiere ersten Absatz...") first_paragraph = self._extract_first_paragraph_from_soup(soup) - logger.debug(" -> Extrahiere Kategorien...") + self.logger.debug(" -> Extrahiere Kategorien...") categories_val = self.extract_categories(soup) - logger.debug(" -> Extrahiere Branche aus Infobox...") + self.logger.debug(" -> Extrahiere Branche aus Infobox...") branche_val = self._extract_infobox_value(soup, 'branche') - logger.debug(" -> Extrahiere Umsatz aus Infobox...") + self.logger.debug(" -> Extrahiere Umsatz aus Infobox...") umsatz_val = self._extract_infobox_value(soup, 'umsatz') - logger.debug(" -> Extrahiere Mitarbeiter aus Infobox...") + self.logger.debug(" -> Extrahiere Mitarbeiter aus Infobox...") mitarbeiter_val = self._extract_infobox_value(soup, 'mitarbeiter') # --- Ergebnis zusammenstellen --- @@ -1873,113 +1887,106 @@ class WikipediaScraper: 'mitarbeiter': mitarbeiter_val, 'categories': categories_val } - logger.info(f" -> Extrahierte Daten: P={first_paragraph[:30]}..., B='{branche_val}', U='{umsatz_val}', M='{mitarbeiter_val}', C={categories_val[:30]}...") + # Logge das Ergebnis (INFO Level für Übersicht) + self.logger.info(f" -> Extrahierte Daten: P={first_paragraph[:30]}..., B='{branche_val}', U='{umsatz_val}', M='{mitarbeiter_val}', C={categories_val[:30]}...") return result - # --- Artikelsuche (ruft Helfer auf) --- - # retry_on_failure hier sinnvoll, da es API-Calls (wikipedia.search/page) macht - @retry_on_failure + @retry_on_failure # Annahme: Decorator existiert def search_company_article(self, company_name, website=None): """ - Sucht einen passenden Wikipedia-Artikel und gibt das page-Objekt zurück. - Nutzt generierte Suchbegriffe und Validierung. + Sucht einen passenden Wikipedia-Artikel und gibt das page-Objekt zurück. (v1.6.5 Logik) """ if not company_name: - logger.warning("Wikipedia search skipped: No company name provided.") + self.logger.warning("Wikipedia search skipped: No company name provided.") return None search_terms = self._generate_search_terms(company_name, website) if not search_terms: - logger.warning(f"Keine Suchbegriffe für '{company_name}' generiert.") + self.logger.warning(f"Keine Suchbegriffe für '{company_name}' generiert.") return None - logger.info(f"Starte Wikipedia-Suche für '{company_name}' (Website: {website}) mit Begriffen: {search_terms}") + self.logger.info(f"Starte Wikipedia-Suche für '{company_name}' (Website: {website}) mit Begriffen: {search_terms}") - # Versuche direkten Match zuerst (oft der beste Treffer) + # Versuche direkten Match zuerst try: - logger.debug(f" -> Versuche direkten Match für '{company_name}'...") + self.logger.debug(f" -> Versuche direkten Match für '{company_name}'...") page = wikipedia.page(company_name, auto_suggest=False, preload=True) - logger.debug(f" -> Direkten Match gefunden: '{page.title}'. Validiere...") + self.logger.debug(f" -> Direkten Match gefunden: '{page.title}'. Validiere...") if self._validate_article(page, company_name, website): - logger.info(f" -> Direkter Match '{page.title}' ({page.url}) erfolgreich validiert.") + # Erfolg bereits hier geloggt durch _validate_article return page else: - logger.debug(f" -> Direkter Match '{page.title}' nicht validiert. Fahre mit Suche fort.") + self.logger.debug(f" -> Direkter Match '{page.title}' nicht validiert. Fahre mit Suche fort.") except wikipedia.exceptions.PageError: - logger.debug(f" -> Kein direkter Artikel für '{company_name}' gefunden.") + self.logger.debug(f" -> Kein direkter Artikel für '{company_name}' gefunden.") except wikipedia.exceptions.DisambiguationError as e: - logger.debug(f" -> '{company_name}' ist eine Begriffsklärungsseite. Optionen: {e.options[:3]}...") - # Optional: Prüfe die erste Option der Begriffsklärung? + self.logger.debug(f" -> '{company_name}' ist eine Begriffsklärungsseite. Optionen: {e.options[:3]}...") + # Optional: Prüfe die erste Option if e.options: try: - first_option_title = e.options[0] - logger.debug(f" -> Prüfe erste Option der Begriffsklärung: '{first_option_title}'") - page = wikipedia.page(first_option_title, auto_suggest=False, preload=True) + option_title = e.options[0] + self.logger.debug(f" -> Prüfe erste Option der Begriffsklärung: '{option_title}'") + page = wikipedia.page(option_title, auto_suggest=False, preload=True) if self._validate_article(page, company_name, website): - logger.info(f" -> Erste Option '{page.title}' ({page.url}) erfolgreich validiert.") - return page + return page # Erfolg wird von _validate_article geloggt else: - logger.debug(f" -> Erste Option '{page.title}' nicht validiert.") + self.logger.debug(f" -> Erste Option '{page.title}' nicht validiert.") except Exception as e_disamb: - logger.warning(f" -> Fehler beim Laden/Validieren der ersten Disambiguation-Option '{e.options[0]}': {e_disamb}") + self.logger.warning(f" -> Fehler beim Laden/Validieren der Disambiguation-Option '{e.options[0]}': {e_disamb}") except Exception as e_direct: - logger.error(f" -> Unerwarteter Fehler beim direkten Zugriff auf '{company_name}': {e_direct}") + # Fehler beim direkten Zugriff loggen + self.logger.error(f" -> Unerwarteter Fehler beim direkten Zugriff auf '{company_name}': {e_direct}") + # Fehler weitergeben, da es ein Problem mit der Lib/Verbindung sein könnte + raise e_direct - # Wenn direkter Match fehlschlägt oder nicht validiert, nutze die generierten Suchbegriffe - logger.debug(f" -> Starte Suche mit generierten Begriffen: {search_terms}") + # Wenn direkter Match fehlschlägt, nutze die generierten Suchbegriffe + self.logger.debug(f" -> Starte Suche mit generierten Begriffen: {search_terms}") for term in search_terms: try: - logger.debug(f" -> Suche mit Begriff: '{term}'...") - results = wikipedia.search(term, results=Config.WIKIPEDIA_SEARCH_RESULTS) - logger.debug(f" -> Suchergebnisse für '{term}': {results}") - if not results: continue # Nächster Begriff, wenn keine Ergebnisse + self.logger.debug(f" -> Suche mit Begriff: '{term}'...") + # Begrenze Anzahl Suchergebnisse + search_results = wikipedia.search(term, results=getattr(Config, 'WIKIPEDIA_SEARCH_RESULTS', 5)) + self.logger.debug(f" -> Suchergebnisse für '{term}': {search_results}") + if not search_results: continue - for title in results: - # Überspringe Titel, die wahrscheinlich keine Firmen sind (heuristisch) - if any(ignore_word in title.lower() for ignore_word in ['liste von', 'disambiguation', 'begriffsklärung']): - logger.debug(f" -> Überspringe wahrscheinlichen Nicht-Artikel: '{title}'") + for title in search_results: + # Heuristische Filterung + if any(ignore in title.lower() for ignore in ['liste von', 'begriffsklärung', '(Begriffsklärung)']): + self.logger.debug(f" -> Überspringe wahrscheinlichen Nicht-Artikel/BK: '{title}'") continue try: - logger.debug(f" -> Prüfe potenziellen Artikel: '{title}'") - # Lade Page-Objekt (preload=True für Effizienz bei Validierung) + self.logger.debug(f" -> Prüfe potenziellen Artikel: '{title}'") page = wikipedia.page(title, auto_suggest=False, preload=True) - # Validierung (nutzt jetzt intern _get_page_soup) if self._validate_article(page, company_name, website): - logger.info(f" -> Valider Artikel gefunden: '{page.title}' ({page.url})") - return page # Gib das validierte Page-Objekt zurück - # else: Artikel gefunden, aber nicht validiert -> nächsten Titel prüfen - # Minimales Delay, um nicht zu schnell zu sein - time.sleep(0.1) + # Erfolg wird von _validate_article geloggt + return page + time.sleep(0.1) # Kleines Delay except wikipedia.exceptions.PageError: - logger.debug(f" -> Seite '{title}' nicht gefunden (PageError).") + self.logger.debug(f" -> Seite '{title}' nicht gefunden (PageError).") continue except wikipedia.exceptions.DisambiguationError as e: - logger.debug(f" -> Seite '{title}' ist Begriffsklärung: {e.options[:3]}...") - # Optional: Versuche ersten Link der Begriffsklärung? + self.logger.debug(f" -> Seite '{title}' ist Begriffsklärung: {e.options[:3]}...") continue except requests.exceptions.RequestException as e_req: - logger.warning(f" -> Netzwerkfehler beim Laden/Validieren von '{title}': {e_req}. Überspringe Titel.") - # Kurze Pause bei Netzwerkfehler + self.logger.warning(f" -> Netzwerkfehler beim Laden/Validieren von '{title}': {e_req}. Überspringe Titel.") time.sleep(1) continue except Exception as e_page: - # Andere Fehler beim Laden/Validieren einer einzelnen Seite - logger.error(f" -> Fehler bei Verarbeitung von Titel '{title}': {type(e_page).__name__} - {e_page}") + self.logger.error(f" -> Fehler bei Verarbeitung von Titel '{title}': {type(e_page).__name__} - {e_page}") continue # Zum nächsten Titel except requests.exceptions.RequestException as e_search_req: - logger.error(f"Netzwerkfehler während Wikipedia-Suche für '{term}': {e_search_req}") - # Bei Netzwerkfehler während der Suche ggf. kurz warten und nächsten Begriff versuchen - time.sleep(2) - continue + self.logger.error(f"Netzwerkfehler während Wikipedia-Suche für '{term}': {e_search_req}") + time.sleep(2) # Längere Pause bei Suchfehler + # Fehler weitergeben für Retry + raise e_search_req except Exception as e_search: - # Fehler bei der Suche selbst - logger.error(f"Allgemeiner Fehler während Wikipedia-Suche für '{term}': {e_search}") - # Hier nicht abbrechen, sondern nächsten Suchbegriff versuchen - continue # Zum nächsten Suchbegriff + self.logger.error(f"Allgemeiner Fehler während Wikipedia-Suche für '{term}': {e_search}") + # Fehler weitergeben für Retry? Eher nicht, nächsten Begriff versuchen. + continue - logger.warning(f"Kein passender & validierter Wikipedia-Artikel für '{company_name}' gefunden nach Prüfung aller Begriffe.") + self.logger.warning(f"Kein passender & validierter Wikipedia-Artikel für '{company_name}' gefunden nach Prüfung aller Begriffe.") return None