app.py

# app.py
from fastapi import FastAPI, Request, Security, Depends, HTTPException, status
from fastapi.responses import JSONResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles
from fastapi.security.http import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
from typing import List
from engine import (
    SearchEngine, 
    get_batch_embeddings, 
    extract_content_with_playwright, 
    normalize_url, 
    ai_gatekeeper_clean, 
    ai_summarizer, 
    init_browser, 
    close_browser,
    ai_research_planner,
    ai_lore_extractor,
    ai_json_healer,
    ai_script_writer
)
import constants
import uvicorn
import asyncio
import json
import requests
from contextlib import asynccontextmanager

from motor.motor_asyncio import AsyncIOMotorClient
from fastapi import BackgroundTasks
import uuid
from datetime import datetime, timezone
from dotenv import load_dotenv
import os
import traceback

load_dotenv() # यह .env फाइल से सारे डेटा उठा लेगा
MONGO_URI = os.getenv("MONGO_URI")

# ⚡ MongoDB Setup
mongo_client = AsyncIOMotorClient(MONGO_URI)
db = mongo_client.spark_studio
history_col = db.research_history

# 🔒 Security Setup
# 💡 auto_error=False लगाने से अगर टोकन नहीं होगा तो ये तुरंत क्रैश नहीं करेगा, बल्कि हमारे फंक्शन को चेक करने का मौका देगा
security = HTTPBearer(auto_error=False) 
API_SECRET_TOKEN = os.getenv("API_SECRET_TOKEN", "sparkling_secret_123")

def verify_api_token(request: Request, credentials: HTTPAuthorizationCredentials = Security(security)):
    # 1. Health check (/) को बिना टोकन के पास होने दें (Cron job के लिए)
    if request.url.path == "/":
        return
    
    # 2. बाकी सभी API के लिए टोकन कड़ाई से चेक करें
    if not credentials or credentials.credentials != API_SECRET_TOKEN:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Not authenticated"
        )

# ⚡ Elite Fix: FastAPI Lifespan (Pre-warms Global Browser AND Database)
@asynccontextmanager
async def lifespan(app: FastAPI):
    print("Application Startup: Warming up Global Browser...")
    await init_browser()
    
    print("Application Startup: Initializing MongoDB Indexes...")
    # ⚡ 29 दिनों (2505600 seconds) का Auto-Delete (TTL) Index
    await history_col.create_index("created_at", expireAfterSeconds=2505600)
    
    yield
    
    print("Application Shutdown: Cleaning up Browser...")
    await close_browser()

# 👉 अब 'app' यहाँ डिफाइन हो रहा है, और lifespan इसके अंदर जा रहा है!
app = FastAPI(title="Sparkling Gyan Content Studio", lifespan=lifespan, dependencies=[Depends(verify_api_token)])

search_tool = SearchEngine()

# 🚦 Semaphore to prevent RAM/Chromium explosion
browser_semaphore = asyncio.Semaphore(3)

class QueryModel(BaseModel):
    query: str
    threshold: float = constants.SEARCH_THRESHOLD

class ExtractModel(BaseModel):
    url: str

class SynthesizeModel(BaseModel):
    query: str
    urls: List[str]

class DeepSearchModel(BaseModel):
    query: str

class ScriptGenModel(BaseModel):
    query: str
    research_data: dict

class DispatchVideoModel(BaseModel):
    script_text: str
    topic: str
    orientation: str = "horizontal"

@app.post("/api/fast-search")
def fast_search(data: QueryModel):
    if not data.query:
        return JSONResponse(status_code=400, content={"error": "No query provided"})
    
    docs = search_tool.perform_search(data.query, max_results=constants.DEFAULT_MAX_RESULTS)
    if not docs:
        return {"results": []}

    texts = [d["title"] + " " + d["body"] for d in docs]
    emb = get_batch_embeddings([data.query] + texts)
    scores = (emb[0] @ emb[1:].T).cpu().numpy().flatten()
    
    ranked_results = []
    for score, d in sorted(zip(scores, docs), key=lambda x: x[0], reverse=True):
        # 🛑 FIX: Added threshold filter here
        if score >= data.threshold:
            ranked_results.append({
                "title": d['title'],
                "snippet": d['body'],
                "url": d['href'],
                "score": f"{score:.4f}"
            })
    return {"results": ranked_results}

