Upload worker.py

worker.py

@@ -0,0 +1,478 @@
+import os
+import sys
+import subprocess
+
+def _ensure_dependencies():
+    try:
+        import redis
+        import playwright
+    except ImportError:
+        print("[-] Missing dependencies. Auto-installing 'redis' and 'playwright'...")
+        subprocess.check_call([sys.executable, "-m", "pip", "install", "redis", "playwright"])
+        print("[-] Installing Playwright chromium browser...")
+        subprocess.check_call([sys.executable, "-m", "playwright", "install", "chromium"])
+        print("[+] Installation complete. Restarting worker...")
+        os.execv(sys.executable, [sys.executable] + sys.argv)
+
+_ensure_dependencies()
+
+import json
+import logging
+import signal
+import asyncio
+import uuid
+import time
+import socket
+from typing import Any, Dict, List, Optional
+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"))
+REDIS_PASSWORD = os.getenv("REDIS_PASSWORD", None)
+
+LEAD_QUEUE = "lead_queue"
+PROCESSING_QUEUE = "processing_queue"
+RESULT_QUEUE = "result_queue"
+ACCOUNT_POOL_READY = "account_pool:ready"
+ACCOUNT_POOL_COOLDOWN = "account_pool:cooldown"
+ACCOUNT_POOL_DONE = "account_pool:done_for_today"
+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
+MAX_RETRIES = 3
+POLL_TIMEOUT = 5
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+logger = logging.getLogger("worker")
+
+_shutdown_requested = False
+
+# Machine identity (1 machine = 1 worker in dashboard)
+MACHINE_ID = socket.gethostname()
+
+# Shared state across all threads on this machine
+_machine_stats = {
+    "total_done": 0,
+    "total_failed": 0,
+    "threads": {},  # task_id -> current_action
+}
+_stats_lock = asyncio.Lock()
+
+def _signal_handler(signum, frame):
+    global _shutdown_requested
+    logger.info(f"Shutdown signal received. Draining worker pool...")
+    _shutdown_requested = True
+
+signal.signal(signal.SIGINT, _signal_handler)
+signal.signal(signal.SIGTERM, _signal_handler)
+
+_redis_pool = None
+
+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
+# -----------------------------------------------------------------------------
+class AutomationEngine:
+    """Robust Playwright automation engine"""
+
+    def __init__(self):
+        self.playwright = None
+        self.browser: Optional[Browser] = None
+        self.context: Optional[BrowserContext] = None
+        self.page: Optional[Page] = None
+
+    async def start(self, headless: bool = True):
+        self.playwright = await async_playwright().start()
+        browser_options = {
+            "headless": headless,
+            "args": [
+                "--no-sandbox",
+                "--disable-setuid-sandbox",
+                "--disable-dev-shm-usage",
+                "--disable-accelerated-2d-canvas",
+                "--no-first-run",
+                "--no-zygote",
+                "--single-process",
+                "--disable-gpu"
+            ]
+        }
+        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
+        await self.page.route("**/*", lambda route: (
+            route.abort() if route.request.resource_type in ["image", "font", "media"]
+            else route.continue_()
+        ))
+
+    async def stop(self):
+        if self.context:
+            await self.context.close()   # 🔥 important
+        if self.browser:
+            await self.browser.close()
+        if self.playwright:
+            await self.playwright.stop()
+
+    async def login_microsoft(self, email: str, temp_pass: str) -> bool:
+        """Login using TAP (Temporary Access Pass) flow"""
+        try:
+            logger.info(f"Logging in as {email} using TAP")
+            login_url = (
+                "https://login.microsoftonline.com/common/oauth2/v2.0/authorize?"
+                "client_id=4765445b-32c6-49b0-83e6-1d93765276ca&"
+                "redirect_uri=https%3A%2F%2Fwww.office.com%2Flandingv2&"
+                "response_type=code%20id_token&"
+                "scope=openid%20profile%20https%3A%2F%2Fwww.office.com%2Fv2%2FOfficeHome.All&"
+                "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)
+            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
+                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}")
+            return False
+
+    async def process_lead(self, email: str) -> tuple[bool, Optional[str]]:
+        """Process a single lead OTP flow"""
+        try:
+            # Click Email method if visible
+            email_btn = self.page.locator('button:has-text("Email")')
+            if await email_btn.is_visible(timeout=3000):
+                await email_btn.click()
+
+            # Fill lead email
+            email_input = self.page.locator('[data-testid="email-input"]')
+            await email_input.wait_for(timeout=10000)
+            await email_input.fill(email)
+            await self.page.locator('[data-testid="reskin-step-next-button"]').click()
+
+            # Wait for OTP or Error
+            await self.page.wait_for_selector(
+                '[data-testid="email-verify-challenge-otp-input"], [data-testid="message-bar-error"]',
+                timeout=15000
+            )
+
+            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"
+                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
+        self.client = redis.Redis(connection_pool=get_redis_pool())
+        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."""
+        leads = []
+        try:
+            # Block for the first item
+            raw_lead = await self.client.brpoplpush(LEAD_QUEUE, PROCESSING_QUEUE, 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)
+                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
+        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)
+            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
+                )
+            
+            # 1. Lease Account from Ready Pool
+            email = await self.client.spop(ACCOUNT_POOL_READY)
+            if not email:
+                self.current_action = "Idle"
+                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
+            self.current_action = "Fetching Leads"
+            batch = await self._fetch_batch()
+            if not batch:
+                # No leads to process, put account back in ready and wait
+                await self.client.sadd(ACCOUNT_POOL_READY, email)
+                self.current_action = "Idle"
+                await asyncio.sleep(1)
+                continue
+            
+            results = []
+            engine = AutomationEngine()
+            
+            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({
+                            "lead_id": lead_id,
+                            "success": False,
+                            "error": "login_failed",
+                            "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)
+                        pipe.lrem(PROCESSING_QUEUE, -1, raw_data)
+                        await pipe.execute()
+                        item["_skip"] = True
+                        continue
+                        
+                    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,
+                            "output": {"error": err} if not success else {},
+                            "error": err if not success else None,
+                            "processing_time_ms": elapsed,
+                            "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}")
+                        
+                        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,
+                                "success": False,
+                                "error": str(e),
+                                "worker_id": self.task_id,
+                                "row_number": lead_data.get("row_number")
+                            })
+                        else:
+                            pipe = self.client.pipeline()
+                            pipe.rpush(LEAD_QUEUE, raw_data)
+                            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)
+            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))
+                
+                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.")
+                else:
+                    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.")
+            
+            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."""
+    client = redis.Redis(connection_pool=get_redis_pool())
+    key = f"worker:machine:{MACHINE_ID}"
+    while not _shutdown_requested:
+        async with _stats_lock:
+            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",
+            "active_threads": busy_count,
+            "max_threads": WORKER_TASKS,
+            "leads_done": done,
+            "leads_failed": failed,
+            "success_rate": success_rate,
+            "current_action": next(
+                (a for a in threads.values() if a not in ("Idle", "Starting")),
+                "Idle"
+            ),
+            "version": "2.0.0",
+            "last_seen": time.time()
+        }
+        await client.setex(key, 30, json.dumps(payload))
+        await asyncio.sleep(5)
+
+async def main():
+    logger.info("="*50)
+    logger.info(f"Starting Async Playwright Worker [{MACHINE_ID}]. Tasks: {WORKER_TASKS}")
+    logger.info("="*50)
+    
+    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())
+    except KeyboardInterrupt:
+        logger.info("Worker node shutting down.")