diff --git a/brancheneinstufung.py b/brancheneinstufung.py index 4e4d8e63..7abb0bc5 100644 --- a/brancheneinstufung.py +++ b/brancheneinstufung.py @@ -1565,80 +1565,42 @@ def get_website_raw(url, max_length=1000, verify_cert=False): debug_print(f"Allgemeiner Fehler beim Scraping von {url}: {e}") return "k.A." -# Diese Funktion bleibt notwendig für den gebündelten OpenAI Call +# Die Hilfsfunktion summarize_batch_openai wird weiterhin benötigt +# (Code dafür bleibt wie in der Antwort von 16:24 Uhr) @retry_on_failure def summarize_batch_openai(tasks_data): - """ - Fasst eine Liste von Rohtexten in einem einzigen OpenAI API Call zusammen. - - Args: - tasks_data (list): Eine Liste von Dictionaries, jedes enthält: - {'row_num': int, 'raw_text': str} - - Returns: - dict: Ein Dictionary, das Zeilennummern auf ihre Zusammenfassungen mappt. - z.B. {2122: "Zusammenfassung A", 2123: "Zusammenfassung B"} - Bei Fehlern oder fehlenden Zusammenfassungen wird "k.A." verwendet. - """ - if not tasks_data: - return {} - + # ... (Implementierung wie zuvor) ... + if not tasks_data: return {} valid_tasks = [t for t in tasks_data if t.get("raw_text") and t["raw_text"] != "k.A." and t["raw_text"].strip()] - if not valid_tasks: - debug_print("Keine gültigen Rohtexte für Batch-Zusammenfassung gefunden.") - # Gib ein leeres Dict zurück, damit die aufrufende Funktion weiß, dass nichts zu tun war - # oder ein Dict mit k.A. für alle ursprünglichen Tasks? Besser letzteres. - return {t['row_num']: "k.A. (Kein gültiger Rohtext)" for t in tasks_data} - - + if not valid_tasks: return {t['row_num']: "k.A. (Kein gültiger Rohtext)" for t in tasks_data} debug_print(f"Starte Batch-Zusammenfassung für {len(valid_tasks)} gültige Texte...") - prompt_parts = [ - "Du bist ein KI-Assistent, der mehrere Website-Texte zusammenfasst.", - "Fasse jeden der folgenden Texte prägnant zusammen (max. 80-100 Wörter pro Text).", - "Konzentriere dich auf Haupttätigkeitsfeld, Produkte/Dienstleistungen und Zielgruppe.", - "Antworte ausschließlich mit den Ergebnissen, jede Zusammenfassung auf einer neuen Zeile, im folgenden Format:", + "Du bist ein KI-Assistent...", # Gekürzt für Lesbarkeit "RESULTAT : ", "\n--- Texte zur Zusammenfassung ---" ] - text_block = "" row_numbers_in_batch = [] total_chars = 0 - # Reduziere das Limit, da die Antwort auch Platz braucht - max_chars_per_batch = 6000 # Vorsichtiger für Prompt + Antwort Tokens - + max_chars_per_batch = 6000 for task in valid_tasks: row_num = task['row_num'] raw_text = task['raw_text'] - # Kürzen nicht mehr nötig, da in get_website_raw schon passiert - # max_raw_length_per_item = 1000 - # if len(raw_text) > max_raw_length_per_item: - # raw_text = raw_text[:max_raw_length_per_item] - entry_text = f"\n--- TEXT Zeile {row_num} ---\n{raw_text}\n--- ENDE TEXT Zeile {row_num} ---\n" if total_chars + len(entry_text) > max_chars_per_batch: - debug_print(f"WARNUNG: Batch-Zeichenlimit ({max_chars_per_batch}) erreicht bei Zeile {row_num}. Dieser Text wird nicht in den Batch aufgenommen.") + debug_print(f"WARNUNG: Batch-Zeichenlimit ({max_chars_per_batch}) erreicht bei Zeile {row_num}.") continue - text_block += entry_text total_chars += len(entry_text) row_numbers_in_batch.append(row_num) - if not row_numbers_in_batch: debug_print("Keine Texte im Batch nach Längenprüfung für OpenAI.") - # Gib k.A. für alle ursprünglichen Tasks zurück, für die wir einen Text hatten return {t['row_num']: "k.A. (Batch-Limit erreicht)" for t in valid_tasks} - prompt_parts.append(text_block) prompt_parts.append("--- Ende der Texte ---") prompt_parts.append("Bitte gib NUR die 'RESULTAT : ...' Zeilen zurück.") final_prompt = "\n".join(prompt_parts) - - # OpenAI API Call chat_response = call_openai_chat(final_prompt, temperature=0.2) - - # Antwort parsen summaries = {row_num: "k.A. (Keine Antwort geparst)" for row_num in row_numbers_in_batch} if chat_response: lines = chat_response.strip().split('\n') @@ -1648,30 +1610,19 @@ def summarize_batch_openai(tasks_data): if match: row_num = int(match.group(1)) summary_text = match.group(2).strip() - if row_num in summaries: - summaries[row_num] = summary_text - parsed_count += 1 - - debug_print(f"Batch-Zusammenfassung: {parsed_count} von {len(row_numbers_in_batch)} Zusammenfassungen erfolgreich geparst.") - if parsed_count < len(row_numbers_in_batch): - debug_print(f"WARNUNG: Nicht alle Zusammenfassungen konnten geparst werden.") - debug_print(f"Komplette ChatGPT Antwort (Batch Summary):\n{chat_response}") - else: - debug_print("Fehler: Keine gültige Antwort von OpenAI für Batch-Zusammenfassung erhalten.") - + if row_num in summaries: summaries[row_num] = summary_text; parsed_count += 1 + debug_print(f"Batch-Zusammenfassung: {parsed_count} von {len(row_numbers_in_batch)} erfolgreich geparst.") + if parsed_count < len(row_numbers_in_batch): debug_print(f"WARNUNG: Nicht alle Zusammenfassungen geparst. Antwort: {chat_response[:500]}...") + else: debug_print("Fehler: Keine gültige Antwort von OpenAI für Batch-Zusammenfassung.") # Füge k.A. für Tasks hinzu, die ursprünglich gültigen Text hatten, aber evtl. wegen Limit nicht im Batch waren for task in valid_tasks: - if task['row_num'] not in summaries: - summaries[task['row_num']] = "k.A. (Nicht im OpenAI-Batch enthalten)" - + if task['row_num'] not in summaries: summaries[task['row_num']] = "k.A. (Nicht im OpenAI-Batch)" # Füge k.A. für Tasks hinzu, die ungültigen Rohtext hatten for task in tasks_data: - if task['row_num'] not in summaries: - summaries[task['row_num']] = "k.A. (Ungültiger Rohtext)" - - + if task['row_num'] not in summaries: summaries[task['row_num']] = "k.A. (Ungültiger Rohtext)" return summaries + @retry_on_failure def scrape_website_details(url): """Extrahiert Title, Description, H1-H3 von einer Website.""" @@ -2316,29 +2267,25 @@ def _process_batch(sheet, batches, row_numbers): # Komplette Funktion process_website_batch (prüft jetzt Timestamp AT mit erzwungenem Debugging) # Komplette Funktion process_website_batch (MIT Batched Google Sheet Updates) +# Komplette Funktion process_website_batch (NEUE STRUKTUR - ECHTER BATCH WORKFLOW) def process_website_batch(sheet_handler, start_row_index_in_sheet, end_row_index_in_sheet): """ - Batch-Prozess für Website-Scraping mit gewünschtem Batch-Workflow: - 1. Sammle URLs für einen Verarbeitungs-Batch. - 2. Scrape Rohtexte parallel für diesen Batch. - 3. Sende gesammelte Rohtexte in einem Call an OpenAI. - 4. Schreibe alle Ergebnisse des Batches gebündelt ins Sheet. + Batch-Prozess für Website-Scraping mit echtem Batch-Workflow: + 1. Sammle Tasks. 2. Scrape Batch parallel. 3. Summarize Batch (OpenAI). 4. Update Sheet Batch. """ - debug_print(f"Starte Website-Scraping (Batch Workflow) für Zeilen {start_row_index_in_sheet} bis {end_row_index_in_sheet}...") + debug_print(f"Starte Website-Scraping (Echter Batch Workflow) für Zeilen {start_row_index_in_sheet} bis {end_row_index_in_sheet}...") # --- Konfiguration --- - PROCESSING_BATCH_SIZE = 15 # Wie viele Websites auf einmal holen/verarbeiten? - OPENAI_BATCH_SIZE_LIMIT = 8 # Wie viele Texte max. pro OpenAI Call (wegen Token Limits) - MAX_SCRAPING_WORKERS = 10 # Wie viele Threads für paralleles Scraping + PROCESSING_BATCH_SIZE = 20 # Wie viele Zeilen pro Verarbeitungs-Batch sammeln? + OPENAI_BATCH_SIZE_LIMIT = 8 # Max. Texte pro OpenAI Call + MAX_SCRAPING_WORKERS = 10 # Threads für Scraping + openai_semaphore = threading.Semaphore(5) # Semaphore für OpenAI Calls # --- Lade Initialdaten --- - if not sheet_handler.load_data(): - debug_print("FEHLER beim Laden der initialen Daten in process_website_batch.") - return + if not sheet_handler.load_data(): return all_data = sheet_handler.get_all_data_with_headers() - if not all_data or len(all_data) <= 5: - debug_print("FEHLER/WARNUNG: Keine Daten zum Verarbeiten gefunden.") - return + if not all_data or len(all_data) <= 5: return + header_rows = 5 # Header-Zeilen annehmen # --- Indizes und Spaltenbuchstaben --- timestamp_col_key = "Website Scrape Timestamp" @@ -2348,34 +2295,32 @@ def process_website_batch(sheet_handler, start_row_index_in_sheet, end_row_index summary_col_idx = COLUMN_MAP.get("Website Zusammenfassung") version_col_idx = COLUMN_MAP.get("Version") if None in [timestamp_col_index, website_col_idx, rohtext_col_idx, summary_col_idx, version_col_idx]: - debug_print(f"FEHLER: Mindestens ein benötigter Spaltenindex für process_website_batch fehlt in COLUMN_MAP.") + debug_print(f"FEHLER: Benötigte Indizes für process_website_batch fehlen.") return ts_col_letter = sheet_handler._get_col_letter(timestamp_col_index + 1) rohtext_col_letter = sheet_handler._get_col_letter(rohtext_col_idx + 1) summary_col_letter = sheet_handler._get_col_letter(summary_col_idx + 1) version_col_letter = sheet_handler._get_col_letter(version_col_idx + 1) - # --- Worker-Funktion nur für Scraping --- + # --- Worker-Funktion für Scraping --- def scrape_raw_text_task(task_info): row_num = task_info['row_num'] url = task_info['url'] raw_text = "k.A." error = None try: - # debug_print(f"Scraping Task Zeile {row_num}: Hole {url}...") # Kann sehr laut werden raw_text = get_website_raw(url) except Exception as e: - error = f"Fehler beim Scraping Zeile {row_num}: {e}" + error = f"Scraping Fehler Zeile {row_num}: {e}" debug_print(error) return {"row_num": row_num, "raw_text": raw_text, "error": error} - # --- Hauptverarbeitung --- - tasks_for_processing_batch = [] # Sammelt Tasks für einen kompletten Verarbeitungsbatch - all_sheet_updates = [] # Sammelt *alle* Sheet-Updates über mehrere Batches hinweg - processed_total_count = 0 - skipped_total_count = 0 - skipped_url_total_count = 0 - error_total_count = 0 + # --- Hauptlogik: Iteriere und sammle Batches --- + tasks_for_current_processing_batch = [] + total_processed_count = 0 + total_skipped_count = 0 + total_skipped_url_count = 0 + total_error_count = 0 for i in range(start_row_index_in_sheet, end_row_index_in_sheet + 1): row_index_in_list = i - 1 @@ -2388,78 +2333,73 @@ def process_website_batch(sheet_handler, start_row_index_in_sheet, end_row_index should_skip = True if should_skip: - skipped_total_count += 1 + total_skipped_count += 1 continue # Gültige URL Prüfung website_url = row[website_col_idx] if len(row) > website_col_idx else "" if not website_url or website_url.strip().lower() == "k.a.": - skipped_url_total_count += 1 + total_skipped_url_count += 1 continue - # Aufgabe zum aktuellen Verarbeitungsbatch hinzufügen - tasks_for_processing_batch.append({"row_num": i, "url": website_url}) + # Aufgabe zum Sammel-Batch hinzufügen + tasks_for_current_processing_batch.append({"row_num": i, "url": website_url}) - # --- Verarbeitungs-Batch ausführen, wenn voll oder letzte Zeile --- - if len(tasks_for_processing_batch) >= PROCESSING_BATCH_SIZE or i == end_row_index_in_sheet: - if tasks_for_processing_batch: - debug_print(f"\n--- Starte Verarbeitungs-Batch für {len(tasks_for_processing_batch)} Zeilen (Start: {tasks_for_processing_batch[0]['row_num']}, Ende: {tasks_for_processing_batch[-1]['row_num']}) ---") + # --- Batch verarbeiten, wenn voll oder letzte Zeile des Gesamtbereichs --- + if len(tasks_for_current_processing_batch) >= PROCESSING_BATCH_SIZE or i == end_row_index_in_sheet: + if tasks_for_current_processing_batch: + batch_start_row = tasks_for_current_processing_batch[0]['row_num'] + batch_end_row = tasks_for_current_processing_batch[-1]['row_num'] + batch_task_count = len(tasks_for_current_processing_batch) + debug_print(f"\n--- Starte Verarbeitungs-Batch ({batch_task_count} Tasks, Zeilen {batch_start_row}-{batch_end_row}) ---") - # --- Schritt 2: Rohtexte parallel scrapen --- - scraping_results_dict = {} # {row_num: {'raw_text': '...', 'error': None}, ...} - debug_print(f" Scrape {len(tasks_for_processing_batch)} Websites parallel (max {MAX_SCRAPING_WORKERS} worker)...") + # --- Phase 1: Paralleles Scraping --- + scraping_results_dict = {} + debug_print(f" 1. Scrape {batch_task_count} Websites parallel (max {MAX_SCRAPING_WORKERS} worker)...") with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_SCRAPING_WORKERS) as executor: - future_to_task = {executor.submit(scrape_raw_text_task, task): task for task in tasks_for_processing_batch} + future_to_task = {executor.submit(scrape_raw_text_task, task): task for task in tasks_for_current_processing_batch} for future in concurrent.futures.as_completed(future_to_task): task = future_to_task[future] - try: - result = future.result() - scraping_results_dict[result['row_num']] = {"raw_text": result['raw_text'], "error": result['error']} + try: result = future.result(); scraping_results_dict[result['row_num']] = result except Exception as exc: - row_num = task['row_num'] - err_msg = f"Generischer Fehler in Scraping Task für Zeile {row_num}: {exc}" - debug_print(err_msg) - scraping_results_dict[row_num] = {"raw_text": "k.A.", "error": err_msg} - error_total_count +=1 - + row_num = task['row_num']; err_msg = f"Generischer Fehler Scraping Task Zeile {row_num}: {exc}" + debug_print(err_msg); scraping_results_dict[row_num] = {"raw_text": "k.A.", "error": err_msg}; total_error_count +=1 debug_print(f" Scraping für Batch beendet.") - # --- Schritt 3: OpenAI Batch(es) vorbereiten und ausführen --- + # --- Phase 2: OpenAI Batch Summarization --- openai_tasks = [] - tasks_without_valid_text = [] - for task_info in tasks_for_processing_batch: + for task_info in tasks_for_current_processing_batch: row_num = task_info['row_num'] scrape_result = scraping_results_dict.get(row_num) - if scrape_result and not scrape_result.get('error') and scrape_result.get('raw_text', 'k.A.') not in ['k.A.', 'k.A. (Nur Cookie-Banner erkannt)'] and str(scrape_result.get('raw_text')).strip(): + # Nur Tasks mit erfolgreichem Scraping und gültigem Text an OpenAI schicken + if scrape_result and not scrape_result.get('error') and \ + scrape_result.get('raw_text', 'k.A.') not in ['k.A.', 'k.A. (Nur Cookie-Banner erkannt)'] and \ + str(scrape_result.get('raw_text')).strip(): openai_tasks.append({'row_num': row_num, 'raw_text': scrape_result['raw_text']}) - else: - # Speichere Zeilen ohne gültigen Text für späteres Update - tasks_without_valid_text.append({'row_num': row_num, 'raw_text': scrape_result.get('raw_text', 'k.A.') if scrape_result else 'k.A. (Scraping fehlgeschlagen)'}) - - all_summaries = {} # Sammelt Ergebnisse aller OpenAI Batches für diesen Verarbeitungsbatch + all_summaries = {} openai_task_batches = [openai_tasks[j:j + OPENAI_BATCH_SIZE_LIMIT] for j in range(0, len(openai_tasks), OPENAI_BATCH_SIZE_LIMIT)] - if openai_task_batches: - debug_print(f" Bereite {len(openai_task_batches)} OpenAI Batch(es) vor (Größe max. {OPENAI_BATCH_SIZE_LIMIT})...") + debug_print(f" 2. Bereite {len(openai_task_batches)} OpenAI Batch(es) vor (Größe max. {OPENAI_BATCH_SIZE_LIMIT})...") for num, openai_batch in enumerate(openai_task_batches): - debug_print(f" Verarbeite OpenAI Batch {num+1}/{len(openai_task_batches)}...") - summaries_result = summarize_batch_openai(openai_batch) + debug_print(f" Verarbeite OpenAI Batch {num+1}/{len(openai_task_batches)} für Zeilen: {[t['row_num'] for t in openai_batch]}") + summaries_result = summarize_batch_openai(openai_batch) # Enthält Fallback für nicht geparste all_summaries.update(summaries_result) - time.sleep(1) # Kurze Pause zwischen OpenAI Batches + if len(openai_task_batches) > 1: time.sleep(1) # Pause nur wenn mehrere OpenAI Batches nötig debug_print(f" OpenAI Batch-Verarbeitung beendet.") else: debug_print(" Keine gültigen Rohtexte für OpenAI Batch-Verarbeitung vorhanden.") - - # --- Schritt 4: Sheet Updates für den kompletten Verarbeitungs-Batch vorbereiten --- - current_timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + # --- Phase 3: Sheet Updates für den Batch vorbereiten & senden --- + batch_sheet_updates = [] + current_timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") # Ein Zeitstempel für den Batch current_version = Config.VERSION - batch_sheet_updates = [] # Updates nur für diesen Verarbeitungsbatch + processed_in_batch_count = 0 - for task_info in tasks_for_processing_batch: + for task_info in tasks_for_current_processing_batch: # Iteriere über die ursprünglichen Tasks des Batches row_num = task_info['row_num'] - raw_text_res = scraping_results_dict.get(row_num, {"raw_text": "k.A. (Fehler)"})["raw_text"] + raw_text_res = scraping_results_dict.get(row_num, {}).get("raw_text", "k.A. (Scraping Fehler)") + # Hole Summary; falls Task nicht in OpenAI Batch war oder Fehler hatte, gibt summarize_batch_openai 'k.A.' zurück summary_res = all_summaries.get(row_num, "k.A. (Keine Zusammenf.)") row_updates = [ @@ -2469,37 +2409,27 @@ def process_website_batch(sheet_handler, start_row_index_in_sheet, end_row_index {'range': f'{version_col_letter}{row_num}', 'values': [[current_version]]} # AP Version ] batch_sheet_updates.extend(row_updates) - processed_total_count += 1 # Zähle alle Zeilen, die diesen Punkt erreichen + processed_in_batch_count += 1 # Zähle alle Zeilen, die im Batch versucht wurden - # Füge Updates dieses Verarbeitungs-Batches zur Gesamtliste hinzu - all_sheet_updates.extend(batch_sheet_updates) - debug_print(f" {len(batch_sheet_updates)} Sheet-Updates für diesen Batch vorbereitet.") - - # Sheet-Updates gebündelt senden (optional, könnte auch am Ende erfolgen) - # Hier: Senden nach jedem Verarbeitungs-Batch (Kompromiss) - if all_sheet_updates: - debug_print(f" Sende {len(all_sheet_updates)} Sheet-Updates für abgeschlossenen Batch...") - success = sheet_handler.batch_update_cells(all_sheet_updates) + if batch_sheet_updates: + debug_print(f" 3. Sende Sheet-Update für {processed_in_batch_count} Zeilen des Batches...") + success = sheet_handler.batch_update_cells(batch_sheet_updates) if success: - debug_print(f" Sheet-Update für Batch bis Zeile {i} erfolgreich.") + debug_print(f" Sheet-Update für Batch {batch_start_row}-{batch_end_row} erfolgreich.") else: - debug_print(f" FEHLER beim Sheet-Update für Batch bis Zeile {i}.") - all_sheet_updates = [] # Liste für den nächsten großen Sheet-Update-Batch leeren + debug_print(f" FEHLER beim Sheet-Update für Batch {batch_start_row}-{batch_end_row}.") + else: + debug_print(f" Keine Sheet-Updates für Batch {batch_start_row}-{batch_end_row} vorbereitet.") - # Verarbeitungs-Batch leeren - tasks_for_processing_batch = [] - # Kurze Pause nach Verarbeitung eines kompletten Batches - time.sleep(2) + # Verarbeitungs-Batch leeren für den nächsten Durchlauf + tasks_for_current_processing_batch = [] + total_processed_count += processed_in_batch_count # Addiere zur Gesamtzahl + debug_print(f"--- Verarbeitungs-Batch {batch_start_row}-{batch_end_row} abgeschlossen ---") + # Keine Pause hier, da die Verarbeitungsschritte Zeit brauchen - # --- Ende der if tasks_for_processing_batch --- - # --- Ende der for-Schleife über alle Zeilen --- + # Kein separater letzter Batch-Send mehr nötig, da nach jedem Batch gesendet wird - # Ggf. letzten, nicht gesendeten Sheet-Update Batch senden (sollte nicht nötig sein, wenn nach jedem Batch gesendet wird) - # if all_sheet_updates: - # debug_print(f"Sende LETZTES Sheet-Update für {len(all_sheet_updates)} Zellen...") - # sheet_handler.batch_update_cells(all_sheet_updates) - - debug_print(f"Website-Scraping (Batch Workflow) abgeschlossen. {processed_total_count} Zeilen verarbeitet (inkl. Fehler), {error_total_count} Fehler aufgetreten, {skipped_count} Zeilen wg. Timestamp übersprungen, {skipped_url_count} Zeilen ohne URL übersprungen.") + debug_print(f"Website-Scraping (Echter Batch Workflow) abgeschlossen. {total_processed_count} Zeilen verarbeitet (inkl. Fehler), {total_error_count} explizite Fehler, {total_skipped_count} Zeilen wg. Timestamp übersprungen, {total_skipped_url_count} Zeilen ohne URL übersprungen.") # Komplette Funktion process_branch_batch (prüft jetzt Timestamp AO mit erzwungenem Debugging) # Komplette Funktion process_branch_batch (MIT Korrektur und Prüfung auf AO)