654 lines
No EOL
27 KiB
Python
654 lines
No EOL
27 KiB
Python
"""
|
|
Webcrawler-Agent für Recherche und Abruf von Informationen aus dem Web.
|
|
Angepasst für die neue chat.py Architektur und chat_registry.py.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import time
|
|
import traceback
|
|
from typing import Dict, Any, List, Optional
|
|
from urllib.parse import quote_plus, unquote
|
|
|
|
from bs4 import BeautifulSoup
|
|
import requests
|
|
from modules.chat_registry import AgentBase
|
|
from modules.utility import APP_CONFIG
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class AgentWebcrawler(AgentBase):
|
|
"""Agent für Webrecherche und Informationsabruf"""
|
|
|
|
def __init__(self):
|
|
"""Initialisiert den Webcrawler-Agent"""
|
|
super().__init__()
|
|
self.name = "Webscraper"
|
|
self.capabilities = "web_search,information_retrieval,data_collection,source_verification,content_integration"
|
|
self.result_format = "SearchResults"
|
|
|
|
# Web-Crawling-Konfiguration
|
|
self.max_url = int(APP_CONFIG.get("Connector_AiWebscraping_MAX_URLS"))
|
|
self.max_key = int(APP_CONFIG.get("Connector_AiWebscraping_MAX_SEARCH_KEYWORDS"))
|
|
self.max_result = int(APP_CONFIG.get("Connector_AiWebscraping_MAX_SEARCH_RESULTS"))
|
|
self.timeout = int(APP_CONFIG.get("Connector_AiWebscraping_TIMEOUT"))
|
|
|
|
def get_agent_info(self) -> Dict[str, Any]:
|
|
"""Gibt Agent-Informationen für die Registry zurück"""
|
|
info = super().get_config()
|
|
info.update({
|
|
"metadata": {
|
|
"max_url": self.max_url,
|
|
"max_result": self.max_result,
|
|
"timeout": self.timeout
|
|
}
|
|
})
|
|
return info
|
|
|
|
async def process_message(self, message: Dict[str, Any], context: Dict[str, Any] = None) -> Dict[str, Any]:
|
|
"""
|
|
Verarbeitet eine Nachricht und führt bei Bedarf eine Webrecherche durch.
|
|
|
|
Args:
|
|
message: Die zu verarbeitende Nachricht
|
|
context: Zusätzlicher Kontext
|
|
|
|
Returns:
|
|
Die generierte Antwort oder Ablehnung, wenn keine Webrecherche erforderlich ist
|
|
"""
|
|
# Workflow-ID aus Kontext oder Nachricht extrahieren
|
|
workflow_id = context.get("workflow_id") if context else message.get("workflow_id", "unknown")
|
|
|
|
# Antwortstruktur erstellen
|
|
response = {
|
|
"role": "assistant",
|
|
"content": "",
|
|
"agent_name": self.name,
|
|
"result_format": self.result_format,
|
|
"workflow_id": workflow_id
|
|
}
|
|
|
|
try:
|
|
# Abfrage aus der Nachricht abrufen
|
|
prompt = message.get("content", "").strip()
|
|
|
|
# Prüfen, ob es sich explizit um eine Webrecherche-Anfrage handelt
|
|
is_web_research = await self._is_web_research_request(prompt)
|
|
|
|
if not is_web_research:
|
|
# Keine Webrecherche-Anfrage ablehnen
|
|
logger.info("Anfrage abgelehnt: keine Webrecherche-Aufgabe")
|
|
response["content"] = "Diese Anfrage scheint keine Webrecherche zu erfordern. Weiterleitung an einen passenderen Agenten."
|
|
response["status"] = "rejected"
|
|
return response
|
|
|
|
# Mit Webrecherche fortfahren
|
|
logger.info(f"Webrecherche für: {prompt[:50]}...")
|
|
|
|
# Suchstrategie vorbereiten
|
|
logger.info("Erstelle Suchstrategie")
|
|
|
|
search_strategy = await self._create_search_strategy(prompt)
|
|
search_keys = search_strategy.get("skey", [])
|
|
search_urls = search_strategy.get("url", [])
|
|
|
|
if search_keys:
|
|
logger.info(f"Suche nach {len(search_keys)} Schlüsselbegriffen: {', '.join(search_keys[:2])}...")
|
|
|
|
if search_urls:
|
|
logger.info(f"Suche in {len(search_urls)} direkten URLs: {', '.join(search_urls[:2])}...")
|
|
|
|
# Suche ausführen
|
|
results = []
|
|
|
|
# Suchbegriffe verarbeiten
|
|
for keyword in search_keys:
|
|
logger.info(f"Suche im Web nach: '{keyword}'")
|
|
keyword_results = self._search_web(keyword)
|
|
results.extend(keyword_results)
|
|
logger.info(f"Gefunden: {len(keyword_results)} Ergebnisse für '{keyword}'")
|
|
|
|
# Direkte URLs verarbeiten
|
|
for url in search_urls:
|
|
logger.info(f"Extrahiere Inhalt von: {url}")
|
|
soup = self._read_url(url)
|
|
|
|
# Titel aus der Seite extrahieren, falls vorhanden
|
|
title = self._extract_title(soup, url)
|
|
|
|
result = self._parse_result(soup, title, url)
|
|
results.append(result)
|
|
logger.info(f"Extrahiert: '{title}' von {url}")
|
|
|
|
# Ergebnisse für die endgültige Ausgabe verarbeiten
|
|
logger.info(f"Analysiere {len(results)} Web-Ergebnisse")
|
|
|
|
# Zusammenfassungen für jedes Ergebnis generieren
|
|
processed_results = []
|
|
for i, result in enumerate(results):
|
|
result_data_limited = self._limit_text(result['data'], max_chars=10000)
|
|
|
|
logger.info(f"Analysiere Ergebnis {i+1}/{len(results)}: {result['title'][:30]}...")
|
|
|
|
content_summary = await self._summarize_result(result_data_limited, prompt)
|
|
|
|
processed_result = {
|
|
"title": result['title'],
|
|
"url": result['url'],
|
|
"snippet": result['snippet'],
|
|
"summary": content_summary
|
|
}
|
|
|
|
processed_results.append(processed_result)
|
|
|
|
# Gesamtzusammenfassung erstellen
|
|
all_summaries = "\n\n".join([r["summary"] for r in processed_results])
|
|
all_summaries_limited = self._limit_text(all_summaries, max_chars=10000)
|
|
|
|
logger.info("Erstelle Gesamtzusammenfassung der Webrecherche")
|
|
|
|
final_summary = await self.ai_service.call_api([
|
|
{"role": "system", "content": "Du erstellst prägnante Zusammenfassungen von Rechercheergebnissen."},
|
|
{"role": "user", "content": f"Bitte fasse diese Erkenntnisse in 5-6 Sätzen zusammen: {all_summaries_limited}\n"}
|
|
])
|
|
|
|
# Sprache der Anfrage ermitteln, um Überschriften in der richtigen Sprache zu verwenden
|
|
headers = await self._get_localized_headers(prompt)
|
|
|
|
# Endgültiges Ergebnis formatieren
|
|
final_result = f"## {headers['web_research_results']}\n\n### {headers['summary']}\n{final_summary}\n\n### {headers['detailed_results']}\n"
|
|
|
|
for i, result in enumerate(processed_results, 1):
|
|
final_result += f"\n\n[{i}] {result['title']}\n{headers['url']}: {result['url']}\n{headers['snippet']}: {result['snippet']}\n{headers['content']}: {result['summary']}"
|
|
|
|
# Inhalt in der Antwort setzen
|
|
response["content"] = final_result
|
|
|
|
logger.info("Webrecherche erfolgreich abgeschlossen")
|
|
|
|
return response
|
|
|
|
except Exception as e:
|
|
error_msg = f"Fehler bei der Webrecherche: {str(e)}"
|
|
logger.error(error_msg)
|
|
response["content"] = f"## Fehler bei der Webrecherche\n\n{error_msg}"
|
|
return response
|
|
|
|
async def _is_web_research_request(self, prompt: str) -> bool:
|
|
"""
|
|
Verwendet KI, um festzustellen, ob eine Anfrage Webrecherche erfordert.
|
|
|
|
Args:
|
|
prompt: Die Benutzeranfrage
|
|
|
|
Returns:
|
|
True, wenn es explizit eine Webrecherche-Anfrage ist, sonst False
|
|
"""
|
|
if not self.ai_service:
|
|
# Fallback zur einfacheren Erkennung, wenn kein KI-Service verfügbar ist
|
|
return self._simple_web_detection(prompt)
|
|
|
|
try:
|
|
# Prompt erstellen, um zu analysieren, ob es sich um eine Webrecherche-Anfrage handelt
|
|
analysis_prompt = f"""
|
|
Analysiere die folgende Anfrage und bestimme, ob sie explizit eine Webrecherche oder Online-Informationen erfordert.
|
|
|
|
ANFRAGE: {prompt}
|
|
|
|
Eine Anfrage erfordert Webrecherche, wenn:
|
|
1. Sie explizit nach der Suche von Informationen online fragt
|
|
2. Sie URLs oder Verweise auf Websites enthält
|
|
3. Sie aktuelle Informationen anfordert, die im Web verfügbar wären
|
|
4. Sie nach Informationen aus Web-Quellen fragt
|
|
5. Sie implizit aktuelle Informationen aus dem Internet erfordert
|
|
|
|
Antworte NUR mit einem einzelnen Wort - entweder "JA", wenn Webrecherche erforderlich ist, oder "NEIN", wenn nicht.
|
|
Füge KEINE Erklärung hinzu, nur die Antwort JA oder NEIN.
|
|
"""
|
|
|
|
# KI zur Analyse aufrufen
|
|
response = await self.ai_service.call_api([
|
|
{"role": "system", "content": "Du bestimmst, ob eine Anfrage Webrecherche erfordert. Antworte immer nur mit JA oder NEIN."},
|
|
{"role": "user", "content": analysis_prompt}
|
|
])
|
|
|
|
# Antwort bereinigen und überprüfen
|
|
response = response.strip().upper()
|
|
|
|
return "JA" in response
|
|
|
|
except Exception as e:
|
|
# Fehler protokollieren, aber nicht fehlschlagen, Fallback zur einfacheren Erkennung
|
|
logger.warning(f"Fehler bei der KI-Erkennung von Webrecherche-Anfragen: {str(e)}")
|
|
return self._simple_web_detection(prompt)
|
|
|
|
def _simple_web_detection(self, prompt: str) -> bool:
|
|
"""
|
|
Einfachere Fallback-Methode zur Erkennung von Webrecherche-Anfragen anhand von URLs.
|
|
|
|
Args:
|
|
prompt: Die Benutzeranfrage
|
|
|
|
Returns:
|
|
True, wenn es klare URL-Indikatoren gibt, sonst False
|
|
"""
|
|
# URLs in der Anfrage deuten stark auf Webrecherche hin
|
|
url_indicators = ["http://", "https://", "www.", ".com", ".org", ".net", ".edu", ".gov"]
|
|
web_terms = ["search", "find online", "look up", "web", "internet", "website", "suche", "finde", "recherchiere"]
|
|
|
|
# Auf URL-Muster in der Anfrage prüfen
|
|
contains_url = any(indicator in prompt.lower() for indicator in url_indicators)
|
|
contains_web_term = any(term in prompt.lower() for term in web_terms)
|
|
|
|
return contains_url or contains_web_term
|
|
|
|
async def _create_search_strategy(self, prompt: str) -> Dict[str, List[str]]:
|
|
"""
|
|
Erstellt eine Suchstrategie basierend auf der Anfrage.
|
|
|
|
Args:
|
|
prompt: Die Benutzeranfrage
|
|
|
|
Returns:
|
|
Suchstrategie mit URLs und Suchbegriffen
|
|
"""
|
|
if not self.ai_service:
|
|
# Fallback zur einfachen Strategie
|
|
return {"skey": [prompt], "url": []}
|
|
|
|
try:
|
|
# KI-Prompt zur Erstellung einer Suchstrategie
|
|
strategy_prompt = f"""Erstelle eine umfassende Webrecherchestrategie für die Aufgabe = '{prompt.replace("'","")}'. Gib die Ergebnisse als Python-Dictionary mit diesen spezifischen Schlüsseln zurück. Wenn bestimmte URLs angegeben sind und die Aufgabe nur die Analyse dieser URLs erfordert, lass 'skey' leer.
|
|
|
|
'url': Eine Liste von maximal {self.max_url} spezifischen URLs, die aus der Aufgabenstellung extrahiert wurden.
|
|
|
|
'skey': Eine Liste von maximal {self.max_key} Schlüsselsätzen, nach denen im Web gesucht werden soll. Diese sollten präzise, vielfältig und gezielt sein, um die relevantesten Informationen zu erhalten.
|
|
|
|
Formatiere deine Antwort als gültiges JSON-Objekt mit diesen beiden Schlüsseln. Füge keinen erklärenden Text oder Markdown außerhalb der Objektdefinition hinzu.
|
|
"""
|
|
|
|
# KI für Suchstrategie aufrufen
|
|
content_text = await self.ai_service.call_api([
|
|
{"role": "system", "content": "Du bist ein Webrecherche-Experte, der präzise Suchstrategien entwickelt."},
|
|
{"role": "user", "content": strategy_prompt}
|
|
])
|
|
|
|
# JSON-Code-Block-Markierungen entfernen, falls vorhanden
|
|
if content_text.startswith("```json"):
|
|
end_marker = "```"
|
|
end_index = content_text.rfind(end_marker)
|
|
if end_index != -1:
|
|
content_text = content_text[7:end_index].strip()
|
|
|
|
# JSON parsen und zurückgeben
|
|
strategy = json.loads(content_text)
|
|
return strategy
|
|
|
|
except Exception as e:
|
|
logger.error(f"Fehler bei der Erstellung der Suchstrategie: {str(e)}")
|
|
# Einfache Fallback-Strategie
|
|
return {"skey": [prompt], "url": []}
|
|
|
|
async def _summarize_result(self, result_data: str, original_prompt: str) -> str:
|
|
"""
|
|
Erstellt eine Zusammenfassung eines Suchergebnisses mit KI.
|
|
|
|
Args:
|
|
result_data: Die zu zusammenfassenden Daten
|
|
original_prompt: Die ursprüngliche Anfrage
|
|
|
|
Returns:
|
|
Zusammenfassung des Ergebnisses
|
|
"""
|
|
if not self.ai_service:
|
|
return "Keine Zusammenfassung verfügbar (KI-Service nicht verfügbar)"
|
|
|
|
try:
|
|
# Anweisungen für die Zusammenfassung
|
|
summary_prompt = f"""
|
|
Fasse dieses Suchergebnis gemäß der ursprünglichen Anfrage in etwa 2000 Zeichen zusammen. Ursprüngliche Anfrage = '{original_prompt.replace("'","")}'
|
|
Konzentriere dich auf die wichtigsten Erkenntnisse und verbinde sie mit der ursprünglichen Anfrage. Du kannst jede Einleitung überspringen.
|
|
Extrahiere nur relevante und hochwertige Informationen im Zusammenhang mit der Anfrage und präsentiere sie in einem klaren Format. Biete eine ausgewogene Ansicht der recherchierten Informationen.
|
|
|
|
Hier ist das Suchergebnis:
|
|
{result_data}
|
|
"""
|
|
|
|
# KI für Zusammenfassung aufrufen
|
|
summary = await self.ai_service.call_api([
|
|
{"role": "system", "content": "Du bist ein Informationsanalyst, der Webinhalte präzise und relevant zusammenfasst."},
|
|
{"role": "user", "content": summary_prompt}
|
|
])
|
|
|
|
# Auf ~2000 Zeichen begrenzen
|
|
return summary[:2000]
|
|
|
|
except Exception as e:
|
|
logger.error(f"Fehler bei der Zusammenfassung des Ergebnisses: {str(e)}")
|
|
return "Fehler bei der Zusammenfassung"
|
|
|
|
async def _get_localized_headers(self, text: str) -> Dict[str, str]:
|
|
"""
|
|
Ermittelt lokalisierte Überschriften für die Webrecherche-Ergebnisse basierend auf der erkannten Sprache.
|
|
|
|
Args:
|
|
text: Text zur Spracherkennung
|
|
|
|
Returns:
|
|
Dictionary mit lokalisierten Überschriften
|
|
"""
|
|
# Standard-Englische Überschriften
|
|
headers = {
|
|
"web_research_results": "Web Research Results",
|
|
"summary": "Summary",
|
|
"detailed_results": "Detailed Results",
|
|
"url": "URL",
|
|
"snippet": "Snippet",
|
|
"content": "Content"
|
|
}
|
|
|
|
if not self.ai_service:
|
|
return headers
|
|
|
|
try:
|
|
# Sprache erkennen
|
|
language_prompt = f"In welcher Sprache ist dieser Text geschrieben? Antworte nur mit dem Sprachnamen: {text[:200]}"
|
|
language = await self.ai_service.call_api([
|
|
{"role": "system", "content": "Du bestimmst die Sprache eines Textes und gibst nur den Sprachnamen zurück."},
|
|
{"role": "user", "content": language_prompt}
|
|
])
|
|
|
|
language = language.strip().lower()
|
|
|
|
# Englische Sprache oder Spracherkennung fehlgeschlagen, Standardüberschriften zurückgeben
|
|
if language in ["english", "en", ""]:
|
|
return headers
|
|
|
|
# Deutsche Überschriften
|
|
if language in ["deutsch", "german", "de"]:
|
|
return {
|
|
"web_research_results": "Webrecherche-Ergebnisse",
|
|
"summary": "Zusammenfassung",
|
|
"detailed_results": "Detaillierte Ergebnisse",
|
|
"url": "URL",
|
|
"snippet": "Ausschnitt",
|
|
"content": "Inhalt"
|
|
}
|
|
|
|
# Französische Überschriften
|
|
if language in ["französisch", "french", "fr"]:
|
|
return {
|
|
"web_research_results": "Résultats de recherche Web",
|
|
"summary": "Résumé",
|
|
"detailed_results": "Résultats détaillés",
|
|
"url": "URL",
|
|
"snippet": "Extrait",
|
|
"content": "Contenu"
|
|
}
|
|
|
|
# Überschriften übersetzen, wenn Sprache erkannt, aber keine vordefinierte Übersetzung
|
|
translation_prompt = f"""
|
|
Übersetze diese Webrecherche-Ergebnisüberschriften ins {language}:
|
|
|
|
Web Research Results
|
|
Summary
|
|
Detailed Results
|
|
URL
|
|
Snippet
|
|
Content
|
|
|
|
Gib ein JSON-Objekt mit diesen Schlüsseln zurück:
|
|
web_research_results, summary, detailed_results, url, snippet, content
|
|
"""
|
|
|
|
# KI für Übersetzung aufrufen
|
|
response = await self.ai_service.call_api([
|
|
{"role": "system", "content": "Du übersetzt Überschriften in die angegebene Sprache und gibst sie als JSON zurück."},
|
|
{"role": "user", "content": translation_prompt}
|
|
])
|
|
|
|
# JSON extrahieren
|
|
import re
|
|
json_match = re.search(r'\{.*\}', response, re.DOTALL)
|
|
|
|
if json_match:
|
|
translated_headers = json.loads(json_match.group(0))
|
|
return translated_headers
|
|
|
|
except Exception as e:
|
|
# Fehler protokollieren, aber mit englischen Überschriften fortfahren
|
|
logger.warning(f"Fehler beim Übersetzen der Überschriften: {str(e)}")
|
|
|
|
return headers
|
|
|
|
def _search_web(self, query: str) -> List[Dict[str, str]]:
|
|
"""
|
|
Führt eine Websuche durch und gibt die Ergebnisse zurück.
|
|
|
|
Args:
|
|
query: Die Suchanfrage
|
|
|
|
Returns:
|
|
Liste von Suchergebnissen
|
|
"""
|
|
formatted_query = quote_plus(query)
|
|
url = f"{APP_CONFIG.get('Connector_AiWebscraping_SEARCH_ENGINE')}{formatted_query}"
|
|
|
|
search_results_soup = self._read_url(url)
|
|
if not isinstance(search_results_soup, BeautifulSoup) or not search_results_soup.select('.result'):
|
|
logger.warning(f"Keine Suchergebnisse gefunden für: {query}")
|
|
return []
|
|
|
|
# Suchergebnisse extrahieren
|
|
results = []
|
|
|
|
# Alle Ergebniscontainer finden
|
|
result_elements = search_results_soup.select('.result')
|
|
|
|
for result in result_elements:
|
|
# Titel extrahieren
|
|
title_element = result.select_one('.result__a')
|
|
title = title_element.text.strip() if title_element else 'Kein Titel'
|
|
|
|
# URL extrahieren (DuckDuckGo verwendet Weiterleitungen)
|
|
url_element = title_element.get('href') if title_element else ''
|
|
extracted_url = 'Keine URL'
|
|
|
|
if url_element:
|
|
# Tatsächliche URL aus DuckDuckGos Weiterleitung extrahieren
|
|
if url_element.startswith('/d.js?q='):
|
|
start = url_element.find('?q=') + 3
|
|
end = url_element.find('&', start) if '&' in url_element[start:] else None
|
|
extracted_url = unquote(url_element[start:end])
|
|
|
|
# Sicherstellen, dass die URL das korrekte Protokollpräfix hat
|
|
if not extracted_url.startswith(('http://', 'https://')):
|
|
if not extracted_url.startswith('//'):
|
|
extracted_url = 'https://' + extracted_url
|
|
else:
|
|
extracted_url = 'https:' + extracted_url
|
|
else:
|
|
extracted_url = url_element
|
|
|
|
# Snippet direkt aus der Suchergebnisseite extrahieren
|
|
snippet_element = result.select_one('.result__snippet')
|
|
snippet = snippet_element.text.strip() if snippet_element else 'Keine Beschreibung'
|
|
|
|
# Tatsächlichen Seiteninhalt für das Datenfeld abrufen
|
|
target_page_soup = self._read_url(extracted_url)
|
|
|
|
# Neue Inhaltsextraktionsmethode verwenden, um Inhaltsgröße zu begrenzen
|
|
content = self._extract_main_content(target_page_soup)
|
|
|
|
results.append({
|
|
'title': title,
|
|
'url': extracted_url,
|
|
'snippet': snippet,
|
|
'data': content
|
|
})
|
|
|
|
# Anzahl der Ergebnisse bei Bedarf begrenzen
|
|
if len(results) >= self.max_result:
|
|
break
|
|
|
|
return results
|
|
|
|
def _read_url(self, url: str) -> BeautifulSoup:
|
|
"""
|
|
Liest eine URL und gibt einen BeautifulSoup-Parser für den Inhalt zurück.
|
|
|
|
Args:
|
|
url: Die zu lesende URL
|
|
|
|
Returns:
|
|
BeautifulSoup-Objekt mit dem Inhalt oder leer bei Fehlern
|
|
"""
|
|
headers = {
|
|
'User-Agent': APP_CONFIG.get("Connector_AiWebscraping_USER_AGENT"),
|
|
'Accept': 'text/html,application/xhtml+xml,application/xml',
|
|
'Accept-Language': 'en-US,en;q=0.9',
|
|
}
|
|
|
|
try:
|
|
# Initiale Anfrage
|
|
response = requests.get(url, headers=headers, timeout=self.timeout)
|
|
|
|
# Abfragen für Status 202
|
|
if response.status_code == 202:
|
|
# Maximal 3 Versuche mit zunehmenden Intervallen
|
|
backoff_times = [0.5, 1.0, 2.0, 5.0]
|
|
|
|
for wait_time in backoff_times:
|
|
time.sleep(wait_time) # Mit zunehmender Zeit warten
|
|
response = requests.get(url, headers=headers, timeout=self.timeout)
|
|
|
|
# Wenn kein 202 mehr, dann abbrechen
|
|
if response.status_code != 202:
|
|
break
|
|
|
|
# Für andere Fehlerstatuscodes einen Fehler auslösen
|
|
response.raise_for_status()
|
|
|
|
# HTML parsen
|
|
return BeautifulSoup(response.text, 'html.parser')
|
|
|
|
except Exception as e:
|
|
logger.error(f"Fehler beim Lesen der URL {url}: {str(e)}")
|
|
# Leeres BeautifulSoup-Objekt erstellen
|
|
return BeautifulSoup("<html><body></body></html>", 'html.parser')
|
|
|
|
def _extract_title(self, soup: BeautifulSoup, url: str) -> str:
|
|
"""
|
|
Extrahiert den Titel aus einer Webseite.
|
|
|
|
Args:
|
|
soup: BeautifulSoup-Objekt der Webseite
|
|
url: URL der Webseite
|
|
|
|
Returns:
|
|
Extrahierter Titel
|
|
"""
|
|
if not isinstance(soup, BeautifulSoup):
|
|
return f"Fehler bei {url}"
|
|
|
|
# Titel aus dem title-Tag extrahieren
|
|
title_tag = soup.find('title')
|
|
title = title_tag.text.strip() if title_tag else "Kein Titel"
|
|
|
|
# Alternative: Auch nach h1-Tags suchen, wenn der title-Tag fehlt
|
|
if title == "Kein Titel":
|
|
h1_tag = soup.find('h1')
|
|
if h1_tag:
|
|
title = h1_tag.text.strip()
|
|
|
|
return title
|
|
|
|
def _extract_main_content(self, soup: BeautifulSoup, max_chars: int = 10000) -> str:
|
|
"""
|
|
Extrahiert den Hauptinhalt aus einer HTML-Seite.
|
|
|
|
Args:
|
|
soup: BeautifulSoup-Objekt der Webseite
|
|
max_chars: Maximale Anzahl von Zeichen
|
|
|
|
Returns:
|
|
Extrahierter Hauptinhalt als String
|
|
"""
|
|
if not isinstance(soup, BeautifulSoup):
|
|
return str(soup)[:max_chars] if soup else ""
|
|
|
|
# Versuchen, Hauptinhaltselemente in Prioritätsreihenfolge zu finden
|
|
main_content = None
|
|
for selector in ['main', 'article', '#content', '.content', '#main', '.main']:
|
|
content = soup.select_one(selector)
|
|
if content:
|
|
main_content = content
|
|
break
|
|
|
|
# Wenn kein Hauptinhalt gefunden wurde, den Body verwenden
|
|
if not main_content:
|
|
main_content = soup.find('body') or soup
|
|
|
|
# Skript-, Style-, Nav-, Footer-Elemente entfernen, die nicht zum Hauptinhalt beitragen
|
|
for element in main_content.select('script, style, nav, footer, header, aside, .sidebar, #sidebar, .comments, #comments, .advertisement, .ads, iframe'):
|
|
element.extract()
|
|
|
|
# Textinhalt extrahieren
|
|
text_content = main_content.get_text(separator=' ', strip=True)
|
|
|
|
# Auf max_chars begrenzen
|
|
return text_content[:max_chars]
|
|
|
|
def _parse_result(self, soup: BeautifulSoup, title: str, url: str) -> Dict[str, str]:
|
|
"""
|
|
Parst ein BeautifulSoup-Objekt in ein Ergebnis-Dictionary.
|
|
|
|
Args:
|
|
soup: BeautifulSoup-Objekt der Webseite
|
|
title: Seitentitel
|
|
url: Seiten-URL
|
|
|
|
Returns:
|
|
Dictionary mit Ergebnisdaten
|
|
"""
|
|
# Inhalt extrahieren
|
|
content = self._extract_main_content(soup)
|
|
|
|
result = {
|
|
'title': title,
|
|
'url': url,
|
|
'snippet': 'Keine Beschreibung', # Standardwert
|
|
'data': content
|
|
}
|
|
return result
|
|
|
|
def _limit_text(self, text: str, max_chars: int = 10000) -> str:
|
|
"""
|
|
Begrenzt den Text auf eine maximale Anzahl von Zeichen.
|
|
|
|
Args:
|
|
text: Eingangstext
|
|
max_chars: Maximale Anzahl von Zeichen
|
|
|
|
Returns:
|
|
Begrenzter Text
|
|
"""
|
|
if not text:
|
|
return ""
|
|
|
|
# Wenn der Text bereits unter dem Limit liegt, unverändert zurückgeben
|
|
if len(text) <= max_chars:
|
|
return text
|
|
|
|
# Andernfalls den Text auf max_chars begrenzen
|
|
return text[:max_chars] + "... [Inhalt aufgrund der Länge gekürzt]"
|
|
|
|
# Singleton-Instanz
|
|
_webcrawler_agent = None
|
|
|
|
def get_webcrawler_agent():
|
|
"""Gibt eine Singleton-Instanz des Webcrawler-Agenten zurück"""
|
|
global _webcrawler_agent
|
|
if _webcrawler_agent is None:
|
|
_webcrawler_agent = AgentWebcrawler()
|
|
return _webcrawler_agent |