| import os
|
| import time
|
| import json
|
| import logging
|
| import re
|
| from typing import Dict, Any
|
| from groq import Groq
|
| from reddit_client import RateLimiter
|
| from config import GROQ_API_KEY, GROQ_MODEL, GROQ_RPS
|
|
|
| logger = logging.getLogger(__name__)
|
| groq_client = Groq(api_key=GROQ_API_KEY, max_retries=0)
|
|
|
| SYSTEM_PROMPT = """You are an expert startup analyst. Analyze the Reddit post below.
|
| Return ONLY valid JSON in this exact schema (no extra text):
|
|
|
| {
|
| "burning_pain": "string",
|
| "customer_type": "string",
|
| "problem_severity_score": integer 1-10,
|
| "competitor_weakness": "string",
|
| "revenue_opportunity": "string",
|
| "market_timing_score": integer 1-10,
|
| "startup_idea": "string",
|
| "difficulty_to_build": integer 1-10,
|
| "potential_pricing_model": "string",
|
| "confidence_score": integer 1-100
|
| }
|
| """
|
|
|
| def estimate_tokens(text: str) -> int:
|
| return len(text) // 4
|
|
|
| def detect_pain_intensity(text: str) -> int:
|
| urgent_words = ["urgent", "critical", "desperate", "wasted", "nothing works", "broken", "frustrated"]
|
| score = 1
|
| for w in urgent_words:
|
| if w in text.lower():
|
| score += 2
|
| return min(score, 10)
|
|
|
| def detect_founder_persona(text: str) -> str:
|
| personas = {
|
| "indie_hacker": ["indie", "solo", "bootstrapping", "side project"],
|
| "startup_founder": ["co-founder", "funding", "investor", "pitch", "scale"],
|
| "developer": ["api", "code", "build", "stack", "github"],
|
| "product_manager": ["roadmap", "feature", "user story", "product"],
|
| "enterprise_buyer": ["enterprise", "b2b", "sales cycle", "procurement"]
|
| }
|
| text_lower = text.lower()
|
| for persona, keywords in personas.items():
|
| if any(k in text_lower for k in keywords):
|
| return persona
|
| return "unknown"
|
|
|
| def estimate_build_difficulty(text: str, insight: Dict) -> int:
|
| difficulty = insight.get("difficulty_to_build", 5)
|
| if any(term in text.lower() for term in ["ai", "machine learning", "blockchain", "hardware"]):
|
| difficulty += 2
|
| if any(term in text.lower() for term in ["api integration", "multi-tenant", "compliance"]):
|
| difficulty += 1
|
| return min(difficulty, 10)
|
|
|
| def detect_investor_signal(text: str) -> bool:
|
| signals = ["funding", "investment", "vc", "venture capital", "angel", "seed round", "series a"]
|
| return any(s in text.lower() for s in signals)
|
|
|
| def detect_early_startup(text: str) -> bool:
|
| phrases = ["i'm building", "working on", "mvp launching", "just launched", "startup idea"]
|
| return any(p in text.lower() for p in phrases)
|
|
|
|
|
| groq_limiter = RateLimiter(GROQ_RPS)
|
|
|
| def call_groq_for_insight(reddit_text: str) -> Dict[str, Any]:
|
| truncated = reddit_text[:2000]
|
| input_tokens = estimate_tokens(SYSTEM_PROMPT) + estimate_tokens(truncated)
|
| output_tokens = 300
|
| total_tokens = input_tokens + output_tokens
|
| required_delay = total_tokens / 300.0
|
| now = time.time()
|
| elapsed = now - groq_limiter.last_request
|
| if elapsed < required_delay:
|
| time.sleep(required_delay - elapsed)
|
| groq_limiter.last_request = time.time()
|
|
|
| max_retries = 2
|
| for attempt in range(max_retries+1):
|
| try:
|
| response = groq_client.chat.completions.create(
|
| model=GROQ_MODEL,
|
| messages=[
|
| {"role": "system", "content": SYSTEM_PROMPT},
|
| {"role": "user", "content": f"Reddit Post:\n{truncated}"}
|
| ],
|
| temperature=0.3,
|
| max_tokens=500,
|
| response_format={"type": "json_object"}
|
| )
|
| raw = response.choices[0].message.content.strip()
|
| if raw.startswith("```json"): raw = raw[7:]
|
| if raw.endswith("```"): raw = raw[:-3]
|
| insight = json.loads(raw)
|
|
|
| insight["pain_intensity"] = detect_pain_intensity(truncated)
|
| insight["founder_persona"] = detect_founder_persona(truncated)
|
| insight["build_difficulty_est"] = estimate_build_difficulty(truncated, insight)
|
| insight["investor_signal"] = detect_investor_signal(truncated)
|
| insight["early_startup_signal"] = detect_early_startup(truncated)
|
| return insight
|
| except Exception as e:
|
| logger.error(f"Groq error attempt {attempt+1}: {e}")
|
| if attempt < max_retries:
|
| time.sleep(2 ** attempt)
|
| fallback = {
|
| "burning_pain": "Error",
|
| "customer_type": "Error",
|
| "problem_severity_score": 5,
|
| "competitor_weakness": "Error",
|
| "revenue_opportunity": "Error",
|
| "market_timing_score": 5,
|
| "startup_idea": "Error",
|
| "difficulty_to_build": 5,
|
| "potential_pricing_model": "Error",
|
| "confidence_score": 0,
|
| "pain_intensity": 0,
|
| "founder_persona": "unknown",
|
| "build_difficulty_est": 5,
|
| "investor_signal": False,
|
| "early_startup_signal": False
|
| }
|
| return fallback |
