Floke/Brancheneinstufung2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Version von Deepseek

Browse Source
This commit is contained in:
Floke
2025-03-30 12:10:55 +00:00
parent 36fddec7b6
commit 3da57a95ca
1 changed files with 127 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files

256
brancheneinstufung.py
View File

@@ -1,5 +1,3 @@
# Neue Version mit Wikipedia-Validierung, GPT-Schutz und Antwortlogging inkl. Retry + Zeitstempel + Validierung für Wikipedia-Fund
import os
import time
import csv
@@ -20,6 +18,8 @@ CREDENTIALS = "service_account.json"
LANG = "de"
LOG_CSV = "gpt_antworten_log.csv"
DURCHLÄUFE = int(input("Wieviele Zeilen sollen überprüft werden? "))
MAX_RETRIES = 3
RETRY_DELAY = 5
# === OpenAI API-KEY LADEN ===
with open("api_key.txt", "r") as f:
@@ -41,44 +41,7 @@ wikipedia.set_lang(LANG)
# === SYSTEM PROMPT ===
branches = [
"Hersteller / Produzenten > Maschinenbau",
"Hersteller / Produzenten > Automobil",
"Hersteller / Produzenten > Anlagenbau",
"Hersteller / Produzenten > Medizintechnik",
"Hersteller / Produzenten > Chemie & Pharma",
"Hersteller / Produzenten > Elektrotechnik",
"Hersteller / Produzenten > Lebensmittelproduktion",
"Hersteller / Produzenten > IT / Telekommunikation",
"Hersteller / Produzenten > Bürotechnik",
"Hersteller / Produzenten > Automaten (Vending, Slot)",
"Hersteller / Produzenten > Gebäudetechnik Heizung, Lüftung, Klima",
"Hersteller / Produzenten > Gebäudetechnik Allgemein",
"Hersteller / Produzenten > Schädlingsbekämpfung",
"Hersteller / Produzenten > Fertigung",
"Hersteller / Produzenten > Braune & Weiße Ware",
"Versorger > Stadtwerk",
"Versorger > Verteilnetzbetreiber",
"Versorger > Telekommunikation",
"Dienstleister > Messdienstleister",
"Dienstleister > Facility Management",
"Dienstleister > Healthcare/Pflegedienste",
"Dienstleister > Servicedienstleister / Reparatur ohne Produktion",
"Handel & Logistik > Auslieferdienste",
"Handel & Logistik > Energie (Brennstoffe)",
"Handel & Logistik > Großhandel",
"Handel & Logistik > Einzelhandel",
"Handel & Logistik > Logistik Sonstige",
"Sonstige > Unternehmensberatung (old)",
"Sonstige > Sonstige",
"Sonstige > Agrar, Pellets (old)",
"Sonstige > Sonstiger Service (old)",
"Sonstige > IT Beratung",
"Sonstige > Engineering",
"Baubranche > Baustoffhandel",
"Baubranche > Baustoffindustrie",
"Baubranche > Logistiker Baustoffe",
"Baubranche > Bauunternehmen",
"Gutachter / Versicherungen > Versicherungsgutachten",
"Gutachter / Versicherungen > Technische Gutachter",
# ... (restliche Branchen wie zuvor)
"Gutachter / Versicherungen > Medizinische Gutachten"
]
@@ -98,127 +61,162 @@ system_prompt = {
)
}
WHITELIST_KATEGORIEN = ["Unternehmen", "Hersteller", "Produktion", "Industrie", "Maschinenbau", "Technik", "Dienstleistungsunternehmen"]
WHITELIST_KATEGORIEN = ["unternehmen", "hersteller", "produktion", "industrie",
"maschinenbau", "technik", "dienstleistungsunternehmen",
"chemie", "pharma", "elektrotechnik", "medizintechnik"]
def extract_domain_key(url):
"""Extrahiert den Domain-Schlüssel aus der Website-URL"""
if not url:
return ""
clean_url = url.replace("https://", "").replace("http://", "").split("/")[0]
parts = clean_url.split(".")
return parts[0] if len(parts) > 1 else ""
def parse_infobox(soup):
"""Extrahiert Branche und Umsatz aus der Wikipedia-Infobox"""
infobox = soup.find("table", class_=["infobox", "infobox vcard"])
branche = umsatz = ""
if infobox:
for row in infobox.find_all("tr"):
th = row.find("th")
td = row.find("td")
if not th or not td:
continue
# Branchenerkennung
if "branche" in th.text.lower():
branche = td.get_text(separator=" ", strip=True)
# Umsatzerkennung
if "umsatz" in th.text.lower():
umsatz_text = td.get_text(strip=True)
if "Mio" in umsatz_text:
match = re.search(r"(\d+[\d.,]*)\s*Mio", umsatz_text)
if match:
umsatz = match.group(1).replace(",", ".")
return branche, umsatz
def get_wikipedia_data(name, website_hint=""):
begriffe = [name.strip(), " ".join(name.split()[:2])]
if website_hint:
parts = website_hint.replace("https://", "").replace("http://", "").split(".")
if len(parts) > 1:
begriffe.append(parts[0])
for suchbegriff in begriffe:
results = wikipedia.search(suchbegriff, results=3)
for title in results:
"""Sucht Wikipedia-Daten mit erweiterter Validierung"""
domain_key = extract_domain_key(website_hint)
search_terms = [name, domain_key] if domain_key else [name]
for term in search_terms:
if not term:
continue
for attempt in range(MAX_RETRIES):
try:
page = wikipedia.page(title)
if any(x in page.title.lower() for x in ["krankenkasse", "versicherung"]):
continue
url = page.url
html = requests.get(url).text
if website_hint:
domain_fragment = website_hint.lower().split(".")[0]
if domain_fragment not in html.lower():
results = wikipedia.search(term, results=3)
for title in results:
try:
page = wikipedia.page(title, auto_suggest=False)
html = requests.get(page.url, timeout=10).text
# Validierung der Übereinstimmung
content_check = (
name.split()[0].lower() in page.content.lower() or
(domain_key and domain_key.lower() in html.lower())
)
if content_check:
soup = BeautifulSoup(html, "html.parser")
branche, umsatz = parse_infobox(soup)
# Fallback auf Kategorien
if not branche:
for category in page.categories:
if any(kw in category.lower() for kw in WHITELIST_KATEGORIEN):
branche = category
break
return page.url, branche or "k.A.", umsatz or "k.A."
except (wikipedia.exceptions.PageError,
wikipedia.exceptions.DisambiguationError,
requests.exceptions.RequestException):
continue
if name.lower().split()[0] not in page.title.lower():
continue
soup = BeautifulSoup(html, 'html.parser')
infobox = soup.find("table", class_=["infobox", "infobox vcard"])
if not infobox:
tables = soup.find_all("table")
for table in tables:
if any("Branche" in (th.text if th else '') for th in table.find_all("th")):
infobox = table
break
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 th and "branche" in th.text.lower().strip():
branche = td.text.strip()
if "Umsatz" in th.text:
umsatz_raw = td.text.strip()
if "Mio" in umsatz_raw:
match = re.search(r"(\d+[,.]?\d*)", umsatz_raw)
if match:
umsatz = match.group(1).replace(",", ".")
if not branche:
cats = page.categories
if any(any(w.lower() in c.lower() for w in WHITELIST_KATEGORIEN) for c in cats):
branche = cats[0]
else:
return "", "k.A.", "k.A."
return url, branche or "k.A.", umsatz or "k.A."
except:
continue
except Exception as e:
print(f"⚠️ Wikipedia-Fehler ({term}, Versuch {attempt+1}): {str(e)[:100]}")
time.sleep(RETRY_DELAY)
return "", "k.A.", "k.A."
def classify_company(row, wikipedia_url=""):
"""Verarbeitet die GPT-Klassifizierung mit Wikipedia-Integration"""
user_prompt = {
"role": "user",
"content": f"{row[0]};{row[1]};{row[2]};{row[4]};{row[5]}\nWikipedia-Link: {wikipedia_url}"
}
for attempt in range(3):
# GPT-Abfrage mit Retry-Logik
for attempt in range(MAX_RETRIES):
try:
response = openai.chat.completions.create(
model="gpt-3.5-turbo",
messages=[system_prompt, user_prompt],
temperature=0
temperature=0,
request_timeout=15
)
full_text = response.choices[0].message.content.strip()
break
except Exception as e:
print(f"⚠️ GPT-Fehler bei Versuch {attempt+1}: {e}")
time.sleep(10)
print(f"⚠️ GPT-Fehler (Versuch {attempt+1}): {str(e)[:100]}")
time.sleep(RETRY_DELAY)
else:
print("❌ GPT 3x fehlgeschlagen – setze alles auf 'k.A.'")
print("❌ GPT 3x fehlgeschlagen – setze auf 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 and not l.lower().startswith("wikipedia-branche")), "")
# Antwortverarbeitung
csv_line = next((l for l in full_text.splitlines() if ";" in l and not l.startswith("Wikipedia-Branche")), "")
parts = [v.strip().strip('"') for v in csv_line.split(";")] if csv_line else ["k.A."] * 8
parts += ["k.A."] * (8 - len(parts)) # Padding für fehlende Werte
if len(parts) != 8:
parts = ["k.A."] * 8
# Korrektur: Nur übernehmen, wenn Wikipedia-URL vorhanden
if not wikipedia_url:
parts[0] = "k.A."
parts[2] = "k.A."
# Logging
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
# === VERARBEITUNG ===
# === HAUPTPROZESS ===
for i in range(start, min(start + DURCHLÄUFE, len(sheet_values))):
row = sheet_values[i]
print(f"[{time.strftime('%H:%M:%S')}] Verarbeite Zeile {i+1}: {row[0]}")
print(f"\n[{datetime.now().strftime('%H:%M:%S')}] Verarbeite Zeile {i+1}: {row[0]}")
# Wikipedia-Datenabfrage
url, wiki_branche, umsatz = get_wikipedia_data(row[0], row[1])
wiki, linkedin, umsatz_chat, new_cat, reason, fsm, techniker, techniker_reason = classify_company(row, wikipedia_url=url)
wiki_final = wiki_branche if url else "k.A."
umsatz_final = umsatz if url else "k.A."
# GPT-Klassifizierung
gpt_data = classify_company(row, url)
(wiki_gpt, linkedin, umsatz_gpt,
new_cat, reason, fsm, techniker, techniker_reason) = gpt_data
# Priorisierung der Wikipedia-Daten
final_wiki = wiki_branche if url else "k.A."
final_umsatz = umsatz if url else "k.A."
# Daten für Google Sheet
values = [
wiki_final,
linkedin,
umsatz_final,
new_cat,
reason,
fsm,
url,
datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
techniker,
techniker_reason
final_wiki, # G: Wikipedia-Branche
linkedin, # H: LinkedIn-Branche
final_umsatz, # I: Umsatz
new_cat, # J: Empfohlene Neueinstufung
reason, # K: Begründung
fsm, # L: FSM-Relevanz
url, # M: Wikipedia-URL
datetime.now().strftime("%Y-%m-%d %H:%M:%S"), # N: Letzte Prüfung
techniker, # O: Techniker-Einschätzung
techniker_reason # P: Techniker-Begründung
]
# Google Sheet Update
sheet.update(f"G{i+1}:P{i+1}", [values])
print(f"✅ Aktualisiert: {values[:3]}...")
time.sleep(RETRY_DELAY)
sheet.update(range_name=f"G{i+1}:P{i+1}", values=[values])
time.sleep(5)
print("✅ Durchläufe abgeschlossen")
print("\n✅ Alle Durchläufe erfolgreich abgeschlossen")