Update app.py

app.py

@@ -3,178 +3,302 @@ import uuid
 import asyncio
 import os
 import uvicorn
+import aiohttp
+import io
+import zipfile
+from bs4 import BeautifulSoup
 from contextlib import asynccontextmanager
-from datetime import datetime, timedelta
-from urllib.parse import urlparse
+from datetime import datetime
+from urllib.parse import urlparse, urljoin
 from fastapi import FastAPI, Query, BackgroundTasks
-from fastapi.responses import JSONResponse, FileResponse # <--- FileResponse add kiya
-from rapidfuzz import process, fuzz
+from fastapi.responses import JSONResponse
+from fastapi.middleware.cors import CORSMiddleware
+from rapidfuzz import fuzz
 from cryptography.fernet import Fernet
 
-# --- 1. SETUP & ENCRYPTION ---
+# --- 1. SETUP & CONFIGURATION ---
 SECRET_KEY = Fernet.generate_key()
 cipher_suite = Fernet(SECRET_KEY)
 
-# Logs directory setup
 LOG_DIR = "logs"
+DATA_DIR = "."
 os.makedirs(LOG_DIR, exist_ok=True)
 
-# Global variables for dataset
 raw_data = []
 search_corpus = []
+unique_domains = set()
+visited_urls = set()
 
-# --- 2. OPTIMIZED CLEANUP TASK ---
-async def cleanup_old_logs():
-    while True:
-        try:
-            if os.path.exists(LOG_DIR):
-                thirty_days_ago = datetime.now() - timedelta(days=30)
-                
-                for filename in os.listdir(LOG_DIR):
-                    if filename.startswith("search_logs_") and filename.endswith(".jsonl"):
-                        date_part = filename.replace("search_logs_", "").replace(".jsonl", "")
-                        try:
-                            file_date = datetime.strptime(date_part, "%Y-%m-%d")
-                            if file_date < thirty_days_ago:
-                                os.remove(os.path.join(LOG_DIR, filename))
-                                print(f"🗑️ Deleted 1-month old log file: {filename}")
-                        except ValueError:
-                            pass
-        except Exception as e:
-            print("Cleanup Error:", e)
-        
-        # Har 24 ghante me ek baar check
-        await asyncio.sleep(24 * 60 * 60)
+# Max queue size to protect Basic CPU RAM
+scrape_queue = asyncio.Queue(maxsize=50000)
+merge_lock = asyncio.Lock()
 
-# --- 3. LIFESPAN (Replaces @app.on_event("startup")) ---
-@asynccontextmanager
-async def lifespan(app: FastAPI):
-    # a. Dataset Loading during startup
-    global raw_data, search_corpus
-    try:
+# Threshold for merging
+NEXT_THRESHOLD = 30000
+
+# --- 2. DATASET MANAGEMENT & DEDUPLICATION ---
+def get_domain(url):
+    return urlparse(url).netloc.lower()
+
+def process_item_into_memory(item):
+    url = item.get("url", "")
+    if url in visited_urls:
+        return # Avoid duplicate URL in memory
+
+    visited_urls.add(url)
+    domain = get_domain(url)
+    unique_domains.add(domain)
+    raw_data.append(item)
+    
+    search_corpus.append({
+        "url": url.lower(),
+        "domain": domain,
+        "title": item.get("title", "").lower(),
+        "description": item.get("description", "").lower()
+    })
+
+def load_all_data():
+    global NEXT_THRESHOLD
+    print("🔄 Scanning existing datasets and date files...")
+    
+    # 1. Load Main Dataset
+    if os.path.exists("dataset.jsonl"):
         with open("dataset.jsonl", "r", encoding="utf-8") as f:
             for line in f:
                 if line.strip():
