app.py

import asyncio
import hashlib
import html
import logging
import random
import re
import sqlite3
import string
import time
from datetime import datetime
from typing import Dict, Optional, Tuple

import httpx

# ================= CONFIGURATION =================
DB_FILE = "owl_enterprise.db"
WORKER_BASE_URL = "https://solvedtokens.scenic-quarry-plus.workers.dev"
NUM_WORKERS = 1
TELEGRAM_BOT_TOKEN = "8281589932:AAEWpeB7DhxxWjZxAmYqX4FvxvoyWxFITlU"
TELEGRAM_CHAT_ID = "-1003786783073"

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger(__name__)


# ================= DATABASE =================
class Database:
    def __init__(self, db_file: str = DB_FILE):
        self.conn = sqlite3.connect(db_file, check_same_thread=False)
        self.init_db()

    def init_db(self):
        c = self.conn.cursor()
        c.executescript(
            """
            PRAGMA journal_mode=WAL;
            PRAGMA synchronous=NORMAL;

            CREATE TABLE IF NOT EXISTS generated_accounts (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                token TEXT NOT NULL,
                userid TEXT NOT NULL,
                email TEXT,
                password TEXT,
                remaining_traffic INTEGER DEFAULT 200,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                assigned_to INTEGER,
                last_checked TIMESTAMP
            );
            """
        )
        self.conn.commit()

    def add_generated_account(self, token: str, userid: str, email: str, password: str, balance: int):
        self.conn.execute(
            """
            INSERT INTO generated_accounts (token, userid, email, password, remaining_traffic, last_checked)
            VALUES (?, ?, ?, ?, ?, ?)
            """,
            (token, userid, email, password, balance, datetime.now().isoformat()),
        )
        self.conn.commit()

    def count_available_accounts(self) -> int:
        c = self.conn.cursor()
        c.execute("SELECT COUNT(*) FROM generated_accounts WHERE assigned_to IS NULL")
        return c.fetchone()[0]


class CloudWorkerClient:
    def __init__(self, base_url: str = WORKER_BASE_URL):
        self.base_url = base_url.rstrip("/")

    async def add_account(self, account: Dict) -> bool:
        payload = {
            "token": account.get("token"),
            "userid": account.get("userid"),
            "email": account.get("email"),
            "password": account.get("password"),
            "remaining_traffic": int(account.get("balance", 0) or 0),
        }
        try:
            async with httpx.AsyncClient(timeout=25.0, follow_redirects=True) as client:
                resp = await client.post(f"{self.base_url}/account", json=payload)
                if resp.status_code not in (200, 201):
                    logger.warning("Cloud save failed: HTTP %s | %s", resp.status_code, resp.text[:300])
                    return False
                data = resp.json() if resp.text else {}
                if not data.get("success"):
                    logger.warning("Cloud save rejected: %s", data)
                    return False
                return True
        except Exception as e:
            logger.warning("Cloud save request failed: %s", e)
            return False


class TelegramNotifier:
    def __init__(self, bot_token: str = TELEGRAM_BOT_TOKEN, chat_id: str = TELEGRAM_CHAT_ID):
        self.bot_token = bot_token
        self.chat_id = str(chat_id)
        self.base_url = f"https://api.telegram.org/bot{self.bot_token}"

    async def send_success(self, account: Dict) -> bool:
        message = (
            "✅ <b>ACC GENERATED SUCCESSFULLY</b>\n\n"
            f"balance: <code>{html.escape(str(account.get('balance', '')))}</code>\n"
            f"token: <code>{html.escape(str(account.get('token', '')))}</code>\n"
            f"userid: <code>{html.escape(str(account.get('userid', '')))}</code>\n"
            f"email: <code>{html.escape(str(account.get('email', '')))}</code>\n"
            f"pass: <code>{html.escape(str(account.get('password', '')))}</code>"
        )

        payload = {
            "chat_id": self.chat_id,
            "text": message,
            "parse_mode": "HTML",
            "disable_web_page_preview": True,
            "message_thread_id": 1725,
        }

        try:
            async with httpx.AsyncClient(timeout=20.0, follow_redirects=True) as client:
                resp = await client.post(f"{self.base_url}/sendMessage", json=payload)
                if resp.status_code != 200:
                    logger.warning("Telegram notify failed: HTTP %s | %s", resp.status_code, resp.text[:300])
                    return False

                data = resp.json() if resp.text else {}
                if not data.get("ok"):
                    logger.warning("Telegram notify rejected: %s", data)
                    return False

                return True
        except Exception as e:
            logger.warning("Telegram notify exception: %s", e)
            return False


