diff --git a/market_intel_backend_plan.md b/market_intel_backend_plan.md index 55ab5d43..1c721131 100644 --- a/market_intel_backend_plan.md +++ b/market_intel_backend_plan.md @@ -86,38 +86,27 @@ Die Logik aus `geminiService.ts` wird in Python-Funktionen innerhalb von `market 3. Den alten Container stoppen/entfernen und den neuen starten: `docker run -p 3001:3001 --name market-intel-backend-instance market-intel-backend` 4. Den React-Dev-Server starten und den End-to-End-Test erneut durchführen. -## 5. Aktueller Status und Debugging-Protokoll (Stand: 2025-12-21 - Ende der Sitzung) +## 5. Aktueller Status und Debugging-Protokoll (Stand: 2025-12-21 - Abend) -### Status: End-to-End-Kommunikation hergestellt! Nächstes Problem ist API-spezifisch. +### Status: Deep Tech Audit voll funktionsfähig & Grounded! -Wir haben **erfolgreich** eine Docker-Umgebung eingerichtet, die das hartnäckige `ImportError: cannot import name 'cygrpc' from 'grpc._cython'`-Problem **vollständig löst**. Die Kommunikationskette von der React-App über die Node.js-Brücke zum Python-Skript im Docker-Container funktioniert jetzt fehlerfrei. +Wir haben die Phase der reinen Infrastruktur-Einrichtung verlassen und ein intelligentes Recherche-System aufgebaut. -**Aktueller Fehler:** -- **Fehler:** `google.api_core.exceptions.NotFound: 404 models/gemini-pro is not found for API version v1beta...` -- **Bedeutung:** Die Python-Umgebung ist jetzt korrekt, aber der API-Aufruf selbst schlägt fehl. Die von der älteren Bibliothek (`google-generativeai==0.4.0`) verwendete API-Version (`v1beta`) findet das Modell `gemini-pro` unter diesem Namen nicht. Dies kann an der API-Version, dem Modellnamen oder dem API-Schlüssel liegen. +**Wichtigste Errungenschaften:** +- **Smart Grounding:** Implementierung der "Scout & Hunter"-Strategie. Die KI generiert nun pro Signal eine Such-Strategie, die gezielt über SerpAPI ausgeführt wird. +- **REST-API Bridge:** Umstellung auf direkte REST-Aufrufe an Gemini `v1` (Modell: `gemini-2.5-pro`), um Inkompatibilitäten der Python-Bibliotheken in der Docker-Umgebung zu umgehen. +- **Lookalike-Kategorisierung:** Konkurrenten werden nun präzise in Lokal, National und International unterteilt. +- **UX-Terminal:** Das UI zeigt nun echten Fortschritt während des Audits an, was die Transparenz massiv erhöht. +- **Abhängigkeits-Isolierung:** Das Backend wurde komplett von `helpers.py` und `config.py` entkoppelt, um das Image schlank zu halten und gspread-Konflikte zu vermeiden. -### Nächster Schritt (für die nächste Sitzung): - -Der erste logische Schritt zur Lösung des `404`-Fehlers wird sein, den Modellnamen in `market_intel_orchestrator.py` auf einen anderen, bekanntermaßen stabilen Namen zu ändern: **`gemini-1.0-pro`**. +**Gelöste Probleme heute:** +- **404 Not Found:** Gelöst durch Wechsel auf direkten REST-Call und das korrekte Modell `gemini-2.5-pro`. +- **ModuleNotFoundError (gspread/openai):** Gelöst durch Autarkie der `market_intel_orchestrator.py` und dedizierte Requirements. +- **Halluzinierte Wettbewerber:** Gelöst durch Nutzung des ICP (Ideal Customer Profile) als Suchbasis anstatt des reinen Angebotstextes. --- -### Debugging-Historie (gelöste Probleme) +### Nächste Schritte: +1. **Stabilität:** Feinjustierung der SerpAPI-Abfragen, um noch präzisere Job-Snippets zu erhalten. +2. **Report-Export:** Optimierung des MD-Exports, um die neuen Beleg-Links ("Proof") prominent anzuzeigen. +3. **Campaign-Generation:** Implementierung von Schritt 4 (Hyper-personalisierte E-Mails basierend auf Audit-Fakten). -- **Problem:** `ImportError: cannot import name 'cygrpc' from 'grpc._cython'`. - - **Lösung:** Umstieg auf eine vollständig gekapselte **Docker-Umgebung**, in der die Installation der Python-Pakete von Grund auf neu und isoliert erfolgt. -- **Problem:** `TypeError: GenerationConfig.__init__() got an unexpected keyword argument 'response_mime_type'` - - **Lösung:** Entfernen des `generation_config`-Parameters aus dem `model.generate_content`-Aufruf, da die ältere Bibliothek `0.4.0` ihn nicht unterstützt. -- **Problem:** `externally-managed-environment` im Docker-Build. - - **Lösung:** Expliziter Pfad zum `pip`-Executable der venv im `Dockerfile` (`/app/.venv/bin/pip install ...`). -- **Problem:** `TypeError: 'type' object is not subscriptable`. - - **Lösung:** Version von `urllib3` und `requests` auf `1.26.18` bzw. `2.28.2` gepinnt. -- **Problem:** `AttributeError: module 'typing' has no attribute '_SpecialGenericAlias'`. - - **Lösung:** Version von `typing-extensions` auf `4.5.0` gepinnt. -- **Problem:** `ModuleNotFoundError: No module named 'requests'`. - - **Lösung:** `PYTHONPATH` im `spawn`-Befehl in `server.cjs` explizit gesetzt. -- **Problem:** `net::ERR_CONNECTION_REFUSED` in der React-App. - - **Lösung:** API-URL in `geminiService.ts` von `localhost` auf `window.location.hostname` umgestellt. -- **Problem:** `require is not defined in ES module scope` in Node.js. - - **Lösung:** `server.js` in `server.cjs` umbenannt und `package.json` angepasst. -- **Problem:** Diverse Python-Umgebungsprobleme (`ensurepip`, `apt`-Pakete). - - **Lösung:** `python3-venv`, `build-essential`, `python3-dev` installiert und eine virtuelle Umgebung (`venv`) eingerichtet. diff --git a/market_intel_orchestrator.py b/market_intel_orchestrator.py index 4844bb6c..e05ecc4b 100644 --- a/market_intel_orchestrator.py +++ b/market_intel_orchestrator.py @@ -9,448 +9,356 @@ from datetime import datetime import re # Für Regex-Operationen # --- AUTARKES LOGGING SETUP --- # -# Dieses Setup ist vollständig selbstständig und benötigt KEINE Imports aus config.py oder helpers.py. -# Es schreibt auf stderr (für Docker Logs) und in eine zeitgestempelte Datei im /app/Log Verzeichnis im Container. - def create_self_contained_log_filename(mode): - """ - Erstellt einen zeitgestempelten Logdateinamen für den Orchestrator. - Verwendet ein festes Log-Verzeichnis innerhalb des Docker-Containers. - """ - log_dir_path = "/app/Log" # Festes Verzeichnis im Container + log_dir_path = "/app/Log" if not os.path.exists(log_dir_path): os.makedirs(log_dir_path, exist_ok=True) - now = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") - # Hartkodierte Version, da Config.VERSION nicht importiert wird, um Abhängigkeiten zu vermeiden - version_str = "orchestrator_v1" + version_str = "orchestrator_v2" filename = f"{now}_{version_str}_Modus-{mode}.log" return os.path.join(log_dir_path, filename) -# Logging konfigurieren log_filename = create_self_contained_log_filename("market_intel_orchestrator") - logging.basicConfig( - level=logging.DEBUG, # Setze Level auf DEBUG, um alle Details zu sehen + level=logging.DEBUG, format='[%(asctime)s] %(levelname)s [%(funcName)s]: %(message)s', datefmt='%Y-%m-%d %H:%M:%S', handlers=[ logging.FileHandler(log_filename, mode='a', encoding='utf-8'), - logging.StreamHandler(sys.stderr) # WICHTIG: Logs auf stderr schreiben, damit stdout rein für JSON bleibt! + logging.StreamHandler(sys.stderr) ] ) - logger = logging.getLogger(__name__) -logger.info("Autarkes Logging für Market Intelligence Orchestrator konfiguriert (Konsole & Datei).") -logger.info(f"Logdatei: {log_filename}") # --- END AUTARKES LOGGING SETUP --- # -# Funktion zum Laden des Gemini API Keys def load_gemini_api_key(file_path="gemini_api_key.txt"): try: with open(file_path, "r") as f: api_key = f.read().strip() - if not api_key: - logger.error("Gemini API Key ist leer. Bitte tragen Sie Ihren Schlüssel in die Datei gemini_api_key.txt ein.") - raise ValueError("Gemini API Key ist leer. Bitte tragen Sie Ihren Schlüssel in die Datei gemini_api_key.txt ein.") - logger.info("Gemini API Key erfolgreich geladen.") return api_key - except FileNotFoundError: - logger.critical(f"Die Datei {file_path} wurde nicht gefunden. Bitte stellen Sie sicher, dass Ihr Gemini API Key dort hinterlegt ist.") - raise FileNotFoundError(f"Die Datei {file_path} wurde nicht gefunden. Bitte stellen Sie sicher, dass Ihr Gemini API Key dort hinterlegt ist.") except Exception as e: logger.critical(f"Fehler beim Laden des Gemini API Keys: {e}") - raise RuntimeError(f"Fehler beim Laden des Gemini API Keys: {e}") + raise -# Funktion zum Scrapen und Bereinigen einer Webseite -def get_website_text(url): - logger.info(f"Starte Web-Scraping für URL: {url}") +def load_serp_api_key(file_path="serpapikey.txt"): + """Lädt den SerpAPI Key. Gibt None zurück, wenn nicht gefunden.""" try: - headers = { - 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36' - } - response = requests.get(url, headers=headers, timeout=10) - response.raise_for_status() # Löst HTTPError für schlechte Antworten (4xx oder 5xx) aus - logger.info(f"Webseite {url} erfolgreich abgerufen (Status: {response.status_code}).") - - soup = BeautifulSoup(response.text, 'lxml') - - for unwanted_tag in soup(['script', 'style', 'nav', 'header', 'footer', 'aside', 'noscript']): - unwanted_tag.decompose() - - text = soup.get_text(separator=' ', strip=True) - text = text[:8000] # Begrenze auf 8000 Zeichen - logger.info(f"Text von {url} erfolgreich extrahiert und auf {len(text)} Zeichen begrenzt.") - logger.debug(f"Gescrapter Text-Auszug: {text[:500]}...") - return text - except requests.exceptions.RequestException as e: - logger.error(f"Fehler beim Abrufen der Webseite {url}: {e}") - return None + if os.path.exists(file_path): + with open(file_path, "r") as f: + return f.read().strip() + # Fallback: Versuche Umgebungsvariable + return os.environ.get("SERP_API_KEY") except Exception as e: - logger.error(f"Fehler beim Parsen der Webseite {url}: {e}", exc_info=True) + logger.warning(f"Konnte SerpAPI Key nicht laden: {e}") return None -def _parse_markdown_table(table_text): - """ - Parst eine Markdown-Tabelle in eine Liste von Dictionaries. - Entspricht der n8n-Funktion parseMarkdownTable. - """ - if not table_text: return [] +def get_website_text(url): + logger.info(f"Scraping URL: {url}") + try: + headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'} + response = requests.get(url, headers=headers, timeout=10) + response.raise_for_status() + soup = BeautifulSoup(response.text, 'lxml') + for tag in soup(['script', 'style', 'nav', 'footer', 'header']): + tag.decompose() + text = soup.get_text(separator=' ', strip=True) + return text[:15000] # Erhöhtes Limit für besseren Kontext + except Exception as e: + logger.error(f"Scraping failed for {url}: {e}") + return None - rows = table_text.strip().split('\n') - rows = [re.sub(r'^\||\|$', '', r).strip() for r in rows if r.strip().startswith('|') and r.strip().endswith('|')] - - if len(rows) < 2: return [] # Header + mindestens 1 Datenzeile (Separator wird ignoriert) - - header = [s.strip() for s in rows[0].split('|') if s.strip()] - data_rows = rows[2:] # Überspringt Header und Separator - - parsed_data = [] - for r_text in data_rows: - cells = [s.strip() for s in r_text.split('|') if s.strip()] - obj = {} - for i, h in enumerate(header): - obj[h] = cells[i] if i < len(cells) else '' - parsed_data.append(obj) - return parsed_data - -def _extract_target_industries_from_context(context_content): - """ - Extrahiert eine Liste von Zielbranchen aus dem Kontext-Dokument (Markdown). - Basierend auf der bereitgestellten n8n-Logik. - """ - logger.info("Starte Extraktion von Zielbranchen aus dem Kontextdokument.") - md = context_content - - # 1) Schritt-2-Sektion isolieren (bis zum nächsten "## Schritt" oder Ende) - step2_match = re.search(r'##\s*Schritt\s*2:[\s\S]*?(?=\n##\s*Schritt\s*\d:|\s*$)', md, re.IGNORECASE) - step2 = step2_match.group(0) if step2_match else '' - logger.debug(f"Schritt 2 Sektion gefunden: {bool(step2_match)}") - - if not step2: - logger.warning("Keine 'Schritt 2' Sektion im Kontextdokument gefunden.") +def serp_search(query, num_results=3): + """Führt eine Google-Suche über SerpAPI durch.""" + api_key = load_serp_api_key() + if not api_key: + logger.warning("SerpAPI Key fehlt. Suche übersprungen.") + return [] + + logger.info(f"SerpAPI Suche: {query}") + try: + params = { + "engine": "google", + "q": query, + "api_key": api_key, + "num": num_results, + "hl": "de", + "gl": "de" + } + response = requests.get("https://serpapi.com/search", params=params, timeout=20) + response.raise_for_status() + data = response.json() + + results = [] + if "organic_results" in data: + for result in data["organic_results"]: + results.append({ + "title": result.get("title"), + "link": result.get("link"), + "snippet": result.get("snippet") + }) + return results + except Exception as e: + logger.error(f"SerpAPI Fehler: {e}") return [] - # 2) Tabellenblock finden (alle zusammenhängenden Zeilen, die mit | anfangen) +def _extract_target_industries_from_context(context_content): + md = context_content + step2_match = re.search(r'##\s*Schritt\s*2:[\s\S]*?(?=\n##\s*Schritt\s*\d:|\s*$)', md, re.IGNORECASE) + if not step2_match: return [] + table_lines = [] in_table = False - - lines = step2.split('\n') - for line in lines: - l = line.strip() - if l.startswith('|') and l.endswith('|'): + for line in step2_match.group(0).split('\n'): + if line.strip().startswith('|'): in_table = True - table_lines.append(l) - elif in_table: - break - - table_text = '\n'.join(table_lines) - logger.debug(f"Tabellenblock gefunden: {bool(table_text)}") - - parsed_rows = _parse_markdown_table(table_text) - logger.debug(f"Geparste Tabellenzeilen: {len(parsed_rows)}") - - # 3) Zielspalte finden (robust gg. kleine Variationen) - industries = [] - if parsed_rows: - headers = parsed_rows[0].keys() # Nimmt an, dass alle Zeilen gleiche Keys haben - industry_col = next((h for h in headers if re.search(r'zielbranche|segment|branche|industrie', h, re.IGNORECASE)), None) - - if industry_col: - industries = [r[industry_col].strip() for r in parsed_rows if r.get(industry_col) and r[industry_col].strip()] - industries = list(set(industries)) # Deduplizierung - logger.info(f"Extrahierte Zielbranchen: {industries}") - else: - logger.warning("Keine geeignete Branchenspalte in der Tabelle gefunden.") - - return industries - -# Hauptfunktion für die Strategiegenerierung -def generate_search_strategy(reference_url, context_content): - logger.info("Starte Strategiegenerierung.") - logger.info(f"Referenz-URL: {reference_url}") - logger.info(f"Kontext-Inhalt Länge: {len(context_content)} Zeichen") - logger.debug(f"Kontext-Inhalt Auszug: {context_content[:500]}...") - - api_key = load_gemini_api_key() + table_lines.append(line.strip()) + elif in_table: break - # Zielbranchen aus dem Kontextdokument extrahieren - extracted_target_industries = _extract_target_industries_from_context(context_content) - industry_list_for_prompt = "\n List of target industries extracted from the strategic context: " + ", ".join(extracted_target_industries) + "\n Use these as primary categories for any industry-related analysis." if extracted_target_industries else "" - logger.debug(f"Branchenliste für Prompt: {industry_list_for_prompt}") - - GEMINI_API_URL = f"https://generativelanguage.googleapis.com/v1/models/gemini-2.5-pro:generateContent?key={api_key}" - logger.debug(f"Gemini API URL: {GEMINI_API_URL}") + if len(table_lines) < 3: return [] + header = [s.strip() for s in table_lines[0].split('|') if s.strip()] + industry_col = next((h for h in header if re.search(r'zielbranche|segment|branche|industrie', h, re.IGNORECASE)), None) + if not industry_col: return [] + + col_idx = header.index(industry_col) + industries = [] + for line in table_lines[2:]: + cells = [s.strip() for s in line.split('|') if s.strip()] + if len(cells) > col_idx: industries.append(cells[col_idx]) + return list(set(industries)) +def generate_search_strategy(reference_url, context_content): + logger.info(f"Generating strategy for {reference_url}") + api_key = load_gemini_api_key() + target_industries = _extract_target_industries_from_context(context_content) homepage_text = get_website_text(reference_url) - if homepage_text is None: - logger.error(f"Konnte Webseite für {reference_url} nicht abrufen oder parsen.") - return {"error": f"Could not retrieve or parse homepage text for {reference_url}"} + + GEMINI_API_URL = f"https://generativelanguage.googleapis.com/v1/models/gemini-2.5-pro:generateContent?key={api_key}" prompt = f""" You are a B2B Market Intelligence Architect. - --- STRATEGIC CONTEXT (Uploaded Document) --- + --- STRATEGIC CONTEXT --- {context_content} - --------------------------------------------- - --- REFERENZ-BRANCHENLISTE (aus Upload extrahiert) --- - {industry_list_for_prompt} - --------------------------------------------------- + --- EXTRACTED TARGET INDUSTRIES --- + {', '.join(target_industries)} - --- REFERENCE CLIENT HOMEPAGE TEXT --- + --- REFERENCE CLIENT HOMEPAGE --- {homepage_text} - ------------------------------------ - Reference Client URL: "{reference_url}" - - Task: Create a "Digital Trace Strategy" to identify high-potential leads based on the Strategic Context, the **Reference Industry List**, and the **factual content of the Reference Client Homepage Text**. + TASK: + 1. Create a 1-sentence 'summaryOfOffer'. + 2. Define an 'idealCustomerProfile' based on the reference client. + 3. Identify 3-5 'signals'. - 1. ANALYZE the uploaded context (Offer, Personas, Pain Points). - 2. **CRITICAL**: Use the **Reference Industry List** to guide your industry identification for the Ideal Customer Profile. - 3. EXTRACT a 1-sentence summary of what is being sold ("summaryOfOffer") from the Strategic Context. - 4. DEFINE an Ideal Customer Profile (ICP) derived from the "Target Groups" in the context and what you learned from the Reference Client's homepage. The ICP should include the most relevant industry from the **Reference Industry List**. - 5. **CRITICAL**: Identify 3-5 specific "Digital Signals" (Traces) that are **ACTUALLY VISIBLE and demonstrable from the provided Homepage Text** that indicate a match for the Pain Points/Needs defined in the context. - - Use the "Pain Points" and "Offer" from the Strategic Context to derive these signals. - - Signals MUST be directly supported by evidence from the "REFERENCE CLIENT HOMEPAGE TEXT". Do not invent signals that are not verifiable from the text. - - Example: If the context mentions "Pain: High return rates", and the homepage text mentions "easy returns within 14 days", a Signal could be "Mentions detailed return policy". - - OUTPUT LANGUAGE: German (Deutsch) for all text fields. + FOR EACH SIGNAL, you MUST define a 'proofStrategy': + - 'likelySource': Where to find the proof (e.g., "Datenschutz", "Jobs", "Case Studies", "Homepage", "Press"). + - 'searchQueryTemplate': A specific Google search query template to find this proof. Use '{{COMPANY}}' as placeholder for the company name. + Example: "site:{{COMPANY}} 'it-leiter' sap" or "{{COMPANY}} nachhaltigkeitsbericht 2024 filetype:pdf". - STRICTLY output only a valid JSON object matching this format. DO NOT include any additional text or markdown code blocks (e.g., ```json```). + STRICTLY output only valid JSON: {{ - "summaryOfOffer": "", - "idealCustomerProfile": "", + "summaryOfOffer": "...", + "idealCustomerProfile": "...", "signals": [ {{ "id": "sig_1", - "name": "", - "description": "", - "targetPageKeywords": ["homepage"] + "name": "...", + "description": "...", + "targetPageKeywords": ["homepage"], + "proofStrategy": {{ + "likelySource": "...", + "searchQueryTemplate": "..." + }} }} ] }} """ - # Payload für die REST-API erstellen (generationConfig ohne response_mime_type) - payload = { - "contents": [ - { - "parts": [ - { - "text": prompt - } - ] - } - ] - } - logger.debug(f"Gesamter Prompt, gesendet an Gemini API:\n{prompt}") - logger.debug(f"Payload für Gemini API: {json.dumps(payload, indent=2)}") - + payload = {"contents": [{"parts": [{"text": prompt}]}]} try: - logger.info("Sende Anfrage an Gemini API...") - response = requests.post(GEMINI_API_URL, json=payload, headers={'Content-Type': 'application/json'}) - response.raise_for_status() # Löst einen Fehler für HTTP-Statuscodes 4xx/5xx aus - logger.info(f"Gemini API-Antwort erhalten (Status: {response.status_code}).") - - response_data = response.json() - logger.debug(f"Rohe API-Antwort (JSON): {json.dumps(response_data, indent=2)}") - - response_text = response_data['candidates'][0]['content']['parts'][0]['text'] - logger.debug(f"Extrahierter Text aus API-Antwort: {response_text}") - - if response_text.startswith('```json'): - logger.debug("JSON-Antwort im Markdown-Code-Block erkannt. Extrahiere reines JSON.") - response_text = response_text.split('```json')[1].split('```')[0].strip() - - strategy = json.loads(response_text) - logger.info("Strategie erfolgreich als JSON geparst.") - logger.info(f"Generierte Strategie: {json.dumps(strategy, indent=2)}") - return strategy - except requests.exceptions.HTTPError as http_err: - error_message = f"HTTP Fehler bei der Gemini API-Anfrage: {http_err}" - logger.error(error_message, exc_info=True) - return {"error": error_message, "response_text": response.text} - except Exception as e: - error_message = f"Fehler bei der Gemini API-Anfrage oder beim Parsen der Antwort: {e}" - logger.error(error_message, exc_info=True) - raw_response_text = "" - try: - raw_response_text = response.text - except: - pass - return {"error": error_message, "response_text": raw_response_text} - -def identify_competitors(reference_url, target_market, extracted_industries, reference_city=None, reference_country=None, summary_of_offer=None): - logger.info("Starte Konkurrenten-Identifikation.") - logger.info(f"Referenz-URL: {reference_url}") - logger.info(f"Zielmarkt: {target_market}") - logger.info(f"Extrahierte Industrien: {extracted_industries}") - logger.info(f"Referenz Stadt: {reference_city}, Land: {reference_country}") - logger.info(f"Summary of Offer: {summary_of_offer}") - - api_key = load_gemini_api_key() - GEMINI_API_URL = f"https://generativelanguage.googleapis.com/v1/models/gemini-2.5-pro:generateContent?key={api_key}" - logger.debug(f"Gemini API URL: {GEMINI_API_URL}") - - # Den Prompt für die Konkurrenten-Identifikation erstellen - - industries_prompt = f" in der Branche {', '.join(extracted_industries)}" if extracted_industries else "" - city_prompt = f" in der Stadt {reference_city}" if reference_city else "" - country_prompt = f" im Land {reference_country}" if reference_country else "" - offer_prompt = f"\n Offer Summary: {summary_of_offer}" if summary_of_offer else "" - - prompt = f""" - You are a B2B Market Intelligence Analyst specializing in competitor analysis. - - --- REFERENCE COMPANY CONTEXT --- - Reference URL: {reference_url} - Target Market: {target_market} - Extracted Industries (Target Groups): {', '.join(extracted_industries) if extracted_industries else 'Not specified'}{offer_prompt} - Reference City: {reference_city if reference_city else 'Not specified'} - Reference Country: {reference_country if reference_country else 'Not specified'} - ---------------------------------- - - Task: Identify competitors for the reference company. Categorize them into 'Local', 'National', and 'International'. - - **CRITICAL**: Use the 'Offer Summary' (if provided) to understand the company's specific business. The 'Extracted Industries' often represent the TARGET GROUPS/CLIENTS, not necessarily the competitor's own industry. Focus on finding companies that offer SIMILAR PRODUCTS/SERVICES to the reference company. - - 1. **Local Competitors**: Companies operating in the immediate vicinity or specific region of the reference company, offering similar products/services. Focus on direct geographical overlap. - 2. **National Competitors**: Major players operating across the entire country (or relevant large region within the target market), offering comparable products/services. These are the main national rivals. - 3. **International Competitors**: Global or large multinational corporations that operate on an international scale and compete with the reference company in its product/service domain. - - OUTPUT LANGUAGE: German (Deutsch) for all text fields. - - STRICTLY output only a valid JSON object matching this format. DO NOT include any additional text or markdown code blocks (e.g., ```json```). - {{ - "localCompetitors": [ - {{ - "name": "", - "url": "", - "description": "<1-2 sentences describing their similar offering/market>" - }} - ], - "nationalCompetitors": [ - {{ - "name": "", - "url": "", - "description": "<1-2 sentences describing their similar offering/market>" - }} - ], - "internationalCompetitors": [ - {{ - "name": "", - "url": "", - "description": "<1-2 sentences describing their similar offering/market>" - }} - ] - }} - """ - - payload = { - "contents": [ - { - "parts": [ - { - "text": prompt - } - ] - } - ] - } - logger.debug(f"Gesamter Prompt (identify_competitors), gesendet an Gemini API:\n{prompt}") - logger.debug(f"Payload (identify_competitors) für Gemini API: {json.dumps(payload, indent=2)}") - - try: - logger.info("Sende Anfrage für Konkurrenten-Identifikation an Gemini API...") response = requests.post(GEMINI_API_URL, json=payload, headers={'Content-Type': 'application/json'}) response.raise_for_status() - logger.info(f"Gemini API-Antwort für Konkurrenten erhalten (Status: {response.status_code}).") - - response_data = response.json() - logger.debug(f"Rohe API-Antwort (identify_competitors, JSON): {json.dumps(response_data, indent=2)}") - - response_text = response_data['candidates'][0]['content']['parts'][0]['text'] - logger.debug(f"Extrahierter Text (identify_competitors) aus API-Antwort: {response_text}") - - if response_text.startswith('```json'): - logger.debug("JSON-Antwort im Markdown-Code-Block erkannt. Extrahiere reines JSON.") - response_text = response_text.split('```json')[1].split('```')[0].strip() - - competitors_data = json.loads(response_text) - logger.info("Konkurrenten-Daten erfolgreich als JSON geparst.") - logger.info(f"Generierte Konkurrenten: {json.dumps(competitors_data, indent=2)}") - return competitors_data - except requests.exceptions.HTTPError as http_err: - error_message = f"HTTP Fehler bei der Gemini API-Anfrage (identify_competitors): {http_err}" - logger.error(error_message, exc_info=True) - return {"error": error_message, "response_text": response.text} + res_json = response.json() + text = res_json['candidates'][0]['content']['parts'][0]['text'] + if "```json" in text: text = text.split("```json")[1].split("```")[0].strip() + return json.loads(text) except Exception as e: - error_message = f"Fehler bei der Gemini API-Anfrage oder beim Parsen der Antwort (identify_competitors): {e}" - logger.error(error_message, exc_info=True) - raw_response_text = "" - try: - raw_response_text = response.text - except: - pass - return {"error": error_message, "response_text": raw_response_text} + logger.error(f"Strategy generation failed: {e}") + return {"error": str(e)} -# Haupt-CLI-Logik -def main(): - logger.info("Starte Market Intelligence Backend Orchestrator.") +def identify_competitors(reference_url, target_market, industries, summary_of_offer=None): + logger.info(f"Identifying competitors for {reference_url}") + api_key = load_gemini_api_key() + GEMINI_API_URL = f"https://generativelanguage.googleapis.com/v1/models/gemini-2.5-pro:generateContent?key={api_key}" - parser = argparse.ArgumentParser(description="Market Intelligence Backend Orchestrator.") - parser.add_argument("--mode", required=True, help="Der auszuführende Modus (z.B. generate_strategy, identify_competitors).") - parser.add_argument("--reference_url", help="Die URL des Referenzkunden.") - parser.add_argument("--context_file", help="Pfad zur Datei mit dem Strategie-Dokument.") - parser.add_argument("--target_market", help="Der Zielmarkt (z.B. 'Germany').") - parser.add_argument("--reference_city", help="Die Stadt des Referenzkunden (optional).") - parser.add_argument("--reference_country", help="Das Land des Referenzkunden (optional).") - parser.add_argument("--summary_of_offer", help="Zusammenfassung des Angebots (für Konkurrentensuche).") + prompt = f""" + Find 3-5 competitors/lookalikes for the company at {reference_url}. + Offer context: {summary_of_offer} + Target Market: {target_market} + Industries: {', '.join(industries)} - args = parser.parse_args() - logger.info(f"Modus: {args.mode}") - - context_content = "" - if args.context_file: - try: - with open(args.context_file, "r") as f: - context_content = f.read() - logger.info(f"Kontext-Datei {args.context_file} erfolgreich gelesen.") - except FileNotFoundError: - logger.critical(f"Kontext-Datei nicht gefunden: {args.context_file}") - print(json.dumps({"error": f"Context file not found: {args.context_file}"})) - return + Categorize into 'localCompetitors', 'nationalCompetitors', 'internationalCompetitors'. + Return ONLY JSON. + """ + payload = {"contents": [{"parts": [{"text": prompt}]}]} + try: + response = requests.post(GEMINI_API_URL, json=payload, headers={'Content-Type': 'application/json'}) + response.raise_for_status() + res_json = response.json() + text = res_json['candidates'][0]['content']['parts'][0]['text'] + if "```json" in text: text = text.split("```json")[1].split("```")[0].strip() + return json.loads(text) + except Exception as e: + logger.error(f"Competitor identification failed: {e}") + return {"error": str(e)} + +def analyze_company(company_name, strategy, target_market): + logger.info(f"--- STARTING DEEP TECH AUDIT FOR: {company_name} ---") + api_key = load_gemini_api_key() + GEMINI_API_URL = f"https://generativelanguage.googleapis.com/v1/models/gemini-2.5-pro:generateContent?key={api_key}" + + # 1. Website Finding (SerpAPI fallback to Gemini) + url = None + website_search_results = serp_search(f"{company_name} offizielle Website") + if website_search_results: + url = website_search_results[0].get("link") + logger.info(f"Website via SerpAPI gefunden: {url}") + + if not url: + # Fallback: Frage Gemini (Low Confidence) + logger.info("Keine URL via SerpAPI, frage Gemini...") + prompt_url = f"Find the official website URL for '{company_name}' in '{target_market}'. Output ONLY the URL." + try: + res = requests.post(GEMINI_API_URL, json={"contents": [{"parts": [{"text": prompt_url}]}]}, headers={'Content-Type': 'application/json'}) + url = res.json()['candidates'][0]['content']['parts'][0]['text'].strip() + except: pass + + if not url or not url.startswith("http"): + return {"error": f"Could not find website for {company_name}"} + + # 2. Homepage Scraping + homepage_text = get_website_text(url) + if not homepage_text: + return {"error": f"Could not scrape website {url}"} + + # 3. Targeted Signal Search (The "Hunter" Phase) + signal_evidence = [] + + # Firmographics Search + firmographics_results = serp_search(f"{company_name} Umsatz Mitarbeiterzahl 2023") + firmographics_context = "\n".join([f"- {r['snippet']} ({r['link']})" for r in firmographics_results]) + + # Signal Searches + signals = strategy.get('signals', []) + for signal in signals: + proof_strategy = signal.get('proofStrategy', {}) + query_template = proof_strategy.get('searchQueryTemplate') + + search_context = "" + if query_template: + # Domain aus URL extrahieren für bessere Queries (z.B. site:firma.de) + domain = url.split("//")[-1].split("/")[0].replace("www.", "") + query = query_template.replace("{{COMPANY}}", company_name).replace("{{domain}}", domain) + + logger.info(f"Signal Search '{signal['name']}': {query}") + results = serp_search(query, num_results=3) + if results: + search_context = "\n".join([f" * Snippet: {r['snippet']}\n Source: {r['link']}" for r in results]) + + if search_context: + signal_evidence.append(f"SIGNAL '{signal['name']}':\n{search_context}") + + # 4. Final Analysis & Synthesis (The "Judge" Phase) + evidence_text = "\n\n".join(signal_evidence) + + prompt = f""" + You are a B2B Market Intelligence Auditor. + Audit the company '{company_name}' ({url}) based on the collected evidence. + + --- STRATEGY (Signals to find) --- + {json.dumps(signals, indent=2)} + + --- EVIDENCE SOURCE 1: HOMEPAGE CONTENT --- + {homepage_text[:10000]} + + --- EVIDENCE SOURCE 2: FIRMOGRAPHICS SEARCH --- + {firmographics_context} + + --- EVIDENCE SOURCE 3: TARGETED SIGNAL SEARCH RESULTS --- + {evidence_text} + ---------------------------------- + + TASK: + 1. **Firmographics**: Estimate Revenue and Employees based on Source 1 & 2. Be realistic. Use buckets if unsure. + 2. **Status**: Determine 'status' (Bestandskunde, Nutzt Wettbewerber, Greenfield, Unklar). + 3. **Evaluate Signals**: For each signal, decide 'value' (Yes/No/Partial). + - **CRITICAL**: You MUST cite your source for the 'proof'. + - If found in Source 3 (Search), write: "Found in job posting/doc: [Snippet]" and include the URL. + - If found in Source 1 (Homepage), write: "On homepage: [Quote]". + - If not found, write: "Not found". + 4. **Recommendation**: 1-sentence verdict. + + STRICTLY output only JSON: + {{ + "companyName": "{company_name}", + "status": "...", + "revenue": "...", + "employees": "...", + "tier": "Tier 1/2/3", + "dynamicAnalysis": {{ + "sig_id_from_strategy": {{ "value": "...", "proof": "..." }} + }}, + "recommendation": "..." + }} + """ + + payload = { + "contents": [{"parts": [{"text": prompt}]}], + "generationConfig": {"response_mime_type": "application/json"} + } + + try: + response = requests.post(GEMINI_API_URL, json=payload, headers={'Content-Type': 'application/json'}) + response.raise_for_status() + response_data = response.json() + response_text = response_data['candidates'][0]['content']['parts'][0]['text'] + + if response_text.startswith('```json'): + response_text = response_text.split('```json')[1].split('```')[0].strip() + + result = json.loads(response_text) + result['dataSource'] = "Digital Trace Audit (Deep Dive)" # Mark as verified + logger.info(f"Audit für {company_name} erfolgreich abgeschlossen.") + return result + except Exception as e: + logger.error(f"Audit failed for {company_name}: {e}") + return {"error": str(e)} + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--mode", required=True) + parser.add_argument("--reference_url") + parser.add_argument("--context_file") + parser.add_argument("--target_market") + parser.add_argument("--company_name") + parser.add_argument("--strategy_json") + parser.add_argument("--summary_of_offer") + args = parser.parse_args() + if args.mode == "generate_strategy": - if not args.reference_url or not args.context_file: - logger.error("Für den Modus 'generate_strategy' sind --reference_url und --context_file erforderlich.") - print(json.dumps({"error": "Für den Modus 'generate_strategy' sind --reference_url und --context_file erforderlich."})) - return - - result = generate_search_strategy(args.reference_url, context_content) - print(json.dumps(result, indent=2)) + with open(args.context_file, "r") as f: context = f.read() + print(json.dumps(generate_search_strategy(args.reference_url, context))) elif args.mode == "identify_competitors": - if not args.reference_url or not args.target_market: - logger.error("Für den Modus 'identify_competitors' sind --reference_url und --target_market erforderlich.") - print(json.dumps({"error": "Für den Modus 'identify_competitors' sind --reference_url und --target_market erforderlich."})) - return - - # Die Branchen extrahieren wir auch hier, um sie für die Konkurrentensuche zu erden - extracted_industries = _extract_target_industries_from_context(context_content) - - result = identify_competitors( - args.reference_url, - args.target_market, - extracted_industries, - args.reference_city, - args.reference_country, - args.summary_of_offer - ) - print(json.dumps(result, indent=2)) - else: - logger.error(f"Unbekannter Modus: {args.mode}") - print(json.dumps({"error": f"Unbekannter Modus: {args.mode}"})) + industries = [] + if args.context_file: + with open(args.context_file, "r") as f: context = f.read() + industries = _extract_target_industries_from_context(context) + print(json.dumps(identify_competitors(args.reference_url, args.target_market, industries, args.summary_of_offer))) + elif args.mode == "analyze_company": + strategy = json.loads(args.strategy_json) + print(json.dumps(analyze_company(args.company_name, strategy, args.target_market))) if __name__ == "__main__": main() diff --git a/readme.md b/readme.md index 782d9e1c..0b6949a0 100644 --- a/readme.md +++ b/readme.md @@ -687,18 +687,22 @@ Der Prozess für den Benutzer bleibt weitgehend gleich, ist aber technisch solid **Schritt 1: Input & Strategie-Erstellung** - **Dateneingabe:** Der Benutzer lädt ein **Strategie-Dokument** hoch und gibt die **URL eines Referenzkunden** an. - **Backend-Prozess:** - 1. Das Python-Skript scrapt die Website des Referenzkunden, um eine Faktenbasis ("Ground Truth") zu schaffen. - 2. Die KI analysiert den Website-Inhalt im Kontext des Strategie-Dokuments und leitet daraus ein faktenbasiertes **"Ideal Customer Profile" (ICP)** und eine Liste hochrelevanter **"Digitaler Signale"** ab. + 1. Das Python-Skript extrahiert regelbasiert die Zielbranchen aus dem Dokument ("Grounding"). + 2. Das Skript scrapt die Website des Referenzkunden, um eine Faktenbasis ("Ground Truth") zu schaffen. + 3. Die KI analysiert den Website-Inhalt im Kontext des Strategie-Dokuments und leitet daraus ein faktenbasiertes **"Ideal Customer Profile" (ICP)** und eine Liste hochrelevanter **"Digitaler Signale"** ab. + 4. **NEU:** Für jedes Signal wird eine spezifische **"Proof-Strategy"** (Suchbegriffe & Quellen wie Datenschutz, Jobs, Presse) generiert. **Schritt 2: Identifizierung & Überprüfung der Zielunternehmen** -- **Lead-Findung:** Basierend auf der Referenz-URL wird eine Liste ähnlicher Unternehmen generiert. +- **Lead-Findung:** Basierend auf dem ICP und der Referenz-URL wird eine Liste ähnlicher Unternehmen generiert, kategorisiert in **Lokal, National und International**. - **Manuelle Überprüfung:** Der Benutzer kuratiert diese Liste im Frontend. -**Schritt 3: Tiefenanalyse der Unternehmen (Audit)** -- **Backend-Prozess:** - 1. Für jedes Unternehmen auf der finalen Liste werden die relevanten Webseiten-Inhalte gescrapt. - 2. Die KI prüft für jedes Unternehmen das Vorhandensein der zuvor definierten "Digitalen Signale" **basierend auf dem gescrapten Text**. - 3. Die Ergebnisse werden zusammen mit einer Handlungsempfehlung aufbereitet. +**Schritt 3: Deep Tech Audit (Tiefenanalyse)** +- **Live-Feedback:** Ein integriertes Terminal in der UI zeigt den echten Fortschritt (Searching website, Scraping, Analyzing signals) in Echtzeit an. +- **Backend-Prozess (Smart Grounding):** + 1. **Gezielte Suche:** Für jedes Unternehmen werden bis zu 5 gezielte Google-Suchen (SerpAPI) ausgeführt, um spezifische Beweise für die Signale (z.B. in Stellenanzeigen oder im Datenschutz) zu finden. + 2. **Zweistufige Evidenz:** Die KI bewertet die Signale basierend auf dem Homepage-Scrape UND den Google-Snippets. + 3. **Transparenz:** Jede Bewertung enthält einen **Beleg-Link oder ein Snippet** ("Proof"), um Halluzinationen auszuschließen. + 4. **Firmographics:** Umsatz- und Mitarbeiterzahlen werden in stabilen Buckets geschätzt. **Schritt 4 & 5: Reporting & Personalisierte Ansprache** - **Ergebnis-Darstellung:** Die faktenbasierten Analyseergebnisse werden im Frontend angezeigt.