browser_pool.py

"""
╔══════════════════════════════════════════════════════════════╗
║   Browser Pool Manager for Titan Downloader                  ║
║   Enhanced: visits target URL, realistic headers             ║
╚══════════════════════════════════════════════════════════════╝
"""
import asyncio
import logging
import os
import tempfile
import urllib.parse
from typing import List, Optional

import yt_dlp
from playwright.async_api import async_playwright, Browser, BrowserContext, Playwright

log = logging.getLogger("titan-dl.browser")

HEADLESS = os.environ.get("BROWSER_HEADLESS", "true").lower() == "true"
PROXY_SERVER = os.environ.get("PROXY_SERVER")
USER_AGENT = (
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
    "AppleWebKit/537.36 (KHTML, like Gecko) "
    "Chrome/131.0.0.0 Safari/537.36"
)

class BrowserPool:
    _playwright: Optional[Playwright] = None
    _browsers: List[Browser] = []
    _lock = asyncio.Lock()
    _initialized = False

    @classmethod
    async def initialize(cls, pool_size: int = 4):
        if cls._initialized:
            return
        cls._playwright = await async_playwright().start()
        for _ in range(pool_size):
            browser = await cls._launch_browser()
            cls._browsers.append(browser)
        cls._initialized = True
        log.info(f"Browser pool initialized with {pool_size} browsers.")

    @classmethod
    async def _launch_browser(cls) -> Browser:
        launch_args = [
            "--no-sandbox",
            "--disable-setuid-sandbox",
            "--disable-dev-shm-usage",
            "--disable-blink-features=AutomationControlled",
            "--disable-web-security",
            "--disable-features=IsolateOrigins,site-per-process",
            "--disable-site-isolation-trials",
        ]
        if PROXY_SERVER:
            launch_args.append(f"--proxy-server={PROXY_SERVER}")

        browser = await cls._playwright.chromium.launch(
            headless=HEADLESS,
            args=launch_args,
        )
        return browser

    @classmethod
    async def acquire_context(cls) -> BrowserContext:
        async with cls._lock:
            if not cls._browsers:
                log.warning("Browser pool exhausted, creating temporary browser.")
                browser = await cls._launch_browser()
            else:
                browser = cls._browsers.pop()

        try:
            context = await browser.new_context(
                user_agent=USER_AGENT,
                viewport={"width": 1920, "height": 1080},
                locale="en-US",
                timezone_id="America/New_York",
                extra_http_headers={
                    "Accept-Language": "en-US,en;q=0.9",
                    "sec-ch-ua": '"Chromium";v="131", "Not_A Brand";v="24"',
                    "sec-ch-ua-mobile": "?0",
                    "sec-ch-ua-platform": '"Windows"',
                },
            )
            context.on("close", lambda: asyncio.create_task(cls._return_browser(browser)))
            return context
        except Exception:
            await cls._return_browser(browser)
            raise

    @classmethod
    async def _return_browser(cls, browser: Browser):
        async with cls._lock:
            if len(cls._browsers) < 4:
                cls._browsers.append(browser)
            else:
                await browser.close()

    @classmethod
    async def close(cls):
        async with cls._lock:
            for browser in cls._browsers:
                await browser.close()
            cls._browsers.clear()
            if cls._playwright:
                await cls._playwright.stop()
            cls._initialized = False


async def extract_with_browser(url: str) -> dict:
    """
    استخراج معلومات الفيديو باستخدام متصفح حقيقي.
    يقوم بزيارة صفحة الفيديو نفسها لتوليد الكوكيز المناسبة.
    """
    context = await BrowserPool.acquire_context()
    try:
        page = await context.new_page()

        # زيارة صفحة الفيديو مباشرة (وليس فقط الصفحة الرئيسية)
        log.info(f"🌐 Visiting video URL with browser: {url[:60]}...")
        await page.goto(url, wait_until="domcontentloaded", timeout=30000)
        # انتظر قليلاً لتحميل بعض العناصر وتنفيذ السكريبتات
        await asyncio.sleep(3)

        # استخراج الكوكيز
        cookies = await context.cookies()
        netscape_cookies = _cookies_to_netscape(cookies)

        # كتابة الكوكيز إلى ملف مؤقت
        with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f:
            f.write(netscape_cookies)
            cookie_file = f.name

        try:
            from main import YDL_OPTS_BASE, _extract_info_ytdlp

            opts = YDL_OPTS_BASE.copy()
            opts["cookiefile"] = cookie_file
            if PROXY_SERVER:
                opts["proxy"] = PROXY_SERVER

            log.info("🔧 Running yt-dlp with fresh browser cookies...")
            data = _extract_info_ytdlp(url, ydl_opts=opts)
            return data
        finally:
            os.unlink(cookie_file)

    finally:
        await context.close()


def _cookies_to_netscape(cookies: list) -> str:
    lines = ["# Netscape HTTP Cookie File", ""]
    for c in cookies:
        domain = c.get("domain", "")
        flag = "TRUE" if domain.startswith(".") else "FALSE"
        secure = "TRUE" if c.get("secure", False) else "FALSE"
        expires = str(int(c.get("expires", -1) or -1))
        name = c.get("name", "")
        value = c.get("value", "")
        path = c.get("path", "/")
        lines.append(f"{domain}\t{flag}\t{path}\t{secure}\t{expires}\t{name}\t{value}")
    return "\n".join(lines)