# extract_insights.py import os import yaml import logging import time import openai import docx # Die neue Bibliothek zur Verarbeitung von Word-Dokumenten from config import Config # --- Konfiguration --- DOCS_SOURCE_FOLDER = "industry_docs" # Der Ordner, in dem Ihre .docx-Dateien liegen OUTPUT_FILE = "marketing_wissen_v1.yaml" MODEL_TO_USE = "gpt-4-turbo" # Empfohlen für komplexe Extraktionsaufgaben # --- Logging einrichten --- logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') def call_openai_with_retry(prompt, max_retries=3, delay=5): """Ruft die OpenAI API mit Retry-Logik auf.""" for attempt in range(max_retries): try: logging.info(f"Sende Prompt an OpenAI (Länge: {len(prompt)} Zeichen)...") response = openai.ChatCompletion.create( model=MODEL_TO_USE, messages=[{"role": "user", "content": prompt}], temperature=0.2, # Niedrige Temperatur für präzise Extraktion max_tokens=1024 ) content = response.choices[0].message['content'].strip() return content except Exception as e: logging.error(f"Fehler bei OpenAI-API-Aufruf: {e}") if attempt < max_retries - 1: logging.info(f"Warte {delay} Sekunden vor dem nächsten Versuch...") time.sleep(delay) else: logging.error("Maximale Anzahl an Wiederholungen erreicht.") return None def read_docx_content(filepath): """Liest den gesamten Textinhalt aus einer .docx-Datei, inklusive Tabellen.""" try: doc = docx.Document(filepath) full_text = [] for para in doc.paragraphs: full_text.append(para.text) for table in doc.tables: for row in table.rows: for cell in row.cells: full_text.append(cell.text) return "\n".join(full_text) except Exception as e: logging.error(f"Fehler beim Lesen der DOCX-Datei {filepath}: {e}") return None def extract_yaml_from_response(response_text): """ Extrahiert sauberen YAML-Code aus einer KI-Antwort, die Markdown-Codeblöcke enthalten kann. """ # Sucht nach dem Start des YAML-Codeblocks if '```yaml' in response_text: # Extrahiert den Teil nach dem ersten ```yaml parts = response_text.split('```yaml', 1) if len(parts) > 1: response_text = parts[1] # Sucht nach dem Start eines generischen Codeblocks elif '```' in response_text: # Extrahiert den Teil nach dem ersten ``` parts = response_text.split('```', 1) if len(parts) > 1: response_text = parts[1] # Entfernt das Ende des Codeblocks if '```' in response_text: response_text = response_text.split('```')[0] return response_text.strip() def generate_extraction_prompt(content, data_to_extract): """Erstellt einen spezialisierten Prompt, um bestimmte Daten zu extrahieren.""" prompts = { "pain_points": ( "Du bist ein Branchenanalyst. Lies das folgende Dokument und extrahiere die 5 wichtigsten operativen " "Herausforderungen (Pain Points) für Unternehmen dieser Branche im Bereich Field Service. " "Formuliere sie als prägnante Stichpunkte.\n\n" "Gib das Ergebnis ausschließlich als YAML-Liste unter dem Schlüssel 'pain_points:' aus. KEINE weiteren Kommentare." ), "key_terms": ( "Du bist ein Fachlexikograf. Lies das folgende Dokument und extrahiere die 10 wichtigsten Fachbegriffe, " "Abkürzungen oder Normen, die im Kontext von Service, Wartung und Technik verwendet werden.\n\n" "Gib das Ergebnis ausschließlich als YAML-Liste unter dem Schlüssel 'key_terms:' aus." ), "summary": ( "Du bist ein Chefredakteur. Lies das folgende Dokument und verfasse eine prägnante Zusammenfassung (max. 3 Sätze) " "über die allgemeine Geschäftslage, die wichtigsten Trends und die Bedeutung des Field Service in dieser Branche.\n\n" "Gib das Ergebnis ausschließlich als einfachen Text unter dem YAML-Schlüssel 'summary:' aus." ) } if data_to_extract not in prompts: raise ValueError(f"Unbekannter Extraktionstyp: {data_to_extract}") return f"{prompts[data_to_extract]}\n\n--- DOKUMENTENINHALT ---\n\n{content}" def main(): """Liest .docx-Dateien, extrahiert Wissen per KI und speichert es als YAML.""" logging.info("Starte die KI-gestützte Extraktion von Branchen-Wissen...") # API-Schlüssel laden Config.load_api_keys() openai.api_key = Config.API_KEYS.get('openai') if not openai.api_key: logging.critical("OpenAI API Key nicht in config.py gefunden. Skript wird beendet.") return if not os.path.exists(DOCS_SOURCE_FOLDER): logging.critical(f"Der Quellordner '{DOCS_SOURCE_FOLDER}' wurde nicht gefunden. Bitte erstellen und die .docx-Dateien dort ablegen.") return knowledge_base = {'Branchen': {}} doc_files = [f for f in os.listdir(DOCS_SOURCE_FOLDER) if f.endswith('.docx')] logging.info(f"Gefundene Dokumente zur Verarbeitung: {', '.join(doc_files)}") for filename in doc_files: # Extrahiere den Branchennamen aus dem Dateinamen # z.B. "Focus_insights_HVAC.docx" -> "Gebäudetechnik Heizung, Lüftung, Klima" # Dies muss manuell oder durch eine Mapping-Tabelle angepasst werden. # Für den Moment nehmen wir den Namen aus der Datei. base_name = os.path.splitext(filename)[0].replace("Focus_insights_", "") # Sie können hier ein Mapping zu den sauberen Namen aus Ihrer `config.py` einfügen. # Beispiel: branch_name = MAPPING.get(base_name, base_name) branch_name = base_name.replace("_", " ") # Einfache Normalisierung für den Start logging.info(f"\n--- Verarbeite Branche: {branch_name} aus Datei {filename} ---") filepath = os.path.join(DOCS_SOURCE_FOLDER, filename) content = read_docx_content(filepath) if not content: continue branch_data = { 'references_DE': '[HIER DEUTSCHE REFERENZKUNDEN EINTRAGEN]', 'references_GB': '[HIER ENGLISCHE REFERENZKUNDEN EINTRAGEN]' } # Extrahiere Pain Points, Key Terms und Summary for data_type in ["pain_points", "key_terms", "summary"]: logging.info(f" -> Extrahiere '{data_type}'...") prompt = generate_extraction_prompt(content, data_type) response_text = call_openai_with_retry(prompt) if response_text: try: # NEU: Erst den sauberen YAML-Teil extrahieren clean_yaml_text = extract_yaml_from_response(response_text) # Dann den sauberen Text parsen parsed_yaml = yaml.safe_load(clean_yaml_text) if parsed_yaml: # Sicherstellen, dass das Ergebnis nicht leer ist branch_data.update(parsed_yaml) else: raise ValueError("Geparsstes YAML ist leer.") except Exception as e: logging.error(f" Fehler beim Parsen der YAML-Antwort für '{data_type}': {e}") # Speichere die *gesamte* ursprüngliche Antwort für Debugging-Zwecke branch_data[data_type] = f"PARSING-FEHLER: {response_text}" time.sleep(2) # Pause zwischen API-Aufrufen knowledge_base['Branchen'][branch_name] = branch_data # Persona-Daten hinzufügen (diese sind statisch) # Hier können Sie die Persona-Daten aus der letzten Iteration einfügen. # ... # Ergebnis in YAML-Datei speichern try: with open(OUTPUT_FILE, 'w', encoding='utf-8') as f: yaml.dump(knowledge_base, f, allow_unicode=True, sort_keys=False, width=120) logging.info(f"\nErfolgreich! Die Wissensbasis wurde in '{OUTPUT_FILE}' gespeichert.") logging.info("BITTE ÜBERPRÜFEN SIE DIESE DATEI UND PASSEN SIE SIE NACH BEDARF AN.") except Exception as e: logging.error(f"Fehler beim Speichern der YAML-Datei: {e}") if __name__ == "__main__": main()