diff --git a/knowledge_base_builder.py b/knowledge_base_builder.py new file mode 100644 index 00000000..8f4d7b92 --- /dev/null +++ b/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() \ No newline at end of file