File size: 5,147 Bytes
0eb4b73 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 | 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)
# Global rate limiter for Groq
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)
# Enhance with extra fields
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 |