Spaces:

paulhemb
/

reddit_intelligence_platform

Sleeping

Update main.py

5b287c8 verified 7 months ago

12 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_auth, supabase_db
from config import GROQ_MODEL, GROQ_API_KEY, SUPABASE_JWT_SECRET

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

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

# CORS
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:
        # 1. Create auth user (using anon client)
        response = supabase_auth.auth.sign_up({
            "email": user.email,
            "password": user.password,
            "options": {"data": {"full_name": user.full_name}}
        })
        user_data = response.user

        # 2. Check if profile already exists (using service client to bypass RLS)
        existing = supabase_db.table("profiles").select("*").eq("id", user_data.id).execute()
        if not existing.data:
            # Create profile record
            supabase_db.table("profiles").insert({
                "id": user_data.id,
                "email": user_data.email,
                "full_name": user.full_name,
                "subscription_tier": "free",
                "subscription_status": "active"
            }).execute()

        # 3. Return profile
        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:
        # If the user already exists in auth but not in profiles, we can still return a helpful message
        raise HTTPException(status_code=400, detail=str(e))

@app.post("/auth/login")
async def login(creds: UserLogin):
    try:
        response = supabase_auth.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)):
    # user already contains profile info from the token validation
    # but we can fetch fresh profile from DB if needed
    profile = supabase_db.table("profiles").select("*").eq("id", user["id"]).execute()
    if not profile.data:
        raise HTTPException(status_code=404, detail="Profile not found")
    p = profile.data[0]
    return UserProfile(
        id=user["id"],
        email=user["email"],
        full_name=p["full_name"],
        subscription_tier=p["subscription_tier"],
        subscription_status=p["subscription_status"],
        created_at=p["created_at"]
    )

# ====================== Subscription Plans ======================
@app.get("/plans", response_model=List[SubscriptionPlan])
async def list_plans():
    plans = supabase_db.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_db.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_db.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_db.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 using service client
            supabase_db.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_db.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)):
    insights = supabase_db.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 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")

    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])
    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_db.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_auth, supabase_db
	from config import GROQ_MODEL, GROQ_API_KEY, SUPABASE_JWT_SECRET

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

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

	# CORS
	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:
	# 1. Create auth user (using anon client)
	response = supabase_auth.auth.sign_up({
	"email": user.email,
	"password": user.password,
	"options": {"data": {"full_name": user.full_name}}
	})
	user_data = response.user

	# 2. Check if profile already exists (using service client to bypass RLS)
	existing = supabase_db.table("profiles").select("*").eq("id", user_data.id).execute()
	if not existing.data:
	# Create profile record
	supabase_db.table("profiles").insert({
	"id": user_data.id,
	"email": user_data.email,
	"full_name": user.full_name,
	"subscription_tier": "free",
	"subscription_status": "active"
	}).execute()

	# 3. Return profile
	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:
	# If the user already exists in auth but not in profiles, we can still return a helpful message
	raise HTTPException(status_code=400, detail=str(e))

	@app.post("/auth/login")
	async def login(creds: UserLogin):
	try:
	response = supabase_auth.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)):
	# user already contains profile info from the token validation
	# but we can fetch fresh profile from DB if needed
	profile = supabase_db.table("profiles").select("*").eq("id", user["id"]).execute()
	if not profile.data:
	raise HTTPException(status_code=404, detail="Profile not found")
	p = profile.data[0]
	return UserProfile(
	id=user["id"],
	email=user["email"],
	full_name=p["full_name"],
	subscription_tier=p["subscription_tier"],
	subscription_status=p["subscription_status"],
	created_at=p["created_at"]
	)

	# ====================== Subscription Plans ======================
	@app.get("/plans", response_model=List[SubscriptionPlan])
	async def list_plans():
	plans = supabase_db.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_db.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_db.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_db.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 using service client
	supabase_db.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_db.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)):
	insights = supabase_db.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 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")

	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])
	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_db.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"}