Files
Brancheneinstufung2/competitor-analysis-app/competitor_analysis_orchestrator.py
Floke 6a7d56a9c9 feat(competitor-analysis): Fix 404, SDK compatibility, and update docs
Resolved multiple issues preventing the 'competitor-analysis' app from running and serving its frontend:

1.  **Fixed Python SyntaxError in Prompts:** Corrected unterminated string literals and ensure proper multi-line string formatting (using .format() instead of f-strings for complex prompts) in .
2.  **Addressed Python SDK Compatibility (google-generativeai==0.3.0):**
    *   Removed  for  and  by adapting the orchestrator to pass JSON schemas as direct Python dictionaries, as required by the older SDK version.
    *   Updated  with detailed guidance on handling / imports and dictionary-based schema definitions for older SDKs.
3.  **Corrected Frontend Build Dependencies:** Moved critical build dependencies (like , , ) from  to  in .
    *   Updated  to include this  pitfall, ensuring frontend build tools are installed in Docker.
4.  **Updated Documentation:**
    *   : Added comprehensive lessons learned regarding  dependencies, Python SDK versioning (specifically  and  imports for ), and robust multi-line prompt handling.
    *   : Integrated specific details of the encountered errors and their solutions, making the migration report a more complete historical record and guide.

These changes collectively fix the 404 error by ensuring the Python backend starts correctly and serves the frontend assets after a successful build.
2026-01-10 09:10:00 +00:00

918 lines
44 KiB
Python

