Spaces:
Sleeping
Sleeping
| 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 |