diff --git a/brancheneinstufung.py b/brancheneinstufung.py index 97b8e960..9a86b07a 100644 --- a/brancheneinstufung.py +++ b/brancheneinstufung.py @@ -1394,290 +1394,210 @@ class WikipediaScraper: debug_print(f"Fehler beim Setzen der Wikipedia-Sprache: {e}") def _get_full_domain(self, website): - """Extrahiert Domain (ohne www, ohne Pfad) aus URL.""" - if not website or not isinstance(website, str): return "" - # Nutze die normalisierte URL - normalized_url = simple_normalize_url(website) - if normalized_url == "k.A.": return "" - # Entferne 'www.' falls vorhanden - if normalized_url.startswith("www."): - return normalized_url[4:] - return normalized_url + # (Unverändert zu deiner Version) + if not website: return ""; website = website.lower().strip() + website = re.sub(r'^https?:\/\/', '', website); website = re.sub(r'^www\.', '', website) + return website.split('/')[0] def _generate_search_terms(self, company_name, website): - """Generiert Suchbegriffe für Wikipedia.""" - terms = set() # Verwende Set, um Duplikate zu vermeiden - - # 1. Domain (ohne www) - full_domain = self._get_full_domain(website) - if full_domain: - terms.add(full_domain.split('.')[0]) # Nur der Domain-Name selbst - - # 2. Normalisierter Firmenname (verschiedene Längen) + # (Unverändert zu deiner Version - leicht optimiert mit Set) + terms = set(); full_domain = self._get_full_domain(website) + if full_domain: terms.add(full_domain) # Ganze Domain kann helfen 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]) # Erstes Wort - if len(name_parts) > 1: - terms.add(" ".join(name_parts[:2])) # Erste zwei Worte - terms.add(normalized_name) # Ganzer normalisierter Name - - # 3. Original Firmenname (falls abweichend und nicht zu lang) - original_name_cleaned = clean_text(company_name).lower() - if original_name_cleaned != normalized_name and len(original_name_cleaned) < 50: - terms.add(original_name_cleaned) - - # Filter leere Strings und konvertiere zu Liste - final_terms = [term for term in terms if term] - debug_print(f"Generierte Wikipedia-Suchbegriffe für '{company_name}': {final_terms}") + 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) + if company_name and company_name.lower() not in terms: terms.add(company_name.lower()) # Original auch dazu + final_terms = [term for term in list(terms)[:5] if term] # Max 5 Begriffe + debug_print(f"Generierte Suchbegriffe: {final_terms}") return final_terms + # retry_on_failure für Requests hinzufügen @retry_on_failure - def _fetch_page_content(self, page_title): - """Lädt eine Wikipedia-Seite sicher.""" + def _get_page_soup(self, url): + """Holt HTML und gibt BeautifulSoup-Objekt zurück.""" try: - # Nutze page() mit auto_suggest=False und preload=True für Effizienz - page = wikipedia.page(page_title, auto_suggest=False, preload=True) - return page - except wikipedia.exceptions.PageError: - debug_print(f"Wikipedia PageError: Seite '{page_title}' nicht gefunden.") + response = requests.get(url, timeout=10) + response.raise_for_status() + response.encoding = response.apparent_encoding # Encoding korrigieren + return BeautifulSoup(response.text, Config.HTML_PARSER) + except requests.exceptions.RequestException as e: + debug_print(f"Fehler beim Abrufen von HTML von {url}: {e}") return None - except wikipedia.exceptions.DisambiguationError as e: - debug_print(f"Wikipedia DisambiguationError für '{page_title}': {e.options[:5]}") - # Optional: Versuche, die erste Option automatisch zu wählen? - # try: - # return wikipedia.page(e.options[0], auto_suggest=False, preload=True) - # except Exception as inner_e: - # debug_print(f"Fehler beim Laden der ersten Disambiguation-Option: {inner_e}") - # return None - return None # Vorerst keine automatische Auswahl except Exception as e: - debug_print(f"Allgemeiner Fehler beim Laden der Wikipedia-Seite '{page_title}': {e}") + debug_print(f"Fehler beim Parsen von HTML von {url}: {e}") return None - @retry_on_failure - def _fetch_page_html(self, page_url): - """ Lädt HTML einer Seite für manuelles Parsing. """ - try: - response = requests.get(page_url, timeout=10) - response.raise_for_status() - return response.text - except requests.exceptions.RequestException as e: - debug_print(f"Fehler beim Abrufen von HTML von {page_url}: {e}") - return None - - def _validate_article(self, page, company_name, website): - """Prüft Ähnlichkeit Titel vs. Name und ob Domain im Artikel vorkommt.""" - if not page: return False - - page_title = page.title - normalized_title = normalize_company_name(page_title) - normalized_company = normalize_company_name(company_name) - - # 1. Ähnlichkeitsprüfung der Namen - name_similarity = fuzzy_similarity(normalized_title, normalized_company) - debug_print(f"Namensähnlichkeit für '{page_title}': {name_similarity:.2f} ('{normalized_title}' vs '{normalized_company}')") - - # 2. Domain-Prüfung - full_domain = self._get_full_domain(website) - domain_found = False - if full_domain: + # (Unverändert zu deiner Version, außer Nutzung von _get_page_soup) + full_domain = self._get_full_domain(website); domain_found = False + if full_domain and page: # Prüfe ob page existiert try: - # Prüfe externe Links zuerst (effizienter) - if hasattr(page, 'externallinks'): - for ext_link in page.externallinks: - if full_domain in ext_link.lower(): - debug_print(f"Domain '{full_domain}' in externem Link gefunden: {ext_link}") - domain_found = True - break - - # Wenn nicht gefunden, prüfe Infobox (aufwändiger, erfordert HTML-Parsing) - if not domain_found: - html_content = self._fetch_page_html(page.url) - if html_content: - soup = BeautifulSoup(html_content, Config.HTML_PARSER) - infobox = soup.find('table', class_=lambda c: c and 'infobox' in c.lower()) - if infobox: - links = infobox.find_all('a', href=True) - for link in links: - href = link.get('href', '').lower() - # Suche nach der Domain in externen Links innerhalb der Infobox - if full_domain in href and ('http://' in href or 'https://' in href): - debug_print(f"Domain '{full_domain}' in Infobox-Link gefunden: {href}") - domain_found = True - break - except Exception as e: - debug_print(f"Fehler bei der Domain-Validierung für '{page_title}': {e}") - - # 3. Entscheidung - # Hohe Ähnlichkeit ODER moderate Ähnlichkeit UND Domain gefunden - threshold = Config.SIMILARITY_THRESHOLD - if name_similarity >= threshold + 0.1: # Bei sehr hoher Ähnlichkeit - debug_print(f"Validierung OK (Hohe Namensähnlichkeit): {page_title}") - return True - if name_similarity >= threshold - 0.1 and domain_found: # Bei moderater Ähnlichkeit, wenn Domain passt - debug_print(f"Validierung OK (Moderate Ähnlichkeit + Domain gefunden): {page_title}") - return True + soup = self._get_page_soup(page.url) # Nutze neue Hilfsfunktion + if soup: + infobox = soup.find('table', class_=lambda c: c and 'infobox' in c.lower()) + if infobox: + # (Link-Extraktion wie gehabt) ... + links = infobox.find_all('a', href=True) + for link in links: + href = link.get('href','').lower() + if href.startswith(('/wiki/datei:', '#')) : continue # Skip Datei- und interne Links + if full_domain in href: debug_print(f"Link-Match Infobox: {href}"); domain_found = True; break + # Prüfe externe Links nur, wenn in Infobox nichts war UND page.externallinks existiert + if not domain_found and hasattr(page, 'externallinks'): + for ext_link in page.externallinks: + if full_domain in ext_link.lower(): debug_print(f"Link-Match ExtLinks: {ext_link}"); domain_found = True; break + except Exception as e: debug_print(f"Fehler Link-Extraktion: {e}") - debug_print(f"Validierung fehlgeschlagen für '{page_title}' (Ähnlichkeit: {name_similarity:.2f}, Domain gefunden: {domain_found})") - return False + # Ähnlichkeitsberechnung (wie gehabt) + normalized_title = normalize_company_name(page.title); normalized_company = normalize_company_name(company_name) + similarity = SequenceMatcher(None, normalized_title, normalized_company).ratio() + debug_print(f"Ähnlichkeit: {similarity:.2f} ('{normalized_title}' vs '{normalized_company}') für {page.title}") + # Schwellenwert-Logik (wie gehabt) + threshold = 0.60 if domain_found else Config.SIMILARITY_THRESHOLD + is_valid = similarity >= threshold + if is_valid: debug_print(f" => Validiert (Schwelle: {threshold:.2f})") + else: debug_print(f" => Nicht validiert (Schwelle: {threshold:.2f})") + return is_valid + # Diese separate Funktion wird nicht mehr benötigt, da wir den soup schon haben + # def extract_first_paragraph(self, page_url): ... - def extract_first_paragraph(self, page_content): - """Extrahiert den ersten sinnvollen Absatz aus dem Seiteninhalt.""" - if not page_content: return "k.A." - # Nutze page.summary, da dies oft der erste Absatz ist - summary = clean_text(page_content) - if len(summary) > 50: - # Begrenze Länge, um nicht zu viel Text zu haben - return summary[:1000] # Max 1000 Zeichen + def _extract_first_paragraph_from_soup(self, soup): + """Extrahiert ersten Absatz aus vorhandenem Soup-Objekt.""" + if not soup: return "k.A." + # Suche nach dem ersten

