From 7bd90f469af21d2846f350fa033b7e085179f566 Mon Sep 17 00:00:00 2001 From: Floke Date: Mon, 31 Mar 2025 12:54:22 +0000 Subject: [PATCH] Deepseek V1 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Klare Trennung der Verantwortlichkeiten: GoogleSheetHandler: Alle Sheet-Interaktionen WikipediaScraper: Suchlogik und Datenextraktion DataProcessor: Steuerung des Gesamtflusses Erweiterte Funktionalitäten: Retry-Decorator für robustere API-Aufrufe Verbesserte Validierung der Artikelrelevanz Flexiblere Suchbegriff-Generierung Zentrale Konfiguration Dokumentation: Ausführliche Docstrings für alle Methoden Kommentare für komplexe Codeabschnitte Klare Parameternamen Wartbarkeit: Einfache Erweiterung durch modularen Aufbau Separierte Geschäftslogik von der Implementierung Konsistente Fehlerbehandlung Die ChatGPT-Bewertung (Teil 3) kann später als separate Klasse hinzugefügt werden, ohne die bestehende Struktur zu beeinflussen. --- brancheneinstufung.py | 407 ++++++++++++++++++++++++------------------ 1 file changed, 235 insertions(+), 172 deletions(-) diff --git a/brancheneinstufung.py b/brancheneinstufung.py index c53004d3..12da4d04 100644 --- a/brancheneinstufung.py +++ b/brancheneinstufung.py @@ -1,190 +1,253 @@ -# Neue Version (1.0.11) mit optimierter Wikipedia-Suchlogik, Original-Auswertung bleibt erhalten - +[file name]: claude.py +[file content begin] import os import time import re import gspread import wikipedia import requests -import openai -import csv from bs4 import BeautifulSoup -from lxml import html as lh from oauth2client.service_account import ServiceAccountCredentials from datetime import datetime from difflib import SequenceMatcher +import csv -# === KONFIGURATION === -VERSION = "1.0.11" -LANG = "de" -CREDENTIALS = "service_account.json" -SHEET_URL = "https://docs.google.com/spreadsheets/d/1u_gHr9JUfmV1-iviRzbSe3575QEp7KLhK5jFV_gJcgo" -DURCHLÄUFE = int(input("Wieviele Zeilen sollen überprüft werden? ")) -MAX_RETRIES = 3 -RETRY_DELAY = 5 -LOG_CSV = "gpt_antworten_log.csv" -SIMILARITY_THRESHOLD = 0.6 +# ==================== KONFIGURATION ==================== +class Config: + VERSION = "1.1.0" + LANG = "de" + CREDENTIALS_FILE = "service_account.json" + SHEET_URL = "https://docs.google.com/spreadsheets/d/1u_gHr9JUfmV1-iviRzbSe3575QEp7KLhK5jFV_gJcgo" + MAX_RETRIES = 3 + RETRY_DELAY = 5 + LOG_CSV = "gpt_antworten_log.csv" + SIMILARITY_THRESHOLD = 0.6 + DEBUG = True + WIKIPEDIA_SEARCH_RESULTS = 8 + HTML_PARSER = "html.parser" -# === OpenAI API-KEY LADEN === -with open("api_key.txt", "r") as f: - openai.api_key = f.read().strip() - -# === GOOGLE SHEET VERBINDUNG === -scope = ["https://www.googleapis.com/auth/spreadsheets"] -creds = ServiceAccountCredentials.from_json_keyfile_name(CREDENTIALS, scope) -sheet = gspread.authorize(creds).open_by_url(SHEET_URL).sheet1 -sheet_values = sheet.get_all_values() - -# === STARTINDEX SUCHEN (Spalte N = Index 13) === -filled_n = [row[13] if len(row) > 13 else '' for row in sheet_values[1:]] -start = next((i + 1 for i, v in enumerate(filled_n, start=1) if not str(v).strip()), len(filled_n) + 1) -print(f"Starte bei Zeile {start+1}") - -wikipedia.set_lang(LANG) - -def similar(a, b): - return SequenceMatcher(None, a.lower(), b.lower()).ratio() - -def get_wikipedia_data(name, website_hint=""): - begriffe = [name.strip(), " ".join(name.split()[:2])] - domain_key = "" - if website_hint: - parts = website_hint.replace("https://", "").replace("http://", "").split(".") - if len(parts) > 1: - domain_key = parts[0] - begriffe.append(domain_key) - - for suchbegriff in begriffe: - try: - results = wikipedia.search(suchbegriff, results=5) - except: - continue - - for title in results: +# ==================== HELPER FUNCTIONS ==================== +def retry_on_failure(func): + """Decorator für Wiederholungsversuche bei Fehlern""" + def wrapper(*args, **kwargs): + for attempt in range(Config.MAX_RETRIES): try: - page = wikipedia.page(title, auto_suggest=False) - html_raw = requests.get(page.url).text - if domain_key and domain_key not in html_raw.lower(): - continue - if similar(page.title, name) < SIMILARITY_THRESHOLD: - continue + return func(*args, **kwargs) + except Exception as e: + print(f"⚠️ Fehler bei {func.__name__} (Versuch {attempt+1}): {str(e)[:100]}") + time.sleep(Config.RETRY_DELAY) + return None + return wrapper - soup = BeautifulSoup(html_raw, 'html.parser') - infobox = soup.find("table", class_="infobox") - branche = umsatz = "" - if infobox: - for row in infobox.find_all("tr"): - th, td = row.find("th"), row.find("td") - if not th or not td: - continue - if "branche" in th.text.lower(): - branche = td.text.strip() - if "umsatz" in th.text.lower(): - umsatz_raw = td.text.strip() - umsatz = re.sub(r"\[[^\]]*\]", "", umsatz_raw) +def debug_print(message): + """Debug-Ausgabe, wenn Config.DEBUG=True""" + if Config.DEBUG: + print(f"[DEBUG] {message}") - if not branche: - cats = page.categories - branche = cats[0] if cats else "k.A." - return page.url, branche or "k.A.", umsatz or "k.A." - except: +def clean_text(text): + """Bereinigt Text von HTML-Entitäten und überflüssigen Whitespaces""" + if not text: + return "k.A." + + # Konvertierung und Säuberung + text = str(text) + text = re.sub(r'\[.*?\]', '', text) # Entferne eckige Klammern mit Inhalt + text = re.sub(r'\(.*?\)', '', text) # Entferne runde Klammern mit Inhalt + text = re.sub(r'<.*?>', '', text) # Entferne HTML-Tags + text = re.sub(r'\s+', ' ', text).strip() + return text if text else "k.A." + +# ==================== GOOGLE SHEET HANDLER ==================== +class GoogleSheetHandler: + """Klasse zur Handhabung der Google Sheets Interaktion""" + + def __init__(self): + self.sheet = None + self.sheet_values = [] + self._connect() + + 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 + ) + 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 (Index 13)""" + 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 + ) + + def update_row(self, row_num, values): + """Aktualisiert eine Zeile im Sheet""" + self.sheet.update( + range_name=f"G{row_num}:Q{row_num}", + values=[values] + ) + +# ==================== WIKIPEDIA SCRAPER ==================== +class WikipediaScraper: + """Klasse zur Handhabung der Wikipedia-Suche und Datenextraktion""" + + def __init__(self): + wikipedia.set_lang(Config.LANG) + + def _generate_search_terms(self, company_name, website_hint=""): + """Generiert Suchbegriffe aus Firmenname und Website""" + search_terms = [company_name.strip()] + + # Zusatzbegriffe aus Firmennamen + name_parts = company_name.split() + if len(name_parts) > 1: + search_terms.append(" ".join(name_parts[:2])) + + # Bereinigung von Rechtsformen + clean_name = re.sub(r'\s+(?:GmbH|AG|KG|OHG|e\.V\.|mbH).*$', '', company_name) + if clean_name != company_name: + search_terms.append(clean_name) + + # Extraktion aus Website + if website_hint: + domain_parts = website_hint.replace("https://", "").replace("http://", "").replace("www.", "").split(".") + if len(domain_parts) > 1 and domain_parts[0] not in ["de", "com", "org"]: + search_terms.append(domain_parts[0]) + + debug_print(f"Generierte Suchbegriffe: {search_terms}") + return search_terms + + def _validate_article(self, page, company_name, domain_hint=""): + """Überprüft ob der Artikel zum Unternehmen passt""" + # Ähnlichkeitsprüfung des Titels + title_similarity = SequenceMatcher( + None, + page.title.lower(), + company_name.lower() + ).ratio() + + # Zusätzliche Domain-Prüfung + if domain_hint: + html_content = requests.get(page.url).text + if domain_hint.lower() not in html_content.lower(): + return False + + return title_similarity >= Config.SIMILARITY_THRESHOLD + + @retry_on_failure + def search_company_article(self, company_name, website_hint=""): + """Hauptfunktion zur Artikelsuche""" + search_terms = self._generate_search_terms(company_name, website_hint) + domain_hint = self._extract_domain_hint(website_hint) + + for term in search_terms: + try: + results = wikipedia.search(term, results=Config.WIKIPEDIA_SEARCH_RESULTS) + debug_print(f"Suchergebnisse für '{term}': {results}") + + for title in results: + try: + page = wikipedia.page(title, auto_suggest=False) + if self._validate_article(page, company_name, domain_hint): + return page + except wikipedia.exceptions.DisambiguationError: + continue + except Exception as e: + debug_print(f"Fehler bei Suche nach {term}: {str(e)}") continue + return None + + def extract_company_data(self, page_url): + """Extrahiert Branche und Umsatz aus dem Wikipedia-Artikel""" + response = requests.get(page_url) + soup = BeautifulSoup(response.text, Config.HTML_PARSER) + + return { + 'branche': self._extract_infobox_value(soup, 'branche'), + 'umsatz': self._extract_infobox_value(soup, 'umsatz'), + 'url': page_url + } + + def _extract_infobox_value(self, soup, target): + """Extrahiert spezifischen Wert aus der Infobox""" + infobox = soup.find('table', class_=lambda c: c and 'infobox' in c.lower()) + if not infobox: + return "k.A." + + # Definiere Keywords für verschiedene Targets + keywords = { + 'branche': ['branche', 'tätigkeitsfeld', 'geschäftsfeld', 'sektor'], + 'umsatz': ['umsatz', 'jahresumsatz', 'konzernumsatz', 'umsatzerlöse'] + }.get(target, []) + + # Durchsuche Infobox-Zeilen + for row in infobox.find_all('tr'): + header = row.find('th') + if header and any(kw in clean_text(header).lower() for kw in keywords): + value = row.find('td') + return clean_text(value) if value else "k.A." + + return "k.A." - return "", "k.A.", "k.A." +# ==================== DATA PROCESSOR ==================== +class DataProcessor: + """Klasse zur Steuerung des Gesamtprozesses""" + + def __init__(self): + self.sheet_handler = GoogleSheetHandler() + self.wiki_scraper = WikipediaScraper() + + def process_rows(self, num_rows): + """Verarbeitet die angegebene Anzahl an Zeilen""" + start_index = self.sheet_handler.get_start_index() + print(f"Starte bei Zeile {start_index+1}") + + for i in range(start_index, min(start_index + num_rows, len(self.sheet_handler.sheet_values))): + row = self.sheet_handler.sheet_values[i] + self._process_single_row(i+1, row) + + def _process_single_row(self, row_num, row_data): + """Verarbeitet eine einzelne Zeile""" + company_name = row_data[0] if len(row_data) > 0 else "" + website = row_data[1] if len(row_data) > 1 else "" + print(f"\n[{datetime.now().strftime('%H:%M:%S')}] Verarbeite Zeile {row_num}: {company_name}") + + # Schritt 1: Wikipedia-Artikel finden + article = self.wiki_scraper.search_company_article(company_name, website) + + # Schritt 2: Daten extrahieren + if article: + company_data = self.wiki_scraper.extract_company_data(article.url) + else: + company_data = {'branche': 'k.A.', 'umsatz': 'k.A.', 'url': ''} + + # Aktualisiere Daten im Sheet + self._update_sheet(row_num, company_data) + time.sleep(Config.RETRY_DELAY) + + def _update_sheet(self, row_num, data): + """Aktualisiert die Zeile mit den neuen Daten""" + current_values = self.sheet_handler.sheet.row_values(row_num) + new_values = [ + data['branche'] if data['branche'] != "k.A." else current_values[6] if len(current_values) > 6 else "k.A.", + "k.A.", # LinkedIn-Branche bleibt unverändert + data['umsatz'] if data['umsatz'] != "k.A." else current_values[8] if len(current_values) > 8 else "k.A.", + "k.A.", "k.A.", "k.A.", + data['url'] if data['url'] else current_values[12] if len(current_values) > 12 else "", + datetime.now().strftime("%Y-%m-%d %H:%M:%S"), + "k.A.", "k.A.", + Config.VERSION + ] + self.sheet_handler.update_row(row_num, new_values) + print(f"✅ Aktualisiert: Branche: {new_values[0]}, Umsatz: {new_values[2]}, URL: {new_values[6]}") -# === VERARBEITUNG === -for i in range(start, min(start + DURCHLÄUFE, len(sheet_values))): - row = sheet_values[i] - print(f"\n[{datetime.now().strftime('%H:%M:%S')}] Verarbeite Zeile {i+1}: {row[0]}") - url, branche, umsatz = get_wikipedia_data(row[0], row[1]) - branche_final = branche if url else "k.A." - umsatz_final = umsatz if url else "k.A." - values = [ - branche_final, - "k.A.", - umsatz_final, - "k.A.", "k.A.", "k.A.", - url, - datetime.now().strftime("%Y-%m-%d %H:%M:%S"), - "k.A.", "k.A.", - VERSION - ] - sheet.update(range_name=f"G{i+1}:Q{i+1}", values=[values]) - print(f"✅ Aktualisiert: {values[:3]}...") - time.sleep(RETRY_DELAY) - -print("\n✅ Wikipedia-Auswertung abgeschlossen") - - - - - - -# === SCHRITT 2: GPT-BEWERTUNG === -def classify_company(row, wikipedia_url=""): - user_prompt = { - "role": "user", - "content": f"{row[0]};{row[1]};{row[2]};{row[4]};{row[5]}\nWikipedia-Link: {wikipedia_url}" - } - for attempt in range(MAX_RETRIES): - try: - response = openai.chat.completions.create( - model="gpt-3.5-turbo", - messages=[ - { - "role": "system", - "content": ( - "Du bist ein Experte für Brancheneinstufung und FSM-Potenzialbewertung.\n" - "Bitte beziehe dich ausschließlich auf das konkret genannte Unternehmen.\n" - "FSM steht für Field Service Management. Ziel ist es, Unternehmen mit >50 Technikern im Außendienst zu identifizieren.\n\n" - "Struktur: Firmenname; Website; Ort; Aktuelle Einstufung; Beschreibung der Branche Extern\n\n" - "Gib deine Antwort im CSV-Format zurück (1 Zeile, 8 Spalten):\n" - "Wikipedia-Branche;LinkedIn-Branche;Umsatz (Mio €);Empfohlene Neueinstufung;Begründung;FSM-Relevanz;Techniker-Einschätzung;Techniker-Begründung" - ) - }, - user_prompt - ], - temperature=0, - timeout=15 - ) - full_text = response.choices[0].message.content.strip() - break - except Exception as e: - print(f"⚠️ GPT-Fehler (Versuch {attempt+1}): {str(e)[:100]}") - time.sleep(RETRY_DELAY) - else: - print("❌ GPT 3x fehlgeschlagen – Standardwerte") - full_text = "k.A.;k.A.;k.A.;k.A.;k.A.;k.A.;k.A.;k.A." - - lines = full_text.splitlines() - csv_line = next((l for l in lines if ";" in l), "") - parts = [v.strip() for v in csv_line.split(";")] if csv_line else ["k.A."] * 8 - - with open(LOG_CSV, "a", newline="", encoding="utf-8") as log: - writer = csv.writer(log, delimiter=";") - writer.writerow([datetime.now().strftime("%Y-%m-%d %H:%M:%S"), row[0], *parts, full_text]) - - return parts - -# === SCHRITT 2 DURCHFÜHREN === -for i in range(start, min(start + DURCHLÄUFE, len(sheet_values))): - row = sheet_values[i] - print(f"\n[{datetime.now().strftime('%H:%M:%S')}] GPT-Bewertung für Zeile {i+1}: {row[0]}") - wiki_url = row[12] if len(row) > 12 else "" - wiki, linkedin, umsatz_chat, new_cat, reason, fsm, techniker, techniker_reason = classify_company(row, wikipedia_url=wiki_url) - values = [ - wiki, - linkedin, - umsatz_chat, - new_cat, - reason, - fsm, - wiki_url, - datetime.now().strftime("%Y-%m-%d %H:%M:%S"), - techniker, - techniker_reason - ] - sheet.update(range_name=f"G{i+1}:P{i+1}", values=[values]) - time.sleep(RETRY_DELAY) - -print("\n✅ GPT-Bewertung abgeschlossen") +# ==================== MAIN EXECUTION ==================== +if __name__ == "__main__": + num_rows = int(input("Wieviele Zeilen sollen überprüft werden? ")) + processor = DataProcessor() + processor.process_rows(num_rows) + print("\n✅ Wikipedia-Auswertung abgeschlossen") +[file content end] \ No newline at end of file