Update app.py

app.py

@@ -51,16 +51,19 @@ DB_PATH = STATE_DIR / "crawler.db"
 TRANCO_URL = os.getenv("TRANCO_URL", "https://tranco-list.eu/top-1m.csv.zip")
 SEED_COUNT = int(os.getenv("SEED_COUNT", "100000"))
 
-GLOBAL_CONCURRENCY = int(os.getenv("GLOBAL_CONCURRENCY", "128")) # HF Space ke liye safe limit
-PER_HOST_CONCURRENCY = int(os.getenv("PER_HOST_CONCURRENCY", "4"))
-REQUEST_TIMEOUT = float(os.getenv("REQUEST_TIMEOUT", "15"))
+# Speed badhane ke liye concurrency 256 kar di gayi hai
+GLOBAL_CONCURRENCY = int(os.getenv("GLOBAL_CONCURRENCY", "256")) 
+PER_HOST_CONCURRENCY = int(os.getenv("PER_HOST_CONCURRENCY", "2")) # Alag domains pe force karne ke liye kam kiya
+REQUEST_TIMEOUT = float(os.getenv("REQUEST_TIMEOUT", "10")) # 15 se 10 kiya taki slow sites time waste na karein
 CONNECT_TIMEOUT = float(os.getenv("CONNECT_TIMEOUT", "5"))
-MAX_REDIRECTS = int(os.getenv("MAX_REDIRECTS", "5"))
+MAX_REDIRECTS = int(os.getenv("MAX_REDIRECTS", "3")) # 5 se 3 kiya
 
 MAX_RESPONSE_BYTES = int(os.getenv("MAX_RESPONSE_BYTES", str(4 * 1024 * 1024)))
 MAX_TEXT_CHARS = int(os.getenv("MAX_TEXT_CHARS", "20000"))
 MAX_KEYWORDS = int(os.getenv("MAX_KEYWORDS", "30"))
-MAX_LINKS_PER_PAGE = int(os.getenv("MAX_LINKS_PER_PAGE", "500"))
+
+# Queue block hone se bachane ke liye 500 se 50 links per page kiya
+MAX_LINKS_PER_PAGE = int(os.getenv("MAX_LINKS_PER_PAGE", "50")) 
 
 DB_FLUSH_INTERVAL = float(os.getenv("DB_FLUSH_INTERVAL", "1.0"))
 HF_UPLOAD_INTERVAL = int(os.getenv("HF_UPLOAD_INTERVAL", str(3 * 60 * 60)))