-                    item = json.loads(line)
-                    raw_data.append(item)
+                    try:
+                        item = json.loads(line)
+                        process_item_into_memory(item)
+                        # Add domain base URL to scraper to find NEW pages on old domains
+                        base_url = f"https://{get_domain(item['url'])}/"
+                        if base_url not in visited_urls:
+                            try:
+                                scrape_queue.put_nowait(base_url)
+                            except asyncio.QueueFull:
+                                pass
+                    except: pass
+
+    # 2. Load ALL date-*.jsonl files
+    for filename in os.listdir(DATA_DIR):
+        if filename.startswith("date-") and filename.endswith(".jsonl"):
+            with open(filename, "r", encoding="utf-8") as f:
+                for line in f:
+                    if line.strip():
+                        try:
+                            item = json.loads(line)
+                            process_item_into_memory(item)
+                        except: pass
+
+    # Set Threshold (e.g., if we have 65k domains, next is 90k)
+    current_domains = len(unique_domains)
+    NEXT_THRESHOLD = ((current_domains // 30000) + 1) * 30000
+    print(f"✅ Startup Complete! Unique URLs: {len(visited_urls)} | Unique Domains: {current_domains} | Next Merge at: {NEXT_THRESHOLD}")
+
+async def check_and_merge_datasets():
+    global NEXT_THRESHOLD
+    if len(unique_domains) >= NEXT_THRESHOLD:
+        async with merge_lock:
+            print(f"🌟 Threshold {NEXT_THRESHOLD} Reached! Merging & Deduplicating Data...")
+            
+            # Since raw_data is already perfectly deduplicated by URL in memory, we just dump it!
+            temp_file = "dataset.jsonl.tmp"
+            with open(temp_file, "w", encoding="utf-8") as f:
+                for item in raw_data:
+                    f.write(json.dumps(item) + "\n")
+            
+            os.replace(temp_file, "dataset.jsonl")
+            
+            # Delete all date-*.jsonl files as they are merged
+            for filename in os.listdir(DATA_DIR):
+                if filename.startswith("date-") and filename.endswith(".jsonl"):
+                    os.remove(filename)
+                    print(f"🗑️ Deleted merged file: {filename}")
+            
+            NEXT_THRESHOLD += 30000
+            print(f"✅ Merge Successful! Next threshold is {NEXT_THRESHOLD}")
+
+
+# --- 3. AUTO TRANCO FETCHER & SPIDER CRAWLER ---
+async def fetch_tranco_seeds(session):
+    print("🌍 Downloading Tranco Top Domains dataset...")
+    try:
+        # Download Tranco top 1 million CSV ZIP
+        async with session.get("https://tranco-list.eu/top-1m.csv.zip", timeout=30) as resp:
+            if resp.status == 200:
+                content = await resp.read()
+                with zipfile.ZipFile(io.BytesIO(content)) as z:
+                    csv_filename = z.namelist()[0]
+                    with z.open(csv_filename) as f:
+                        lines = f.read().decode('utf-8').splitlines()
+                        # Take top 5000 domains to start seeding
+                        for line in lines[:5000]:
+                            parts = line.split(",")
+                            if len(parts) >= 2:
+                                domain = parts[1].strip()
+                                try:
+                                    scrape_queue.put_nowait(f"https://{domain}/")
+                                except asyncio.QueueFull:
+                                    break
+                print("✅ Tranco dataset loaded into Scraper Queue!")
+    except Exception as e:
+        print("⚠️ Failed to download Tranco, using fallback seeds.", e)
+        fallback = ["https://wikipedia.org/", "https://github.com/", "https://bbc.com/"]
+        for url in fallback:
+            await scrape_queue.put(url)
+
+async def extract_page_data(session, url):
+    headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) NebulixBot/1.0"}
+    try:
+        async with session.get(url, headers=headers, timeout=10) as response:
+            if response.status != 200: return None
+            
+            html = await response.text()
+            soup = BeautifulSoup(html, 'html.parser')
+            
+            title = soup.title.string if soup.title else ""
+            desc_tag = soup.find("meta", {"name": "description"})
+            description = desc_tag["content"] if desc_tag else ""
+            
+            fav_tag = soup.find("link", {"rel": lambda x: x and 'icon' in x.lower()})
+            favicon = urljoin(url, fav_tag["href"]) if fav_tag else ""
+            
+            images = [urljoin(url, img['src']) for img in soup.find_all('img', src=True)][:10]
+            videos = [urljoin(url, vid['src']) for vid in soup.find_all(['video', 'iframe'], src=True)][:5]
+            
+            docs = []
+            new_links = []
+            for a in soup.find_all('a', href=True):
+                href = urljoin(url, a['href'])
+                if href.endswith(('.pdf', '.doc', '.docx')):
+                    docs.append(href)
+                # Ensure it's a web link and user requirement: filter out 'tranco' keyword in URLs
+                elif href.startswith('http') and "tranco" not in href.lower():
+                    new_links.append(href)
+            
+            return {
+                "url": url,
+                "title": title.strip() if title else "",
+                "description": description.strip() if description else "",
+                "favicon": favicon,
+                "images": images,
+                "videos": videos,
+                "docs": docs[:5],
+                "new_links": new_links[:25] # Spider branching limit
+            }
+    except Exception: return None
+
+async def scraper_worker(worker_id, session):
+    while True:
+        url = await scrape_queue.get()
+        
+        if url not in visited_urls:
+            visited_urls.add(url)
+            
+            data = await extract_page_data(session, url)
+            if data:
+                # 1. SPIDER EFFECT: Put new discovered links into the queue
+                for link in data.pop("new_links"):
+                    if link not in visited_urls:
+                        try:
+                            scrape_queue.put_nowait(link)
+                        except asyncio.QueueFull:
+                            pass # If queue is full, ignore to save memory
+                
+                # 2. Save data to TODAY's date file
+                date_str = datetime.now().strftime("%Y-%m-%d")
+                daily_file = f"date-{date_str}.jsonl"
+                
+                async with merge_lock:
+                    with open(daily_file, "a", encoding="utf-8") as f:
+                        f.write(json.dumps(data) + "\n")
                     
-                    domain = urlparse(item.get("url", "")).netloc
-                    text_to_search = f"{domain} {item.get('title', '')} {item.get('description', '')}"
-                    search_corpus.append(text_to_search.lower())
-        print(f"✅ Successfully loaded {len(raw_data)} records from dataset.jsonl")
-    except FileNotFoundError:
-        print("❌ ERROR: dataset.jsonl file not found in the directory!")
-
-    # b. Start background cleanup task
-    cleanup_task = asyncio.create_task(cleanup_old_logs())
-    
-    yield # App jab tak chalegi, yaha yield rahega
-    
-    # c. App shutdown hone par task cancel karna
-    cleanup_task.cancel()
+                    # Update Memory without duplicate (handled in function)
+                    process_item_into_memory(data)
+                
+                # 3. Check Threshold for Auto-Merge
+                await check_and_merge_datasets()
+                
+        scrape_queue.task_done()
+        await asyncio.sleep(0.05) # Yield event loop
 
-# Pass lifespan to FastAPI
-app = FastAPI(title="Fast Search API", lifespan=lifespan)
+async def start_scraper():
+    async with aiohttp.ClientSession() as session:
+        # Fetch initial seeds if queue is empty
+        if scrape_queue.empty():
+            await fetch_tranco_seeds(session)
+        
+        # 15 Workers for full CPU utilization without crashing HF Basic Space
+        workers = [asyncio.create_task(scraper_worker(i, session)) for i in range(15)]
+        await asyncio.gather(*workers)
 
 
-# --- 4. DATE-BASED LOGGING & ENCRYPTION ---
+# --- 4. FASTAPI & LOGGING LIFESPAN ---
 async def log_query_encrypted(req_id: str, query: str):
-    encrypted_q = cipher_suite.encrypt(query.encode()).decode()
-    now = datetime.now()
-    
-    date_str = now.strftime("%Y-%m-%d")
-    log_file = os.path.join(LOG_DIR, f"search_logs_{date_str}.jsonl")
-    
-    log_entry = {
-        "id": req_id,
-        "query": encrypted_q,
-        "timestamp": now.isoformat()
-    }
-    
-    with open(log_file, "a", encoding="utf-8") as f:
-        f.write(json.dumps(log_entry) + "\n")
-
-
-# --- 5. FAST AI / NLP LOGIC ---
-def check_self_harm(query: str) -> bool:
-    harm_keywords = ["suicide", "kill myself", "die", "harm myself", "end my life", "suiside"]
-    return any(word in query.lower() for word in harm_keywords)
-
-def check_adult_content(query: str) -> bool:
-    adult_keywords = ["porn", "sex", "xxx", "nude", "adult"]
-    return any(word in query.lower() for word in adult_keywords)
-
-def ai_fallback_search(query: str):
-    return {
-        "ai_fallback_triggered": True,
-        "message": f"we dont find anything but we are trying to be the perfect sorry but our ai find this {query}",
-        "results": [
-            {
-                "url": f"https://www.google.com/search?q={query.replace(' ', '+')}",
-                "title": f"AI Web Search: {query}",
-                "description": "Local DB me result na milne par ye AI generated fallback result hai.",
-                "images": [],
-                "videos": [],
-                "docs": []
-            }
-        ]
-    }
-
-# --- 0. SERVE HTML UI ---
-@app.get("/")
-async def serve_ui():
-    # Ye root URL (e.g. https://nebulixlabs-nutral-sr-image-enhancer.hf.space/) par HTML load karega
-    return FileResponse("index.html")
-# --- 6. SEARCH API ENDPOINT ---
+    try:
+        encrypted_q = cipher_suite.encrypt(query.encode()).decode()
+        now = datetime.now()
+        date_str = now.strftime("%Y-%m-%d")
+        log_file = os.path.join(LOG_DIR, f"search_logs_{date_str}.jsonl")
+        
+        log_entry = {"id": req_id, "query": encrypted_q, "timestamp": now.isoformat()}
+        with open(log_file, "a", encoding="utf-8") as f:
+            f.write(json.dumps(log_entry) + "\n")
+            f.flush()
+    except Exception as e: print("Log Error:", e)
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    load_all_data() # Load existing dataset + date files + deduplicate
+    scraper_task = asyncio.create_task(start_scraper()) # Start Spider Background
+    yield
+    scraper_task.cancel()
+
+app = FastAPI(title="Spider Search Engine API", lifespan=lifespan)
+app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"])
+
+
+# --- 5. SMART SEARCH ENDPOINT ---
 @app.get("/search")
-async def search(q: str = Query(..., description="Search query"), background_tasks: BackgroundTasks = BackgroundTasks()):
+async def search(q: str = Query(..., description="Search Query"), background_tasks: BackgroundTasks = BackgroundTasks()):
     req_id = str(uuid.uuid4())
-    
-    # Background Logging
     background_tasks.add_task(log_query_encrypted, req_id, q)
     
-    # Self-Harm Check
-    if check_self_harm(q):
-        return JSONResponse(
-            status_code=400,
-            content={
-                "request_id": req_id,
-                "error": "Safety Alert",
-                "message": "Please do not harm yourself. Help is available. You are valuable."
-            }
-        )
-    
-    is_adult = check_adult_content(q)
+    is_adult = any(word in q.lower() for word in ["porn", "sex", "xxx", "nude", "adult"])
+    query_lower = q.lower()
+    scored_results = []
     
-    final_results = []
-    
-    # RapidFuzz search algorithm
-    if search_corpus:
-        results = process.extract(
-            q.lower(), 
-            search_corpus, 
-            limit=30, 
-            scorer=fuzz.WRatio
-        )
+    async with merge_lock:
+        for index, data in enumerate(search_corpus):
+            title_score = fuzz.token_set_ratio(query_lower, data["title"])
+            desc_score = fuzz.token_set_ratio(query_lower, data["description"])
+            domain_score = fuzz.token_set_ratio(query_lower, data["domain"])
+            
+            # Weighted scoring mechanism
+            total_score = (title_score * 0.50) + (desc_score * 0.35) + (domain_score * 0.15)
+            
+            if total_score > 40:
+                scored_results.append((total_score, index))
+        
+        scored_results.sort(key=lambda x: x[0], reverse=True)
+        top_30 = scored_results[:30]
         
-        for match_str, score, index in results:
-            if score > 40:
-                result_item = raw_data[index].copy()
-                result_item["match_score"] = round(score, 2)
-                final_results.append(result_item)
+        final_results = []
+        for score, index in top_30:
+            item = raw_data[index].copy()
+            item["match_score"] = round(score, 2)
+            final_results.append(item)
             
-    # Fallback to AI if no results found
     if len(final_results) == 0:
-        response_data = ai_fallback_search(q)
-    else:
         response_data = {
-            "total_results": len(final_results),
-            "results": final_results
+            "ai_fallback_triggered": True,
+            "message": "AI generated fallback result.",
+            "results": [{"url": f"https://www.google.com/search?q={q.replace(' ', '+')}", "title": f"Web Search: {q}", "description": "Local DB empty for this query."}]
         }
+    else:
+        response_data = {"total_results": len(final_results), "results": final_results}
         
     response_data["request_id"] = req_id
 
-    # Adult Content Safety Modification
     if is_adult:
         response_data["safety_message"] = "you are safe you can do anything"
         headers = {"X-Safety-Message": "you are safe you can do anything"}
@@ -182,7 +306,5 @@ async def search(q: str = Query(..., description="Search query"), background_tas
     
     return response_data
 
-# --- 7. START SERVER FOR GRADIO SPACE ---
-# Ye block Gradio Space ko bata dega ki port 7860 par API continuously chalani hai.
 if __name__ == "__main__":
     uvicorn.run(app, host="0.0.0.0", port=7860)