Spaces:

paulhemb
/

ConsumerIntelligence

Sleeping

Update main.py

d6bf870 verified 7 months ago

11.5 kB

from fastapi import FastAPI, HTTPException, Depends, BackgroundTasks
from fastapi.middleware.cors import CORSMiddleware
from datetime import datetime
from typing import List, Dict, Any
import logging

from auth import get_current_user
from models import (
    AnalysisRequest, AnalysisResponse, InsightResponse, QueryRequest, QueryResponse,
    MarketIntelligenceRequest, MarketIntelligenceResponse, GeneratePlanRequest, GeneratePlanResponse,
    TrendRequest, TrendResponse, UserSignup, UserLogin, UserProfile, SubscriptionPlan
)
from reddit_client import RedditClient
from analyzer import call_groq_for_insight
from serp_client import get_search_volume, search_competitors, search_funding_news
from supabase_client import supabase
from config import GROQ_MODEL, GROQ_API_KEY

# Groq client is imported in analyzer
# Reddit client
reddit_client = RedditClient()

app = FastAPI(title="Founder Intelligence Engine API", version="2.0")

# CORS – allow all origins for development; restrict in production
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

logger = logging.getLogger(__name__)

# ====================== Root ======================
@app.get("/")
async def root():
    return {"message": "Founder Intelligence Engine API is running"}

# ====================== Auth Endpoints ======================
@app.post("/auth/signup", response_model=UserProfile)
async def signup(user: UserSignup):
    try:
        response = supabase.auth.sign_up({
            "email": user.email,
            "password": user.password,
            "options": {"data": {"full_name": user.full_name}}
        })
        user_data = response.user
        # Create profile record
        profile = supabase.table("profiles").insert({
            "id": user_data.id,
            "email": user_data.email,
            "full_name": user.full_name,
            "subscription_tier": "free",
            "subscription_status": "active"
        }).execute()
        return UserProfile(
            id=user_data.id,
            email=user_data.email,
            full_name=user.full_name,
            subscription_tier="free",
            subscription_status="active",
            created_at=user_data.created_at
        )
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

@app.post("/auth/login")
async def login(creds: UserLogin):
    try:
        response = supabase.auth.sign_in_with_password({
            "email": creds.email,
            "password": creds.password
        })
        return {"access_token": response.session.access_token, "token_type": "bearer"}
    except Exception as e:
        raise HTTPException(status_code=401, detail="Invalid credentials")

@app.get("/auth/me", response_model=UserProfile)
async def get_me(user=Depends(get_current_user)):
    return UserProfile(
        id=user["id"],
        email=user["email"],
        full_name=user["profile"]["full_name"],
        subscription_tier=user["profile"]["subscription_tier"],
        subscription_status=user["profile"]["subscription_status"],
        created_at=user["profile"]["created_at"]
    )

# ====================== Subscription Plans ======================
@app.get("/plans", response_model=List[SubscriptionPlan])
async def list_plans():
    plans = supabase.table("plans").select("*").execute()
    return [SubscriptionPlan(**p) for p in plans.data]

# ====================== Helper: Check subscription limits ======================
def check_subscription_limit(user_id: str, requested_posts: int):
    """Raise 403 if user exceeds plan limits."""
    profile = supabase.table("profiles").select("subscription_tier").eq("id", user_id).execute()
    tier = profile.data[0]["subscription_tier"] if profile.data else "free"
    # Fetch plan features
    plan = supabase.table("plans").select("features").eq("name", tier.capitalize()).execute()
    if plan.data:
        features = plan.data[0]["features"]
        posts_limit = features.get("posts_per_month", 10)  # default free limit
        # Count posts this month for this user
        start_of_month = datetime.now().replace(day=1, hour=0, minute=0, second=0).isoformat()
        count_res = supabase.table("insights").select("id", count="exact").eq("user_id", user_id).gte("created_at", start_of_month).execute()
        current_count = count_res.count if hasattr(count_res, 'count') else 0
        if current_count + requested_posts > posts_limit:
            raise HTTPException(status_code=403, detail=f"Monthly post limit of {posts_limit} reached. Upgrade your plan.")
    # else no limit