# ================= AUTOMATION (FULL LOAD FLOW) =================
class AsyncOwlProxyAutomation:
    def __init__(self):
        self.base_url = "https://api.owlproxy.com"
        self.mail_api = "https://api.mail.tm"

        self.base_headers = {
            "accept": "application/json, text/plain, */*",
            "accept-language": "en-US,en;q=0.9",
            "appversion": "2005400",
            "clienttype": "web",
            "content-type": "application/json",
            "priority": "u=1, i",
            "requestsource": "wechat-miniapp",
            "sec-ch-ua": '"Chromium";v="142", "Brave";v="142", "Not_A Brand";v="99"',
            "sec-ch-ua-mobile": "?0",
            "sec-ch-ua-platform": '"Windows"',
            "sec-fetch-dest": "empty",
            "sec-fetch-mode": "cors",
            "sec-fetch-site": "same-site",
            "sec-gpc": "1",
            "suppliertype": "0",
            "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/142.0.0.0 Safari/537.36",
            "origin": "https://proxy.owlproxy.com",
            "referer": "https://proxy.owlproxy.com/",
        }

        self.token: Optional[str] = None
        self.userid: Optional[str] = None
        self.email: Optional[str] = None
        self.mail_token: Optional[str] = None

    async def random_delay(self, min_seconds: float = 1.0, max_seconds: float = 3.0):
        await asyncio.sleep(random.uniform(min_seconds, max_seconds))

    @staticmethod
    def parse_json(response: httpx.Response) -> Dict:
        try:
            return response.json() if response.text else {}
        except Exception:
            return {}

    async def create_temp_email(self, client: httpx.AsyncClient) -> Tuple[bool, Optional[str], Optional[str]]:
        try:
            domains_resp = await client.get(f"{self.mail_api}/domains")
            if domains_resp.status_code == 200:
                domains = self.parse_json(domains_resp)
                domain = domains["hydra:member"][0]["domain"]
            else:
                domain = "sharebot.net"

            random_string = "".join(random.choices(string.ascii_lowercase + string.digits, k=10))
            email = f"{random_string}@{domain}"
            password = "".join(random.choices(string.ascii_letters + string.digits, k=12))

            account_data = {"address": email, "password": password}
            create_resp = await client.post(f"{self.mail_api}/accounts", json=account_data)
            if create_resp.status_code not in (200, 201):
                return False, None, None

            login_resp = await client.post(f"{self.mail_api}/token", json=account_data)
            if login_resp.status_code != 200:
                return False, None, None

            token = self.parse_json(login_resp).get("token")
            if not token:
                return False, None, None

            self.email = email
            self.mail_token = token
            return True, email, token
        except Exception as e:
            logger.error("create_temp_email error: %s", e)
            return False, None, None

    async def request_verification_code(self, client: httpx.AsyncClient, email: str) -> bool:
        try:
            payload = {"smsType": 2, "mobilePhone": email}
            resp = await client.post(f"{self.base_url}/owlproxy/api/sms/smsSend", headers=self.base_headers, json=payload)
            data = self.parse_json(resp)
            return data.get("code") == 200
        except Exception as e:
            logger.error("request_verification_code error: %s", e)
            return False

    async def wait_for_email(self, client: httpx.AsyncClient, timeout: int = 120) -> Optional[str]:
        if not self.mail_token:
            return None

        start_time = time.time()
        mail_headers = {"Authorization": f"Bearer {self.mail_token}"}

        while time.time() - start_time < timeout:
            try:
                msg_list_resp = await client.get(f"{self.mail_api}/messages", headers=mail_headers)
                if msg_list_resp.status_code == 200:
                    messages = self.parse_json(msg_list_resp)
                    members = messages.get("hydra:member", [])
                    if members:
                        msg_id = members[0].get("id")
                        msg_resp = await client.get(f"{self.mail_api}/messages/{msg_id}", headers=mail_headers)
                        if msg_resp.status_code == 200:
                            text = self.parse_json(msg_resp).get("text", "")
                            patterns = [
                                r"Verification Code:\s*(\d{6})",
                                r"Code:\s*(\d{6})",
                                r"\b(\d{6})\b",
                            ]
                            for pattern in patterns:
                                match = re.search(pattern, text, re.IGNORECASE)
                                if match:
                                    return match.group(1)
            except Exception as e:
                logger.warning("wait_for_email poll error: %s", e)

            await asyncio.sleep(3)

        return None

    async def register_account(self, client: httpx.AsyncClient, email: str, otp: str, password: str) -> bool:
        try:
            payload = {
                "mobilePhone": email,
                "loginType": 0,
                "verifyCode": otp,
                "channel": "web",
                "password": hashlib.md5(password.encode()).hexdigest(),
            }
            resp = await client.post(f"{self.base_url}/owlproxy/api/user/login", headers=self.base_headers, json=payload)
            data = self.parse_json(resp)
            if data.get("code") != 200:
                return False

            body = data.get("data", {})
            self.token = body.get("token")
            self.userid = body.get("userId")
            return bool(self.token and self.userid)
        except Exception as e:
            logger.error("register_account error: %s", e)
            return False

    async def login_with_password(self, client: httpx.AsyncClient, email: str, password: str) -> bool:
        try:
            payload = {
                "mobilePhone": email,
                "loginType": 1,
                "password": hashlib.md5(password.encode()).hexdigest(),
                "channel": "web",
            }
            resp = await client.post(f"{self.base_url}/owlproxy/api/user/login", headers=self.base_headers, json=payload)
            data = self.parse_json(resp)
            if data.get("code") != 200:
                return False

            body = data.get("data", {})
            self.token = body.get("token")
            self.userid = body.get("userId")
            return bool(self.token and self.userid)
        except Exception as e:
            logger.error("login_with_password error: %s", e)
            return False

    async def post_login_sequence(self, client: httpx.AsyncClient) -> bool:
        if not self.token or not self.userid:
            return False

        headers = self.base_headers.copy()
        headers["token"] = self.token
        headers["userid"] = str(self.userid)

        endpoints = [
            "/owlproxy/api/user/getUserInfo",
            "/owlproxy/api/configure/getCommonConfig",
            "/owlproxy/api/newUserGuide/getNewUserGuideConfig_V2",
            "/owlproxy/api/vpopDialog/getUserActiveBanner",
            "/owlproxy/api/vcDynamicGood/getDynamicProxyRegion",
            "/owlproxy/api/vcDynamicGood/queryCurrentTrafficBalance",
            "/owlproxy/api/vcDynamicGood/getDynamicProxyHost",
        ]

        for endpoint in endpoints:
            try:
                await client.get(f"{self.base_url}{endpoint}", headers=headers)
            except Exception as e:
                logger.warning("post_login endpoint failed (%s): %s", endpoint, e)
            await self.random_delay(1, 2)

        return True

    async def claim_offer_with_retry(self, client: httpx.AsyncClient, guide_id: int = 10003, max_retries: int = 5) -> bool:
        if not self.token or not self.userid:
            return False

        headers = self.base_headers.copy()
        headers["token"] = self.token
        headers["userid"] = str(self.userid)

        for attempt in range(max_retries):
            try:
                await asyncio.sleep(random.uniform(2, 5) * (attempt + 1))
                resp = await client.get(
                    f"{self.base_url}/owlproxy/api/newUserGuide/getNewUserReceiveTraffic",
                    headers=headers,
                    params={"guideId": guide_id},
                )
                data = self.parse_json(resp)
                if data.get("code") == 200:
                    return True
                if data.get("code") == 3316:
                    await asyncio.sleep((2**attempt) * random.uniform(1, 3))
                    continue
                return False
            except Exception as e:
                logger.warning("claim attempt %s failed: %s", attempt + 1, e)
                await asyncio.sleep(5)
        return False

    async def query_balance(self, client: httpx.AsyncClient) -> int:
        if not self.token or not self.userid:
            return 0

        headers = self.base_headers.copy()
        headers["token"] = self.token
        headers["userid"] = str(self.userid)

        try:
            resp = await client.get(f"{self.base_url}/owlproxy/api/vcDynamicGood/queryCurrentTrafficBalance", headers=headers)
            if resp.status_code == 200:
                data = self.parse_json(resp)
                balance = data.get("data")
                if isinstance(balance, int):
                    return balance
        except Exception as e:
            logger.warning("query_balance failed: %s", e)

        return 200 if self.token else 0

    async def run_full_flow(self) -> Optional[Dict]:
        try:
            async with httpx.AsyncClient(
                timeout=30.0,
                follow_redirects=True,
                verify=False
            ) as client:
                success, email, _ = await self.create_temp_email(client)
                if not success or not email:
                    return None

                await self.random_delay(2, 3)

                if not await self.request_verification_code(client, email):
                    return None

                await self.random_delay(1, 2)

                otp = await self.wait_for_email(client, timeout=120)
                if not otp:
                    return None

                await self.random_delay(1, 2)

                password = "".join(random.choices(string.ascii_letters + string.digits, k=12))
                if not await self.register_account(client, email, otp, password):
                    return None

                await self.random_delay(2, 4)

                if not await self.login_with_password(client, email, password):
                    return None

                await self.random_delay(2, 3)

                await self.post_login_sequence(client)
                await self.random_delay(2, 3)

                claim_success = await self.claim_offer_with_retry(client, guide_id=10003, max_retries=5)
                balance = await self.query_balance(client)

                return {
                    "email": email,
                    "password": password,
                    "userid": str(self.userid),
                    "token": self.token,
                    "claim_success": claim_success,
                    "balance": balance,
                }
        except Exception as e:
            logger.error("run_full_flow fatal: %s", e)
            return None


