Update nach der ersten Version die einigermaßen funktioniert

URL-Normalisierung:
Die Methode _normalize_domain entfernt nun explizit Protokoll und „www“, sodass z. B. aus „www.heilemann-leo.de“ einfach „heilemann-leo.de“ wird.

Suchlogik:
In _generate_search_terms wird zuerst der normalisierte Domainname verwendet. Erst danach kommen der bereinigte Firmenname und ergänzende Begriffe. So wird irrelevanter Suchbegriff wie „www“ vermieden.

Validierung:
Die Artikel werden auf ihre Relevanz (unter anderem mittels Ähnlichkeitsvergleich) geprüft, sodass nur Artikel mit ausreichend hoher Ähnlichkeit akzeptiert werden.
This commit is contained in:
2025-03-31 18:25:48 +00:00
parent f046311521
commit bd64fc7121

View File

@@ -63,27 +63,19 @@ class GoogleSheetHandler:
def _connect(self):
"""Stellt Verbindung zum Google Sheet her"""
scope = ["https://www.googleapis.com/auth/spreadsheets"]
creds = ServiceAccountCredentials.from_json_keyfile_name(
Config.CREDENTIALS_FILE, scope
)
creds = ServiceAccountCredentials.from_json_keyfile_name(Config.CREDENTIALS_FILE, scope)
self.sheet = gspread.authorize(creds).open_by_url(Config.SHEET_URL).sheet1
self.sheet_values = self.sheet.get_all_values()
def get_start_index(self):
"""Ermittelt die erste leere Zeile in Spalte N"""
filled_n = [row[13] if len(row) > 13 else '' for row in self.sheet_values[1:]]
return next(
(i + 1 for i, v in enumerate(filled_n, start=1) if not str(v).strip()),
len(filled_n) + 1
)
return next((i + 1 for i, v in enumerate(filled_n, start=1) if not str(v).strip()), len(filled_n) + 1)
def update_row(self, row_num, values):
"""Aktualisiert eine Zeile im Sheet"""
# ACHTUNG: Bereich auf G bis R erweitern, um 12 Spalten zu umfassen.
self.sheet.update(
range_name=f"G{row_num}:R{row_num}",
values=[values]
)
# Bereich G bis R umfasst 12 Spalten
self.sheet.update(range_name=f"G{row_num}:R{row_num}", values=[values])
# ==================== WIKIPEDIA SCRAPER ====================
class WikipediaScraper:
@@ -93,35 +85,37 @@ class WikipediaScraper:
wikipedia.set_lang(Config.LANG)
def _normalize_domain(self, website):
"""Normalisiert URLs zu reinen Domainnamen"""
"""Normalisiert URLs zu reinen Domainnamen ohne Protokoll und www"""
if not website:
return ""
domain = re.sub(r'^https?:\/\/(www\.)?', '', website.lower())
domain = re.sub(r'\/.*$', '', domain)
domain = domain.split('.')[0]
debug_print(f"Normalisierte Domain: {domain}")
return domain
website = website.lower().strip()
website = re.sub(r'^https?:\/\/', '', website) # Entferne http/https
website = re.sub(r'^www\.', '', website) # Entferne führendes www.
website = re.sub(r'\/.*$', '', website) # Entferne Pfad
debug_print(f"Normalisierte Domain: {website}")
return website
def _generate_search_terms(self, company_name, website):
"""Generiert Suchbegriffe mit optimierter URL-Verarbeitung"""
"""Generiert Suchbegriffe, zuerst basierend auf der URL, dann Firmenname"""
terms = []
clean_name = re.sub(
r'\s+(GmbH|AG|KG|Co\. KG|e\.V\.|mbH|& Co).*$',
'',
company_name
).strip()
terms.extend([company_name.strip(), clean_name])
domain = self._normalize_domain(website)
if domain and domain not in ["de", "com", "org"]:
terms.append(domain)
normalized_url = self._normalize_domain(website)
if normalized_url:
terms.append(normalized_url)
clean_name = re.sub(r'\s+(GmbH|AG|KG|Co\. KG|e\.V\.|mbH|& Co).*$', '', company_name).strip()
if clean_name and clean_name not in terms:
terms.append(clean_name)
if company_name.strip() and company_name.strip() not in terms:
terms.append(company_name.strip())
name_parts = [p for p in re.split(r'\W+', clean_name) if p and len(p) > 3]
if len(name_parts) >= 2:
terms.append(" ".join(name_parts[:2]))
debug_print(f"Generierte Suchbegriffe: {list(set(terms))}")
return list(set(terms))
candidate = " ".join(name_parts[:2])
if candidate not in terms:
terms.append(candidate)
debug_print(f"Generierte Suchbegriffe: {terms}")
return terms
def _validate_article(self, page, company_name, domain_hint):
"""Überprüft Artikelrelevanz"""
"""Überprüft Artikelrelevanz basierend auf Ähnlichkeit und Domain-Hinweis"""
clean_title = re.sub(r'\(.*?\)', '', page.title).lower()
clean_company = re.sub(r'[^a-zäöüß ]', '', company_name.lower())
similarity = SequenceMatcher(None, clean_title, clean_company).ratio()
@@ -138,7 +132,7 @@ class WikipediaScraper:
@retry_on_failure
def search_company_article(self, company_name, website):
"""Hauptfunktion zur Artikelsuche"""
"""Sucht zuerst nach der URL, dann nach dem Firmennamen"""
search_terms = self._generate_search_terms(company_name, website)
domain_hint = self._normalize_domain(website)
for term in search_terms:
@@ -150,8 +144,7 @@ class WikipediaScraper:
page = wikipedia.page(title, auto_suggest=False)
if self._validate_article(page, company_name, domain_hint):
return page
except (wikipedia.exceptions.DisambiguationError,
wikipedia.exceptions.PageError) as e:
except (wikipedia.exceptions.DisambiguationError, wikipedia.exceptions.PageError) as e:
debug_print(f"Seitenfehler: {str(e)}")
continue
except Exception as e:
@@ -160,23 +153,13 @@ class WikipediaScraper:
return None
def _extract_infobox_value(self, soup, target):
"""Extrahiert Werte aus der Infobox (Fallback-Methode)"""
infobox = soup.find('table', class_=lambda c: c and any(
kw in c.lower() for kw in ['infobox', 'vcard', 'unternehmen']
))
"""Extrahiert Werte aus der Infobox (Fallback)"""
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 = {
'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'
]
'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']
}[target]
for row in infobox.find_all('tr'):
header = row.find('th')
@@ -190,13 +173,7 @@ class WikipediaScraper:
clean_val = re.sub(r'\[.*?\]|\(.*?\)', '', raw_value)
return ' '.join(clean_val.split()).strip()
if target == 'umsatz':
match = re.search(
r'(\d{1,3}(?:[.,]\d{3})*)\s*'
r'(?:Mio\.?|Millionen|Mrd\.?|Milliarden)?\s*'
r'€?',
raw_value.replace('.', '').replace(',', '.'),
re.IGNORECASE
)
match = re.search(r'(\d{1,3}(?:[.,]\d{3})*)\s*(?:Mio\.?|Millionen|Mrd\.?|Milliarden)?\s*€?', raw_value.replace('.', '').replace(',', '.'), re.IGNORECASE)
if match:
num = float(match.group(1))
if 'mrd' in raw_value.lower():
@@ -207,22 +184,18 @@ class WikipediaScraper:
def extract_full_infobox(self, soup):
"""Extrahiert die komplette Infobox als Text"""
infobox = soup.find('table', class_=lambda c: c and any(
kw in c.lower() for kw in ['infobox', 'vcard', 'unternehmen']
))
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."
return clean_text(infobox.get_text(separator=' | '))
def extract_fields_from_infobox_text(self, infobox_text, field_names):
"""Extrahiert die gewünschten Felder (z.B. Branche, Umsatz) aus dem Infobox-Text.
Es wird angenommen, dass die Felder durch ' | ' getrennt sind."""
"""Extrahiert die gewünschten Felder aus dem Infobox-Text (getrennt durch ' | ')"""
result = {}
tokens = [token.strip() for token in infobox_text.split("|") if token.strip()]
for i, token in enumerate(tokens):
for field in field_names:
if token.lower() == field.lower():
# Nächstes nicht-leeres Token als Wert
j = i + 1
while j < len(tokens) and not tokens[j]:
j += 1
@@ -236,19 +209,11 @@ class WikipediaScraper:
try:
response = requests.get(page_url)
soup = BeautifulSoup(response.text, Config.HTML_PARSER)
# Gesamte Infobox extrahieren
full_infobox = self.extract_full_infobox(soup)
# Versuch, Felder aus dem kompletten Infobox-Text herauszulesen
extracted_fields = self.extract_fields_from_infobox_text(full_infobox, ['Branche', 'Umsatz'])
# Fallback: Falls kein Wert gefunden wird, dann nutze die alte Methode
branche_val = extracted_fields.get('Branche', self._extract_infobox_value(soup, 'branche'))
umsatz_val = extracted_fields.get('Umsatz', self._extract_infobox_value(soup, 'umsatz'))
return {
'full_infobox': full_infobox,
'branche': branche_val,
'umsatz': umsatz_val,
'url': page_url
}
umsatz_val = extracted_fields.get('Umsatz', self._extract_infobox_value(soup, 'umsatz'))
return {'full_infobox': full_infobox, 'branche': branche_val, 'umsatz': umsatz_val, 'url': page_url}
except Exception as e:
debug_print(f"Extraktionsfehler: {str(e)}")
return {'branche': 'k.A.', 'umsatz': 'k.A.', 'url': page_url, 'full_infobox': 'k.A.'}