# ====================== Analysis Endpoints ======================
@app.post("/analyze", response_model=AnalysisResponse)
async def analyze(request: AnalysisRequest, user=Depends(get_current_user)):
    user_id = user["id"]
    total_posts = len(request.subreddits) * request.posts_per_subreddit
    check_subscription_limit(user_id, total_posts)

    all_results = []
    for sub in request.subreddits:
        posts = reddit_client.get_top_posts(sub, limit=request.posts_per_subreddit, time_filter=request.time_filter)
        for post in posts:
            text = f"Title: {post.get('title','')}\nBody: {post.get('selftext','')[:1500]}"
            insight = call_groq_for_insight(text)
            # Compute opportunity score
            pain = insight.get("problem_severity_score", 5)
            timing = insight.get("market_timing_score", 5)
            rev = len(insight.get("revenue_opportunity", "")) / 20
            comp = 10 - len(insight.get("competitor_weakness", "")) / 20
            opp_score = int((pain + timing + rev + comp) / 4 * 10)
            insight["opportunity_score"] = min(100, max(0, opp_score))

            result = {
                "user_id": user_id,
                "subreddit": sub,
                "post_title": post.get("title"),
                "post_url": post.get("url"),
                "timestamp": datetime.utcnow().isoformat(),
                "insight_data": insight,
                "opportunity_score": insight["opportunity_score"]
            }
            # Insert into Supabase
            supabase.table("insights").insert(result).execute()
            all_results.append({
                "subreddit": sub,
                "post_title": post.get("title"),
                "post_url": post.get("url"),
                "timestamp": result["timestamp"],
                "model": GROQ_MODEL,
                "insight": insight
            })
    return AnalysisResponse(status="success", message=f"Analyzed {len(all_results)} posts", insights=all_results)

@app.get("/insights", response_model=List[InsightResponse])
async def get_insights(user=Depends(get_current_user)):
    res = supabase.table("insights").select("*").eq("user_id", user["id"]).order("created_at", desc=True).execute()
    insights = []
    for row in res.data:
        insights.append(InsightResponse(
            id=row["id"],
            subreddit=row["subreddit"],
            post_title=row["post_title"],
            post_url=row["post_url"],
            timestamp=row["timestamp"],
            model=GROQ_MODEL,
            insight=row["insight_data"]
        ))
    return insights

# ====================== Query / RAG ======================
@app.post("/query", response_model=QueryResponse)
async def query_insights(request: QueryRequest, user=Depends(get_current_user)):
    # Simple keyword search over user's insights (full text search on insight_data)
    insights = supabase.table("insights").select("*").eq("user_id", user["id"]).execute()
    relevant = []
    for row in insights.data:
        data = row["insight_data"]
        text = f"{data.get('burning_pain','')} {data.get('startup_idea','')} {data.get('customer_type','')}"
        if request.query.lower() in text.lower():
            relevant.append(row)
    # If no exact matches, return top N by opportunity_score
    if not relevant:
        relevant = sorted(insights.data, key=lambda x: x["opportunity_score"], reverse=True)[:request.n_results]
    else:
        relevant = relevant[:request.n_results]

    if not relevant:
        raise HTTPException(status_code=404, detail="No relevant insights found")

    # Build context
    context = "\n\n".join([f"From r/{r['subreddit']}: {r['insight_data'].get('burning_pain','')} {r['insight_data'].get('startup_idea','')}" for r in relevant])
    # Call Groq to answer
    from groq import Groq
    groq = Groq(api_key=GROQ_API_KEY)
    try:
        response = groq.chat.completions.create(
            model=GROQ_MODEL,
            messages=[
                {"role": "system", "content": "You are a startup advisor. Use the provided Reddit insights to answer the user's question concisely."},
                {"role": "user", "content": f"Context:\n{context}\n\nQuestion: {request.query}"}
            ],
            temperature=0.5
        )
        answer = response.choices[0].message.content
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Groq error: {e}")
    sources = [{"title": r["post_title"], "subreddit": r["subreddit"], "url": r["post_url"]} for r in relevant]
    return QueryResponse(answer=answer, sources=sources)

