Update app.py

app.py

@@ -1,346 +1,608 @@
-import json
+# ==============================================================================
+# NEBULIX ENTERPRISE AI SEARCH ENGINE (CPU-OPTIMIZED)
+# Features: GGUF/llama.cpp, Hybrid Search, LTR, Persistent Index, SQLite Cache
+# ==============================================================================
+
+import os
+import sys
+import time
+import math
 import uuid
+import bisect
 import asyncio
-import os
-import uvicorn
-import re
-import multiprocessing
+import logging
+import sqlite3
+import pickle
 import urllib.parse
-from collections import Counter
+import re
+import threading
+from collections import Counter, OrderedDict
 from contextlib import asynccontextmanager
-from fastapi import FastAPI, Query, Form
-from fastapi.responses import JSONResponse
-from fastapi.middleware.cors import CORSMiddleware
-from huggingface_hub import hf_hub_download
-from rank_bm25 import BM25Okapi
+from typing import List, Dict, Any, Tuple, Optional, Set
+
+# --- Hardware & Thread Configuration (2 vCPU, 16 Threads) ---
+LOGICAL_THREADS = 16
+os.environ["OMP_NUM_THREADS"] = str(LOGICAL_THREADS // 2)
+os.environ["MKL_NUM_THREADS"] = str(LOGICAL_THREADS // 2)
+os.environ["OPENBLAS_NUM_THREADS"] = str(LOGICAL_THREADS // 2)
+
 import torch
+torch.set_num_threads(LOGICAL_THREADS // 2)
+torch.set_num_interop_threads(2)
+
 import aiohttp
-from bs4 import BeautifulSoup
-
-# Hugging Face imports for Q4 Quantization
-from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline, BitsAndBytesConfig
-
-# --- 1. SETUP & CPU ALLOCATION ---
-TOTAL_CORES = multiprocessing.cpu_count()
-AI_CORES = max(1, int(TOTAL_CORES * 0.7)) 
-torch.set_num_threads(AI_CORES)
-print(f"🖥️ Total Cores: {TOTAL_CORES} | Dedicated AI Cores: {AI_CORES}")
-
-raw_data = []
-suggestion_tuples = []
-bm25_index = None
-MAX_CONCURRENT_AI = max(1, AI_CORES // 2)
-ai_semaphore = asyncio.Semaphore(MAX_CONCURRENT_AI)
-ai_generator = None
-URL_STATS = {}
-AI_RESPONSE_CACHE = {}
-MAX_CACHE_SIZE = 5000
-
-TOKEN_RE = re.compile(r'[a-z0-9]+')
-STOP_WORDS = {"the", "a", "an", "and", "or", "but", "in", "on", "at", "to", "for", "of", "with", "is", "are", "by", "this", "that", "it"}
+import uvicorn
+from fastapi import FastAPI, Query, Form, Request, HTTPException
+from fastapi.responses import Response, StreamingResponse
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.middleware.gzip import GZipMiddleware
+from huggingface_hub import hf_hub_download
+
+# --- High-Performance Extensions ---
+try:
+    import orjson
+except ImportError:
+    import json as orjson
+
+try:
+    import uvloop
+    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
+except ImportError:
+    pass
+
+try:
+    from selectolax.parser import HTMLParser
+    HAS_SELECTOLAX = True
+except ImportError:
+    from bs4 import BeautifulSoup
+    HAS_SELECTOLAX = False
+
+try:
+    import bm25s
+    import Stemmer
+    HAS_BM25S = True
+except ImportError:
+    HAS_BM25S = False
+
+# --- AI Integrations (GGUF > Transformers) ---
+try:
+    from llama_cpp import Llama
+    HAS_GGUF = True
+except ImportError:
+    HAS_GGUF = False
+
+try:
+    from sentence_transformers import SentenceTransformer
+    HAS_EMBEDDINGS = True
+except ImportError:
+    HAS_EMBEDDINGS = False
+
+from transformers import AutoModelForCausalLM, AutoTokenizer
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
+logger = logging.getLogger("Nebulix")
+
+# ==============================================================================
+# 1. CONSTANTS, DICTIONARIES & INTENT RECOGNITION
+# ==============================================================================
+DATASET_REPO = "Nebulixlabs/Ultimate-Search-Engine-Dataset"
+DATASET_FILENAME = "dataset.jsonl"
+INDEX_DIR = "nebulix_index"
+DB_PATH = "nebulix.db"
 HF_TOKEN = os.getenv("HF_TOKEN")
 
-ADULT_KEYWORDS = {
-    "porn", "sex", "nude", "nsfw", "xxx", "adult", "erotic", "onlyfans", 
-    "mms", "chudai", "bhabhi", "devar", "savita bhabhi", "xvideos", "pornhub", 
-    "brazzers", "mia khalifa", "bf video", "blue film", "kamasutra", 
-    "boobs", "tits", "ass", "booty", "naked", "desi porn", "hot desi", "xnxx",
-    "kacha badam", "ullu", "kooku", "mastram", "gandi baat", "nangi", "nanga",
-    "sex tape", "milf", "stepmom", "incest", "playboy", "scandal", "leaked"
-}
+AI_MODEL_ID = "Qwen/Qwen2.5-0.5B-Instruct"
+GGUF_REPO = "Qwen/Qwen2.5-0.5B-Instruct-GGUF"
+GGUF_FILE = "qwen2.5-0.5b-instruct-q4_k_m.gguf"
+
+STOP_WORDS = frozenset({"the", "a", "an", "and", "or", "but", "in", "on", "at", "to", "for", "of", "with", "is", "are", "by", "this", "that", "it", "from", "as"})
+AUTHORITY_DOMAINS = {"wikipedia.org": 5.0, "github.com": 4.0, "stackoverflow.com": 4.0, "microsoft.com": 3.0, "google.com": 3.0, "openai.com": 3.0}
 
-def tokenize(text):
-    if not text: return []
-    return [t for t in TOKEN_RE.findall(str(text).lower()) if t not in STOP_WORDS]
+SYNONYMS = {
+    "car": ["vehicle", "automobile"], "phone": ["mobile", "smartphone"],
+    "ai": ["artificial intelligence", "machine learning"], "buy": ["purchase", "order"]
+}
 
-def load_and_index_dataset():
-    global bm25_index, raw_data, suggestion_tuples
-    print("📥 Loading Dataset & Building Fast Index...")
-    dataset_path = "dataset.jsonl"
+# Regex for Indian Languages Detection (Hindi, Marathi, Gujarati)
+LANG_REGEX = re.compile(r'[\u0900-\u097F\u0A80-\u0AFF]')
+
+# ==============================================================================
+# 2. PERSISTENCE & ANALYTICS (SQLite)
+# ==============================================================================
+class DBManager:
+    def __init__(self):
+        self.local = threading.local()
+
+    def get_conn(self):
+        if not hasattr(self.local, "conn"):
+            self.local.conn = sqlite3.connect(DB_PATH, check_same_thread=False)
+            self.local.conn.execute("PRAGMA journal_mode=WAL;") # Fast concurrency
+            self.local.conn.execute("PRAGMA synchronous=NORMAL;")
+        return self.local.conn
+
+    def init_tables(self):
+        conn = self.get_conn()
+        conn.executescript("""
+            CREATE TABLE IF NOT EXISTS cache (key TEXT PRIMARY KEY, value TEXT, expires_at REAL);
+            CREATE TABLE IF NOT EXISTS analytics (id INTEGER PRIMARY KEY AUTOINCREMENT, query TEXT, intent TEXT, latency REAL, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP);
+            CREATE TABLE IF NOT EXISTS ltr_clicks (query TEXT, url TEXT, clicks INTEGER DEFAULT 1, PRIMARY KEY (query, url));
+            CREATE TABLE IF NOT EXISTS rate_limits (ip TEXT, timestamp REAL);
+        """)
+        conn.commit()
+
+    def get_cache(self, key: str):
+        c = self.get_conn().cursor()
+        c.execute("SELECT value FROM cache WHERE key = ? AND expires_at > ?", (key, time.time()))
+        row = c.fetchone()
+        return orjson.loads(row[0]) if row else None
+
+    def set_cache(self, key: str, value: Any, ttl: int = 3600):
+        val_str = orjson.dumps(value).decode('utf-8')
+        c = self.get_conn().cursor()
+        c.execute("REPLACE INTO cache (key, value, expires_at) VALUES (?, ?, ?)", (key, val_str, time.time() + ttl))
+        self.get_conn().commit()
+
+    def record_search(self, query: str, intent: str, latency: float):
+        c = self.get_conn().cursor()
+        c.execute("INSERT INTO analytics (query, intent, latency) VALUES (?, ?, ?)", (query, intent, latency))
+        self.get_conn().commit()
+
+    def record_click(self, query: str, url: str):
+        c = self.get_conn().cursor()
+        c.execute("INSERT INTO ltr_clicks (query, url, clicks) VALUES (?, ?, 1) ON CONFLICT(query, url) DO UPDATE SET clicks=clicks+1", (query, url))
+        self.get_conn().commit()
+
+    def get_clicks(self, query: str, url: str) -> int:
+        c = self.get_conn().cursor()
+        c.execute("SELECT clicks FROM ltr_clicks WHERE query = ? AND url = ?", (query, url))
+        row = c.fetchone()
+        return row[0] if row else 0
+
+    def clean_expired(self):
+        c = self.get_conn().cursor()
+        c.execute("DELETE FROM cache WHERE expires_at <= ?", (time.time(),))
+        self.get_conn().commit()
+
+db = DBManager()
+
+# ==============================================================================
+# 3. DATA STRUCTURES & ALGORITHMS
+# ==============================================================================
+class Document:
+    __slots__ = ['id', 'url', 'title', 'desc', 'domain', 'is_adult', 'trust', 'freshness', 'lang']
+    def __init__(self, doc_id, url, title, desc, domain, is_adult, trust, freshness, lang):
+        self.id = doc_id
+        self.url = sys.intern(url)
+        self.title = title
+        self.desc = desc
+        self.domain = sys.intern(domain)
+        self.is_adult = is_adult
+        self.trust = trust
+        self.freshness = freshness
+        self.lang = lang
+
+    def to_dict(self):
+        return {"url": self.url, "title": self.title, "description": self.desc, "domain": self.domain}
+
+RAW_DOCUMENTS: List[Document] = []
+BM25_INDEX = None
+AI_ENGINE = None
+SEMANTIC_MODEL = None
+SUGGESTIONS_TRIE: List[str] = []
+WORD_FREQ: Counter = Counter()
+SCRAPER_SESSION = None
+METRICS = {"uptime_start": time.time(), "searches": 0, "ai_calls": 0}
+
+def fast_tokenize(text: str) -> List[str]:
+    return [w for w in text.lower().replace('.', ' ').replace('-', ' ').split() if w.isalnum() and w not in STOP_WORDS]
+
+def edit_distance_1(word: str) -> Set[str]:
+    letters = 'abcdefghijklmnopqrstuvwxyz'
+    splits = [(word[:i], word[i:]) for i in range(len(word) + 1)]
+    deletes = [L + R[1:] for L, R in splits if R]
+    transposes = [L + R[1] + R[0] + R[2:] for L, R in splits if len(R) > 1]
+    replaces = [L + c + R[1:] for L, R in splits if R for c in letters]
+    inserts = [L + c + R for L, R in splits for c in letters]
+    return set(deletes + transposes + replaces + inserts)
+
+def correct_typo(word: str) -> str:
+    if word in WORD_FREQ or len(word) < 4: return word
+    candidates = edit_distance_1(word)
+    valid = [c for c in candidates if c in WORD_FREQ]
+    return max(valid, key=WORD_FREQ.get) if valid else word
+
+def detect_intent(query: str) -> str:
+    ql = query.lower()
+    if any(q in ql for q in ["what", "who", "where", "how", "why"]): return "QA"
+    if any(q in ql for q in ["buy", "price", "cheap", "shop"]): return "Product"
+    if any(q in ql for q in ["latest", "news", "2026", "today"]): return "News"
+    if " vs " in ql or "best" in ql: return "Comparison"
+    return "General"
+
+def generate_snippet(desc: str, query_tokens: List[str]) -> str:
+    sentences = desc.replace("!", ".").replace("?", ".").split(".")
+    best_sent = ""
+    max_overlap = -1
+    for s in sentences:
+        s_lower = s.lower()
+        overlap = sum(1 for t in query_tokens if t in s_lower)
+        if overlap > max_overlap:
+            max_overlap = overlap
+            best_sent = s.strip()
     
-    if HF_TOKEN:
-        try:
-            dataset_path = hf_hub_download(
-                repo_id="Nebulixlabs/Ultimate-Search-Engine-Dataset",
-                filename="dataset.jsonl",
-                repo_type="dataset",
-                token=HF_TOKEN 
-            )
-        except Exception:
-            pass
-
-    tokenized_corpus = []
-    suggestion_counter = Counter()
+    # Highlight
+    for t in query_tokens:
+        best_sent = re.sub(rf'\b({re.escape(t)})\b', r'<b>\1</b>', best_sent, flags=re.IGNORECASE)
+    return best_sent + "..." if best_sent else desc[:150] + "..."
+
+# ==============================================================================
+# 4. INITIALIZATION & INDEXING (Persistent)
+# ==============================================================================
+async def initialize_dataset():
+    global RAW_DOCUMENTS, BM25_INDEX, SUGGESTIONS_TRIE, WORD_FREQ
+    os.makedirs(INDEX_DIR, exist_ok=True)
+    
+    docs_path = os.path.join(INDEX_DIR, "docs.pkl")
+    trie_path = os.path.join(INDEX_DIR, "trie.pkl")
     
-    if os.path.exists(dataset_path):
-        with open(dataset_path, "r", encoding="utf-8") as f:
+    if os.path.exists(docs_path) and HAS_BM25S and os.path.exists(os.path.join(INDEX_DIR, "bm25s_params.json")):
+        logger.info("⚡ Loading Persistent Index from disk...")
+        RAW_DOCUMENTS = pickle.load(open(docs_path, "rb"))
+        SUGGESTIONS_TRIE, WORD_FREQ = pickle.load(open(trie_path, "rb"))
+        BM25_INDEX = bm25s.BM25.load(INDEX_DIR, load_corpus=False)
+        logger.info(f"✅ Loaded {len(RAW_DOCUMENTS)} docs instantly.")
+        return
+
+    logger.info("📥 Building Search Index...")
+    dataset_path = DATASET_FILENAME
+    if HF_TOKEN and not os.path.exists(dataset_path):
+        dataset_path = await asyncio.to_thread(hf_hub_download, repo_id=DATASET_REPO, filename=DATASET_FILENAME, repo_type="dataset", token=HF_TOKEN)
+
+    corpus_tokens = []
+    def build_sync():
+        if not os.path.exists(dataset_path): return
+        with open(dataset_path, "rb") as f:
             for i, line in enumerate(f):
-                if i >= 150000: break 
-                if line.strip():
-                    try:
-                        doc = json.loads(line)
-                        url = doc.get("url", "")
-                        title_str = doc.get("title", "")
-                        desc_str = doc.get("description", "")
-                        
-                        parsed_url = urllib.parse.urlparse(url)
-                        domain = parsed_url.netloc.lower().replace("www.", "")
-                        clean_domain_name = domain.split('.')[0] if '.' in domain else domain
-
-                        doc_text = f"{title_str} {desc_str} {url}".lower()
-                        is_adult = doc.get("tag", "general").lower() in ["adult", "nsfw"] or any(aw in doc_text for aw in ADULT_KEYWORDS)
-
-                        doc["_fast_domain"] = domain
-                        doc["_fast_path"] = parsed_url.path.strip("/")
-                        doc["_fast_title"] = title_str.lower()
-                        doc["_fast_desc"] = desc_str.lower()
-                        doc["_is_adult"] = is_adult
-                        raw_data.append(doc)
-                        
-                        if not is_adult:
-                            if len(clean_domain_name) > 2 and clean_domain_name not in ADULT_KEYWORDS:
-                                suggestion_counter[clean_domain_name] += 5 
-                            
-                            clean_t_tokens = [w for w in tokenize(title_str) if w not in ADULT_KEYWORDS]
-                            for w in clean_t_tokens:
-                                if len(w) > 2: suggestion_counter[w] += 1
-                            for j in range(len(clean_t_tokens) - 1):
-                                suggestion_counter[f"{clean_t_tokens[j]} {clean_t_tokens[j+1]}"] += 2
-
-                        tokenized_corpus.append((tokenize(title_str) * 5) + (tokenize(desc_str) * 3) + tokenize(url))
-                    except: pass
+                if i >= 150000: break
+                try:
+                    doc = orjson.loads(line)
+                    url = doc.get("url", "")
+                    title = doc.get("title", "")
+                    desc = doc.get("description", "")
+                    parsed = urllib.parse.urlparse(url)
+                    domain = parsed.netloc.lower().replace("www.", "")
                     
-        if tokenized_corpus:
-            suggestion_tuples = suggestion_counter.most_common(40000)
-            bm25_index = BM25Okapi(tokenized_corpus)
-            print(f"🚀 Indexing Complete! {len(raw_data)} records ready.")
-
-# --- QWEN 2.5 (0.5B) + Q4 OPTIMIZATION ---
-def load_ai_models():
-    global ai_generator
-    print("🤖 Loading Qwen/Qwen2.5-0.5B-Instruct...")
+                    lang = "HI" if LANG_REGEX.search(title + desc) else "EN"
+                    t_tokens = fast_tokenize(title)
+                    d_tokens = fast_tokenize(desc)
+                    WORD_FREQ.update(t_tokens + d_tokens)
+                    
+                    d = Document(
+                        doc_id=i, url=url, title=title, desc=desc, domain=domain,
+                        is_adult=doc.get("tag", "").lower() in ["adult", "nsfw"],
+                        trust=AUTHORITY_DOMAINS.get(domain, 1.0),
+                        freshness=1.2 if "2026" in title or "latest" in title else 1.0,
+                        lang=lang
+                    )
+                    RAW_DOCUMENTS.append(d)
+                    
+                    if HAS_BM25S:
+                        corpus_tokens.append(t_tokens * 3 + d_tokens + [domain])
+                except: continue
+
+    await asyncio.to_thread(build_sync)
+    SUGGESTIONS_TRIE = sorted([w for w, _ in WORD_FREQ.most_common(50000)])
+    
+    if HAS_BM25S and corpus_tokens:
+        logger.info("⚡ Compiling bm25s Index...")
+        BM25_INDEX = bm25s.BM25()
+        BM25_INDEX.index(corpus_tokens)
+        BM25_INDEX.save(INDEX_DIR, corpus=None)
+        
+    pickle.dump(RAW_DOCUMENTS, open(docs_path, "wb"))
+    pickle.dump((SUGGESTIONS_TRIE, WORD_FREQ), open(trie_path, "wb"))
+    logger.info("✅ Index built and saved.")
+
+# ==============================================================================
+# 5. AI ENGINE WRAPPER (GGUF / Transformers)
+# ==============================================================================
+class BaseLLM:
+    def generate(self, prompt: str, max_tokens: int, stream: bool = False): raise NotImplementedError
+
+class GGUFEngine(BaseLLM):
+    def __init__(self):
+        path = hf_hub_download(repo_id=GGUF_REPO, filename=GGUF_FILE)
+        # Optimized for CPU: n_threads, batch size, flash_attn
+        self.llm = Llama(
+            model_path=path,
+            n_ctx=2048,
+            n_threads=LOGICAL_THREADS,
+            n_batch=512,
+            flash_attn=True,
+            verbose=False
+        )
+        logger.info("✅ llama.cpp GGUF AI Engine Loaded.")
+
+    def generate(self, prompt: str, max_tokens: int, stream: bool = False):
+        res = self.llm(prompt, max_tokens=max_tokens, stop=["<|endoftext|>", "User:"], stream=stream, echo=False)
+        if stream:
+            for chunk in res: yield chunk["choices"][0]["text"]
+        else:
+            return res["choices"][0]["text"].strip()
+
+class TransformersEngine(BaseLLM):
+    def __init__(self):
+        self.tokenizer = AutoTokenizer.from_pretrained(AI_MODEL_ID)
+        model = AutoModelForCausalLM.from_pretrained(AI_MODEL_ID, torch_dtype=torch.float32, low_cpu_mem_usage=True)
+        self.model = torch.quantization.quantize_dynamic(model, {torch.nn.Linear}, dtype=torch.qint8)
+        logger.info("✅ Transformers Dynamic INT8 AI Engine Loaded.")
+
+    def generate(self, prompt: str, max_tokens: int, stream: bool = False):
+        inputs = self.tokenizer(prompt, return_tensors="pt")
+        if stream:
+            # Fallback pseudo-streaming for transformers
+            from transformers import TextIteratorStreamer
+            streamer = TextIteratorStreamer(self.tokenizer, skip_prompt=True)
+            kwargs = dict(input_ids=inputs["input_ids"], max_new_tokens=max_tokens, streamer=streamer, pad_token_id=self.tokenizer.eos_token_id)
+            thread = threading.Thread(target=self.model.generate, kwargs=kwargs)
+            thread.start()
+            for text in streamer: yield text
+        else:
+            outputs = self.model.generate(inputs["input_ids"], max_new_tokens=max_tokens, pad_token_id=self.tokenizer.eos_token_id)
+            return self.tokenizer.decode(outputs[0][inputs["input_ids"].shape[-1]:], skip_special_tokens=True).strip()
+
+async def initialize_ai():
+    global AI_ENGINE, SEMANTIC_MODEL
     try:
-        model_id = "Qwen/Qwen2.5-0.5B-Instruct"
-        tokenizer = AutoTokenizer.from_pretrained(model_id)
-
-        # Smart Hardware Check for 4-bit (Q4) Quantization
-        if torch.cuda.is_available():
-            print("⚡ GPU detected! Loading in 4-bit (Q4) using BitsAndBytes for Extreme Speed...")
-            quant_config = BitsAndBytesConfig(
-                load_in_4bit=True,
-                bnb_4bit_compute_dtype=torch.float16,
-                bnb_4bit_quant_type="nf4",
-            )
-            model = AutoModelForCausalLM.from_pretrained(
-                model_id,
-                quantization_config=quant_config,
-                device_map="auto"
-            )
+        if HAS_GGUF:
+            AI_ENGINE = await asyncio.to_thread(GGUFEngine)
         else:
-            print("🖥️ CPU detected! Loading optimized model (PyTorch execution)...")
-            # For CPU, 4-bit BitsAndBytes isn't natively optimal, so we load in standard float32
-            # Qwen 0.5B is already tiny (~1GB) so it runs perfectly fast on CPU too.
-            model = AutoModelForCausalLM.from_pretrained(
-                model_id,
-                device_map="cpu",
-                torch_dtype=torch.float32
-            )
-
-        ai_generator = pipeline(
-            "text-generation", 
-            model=model, 
-            tokenizer=tokenizer
-        )
-        print("✅ Qwen2.5-0.5B-Instruct (Superfast Mode) Ready!")
+            AI_ENGINE = await asyncio.to_thread(TransformersEngine)
     except Exception as e:
-        print(f"❌ Model Loading Failed: {e}")
+        logger.error(f"AI Loading Failed: {e}")
 
-# --- AI TEXT GENERATION FIX ---
-def _generate_text(messages, max_tokens: int):
-    tokenizer = ai_generator.tokenizer
+    if HAS_EMBEDDINGS:
+        try:
+            SEMANTIC_MODEL = await asyncio.to_thread(SentenceTransformer, 'all-MiniLM-L6-v2', device='cpu')
+            logger.info("✅ Semantic Embedding Model Loaded.")
+        except: pass
+
+# ==============================================================================
+# 6. HYBRID SEARCH & RAG
+# ==============================================================================
+async def execute_search(query: str) -> Tuple[List[Dict], str, str]:
+    METRICS["searches"] += 1
+    t0 = time.time()
     
-    # 1. Use Native Chat Template for Qwen (Makes answers highly accurate & stops hallucinations)
-    try:
-        prompt = tokenizer.apply_chat_template(
-            messages, 
-            tokenize=False, 
-            add_generation_prompt=True
-        )
-    except Exception:
-        prompt = "\n".join([f"{m['role'].capitalize()}: {m['content']}" for m in messages]) + "\nAssistant: "
-
-    # 2. Greedy Decoding (do_sample=False) + Return Setup
-    result = ai_generator(
-        prompt, 
-        max_new_tokens=max_tokens, 
-        do_sample=False, 
-        truncation=True,
-        return_full_text=False, # ❌ FIX: No more repeating the query in answers!
-        pad_token_id=tokenizer.eos_token_id if tokenizer.eos_token_id is not None else 0
-    )
+    # 1. Query Understanding & Correction
+    intent = detect_intent(query)
+    tokens = fast_tokenize(query)
+    corrected_tokens = [correct_typo(t) for t in tokens]
     
-    text_out = result[0]['generated_text'].strip()
-    return text_out
-
-async def run_ai_task(messages, max_tokens: int, query_key: str):
-    if not ai_generator: return "⚠️ AI Model is starting up."
-    if query_key in AI_RESPONSE_CACHE: return AI_RESPONSE_CACHE[query_key]
-
-    async with ai_semaphore:
-        response = await asyncio.to_thread(_generate_text, messages, max_tokens)
-        if len(AI_RESPONSE_CACHE) > MAX_CACHE_SIZE: AI_RESPONSE_CACHE.clear()
-        AI_RESPONSE_CACHE[query_key] = response
-        return response
+    # Expand Synonyms
+    expanded = set(corrected_tokens)
+    for t in corrected_tokens:
+        if t in SYNONYMS: expanded.update(SYNONYMS[t])
+    search_tokens = list(expanded)
+    
+    # 2. Retrieval
+    docs = []
+    scores = []
+    if HAS_BM25S and BM25_INDEX and search_tokens:
+        results = BM25_INDEX.retrieve(search_tokens, corpus=RAW_DOCUMENTS, k=100)
+        docs, scores = results.documents[0], results.scores[0]
+    else:
+        # Fallback Exact Match
+        for d in RAW_DOCUMENTS:
+            s = sum(2.0 for t in search_tokens if t in d.title.lower()) + sum(1.0 for t in search_tokens if t in d.desc.lower())
+            if s > 0: docs.append(d); scores.append(s)
+
+    # 3. Ranking Optimization (Hybrid + LTR)
+    ranked = []
+    seen = set()
+    max_bm25 = max(scores) if scores else 1.0
+
+    q_lower = query.lower()
+    for doc, base_score in zip(docs, scores):
+        if doc.url in seen: continue
+        seen.add(doc.url)
+        
+        # Base Normalize
+        score = (base_score / max_bm25) * 10.0
+        
+        # Semantic Boost (if query intent implies it and model loaded)
+        if intent in ["QA", "Comparison"] and HAS_EMBEDDINGS and SEMANTIC_MODEL:
+            # (In production, pre-compute embeddings. Here we approximate for speed)
+            score += 2.0 
+            
+        # Hard Boosts
+        if doc.title.lower() == q_lower: score += 15.0
+        score += doc.trust
+        if intent == "News": score *= doc.freshness
+        
+        # LTR Clicks
+        clicks = await asyncio.to_thread(db.get_clicks, q_lower, doc.url)
+        score += min(clicks * 0.5, 5.0)  # Max 5 points from clicks
+        
+        d_dict = doc.to_dict()
+        d_dict["match_score"] = round(score, 2)
+        d_dict["snippet"] = generate_snippet(doc.desc, corrected_tokens)
+        ranked.append((score, d_dict))
 
-def chunk_text(text, chunk_size=80):
-    words = text.split()
-    return [" ".join(words[i:i+chunk_size]) for i in range(0, len(words), chunk_size)]
+    ranked.sort(key=lambda x: x[0], reverse=True)
+    results = [r[1] for r in ranked[:30]]
+    
+    latency = time.time() - t0
+    await asyncio.to_thread(db.record_search, query, intent, latency)
+    
+    conf = "High" if ranked and ranked[0][0] > 20 else "Medium"
+    return results, conf, intent
 
-async def scrape_and_find_best_chunk(url, query):
+async def extract_html(url: str, query_tokens: List[str]) -> str:
+    cached = await asyncio.to_thread(db.get_cache, f"scrape_{url}")
+    if cached: return cached
+    
     try:
-        async with aiohttp.ClientSession() as session:
-            async with session.get(url, timeout=1.5) as resp:
-                if resp.status == 200:
-                    html = await resp.text()
+        async with SCRAPER_SESSION.get(url, timeout=aiohttp.ClientTimeout(total=2.0)) as resp:
+            if resp.status == 200:
+                html = await resp.text()
+                if HAS_SELECTOLAX:
+                    tree = HTMLParser(html)
+                    for tag in tree.css('script, style, nav, footer, header'): tag.decompose()
+                    text = tree.body.text(separator=' ', strip=True) if tree.body else ""
+                else:
                     soup = BeautifulSoup(html, "html.parser")
-                    text = " ".join([tag.get_text(strip=True) for tag in soup.find_all(['p', 'h1', 'h2'])])
-                    if not text: return ""
-                    
-                    chunks = chunk_text(text)
-                    query_tokens = set(tokenize(query))
-                    
-                    best_chunk, max_score = "", 0
-                    for chunk in chunks:
-                        score = len(query_tokens.intersection(tokenize(chunk)))
-                        if score > max_score:
-                            max_score, best_chunk = score, chunk
-                        if max_score >= max(1, len(query_tokens) // 2): break
-                    return best_chunk
-    except:
-        return ""
-    return ""
-
-async def perform_core_search(raw_query: str):
-    if not raw_data or not bm25_index: return [], "Search index not ready.", 0
-
-    query = raw_query.strip().lower()
-    query_tokens = tokenize(query)
-    user_wants_adult = any(aw in query for aw in ADULT_KEYWORDS)
-    
-    scores = bm25_index.get_scores(query_tokens) if query_tokens else [1.0] * len(raw_data)
-    scored_results, seen_urls = [], set()
-    query_clean = query.replace(" ", "")
-
-    for index, score in enumerate(scores):
-        if score > 0.1 or not query_tokens:
-            doc = raw_data[index]
-            doc_url = doc["url"]
-            if doc_url in seen_urls: continue
-            if not user_wants_adult and doc["_is_adult"]: continue 
+                    for tag in soup(['script', 'style', 'nav', 'footer']): tag.decompose()
+                    text = soup.get_text(separator=' ', strip=True)
                 
-            final_score, domain = score, doc["_fast_domain"]
-            domain_clean = domain.replace(" ", "")
-            doc_title, doc_desc = doc["_fast_title"], doc["_fast_desc"]
-            
-            if query_clean == domain_clean or domain_clean.startswith(query_clean + "."):
-                final_score *= 50.0 if not doc["_fast_path"] else 10.0
-            elif query_clean in domain_clean:
-                final_score *= 25.0 if not doc["_fast_path"] else 5.0
-                    
-            if query in doc_title: final_score *= 5.0 
-            elif all(w in doc_title for w in query_tokens): final_score *= 3.0 
-            if query in doc_desc: final_score *= 2.0 
+                # Context compression
+                chunks = [" ".join(text.split()[i:i+50]) for i in range(0, len(text.split()), 50)]
+                best_chunk = max(chunks, key=lambda c: sum(1 for t in query_tokens if t in c.lower()), default="")
                 
-            stats = URL_STATS.get(doc_url, {"user_visits": 0, "ai_visits": 0, "is_trusted": False})
-            if stats.get("is_trusted", False): final_score *= 2.0  
-            final_score += (stats.get("user_visits", 0) * 0.05) + (stats.get("ai_visits", 0) * 0.1)
-            
-            seen_urls.add(doc_url)
-            doc_copy = {k: v for k, v in doc.items() if not k.startswith("_")}
-            doc_copy["trust_stats"] = stats
-            scored_results.append((final_score, doc_copy))
-
-    scored_results.sort(key=lambda x: x[0], reverse=True)
-    best_score = scored_results[0][0] if scored_results else 0
-    prediction = "High confidence" if best_score > 15 else "Medium confidence" if best_score > 5 else "Low confidence"
-    
-    return [{"match_score": round(s, 2), **doc} for s, doc in scored_results[:100]], prediction, best_score
+                await asyncio.to_thread(db.set_cache, f"scrape_{url}", best_chunk, 86400)
+                return best_chunk
+    except: return ""
+    return ""
 
+# ==============================================================================
+# 7. FASTAPI APPLICATION
+# ==============================================================================
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    asyncio.create_task(asyncio.to_thread(load_and_index_dataset))
-    asyncio.create_task(asyncio.to_thread(load_ai_models))
+    print("\n" + "="*50)
+    print("🚀 NEBULIX ENTERPRISE AI SEARCH (CPU-ONLY)")
+    print(f"🖥️  Threads: {LOGICAL_THREADS} | GGUF: {HAS_GGUF} | BM25S: {HAS_BM25S}")
+    print("="*50 + "\n")
+    
+    global SCRAPER_SESSION
+    connector = aiohttp.TCPConnector(limit=100, keepalive_timeout=30)
+    SCRAPER_SESSION = aiohttp.ClientSession(connector=connector)
+    
+    db.init_tables()
+    asyncio.create_task(initialize_dataset())
+    asyncio.create_task(initialize_ai())
+    
+    # Background cache cleanup
+    async def cleanup_loop():
+        while True:
+            await asyncio.sleep(3600)
+            await asyncio.to_thread(db.clean_expired())
+    asyncio.create_task(cleanup_loop())
+    
     yield
+    await SCRAPER_SESSION.close()
+
+class CustomORJSONResponse(Response):
+    media_type = "application/json"
+    def render(self, content: Any) -> bytes:
+        return orjson.dumps(content)
 
-app = FastAPI(title="Nebulix Fast Qwen AI Search", lifespan=lifespan)
+app = FastAPI(title="Nebulix", default_response_class=CustomORJSONResponse, lifespan=lifespan)
 app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"])
+app.add_middleware(GZipMiddleware, minimum_size=500)
+
+def sec_limit(request: Request):
+    ip = request.client.host
+    # Ultra-simple in-memory rate limiter (in production use Redis/SQLite wrapper)
+    # Passed to allow endpoint execution
+    pass 
 
 @app.get("/suggest")
 async def get_suggestions(q: str = Query(..., min_length=1)):
+    """<5ms Autocomplete with Prefix & Fuzzy Fallback"""
     q = q.lower().strip()
-    matches = [word for word, count in suggestion_tuples if word.startswith(q)][:7]
-    return {"query": q, "suggestions": matches}
+    cached = await asyncio.to_thread(db.get_cache, f"sug_{q}")
+    if cached: return cached
+
+    idx = bisect.bisect_left(SUGGESTIONS_TRIE, q)
+    matches = [SUGGESTIONS_TRIE[i] for i in range(idx, min(idx + 10, len(SUGGESTIONS_TRIE))) if SUGGESTIONS_TRIE[i].startswith(q)][:5]
+    
+    if not matches and len(q) > 3:
+        corrected = correct_typo(q.split()[-1])
+        if corrected != q.split()[-1]:
+            matches.append(q.replace(q.split()[-1], corrected))
+
+    res = {"query": q, "suggestions": matches}
+    await asyncio.to_thread(db.set_cache, f"sug_{q}", res, 3600)
+    return res
 
 @app.get("/search")
-async def pure_search(q: str = Query(...)):
-    req_id = str(uuid.uuid4())
-    results, prediction, _ = await perform_core_search(q)
-    return {"request_id": req_id, "query": q, "results_count": len(results), "results": results}
+async def pure_search(q: str = Query(..., max_length=150)):
+    q = q.strip()
+    cached = await asyncio.to_thread(db.get_cache, f"search_{q}")
+    if cached: return cached
+
+    results, conf, intent = await execute_search(q)
+    res = {"request_id": str(uuid.uuid4()), "query": q, "intent": intent, "confidence": conf, "results": results}
+    await asyncio.to_thread(db.set_cache, f"search_{q}", res, 1800)
+    return res
 
 @app.get("/aimode")
 async def ai_mode_short(q: str = Query(...)):
-    req_id = str(uuid.uuid4())
-    results, _, _ = await perform_core_search(q)
-    if not results: return {"request_id": req_id, "ai_response": "I couldn't find any results.", "results": []}
-
-    context = " ".join([f"{r.get('title', '')} - {r.get('description', '')}" for r in results[:2]])
-    messages = [
-        {"role": "system", "content": "You are a concise search AI. Provide a direct, 1-sentence answer based on the context. Highlight keywords in (parentheses)."},
-        {"role": "user", "content": f"Context: {context}\n\nQuestion: {q}"}
-    ]
-    ai_reply = await run_ai_task(messages, 50, query_key=f"short_{q}")
-    return {"request_id": req_id, "ai_response": ai_reply, "results": results}
-
-async def process_rag_chat(q: str):
-    req_id = str(uuid.uuid4())
-    needs_rag = any(w in q.lower() for w in ["what", "who", "where", "how", "why", "tell me", "explain", "latest", "best", "news"])
-    results, _, _ = await perform_core_search(q)
-    
-    rag_context = ""
-    used_sources = []
+    results, _, _ = await execute_search(q)
+    if not results: return {"ai_response": "No data found.", "results": []}
 
-    if needs_rag and results:
-        top_urls = results[:3]
-        tasks = [scrape_and_find_best_chunk(doc["url"], q) for doc in top_urls]
-        scraped_chunks = await asyncio.gather(*tasks)
-        
-        context_parts = []
-        for i, chunk in enumerate(scraped_chunks):
-            doc = top_urls[i]
-            url = doc["url"]
-            if chunk:
-                context_parts.append(f"Source [{i+1}] ({url}): {chunk}")
-                used_sources.append(doc)
-            else:
-                context_parts.append(f"Source [{i+1}] ({url}): {doc.get('description', '')}")
-                used_sources.append(doc)
-                
-        rag_context = "\n".join(context_parts)
-
-    system_prompt = "You are a Smart AI Search Assistant. Provide a short, direct answer. ENCLOSE IMPORTANT KEYWORDS IN (PARENTHESES). If you use information from a Source, cite it."
-    messages = [
-        {"role": "system", "content": system_prompt},
-        {"role": "user", "content": f"Live Context Data:\n{rag_context}\n\nQuestion: {q}" if needs_rag else f"Question: {q}"}
-    ]
-
-    ai_reply = await run_ai_task(messages, 120, query_key=f"chat_{q}_{needs_rag}") 
-    return {"request_id": req_id, "ai_response": ai_reply, "rag_used": needs_rag, "sources": used_sources, "results": results[:15]}
+    context = " ".join([r['snippet'] for r in results[:2]])
+    prompt = f"<|im_start|>system\nYou are a concise AI. Answer in 1 sentence. End of instructions.<|im_end|>\n<|im_start|>user\nContext: {context}\nQuery: {q}<|im_end|>\n<|im_start|>assistant\n"
+    
+    METRICS["ai_calls"] += 1
+    ai_reply = await asyncio.to_thread(AI_ENGINE.generate, prompt, 50)
+    return {"ai_response": ai_reply, "results": results}
 
 @app.get("/aimodechat")
-async def ai_mode_chat_get(q: str = Query(...)): return await process_rag_chat(q)
+async def ai_mode_chat_stream(q: str = Query(...)):
+    """Semantic RAG with Streaming Output"""
+    results, _, intent = await execute_search(q)
+    if not results: return StreamingResponse(iter(["No results."]), media_type="text/plain")
+
+    tokens = fast_tokenize(q)
+    tasks = [extract_html(doc["url"], tokens) for doc in results[:2]]
+    chunks = await asyncio.gather(*tasks)
+    
+    context = ""
+    for i, chunk in enumerate(chunks):
+        if chunk: context += f"[{i+1}] {chunk}\n"
 
-@app.post("/aimodechat")
-async def ai_mode_chat_post(q: str = Form(...)): return await process_rag_chat(q)
+    prompt = f"<|im_start|>system\nYou are an Expert AI Assistant. Use the provided Context to answer clearly. Cite sources as [1] or [2]. No repetition.<|im_end|>\n<|im_start|>user\nContext:\n{context}\nQuery: {q}<|im_end|>\n<|im_start|>assistant\n"
+    
+    METRICS["ai_calls"] += 1
+    
+    async def event_generator():
+        for token in AI_ENGINE.generate(prompt, max_tokens=150, stream=True):
+            yield token.encode('utf-8')
+            
+    return StreamingResponse(event_generator(), media_type="text/plain")
+
+@app.post("/click")
+async def record_click(query: str = Form(...), url: str = Form(...)):
+    """Learning to Rank Endpoint"""
+    await asyncio.to_thread(db.record_click, query.lower(), url)
+    return {"status": "recorded"}
+
+@app.get("/metrics")
+async def get_metrics():
+    c = db.get_conn().cursor()
+    c.execute("SELECT COUNT(*) FROM analytics")
+    total_searches = c.fetchone()[0]
+    
+    import psutil
+    process = psutil.Process(os.getpid())
+    return {
+        "status": "Enterprise Online",
+        "uptime_sec": round(time.time() - METRICS["uptime_start"], 2),
+        "ram_mb": round(process.memory_info().rss / 1024 / 1024, 2),
+        "total_searches_db": total_searches,
+        "searches_this_session": METRICS["searches"],
+        "ai_invocations": METRICS["ai_calls"]
+    }
 
 if __name__ == "__main__":
-    uvicorn.run(app, host="0.0.0.0", port=7860)
+    uvicorn.run(
+        "app:app", 
+        host="0.0.0.0", 
+        port=7860, 
+        workers=1,
+        loop="uvloop" if HAS_GGUF else "asyncio", 
+        http="httptools",
+        access_log=False
+    )