import os
import json
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
from typing import List, Dict, Any, Optional
import google.generativeai as genai
from google.generativeai.types import HarmCategory, HarmBlockThreshold
# Load environment variables
load_dotenv()
API_KEY = os.getenv("GEMINI_API_KEY")
# Fallback: check for API Key in a file (common in this project's Docker setup)
if not API_KEY:
key_file_path = os.getenv("GEMINI_API_KEY_FILE", "/app/gemini_api_key.txt")
if os.path.exists(key_file_path):
with open(key_file_path, 'r') as f:
API_KEY = f.read().strip()
if not API_KEY:
raise ValueError("GEMINI_API_KEY environment variable or file not set")
genai.configure(api_key=API_KEY)
# Use a candidate list for models as per migration guide
MODEL_CANDIDATES = ['gemini-1.5-pro', 'gemini-1.0-pro'] # Added 1.0-pro as fallback
model = None
for candidate in MODEL_CANDIDATES:
try:
model = genai.GenerativeModel(candidate)
print(f"DEBUG: Using Gemini model: {candidate}")
break
except Exception as e:
print(f"DEBUG: Could not load model {candidate}: {e}")
if "404" in str(e):
continue
raise e
if not model:
raise ValueError(f"No suitable Gemini model found from candidates: {MODEL_CANDIDATES}")
app = FastAPI()
# Configure CORS
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
def parse_json_response(text: str) -> Any:
"""Parses JSON response, stripping markdown code blocks."""
try:
# Clean the text, removing markdown code block fences
cleaned_text = text.replace('```json', '').replace('```', '').strip()
# Handle cases where the model might return a list instead of a direct object
result = json.loads(cleaned_text)
if isinstance(result, list) and result:
# If it's a list, assume the first element is the intended object
return result[0]
return result
except json.JSONDecodeError as e:
print(f"Failed to parse JSON: {e}\nOriginal text: {text}")
raise ValueError("Invalid JSON response from API")
# --- Schemas (ported from TypeScript) ---
evidence_schema = {
"type": "object",
"properties": {
"url": {"type": "string"},
"snippet": {"type": "string"},
},
"required": ['url', 'snippet']
}
product_schema = {
"type": "object",
"properties": {
"name": {"type": "string", "description": "Product name"},
"purpose": {"type": "string", "description": "Purpose description (1-2 sentences)"},
"evidence": {"type": "array", "items": evidence_schema},
},
"required": ['name', 'purpose', 'evidence']
}
industry_schema = {
"type": "object",
"properties": {
"name": {"type": "string", "description": "Name of the target industry"},
"evidence": {"type": "array", "items": evidence_schema},
},
"required": ['name', 'evidence']
}
# --- Request Models for FastAPI ---
class ProductDetailsRequest(BaseModel):
name: str
url: str
language: str
class FetchStep1DataRequest(BaseModel):
start_url: str
language: str
class ProductModel(BaseModel):
name: str
purpose: str
evidence: List[Dict[str, str]]
class TargetIndustryModel(BaseModel):
name: str
evidence: List[Dict[str, str]]
class FetchStep2DataRequest(BaseModel):
products: List[ProductModel]
industries: List[TargetIndustryModel]
language: str
class KeywordModel(BaseModel):
term: str
rationale: str
class FetchStep3DataRequest(BaseModel):
keywords: List[KeywordModel]
market_scope: str
language: str
class CompanyModel(BaseModel):
name: str
start_url: str
class CompetitorCandidateModel(BaseModel):
name: str
url: str
confidence: float
why: str
evidence: List[Dict[str, str]]
class FetchStep4DataRequest(BaseModel):
company: CompanyModel
competitors: List[CompetitorCandidateModel]
language: str
class AnalysisModel(BaseModel):
competitor: Dict[str, str]
portfolio: List[Dict[str, str]]
target_industries: List[str]
delivery_model: str
overlap_score: int
differentiators: List[str]
evidence: List[Dict[str, str]]
class FetchStep5DataSilverBulletsRequest(BaseModel):
company: CompanyModel
analyses: List[AnalysisModel]
language: str
class SilverBulletModel(BaseModel):
competitor_name: str
statement: str
class FetchStep6DataConclusionRequest(BaseModel):
company: CompanyModel
products: List[ProductModel]
industries: List[TargetIndustryModel]
analyses: List[AnalysisModel]
silver_bullets: List[SilverBulletModel]
language: str
class FetchStep7DataBattlecardsRequest(BaseModel):
company: CompanyModel
analyses: List[AnalysisModel]
silver_bullets: List[SilverBulletModel]
language: str
class ShortlistedCompetitorModel(BaseModel):
name: str
url: str
class FetchStep8DataReferenceAnalysisRequest(BaseModel):
competitors: List[ShortlistedCompetitorModel]
language: str
# --- Endpoints ---
@app.post("/api/fetchProductDetails")
async def fetch_product_details(request: ProductDetailsRequest):
prompts = {
"de": f"""
Analysiere die Webseite {request.url} und beschreibe den Zweck des Produkts "{request.name}" in 1-2 Sätzen. Gib auch die genaue URL und ein relevantes Snippet als Beleg an.
Das "name" Feld im JSON soll der offizielle Produktname sein, wie er auf der Seite gefunden wird, oder "{request.name}" falls nicht eindeutig.
Antworte ausschließlich im JSON-Format.
""",
"en": f"""
Analyze the website {request.url} and describe the purpose of the product "{request.name}" in 1-2 sentences. Also provide the exact URL and a relevant snippet as evidence.
The "name" field in the JSON should be the official product name as found on the page, or "{request.name}" if not clearly stated.
Respond exclusively in JSON format.
"""
}
try:
response = await model.generate_content_async(
prompts[request.language],
generation_config=genai.GenerationConfig(
response_mime_type="application/json",
response_schema=product_schema
),
safety_settings={
HarmCategory.HARM_CATEGORY_HARASSMENT: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_HATE_SPEECH: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: HarmBlockThreshold.BLOCK_NONE,
},
)
return parse_json_response(response.text)
except Exception as e:
print(f"Error in fetch_product_details: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/fetchStep1Data")
async def fetch_step1_data(request: FetchStep1DataRequest):
prompts = {
"de": f"""
Rolle: Research-Agent für B2B-Software-Wettbewerbsanalyse.
Aufgabe: Analysiere die Website {request.start_url} und identifiziere die Hauptprodukte/Lösungen und deren Zielbranchen.
Regeln:
1. Konzentriere dich auf offizielle Produkt- und Lösungsseiten.
2. Jede Information (Produkt, Branche) muss mit einer URL und einem kurzen Snippet belegt werden.
3. Sei präzise und vermeide Spekulationen.
Antworte ausschließlich im JSON-Format.
""",
"en": f"""
Role: Research Agent for B2B Software Competitor Analysis.
Task: Analyze the website {request.start_url} and identify the main products/solutions and their target industries.
Rules:
1. Focus on official product and solution pages.
2. Every piece of information (product, industry) must be backed by a URL and a short snippet as evidence.
3. Be precise and avoid speculation.
Respond exclusively in JSON format.
"""
}
response_schema = {
"type": "object",
"properties": {
"products": {"type": "array", "items": product_schema},
"target_industries": {"type": "array", "items": industry_schema},
},
"required": ['products', 'target_industries']
}
try:
response = await model.generate_content_async(
prompts[request.language],
generation_config=genai.GenerationConfig(
response_mime_type="application/json",
response_schema=response_schema
),
safety_settings={
HarmCategory.HARM_CATEGORY_HARASSMENT: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_HATE_SPEECH: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: HarmBlockThreshold.BLOCK_NONE,
},
)
return parse_json_response(response.text)
except Exception as e:
print(f"Error in fetch_step1_data: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/fetchStep2Data")
async def fetch_step2_data(request: FetchStep2DataRequest):
prompts = {
"de": f"""
Rolle: Research-Agent.
Aufgabe: Leite aus den folgenden Produkt- und Brancheninformationen 10-25 präzise deutsche und englische Keywords/Suchphrasen für die Wettbewerbsrecherche ab.
Kontext:
Produkte: {', '.join([f'{p.name} ({p.purpose})' for p in request.products])}
Branchen: {', '.join([i.name for i in request.industries])}
Regeln:
1. Erstelle Cluster: Produktkategorie, Funktionskern, Zielbranchen, Synonyme/Englischvarianten.
2. Gib für jedes Keyword eine kurze Begründung ("rationale").
3. Antworte ausschließlich im JSON-Format.
""",
"en": f"""
Role: Research Agent.
Task: From the following product and industry information, derive 10-25 precise English keywords/search phrases for competitor research.
Context:
Products: {', '.join([f'{p.name} ({p.purpose})' for p in request.products])}
Industries: {', '.join([i.name for i in request.industries])}
Rules:
1. Create clusters: Product category, core function, target industries, synonyms.
2. Provide a brief rationale for each keyword.
3. Respond exclusively in JSON format.
"""
}
response_schema = {
"type": "object",
"properties": {
"keywords": {
"type": "array",
"items": {
"type": "object",
"properties": {
"term": {"type": "string"},
"rationale": {"type": "string"}
},
"required": ['term', 'rationale']
}
}
},
"required": ['keywords']
}
try:
response = await model.generate_content_async(
prompts[request.language],
generation_config=genai.GenerationConfig(
response_mime_type="application/json",
response_schema=response_schema
),
safety_settings={
HarmCategory.HARM_CATEGORY_HARASSMENT: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_HATE_SPEECH: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: HarmBlockThreshold.BLOCK_NONE,
},
)
return parse_json_response(response.text)
except Exception as e:
print(f"Error in fetch_step2_data: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/fetchStep3Data")
async def fetch_step3_data(request: FetchStep3DataRequest):
prompts = {
"de": f"""
Rolle: Research-Agent.
Aufgabe: Finde relevante Wettbewerber basierend auf den folgenden Keywords. Fokussiere dich auf den Markt: {request.market_scope}.
Keywords: {', '.join([k.term for k in request.keywords])}
Regeln:
1. Suche nach Software-Anbietern, nicht nach Resellern, Beratungen oder Implementierungspartnern.
2. Gib für jeden Kandidaten an: Name, URL, Eignungsbegründung (why), Confidence-Score (0.0-1.0) und Belege (URL+Snippet).
3. Antworte ausschließlich im JSON-Format.
""",
"en": f"""
Role: Research Agent.
Task: Find relevant competitors based on the following keywords. Focus on the market: {request.market_scope}.
Keywords: {', '.join([k.term for k in request.keywords])}
Rules:
1. Search for software vendors, not resellers, consultants, or implementation partners.
2. For each candidate, provide: Name, URL, justification for inclusion (why), confidence score (0.0-1.0), and evidence (URL+snippet).
3. Respond exclusively in JSON format.
"""
}
response_schema = {
"type": "object",
"properties": {
"competitor_candidates": {
"type": "array",
"items": {
"type": "object",
"properties": {
"name": {"type": "string"},
"url": {"type": "string"},
"confidence": {"type": "number"},
"why": {"type": "string"},
"evidence": {"type": "array", "items": evidence_schema}
},
"required": ['name', 'url', 'confidence', 'why', 'evidence']
}
}
},
"required": ['competitor_candidates']
}
try:
response = await model.generate_content_async(
prompts[request.language],
generation_config=genai.GenerationConfig(
response_mime_type="application/json",
response_schema=response_schema
),
safety_settings={
HarmCategory.HARM_CATEGORY_HARASSMENT: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_HATE_SPEECH: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: HarmBlockThreshold.BLOCK_NONE,
},
)
return parse_json_response(response.text)
except Exception as e:
print(f"Error in fetch_step3_data: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/fetchStep4Data")
async def fetch_step4_data(request: FetchStep4DataRequest):
competitors_summary = '\n'.join([f'- {c.name}: {c.url}' for c in request.competitors])
prompts = {
"de": f"""
Rolle: Research-Agent.
Aufgabe: Führe eine detaillierte Portfolio- & Positionierungsanalyse für jeden der folgenden Wettbewerber durch. Vergleiche sie mit dem Ausgangsunternehmen {request.company.name} ({request.company.start_url}).
Wettbewerber:
{competitors_summary}
Analyse-Punkte pro Wettbewerber:
1. portfolio: Kernprodukte (max. 5) mit kurzem Zweck.
2. target_industries: Hauptzielbranchen.
3. delivery_model: Geschäfts-/Bereitstellungsmodell (z.B. SaaS, On-Premise), falls ersichtlich.
4. overlap_score: 0-100, basierend auf Produktfunktion, Zielbranche, Terminologie im Vergleich zum Ausgangsunternehmen.
5. differentiators: 3-5 Bulletpoints zu Alleinstellungsmerkmalen oder Unterschieden.
6. evidence: Wichtige Beleg-URLs mit Snippets.
Regeln:
1. Jede Behauptung mit Quellen belegen.
2. Antworte ausschließlich im JSON-Format.
""",
"en": f"""
Role: Research Agent.
Task: Conduct a detailed portfolio & positioning analysis for each of the following competitors. Compare them with the initial company {request.company.name} ({request.company.start_url}).
Competitors:
{competitors_summary}
Analysis points per competitor:
1. portfolio: Core products (max 5) with a brief purpose.
2. target_industries: Main target industries.
3. delivery_model: Business/delivery model (e.g., SaaS, On-Premise), if apparent.
4. overlap_score: 0-100, based on product function, target industry, terminology compared to the initial company.
5. differentiators: 3-5 bullet points on unique selling points or differences.
6. evidence: Key supporting URLs with snippets.
Rules:
1. Back up every claim with sources.
2. Respond exclusively in JSON format.
"""
}
response_schema = {
"type": "object",
"properties": {
"analyses": {
"type": "array",
"items": {
"type": "object",
"properties": {
"competitor": {"type": "object", "properties": {"name": {"type": "string"}, "url": {"type": "string"}}}
"portfolio": {"type": "array", "items": {"type": "object", "properties": {"product": {"type": "string"}, "purpose": {"type": "string"}}}}
"target_industries": {"type": "array", "items": {"type": "string"}},
"delivery_model": {"type": "string"},
"overlap_score": {"type": "integer"},
"differentiators": {"type": "array", "items": {"type": "string"}},
"evidence": {"type": "array", "items": evidence_schema}
},
"required": ['competitor', 'portfolio', 'target_industries', 'delivery_model', 'overlap_score', 'differentiators', 'evidence']
}
}
},
"required": ['analyses']
}
try:
response = await model.generate_content_async(
prompts[request.language],
generation_config=genai.GenerationConfig(
response_mime_type="application/json",
response_schema=response_schema
),
safety_settings={
HarmCategory.HARM_CATEGORY_HARASSMENT: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_HATE_SPEECH: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: HarmBlockThreshold.BLOCK_NONE,
},
)
return parse_json_response(response.text)
except Exception as e:
print(f"Error in fetch_step4_data: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/fetchStep5Data_SilverBullets")
async def fetch_step5_data_silver_bullets(request: FetchStep5DataSilverBulletsRequest):
competitor_data_summary = '\n'.join([
f"- Competitor: {a.competitor['name']}\n - Portfolio Focus: {', '.join([p['product'] for p in a.portfolio]) or 'N/A'}\n - Differentiators: {'; '.join(a.differentiators)}"
for a in request.analyses
])
prompts = {
"de": f"""
Rolle: Strategieberater.
Aufgabe: Erstelle für das Unternehmen "{request.company.name}" eine "Silver Bullet" für jeden Wettbewerber. Eine "Silver Bullet" ist ein prägnanter Satz (max. 25 Wörter), der im Vertriebsgespräch genutzt werden kann, um sich vom jeweiligen Wettbewerber abzugrenzen.
Kontext: {request.company.name} wird mit den folgenden Wettbewerbern verglichen:
{competitor_data_summary.replace("Competitor:", "Wettbewerber:").replace("Portfolio Focus:", "Portfolio-Fokus:").replace("Differentiators:", "Alleinstellungsmerkmale:")}
Regeln:
1. Formuliere für JEDEN Wettbewerber einen einzigen, schlagkräftigen Satz.
2. Der Satz soll eine Schwäche des Wettbewerbers oder eine Stärke von {request.company.name} im direkten Vergleich hervorheben.
3. Sei prägnant und überzeugend.
4. Antworte ausschließlich im JSON-Format.
""",
"en": f"""
Role: Strategy Consultant.
Task: Create a "Silver Bullet" for the company "{request.company.name}" for each competitor. A "Silver Bullet" is a concise sentence (max 25 words) that can be used in a sales pitch to differentiate from the respective competitor.
Context: {request.company.name} is being compared with the following competitors:
{competitor_data_summary}
Rules:
1. Formulate a single, powerful sentence for EACH competitor.
2. The sentence should highlight a weakness of the competitor or a strength of {request.company.name} in direct comparison.
3. Be concise and persuasive.
4. Respond exclusively in JSON format.
"""
}
response_schema = {
"type": "object",
"properties": {
"silver_bullets": {
"type": "array",
"items": {
"type": "object",
"properties": {
"competitor_name": {"type": "string"},
"statement": {"type": "string"}
},
"required": ['competitor_name', 'statement']
}
}
},
"required": ['silver_bullets']
}
try:
response = await model.generate_content_async(
prompts[request.language],
generation_config=genai.GenerationConfig(
response_mime_type="application/json",
response_schema=response_schema
),
safety_settings={
HarmCategory.HARM_CATEGORY_HARASSMENT: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_HATE_SPEECH: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: HarmBlockThreshold.BLOCK_NONE,
},
)
return parse_json_response(response.text)
except Exception as e:
print(f"Error in fetch_step5_data_silver_bullets: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/fetchStep6Data_Conclusion")
async def fetch_step6_data_conclusion(request: FetchStep6DataConclusionRequest):
competitor_data_summary = '\n\n'.join([
f"- Competitor: {a.competitor['name']}\n - Portfolio: {', '.join([p['product'] for p in a.portfolio]) or 'N/A'}\n - Target Industries: {', '.join(a.target_industries) or 'N/A'}\n - Overlap Score: {a.overlap_score}\n - Differentiators: {'; '.join(a.differentiators)}"
for a in request.analyses
])
silver_bullets_summary = '\n'.join([f"- Gegen {sb.competitor_name}: \"{sb.statement}\"" for sb in request.silver_bullets])
if request.language == 'en':
silver_bullets_summary = '\n'.join([f"- Against {sb.competitor_name}: \"{sb.statement}\"" for sb in request.silver_bullets])
prompts = {
"de": """
Rolle: Research-Agent.
Aufgabe: Erstelle ein Fazit der Wettbewerbsanalyse für {company_name}.
Ausgangsunternehmen ({company_name}) Daten:
- Produkte: {products_summary}
- Branchen: {industries_summary}
Zusammengefasste Wettbewerber-Daten:
{competitor_data_summary_de}
Strategische Positionierung ("Silver Bullets"):
{silver_bullets_summary_de}
Erstelle:
1. product_matrix: Analysiere ALLE Produkte von {company_name} und den Wettbewerbern. Identifiziere 5-10 generische Produktkategorien oder Kernfunktionalitäten (z.B. "Mobile Lösung", "Disposition & Planung", "Asset Management"). Erstelle dann eine Matrix basierend auf diesen generischen Kategorien. Jedes Element im Array repräsentiert eine dieser Kategorien. Jedes Element hat ein "product" (string, der Name der generischen Kategorie) und ein "availability" (Array). Das "availability"-Array enthält Objekte mit "competitor" (string) und "has_offering" (boolean), für JEDEN Anbieter (inkl. {company_name}), das anzeigt, ob der Anbieter eine Lösung in dieser Kategorie hat.
2. industry_matrix: Ein Array. Jedes Element repräsentiert eine Branche (von ALLEN Anbietern, inkl. {company_name}). Jedes Element hat eine "industry" (string) und ein "availability" (Array). Das "availability"-Array enthält Objekte mit "competitor" (string) und "has_offering" (boolean), für JEDEN Anbieter (inkl. {company_name}).
3. overlap_scores: Eine Liste der Wettbewerber und ihrer Overlap-Scores.
4. summary: Eine knappe Einordnung (2-3 Sätze), wer sich worauf positioniert.
5. opportunities: Wo liegen Lücken oder Chancen für {company_name}?
6. next_questions: Max. 5 offene Fragen oder nächste Schritte für den User.
Regeln:
1. Antworte ausschließlich im JSON-Format gemäß dem vorgegebenen Schema.
".format(
company_name=request.company.name,
products_summary=', '.join([p.name for p in request.products]),
industries_summary=', '.join([i.name for i in request.industries]),
competitor_data_summary_de=competitor_data_summary.replace("Competitor:", "Wettbewerber:").replace("Portfolio:", "Portfolio:").replace("Target Industries:", "Zielbranchen:").replace("Overlap Score:", "Overlap Score:").replace("Differentiators:", "Alleinstellungsmerkmale:"),
silver_bullets_summary_de=silver_bullets_summary
),
"en": """
Role: Research Agent.
Task: Create a conclusion for the competitive analysis for {company_name}.
Initial Company ({company_name}) Data:
- Products: {products_summary}
- Industries: {industries_summary}
Summarized Competitor Data:
{competitor_data_summary_en}
Strategic Positioning ("Silver Bullets"):
{silver_bullets_summary_en}
Create:
1. product_matrix: Analyze ALL products from {company_name} and the competitors. Identify 5-10 generic product categories or core functionalities (e.g., "Mobile Solution", "Dispatch & Planning", "Asset Management"). Then create a matrix based on these generic categories. Each element in the array represents one of these categories. Each element has a "product" (string, the name of the generic category) and an "availability" (array). The "availability"-array contains objects with "competitor" (string) and "has_offering" (boolean), for EVERY provider (incl. {company_name}), indicating if the provider has a solution in this category.
2. industry_matrix: An array. Each element represents an industry (from ALL providers, incl. {company_name}). Each element has an "industry" (string) and an "availability" (array). The "availability"-array contains objects with "competitor" (string) and "has_offering" (boolean), for EVERY provider (incl. {company_name}).
3. overlap_scores: A list of competitors and their overlap scores.
4. summary: A brief assessment (2-3 sentences) of who is positioned where.
5. opportunities: Where are the gaps or opportunities for {company_name}?
6. next_questions: Max 5 open questions or next steps for the user.
Rules:
1. Respond exclusively in JSON format according to the provided schema.
".format(
company_name=request.company.name,
products_summary=', '.join([p.name for p in request.products]),
industries_summary=', '.join([i.name for i in request.industries]),
competitor_data_summary_en=competitor_data_summary,
silver_bullets_summary_en=silver_bullets_summary
)
}
response_schema = {
"type": "object",
"properties": {
"conclusion": {
"type": "object",
"properties": {
"product_matrix": {
"type": "array",
"description": "Array representing a feature-based product comparison. Each item is a generic product category or core functionality.",
"items": {
"type": "object",
"properties": {
"product": {"type": "string", "description": "Name of the generic product category/feature."},
"availability": {
"type": "array",
"description": "Which competitors offer this product.",
"items": {
"type": "object",
"properties": {
"competitor": {"type": "string"},
"has_offering": {"type": "boolean", "description": "True if the competitor has a similar offering."}
},
"required": ['competitor', 'has_offering']
}
}
},
"required": ['product', 'availability']
}
},
"industry_matrix": {
"type": "array",
"description": "Array representing industry comparison. Each item is an industry.",
"items": {
"type": "object",
"properties": {
"industry": {"type": "string", "description": "Name of the industry."},
"availability": {
"type": "array",
"description": "Which competitors serve this industry.",
"items": {
"type": "object",
"properties": {
"competitor": {"type": "string"},
"has_offering": {"type": "boolean", "description": "True if the competitor serves this industry."}
},
"required": ['competitor', 'has_offering']
}
}
},
"required": ['industry', 'availability']
}
},
"overlap_scores": {"type": "array", "items": {"type": "object", "properties": {"competitor": {"type": "string"}, "score": {"type": "number"}}}},
"summary": {"type": "string"},
"opportunities": {"type": "string"},
"next_questions": {"type": "array", "items": {"type": "string"}}
},
"required": ['product_matrix', 'industry_matrix', 'overlap_scores', 'summary', 'opportunities', 'next_questions']
}
},
"required": ['conclusion']
}
try:
response = await model.generate_content_async(
prompts[request.language],
generation_config=genai.GenerationConfig(
response_mime_type="application/json",
response_schema=response_schema
),
safety_settings={
HarmCategory.HARM_CATEGORY_HARASSMENT: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_HATE_SPEECH: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: HarmBlockThreshold.BLOCK_NONE,
},
)
return parse_json_response(response.text)
except Exception as e:
print(f"Error in fetch_step6_data_conclusion: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/fetchStep7Data_Battlecards")
async def fetch_step7_data_battlecards(request: FetchStep7DataBattlecardsRequest):
competitor_data_summary = '\n\n'.join([
f"- Competitor: {a.competitor['name']}\n - Portfolio Focus: {', '.join([p['product'] for p in a.portfolio]) or 'N/A'}\n - Target Industries: {', '.join(a.target_industries)}\n - Differentiators: {'; '.join(a.differentiators)}\n - Silver Bullet against this competitor: \"{next((sb.statement for sb in request.silver_bullets if sb.competitor_name == a.competitor['name']), 'Not found.')}\""
for a in request.analyses
])
prompts = {
"de": f"""
Rolle: Vertriebsstratege und Coach für das B2B-Softwareunternehmen "{request.company.name}".
Aufgabe: Erstelle für jeden der folgenden Wettbewerber eine detaillierte "Sales Battlecard". Diese Battlecard soll dem Vertriebsteam konkrete, handlungsorientierte Argumente für Kundengespräche an die Hand geben.
Informationen über das eigene Unternehmen: {request.company.name}
Detaillierte Analyse der Wettbewerber:
{competitor_data_summary.replace("Competitor:", "Wettbewerber:").replace("Portfolio Focus:", "Portfolio-Fokus:").replace("Target Industries:", "Zielbranchen:").replace("Differentiators:", "Alleinstellungsmerkmale:").replace("Silver Bullet against this competitor:", "Silver Bullet gegen diesen Wettbewerber:")}
Für JEDEN Wettbewerber, erstelle die folgenden Sektionen einer Battlecard im JSON-Format:
1. **competitor_name**: Der Name des Wettbewerbers.
2. **competitor_profile**:
* **focus**: Fasse den Kernfokus des Wettbewerbers (Produkte & Branchen) in einem Satz zusammen.
* **positioning**: Beschreibe die Kernpositionierung des Wettbewerbers in 1-2 Sätzen.
3. **strengths_vs_weaknesses**: Formuliere 3-4 prägnante Stichpunkte. Jeder Stichpunkt soll eine Stärke von "{request.company.name}" einer vermuteten Schwäche des Wettbewerbers gegenüberstellen. Beginne die Sätze z.B. mit "Während [Wettbewerber]..., bieten wir...".
4. **landmine_questions**: Formuliere 3-5 intelligente, offene "Landminen"-Fragen, die ein Vertriebsmitarbeiter einem potenziellen Kunden stellen kann. Diese Fragen sollen die Schwächen des Wettbewerbers aufdecken oder die Stärken von "{request.company.name}" betonen, ohne den Wettbewerber direkt anzugreifen.
5. **silver_bullet**: Übernimm die bereits formulierte "Silver Bullet" für diesen Wettbewerber.
Regeln:
- Sei präzise, überzeugend und nutze eine aktive, vertriebsorientierte Sprache.
- Die "landmine_questions" müssen so formuliert sein, dass sie den Kunden zum Nachdenken anregen und ihn in Richtung der Vorteile von "{request.company.name}" lenken.
- Antworte ausschließlich im JSON-Format gemäß dem vorgegebenen Schema für ein Array von Battlecards.
",
"en": f"""
Role: Sales Strategist and Coach for the B2B software company "{request.company.name}".
Task: Create a detailed "Sales Battlecard" for each of the following competitors. This battlecard should provide the sales team with concrete, actionable arguments for customer conversations.
Information about our own company: {request.company.name}
Detailed analysis of competitors:
{competitor_data_summary}
For EACH competitor, create the following sections of a battlecard in JSON format:
1. **competitor_name**: The name of the competitor.
2. **competitor_profile**:
* **focus**: Summarize the competitor's core focus (products & industries) in one sentence.
* **positioning**: Describe the competitor's core positioning in 1-2 sentences.
3. **strengths_vs_weaknesses**: Formulate 3-4 concise bullet points. Each point should contrast a strength of "{request.company.name}" with a presumed weakness of the competitor. Start sentences with, for example, "While [Competitor]..., we offer...".
4. **landmine_questions**: Formulate 3-5 intelligent, open-ended "landmine" questions that a sales representative can ask a potential customer. These questions should uncover the competitor's weaknesses or emphasize the strengths of "{request.company.name}" without attacking the competitor directly.
5. **silver_bullet**: Use the "Silver Bullet" already formulated for this competitor.
Rules:
- Be precise, persuasive, and use active, sales-oriented language.
- The "landmine_questions" must be formulated to make the customer think and guide them towards the advantages of "{request.company.name}".
- Respond exclusively in JSON format according to the specified schema for an array of battlecards.
"
}
response_schema = {
"type": "object",
"properties": {
"battlecards": {
"type": "array",
"items": {
"type": "object",
"properties": {
"competitor_name": {"type": "string"},
"competitor_profile": {
"type": "object",
"properties": {
"focus": {"type": "string"},
"positioning": {"type": "string"}
},
"required": ['focus', 'positioning']
},
"strengths_vs_weaknesses": {"type": "array", "items": {"type": "string"}},
"landmine_questions": {"type": "array", "items": {"type": "string"}},
"silver_bullet": {"type": "string"}
},
"required": ['competitor_name', 'competitor_profile', 'strengths_vs_weaknesses', 'landmine_questions', 'silver_bullet']
}
}
},
"required": ['battlecards']
}
try:
response = await model.generate_content_async(
prompts[request.language],
generation_config=genai.GenerationConfig(
response_mime_type="application/json",
response_schema=response_schema
),
safety_settings={
HarmCategory.HARM_CATEGORY_HARASSMENT: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_HATE_SPEECH: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: HarmBlockThreshold.BLOCK_NONE,
},
)
return parse_json_response(response.text)
except Exception as e:
print(f"Error in fetch_step7_data_battlecards: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/fetchStep8Data_ReferenceAnalysis")
async def fetch_step8_data_reference_analysis(request: FetchStep8DataReferenceAnalysisRequest):
competitors_summary = '\n'.join([f'- {c.name}: {c.url}' for c in request.competitors])
prompts = {
"de": """
Rolle: Faktentreuer Research-Agent. Deine Antworten MÜSSEN ausschließlich auf den Ergebnissen der von dir durchgeführten Websuche basieren.
Aufgabe: Führe für jeden der folgenden Wettbewerber eine Websuche durch, um deren offizielle Referenzkunden, Case Studies oder Success Stories zu finden.
Wettbewerber:
{competitors_summary_de}
ABLAUF FÜR JEDEN WETTBEWERBER:
1. **SUCHE**: Führe eine gezielte Suche durch mit Phrasen wie "[Wettbewerber-Name] Referenzen", "[Wettbewerber-Name] Case Studies", "[Wettbewerber-Name] Kunden".
2. **VALIDIERUNG**: Analysiere die Suchergebnisse. Konzentriere dich AUSSCHLIESSLICH auf Links, die zur offiziellen Domain des Wettbewerbers gehören (z.B. `wettbewerber.com/referenzen`). Ignoriere Pressemitteilungen auf Drittseiten, Partnerlisten oder Nachrichtenartikel.
3. **EXTRAKTION**: Extrahiere die geforderten Informationen NUR von diesen validierten, offiziellen Seiten.
SEHR WICHTIGE REGELN ZUR VERMEIDUNG VON FALSCHINFORMATIONEN:
- **NUR GEFUNDENE DATEN**: Gib NUR Kunden an, für die du eine dedizierte Case-Study-Seite oder einen klaren Testimonial-Abschnitt auf der OFFIZIELLEN Website des Wettbewerbers gefunden hast.
- **KEINE HALLUZINATION**: Erfinde KEINE Kunden, Branchen oder Zitate. Wenn du für einen Wettbewerber absolut nichts findest, gib ein leeres "references" Array zurück. Dies ist besser als falsche Informationen.
- **DIREKTER LINK**: Die 'case_study_url' MUSS der exakte, funktionierende Link zur Seite sein, auf der die Informationen gefunden wurden.
Antworte AUSSCHLIESSLICH im JSON-Format, eingeschlossen in einem Markdown-Codeblock.
Extrahiere für JEDEN GEFUNDENEN UND VERIFIZIERTEN Referenzkunden (max. 5 pro Wettbewerber) die geforderten Felder.
".format(
competitors_summary_de=competitors_summary
),
"en": """
Role: Fact-based Research Agent. Your answers MUST be based solely on the results of the web search you perform.
Task: For each of the following competitors, conduct a web search to find their official reference customers, case studies, or success stories.
Competitors:
{competitors_summary_en}
PROCESS FOR EACH COMPETITOR:
1. **SEARCH**: Conduct a targeted search with phrases like "[Competitor Name] references", "[Competitor Name] case studies", "[Competitor Name] customers".
2. **VALIDATION**: Analyze the search results. Focus EXCLUSIVELY on links that belong to the competitor's official domain (e.g., `competitor.com/references`). Ignore press releases on third-party sites, partner lists, or news articles.
3. **EXTRACTION**: Extract the required information ONLY from these validated, official pages.
VERY IMPORTANT RULES TO AVOID MISINFORMATION:
- **ONLY FOUND DATA**: ONLY list customers for whom you have found a dedicated case study page or a clear testimonial section on the OFFICIAL website of the competitor.
- **NO HALLUCINATION**: DO NOT invent customers, industries, or quotes. If you find absolutely nothing for a competitor, return an empty "references" array. This is better than false information.
- **DIRECT LINK**: The 'case_study_url' MUST be the exact, working link to the page where the information was found.
Respond EXCLUSIVELY in JSON format, enclosed in a markdown code block.
For EACH FOUND AND VERIFIED reference customer (max 5 per competitor), extract the required fields.
".format(
competitors_summary_en=competitors_summary
)
}
response_schema = {
"type": "object",
"properties": {
"reference_analysis": {
"type": "array",
"items": {
"type": "object",
"properties": {
"competitor_name": {"type": "string"},
"references": {
"type": "array",
"items": {
"type": "object",
"properties": {
"name": {"type": "string"},
"industry": {"type": "string"},
"testimonial_snippet": {"type": "string"},
"case_study_url": {"type": "string"}
},
"required": ["name", "industry", "testimonial_snippet", "case_study_url"]
}
}
},
"required": ["competitor_name", "references"]
}
}
},
"required": ['reference_analysis']
}
try:
response = await model.generate_content_async(
prompts[request.language],
generation_config=genai.GenerationConfig(
response_mime_type="application/json",
response_schema=response_schema
),
tools=[genai.types.Tool(google_search_retrieval=genai.types.GoogleSearchRetrieval())], # Correct way to enable search in Python SDK
safety_settings={
HarmCategory.HARM_CATEGORY_HARASSMENT: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_HATE_SPEECH: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: HarmBlockThreshold.BLOCK_NONE,
},
)
grounding_metadata = [chunk.to_dict() for chunk in response.candidates[0].grounding_metadata.grounding_chunks] if response.candidates[0].grounding_metadata else []
parsed_data = parse_json_response(response.text)
return {**parsed_data, "groundingMetadata": grounding_metadata}
except Exception as e:
print(f"Error in fetch_step8_data_reference_analysis: {e}")
raise HTTPException(status_code=500, detail=str(e))
# Mount static files AFTER all API routes
if os.path.exists("dist"):
app.mount("/", StaticFiles(directory="dist", html=True), name="static")
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)