@app.post("/api/deep-search")
def deep_search(data: QueryModel):
    curr_k, seen = 50, set()
    all_docs, all_scores = [], []
    best_score = 0
    
    while curr_k <= constants.DEEP_SEARCH_LIMIT:
        docs = search_tool.perform_search(data.query, max_results=curr_k)
        new_docs = []
        for d in docs:
            norm_url = normalize_url(d['href'])
            if norm_url not in seen:
                seen.add(norm_url)
                new_docs.append(d)
                
        if new_docs:
            texts = [d["title"] + " " + d["body"] for d in new_docs]
            emb = get_batch_embeddings([data.query] + texts)
            scores = (emb[0] @ emb[1:].T).cpu().numpy().flatten()
            all_scores.extend(scores.tolist())
            all_docs.extend(new_docs)
            
        best_score = max(all_scores) if all_scores else 0
        if best_score >= data.threshold:
            break
        curr_k += 50

    ranked = sorted(zip(all_scores, all_docs), key=lambda x: x[0], reverse=True)
    
    # 🛑 FIX: Filter by threshold before taking top 10
    filtered_ranked = [(s, d) for s, d in ranked if s >= data.threshold][:10]
    
    results = [{"title": d['title'], "snippet": d['body'], "url": d['href'], "score": f"{s:.4f}"} for s, d in filtered_ranked]
    
    return {
        "status": f"स्कैनिंग: {len(all_docs)} साइट्स | बेस्ट मैच: {best_score:.4f} | पास हुए: {len(filtered_ranked)}",
        "results": results
    }

@app.post("/api/extract")
async def extract(data: ExtractModel):
    if not data.url:
        return JSONResponse(status_code=400, content={"error": "No URL provided"})
    
    async with browser_semaphore:
        content = await extract_content_with_playwright(data.url)
    return content

async def process_single_url_for_synthesis(url):
    """एक URL का पूरा प्रोसेस Semaphore के साथ: Scrape -> Clean -> Structuring"""
    async with browser_semaphore:
        content = await extract_content_with_playwright(url)
    
    if "text" in content and content["text"] != "Content not found.":
        clean_text = await ai_gatekeeper_clean(content["text"])
        return {
            "url": url,
            "content": clean_text
        }
    return None

@app.post("/api/synthesize")
async def synthesize_research(data: SynthesizeModel):
    if not data.urls or not data.query:
        return JSONResponse(status_code=400, content={"error": "URLs and Query required"})
        
    tasks = [process_single_url_for_synthesis(url) for url in data.urls]
    results = await asyncio.gather(*tasks)
    
    structured_data_list = [res for res in results if res]
            
    if not structured_data_list:
        return JSONResponse(status_code=500, content={"error": "Failed to extract valid text from URLs."})
        
    final_json_data = await ai_summarizer(structured_data_list, data.query)
    
    if "error" in final_json_data:
        return JSONResponse(status_code=500, content={"error": "JSON Parsing Failed", "raw_data": final_json_data.get("raw_output")})
    
    return {"summary_json": final_json_data}

