Upload 5 files

hf_runner.py

@@ -1,4 +1,6 @@
 import os
+import sys
+import time
 import subprocess
 import threading
 from flask import Flask
@@ -10,11 +12,23 @@ def home():
     return "Worker is Running! 🚀"
 
 def run_worker():
-    print("Starting Worker...")
-    subprocess.run(["python", "worker.py"])
+    """
+    Worker ko background mein run karta hai.
+    FIX #1: Auto-restart loop — agar worker crash ho toh 5s baad restart hoga
+    FIX #2: stdout/stderr properly piped taaki HF logs mein dikhe
+    """
+    while True:
+        print("[hf_runner] Starting worker.py...", flush=True)
+        result = subprocess.run(
+            ["python", "-u", "worker.py"],   # -u = unbuffered output
+            stdout=sys.stdout,
+            stderr=sys.stderr
+        )
+        print(f"[hf_runner] Worker exited with code {result.returncode}. Restarting in 5s...", flush=True)
+        time.sleep(5)
 
 if __name__ == "__main__":
-    # Start worker in background
+    # Start worker in background thread with auto-restart
     threading.Thread(target=run_worker, daemon=True).start()
-    # Start dummy server for Hugging Face health check
+    # Start dummy Flask server for Hugging Face health check (port 7860 required)
     app.run(host="0.0.0.0", port=7860)

worker.py

@@ -29,8 +29,6 @@ import redis.asyncio as redis
 # Playwright imports
 from playwright.async_api import async_playwright, Page, Browser, BrowserContext
 
-# -----------------------------------------------------------------------------
-# Configuration
 # -----------------------------------------------------------------------------
 REDIS_HOST = os.getenv("REDIS_HOST", "13.235.87.209")
 REDIS_PORT = int(os.getenv("REDIS_PORT", "6379"))
@@ -47,7 +45,7 @@ RETRY_COUNT_KEY = "retry_count"
 WORKER_TASKS = int(os.getenv("WORKER_THREADS", "5"))
 BATCH_SIZE = 10
 MAX_DAILY_LEADS = 40
-COOLDOWN_SECONDS = 3600 # 1 hour
+COOLDOWN_SECONDS = 3600  # 1 hour
 MAX_RETRIES = 3
 POLL_TIMEOUT = 5
 
@@ -59,36 +57,48 @@ _shutdown_requested = False
 # Machine identity (1 machine = 1 worker in dashboard)
 MACHINE_ID = socket.gethostname()
 
-# Shared state across all threads on this machine
+# Shared state across all tasks on this machine
 _machine_stats = {
     "total_done": 0,
     "total_failed": 0,
     "threads": {},  # task_id -> current_action
 }
-_stats_lock = asyncio.Lock()
+
+# FIX #7: asyncio.Lock() module level par nahi, lazy init bhi nahi — main() ke andar bano
+_stats_lock: asyncio.Lock = None  # type: ignore
 
 def _signal_handler(signum, frame):
     global _shutdown_requested
-    logger.info(f"Shutdown signal received. Draining worker pool...")
+    logger.info("Shutdown signal received. Draining worker pool...")
     _shutdown_requested = True
 
 signal.signal(signal.SIGINT, _signal_handler)
 signal.signal(signal.SIGTERM, _signal_handler)
 
-_redis_pool = None
+# FIX #7: Redis pool eager init — module level par, race condition nahi hogi
+_redis_pool = redis.ConnectionPool(
+    host=REDIS_HOST, port=REDIS_PORT, password=REDIS_PASSWORD,
+    decode_responses=True, max_connections=50
+)
 
 def get_redis_pool():
-    global _redis_pool
-    if _redis_pool is None:
-        _redis_pool = redis.ConnectionPool(
-            host=REDIS_HOST, port=REDIS_PORT, password=REDIS_PASSWORD,
-            decode_responses=True, max_connections=50
-        )
     return _redis_pool
 