# ====================== Market Intelligence ======================
@app.post("/market-intelligence", response_model=MarketIntelligenceResponse)
async def market_intelligence(request: MarketIntelligenceRequest, user=Depends(get_current_user)):
    interest = get_search_volume(request.idea)
    competitors = search_competitors(request.idea)
    news = search_funding_news(request.idea)
    return MarketIntelligenceResponse(
        interest_score=interest,
        competitors=competitors,
        funding_news=news
    )

# ====================== Startup Builder ======================
@app.post("/generate-plan", response_model=GeneratePlanResponse)
async def generate_plan(request: GeneratePlanRequest, user=Depends(get_current_user)):
    from groq import Groq
    groq = Groq(api_key=GROQ_API_KEY)
    prompt = f"""Based on the following startup insight, generate:
- MVP Plan (key features for first version)
- Product Architecture (high-level tech stack)
- Go-to-Market Strategy (channels, pricing)
- Unique Moat (competitive advantage)
- Risk Assessment (legal, technical, market)
Insight: {request.insight}
"""
    try:
        response = groq.chat.completions.create(
            model=GROQ_MODEL,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.6
        )
        plan = response.choices[0].message.content
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Groq error: {e}")
    return GeneratePlanResponse(plan=plan)

# ====================== Trends ======================
@app.get("/trends", response_model=TrendResponse)
async def trends(subreddit: str = None, user=Depends(get_current_user)):
    query = supabase.table("insights").select("timestamp", "subreddit").eq("user_id", user["id"])
    if subreddit:
        query = query.eq("subreddit", subreddit)
    res = query.execute()
    from collections import defaultdict
    data = defaultdict(int)
    for row in res.data:
        date = row["timestamp"].split("T")[0]
        data[date] += 1
    dates = sorted(data.keys())
    counts = [data[d] for d in dates]
    return TrendResponse(dates=dates, counts=counts)