# ================= WORKER LOOP =================
async def refill_worker(worker_id: int, db: Database, cloud: CloudWorkerClient, notifier: TelegramNotifier):
    logger.info("Worker %s started", worker_id)
    while True:
        try:
            logger.info("[Worker %s] Generating without proxy (direct connection)", worker_id)

            automation = AsyncOwlProxyAutomation()
            result = await automation.run_full_flow()
            result_dict: Dict = result if isinstance(result, dict) else {}

            is_eligible = (
                bool(result_dict)
                and bool(result_dict.get("token"))
                and bool(result_dict.get("userid"))
                and bool(result_dict.get("claim_success"))
                and int(result_dict.get("balance", 0) or 0) == 200
            )

            if is_eligible:
                saved = await cloud.add_account(result_dict)
                if saved:
                    logger.info(
                        "[Worker %s] CLOUD SAVED (claim_success + balance=200) -> %s | userid=%s | balance=%s",
                        worker_id,
                        result_dict.get("email"),
                        result_dict.get("userid"),
                        result_dict.get("balance"),
                    )
                    sent = await notifier.send_success(result_dict)
                    if sent:
                        logger.info("[Worker %s] Telegram notification sent", worker_id)
                    else:
                        logger.warning("[Worker %s] Telegram notification failed", worker_id)
                else:
                    logger.warning("[Worker %s] Eligible account generated but cloud save failed", worker_id)
            else:
                if result_dict:
                    logger.warning(
                        "[Worker %s] Skipped save (claim_success=%s, balance=%s)",
                        worker_id,
                        result_dict.get("claim_success"),
                        result_dict.get("balance"),
                    )
                else:
                    logger.warning("[Worker %s] Flow failed, retrying...", worker_id)
                await asyncio.sleep(5)

        except Exception as e:
            logger.error("[Worker %s] Fatal loop error: %s", worker_id, e)
            await asyncio.sleep(10)

        await asyncio.sleep(random.uniform(4, 9))


async def main():
    db = Database()
    cloud = CloudWorkerClient()
    notifier = TelegramNotifier()

    print("\n" + "="*40)
    print("   OWL PROXY GENERATOR (DIRECT MODE)")
    print("="*40)
    print(f"Starting continuous generator with {NUM_WORKERS} workers")
    print("Workers will use direct connection (no proxies)")
    print("Press Ctrl+C to stop\n")

    logger.info("Starting continuous generator with %s workers (direct connection)", NUM_WORKERS)
    tasks = [asyncio.create_task(refill_worker(i + 1, db, cloud, notifier)) for i in range(NUM_WORKERS)]
    await asyncio.gather(*tasks)


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        logger.info("Stopped by user")