-Tag, der direkt unter dem Hauptinhalt (z.B. div.mw-parser-output) liegt, + # oder einfach den ersten

nach der Infobox? Sicherer ist oft der erste

generell. + paragraphs = soup.find_all('p', recursive=True) # Finde alle

+ for p in paragraphs: + # Ignoriere leere

oder solche in Tabellen/Sidebars etc. (optional) + # if p.find_parent(['table', 'aside']): continue + text = clean_text(p.get_text()) + if len(text) > 50: # Nimm den ersten Absatz mit signifikanter Länge + return text[:1000] # Begrenze Länge return "k.A." - - def _extract_infobox_data(self, page_url): - """Extrahiert Branche, Umsatz, Mitarbeiter aus der Infobox (via HTML).""" - data = {'branche': 'k.A.', 'umsatz': 'k.A.', 'mitarbeiter': 'k.A.'} - html_content = self._fetch_page_html(page_url) - if not html_content: return data - - try: - soup = BeautifulSoup(html_content, Config.HTML_PARSER) - # Finde Infobox (flexiblere Suche nach Klassen) - infobox = soup.find('table', class_=lambda c: c and any(kw in c.lower() for kw in ['infobox', 'vcard', 'unternehmen'])) - if not infobox: return data - - # Definiere Keywords für jede Information - keywords_map = { - 'branche': ['branche', 'industrie', 'tätigkeit', 'geschäftsfeld', 'sektor', 'produkte', 'leistungen', 'wirtschaftszweig'], - 'umsatz': ['umsatz', 'jahresumsatz', 'erlöse', 'umsatzerlöse', 'einnahmen', 'ergebnis'], - 'mitarbeiter': ['mitarbeiter', 'beschäftigte', 'personal', 'mitarbeiterzahl', 'angestellte', 'belegschaft'] - } - - rows = infobox.find_all('tr') - for row in rows: - header = row.find('th') - value_cell = row.find('td') - if header and value_cell: - header_text = clean_text(header.get_text()).lower() - raw_value_text = value_cell.get_text(separator=' ', strip=True) # Text aus der Zelle holen - - # Suche nach Keywords in der Kopfzeile - for key, keywords in keywords_map.items(): - if any(kw in header_text for kw in keywords): - # Wenn ein Keyword passt, verarbeite den Wert - if key == 'branche': - # Für Branche: Bereinige Referenzen und Klammern, bevor clean_text - cleaned_branch = re.sub(r'\[.*?\]|\(.*?\)', '', raw_value_text) - data['branche'] = clean_text(cleaned_branch) - elif key == 'umsatz': - data['umsatz'] = extract_numeric_value(raw_value_text, is_umsatz=True) - elif key == 'mitarbeiter': - data['mitarbeiter'] = extract_numeric_value(raw_value_text, is_umsatz=False) - # Optional: break, wenn ein Wert für diese Zeile gefunden wurde? - # break # Verhindert, dass z.B. "Umsatz" auch als "Ergebnis" interpretiert wird, falls beide Keywords passen - - # Fallback: Manchmal steht die Branche ohne explizites th da - if data['branche'] == 'k.A.': - possible_branches = infobox.select('tr > td[colspan="2"]') # Suche nach Zellen über 2 Spalten - for pb in possible_branches: - pb_text = clean_text(pb.get_text()) - # Prüfe, ob Text nach Branche aussieht (keine Zahlen, nicht zu lang) - if pb_text and not any(char.isdigit() for char in pb_text) and len(pb_text) < 100: - is_likely_branch = True - for kw_list in keywords_map.values(): # Nicht mit anderen Keywords verwechseln - if any(kw in pb_text.lower() for kw in kw_list): - is_likely_branch = False - break - if is_likely_branch: - data['branche'] = pb_text - break - - except Exception as e: - debug_print(f"Fehler beim Parsen der Infobox von {page_url}: {e}") - - return data - - - def extract_categories(self, page_url): - """Extrahiert Kategorien (via HTML).""" - html_content = self._fetch_page_html(page_url) - if not html_content: return "k.A." - try: - soup = BeautifulSoup(html_content, Config.HTML_PARSER) - cat_div = soup.find('div', id="mw-normal-catlinks") - if cat_div: - ul = cat_div.find('ul') - if ul: - cats = [clean_text(li.get_text()) for li in ul.find_all('li')] - # Filtere leere Kategorien und Standardkategorien - cats = [cat for cat in cats if cat and cat != "Kategorien:" and "Wikipedia:" not in cat] - return ", ".join(cats) if cats else "k.A." - except Exception as e: - debug_print(f"Fehler beim Extrahieren der Kategorien von {page_url}: {e}") + def extract_categories(self, soup): + # (Unverändert zu deiner Version) + if not soup: return "k.A." + cat_div = soup.find('div', id="mw-normal-catlinks"); + if cat_div: + ul = cat_div.find('ul') + if ul: + cats = [clean_text(li.get_text()) for li in ul.find_all('li') if clean_text(li.get_text()) and "Kategorien:" not in clean_text(li.get_text())] + return ", ".join(cats) if cats else "k.A." return "k.A." + def _extract_infobox_value(self, soup, target): + # (Unverändert zu deiner Version) + if not soup: return "k.A." + infobox = soup.find('table', class_=lambda c: c and any(kw in c.lower() for kw in ['infobox', 'vcard', 'unternehmen'])) + if not infobox: return "k.A." + keywords_map = { # ... (wie gehabt) + 'branche': ['branche', 'industrie', 'tätigkeit', 'geschäftsfeld', 'sektor', 'produkte', 'leistungen', 'aktivitäten', 'wirtschaftszweig'], + 'umsatz': ['umsatz', 'jahresumsatz', 'konzernumsatz', 'gesamtumsatz', 'erlöse', 'umsatzerlöse', 'einnahmen', 'ergebnis', 'jahresergebnis'], + 'mitarbeiter': ['mitarbeiter', 'beschäftigte', 'personal', 'mitarbeiterzahl', 'angestellte', 'belegschaft', 'personalstärke'] + } + keywords = keywords_map.get(target, []) + for row in infobox.find_all('tr'): + header = row.find('th'); value_cell = row.find('td') + if header and value_cell: + header_text = clean_text(header.get_text()).lower() + if any(kw in header_text for kw in keywords): + raw_value = value_cell.get_text(separator=' ', strip=True) # Nimm Text inkl. Unterelementen + cleaned_raw_value = clean_text(raw_value) # Bereinige Whitespace etc. + if target == 'branche': + # Entferne Referenzen etc. NACH get_text + clean_val = re.sub(r'\[\d+\]', '', cleaned_raw_value) # Entferne [1], [2] + clean_val = re.sub(r'\([^)]*\)', '', clean_val) # Entferne (...) + return clean_val.strip() if clean_val else "k.A." + elif target == 'umsatz': return extract_numeric_value(cleaned_raw_value, is_umsatz=True) + elif target == 'mitarbeiter': return extract_numeric_value(cleaned_raw_value, is_umsatz=False) + return "k.A." + + # Diese Funktionen sind jetzt überflüssig, da _extract_infobox_value die Arbeit macht + # def extract_full_infobox(self, soup): ... + # def extract_fields_from_infobox_text(self, infobox_text, field_names): ... + + # --- NEUE, KOMBINIERTE extract_company_data --- def extract_company_data(self, page_url): - """Extrahiert alle relevanten Daten von einer Wikipedia-Seite.""" - default_data = { - '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 page_url == 'k.A.': - return default_data + """ + Extrahiert Firmendaten von einer Wikipedia-URL. Holt die Seite nur einmal. + Priorisiert das Parsen von th/td in der Infobox. + """ + 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: + return default_result - # Lade Seiteninhalt über die wikipedia library (für summary) - page = self._fetch_page_content(page_url.split('/')[-1]) # Nutze Titel aus URL - if not page: - # Wenn Seite nicht geladen werden kann, HTML trotzdem versuchen zu parsen - debug_print(f"Konnte Seite '{page_url}' nicht über Wikipedia-Lib laden, versuche HTML-Parsing.") - - # Extrahiere Daten, die HTML benötigen - infobox_data = self._extract_infobox_data(page_url) - categories_val = self.extract_categories(page_url) - - # Extrahiere Absatz (nutze page.summary wenn verfügbar, sonst leer) - first_paragraph = self.extract_first_paragraph(page.summary) if page else "k.A." + debug_print(f"Extrahiere Daten für Wiki-URL: {page_url}") + soup = self._get_page_soup(page_url) # Hole und parse die Seite EINMAL - # Kombiniere Ergebnisse - company_data = { + if not soup: + debug_print(" -> Fehler: Konnte Seite nicht laden oder parsen.") + # Optional: Versuche, page object für summary zu holen? Eher nicht, wenn HTML fehlscglug. + return default_result + + # --- Extrahiere Daten aus dem Soup-Objekt --- + first_paragraph = self._extract_first_paragraph_from_soup(soup) + categories_val = self.extract_categories(soup) + # Nutze _extract_infobox_value für robustere Extraktion + branche_val = self._extract_infobox_value(soup, 'branche') + umsatz_val = self._extract_infobox_value(soup, 'umsatz') + mitarbeiter_val = self._extract_infobox_value(soup, 'mitarbeiter') + + # --- Ergebnis zusammenstellen --- + result = { 'url': page_url, 'first_paragraph': first_paragraph, - 'branche': infobox_data['branche'], - 'umsatz': infobox_data['umsatz'], - 'mitarbeiter': infobox_data['mitarbeiter'], + 'branche': branche_val, + 'umsatz': umsatz_val, + 'mitarbeiter': mitarbeiter_val, 'categories': categories_val + # 'full_infobox' wird nicht mehr extrahiert } - # debug_print(f"Extrahierte Wiki-Daten für {page_url}: {company_data}") - return company_data + debug_print(f" -> Extrahierte Daten: P={first_paragraph[:30]}..., B='{branche_val}', U='{umsatz_val}', M='{mitarbeiter_val}', C={categories_val[:30]}...") + return result - # retry_on_failure ist hier schon drauf + # --- search_company_article bleibt fast gleich, nutzt aber intern _get_page_soup --- + @retry_on_failure def search_company_article(self, company_name, website): - """Sucht nach einem passenden Wikipedia-Artikel.""" + """Sucht einen passenden Wikipedia-Artikel und gibt das page-Objekt zurück.""" search_terms = self._generate_search_terms(company_name, website) - if not search_terms: - debug_print("Keine Suchbegriffe generiert, Wikipedia-Suche übersprungen.") - return None + if not search_terms: return None for term in search_terms: try: - # wikipedia.search gibt Titel zurück results = wikipedia.search(term, results=Config.WIKIPEDIA_SEARCH_RESULTS) - debug_print(f"Wikipedia-Suchergebnisse für '{term}': {results}") + debug_print(f"Suchergebnisse für '{term}': {results}") for title in results: - page = self._fetch_page_content(title) - if page and self._validate_article(page, company_name, website): - debug_print(f"Passenden Wikipedia-Artikel gefunden: {page.url}") - return page # Gib das Page-Objekt zurück - except Exception as e: - # Fehler bei der Suche selbst (Netzwerk etc.) - debug_print(f"Fehler während der Wikipedia-Suche für '{term}': {e}") + try: + # Versuche, das Page-Objekt zu laden (für Validierung mit page.externallinks etc.) + page = wikipedia.page(title, auto_suggest=False, preload=True) # Preload für Effizienz + # Validierung prüft jetzt intern mit _get_page_soup + if self._validate_article(page, company_name, website): + debug_print(f"Valider Artikel gefunden: {page.url}") + return page # Gib das Page-Objekt zurück + # else: Artikel gefunden, aber nicht validiert -> nächsten Titel prüfen + except wikipedia.exceptions.PageError: + debug_print(f" -> Seite '{title}' nicht gefunden (PageError).") + continue + except wikipedia.exceptions.DisambiguationError as e: + debug_print(f" -> Seite '{title}' ist Begriffsklärung: {e.options[:3]}...") + # Optional: Versuche ersten Link der Begriffsklärung? Vorerst nicht. + continue + except Exception as e_page: + # Andere Fehler beim Laden/Validieren einer einzelnen Seite + debug_print(f" -> Fehler bei Verarbeitung von Titel '{title}': {e_page}") + continue # Zum nächsten Titel + except Exception as e_search: + # Fehler bei der Suche selbst (z.B. Netzwerk zu Wikipedia) + debug_print(f"Fehler während Wikipedia-Suche für '{term}': {e_search}") # Hier nicht abbrechen, sondern nächsten Suchbegriff versuchen continue # Zum nächsten Suchbegriff - debug_print(f"Kein passender Wikipedia-Artikel für '{company_name}' gefunden.") + debug_print(f"Kein passender Wikipedia-Artikel für '{company_name}' gefunden nach Prüfung aller Begriffe.") return None # ==================== WEBSITE SCRAPING ====================