Deepseek V1
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.
This commit is contained in:
@@ -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]
|
||||
Reference in New Issue
Block a user