mapak/Vero6

1import os2import sys3import asyncio4import json5import socket6import threading7import logging8import pyotp9import aiohttp10from datetime import datetime11from typing import Dict, List, Optional, Any12from dataclasses import dataclass, asdict13from playwright.async_api import async_playwright, Page, Browser, BrowserContext14 15# -----------------------------------------------------------------------------16# Configuration (Synced with worker_linux.py)17# -----------------------------------------------------------------------------18@dataclass19class WorkerConfig:20    api_url: str = os.getenv("API_URL", "http://43.134.39.181:3100")21    # Dynamic ID: [Username]-[SpaceName]22    worker_id: str = os.getenv("SPACE_ID", "Pk-user-space").replace("/", "-")23    heartbeat_interval: int = 524    task_poll_interval: int = 525    max_retries: int = 326    request_timeout: int = 3027    os_type: str = "linux_docker_hf"28 29CONFIG = WorkerConfig()30 31@dataclass32class WorkerState:33    status: str = "online"34    active_threads: int = 035    max_threads: int = 136    current_action: str = "Idle"37    leads_done: int = 038    leads_failed: int = 039    version: str = "4.2.0-Docker"40 41    def to_dict(self) -> Dict[str, Any]:42        return asdict(self)43 44STATE = WorkerState()45STATE_LOCK = threading.Lock()46 47logging.basicConfig(48    level=logging.INFO, 49    format='%(asctime)s - %(levelname)s - %(message)s',50    handlers=[logging.StreamHandler(sys.stdout)]51)52logger = logging.getLogger("worker_docker")53 54# -----------------------------------------------------------------------------55# API Client56# -----------------------------------------------------------------------------57class APIClient:58    def __init__(self, base_url: str):59        self.base_url = base_url60        self.session: Optional[aiohttp.ClientSession] = None61 62    async def connect(self):63        timeout = aiohttp.ClientTimeout(total=CONFIG.request_timeout)64        self.session = aiohttp.ClientSession(timeout=timeout)65 66    async def request(self, method: str, endpoint: str, **kwargs) -> Optional[Dict]:67        url = f"{self.base_url.rstrip('/')}{endpoint}"68        for attempt in range(CONFIG.max_retries):69            try:70                if not self.session or self.session.closed: await self.connect()71                async with self.session.request(method, url, **kwargs) as resp:72                    if resp.status == 200: return await resp.json()73            except Exception as e:74                logger.warning(f"API Attempt {attempt+1} failed: {e}")75            await asyncio.sleep(2 ** attempt)76        return None77 78    async def heartbeat(self, state: WorkerState):79        data = {**state.to_dict(), "worker_id": CONFIG.worker_id, "ip_address": "127.0.0.1"}80        return await self.request("POST", "/api/worker/heartbeat", json=data)81 82    async def request_task(self):83        return await self.request("POST", "/api/worker/task/request", json={"worker_id": CONFIG.worker_id})84 85    async def report_task(self, task_id, campaign_id, lead_id, email, status, error=None):86        data = {87            "task_id": task_id, "worker_id": CONFIG.worker_id, "campaign_id": campaign_id,88            "lead_id": lead_id, "email": email, "status": status, "error_message": error89        }90        return await self.request("POST", "/api/worker/task/report", json=data)91 92# -----------------------------------------------------------------------------93# Automation Engine94# -----------------------------------------------------------------------------95class AutomationEngine:96    def __init__(self):97        self.playwright = None98        self.browser = None99        self.context = None100        self.page = None101 102    async def start(self, headless: bool = True):103        self.playwright = await async_playwright().start()104        self.browser = await self.playwright.chromium.launch(105            headless=headless,106            args=["--no-sandbox", "--disable-setuid-sandbox", "--disable-dev-shm-usage"]107        )108        self.context = await self.browser.new_context()109        self.page = await self.context.new_page()110 111    async def stop(self):112        if self.browser: await self.browser.close()113        if self.playwright: await self.playwright.stop()114 115    async def login_microsoft(self, email: str, temp_pass: str) -> bool:116        try:117            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"118            await self.page.goto(login_url, timeout=60000)119            await self.page.get_by_role("textbox", name="Enter your email, phone, or").fill(email)120            await self.page.get_by_role("button", name="Next").click()121            await asyncio.sleep(2)122            await self.page.get_by_role("textbox", name="Temporary Access Pass").fill(temp_pass)123            await self.page.get_by_role("button", name="Sign in").click()124            await asyncio.sleep(4)125            if await self.page.get_by_role("heading", name="Let's keep your account secure").is_visible(timeout=5000):126                await self.page.get_by_role("button", name="Next").click()127                await asyncio.sleep(2)128            try:129                await self.page.get_by_test_id("choose-different-method-link").click()130                await self.page.get_by_role("button", name="Email Receive a code to reset").click()131            except: pass132            return True133        except Exception as e:134            logger.error(f"Login failed: {e}")135            return False136 137    async def process_lead(self, email: str) -> tuple[bool, Optional[str]]:138        try:139            await self.page.fill('[data-testid="email-input"]', email)140            await self.page.click('button[data-testid="reskin-step-next-button"]')141            await self.page.wait_for_selector('[data-testid="email-verify-challenge-otp-input"], [data-testid="message-bar-error"]', timeout=30000)142            if await self.page.is_visible('[data-testid="message-bar-error"]'):143                return False, "rate_limit"144            otp_input = self.page.get_by_test_id("email-verify-challenge-otp-input")145            if not await otp_input.is_visible(timeout=5000): return False, "OTP_not_visible"146            await otp_input.click()147            await self.page.get_by_test_id("backButton").click()148            return True, None149        except Exception as e:150            return False, str(e)151 152# -----------------------------------------------------------------------------153# Worker154# -----------------------------------------------------------------------------155class Worker:156    def __init__(self):157        self.api = APIClient(CONFIG.api_url)158        self._running = True159 160    async def run(self):161        await self.api.connect()162        asyncio.create_task(self._heartbeat())163        logger.info(f"Worker {CONFIG.worker_id} started. Monitoring tasks...")164        165        while self._running:166            try:167                if STATE.active_threads >= STATE.max_threads:168                    await asyncio.sleep(5); continue169                resp = await self.api.request_task()170                if resp and resp.get("status") == "ok":171                    task = resp.get("task")172                    173                    # SYNC THREADS WITH PANEL SETTINGS174                    with STATE_LOCK:175                        campaign_threads = task.get("settings", {}).get("threads", 1)176                        if campaign_threads > 1:177                            STATE.max_threads = campaign_threads178                            logger.info(f"Thread limit updated from panel: {STATE.max_threads}")179                        STATE.active_threads += 1180                    181                    logger.info(f"Task received: {task['task_id']} | Active Threads: {STATE.active_threads}/{STATE.max_threads}")182                    asyncio.create_task(self._process_task(task))183            except Exception as e:184                logger.error(f"Error in main loop: {e}")185            await asyncio.sleep(CONFIG.task_poll_interval)186 187    async def _heartbeat(self):188        while self._running:189            try: 190                await self.api.heartbeat(STATE)191                logger.info(f"Heartbeat sent - Status: {STATE.status}, Done: {STATE.leads_done}")192            except: pass193            await asyncio.sleep(CONFIG.heartbeat_interval)194 195    async def _process_task(self, task):196        global STATE197        with STATE_LOCK: STATE.status = "busy"198        engine = AutomationEngine()199        try:200            await engine.start(headless=True)201            acc = task["account"]202            tap = acc.get("temppass") or acc.get("temp_pass") or acc.get("password")203            if await engine.login_microsoft(acc["email"], tap):204                for i, lead in enumerate(task["leads"][:10]):205                    email = lead["email"] if isinstance(lead, dict) else lead206                    with STATE_LOCK: STATE.current_action = f"Processing {email}"207                    logger.info(STATE.current_action)208                    ok, err = await engine.process_lead(email)209                    await self.api.report_task(task["task_id"], task["campaign_id"], lead.get("id", 0) if isinstance(lead, dict) else 0, email, "success" if ok else "failed", err)210                    if ok: STATE.leads_done += 1211                    else: STATE.leads_failed += 1212                    if err == "rate_limit": break213        finally:214            await engine.stop()215            with STATE_LOCK: STATE.active_threads -= 1; STATE.status = "online"; STATE.current_action = "Idle"216 217async def main():218    # Start worker and keep alive219    worker = Worker()220    await worker.run()221 222if __name__ == "__main__":223    try:224        asyncio.run(main())225    except KeyboardInterrupt:226        logger.info("Worker stopped by user.")227