-# -----------------------------------------------------------------------------
-# Automation Engine
-# -----------------------------------------------------------------------------
+
+# FIX #12: Redis startup connection check
+async def check_redis_connection():
+    """Startup mein Redis reachability check karo"""
+    client = redis.Redis(connection_pool=get_redis_pool())
+    try:
+        await client.ping()
+        logger.info(f"✅ Redis connected: {REDIS_HOST}:{REDIS_PORT}")
+    except Exception as e:
+        logger.critical(f"❌ Cannot connect to Redis at {REDIS_HOST}:{REDIS_PORT} — {e}")
+        sys.exit(1)
+    finally:
+        await client.aclose()
+
+
 class AutomationEngine:
     """Robust Playwright automation engine"""
 
@@ -116,7 +126,7 @@ class AutomationEngine:
         self.browser = await self.playwright.chromium.launch(**browser_options)
         self.context = await self.browser.new_context()
         self.page = await self.context.new_page()
-        # Abort resource-heavy requests
+        # Abort resource-heavy requests (images/fonts/media) for speed
         await self.page.route("**/*", lambda route: (
             route.abort() if route.request.resource_type in ["image", "font", "media"]
             else route.continue_()
@@ -124,7 +134,7 @@ class AutomationEngine:
 
     async def stop(self):
         if self.context:
-            await self.context.close()   # 🔥 important
+            await self.context.close()
         if self.browser:
             await self.browser.close()
         if self.playwright:
@@ -143,47 +153,45 @@ class AutomationEngine:
                 "response_mode=form_post"
             )
             await self.page.goto(login_url, timeout=60000)
-            
+
             # Email step
             await self.page.get_by_role("textbox", name="Enter your email, phone, or").fill(email)
             await self.page.get_by_role("button", name="Next").click()
             await asyncio.sleep(2)
-            
+
             # TAP step
             tap_field = self.page.get_by_role("textbox", name="Temporary Access Pass")
             await tap_field.fill(temp_pass)
             await self.page.get_by_role("button", name="Sign in").click()
-            
+
             # Check for incorrect TAP immediately
             if await self.page.locator('text="Your Temporary Access Pass is incorrect."').is_visible(timeout=5000):
                 logger.error(f"Incorrect Temporary Access Pass for {email}")
                 return False
-                
-            # Wait a bit longer for the "Keep your account secure" or "Authenticator" page to load
+
             await asyncio.sleep(4)
-            
+
             # Handling potential security prompts
             if await self.page.get_by_role("heading", name="Let's keep your account secure").is_visible(timeout=5000):
                 await self.page.get_by_role("button", name="Next").click()
                 await asyncio.sleep(2)
-            
-            # Choose different method -> Email (using the button from your HTML snippet)
+
+            # Choose different method -> Email
             try:
-                # Wait for the page to settle
                 await asyncio.sleep(2)
                 await self.page.wait_for_load_state("domcontentloaded", timeout=20000)
-                
+
                 selector = 'button[data-testid="choose-different-method-link"]:has-text("Set up a different way to sign in")'
                 await self.page.wait_for_selector(selector, timeout=15000)
                 await self.page.locator(selector).click()
-                
-                # Now click the Email option
+
+                # Click the Email option
                 await self.page.get_by_role("button", name="Email Receive a code to reset").click()
                 await asyncio.sleep(2)
             except Exception as e:
                 logger.warning(f"Could not switch to Email method: {e}")
                 return False
-                
+
             return True
         except Exception as e:
             logger.error(f"Login failed for {email}: {e}")
@@ -211,22 +219,20 @@ class AutomationEngine:
 
             if await self.page.locator('[data-testid="message-bar-error"]').is_visible():
                 text = await self.page.locator('[data-testid="message-bar-error"]').inner_text()
-                if "too many" in text.lower(): return False, "rate_limit"
+                if "too many" in text.lower():
+                    return False, "rate_limit"
                 return False, text
 
             if await self.page.locator('[data-testid="email-verify-challenge-otp-input"]').is_visible():
                 await self.page.locator('[data-testid="backButton"]').click()
                 logger.info(f"OTP Flow Success: {email}")
                 return True, None
-            
+
             return False, "otp_not_found"
         except Exception as e:
             return False, str(e)
 
 
-# -----------------------------------------------------------------------------
-# Worker Async Task Logic
-# -----------------------------------------------------------------------------
 class WorkerTask:
     def __init__(self, task_id: str):
         self.task_id = task_id
@@ -234,60 +240,61 @@ class WorkerTask:
         self.leads_done = 0
         self.leads_failed = 0
         self.current_action = "Starting"
-        
+
     async def _fetch_batch(self) -> List[Dict[str, Any]]:
-        """Fetch a batch of up to BATCH_SIZE using BRPOPLPUSH for the first, then RPOPLPUSH."""
+        """Fetch a batch of up to BATCH_SIZE leads from Redis."""
         leads = []
         try:
-            # Block for the first item
-            raw_lead = await self.client.brpoplpush(LEAD_QUEUE, PROCESSING_QUEUE, timeout=POLL_TIMEOUT)
+            # FIX #8: brpoplpush deprecated — use blmove instead (Redis 6.2+)
+            raw_lead = await self.client.blmove(
+                LEAD_QUEUE, PROCESSING_QUEUE, "RIGHT", "LEFT", timeout=POLL_TIMEOUT
+            )
             if not raw_lead:
                 return leads
-            
+
             leads.append({"_raw": raw_lead, "data": json.loads(raw_lead)})
-            
+
             # Non-blocking fetch for the rest of the batch
             for _ in range(BATCH_SIZE - 1):
-                raw = await self.client.rpoplpush(LEAD_QUEUE, PROCESSING_QUEUE)
+                raw = await self.client.lmove(LEAD_QUEUE, PROCESSING_QUEUE, "RIGHT", "LEFT")
                 if not raw:
                     break
                 leads.append({"_raw": raw, "data": json.loads(raw)})
-                
+
         except Exception as e:
             logger.error(f"Error fetching batch: {e}")
-            
+
         return leads
 
     async def _cleanup_and_push_results(self, leads: List[Dict[str, Any]], results: List[Dict[str, Any]]):
-        if not leads: return
+        if not leads:
+            return
+        if len(leads) != len(results):
+            logger.error(f"Mismatch: {len(leads)} leads but {len(results)} results — some leads may be stuck!")
         try:
             pipe = self.client.pipeline()
             for lead, result in zip(leads, results):
                 pipe.lpush(RESULT_QUEUE, json.dumps(result))
                 pipe.lrem(PROCESSING_QUEUE, -1, lead["_raw"])
-                
+
                 lead_id = lead["data"].get("id", str(uuid.uuid4()))
                 if result["success"]:
                     pipe.hdel(RETRY_COUNT_KEY, lead_id)
-                    
+
             await pipe.execute()
         except Exception as e:
             logger.error(f"Error in cleanup and push: {e}")
 
     async def run(self):
         logger.info(f"Task {self.task_id} started. Batch Size: {BATCH_SIZE}")
-        
+
         while not _shutdown_requested:
-            # Update shared machine stats (no per-thread Redis write)
+            # Update shared machine stats
             async with _stats_lock:
                 _machine_stats["threads"][self.task_id] = self.current_action
-                _machine_stats["total_done"] = sum(
-                    t.leads_done for t in _all_worker_tasks
-                )
-                _machine_stats["total_failed"] = sum(
-                    t.leads_failed for t in _all_worker_tasks
-                )
-            
+                _machine_stats["total_done"] = sum(t.leads_done for t in _all_worker_tasks)
+                _machine_stats["total_failed"] = sum(t.leads_failed for t in _all_worker_tasks)
+
             # 1. Lease Account from Ready Pool
             email = await self.client.spop(ACCOUNT_POOL_READY)
             if not email:
@@ -295,39 +302,41 @@ class WorkerTask:
                 logger.debug(f"Task {self.task_id} waiting for accounts in '{ACCOUNT_POOL_READY}'...")
                 await asyncio.sleep(5)
                 continue
-                
+
             acc_key = f"account:{email}"
             acc_pass = await self.client.hget(acc_key, "temp_pass")
-            
-            # Fetch Leads
+
+            # 2. Fetch Leads
             self.current_action = "Fetching Leads"
             batch = await self._fetch_batch()
             if not batch:
-                # No leads to process, put account back in ready and wait
+                # No leads, return account to ready pool
                 await self.client.sadd(ACCOUNT_POOL_READY, email)
                 self.current_action = "Idle"
                 await asyncio.sleep(1)
                 continue
-            
+
             results = []
             engine = AutomationEngine()
-            
+
+            # FIX #2: login_success default False PEHLE try block se — UnboundLocalError prevent
+            login_success = False
+
             try:
-                # Set headless=False so the user can visually debug the Microsoft login flow
                 await engine.start(headless=True)
-                
+
                 # 3. Login
                 self.current_action = f"Logging in: {email}"
                 login_success = await engine.login_microsoft(email, acc_pass)
-                
+
                 for item in batch:
                     lead_data = item["data"]
                     lead_id = lead_data.get("id", str(uuid.uuid4()))
                     raw_data = item["_raw"]
                     lead_email = lead_data.get("email")
-                    
+
                     self.current_action = f"Processing: {lead_email}"
-                    
+
                     if not login_success:
                         # Fail entire batch if login fails
                         results.append({
@@ -337,7 +346,7 @@ class WorkerTask:
                             "worker_id": self.task_id,
                             "row_number": lead_data.get("row_number")
                         })
-                        
+
                         # Requeue the raw_data
                         pipe = self.client.pipeline()
                         pipe.rpush(LEAD_QUEUE, raw_data)
@@ -345,15 +354,17 @@ class WorkerTask:
                         await pipe.execute()
                         item["_skip"] = True
                         continue
-                        
+
+                    # FIX #3: retries default 0 PEHLE — UnboundLocalError prevent
+                    retries = 0
                     try:
                         retries = int(await self.client.hget(RETRY_COUNT_KEY, lead_id) or 0)
                         start_time = time.time()
-                        
+
                         # Run Automation
                         success, err = await engine.process_lead(lead_email)
                         elapsed = (time.time() - start_time) * 1000
-                        
+
                         results.append({
                             "lead_id": lead_id,
                             "success": success,
@@ -363,23 +374,23 @@ class WorkerTask:
                             "worker_id": self.task_id,
                             "row_number": lead_data.get("row_number")
                         })
-                        
+
                         if success:
                             self.leads_done += 1
                             logger.info(f"[+] Successfully sent lead to: {lead_email}")
                         else:
                             self.leads_failed += 1
-                            logger.warning(f"[-] Failed to send lead to: {lead_email}. Reason: {err}")
-                        
+                            logger.warning(f"[-] Failed lead: {lead_email}. Reason: {err}")
+
                         if err == "rate_limit":
                             logger.warning("Rate limit hit, stopping batch")
                             break
-                        
+
                     except Exception as e:
                         logger.error(f"Error processing lead {lead_id}: {e}")
                         retries += 1
                         await self.client.hset(RETRY_COUNT_KEY, lead_id, retries)
-                        
+
                         if retries >= MAX_RETRIES:
                             results.append({
                                 "lead_id": lead_id,
@@ -394,20 +405,21 @@ class WorkerTask:
                             pipe.lrem(PROCESSING_QUEUE, -1, raw_data)
                             await pipe.execute()
                             item["_skip"] = True
-                            
+
             except Exception as e:
                 logger.error(f"Engine exception: {e}")
             finally:
                 await engine.stop()
-            
-            # 4. Handle Account Routing (Cooldown / Done)
+
+            # 4. Handle Account Routing (Cooldown / Done / Pending Refresh)
             if not login_success:
                 await self.client.sadd("account_pool:pending_refresh", email)
                 logger.info(f"Account {email} login failed. Moved to pending_refresh.")
             else:
-                # Increment usage
-                leads_sent = await self.client.hincrby(acc_key, "leads_sent_today", len(batch))
-                
+                # FIX #6: Count only actually processed leads (not skipped)
+                actual_processed = len([b for b in batch if not b.get("_skip")])
+                leads_sent = await self.client.hincrby(acc_key, "leads_sent_today", actual_processed)
+
                 if leads_sent >= MAX_DAILY_LEADS:
                     await self.client.sadd(ACCOUNT_POOL_DONE, email)
                     logger.info(f"Account {email} hit daily limit ({leads_sent}/{MAX_DAILY_LEADS}). Moved to Done.")
@@ -415,20 +427,20 @@ class WorkerTask:
                     now = time.time()
                     await self.client.hset(acc_key, "cooldown_until", now + COOLDOWN_SECONDS)
                     await self.client.sadd(ACCOUNT_POOL_COOLDOWN, email)
-                    logger.info(f"Account {email} finished batch ({leads_sent}/{MAX_DAILY_LEADS}). Moved to Cooldown for 1 hour.")
-            
+                    logger.info(f"Account {email} on Cooldown ({leads_sent}/{MAX_DAILY_LEADS}). 1 hour rest.")
+
             final_batch = [b for b in batch if not b.get("_skip")]
-            # Filter results for leads actually in final_batch
             valid_ids = {b["data"].get("id") for b in final_batch}
             final_results = [r for r in results if r["lead_id"] in valid_ids]
-            
+
             await self._cleanup_and_push_results(final_batch, final_results)
 
-# Global list of all WorkerTask instances on this machine
+
 _all_worker_tasks: list["WorkerTask"] = []
 
+
 async def machine_heartbeat_loop():
-    """Single coroutine that sends ONE heartbeat for the whole machine every 5s."""
+    """Send ONE heartbeat per machine every 5s."""
     client = redis.Redis(connection_pool=get_redis_pool())
     key = f"worker:machine:{MACHINE_ID}"
     while not _shutdown_requested:
@@ -436,11 +448,11 @@ async def machine_heartbeat_loop():
             done = _machine_stats["total_done"]
             failed = _machine_stats["total_failed"]
             threads = dict(_machine_stats["threads"])
-        
+
         total = done + failed
         success_rate = round((done / total * 100), 1) if total > 0 else 0
         busy_count = sum(1 for a in threads.values() if a not in ("Idle", "Starting"))
-        
+
         payload = {
             "id": MACHINE_ID,
             "status": "busy" if busy_count > 0 else "online",
@@ -453,24 +465,34 @@ async def machine_heartbeat_loop():
                 (a for a in threads.values() if a not in ("Idle", "Starting")),
                 "Idle"
             ),
-            "version": "2.0.0",
+            "version": "2.1.0",
             "last_seen": time.time()
         }
         await client.setex(key, 30, json.dumps(payload))
         await asyncio.sleep(5)
 
+
 async def main():
-    logger.info("="*50)
+    global _stats_lock
+
+    # FIX #13: asyncio.Lock() event loop ke andar banao, module level par nahi
+    _stats_lock = asyncio.Lock()
+
+    logger.info("=" * 50)
     logger.info(f"Starting Async Playwright Worker [{MACHINE_ID}]. Tasks: {WORKER_TASKS}")
-    logger.info("="*50)
-    
+    logger.info("=" * 50)
+
+    # FIX #12: Redis connection check at startup
+    await check_redis_connection()
+
     for i in range(WORKER_TASKS):
         t = WorkerTask(f"worker-{uuid.uuid4().hex[:6]}")
         _all_worker_tasks.append(t)
-    
+
     task_coroutines = [t.run() for t in _all_worker_tasks]
     await asyncio.gather(machine_heartbeat_loop(), *task_coroutines)
 
+
 if __name__ == "__main__":
     try:
         asyncio.run(main())