466 lines
26 KiB
Python
466 lines
26 KiB
Python
import os
|
|
import json
|
|
import asyncio
|
|
import logging
|
|
import time
|
|
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
|
|
from urllib.parse import urljoin, urlparse
|
|
|
|
# --- DEPENDENCIES ---
|
|
import requests
|
|
from bs4 import BeautifulSoup
|
|
from serpapi import GoogleSearch
|
|
|
|
# --- DUAL SDK IMPORTS ---
|
|
try:
|
|
from google import genai
|
|
from google.genai import types
|
|
HAS_NEW_GENAI = True
|
|
logging.info("✅ SUCCESS: Loaded 'google-genai' SDK.")
|
|
except ImportError:
|
|
HAS_NEW_GENAI = False
|
|
logging.warning("⚠️ WARNING: 'google-genai' not found. Fallback.")
|
|
|
|
try:
|
|
import google.generativeai as old_genai
|
|
HAS_OLD_GENAI = True
|
|
logging.info("✅ SUCCESS: Loaded legacy 'google.generativeai' SDK.")
|
|
except ImportError:
|
|
HAS_OLD_GENAI = False
|
|
logging.warning("⚠️ WARNING: Legacy 'google.generativeai' not found.")
|
|
|
|
# --- ENV & LOGGING SETUP ---
|
|
load_dotenv()
|
|
API_KEY = os.getenv("GEMINI_API_KEY") or (open("/app/gemini_api_key.txt").read().strip() if os.path.exists("/app/gemini_api_key.txt") else None)
|
|
SERPAPI_KEY = os.getenv("SERPAPI_KEY")
|
|
if not API_KEY: raise ValueError("GEMINI_API_KEY not set.")
|
|
if HAS_OLD_GENAI: old_genai.configure(api_key=API_KEY)
|
|
|
|
os.makedirs("/app/Log_from_docker", exist_ok=True)
|
|
logging.basicConfig(level=logging.DEBUG, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[logging.FileHandler("/app/Log_from_docker/competitor_analysis_debug.log"), logging.StreamHandler()], force=True)
|
|
|
|
app = FastAPI()
|
|
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"])
|
|
|
|
# --- V5 STRATEGY: CORE LOGIC ---
|
|
|
|
CANONICAL_PRODUCT_MASTER_LIST = {
|
|
"Pudu": ["BellaBot", "KettyBot", "HolaBot", "PuduBot 2", "SwiftBot", "FlashBot", "Pudu CC1", "Pudu SH1"],
|
|
"Gausium": ["Scrubber 50 Pro", "Scrubber 75", "Vacuum 40", "Phantas", "Sweeper 111"],
|
|
"Keenon": ["DINERBOT T1", "DINERBOT T2", "DINERBOT T5", "DINERBOT T6", "BUTLERBOT W3", "GUIDERBOT G2"],
|
|
"Lionsbot": ["LeoBot", "Rex"],
|
|
"Nexaro": ["Nexaro NR 1500"]
|
|
}
|
|
|
|
def scrape_text_from_url(url: str) -> str:
|
|
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, verify=False)
|
|
response.raise_for_status()
|
|
soup = BeautifulSoup(response.content, 'html.parser')
|
|
for element in soup(['script', 'style', 'nav', 'footer', 'aside', 'header']): element.decompose()
|
|
return ' '.join(soup.stripped_strings)
|
|
except Exception as e:
|
|
logging.warning(f"Failed to scrape {url}: {e}")
|
|
return ""
|
|
|
|
async def discover_and_scrape_website(start_url: str, keywords: List[str], manual_urls: List[str] = None) -> str:
|
|
logging.info(f"Scraping {start_url} with manual URLs: {manual_urls}")
|
|
urls_to_scrape = {start_url} if start_url else set()
|
|
|
|
# Add manual URLs first (high priority)
|
|
if manual_urls:
|
|
for url in manual_urls:
|
|
urls_to_scrape.add(url)
|
|
|
|
if start_url:
|
|
try:
|
|
base_domain = urlparse(start_url).netloc
|
|
r = requests.get(start_url, timeout=10, verify=False)
|
|
soup = BeautifulSoup(r.content, 'html.parser')
|
|
for a in soup.find_all('a', href=True):
|
|
href = a['href']
|
|
link_text = a.get_text().lower()
|
|
if any(k in href.lower() or k in link_text for k in keywords):
|
|
full_url = urljoin(start_url, href)
|
|
if urlparse(full_url).netloc == base_domain: urls_to_scrape.add(full_url)
|
|
except Exception as e:
|
|
logging.error(f"Failed to get links from {start_url}: {e}")
|
|
|
|
urls_list = list(urls_to_scrape)[:8]
|
|
tasks = [asyncio.to_thread(scrape_text_from_url, url) for url in urls_list]
|
|
scraped_contents = await asyncio.gather(*tasks)
|
|
return "\n\n---".join(c for c in scraped_contents if c)[:60000]
|
|
|
|
def parse_json_response(response_text: str) -> Any:
|
|
try:
|
|
if not response_text: return {}
|
|
cleaned_text = response_text.strip()
|
|
if cleaned_text.startswith("```"):
|
|
lines = cleaned_text.splitlines()
|
|
if lines[0].lower().startswith("```json"): lines = lines[1:]
|
|
if lines[-1].startswith("```"): lines = lines[:-1]
|
|
cleaned_text = "\n".join(lines).strip()
|
|
result = json.loads(cleaned_text)
|
|
return result[0] if isinstance(result, list) and result else result
|
|
except Exception as e:
|
|
logging.error(f"CRITICAL: Failed to parse JSON. Error: {e}\nRaw Text: {response_text[:500]}")
|
|
return {}
|
|
|
|
async def call_gemini_robustly(prompt: str, schema: dict):
|
|
last_err = None
|
|
if HAS_OLD_GENAI:
|
|
try:
|
|
logging.debug("Attempting Legacy SDK with gemini-2.0-flash (as per project conventions)")
|
|
gen_config = {"temperature": 0.2, "response_mime_type": "application/json", "max_output_tokens": 8192}
|
|
if schema: gen_config["response_schema"] = schema
|
|
model = old_genai.GenerativeModel('gemini-2.0-flash', generation_config=gen_config)
|
|
response = await model.generate_content_async(prompt)
|
|
return parse_json_response(response.text)
|
|
except Exception as e:
|
|
last_err = e
|
|
logging.warning(f"Legacy SDK failed: {e}")
|
|
|
|
if HAS_NEW_GENAI:
|
|
try:
|
|
logging.debug("Attempting Modern SDK with gemini-1.5-flash as fallback")
|
|
client_new = genai.Client(api_key=API_KEY)
|
|
config_dict = { "temperature": 0.2, "max_output_tokens": 8192, "response_mime_type": "application/json" }
|
|
if schema: config_dict["response_schema"] = schema
|
|
generation_config = types.GenerationConfig(**config_dict)
|
|
response = await client_new.models.generate_content(
|
|
model='gemini-1.5-flash',
|
|
contents=prompt,
|
|
generation_config=generation_config
|
|
)
|
|
return parse_json_response(response.text)
|
|
except Exception as e:
|
|
logging.error(f"Modern SDK fallback failed: {e}")
|
|
raise HTTPException(status_code=500, detail=str(last_err or e))
|
|
|
|
raise HTTPException(status_code=500, detail=f"All Gemini SDKs failed. Last error: {last_err}")
|
|
|
|
def clean_raw_product_list(raw_list: List[str]) -> List[str]:
|
|
cleaned = set()
|
|
for item in raw_list:
|
|
parts = [p.strip() for p in item.replace(' und ', ',').replace(';', ',').split(',')]
|
|
for part in parts:
|
|
if part: cleaned.add(part)
|
|
return sorted(list(cleaned))
|
|
|
|
async def extract_raw_data_phase1(competitor: Any, my_company: Any) -> Optional[Dict]:
|
|
c_name = competitor.get('name', 'Unknown')
|
|
c_url = competitor.get('url', '')
|
|
manual_urls = competitor.get('manual_urls', [])
|
|
|
|
logging.debug(f"➡️ [P1] Start: {c_name}")
|
|
content = await discover_and_scrape_website(c_url, ['product', 'solution', 'roboter', 'portfolio'], manual_urls)
|
|
context_text = content if content else "No website data."
|
|
|
|
product_prompt = f"Extract all specific product names from this text. Ignore general categories. TEXT: {context_text}"
|
|
product_schema = {"type": "object", "properties": {"products": {"type": "array", "items": {"type": "string"}}}, "required": ["products"]}
|
|
|
|
profile_prompt = f"Analyze competitor '{c_name}' based on this text. Focus on strategy (target industries, delivery model, differentiators), not a list of products. TEXT: {context_text}"
|
|
profile_schema = {"type": "object", "properties": {"target_industries": {"type": "array", "items": {"type": "string"}},"delivery_model": {"type": "string"},"differentiators": {"type": "array", "items": {"type": "string"}},"overlap_score": {"type": "integer"}},"required": ['target_industries', 'delivery_model', 'differentiators', 'overlap_score']}
|
|
|
|
try:
|
|
product_task = call_gemini_robustly(product_prompt, product_schema)
|
|
profile_task = call_gemini_robustly(profile_prompt, profile_schema)
|
|
product_result, profile_result = await asyncio.gather(product_task, profile_task)
|
|
|
|
if not profile_result: return None
|
|
|
|
cleaned_products = clean_raw_product_list(product_result.get('products', []) if product_result else [])
|
|
logging.debug(f"✅ [P1] OK: {c_name} ({len(cleaned_products)} products)")
|
|
return {"competitor": {"name": c_name, "url": c_url},"cleaned_products": cleaned_products,"profile": profile_result,"raw_text": context_text}
|
|
except Exception as e:
|
|
logging.error(f"❌ [P1] Fail: {c_name}: {e}")
|
|
return None
|
|
|
|
async def enrich_product_details_phase3(product_name: str, context_text: str) -> Dict:
|
|
logging.debug(f" [P3] Enrich: {product_name} (CoT)")
|
|
prompt = f"""Analyze the product '{product_name}' based on the provided text.
|
|
|
|
TEXT:
|
|
{context_text}
|
|
|
|
STANDARD CATEGORIES:
|
|
- \"Cleaning (Indoor)\"\n- \"Cleaning (Outdoor)\"\n- \"Transport/Logistics\"\n- \"Service/Gastro\"\n- \"Security/Inspection\"\n- \"Software/Fleet Mgmt\"\n- \"Other\"
|
|
|
|
INSTRUCTIONS:
|
|
1. Scan the text for all mentions of '{product_name}'.
|
|
2. Synthesize a detailed description of its purpose ("purpose"). What does it do? Who is it for? Be specific and descriptive (2-3 sentences).
|
|
3. Determine the best fitting category from the list above.
|
|
|
|
Output the result as a single JSON object.
|
|
"""
|
|
schema = {"type": "object", "properties": {"product": {"type": "string"},"purpose": {"type": "string"},"category": {"type": "string", "enum": ["Cleaning (Indoor)", "Cleaning (Outdoor)", "Transport/Logistics", "Service/Gastro", "Security/Inspection", "Software/Fleet Mgmt", "Other"]}},"required": ["product", "purpose", "category"]}
|
|
try:
|
|
result = await call_gemini_robustly(prompt, schema)
|
|
return result if result and result.get('product') else {"product": product_name, "purpose": "N/A", "category": "Other"}
|
|
except Exception:
|
|
return {"product": product_name, "purpose": "Error", "category": "Other"}
|
|
|
|
async def analyze_single_competitor_references(competitor: Any) -> Optional[Dict]:
|
|
c_name, c_url = competitor.get('name', 'Unknown'), competitor.get('url', '')
|
|
logging.debug(f"➡️ [Ref] Analyzing references for: {c_name}")
|
|
content = await discover_and_scrape_website(c_url, ['referenz', 'kunde', 'case', 'erfolg'])
|
|
context_text = content if content else "No reference data."
|
|
prompt = f"Extract reference customers from this text for '{c_name}'. If no specific names, describe typical customer profiles. TEXT: {context_text}"
|
|
schema = {"type": "object","properties": {"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"]}}},"required": ["references"]}
|
|
try:
|
|
result = await call_gemini_robustly(prompt, schema)
|
|
return {"competitor_name": c_name, "references": result.get('references', [])} if result else {"competitor_name": c_name, "references": []}
|
|
except Exception as e:
|
|
logging.error(f"❌ [Ref] Fail: {c_name}: {e}")
|
|
return {"competitor_name": c_name, "references": []}
|
|
|
|
# --- FastAPI Models ---
|
|
class ProductDetailsRequest(BaseModel): name: str; url: str; language: str
|
|
class FetchStep1DataRequest(BaseModel): start_url: str; language: str
|
|
class FetchStep2DataRequest(BaseModel): products: List[Any]; industries: List[Any]; language: str
|
|
class FetchStep3DataRequest(BaseModel): keywords: List[Any]; market_scope: str; language: str
|
|
class StepRequest(BaseModel):
|
|
company: Any = {}
|
|
competitors: List[Any] = []
|
|
analyses: List[Any] = []
|
|
products: List[Any] = []
|
|
industries: List[Any] = []
|
|
silver_bullets: List[Any] = []
|
|
class ReanalyzeRequest(BaseModel):
|
|
company: Any
|
|
competitor: Any
|
|
manual_urls: List[str]
|
|
|
|
# --- Endpoints ---
|
|
|
|
# Step 0: Product Details
|
|
@app.post("/api/fetchProductDetails")
|
|
async def fetch_product_details(request: ProductDetailsRequest):
|
|
prompt = f"Analysiere die URL {request.url} und beschreibe den Zweck von '{request.name}' in 1-2 Sätzen. Antworte JSON."
|
|
schema = {"type": "object", "properties": {"name": {"type": "string"}, "purpose": {"type": "string"}, "evidence": {"type": "array", "items": {"type": "object", "properties": {"url": {"type": "string"}, "snippet": {"type": "string"}}, "required": ['url', 'snippet']}}}, "required": ['name', 'purpose', 'evidence']}
|
|
return await call_gemini_robustly(prompt, schema)
|
|
|
|
# Step 1: Extraction
|
|
@app.post("/api/fetchStep1Data")
|
|
async def fetch_step1_data(request: FetchStep1DataRequest):
|
|
grounding_text = await discover_and_scrape_website(request.start_url, ['product', 'solution', 'roboter', 'portfolio'])
|
|
prompt = f"Extrahiere Hauptprodukte und Zielbranchen aus dem Text. TEXT: {grounding_text}"
|
|
schema = {"type": "object", "properties": {"products": {"type": "array", "items": {"type": "object", "properties": {"name": {"type": "string"}, "purpose": {"type": "string"}, "evidence": {"type": "array", "items": {"type": "object", "properties": {"url": {"type": "string"}, "snippet": {"type": "string"}}, "required": ['url', 'snippet']}}}, "required": ['name', 'purpose', 'evidence']}}, "target_industries": {"type": "array", "items": {"type": "object", "properties": {"name": {"type": "string"}, "evidence": {"type": "array", "items": {"type": "object", "properties": {"url": {"type": "string"}, "snippet": {"type": "string"}}, "required": ['url', 'snippet']}}}, "required": ['name', 'evidence']}}}, "required": ['products', 'target_industries']}
|
|
return await call_gemini_robustly(prompt, schema)
|
|
|
|
# Step 2: Keywords
|
|
@app.post("/api/fetchStep2Data")
|
|
async def fetch_step2_data(request: FetchStep2DataRequest):
|
|
p_names = [p.get('name') if isinstance(p, dict) else getattr(p, 'name', str(p)) for p in request.products]
|
|
prompt = f"Leite Keywords für Recherche ab: {', '.join(p_names)}"
|
|
schema = {"type": "object", "properties": {"keywords": {"type": "array", "items": {"type": "object", "properties": {"term": {"type": "string"}, "rationale": {"type": "string"}}, "required": ['term', 'rationale']}}}, "required": ['keywords']}
|
|
return await call_gemini_robustly(prompt, schema)
|
|
|
|
# Step 3: Competitors
|
|
@app.post("/api/fetchStep3Data")
|
|
async def fetch_step3_data(request: FetchStep3DataRequest):
|
|
k_terms = [k.get('term') if isinstance(k, dict) else getattr(k, 'term', str(k)) for k in request.keywords]
|
|
prompt = f"Finde Wettbewerber für Markt {request.market_scope} basierend auf: {', '.join(k_terms)}"
|
|
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": {"type": "object", "properties": {"url": {"type": "string"}, "snippet": {"type": "string"}}, "required": ['url', 'snippet']}}}, "required": ['name', 'url', 'confidence', 'why', 'evidence']}}}, "required": ['competitor_candidates']}
|
|
return await call_gemini_robustly(prompt, schema)
|
|
|
|
@app.post("/api/fetchStep4Data")
|
|
async def fetch_step4_data(request: StepRequest):
|
|
logging.info("=== V5 PIPELINE START ===")
|
|
phase1_results = await asyncio.gather(*[extract_raw_data_phase1(c, request.company) for c in request.competitors])
|
|
valid_phase1 = [r for r in phase1_results if r]
|
|
if not valid_phase1: raise HTTPException(500, "P1 failed for all.")
|
|
|
|
global_products = {p for r in valid_phase1 for p in r['cleaned_products']}
|
|
canon_prompt = f"""Du bist ein Daten-Normalisierer. Ordne die rohen Produktnamen den kanonischen Namen aus der Grounded Truth zu.
|
|
|
|
GROUNDED TRUTH (Hersteller-Masterliste):
|
|
{json.dumps(CANONICAL_PRODUCT_MASTER_LIST, indent=2)}
|
|
|
|
ROHE PRODUKT-ERWÄHNUNGEN:
|
|
{json.dumps(list(global_products))}
|
|
|
|
AUFGABE:
|
|
Antworte mit einer JSON-Liste von Objekten. Jedes Objekt soll einen kanonischen Namen und seine gefundenen Variationen enthalten.
|
|
"""
|
|
canon_schema = {"type": "object","properties": {"mapping": {"type": "array","items": {"type": "object","properties": {"canonical_name": {"type": "string"},"variations": {"type": "array", "items": {"type": "string"}}},"required": ["canonical_name", "variations"]}}},"required": ["mapping"]}
|
|
canon_result = await call_gemini_robustly(canon_prompt, canon_schema)
|
|
if not (canon_map_list := canon_result.get('mapping')): raise HTTPException(500, "P2 canonization failed.")
|
|
|
|
inverted_map = {raw: item['canonical_name'] for item in canon_map_list for raw in item['variations']}
|
|
|
|
final_analyses = []
|
|
for comp_data in valid_phase1:
|
|
can_prods = {inverted_map.get(p) for p in comp_data['cleaned_products'] if inverted_map.get(p)}
|
|
enriched_portfolio = await asyncio.gather(*[enrich_product_details_phase3(p, comp_data['raw_text']) for p in can_prods])
|
|
final_analyses.append({"competitor": comp_data['competitor'],"portfolio": enriched_portfolio,**comp_data['profile']})
|
|
logging.info("=== V5 PIPELINE COMPLETE ===")
|
|
return {"analyses": final_analyses}
|
|
|
|
@app.post("/api/reanalyzeCompetitor")
|
|
async def reanalyze_competitor(request: ReanalyzeRequest):
|
|
logging.info(f"=== RE-ANALYZING COMPETITOR: {request.competitor.get('name')} ===")
|
|
|
|
# 1. Update competitor object with new manual URLs
|
|
competitor_data = request.competitor
|
|
competitor_data['manual_urls'] = request.manual_urls
|
|
|
|
# 2. Run Phase 1 (Scraping & Raw Extraction) for just this competitor
|
|
phase1_result = await extract_raw_data_phase1(competitor_data, request.company)
|
|
if not phase1_result:
|
|
raise HTTPException(500, "Phase 1 failed during re-analysis.")
|
|
|
|
# 3. Phase 2 (Canonization) - We map just this competitor's products against the Master List
|
|
# Note: We don't have the global context of other competitors here, but mapping against
|
|
# the static CANONICAL_PRODUCT_MASTER_LIST is sufficient and robust.
|
|
raw_products = phase1_result['cleaned_products']
|
|
canon_prompt = f"""Du bist ein Daten-Normalisierer. Ordne die rohen Produktnamen den kanonischen Namen aus der Grounded Truth zu.
|
|
|
|
GROUNDED TRUTH (Hersteller-Masterliste):
|
|
{json.dumps(CANONICAL_PRODUCT_MASTER_LIST, indent=2)}
|
|
|
|
ROHE PRODUKT-ERWÄHNUNGEN:
|
|
{json.dumps(list(raw_products))}
|
|
|
|
AUFGABE:
|
|
Antworte mit einer JSON-Liste von Objekten. Jedes Objekt soll einen kanonischen Namen und seine gefundenen Variationen enthalten.
|
|
"""
|
|
canon_schema = {"type": "object","properties": {"mapping": {"type": "array","items": {"type": "object","properties": {"canonical_name": {"type": "string"},"variations": {"type": "array", "items": {"type": "string"}}},"required": ["canonical_name", "variations"]}}},"required": ["mapping"]}
|
|
canon_result = await call_gemini_robustly(canon_prompt, canon_schema)
|
|
if not (canon_map_list := canon_result.get('mapping')):
|
|
canon_map_list = [] # Fallback if empty
|
|
|
|
inverted_map = {raw: item['canonical_name'] for item in canon_map_list for raw in item['variations']}
|
|
|
|
# 4. Phase 3 (Enrichment)
|
|
can_prods = {inverted_map.get(p) for p in raw_products if inverted_map.get(p)}
|
|
enriched_portfolio = await asyncio.gather(*[enrich_product_details_phase3(p, phase1_result['raw_text']) for p in can_prods])
|
|
|
|
final_analysis = {
|
|
"competitor": phase1_result['competitor'],
|
|
"portfolio": enriched_portfolio,
|
|
**phase1_result['profile']
|
|
}
|
|
|
|
logging.info("=== RE-ANALYSIS COMPLETE ===")
|
|
return final_analysis
|
|
|
|
@app.post("/api/fetchStep5Data_SilverBullets")
|
|
async def fetch_step5_data_silver_bullets(request: StepRequest):
|
|
logging.info("=== V5 Step 5 START: Silver Bullets ===")
|
|
my_name = request.company.get('name', 'My Company')
|
|
lines = [f"- {a.get('competitor', {}).get('name', 'Unknown')}: {', '.join(a.get('profile', {}).get('differentiators', []))}" for a in request.analyses]
|
|
prompt = f"Create 'Silver Bullet' positioning statements for '{my_name}' against these competitors:\n" + "\n".join(lines)
|
|
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"]}
|
|
result = await call_gemini_robustly(prompt, schema)
|
|
logging.info("=== V5 Step 5 COMPLETE ===")
|
|
return result
|
|
|
|
@app.post("/api/fetchStep6Data_Conclusion")
|
|
async def fetch_step6_data_conclusion(request: StepRequest):
|
|
logging.info("=== V5 Step 6 FINAL START: Conclusion ===")
|
|
my_name = request.company.get('name', 'My Company')
|
|
|
|
# --- PART 1: Build Matrices in Python (Deterministic) ---
|
|
product_mapping_rules = {
|
|
"Reinigungsroboter": ["Cleaning (Indoor)", "Cleaning (Outdoor)"],
|
|
"Lieferroboter": ["Transport/Logistics"],
|
|
"Serviceroboter": ["Service/Gastro"]
|
|
}
|
|
competitor_category_map = {
|
|
a.get('competitor', {}).get('name'): set(p.get('category') for p in a.get('portfolio', []) if p.get('category'))
|
|
for a in request.analyses
|
|
}
|
|
competitor_industry_map = {
|
|
a.get('competitor', {}).get('name'): set(a.get('target_industries', [])) # Note: target_industries is at root level in V5 final structure
|
|
for a in request.analyses
|
|
}
|
|
competitor_names = [a.get('competitor', {}).get('name') for a in request.analyses]
|
|
|
|
product_matrix = []
|
|
for my_product in request.products:
|
|
product_name = my_product.get('name')
|
|
mapped_categories = product_mapping_rules.get(product_name, ["Other"])
|
|
availability = []
|
|
for comp_name in competitor_names:
|
|
comp_categories = competitor_category_map.get(comp_name, set())
|
|
has_offering = any(mc in comp_categories for mc in mapped_categories)
|
|
availability.append({"competitor": comp_name, "has_offering": has_offering})
|
|
product_matrix.append({"product": product_name, "availability": availability})
|
|
|
|
industry_matrix = []
|
|
for my_industry in request.industries:
|
|
industry_name = my_industry.get('name')
|
|
availability = []
|
|
for comp_name in competitor_names:
|
|
has_offering = industry_name in competitor_industry_map.get(comp_name, set())
|
|
availability.append({"competitor": comp_name, "has_offering": has_offering})
|
|
industry_matrix.append({"industry": industry_name, "availability": availability})
|
|
|
|
overlap_scores = [{"competitor": a.get('competitor', {}).get('name'), "score": a.get('overlap_score', 0)} for a in request.analyses]
|
|
|
|
logging.info("Python-side matrix generation complete.")
|
|
|
|
# --- PART 2: Call LLM for Summary ONLY ---
|
|
prompt = f"""As a strategy consultant, analyze the following market data for '{my_name}' and provide a strategic summary.
|
|
|
|
Product Competitive Matrix:
|
|
{json.dumps(product_matrix, indent=2)}
|
|
|
|
Industry Overlap Matrix:
|
|
{json.dumps(industry_matrix, indent=2)}
|
|
|
|
Task:
|
|
Based ONLY on the data above, provide a concise strategic summary.
|
|
- \"summary\": A brief overview of the competitive landscape.
|
|
- \"opportunities\": 2-3 actionable opportunities.
|
|
- \"next_questions\": 2-3 strategic questions.
|
|
"""
|
|
schema = {"type": "object","properties": {"summary": {"type": "string"},"opportunities": {"type": "string"},"next_questions": {"type": "array", "items": {"type": "string"}}},"required": ["summary", "opportunities", "next_questions"]}
|
|
|
|
summary_result = await call_gemini_robustly(prompt, schema)
|
|
if not summary_result: raise HTTPException(500, "Failed to generate summary from LLM.")
|
|
|
|
final_conclusion = {
|
|
"product_matrix": product_matrix,
|
|
"industry_matrix": industry_matrix,
|
|
"overlap_scores": overlap_scores,
|
|
**summary_result
|
|
}
|
|
logging.info("=== V5 Step 6 FINAL COMPLETE ===")
|
|
return {"conclusion": final_conclusion}
|
|
|
|
@app.post("/api/fetchStep7Data_Battlecards")
|
|
async def fetch_step7_data_battlecards(request: StepRequest):
|
|
logging.info("=== V5 Step 7 START: Battlecards ===")
|
|
my_name = request.company.get('name', 'My Company')
|
|
comp_context = [f"- {a.get('competitor', {}).get('name', 'Unknown')}: {', '.join(a.get('differentiators', [])[:3])}" for a in request.analyses]
|
|
bullets_context = [f"- {sb.get('competitor_name')}: {sb.get('statement')}" for sb in request.silver_bullets]
|
|
prompt = f"Create Sales Battlecards for '{my_name}' against competitors.\nCompetitors: {' '.join(comp_context)}\nBullets: {' '.join(bullets_context)}"
|
|
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"}}},"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"]}
|
|
result = await call_gemini_robustly(prompt, schema)
|
|
logging.info("=== V5 Step 7 COMPLETE ===")
|
|
return result
|
|
|
|
@app.post("/api/fetchStep8Data_ReferenceAnalysis")
|
|
async def fetch_step8_data_reference_analysis(request: StepRequest):
|
|
logging.info("=== V5 Step 8 START: References ===")
|
|
tasks = [analyze_single_competitor_references(c) for c in request.competitors]
|
|
results = await asyncio.gather(*tasks)
|
|
logging.info("=== V5 Step 8 COMPLETE ===")
|
|
return {"reference_analysis": [r for r in results if r]}
|
|
|
|
# Static Files
|
|
dist_path = os.path.join(os.getcwd(), "dist")
|
|
if os.path.exists(dist_path):
|
|
app.mount("/", StaticFiles(directory=dist_path, html=True), name="static")
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(app, host="0.0.0.0", port=8000) |