# --- 🚀 Background Autonomous Agent Logic ---
async def run_autonomous_agent_bg(task_id: str, query: str):
    async def log_db(msg, color="var(--text-secondary)"):
        await history_col.update_one({"task_id": task_id}, {"$push": {"logs": {"msg": msg, "color": color}}})

    try:
        await log_db(f"Target locked: {query}", "var(--accent)")
        await log_db("Phase 1: Engaging AI Planner for deep sub-queries...", "var(--text-secondary)")
        
        # 1. Planning
        sub_queries = await ai_research_planner(query)
        if not isinstance(sub_queries, list):
            sub_queries = [f"{query} scientific explanation", f"{query} history", f"{query} mysteries"]

        await log_db(f"AI Strategy formulated. Parallel searching for: {sub_queries}", "var(--success)")
        all_queries = [query] + sub_queries
        
        # 2. Parallel Hunting
        await log_db("Phase 2: Harvesting raw nodes from Deep Web...", "var(--text-secondary)")
        all_docs = []
        for q in all_queries:
            docs = search_tool.perform_search(q, max_results=15)
            all_docs.extend(docs)
            await log_db(f"Scraped {len(docs)} initial links for '{q}'", "var(--text-muted)")
            
        seen = set()
        unique_docs = []
        for d in all_docs:
            norm_url = normalize_url(d['href'])
            if norm_url not in seen:
                seen.add(norm_url)
                unique_docs.append(d)
                
        # 3. Semantic Reranking
        await log_db(f"Phase 3: Found {len(unique_docs)} unique nodes. Running semantic reranking...", "var(--text-secondary)")
        texts = [d["title"] + " " + d["body"] for d in unique_docs]
        emb = get_batch_embeddings([query] + texts)
        scores = (emb[0] @ emb[1:].T).cpu().numpy().flatten()
        
        # 🛑 FIX: Cleaned up duplicates and added Strict Threshold & Safety Bouncer
        ranked = sorted(zip(scores, unique_docs), key=lambda x: x[0], reverse=True)
        
        MIN_THRESHOLD = getattr(constants, 'SEARCH_THRESHOLD', 0.5) 
        BANNED_WORDS = ['porn', 'xxx', 'sex', 'casino', 'betting', 'escort', 'viagra']
        
        valid_nodes = []
        for s, d in ranked:
            url_lower = d['href'].lower()
            # सिर्फ वही लिंक्स पास होंगे जो थ्रेशोल्ड क्लियर करें और जिनमें गंदे कीवर्ड्स न हों
            if s >= MIN_THRESHOLD and not any(banned in url_lower for banned in BANNED_WORDS):
                valid_nodes.append((s, d))
                
        top_10_clean = valid_nodes[:10]
        
        if not top_10_clean:
            await log_db("FATAL: All scraped nodes were irrelevant, spam, or scored below threshold. Aborting to save resources.", "var(--danger)")
            await history_col.update_one({"task_id": task_id}, {"$set": {"status": "failed", "error": "No relevant data passed the AI embedding filter."}})
            return
            
        top_urls = [d['href'] for s, d in top_10_clean]
        top_results_rich = [{"title": d['title'], "snippet": d['body'], "url": d['href']} for s, d in top_10_clean]
        
        await log_db(f"{len(top_urls)} High-Tension Nodes passed security. Commencing extraction...", "var(--success)")
        for i, url in enumerate(top_urls, 1):
            await log_db(f"Targeting Node [{i}/{len(top_urls)}]: {url}", "var(--accent-dim)")
            
        # 4. Bypassing firewalls & extracting context
        await log_db("Phase 4: Bypassing firewalls & extracting context (Parallel)...", "var(--text-secondary)")
        tasks = [process_single_url_for_synthesis(url) for url in top_urls]
        results = await asyncio.gather(*tasks)
        structured_data_list = [res for res in results if res]
                
        if not structured_data_list:
            await log_db("FATAL: Extraction failed. All nodes blocked.", "var(--danger)")
            await history_col.update_one({"task_id": task_id}, {"$set": {"status": "failed", "error": "All nodes blocked or no data extracted."}})
            return
            
        await log_db(f"Extraction successful for {len(structured_data_list)} nodes. Junk cleaned.", "var(--success)")
        
        # 5. Engaging Dual-Core AI & Gatekeeper
        await log_db("Phase 5: Engaging Dual-Core AI...", "var(--text-secondary)")
        task_sci = asyncio.create_task(ai_summarizer(structured_data_list, query))
        task_lore = asyncio.create_task(ai_lore_extractor(structured_data_list, query))
        
        await asyncio.gather(task_sci, task_lore)
        
        try:
            sci_data = task_sci.result()
        except Exception as e:
            sci_data = {"error": str(e), "raw_output": ""}
            
        try:
            lore_data = task_lore.result()
        except Exception as e:
            lore_data = {"error": str(e), "raw_output": ""}

        # 🏥 THE GATEKEEPER: अगर JSON टूटा है, तो उसे ठीक करो!
        if "error" in sci_data and "raw_output" in sci_data:
            await log_db("⚠️ Scientist AI JSON broken. Engaging Gatekeeper Healer...", "var(--danger)")
            healed_sci = await ai_json_healer(sci_data["raw_output"])
            if "error" not in healed_sci:
                sci_data = healed_sci
                await log_db("✅ Gatekeeper successfully healed Scientist Data!", "var(--success)")

        if "error" in lore_data and "raw_output" in lore_data:
            await log_db("⚠️ Storyteller AI JSON broken. Engaging Gatekeeper Healer...", "var(--danger)")
            healed_lore = await ai_json_healer(lore_data["raw_output"])
            if "error" not in healed_lore:
                lore_data = healed_lore
                await log_db("✅ Gatekeeper successfully healed Storyteller Data!", "var(--success)")

        final_data = {"conflict_engine": sci_data, "lore_engine": lore_data}
        
        # 🎯 काम खत्म! DB में फाइनल रिज़ल्ट सेव करो
        await history_col.update_one(
            {"task_id": task_id}, 
            {"$set": {"status": "complete", "result": final_data, "urls": top_results_rich}} 
        )
        await log_db("Autonomous Research Complete & Saved to History.", "var(--success)")

    except Exception as e:
        await history_col.update_one(
            {"task_id": task_id}, 
            {"$set": {"status": "failed", "error": str(e)}}
        )
        await log_db(f"CRITICAL CRASH: {str(e)}", "var(--danger)")


