Update app.py

app.py

@@ -5,21 +5,20 @@ import os
 import uvicorn
 import re
 import math
-import time
 import multiprocessing
 import urllib.parse
 import difflib
+import gc
+import sys
+import array
 from datetime import datetime
 from collections import Counter, defaultdict
 from contextlib import asynccontextmanager
 from typing import List, Dict, Any, Tuple, Set
 
 from fastapi import FastAPI, Query, Form
-from fastapi.responses import JSONResponse
 from fastapi.middleware.cors import CORSMiddleware
 from huggingface_hub import snapshot_download
-import aiohttp
-from bs4 import BeautifulSoup
 import torch
 
 # Hugging Face imports
@@ -39,10 +38,6 @@ HF_TOKEN = os.getenv("HF_TOKEN")
 AI_RESPONSE_CACHE = {}
 MAX_AI_CACHE_SIZE = 10000
 
-CONTENT_CACHE = {}
-MAX_CONTENT_CONCURRENCY = 10
-content_semaphore = asyncio.Semaphore(MAX_CONTENT_CONCURRENCY)
-
 # --- 1. DETERMINISTIC PREPROCESSING & CONSTANTS ---
 STOP_WORDS = {"the", "a", "an", "and", "or", "but", "in", "on", "at", "to", "for", "of", "with", "is", "are", "by", "this", "that", "it", "what", "who", "where", "when", "why", "how"}
 ADULT_KEYWORDS = {
@@ -78,7 +73,6 @@ def normalize_text(text: str) -> str:
     if not text: return ""
     text = str(text).lower()
     text = re.sub(r'[\'’]', '', text)
-    # \w allows words from any language (Hindi, Marathi, etc.)
     text = re.sub(r'[^\w\s\-]', ' ', text)
     return " ".join(text.split())
 
@@ -91,146 +85,126 @@ def tokenize(text: str, remove_stopwords: bool = True) -> List[str]:
 # --- 2. DETERMINISTIC QUERY INTENT CLASSIFIER ---
 def detect_query_intent(query_tokens: List[str]) -> Dict[str, Any]:
     intents = []
-    
     news_terms = {"latest", "news", "today", "breaking", "recent", "updates", "current", "aaj"}
     if any(term in query_tokens for term in news_terms): intents.append("NEWS")
-        
     def_terms = {"definition", "meaning", "kya", "matlab"}
     if any(term in query_tokens for term in def_terms): intents.append("DEFINITION")
-        
     how_terms = {"tutorial", "guide", "steps", "learn", "how", "kaise"}
     if any(term in query_tokens for term in how_terms): intents.append("HOW_TO")
-        
     comp_terms = {"vs", "versus", "compare", "comparison", "difference", "best", "accha"}
     if any(term in query_tokens for term in comp_terms): intents.append("COMPARISON")
-        
     local_terms = {"near", "restaurant", "restaurants", "location", "weather", "paas"}
     if any(term in query_tokens for term in local_terms): intents.append("LOCAL")
-        
     if len(query_tokens) <= 2 and not intents: intents.append("NAVIGATIONAL")
-        
     return {"intent": intents[0] if intents else "GENERAL", "signals": intents}
 
-# --- 3. DEDICATED WEB CONTENT EXTRACTION ---
-async def extract_page_content(url: str) -> Dict[str, Any]:
-    if url in search_index.url_to_doc:
-        doc = search_index.url_to_doc[url]
-        if doc.get("_fast_content"):
-            return {"url": url, "title": doc.get("_fast_title", ""), "content": doc.get("_fast_content", ""), "success": True}
-    
-    if url in CONTENT_CACHE:
-        return CONTENT_CACHE[url]
-        
-    async with content_semaphore:
-        try:
-            async with aiohttp.ClientSession() as session:
-                async with session.get(url, timeout=3.0, headers={"User-Agent": "Mozilla/5.0 NebulixBot"}) as resp:
-                    if resp.status != 200:
-                        return {"url": url, "title": "", "content": "", "success": False, "error": f"HTTP {resp.status}"}
-                    
-                    html = await resp.text()
-                    soup = BeautifulSoup(html, "html.parser")
-                    
-                    for tag in soup(['script', 'style', 'nav', 'footer', 'header', 'aside', 'noscript', 'form', 'button', 'iframe']):
-                        tag.decompose()
-                        
-                    blocks = []
-                    for tag in soup.find_all(['h1', 'h2', 'h3', 'p', 'article', 'main', 'li']):
-                        text = tag.get_text(separator=' ', strip=True)
-                        if len(text) > 30: blocks.append(text)
-                            
-                    clean_blocks = []
-                    seen_blocks = set()
-                    for b in blocks:
-                        if b not in seen_blocks:
-                            seen_blocks.add(b)
-                            clean_blocks.append(b)
-                            
-                    content_str = "\n\n".join(clean_blocks)
-                    page_title = soup.title.string.strip() if soup.title and soup.title.string else ""
-                    
-                    result = {"url": url, "title": page_title, "content": content_str, "success": True}
-                    CONTENT_CACHE[url] = result
-                    return result
-        except Exception as e:
-            return {"url": url, "title": "", "content": "", "success": False, "error": str(e)}
-
+# --- 3. INTERNAL RAG CHUNKING LOGIC ---
 def rank_chunks_for_rag(text: str, query_tokens: List[str]) -> List[str]:
     if not text: return []
     chunks = text.split("\n\n")
     q_set = set(query_tokens)
     scored_chunks = []
-    
     for c in chunks:
         c_toks = set(tokenize(c))
         score = len(q_set.intersection(c_toks))
         if score > 0: scored_chunks.append((score, c))
-            
     if not scored_chunks and chunks: scored_chunks = [(0, c) for c in chunks[:3]]
     scored_chunks.sort(reverse=True, key=lambda x: x[0])
     return [c for score, c in scored_chunks[:2]] 
 
-# --- 4. CORE DETERMINISTIC SEARCH INDEX ---
+
+# --- 4. CORE DETERMINISTIC SEARCH INDEX (STREAMING ARCHITECTURE) ---
 class HybridSearchIndex:
     def __init__(self):
         self.docs = []
-        self.url_to_doc = {}
         self.total_docs = 0
+        self.total_len = 0
         self.vocab = set()
         self.high_freq_vocab = []
         self.df = defaultdict(int)
         self.avg_doc_len = 0.0
-        self.inverted_index = defaultdict(set)
+        
+        # MEMORY SAVER: C-style array index, saves >90% memory over sets
+        self.inverted_index = defaultdict(lambda: array.array('I')) 
+        
+        self.suggestion_counter = Counter()
         self.suggestions = []
         self.k1 = 1.5
         self.b = 0.75
 
-    def build(self, raw_data: List[dict]):
-        print(f"📥 Building Advanced Deterministic Search Index from {len(raw_data)} documents...")
-        total_len = 0
-        suggestion_counter = Counter()
-
-        for doc_id, doc in enumerate(raw_data):
-            self.docs.append(doc)
-            self.url_to_doc[doc.get("url", "")] = doc
-            
-            # Using partial content tokens to save memory while maximizing recall
-            title_tokens = tokenize(doc.get("_fast_title", ""))
-            desc_tokens = tokenize(doc.get("_fast_desc", ""))
-            domain_tokens = tokenize(doc.get("_fast_domain", ""))
-            headings_tokens = tokenize(doc.get("_fast_headings", ""))
-            snippet_tokens = tokenize(doc.get("_fast_snippet", ""))
-            content_tokens = tokenize(doc.get("_fast_content", ""))[:400] 
-            
-            all_tokens = title_tokens + desc_tokens + domain_tokens + headings_tokens + snippet_tokens + content_tokens
-            
-            doc["_token_len"] = len(all_tokens)
-            total_len += len(all_tokens)
+    def add_document(self, doc_json: dict):
+        doc_id = len(self.docs)
+        
+        metadata = doc_json.get("metadata") or {}
+        content_dict = doc_json.get("content") or {}
+        
+        title_str = metadata.get("title") or ""
+        desc_str = metadata.get("description") or ""
+        snippet_str = content_dict.get("snippet") or ""
+        main_text = (content_dict.get("main_text") or "")[:800] # Safe crop for 16GB
+        
+        headings = content_dict.get("headings") or {}
+        h1_str = " ".join(headings.get("h1") or [])
+        h2_str = " ".join(headings.get("h2") or [])
+        
+        url = doc_json.get("url", "")
+        parsed_url = urllib.parse.urlparse(url)
+        
+        # MEMORY SAVER: sys.intern deduplicates domain strings globally in RAM
+        domain_str = sys.intern(parsed_url.netloc.lower().replace("www.", ""))
+        
+        clean_doc = {
+            "url": url,
+            "_fast_domain": domain_str,
+            "_fast_path": parsed_url.path.strip("/"),
+            "_fast_title": title_str[:150],
+            "_fast_desc": desc_str[:250],
+            "_fast_snippet": snippet_str[:300],
+            "_fast_content": main_text,
+            "_fast_headings": (h1_str + " " + h2_str)[:200]
+        }
+        
+        title_tokens = tokenize(clean_doc["_fast_title"])
+        desc_tokens = tokenize(clean_doc["_fast_desc"])
+        domain_tokens = tokenize(clean_doc["_fast_domain"])
+        headings_tokens = tokenize(clean_doc["_fast_headings"])
+        snippet_tokens = tokenize(clean_doc["_fast_snippet"])
+        content_tokens = tokenize(clean_doc["_fast_content"])[:250] 
+        
+        all_tokens = title_tokens + desc_tokens + domain_tokens + headings_tokens + snippet_tokens + content_tokens
+        
+        clean_doc["_token_len"] = len(all_tokens)
+        self.total_len += len(all_tokens)
+        
+        combined_adult_check = f"{title_str} {desc_str} {url} {snippet_str} {main_text[:200]}".lower()
+        clean_doc["_is_adult"] = bool(ADULT_REGEX.search(combined_adult_check))
+        
+        self.docs.append(clean_doc)
+        
+        distinct_tokens = set(all_tokens)
+        for term in distinct_tokens:
+            self.inverted_index[term].append(doc_id)
+            self.df[term] += 1
+            self.vocab.add(term)
             
-            # Use distinct tokens to speed up df counting
-            distinct_tokens = set(all_tokens)
-            for term in distinct_tokens:
-                self.inverted_index[term].add(doc_id)
-                self.df[term] += 1
-                self.vocab.add(term)
-                
-            if not doc.get("_is_adult", False):
-                clean_title = [w for w in title_tokens if w not in ADULT_KEYWORDS and len(w) > 2]
-                for w in clean_title: suggestion_counter[w] += 1
-                
-                d_clean = doc.get("_fast_domain", "").split('.')[0]
-                if len(d_clean) > 2 and d_clean not in ADULT_KEYWORDS: suggestion_counter[d_clean] += 5
+        if not clean_doc["_is_adult"]:
+            clean_title = [w for w in title_tokens if w not in ADULT_KEYWORDS and len(w) > 2]
+            for w in clean_title: self.suggestion_counter[w] += 1
+            if len(domain_str) > 2 and domain_str not in ADULT_KEYWORDS: self.suggestion_counter[domain_str] += 5
 
+    def finalize_build(self):
         self.total_docs = len(self.docs)
-        self.avg_doc_len = total_len / max(1, self.total_docs)
-        self.suggestions = suggestion_counter.most_common(20000)
-        # Store high freq words for fast spell correction (Did you mean)
+        self.avg_doc_len = self.total_len / max(1, self.total_docs)
+        self.suggestions = self.suggestion_counter.most_common(20000)
         self.high_freq_vocab = [word for word, count in Counter(self.df).most_common(15000)]
-        print(f"🚀 Indexing Complete! {self.total_docs} records fully embedded.")
+        
+        self.suggestion_counter.clear()
+        gc.collect()
+        print(f"🚀 Indexing Complete! Exactly {self.total_docs} records fully embedded inside Safe RAM Limits.")
 
     def idf(self, term: str) -> float:
         df = self.df.get(term, 0)
-        total = max(1, self.total_docs) # Prevents division by zero
+        total = max(1, self.total_docs) 
         return math.log(1.0 + (total - df + 0.5) / (df + 0.5))
 
     def _get_candidates(self, query_tokens: List[str]) -> Set[int]:
@@ -238,7 +212,6 @@ class HybridSearchIndex:
         for term in query_tokens:
             if term in self.inverted_index: candidates.update(self.inverted_index[term])
         
-        # Fast fallback for slight misspellings during retrieval
         if len(candidates) < 10 and self.high_freq_vocab:
             for term in query_tokens:
                 if term not in self.inverted_index:
@@ -263,7 +236,7 @@ class HybridSearchIndex:
         scores = {"bm25": 0.0, "title": 0.0, "phrase": 0.0, "exact": 0.0, "url": 0.0, "domain": 0.0, "quality": 0.0, "freshness": 0.0, "proximity": 0.0, "root_domain": 0.0, "headings": 0.0, "body": 0.0}
 
         doc_len = doc.get("_token_len", 1)
-        avg_dl = max(1.0, self.avg_doc_len) # Fixed ZeroDivisionError
+        avg_dl = max(1.0, self.avg_doc_len)
         
         content_lower = content.lower()
         snippet_lower = snippet.lower()
@@ -330,27 +303,23 @@ class HybridSearchIndex:
         if len(desc_toks) >= 10: q_score += 0.2
         if doc.get("url", "").startswith("https"): q_score += 0.2
         
-        # Code-based Domain Authority Boost
         if domain.endswith(".edu") or domain.endswith(".gov"): q_score += 0.4
         elif domain.endswith(".org"): q_score += 0.15
 
         if len(title_toks) > 0 and (len(set(title_toks)) / len(title_toks)) < 0.5: q_score -= 0.3
         scores["quality"] = max(0.0, min(1.0, q_score))
 
-        if url == "" or url == "/":
-            scores["root_domain"] = 1.0
+        if url == "" or url == "/": scores["root_domain"] = 1.0
 
         return scores
 
     def search(self, query: str, debug: bool = False) -> Tuple[List[dict], Dict[str, Any], str]:
-        # Handle search request if indexing is still loading or totally empty
         if self.total_docs == 0:
             return [], {"intent": "GENERAL", "signals": []}, None
 
         query_norm = normalize_text(query)
         raw_tokens = query_norm.split()
         
-        # Spell Correction ("Did you mean / Showing results for")
         corrected_tokens = []
         is_corrected = False
         
@@ -359,7 +328,6 @@ class HybridSearchIndex:
                 if t in self.vocab or t in EXPANSION_DICT or t in ADULT_KEYWORDS:
                     corrected_tokens.append(t)
                 else:
-                    # Fast correction using pre-computed high frequency words
                     matches = difflib.get_close_matches(t, self.high_freq_vocab, n=1, cutoff=0.85)
                     if matches:
                         corrected_tokens.append(matches[0])
@@ -377,11 +345,9 @@ class HybridSearchIndex:
             if t in EXPANSION_DICT: expanded_tokens.extend(EXPANSION_DICT[t])
         
         sorted_tokens = list(dict.fromkeys(expanded_tokens)) 
-        if not sorted_tokens:
-            sorted_tokens = corrected_tokens 
+        if not sorted_tokens: sorted_tokens = corrected_tokens 
             
         sorted_tokens.sort(key=lambda t: self.idf(t), reverse=True)
-        
         intent_info = detect_query_intent(sorted_tokens)
         intent = intent_info["intent"]
         user_wants_adult = any(aw in query_norm for aw in ADULT_KEYWORDS)
@@ -438,7 +404,7 @@ class HybridSearchIndex:
             
             if final_score > 0.05:
                 doc_out = {
-                    "doc_id": doc.get("doc_id", ""),
+                    "doc_id": str(doc_id),
                     "title": doc.get("_fast_title", ""),
                     "description": doc.get("_fast_desc", ""),
                     "snippet": doc.get("_fast_snippet", ""),
@@ -469,8 +435,7 @@ class HybridSearchIndex:
 search_index = HybridSearchIndex()
 
 def load_and_index_dataset():
-    print("📦 Automatically Downloading 'Nebulixlabs/web-search-dataset' ...")
-    dataset_path = "."
+    print("📦 Automatically Downloading and Streaming 'web-search-dataset' ...")
     try:
         dataset_path = snapshot_download(
             repo_id="Nebulixlabs/web-search-dataset", 
@@ -480,14 +445,15 @@ def load_and_index_dataset():
         )
     except Exception as e:
         print("⚠️ Direct HF Hub fetch failed, attempting local 'data' folder mapping. Error:", e)
+        dataset_path = "."
         
-    raw_data = []
     data_folder = os.path.join(dataset_path, "data")
-    if not os.path.exists(data_folder):
-        data_folder = "data" 
+    if not os.path.exists(data_folder): data_folder = "data" 
         
     if os.path.exists(data_folder):
         jsonl_files = [f for f in os.listdir(data_folder) if f.endswith(".jsonl")]
+        docs_processed = 0
+        
         for file in jsonl_files:
             file_path = os.path.join(data_folder, file)
             with open(file_path, "r", encoding="utf-8") as f:
@@ -495,39 +461,16 @@ def load_and_index_dataset():
                     if not line.strip(): continue
                     try:
                         doc = json.loads(line)
-                        url = doc.get("url", "")
-                        
-                        metadata = doc.get("metadata") or {}
-                        content_dict = doc.get("content") or {}
-                        
-                        title_str = metadata.get("title") or ""
-                        desc_str = metadata.get("description") or ""
-                        snippet_str = content_dict.get("snippet") or ""
-                        main_text = content_dict.get("main_text") or ""
-                        
-                        headings = content_dict.get("headings") or {}
-                        h1_str = " ".join(headings.get("h1") or [])
-                        h2_str = " ".join(headings.get("h2") or [])
-                        
-                        parsed_url = urllib.parse.urlparse(url)
+                        search_index.add_document(doc)
+                        docs_processed += 1
                         
-                        doc["_fast_domain"] = doc.get("domain") or parsed_url.netloc.lower().replace("www.", "")
-                        doc["_fast_path"] = parsed_url.path.strip("/")
-                        doc["_fast_title"] = title_str
-                        doc["_fast_desc"] = desc_str
-                        doc["_fast_snippet"] = snippet_str
-                        doc["_fast_content"] = main_text
-                        doc["_fast_headings"] = h1_str + " " + h2_str
-                        
-                        combined_adult_check = f"{title_str} {desc_str} {url} {snippet_str} {main_text[:200]}".lower()
-                        # Fast Regex Adult Match
-                        doc["_is_adult"] = bool(ADULT_REGEX.search(combined_adult_check))
-                        
-                        raw_data.append(doc)
+                        if docs_processed % 50000 == 0:
+                            gc.collect()
+                            print(f"🔄 Streamed & Indexed {docs_processed} docs so far...")
                     except Exception: 
                         pass
-            
-    search_index.build(raw_data)
+                        
+    search_index.finalize_build()
 
 # --- AI GENERATION MODULE ---
 def load_ai_models():
@@ -554,22 +497,19 @@ async def run_ai_task(messages, max_tokens: int, query_key: str):
     
     try: await asyncio.wait_for(ai_semaphore.acquire(), timeout=0.1)
     except asyncio.TimeoutError:
-        return "⚠️ AI Servers are fully occupied right now. Displaying your ultra-fast search results below! (Please try AI Mode again in a few seconds)"
+        return "⚠️ AI Servers are fully occupied right now. Displaying your ultra-fast search results below!"
 
     try:
         def _gen():
             tokenizer = ai_generator.tokenizer
             try: prompt = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
             except: prompt = "\n".join([f"{m['role'].capitalize()}: {m['content']}" for m in messages]) + "\nAssistant: "
-
             result = ai_generator(prompt, max_new_tokens=max_tokens, do_sample=False, truncation=True, return_full_text=False)
             return result[0]['generated_text'].strip()
 
         response = await asyncio.to_thread(_gen)
-        
         if len(AI_RESPONSE_CACHE) > MAX_AI_CACHE_SIZE: AI_RESPONSE_CACHE.clear()
         AI_RESPONSE_CACHE[query_key] = response
-        
         return response
     finally:
         ai_semaphore.release()
@@ -591,11 +531,6 @@ async def get_suggestions(q: str = Query(..., min_length=1)):
     matches = [word for word, count in search_index.suggestions if word.startswith(q)][:7]
     return {"query": q, "suggestions": matches}
 
-@app.get("/content")
-async def get_content_endpoint(url: str = Query(...)):
-    res = await extract_page_content(url)
-    return res
-
 @app.get("/search")
 async def pure_search(q: str = Query(...), content: bool = Query(False), debug: bool = Query(False)):
     req_id = str(uuid.uuid4())
@@ -603,8 +538,7 @@ async def pure_search(q: str = Query(...), content: bool = Query(False), debug:
     top_20 = results[:STRICT_RESULT_LIMIT]
     
     for r in top_20:
-        if content:
-            r["content"] = r.get("_internal_content", "")
+        if content: r["content"] = r.get("_internal_content", "")
         r.pop("_internal_content", None)
             
     return {
@@ -628,14 +562,9 @@ async def process_aimode(q: str, include_content: bool, debug: bool, max_ai_toke
             r.pop("_internal_content", None)
             
         return {
-            "request_id": req_id, 
-            "query": q,
-            "corrected_query": corrected_query, 
-            "ai_response": AI_RESPONSE_CACHE[cache_key], 
-            "results_count": len(top_20), 
-            "results": top_20,
-            "sources": [],
-            "cached": True
+            "request_id": req_id, "query": q, "corrected_query": corrected_query, 
+            "ai_response": AI_RESPONSE_CACHE[cache_key], "results_count": len(top_20), 
+            "results": top_20, "sources": [], "cached": True
         }
 
     query_tokens = tokenize(q)
@@ -646,9 +575,7 @@ async def process_aimode(q: str, include_content: bool, debug: bool, max_ai_toke
         valid_sources_count = 0
         for r in top_20[:3]:
             doc_content = r.get("_internal_content", "")
-            
-            if include_content: 
-                r["content"] = doc_content
+            if include_content: r["content"] = doc_content
                 
             if doc_content and valid_sources_count < 3:
                 best_chunks = rank_chunks_for_rag(doc_content, query_tokens)
@@ -657,8 +584,7 @@ async def process_aimode(q: str, include_content: bool, debug: bool, max_ai_toke
                     used_sources.append({"id": valid_sources_count+1, "title": r["title"], "url": r["url"]})
                     valid_sources_count += 1
         
-        for r in top_20:
-            r.pop("_internal_content", None)
+        for r in top_20: r.pop("_internal_content", None)
                     
     system_prompt = (
         "You are a helpful AI. Answer directly using the context. "
@@ -682,14 +608,9 @@ async def process_aimode(q: str, include_content: bool, debug: bool, max_ai_toke
     ai_reply = await run_ai_task(messages, max_tokens=max_ai_tokens, query_key=cache_key)
     
     return {
-        "request_id": req_id, 
-        "query": q,
-        "corrected_query": corrected_query,
-        "ai_response": ai_reply, 
-        "results_count": len(top_20), 
-        "results": top_20,
-        "sources": used_sources,
-        "cached": False
+        "request_id": req_id, "query": q, "corrected_query": corrected_query,
+        "ai_response": ai_reply, "results_count": len(top_20), 
+        "results": top_20, "sources": used_sources, "cached": False
     }
 
 @app.get("/aimode")