bugfix
This commit is contained in:
@@ -1505,13 +1505,24 @@ class GoogleSheetHandler:
|
|||||||
class WikipediaScraper:
|
class WikipediaScraper:
|
||||||
"""
|
"""
|
||||||
Handles searching Wikipedia articles and extracting relevant company data.
|
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
|
def __init__(self, user_agent=None):
|
||||||
# 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)')
|
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 = requests.Session()
|
||||||
self.session.headers.update({'User-Agent': self.user_agent})
|
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
|
# Erweiterte Keyword Map für robustere Infobox-Extraktion
|
||||||
self.keywords_map = {
|
self.keywords_map = {
|
||||||
@@ -1520,348 +1531,351 @@ class WikipediaScraper:
|
|||||||
'mitarbeiter': ['mitarbeiter', 'mitarbeiterzahl', 'beschäftigte', 'employees', 'number of employees', 'personal', 'belegschaft']
|
'mitarbeiter': ['mitarbeiter', 'mitarbeiterzahl', 'beschäftigte', 'employees', 'number of employees', 'personal', 'belegschaft']
|
||||||
}
|
}
|
||||||
try:
|
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
|
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:
|
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):
|
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 ""
|
if not website or not isinstance(website, str): return ""
|
||||||
website = website.lower().strip()
|
website_lower = website.lower().strip()
|
||||||
website = re.sub(r'^https?:\/\/', '', website)
|
if not website_lower or website_lower == 'k.a.': return ""
|
||||||
website = re.sub(r'^www\.', '', website)
|
# Schema entfernen
|
||||||
return website.split('/')[0]
|
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):
|
def _generate_search_terms(self, company_name, website):
|
||||||
terms = set(); full_domain = self._get_full_domain(website)
|
"""Generiert eine Liste von Suchbegriffen für die Wikipedia-Suche."""
|
||||||
if full_domain: terms.add(full_domain)
|
if not company_name: return []
|
||||||
normalized_name = normalize_company_name(company_name) # Annahme: normalize_company_name existiert
|
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:
|
if normalized_name:
|
||||||
name_parts = normalized_name.split()
|
name_parts = normalized_name.split()
|
||||||
if len(name_parts) > 0: terms.add(name_parts[0])
|
if len(name_parts) > 0: terms.add(name_parts[0]) # Erstes Wort
|
||||||
if len(name_parts) > 1: terms.add(" ".join(name_parts[:2]))
|
if len(name_parts) > 1: terms.add(" ".join(name_parts[:2])) # Erste zwei Worte
|
||||||
terms.add(normalized_name)
|
terms.add(normalized_name) # Ganzer normalisierter 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())
|
|
||||||
|
|
||||||
# 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
|
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
|
return final_terms
|
||||||
|
|
||||||
# retry_on_failure Decorator hier hinzufügen!
|
@retry_on_failure # Annahme: Decorator existiert und behandelt Exceptions
|
||||||
@retry_on_failure
|
|
||||||
def _get_page_soup(self, url):
|
def _get_page_soup(self, url):
|
||||||
"""Holt HTML von einer URL und gibt ein BeautifulSoup-Objekt zurück."""
|
"""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"):
|
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
|
return None
|
||||||
try:
|
try:
|
||||||
logger.debug(f"_get_page_soup: Rufe URL ab: {url}")
|
self.logger.debug(f"_get_page_soup: Rufe URL ab: {url}")
|
||||||
response = self.session.get(url, timeout=15) # Timeout erhöhen
|
response = self.session.get(url, timeout=20) # Timeout etwas höher
|
||||||
response.raise_for_status() # Fehler für 4xx/5xx auslösen
|
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'
|
response.encoding = 'utf-8'
|
||||||
soup = BeautifulSoup(response.text, Config.HTML_PARSER) # Nutze Parser aus Config
|
soup = BeautifulSoup(response.text, getattr(Config, 'HTML_PARSER', 'html.parser')) # Nutze Parser aus Config
|
||||||
logger.debug(f"_get_page_soup: Parsen von {url} erfolgreich.")
|
self.logger.debug(f"_get_page_soup: Parsen von {url} erfolgreich.")
|
||||||
return soup
|
return soup
|
||||||
except requests.exceptions.Timeout:
|
except requests.exceptions.Timeout:
|
||||||
logger.error(f"_get_page_soup: Timeout beim Abrufen von {url}")
|
self.logger.error(f"_get_page_soup: Timeout beim Abrufen von {url}")
|
||||||
return None
|
raise # Fehler weitergeben für Retry
|
||||||
except requests.exceptions.RequestException as e:
|
except requests.exceptions.RequestException as e:
|
||||||
logger.error(f"_get_page_soup: Fehler beim Abrufen von HTML von {url}: {e}")
|
self.logger.error(f"_get_page_soup: Netzwerkfehler beim Abrufen von HTML von {url}: {e}")
|
||||||
# Hier keinen Fehler weitergeben, damit der Prozess weiterläuft, aber None zurückgeben
|
raise e # Fehler weitergeben für Retry
|
||||||
return None
|
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error(f"_get_page_soup: Fehler beim Parsen von HTML von {url}: {e}")
|
self.logger.error(f"_get_page_soup: Fehler beim Parsen von HTML von {url}: {e}")
|
||||||
return None
|
# 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):
|
def _validate_article(self, page, company_name, website):
|
||||||
"""
|
"""
|
||||||
Validiert, ob ein Wikipedia-Artikel zum Unternehmen passt.
|
Validiert, ob ein Wikipedia-Artikel zum Unternehmen passt.
|
||||||
Prüft Titelähnlichkeit und ob die Firmenwebsite verlinkt ist.
|
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
|
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)
|
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)
|
normalized_title = normalize_company_name(page.title)
|
||||||
|
|
||||||
# 1. Titelähnlichkeit prüfen
|
# 1. Titelähnlichkeit
|
||||||
similarity = SequenceMatcher(None, normalized_title, normalized_company).ratio()
|
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)
|
# 2. Link-Prüfung (nur wenn Domain vorhanden)
|
||||||
domain_found = False
|
domain_found = False
|
||||||
if full_domain:
|
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
|
soup = self._get_page_soup(page.url) # Hole Soup für Link-Prüfung
|
||||||
if soup:
|
if soup:
|
||||||
# Suche zuerst in der Infobox (häufigster Ort für offizielle Links)
|
# Suche zuerst in der Infobox
|
||||||
infobox = soup.select_one('table[class*="infobox"]') # Flexibler Selektor
|
infobox = soup.select_one('table[class*="infobox"]')
|
||||||
if infobox:
|
if infobox:
|
||||||
website_links = infobox.find_all('a', href=True)
|
website_links = infobox.find_all('a', href=True)
|
||||||
for link in website_links:
|
for link in website_links:
|
||||||
href = link.get('href', '')
|
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 self._get_full_domain(href): # Vergleiche Domains
|
||||||
if href.startswith('http') and full_domain in href.lower():
|
|
||||||
# Zusätzliche Prüfung: Ist der Link-Text relevant? (Optional)
|
|
||||||
link_text = link.get_text(strip=True).lower()
|
link_text = link.get_text(strip=True).lower()
|
||||||
if any(kw in link_text for kw in ['website', 'webseite', 'offizielle']):
|
# Prüfe, ob Keyword im Link-Text ODER in der Tabellenüberschrift der Zeile
|
||||||
debug_print(f" -> Domain '{full_domain}' in Infobox-Link gefunden (Text: '{link_text}', URL: {href})")
|
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
|
domain_found = True
|
||||||
break
|
break
|
||||||
else: # Auch akzeptieren, wenn der Text nicht passt, aber die URL klar ist
|
else: # Akzeptiere auch ohne Keyword, wenn URL eindeutig scheint
|
||||||
debug_print(f" -> Domain '{full_domain}' in Infobox-Link gefunden (URL: {href})")
|
self.logger.debug(f" -> Domain '{full_domain}' in Infobox-Link gefunden (URL: {href}, kein Keyword-Match im Text/TH)")
|
||||||
domain_found = True
|
domain_found = True
|
||||||
break
|
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:
|
if not domain_found:
|
||||||
# Die wikipedia library liefert externe Links manchmal nicht zuverlässig, daher besser alle Links parsen
|
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)
|
all_links = soup.find_all('a', href=True)
|
||||||
|
|
||||||
for link in all_links:
|
for link in all_links:
|
||||||
href = link.get('href', '')
|
href = link.get('href', '')
|
||||||
# Klasse 'external text' ist ein guter Indikator, aber nicht immer vorhanden
|
if href.startswith('http') and full_domain in self._get_full_domain(href):
|
||||||
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():
|
|
||||||
# Ignoriere Links zu anderen Wikimedia-Projekten etc.
|
# Ignoriere Links zu anderen Wikimedia-Projekten etc.
|
||||||
if not any(site in href for site in ['wikipedia.org', 'wikimedia.org', 'wikidata.org']):
|
if not any(site in href for site in ['wikipedia.org', 'wikimedia.org', 'wikidata.org', 'archive.org', 'webcitation.org']):
|
||||||
debug_print(f" -> Domain '{full_domain}' in externem Link gefunden (URL: {href}, External Class? {is_external})")
|
self.logger.debug(f" -> Domain '{full_domain}' in externem Link gefunden (URL: {href})")
|
||||||
domain_found = True
|
domain_found = True
|
||||||
break # Erster Treffer reicht
|
break # Erster Treffer reicht
|
||||||
else:
|
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:
|
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
|
# 3. Entscheidung
|
||||||
threshold = Config.SIMILARITY_THRESHOLD # Standard-Schwelle
|
threshold = getattr(Config, 'SIMILARITY_THRESHOLD', 0.65) # Standard-Schwelle aus Config
|
||||||
if domain_found:
|
if domain_found:
|
||||||
threshold = max(0.4, threshold - 0.2) # Schwelle lockern, wenn Domain passt (aber nicht zu sehr)
|
# Schwelle signifikant lockern, wenn Domain passt
|
||||||
debug_print(f" -> Domain gefunden, Schwelle angepasst auf {threshold:.2f}")
|
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
|
is_valid = similarity >= threshold
|
||||||
if is_valid:
|
log_level = logging.INFO if is_valid else logging.DEBUG # Logge Erfolg als INFO
|
||||||
debug_print(f" => Artikel '{page.title}' VALIDiert (Ähnlichkeit >= {threshold:.2f}, Domain gefunden? {domain_found})")
|
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})")
|
||||||
else:
|
|
||||||
debug_print(f" => Artikel '{page.title}' NICHT validiert (Ähnlichkeit < {threshold:.2f}, Domain gefunden? {domain_found})")
|
|
||||||
return is_valid
|
return is_valid
|
||||||
|
|
||||||
# Beibehalten: Extrahiere Kategorien
|
|
||||||
def extract_categories(self, soup):
|
def extract_categories(self, soup):
|
||||||
|
"""Extrahiert Wikipedia-Kategorien aus dem Soup-Objekt."""
|
||||||
if not soup: return "k.A."
|
if not soup: return "k.A."
|
||||||
|
cats_filtered = []
|
||||||
try:
|
try:
|
||||||
cat_div = soup.find('div', id="mw-normal-catlinks")
|
cat_div = soup.find('div', id="mw-normal-catlinks")
|
||||||
if cat_div:
|
if cat_div:
|
||||||
ul = cat_div.find('ul')
|
ul = cat_div.find('ul')
|
||||||
if 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 = [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()]
|
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:
|
except Exception as e:
|
||||||
logger.error(f"Fehler beim Extrahieren der Kategorien: {e}")
|
self.logger.error(f"Fehler beim Extrahieren der Kategorien: {e}")
|
||||||
return "k.A."
|
|
||||||
|
return ", ".join(cats_filtered) if cats_filtered else "k.A."
|
||||||
|
|
||||||
# Beibehalten: Extrahiere ersten Absatz
|
|
||||||
def _extract_first_paragraph_from_soup(self, soup):
|
def _extract_first_paragraph_from_soup(self, soup):
|
||||||
|
"""Extrahiert den ersten aussagekräftigen Absatz aus dem Soup-Objekt."""
|
||||||
if not soup: return "k.A."
|
if not soup: return "k.A."
|
||||||
|
paragraph_text = "k.A."
|
||||||
try:
|
try:
|
||||||
# Finde das Haupt-Inhaltsdiv (üblich bei MediaWiki)
|
# Suche bevorzugt im Hauptinhalt
|
||||||
content_div = soup.find('div', class_='mw-parser-output')
|
content_div = soup.find('div', class_='mw-parser-output')
|
||||||
if not content_div:
|
search_area = content_div if content_div else soup # Fallback auf ganzen Soup
|
||||||
content_div = soup.find('div', id='bodyContent') # Fallback
|
|
||||||
if not content_div:
|
|
||||||
content_div = soup # Fallback auf ganzen Soup
|
|
||||||
|
|
||||||
# Suche nach dem ersten <p>-Tag, der direkten Text enthält (keine leeren <p>)
|
# Finde alle <p>-Tags direkt unter dem Suchbereich (oft relevanter)
|
||||||
paragraphs = content_div.find_all('p', recursive=False) # Nur direkte Kinder suchen? Oft besser
|
paragraphs = search_area.find_all('p', recursive=False)
|
||||||
if not paragraphs:
|
if not paragraphs: # Fallback: Alle <p>-Tags suchen
|
||||||
paragraphs = content_div.find_all('p', recursive=True) # Fallback: alle <p>
|
paragraphs = search_area.find_all('p', recursive=True)
|
||||||
|
self.logger.debug(f"Suche ersten Absatz in {len(paragraphs)} gefundenen <p>-Tags...")
|
||||||
|
|
||||||
for p in paragraphs:
|
for idx, p in enumerate(paragraphs):
|
||||||
# Ignoriere <p>-Tags, die nur ein Bild oder eine Tabelle enthalten (heuristisch)
|
# Überspringe leere oder zu kurze Absätze
|
||||||
if p.find(['img', 'table', 'figure'], recursive=False): continue
|
if not p.get_text(strip=True):
|
||||||
# Entferne Referenz-Links [1], [2] etc. VOR der Textextraktion
|
self.logger.debug(f" -> Überspringe leeres <p> 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 <p> Tag (Index {idx}), da er Blockelemente enthält.")
|
||||||
|
continue
|
||||||
|
|
||||||
|
# Entferne Referenz-Links `[1]`, `[2]` etc.
|
||||||
for sup in p.find_all('sup', class_='reference'):
|
for sup in p.find_all('sup', class_='reference'):
|
||||||
sup.decompose()
|
sup.decompose()
|
||||||
# Extrahiere und bereinige Text
|
# Entferne Koordinaten (oft am Anfang in <span>)
|
||||||
text = p.get_text(separator=' ', strip=True)
|
for span in p.find_all('span', id='coordinates'):
|
||||||
text = clean_text(text) # Annahme: clean_text entfernt u.a. doppelte Leerzeichen
|
span.decompose()
|
||||||
# Nimm den ersten Absatz mit signifikanter Länge
|
|
||||||
if text and len(text) > 30: # Mindestlänge, um leere oder irrelevante <p> zu überspringen
|
text = clean_text(p.get_text(separator=' ', strip=True))
|
||||||
logger.debug(f"Ersten Absatz gefunden: {text[:100]}...")
|
|
||||||
return text[:1000] # Begrenze Länge
|
# Prüfe Mindestlänge
|
||||||
except Exception as e:
|
if text and len(text) > 40: # Mindestlänge etwas erhöht
|
||||||
logger.error(f"Fehler beim Extrahieren des ersten Absatzes: {e}")
|
self.logger.debug(f" -> Ersten gültigen Absatz (Index {idx}) gefunden: {text[:100]}...")
|
||||||
return "k.A."
|
paragraph_text = text[:1000] # Begrenze Länge
|
||||||
|
break # Nimm den ersten passenden
|
||||||
|
else:
|
||||||
|
self.logger.debug(f" -> Überspringe <p> 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):
|
def _extract_infobox_value(self, soup, target):
|
||||||
"""
|
"""
|
||||||
Extrahiert gezielt Branche, Umsatz oder Mitarbeiter aus der Infobox.
|
Extrahiert gezielt Branche, Umsatz oder Mitarbeiter aus der Infobox. (v1.6.5 Logik)
|
||||||
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.".
|
|
||||||
"""
|
"""
|
||||||
|
# --- 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:
|
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."
|
return "k.A."
|
||||||
|
|
||||||
keywords = self.keywords_map[target]
|
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
|
# Flexiblere Suche nach der Infobox
|
||||||
infobox = soup.select_one('table[class*="infobox"]')
|
infobox = soup.select_one('table[class*="infobox"]')
|
||||||
if not infobox:
|
# --- DEBUG LOG: Infobox gefunden? ---
|
||||||
logger.debug(" -> Keine Infobox mit Klasse '*infobox*' gefunden.")
|
if infobox:
|
||||||
# Fallback: Suche nach Tabellen mit typischen Attributen, falls Klasse fehlt
|
self.logger.debug(f" -> Infobox gefunden (via select_one 'table[class*=\"infobox\"]')")
|
||||||
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:
|
else:
|
||||||
logger.debug(" -> Infobox über summary='Infobox...' gefunden.")
|
self.logger.debug(" -> KEINE Infobox via select_one 'table[class*=\"infobox\"]' gefunden.")
|
||||||
else:
|
# Optional: Fallback-Suche (hier ausgelassen für Klarheit)
|
||||||
logger.debug(f" -> Infobox gefunden (Selektor 'table[class*=\"infobox\"]'). Erste Zeilen: {str(infobox.find('tr'))[:100]}...")
|
return "k.A." # Frühzeitiger Ausstieg
|
||||||
|
|
||||||
|
value_found = "k.A."
|
||||||
value_found = "k.A." # Standardwert
|
|
||||||
|
|
||||||
# Iteriere durch alle Zeilen (tr) der Infobox
|
|
||||||
try:
|
try:
|
||||||
rows = infobox.find_all('tr')
|
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):
|
for idx, row in enumerate(rows):
|
||||||
header = row.find('th') # Finde die Header-Zelle
|
header_cells = row.find_all('th', recursive=False) # Nur direkte th Kinder
|
||||||
value_cell = row.find('td') # Finde die Daten-Zelle
|
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 = header.get_text(strip=True)
|
||||||
header_text_lower = header_text.lower()
|
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)
|
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]
|
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
|
matched_keyword = None
|
||||||
for kw in keywords:
|
for kw in keywords:
|
||||||
# Suche nach dem Keyword als ganzes Wort oder am Anfang/Ende?
|
# Prüfe ob Keyword im (nicht-normalisierten) Lowercase-Header vorkommt
|
||||||
# Einfacher: `in`-Operator prüft auf Substring
|
if kw in header_text_lower:
|
||||||
if kw in header_text_lower: # Prüfe im weniger aggressiv normalisierten Text
|
|
||||||
matched_keyword = kw
|
matched_keyword = kw
|
||||||
break
|
break
|
||||||
|
|
||||||
if matched_keyword:
|
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 ---
|
# Störende Elemente (Referenzen, unsichtbare Spans) im Value entfernen
|
||||||
# 1. Störende Elemente entfernen (Referenzen etc.)
|
for sup in value_cell.find_all(['sup', 'span']):
|
||||||
for sup in value_cell.find_all(['sup', 'span']): # Entferne <sup> und ggf. überflüssige <span>
|
if (sup.name == 'sup' and sup.has_attr('class') and 'reference' in sup['class']) or \
|
||||||
# Spezifischer: Nur Referenzen entfernen?
|
(sup.name == 'span' and sup.get('style') and 'display:none' in sup['style']):
|
||||||
if sup.has_attr('class') and 'reference' in sup['class']:
|
self.logger.debug(f" -> Entferne störendes Element: {sup.get_text(strip=True)}")
|
||||||
logger.debug(f" -> Entferne Referenz: {sup.get_text(strip=True)}")
|
|
||||||
sup.decompose()
|
|
||||||
# Entferne unsichtbare Elemente (manchmal in <span>)
|
|
||||||
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()
|
sup.decompose()
|
||||||
|
|
||||||
|
# Text extrahieren (mit Leerzeichen für <br>) und bereinigen
|
||||||
# 2. Text extrahieren (mit Leerzeichen für <br>)
|
|
||||||
raw_value_text = value_cell.get_text(separator=' ', strip=True)
|
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)
|
# Spezifische Verarbeitung
|
||||||
cleaned_raw_value = clean_text(raw_value_text) # Annahme: clean_text ist definiert
|
|
||||||
|
|
||||||
# 4. Spezifische Verarbeitung je nach Zieltyp
|
|
||||||
if target == 'branche':
|
if target == 'branche':
|
||||||
# Entferne oft vorkommende Zusätze wie (Stand: ...), [1], etc.
|
# Einfache Bereinigung für Branche
|
||||||
clean_val = re.sub(r'\s*\[\d+\]', '', cleaned_raw_value) # Referenzen (falls clean_text sie nicht erwischt)
|
clean_val = re.sub(r'\s*\([^)]*\)', '', cleaned_raw_value).strip() # Klammerzusätze entfernen
|
||||||
clean_val = re.sub(r'\s*\([^)]*\)', '', clean_val).strip() # Klammerzusätze
|
# 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."
|
value_found = clean_val if clean_val else "k.A."
|
||||||
logger.debug(f" --> Branche extrahiert: '{value_found}'")
|
self.logger.debug(f" --> Branche extrahiert: '{value_found}'")
|
||||||
|
|
||||||
elif target == 'umsatz':
|
elif target == 'umsatz':
|
||||||
# extract_numeric_value sollte robust sein
|
numeric_val = extract_numeric_value(cleaned_raw_value, is_umsatz=True)
|
||||||
numeric_val = extract_numeric_value(cleaned_raw_value, is_umsatz=True) # Annahme: existiert
|
|
||||||
value_found = numeric_val
|
value_found = numeric_val
|
||||||
logger.debug(f" --> Umsatz extrahiert (aus '{cleaned_raw_value}'): '{value_found}'")
|
self.logger.debug(f" --> Umsatz extrahiert (aus '{cleaned_raw_value}'): '{value_found}'")
|
||||||
|
|
||||||
elif target == 'mitarbeiter':
|
elif target == 'mitarbeiter':
|
||||||
# extract_numeric_value sollte robust sein
|
numeric_val = extract_numeric_value(cleaned_raw_value, is_umsatz=False)
|
||||||
numeric_val = extract_numeric_value(cleaned_raw_value, is_umsatz=False) # Annahme: existiert
|
|
||||||
value_found = numeric_val
|
value_found = numeric_val
|
||||||
logger.debug(f" --> Mitarbeiter extrahiert (aus '{cleaned_raw_value}'): '{value_found}'")
|
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: Ersten Treffer für das gesuchte Ziel nehmen und Zeilenschleife verlassen
|
||||||
break # WICHTIG: Verhindert Überschreiben durch spätere Zeilen
|
else:
|
||||||
|
# Logge Zeilen, die nicht dem Key-Value-Muster entsprechen (optional auf DEBUG)
|
||||||
# else: # Optional: Loggen, wenn kein Keyword passte
|
# self.logger.debug(f" -> Überspringe Zeile {idx}: Struktur passt nicht (TH:{len(header_cells)}, TD:{len(value_cells)})")
|
||||||
# logger.debug(f" --- Kein passendes Keyword in TH '{header_text}' gefunden.")
|
pass
|
||||||
|
|
||||||
# Ende der Zeilenschleife
|
# Ende der Zeilenschleife
|
||||||
if value_found != "k.A.":
|
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:
|
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:
|
except Exception as e:
|
||||||
logger.error(f"Fehler beim Durchlaufen der Infobox-Zeilen für '{target}': {e}")
|
self.logger.exception(f"Fehler beim Durchlaufen der Infobox-Zeilen für '{target}': {e}")
|
||||||
# Gib "k.A." zurück, wenn Fehler beim Parsen auftreten
|
return "k.A." # Fehler beim Parsen
|
||||||
return "k.A."
|
|
||||||
|
|
||||||
return value_found
|
return value_found
|
||||||
|
|
||||||
# --- Kombinierte Datenextraktion (ruft Helfer auf) ---
|
|
||||||
def extract_company_data(self, page_url):
|
def extract_company_data(self, page_url):
|
||||||
"""
|
"""
|
||||||
Extrahiert Firmendaten von einer Wikipedia-URL. Holt die Seite nur einmal
|
Extrahiert Firmendaten von einer Wikipedia-URL (v1.6.5 Logik).
|
||||||
und verwendet die verbesserte Infobox-Extraktion.
|
|
||||||
"""
|
"""
|
||||||
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.'}
|
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:
|
if not page_url or not isinstance(page_url, str) or "wikipedia.org" not in page_url.lower():
|
||||||
logger.warning(f"extract_company_data: Ungültige URL '{page_url}'.")
|
self.logger.warning(f"extract_company_data: Ungültige URL '{page_url}'.")
|
||||||
return default_result
|
return default_result
|
||||||
|
|
||||||
logger.info(f"Extrahiere Daten für Wiki-URL: {page_url}")
|
self.logger.info(f"Extrahiere Daten für Wiki-URL: {page_url}")
|
||||||
soup = self._get_page_soup(page_url) # Hole und parse die Seite EINMAL
|
soup = self._get_page_soup(page_url)
|
||||||
|
|
||||||
if not soup:
|
if not soup:
|
||||||
logger.error(f" -> Fehler: Konnte Seite {page_url} nicht laden oder parsen.")
|
self.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 # Behalte URL im Ergebnis
|
||||||
default_result['url'] = page_url
|
|
||||||
return default_result
|
return default_result
|
||||||
|
|
||||||
# --- Extrahiere Daten aus dem Soup-Objekt ---
|
# --- 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)
|
first_paragraph = self._extract_first_paragraph_from_soup(soup)
|
||||||
logger.debug(" -> Extrahiere Kategorien...")
|
self.logger.debug(" -> Extrahiere Kategorien...")
|
||||||
categories_val = self.extract_categories(soup)
|
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')
|
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')
|
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')
|
mitarbeiter_val = self._extract_infobox_value(soup, 'mitarbeiter')
|
||||||
|
|
||||||
# --- Ergebnis zusammenstellen ---
|
# --- Ergebnis zusammenstellen ---
|
||||||
@@ -1873,113 +1887,106 @@ class WikipediaScraper:
|
|||||||
'mitarbeiter': mitarbeiter_val,
|
'mitarbeiter': mitarbeiter_val,
|
||||||
'categories': categories_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
|
return result
|
||||||
|
|
||||||
# --- Artikelsuche (ruft Helfer auf) ---
|
@retry_on_failure # Annahme: Decorator existiert
|
||||||
# retry_on_failure hier sinnvoll, da es API-Calls (wikipedia.search/page) macht
|
|
||||||
@retry_on_failure
|
|
||||||
def search_company_article(self, company_name, website=None):
|
def search_company_article(self, company_name, website=None):
|
||||||
"""
|
"""
|
||||||
Sucht einen passenden Wikipedia-Artikel und gibt das page-Objekt zurück.
|
Sucht einen passenden Wikipedia-Artikel und gibt das page-Objekt zurück. (v1.6.5 Logik)
|
||||||
Nutzt generierte Suchbegriffe und Validierung.
|
|
||||||
"""
|
"""
|
||||||
if not company_name:
|
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
|
return None
|
||||||
|
|
||||||
search_terms = self._generate_search_terms(company_name, website)
|
search_terms = self._generate_search_terms(company_name, website)
|
||||||
if not search_terms:
|
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
|
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:
|
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)
|
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):
|
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
|
return page
|
||||||
else:
|
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:
|
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:
|
except wikipedia.exceptions.DisambiguationError as e:
|
||||||
logger.debug(f" -> '{company_name}' ist eine Begriffsklärungsseite. Optionen: {e.options[:3]}...")
|
self.logger.debug(f" -> '{company_name}' ist eine Begriffsklärungsseite. Optionen: {e.options[:3]}...")
|
||||||
# Optional: Prüfe die erste Option der Begriffsklärung?
|
# Optional: Prüfe die erste Option
|
||||||
if e.options:
|
if e.options:
|
||||||
try:
|
try:
|
||||||
first_option_title = e.options[0]
|
option_title = e.options[0]
|
||||||
logger.debug(f" -> Prüfe erste Option der Begriffsklärung: '{first_option_title}'")
|
self.logger.debug(f" -> Prüfe erste Option der Begriffsklärung: '{option_title}'")
|
||||||
page = wikipedia.page(first_option_title, auto_suggest=False, preload=True)
|
page = wikipedia.page(option_title, auto_suggest=False, preload=True)
|
||||||
if self._validate_article(page, company_name, website):
|
if self._validate_article(page, company_name, website):
|
||||||
logger.info(f" -> Erste Option '{page.title}' ({page.url}) erfolgreich validiert.")
|
return page # Erfolg wird von _validate_article geloggt
|
||||||
return page
|
|
||||||
else:
|
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:
|
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:
|
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
|
# Wenn direkter Match fehlschlägt, nutze die generierten Suchbegriffe
|
||||||
logger.debug(f" -> Starte Suche mit generierten Begriffen: {search_terms}")
|
self.logger.debug(f" -> Starte Suche mit generierten Begriffen: {search_terms}")
|
||||||
for term in search_terms:
|
for term in search_terms:
|
||||||
try:
|
try:
|
||||||
logger.debug(f" -> Suche mit Begriff: '{term}'...")
|
self.logger.debug(f" -> Suche mit Begriff: '{term}'...")
|
||||||
results = wikipedia.search(term, results=Config.WIKIPEDIA_SEARCH_RESULTS)
|
# Begrenze Anzahl Suchergebnisse
|
||||||
logger.debug(f" -> Suchergebnisse für '{term}': {results}")
|
search_results = wikipedia.search(term, results=getattr(Config, 'WIKIPEDIA_SEARCH_RESULTS', 5))
|
||||||
if not results: continue # Nächster Begriff, wenn keine Ergebnisse
|
self.logger.debug(f" -> Suchergebnisse für '{term}': {search_results}")
|
||||||
|
if not search_results: continue
|
||||||
|
|
||||||
for title in results:
|
for title in search_results:
|
||||||
# Überspringe Titel, die wahrscheinlich keine Firmen sind (heuristisch)
|
# Heuristische Filterung
|
||||||
if any(ignore_word in title.lower() for ignore_word in ['liste von', 'disambiguation', 'begriffsklärung']):
|
if any(ignore in title.lower() for ignore in ['liste von', 'begriffsklärung', '(Begriffsklärung)']):
|
||||||
logger.debug(f" -> Überspringe wahrscheinlichen Nicht-Artikel: '{title}'")
|
self.logger.debug(f" -> Überspringe wahrscheinlichen Nicht-Artikel/BK: '{title}'")
|
||||||
continue
|
continue
|
||||||
try:
|
try:
|
||||||
logger.debug(f" -> Prüfe potenziellen Artikel: '{title}'")
|
self.logger.debug(f" -> Prüfe potenziellen Artikel: '{title}'")
|
||||||
# Lade Page-Objekt (preload=True für Effizienz bei Validierung)
|
|
||||||
page = wikipedia.page(title, auto_suggest=False, preload=True)
|
page = wikipedia.page(title, auto_suggest=False, preload=True)
|
||||||
# Validierung (nutzt jetzt intern _get_page_soup)
|
|
||||||
if self._validate_article(page, company_name, website):
|
if self._validate_article(page, company_name, website):
|
||||||
logger.info(f" -> Valider Artikel gefunden: '{page.title}' ({page.url})")
|
# Erfolg wird von _validate_article geloggt
|
||||||
return page # Gib das validierte Page-Objekt zurück
|
return page
|
||||||
# else: Artikel gefunden, aber nicht validiert -> nächsten Titel prüfen
|
time.sleep(0.1) # Kleines Delay
|
||||||
# Minimales Delay, um nicht zu schnell zu sein
|
|
||||||
time.sleep(0.1)
|
|
||||||
|
|
||||||
except wikipedia.exceptions.PageError:
|
except wikipedia.exceptions.PageError:
|
||||||
logger.debug(f" -> Seite '{title}' nicht gefunden (PageError).")
|
self.logger.debug(f" -> Seite '{title}' nicht gefunden (PageError).")
|
||||||
continue
|
continue
|
||||||
except wikipedia.exceptions.DisambiguationError as e:
|
except wikipedia.exceptions.DisambiguationError as e:
|
||||||
logger.debug(f" -> Seite '{title}' ist Begriffsklärung: {e.options[:3]}...")
|
self.logger.debug(f" -> Seite '{title}' ist Begriffsklärung: {e.options[:3]}...")
|
||||||
# Optional: Versuche ersten Link der Begriffsklärung?
|
|
||||||
continue
|
continue
|
||||||
except requests.exceptions.RequestException as e_req:
|
except requests.exceptions.RequestException as e_req:
|
||||||
logger.warning(f" -> Netzwerkfehler beim Laden/Validieren von '{title}': {e_req}. Überspringe Titel.")
|
self.logger.warning(f" -> Netzwerkfehler beim Laden/Validieren von '{title}': {e_req}. Überspringe Titel.")
|
||||||
# Kurze Pause bei Netzwerkfehler
|
|
||||||
time.sleep(1)
|
time.sleep(1)
|
||||||
continue
|
continue
|
||||||
except Exception as e_page:
|
except Exception as e_page:
|
||||||
# Andere Fehler beim Laden/Validieren einer einzelnen Seite
|
self.logger.error(f" -> Fehler bei Verarbeitung von Titel '{title}': {type(e_page).__name__} - {e_page}")
|
||||||
logger.error(f" -> Fehler bei Verarbeitung von Titel '{title}': {type(e_page).__name__} - {e_page}")
|
|
||||||
continue # Zum nächsten Titel
|
continue # Zum nächsten Titel
|
||||||
|
|
||||||
except requests.exceptions.RequestException as e_search_req:
|
except requests.exceptions.RequestException as e_search_req:
|
||||||
logger.error(f"Netzwerkfehler während Wikipedia-Suche für '{term}': {e_search_req}")
|
self.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) # Längere Pause bei Suchfehler
|
||||||
time.sleep(2)
|
# Fehler weitergeben für Retry
|
||||||
continue
|
raise e_search_req
|
||||||
except Exception as e_search:
|
except Exception as e_search:
|
||||||
# Fehler bei der Suche selbst
|
self.logger.error(f"Allgemeiner Fehler während Wikipedia-Suche für '{term}': {e_search}")
|
||||||
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.
|
||||||
# Hier nicht abbrechen, sondern nächsten Suchbegriff versuchen
|
continue
|
||||||
continue # Zum nächsten Suchbegriff
|
|
||||||
|
|
||||||
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
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user