# --- 🚀 Script Generation & Dispatch Endpoints ---
@app.post("/api/research/start")
async def start_research(data: DeepSearchModel, background_tasks: BackgroundTasks):
    task_id = str(uuid.uuid4())
    
    # 1. डेटाबेस में इनिशियल एंट्री बनाओ
    await history_col.insert_one({
        "task_id": task_id,
        "query": data.query,
        "status": "processing",
        "logs": [{"msg": "Initializing Autonomous Agent in Background...", "color": "var(--accent)"}],
        "result": None,
        "urls": [],
        "created_at": datetime.now(timezone.utc)
    })
    
    # 2. काम को बैकग्राउंड में धकेल दो
    background_tasks.add_task(run_autonomous_agent_bg, task_id, data.query)
    
    # 3. तुरंत UI को task_id दे दो
    return {"task_id": task_id}

# यह डिक्शनरी मेमोरी में टास्क का स्टेटस सेव करेगी
SCRIPT_TASKS = {} 

async def background_script_generator(task_id: str, research_data: dict, query: str):
    try:
        SCRIPT_TASKS[task_id] = {"status": "processing"}
        final_script = await ai_script_writer(research_data, query)
        SCRIPT_TASKS[task_id] = {"status": "completed", "script": final_script}
    except Exception as e:
        SCRIPT_TASKS[task_id] = {"status": "failed", "error": str(e)}

@app.post("/api/generate-script-async")
async def generate_script_async(request: Request, background_tasks: BackgroundTasks):
    data = await request.json()
    task_id = str(uuid.uuid4())
    SCRIPT_TASKS[task_id] = {"status": "pending"}
    
    # काम को बैकग्राउंड में धकेल दिया
    background_tasks.add_task(background_script_generator, task_id, data.get("research_data"), data.get("query"))
    
    # फ्रंटएंड को तुरंत 1 सेकंड में जवाब मिल जाएगा
    return {"task_id": task_id, "status": "processing"}

@app.get("/api/script-status/{task_id}")
async def get_script_status(task_id: str):
    if task_id not in SCRIPT_TASKS:
        raise HTTPException(status_code=404, detail="Task not found")
    return SCRIPT_TASKS[task_id]

# --- 🚀 Status Checking Endpoint ---
@app.get("/api/research/status/{task_id}")
async def get_status(task_id: str):
    task = await history_col.find_one({"task_id": task_id}, {"_id": 0})
    if not task:
        return JSONResponse(status_code=404, content={"error": "Task not found"})
    return task

# --- 🚀 History Fetching Endpoint ---
@app.get("/api/history")
async def get_history():
    cursor = history_col.find({}, {"_id": 0, "task_id": 1, "query": 1, "status": 1, "created_at": 1}).sort("created_at", -1)
    history = await cursor.to_list(length=100)
    return {"history": history}

@app.post("/api/generate-script")
async def generate_script(data: ScriptGenModel):
    try:
        script_text = await ai_script_writer(data.research_data, data.query)
        return {"script": script_text}
    except Exception as e:
        return JSONResponse(status_code=500, content={"error": str(e)})

@app.post("/api/dispatch-to-creator")
async def dispatch_to_creator(data: DispatchVideoModel):
    print(f"🚀 Dispatching Script to Vidspark API for topic: {data.topic}")
    
    headers = {
        "X-Vidspark-Key": constants.VIDSPARK_API_KEY
    }
    
    payload = {
        "script_text": (None, data.script_text),
        "orientation": (None, data.orientation)
    }
    
    try:
        response = requests.post(constants.VIDSPARK_URL, headers=headers, files=payload)
        
        if response.status_code == 200:
            res_data = response.json()
            return {
                "status": "success", 
                "task_id": res_data.get("task_id"),
                "message": res_data.get("message", "Video generation started!")
            }
        else:
            return JSONResponse(status_code=response.status_code, content={"error": response.text})
            
    except Exception as e:
        return JSONResponse(status_code=500, content={"error": f"Failed to connect to Vidspark API: {str(e)}"})

@app.get("/")
def index():
    return {"status": "success", "message": "API is up and running. 🚀"}
    
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=7860)