@app.get("/test")
async def test():
    return {"test": "ok"}

	from fastapi import FastAPI, HTTPException, Depends, BackgroundTasks
	from fastapi.middleware.cors import CORSMiddleware
	from datetime import datetime
	from typing import List, Dict, Any
	import logging

	from auth import get_current_user
	from models import (
	AnalysisRequest, AnalysisResponse, InsightResponse, QueryRequest, QueryResponse,
	MarketIntelligenceRequest, MarketIntelligenceResponse, GeneratePlanRequest, GeneratePlanResponse,
	TrendRequest, TrendResponse, UserSignup, UserLogin, UserProfile, SubscriptionPlan
	)
	from reddit_client import RedditClient
	from analyzer import call_groq_for_insight
	from serp_client import get_search_volume, search_competitors, search_funding_news
	from supabase_client import supabase
	from config import GROQ_MODEL, GROQ_API_KEY

	# Groq client is imported in analyzer
	# Reddit client
	reddit_client = RedditClient()

	app = FastAPI(title="Founder Intelligence Engine API", version="2.0")

	# CORS – allow all origins for development; restrict in production
	app.add_middleware(
	CORSMiddleware,
	allow_origins=["*"],
	allow_credentials=True,
	allow_methods=["*"],
	allow_headers=["*"],
	)

	logger = logging.getLogger(__name__)

	# ====================== Root ======================
	@app.get("/")
	async def root():
	return {"message": "Founder Intelligence Engine API is running"}

	# ====================== Auth Endpoints ======================
	@app.post("/auth/signup", response_model=UserProfile)
	async def signup(user: UserSignup):
	try:
	response = supabase.auth.sign_up({
	"email": user.email,
	"password": user.password,
	"options": {"data": {"full_name": user.full_name}}
	})
	user_data = response.user
	# Create profile record
	profile = supabase.table("profiles").insert({
	"id": user_data.id,
	"email": user_data.email,
	"full_name": user.full_name,
	"subscription_tier": "free",
	"subscription_status": "active"
	}).execute()
	return UserProfile(
	id=user_data.id,
	email=user_data.email,
	full_name=user.full_name,
	subscription_tier="free",
	subscription_status="active",
	created_at=user_data.created_at
	)
	except Exception as e:
	raise HTTPException(status_code=400, detail=str(e))

	@app.post("/auth/login")
	async def login(creds: UserLogin):
	try:
	response = supabase.auth.sign_in_with_password({
	"email": creds.email,
	"password": creds.password
	})
	return {"access_token": response.session.access_token, "token_type": "bearer"}
	except Exception as e:
	raise HTTPException(status_code=401, detail="Invalid credentials")

	@app.get("/auth/me", response_model=UserProfile)
	async def get_me(user=Depends(get_current_user)):
	return UserProfile(
	id=user["id"],
	email=user["email"],
	full_name=user["profile"]["full_name"],
	subscription_tier=user["profile"]["subscription_tier"],
	subscription_status=user["profile"]["subscription_status"],
	created_at=user["profile"]["created_at"]
	)

	# ====================== Subscription Plans ======================
	@app.get("/plans", response_model=List[SubscriptionPlan])
	async def list_plans():
	plans = supabase.table("plans").select("*").execute()
	return [SubscriptionPlan(**p) for p in plans.data]

	# ====================== Helper: Check subscription limits ======================
	def check_subscription_limit(user_id: str, requested_posts: int):
	"""Raise 403 if user exceeds plan limits."""
	profile = supabase.table("profiles").select("subscription_tier").eq("id", user_id).execute()
	tier = profile.data[0]["subscription_tier"] if profile.data else "free"
	# Fetch plan features
	plan = supabase.table("plans").select("features").eq("name", tier.capitalize()).execute()
	if plan.data:
	features = plan.data[0]["features"]
	posts_limit = features.get("posts_per_month", 10) # default free limit
	# Count posts this month for this user
	start_of_month = datetime.now().replace(day=1, hour=0, minute=0, second=0).isoformat()
	count_res = supabase.table("insights").select("id", count="exact").eq("user_id", user_id).gte("created_at", start_of_month).execute()
	current_count = count_res.count if hasattr(count_res, 'count') else 0
	if current_count + requested_posts > posts_limit:
	raise HTTPException(status_code=403, detail=f"Monthly post limit of {posts_limit} reached. Upgrade your plan.")
	# else no limit

	# ====================== Analysis Endpoints ======================
	@app.post("/analyze", response_model=AnalysisResponse)
	async def analyze(request: AnalysisRequest, user=Depends(get_current_user)):
	user_id = user["id"]
	total_posts = len(request.subreddits) * request.posts_per_subreddit
	check_subscription_limit(user_id, total_posts)

	all_results = []
	for sub in request.subreddits:
	posts = reddit_client.get_top_posts(sub, limit=request.posts_per_subreddit, time_filter=request.time_filter)
	for post in posts:
	text = f"Title: {post.get('title','')}\nBody: {post.get('selftext','')[:1500]}"
	insight = call_groq_for_insight(text)
	# Compute opportunity score
	pain = insight.get("problem_severity_score", 5)
	timing = insight.get("market_timing_score", 5)
	rev = len(insight.get("revenue_opportunity", "")) / 20
	comp = 10 - len(insight.get("competitor_weakness", "")) / 20
	opp_score = int((pain + timing + rev + comp) / 4 * 10)
	insight["opportunity_score"] = min(100, max(0, opp_score))

	result = {
	"user_id": user_id,
	"subreddit": sub,
	"post_title": post.get("title"),
	"post_url": post.get("url"),
	"timestamp": datetime.utcnow().isoformat(),
	"insight_data": insight,
	"opportunity_score": insight["opportunity_score"]
	}
	# Insert into Supabase
	supabase.table("insights").insert(result).execute()
	all_results.append({
	"subreddit": sub,
	"post_title": post.get("title"),
	"post_url": post.get("url"),
	"timestamp": result["timestamp"],
	"model": GROQ_MODEL,
	"insight": insight
	})
	return AnalysisResponse(status="success", message=f"Analyzed {len(all_results)} posts", insights=all_results)

	@app.get("/insights", response_model=List[InsightResponse])
	async def get_insights(user=Depends(get_current_user)):
	res = supabase.table("insights").select("*").eq("user_id", user["id"]).order("created_at", desc=True).execute()
	insights = []
	for row in res.data:
	insights.append(InsightResponse(
	id=row["id"],
	subreddit=row["subreddit"],
	post_title=row["post_title"],
	post_url=row["post_url"],
	timestamp=row["timestamp"],
	model=GROQ_MODEL,
	insight=row["insight_data"]
	))
	return insights

	# ====================== Query / RAG ======================
	@app.post("/query", response_model=QueryResponse)
	async def query_insights(request: QueryRequest, user=Depends(get_current_user)):
	# Simple keyword search over user's insights (full text search on insight_data)
	insights = supabase.table("insights").select("*").eq("user_id", user["id"]).execute()
	relevant = []
	for row in insights.data:
	data = row["insight_data"]
	text = f"{data.get('burning_pain','')} {data.get('startup_idea','')} {data.get('customer_type','')}"
	if request.query.lower() in text.lower():
	relevant.append(row)
	# If no exact matches, return top N by opportunity_score
	if not relevant:
	relevant = sorted(insights.data, key=lambda x: x["opportunity_score"], reverse=True)[:request.n_results]
	else:
	relevant = relevant[:request.n_results]

	if not relevant:
	raise HTTPException(status_code=404, detail="No relevant insights found")

	# Build context
	context = "\n\n".join([f"From r/{r['subreddit']}: {r['insight_data'].get('burning_pain','')} {r['insight_data'].get('startup_idea','')}" for r in relevant])
	# Call Groq to answer
	from groq import Groq
	groq = Groq(api_key=GROQ_API_KEY)
	try:
	response = groq.chat.completions.create(
	model=GROQ_MODEL,
	messages=[
	{"role": "system", "content": "You are a startup advisor. Use the provided Reddit insights to answer the user's question concisely."},
	{"role": "user", "content": f"Context:\n{context}\n\nQuestion: {request.query}"}
	],
	temperature=0.5
	)
	answer = response.choices[0].message.content
	except Exception as e:
	raise HTTPException(status_code=500, detail=f"Groq error: {e}")
	sources = [{"title": r["post_title"], "subreddit": r["subreddit"], "url": r["post_url"]} for r in relevant]
	return QueryResponse(answer=answer, sources=sources)

	# ====================== Market Intelligence ======================
	@app.post("/market-intelligence", response_model=MarketIntelligenceResponse)
	async def market_intelligence(request: MarketIntelligenceRequest, user=Depends(get_current_user)):
	interest = get_search_volume(request.idea)
	competitors = search_competitors(request.idea)
	news = search_funding_news(request.idea)
	return MarketIntelligenceResponse(
	interest_score=interest,
	competitors=competitors,
	funding_news=news
	)

	# ====================== Startup Builder ======================
	@app.post("/generate-plan", response_model=GeneratePlanResponse)
	async def generate_plan(request: GeneratePlanRequest, user=Depends(get_current_user)):
	from groq import Groq
	groq = Groq(api_key=GROQ_API_KEY)
	prompt = f"""Based on the following startup insight, generate:
	- MVP Plan (key features for first version)
	- Product Architecture (high-level tech stack)
	- Go-to-Market Strategy (channels, pricing)
	- Unique Moat (competitive advantage)
	- Risk Assessment (legal, technical, market)
	Insight: {request.insight}
	"""
	try:
	response = groq.chat.completions.create(
	model=GROQ_MODEL,
	messages=[{"role": "user", "content": prompt}],
	temperature=0.6
	)
	plan = response.choices[0].message.content
	except Exception as e:
	raise HTTPException(status_code=500, detail=f"Groq error: {e}")
	return GeneratePlanResponse(plan=plan)

	# ====================== Trends ======================
	@app.get("/trends", response_model=TrendResponse)
	async def trends(subreddit: str = None, user=Depends(get_current_user)):
	query = supabase.table("insights").select("timestamp", "subreddit").eq("user_id", user["id"])
	if subreddit:
	query = query.eq("subreddit", subreddit)
	res = query.execute()
	from collections import defaultdict
	data = defaultdict(int)
	for row in res.data:
	date = row["timestamp"].split("T")[0]
	data[date] += 1
	dates = sorted(data.keys())
	counts = [data[d] for d in dates]
	return TrendResponse(dates=dates, counts=counts)

	@app.get("/test")
	async def test():
	return {"test": "ok"}