@@ -76,14 +79,14 @@ USER_AGENT = os.getenv("USER_AGENT", "NebulixSearchBot/1.0 (+https://nebulixlabs
 # APP & RUNTIME STATE
 # ============================================================
 
-app = FastAPI(title="Nebulix Search Crawler", version="4.1.0")
+app = FastAPI(title="Nebulix Search Crawler", version="4.2.0")
 
 db: Optional[aiosqlite.Connection] = None
 http_session: Optional[aiohttp.ClientSession] = None
 
 frontier = asyncio.PriorityQueue()
-record_queue = asyncio.Queue(maxsize=5000)
-db_event_queue = asyncio.Queue(maxsize=20000)
+record_queue = asyncio.Queue(maxsize=10000)
+db_event_queue = asyncio.Queue(maxsize=100000) # Queue limit badha di gayi hai
 
 db_lock = asyncio.Lock()
 host_state_lock = asyncio.Lock()
@@ -170,6 +173,7 @@ async def init_db():
     await db.execute("PRAGMA journal_mode=WAL")
     await db.execute("PRAGMA synchronous=NORMAL")
     await db.execute("PRAGMA temp_store=MEMORY")
+    await db.execute("PRAGMA mmap_size=3000000000") # Speed up SQLite
     
     await db.execute("""
         CREATE TABLE IF NOT EXISTS urls (
@@ -198,7 +202,7 @@ async def db_writer():
             batch.append(first)
         except asyncio.TimeoutError: pass
 
-        while len(batch) < 5000:
+        while len(batch) < 10000:
             try: batch.append(db_event_queue.get_nowait())
             except asyncio.QueueEmpty: break
 
@@ -238,10 +242,15 @@ async def db_writer():
 
 async def frontier_feeder():
     while True:
+        # RAM ko free rakhne ke liye memory queue ko 5000 par limit kiya
+        if frontier.qsize() > 5000:
+            await asyncio.sleep(0.5)
+            continue
+            
         try:
             rows = []
             async with db_lock:
-                cursor = await db.execute("SELECT url_hash, url, priority, depth FROM urls WHERE status = 0 ORDER BY priority ASC, discovered_at ASC LIMIT 2000")
+                cursor = await db.execute("SELECT url_hash, url, priority, depth FROM urls WHERE status = 0 ORDER BY priority ASC, discovered_at ASC LIMIT 3000")
                 rows = await cursor.fetchall()
 
                 if rows:
@@ -260,7 +269,7 @@ async def frontier_feeder():
             await asyncio.sleep(1)
 
 # ============================================================
-# TRANCO SEED (RESTORED - VERY IMPORTANT)
+# TRANCO SEED
 # ============================================================
 
 async def seed_tranco():
@@ -437,15 +446,12 @@ async def fetch_page(session, url):
     async with semaphore:
         await respect_host_delay(host, DEFAULT_HOST_DELAY)
         timeout = ClientTimeout(total=REQUEST_TIMEOUT, connect=CONNECT_TIMEOUT)
-        for attempt in range(2):
+        
+        # Dead websites par rukne ke bajaye aage badhne ke liye sirf 1 attempt
+        for attempt in range(1):
             try:
                 async with session.get(url, timeout=timeout, allow_redirects=True, max_redirects=MAX_REDIRECTS) as response:
                     status = response.status
-                    if status in {429, 500, 502, 503, 504}:
-                        if attempt == 0:
-                            await asyncio.sleep(0.5)
-                            continue
-                        return None
                     if status != 200: return None
 
                     content_type = response.headers.get("Content-Type", "").lower()
@@ -463,10 +469,7 @@ async def fetch_page(session, url):
                     body = b"".join(chunks)
                     async with stats_lock: stats["bytes_downloaded"] += len(body)
                     return {"url": str(response.url), "status": status, "html": body, "headers": {str(k).lower(): str(v) for k, v in response.headers.items()}}
-            except (asyncio.TimeoutError, aiohttp.ClientError):
-                if attempt == 0:
-                    await asyncio.sleep(0.25)
-                    continue
+            except Exception:
                 return None
         return None
 
@@ -529,7 +532,7 @@ async def record_writer():
             batch.append(first)
         except asyncio.TimeoutError: pass
 
-        while len(batch) < 500:
+        while len(batch) < 1000:
             try: batch.append(record_queue.get_nowait())
             except asyncio.QueueEmpty: break
 
@@ -605,7 +608,7 @@ async def hf_upload_loop():
         except Exception: pass
 
 # ============================================================
-# CRAWLER WORKER
+# CRAWLER WORKER (FIXED PRIORITY & BLOCKING)
 # ============================================================
 
 async def crawl_worker(worker_id):
@@ -618,7 +621,8 @@ async def crawl_worker(worker_id):
             response = await fetch_page(http_session, url)
 
             if response is None:
-                await db_event_queue.put(("failed", page_hash))
+                if not db_event_queue.full():
+                    db_event_queue.put_nowait(("failed", page_hash))
                 async with stats_lock: stats["failed"] += 1
                 continue
 
@@ -627,11 +631,13 @@ async def crawl_worker(worker_id):
             record, discovered_urls, noindex = parsed
 
             if noindex:
-                await db_event_queue.put(("done", page_hash))
+                if not db_event_queue.full():
+                    db_event_queue.put_nowait(("done", page_hash))
                 async with stats_lock: stats["noindex"] += 1
             else:
                 await persist_record(record)
-                await db_event_queue.put(("done", page_hash))
+                if not db_event_queue.full():
+                    db_event_queue.put_nowait(("done", page_hash))
                 async with stats_lock:
                     stats["successful"] += 1
                     recent_successes.append(time.monotonic())
@@ -641,20 +647,26 @@ async def crawl_worker(worker_id):
 
             for discovered_url in discovered_urls:
                 target_host = urllib.parse.urlsplit(discovered_url).hostname or ""
-                next_priority = priority + 10 if target_host == source_host else priority + 100
+                
+                # FIX: Breadth-First Search. Alag domain ko +10 (High Priority), Same domain ko +100 (Low Priority)
+                # Isse workers alag-alag websites pe jayenge aur wait nahi karenge.
+                next_priority = priority + 10 if target_host != source_host else priority + 100
+                
                 normalized = normalize_url(discovered_url)
                 if normalized:
-                    await db_event_queue.put(("discover", normalized, next_priority, next_depth))
-                    async with stats_lock: stats["urls_discovered"] += 1
+                    if not db_event_queue.full():
+                        db_event_queue.put_nowait(("discover", normalized, next_priority, next_depth))
+                        async with stats_lock: stats["urls_discovered"] += 1
 
         except Exception as exc:
-            await db_event_queue.put(("failed", page_hash))
+            if not db_event_queue.full():
+                db_event_queue.put_nowait(("failed", page_hash))
             async with stats_lock: stats["failed"] += 1
         finally:
             frontier.task_done()
 
 # ============================================================
-# API ENDPOINTS (RESTORED FULL STATUS)
+# API ENDPOINTS
 # ============================================================
 
 async def get_db_counts():
@@ -740,16 +752,15 @@ async def force_save_data():
 async def start_app():
     global http_session
 
-    # HF Space safe CPU limit (Max 16 threads to avoid RAM crash)
+    # CPU ka 70-80% use karne ke liye ThreadPool badha diya gaya hai
     loop = asyncio.get_running_loop()
-    safe_threads = min(16, (os.cpu_count() or 2) * 4)
+    safe_threads = min(48, (os.cpu_count() or 4) * 8) 
     executor = concurrent.futures.ThreadPoolExecutor(max_workers=safe_threads)
     loop.set_default_executor(executor)
 
     await init_db()
     await reset_in_progress()
 
-    # Check if DB is empty, then seed Tranco
     async with db_lock:
         async with db.execute("SELECT COUNT(*) FROM urls") as cursor:
             count = (await cursor.fetchone())[0]
@@ -773,7 +784,7 @@ async def start_app():
     for i in range(GLOBAL_CONCURRENCY):
         worker_tasks.append(asyncio.create_task(crawl_worker(i)))
 
-    print(f"Crawler started with {GLOBAL_CONCURRENCY} workers and {safe_threads} CPU threads.")
+    print(f"Crawler started with {GLOBAL_CONCURRENCY} workers and {safe_threads} CPU threads for maximum speed.")
 
 @asynccontextmanager
 async def lifespan(_app):