knowledge_base_builder.py hinzugefügt

knowledge_base_builder.py

@@ -0,0 +1,148 @@
+__version__ = "v1.0.0"
+
+import logging
+import json
+import re
+from collections import Counter
+import pandas as pd
+
+# Importiere die existierenden, robusten Handler und Konfigurationen
+from google_sheet_handler import GoogleSheetHandler
+from config import Config
+
+# --- Konfiguration ---
+# Name des Tabellenblatts, das die Rohdaten der Jobtitel enthält
+SOURCE_SHEET_NAME = "CRM_Jobtitles"
+# Namen der finalen Ausgabedateien
+EXACT_MATCH_OUTPUT_FILE = "exact_match_map.json"
+KEYWORD_RULES_OUTPUT_FILE = "keyword_rules.json"
+
+# Priorisierung der Departments (von spezifisch zu allgemein)
+# Niedrigere Zahl = höhere Priorität
+DEPARTMENT_PRIORITIES = {
+    "Fuhrparkmanagement": 1,
+    "Production Maintenance / Wartung Produktion": 1,
+    "Utility Maintenance": 1,
+    "Baustofflogistik": 1,
+    "Baustoffherstellung": 1,
+    "Legal": 1,
+    "Technik": 2,
+    "IT": 2,
+    "Finanzen": 3,
+    "Procurement / Einkauf": 3,
+    "Supply Chain Management": 3,
+    "Field Service Management / Kundenservice": 4,
+    "Logistik": 4,
+    "Transportwesen": 4,
+    "Vertrieb": 5,
+    "Berater": 6,
+    "Management / GF / C-Level": 7,
+    "Undefined": 99  # Niedrigste Priorität
+}
+
+# Stoppwörter: Häufige Wörter in Jobtiteln, die wenig Aussagekraft für die Abteilung haben
+STOP_WORDS = {
+    'manager', 'leiter', 'head', 'lead', 'senior', 'junior', 'direktor', 'director',
+    'verantwortlicher', 'beauftragter', 'referent', 'sachbearbeiter', 'mitarbeiter',
+    'spezialist', 'specialist', 'expert', 'experte', 'consultant', 'berater',
+    'assistant', 'assistenz', 'teamleiter', 'teamlead', 'abteilungsleiter',
+    'bereichsleiter', 'gruppenleiter', 'geschäftsführer', 'vorstand', 'ceo', 'cio',
+    'cfo', 'cto', 'coo', 'von', 'of', 'und', 'für', 'der', 'die', 'das', '&'
+}
+
+
+def build_knowledge_base():
+    """
+    Hauptfunktion zur Erstellung der Wissensbasis.
+    Liest die Rohdaten aus Google Sheets, analysiert sie und erstellt zwei JSON-Dateien.
+    """
+    logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+    logger = logging.getLogger(__name__)
+
+    logger.info(f"Starte Erstellung der Wissensbasis (Version {__version__})...")
+
+    # 1. Daten aus Google Sheet laden
+    try:
+        gsh = GoogleSheetHandler()
+        df = gsh.get_sheet_as_dataframe(SOURCE_SHEET_NAME)
+
+        if df is None or df.empty:
+            logger.critical(f"Konnte keine Daten aus '{SOURCE_SHEET_NAME}' laden oder das Tabellenblatt ist leer. Abbruch.")
+            return
+        
+        # Spaltennamen normalisieren (z.B. Leerzeichen am Ende entfernen)
+        df.columns = [col.strip() for col in df.columns]
+        
+        if "Job Title" not in df.columns or "Department" not in df.columns:
+            logger.critical(f"Benötigte Spalten 'Job Title' und/oder 'Department' nicht in '{SOURCE_SHEET_NAME}' gefunden. Abbruch.")
+            return
+
+    except Exception as e:
+        logger.critical(f"Ein kritischer Fehler ist beim Laden der Google Sheet Daten aufgetreten: {e}")
+        return
+
+    logger.info(f"{len(df)} Zeilen erfolgreich aus '{SOURCE_SHEET_NAME}' geladen.")
+    
+    # 2. Daten bereinigen und vorbereiten
+    df.dropna(subset=["Job Title", "Department"], inplace=True)
+    df = df[df["Job Title"].str.strip() != '']
+    df['normalized_title'] = df['Job Title'].str.lower().str.strip()
+    logger.info(f"{len(df)} Zeilen nach Bereinigung (Entfernen leerer Jobtitel/Departments).")
+
+    # 3. Stufe 1: "Primary Mapping" für exakte Treffer erstellen
+    logger.info("Erstelle 'Primary Mapping' für exakte Treffer (Stufe 1)...")
+    # Für jeden Jobtitel, finde das am häufigsten zugewiesene Department
+    # .mode()[0] ist ein robuster Weg, den häufigsten Wert zu bekommen
+    exact_match_map = df.groupby('normalized_title')['Department'].apply(lambda x: x.mode()[0]).to_dict()
+    
+    try:
+        with open(EXACT_MATCH_OUTPUT_FILE, 'w', encoding='utf-8') as f:
+            json.dump(exact_match_map, f, indent=4, ensure_ascii=False)
+        logger.info(f"-> '{EXACT_MATCH_OUTPUT_FILE}' mit {len(exact_match_map)} einzigartigen Jobtiteln erfolgreich erstellt.")
+    except IOError as e:
+        logger.error(f"Fehler beim Schreiben der Datei '{EXACT_MATCH_OUTPUT_FILE}': {e}")
+        return
+
+    # 4. Stufe 2: "Keyword-Datenbank" für regelbasiertes Matching erstellen
+    logger.info("Erstelle 'Keyword-Datenbank' mit Prioritäten (Stufe 2)...")
+    
+    # Ordne jedem Department eine Liste seiner (normalisierten) Jobtitel zu
+    titles_by_department = df.groupby('Department')['normalized_title'].apply(list).to_dict()
+    
+    keyword_rules = {}
+    for department, titles in titles_by_department.items():
+        all_words = []
+        # Zerlege alle Jobtitel in einzelne Wörter
+        for title in titles:
+            words = re.split(r'[\s/(),-]+', title) # Trennt bei Leerzeichen, /, (, ), -
+            all_words.extend([word for word in words if word])
+
+        # Zähle die Worthäufigkeiten und filtere die relevantesten
+        word_counts = Counter(all_words)
+        top_keywords = []
+        for word, count in word_counts.most_common(50): # Nimm die 50 häufigsten als Kandidaten
+            # Keyword muss aussagekräftig sein
+            if word not in STOP_WORDS and (len(word) > 2 or word in {'it', 'edv'}):
+                top_keywords.append(word)
+
+        if top_keywords:
+            priority = DEPARTMENT_PRIORITIES.get(department, 99) # 99 als Fallback-Priorität
+            keyword_rules[department] = {
+                "priority": priority,
+                "keywords": sorted(top_keywords)
+            }
+            logger.debug(f"  - Department '{department}' (Prio {priority}): {len(top_keywords)} Keywords gefunden (z.B. {top_keywords[:5]}).")
+
+    try:
+        with open(KEYWORD_RULES_OUTPUT_FILE, 'w', encoding='utf-8') as f:
+            json.dump(keyword_rules, f, indent=4, ensure_ascii=False)
+        logger.info(f"-> '{KEYWORD_RULES_OUTPUT_FILE}' mit Regeln für {len(keyword_rules)} Departments erfolgreich erstellt.")
+    except IOError as e:
+        logger.error(f"Fehler beim Schreiben der Datei '{KEYWORD_RULES_OUTPUT_FILE}': {e}")
+        return
+        
+    logger.info("Wissensbasis erfolgreich erstellt.")
+
+
+if __name__ == "__main__":
+    build_knowledge_base()