gpt4free

from __future__ import annotations

import asyncio
import requests

from ..requests.cdp import SyncCDPSession
from .. import debug
from .template import OpenaiTemplate


def _get_turnstile_token_sync(model: str) -> str:
    """
    Synchronous Turnstile token retrieval using SyncCDPSession with retries.
    Uses a blocking recv() loop — no async timeouts, waits as long as needed.
    Designed to be run via asyncio.run_in_executor() from async context.
    """
    import time

    for attempt in range(1):
        session = SyncCDPSession(headless=False)
        session.start_chrome()

        try:
            url = f"https://deepinfra.com/{model}"
            debug.log(f"[DeepInfra] Navigating to {url} (Attempt {attempt + 1}/3)...")
            session.navigate(url)

            # Inject completions request blocker
            fetch_blocker_js = """
            const origFetch = window.fetch;
            window.fetch = async function(...args) {
                let url = args[0];
                if (typeof url === 'string' && url.includes('/chat/completions')) {
                    return new Response('{}', {status: 200});
                }
                return origFetch.apply(this, args);
            };
            """
            session.evaluate_js(fetch_blocker_js)

            # Try to click "Accept" on cookies consent popup if present
            session.evaluate_js(
                """
            (() => {
                const btn = Array.from(document.querySelectorAll('button')).find(b => b.textContent.trim() === 'Accept');
                if (btn) btn.click();
            })()
            """
            )

            # Click on an empty page area to give window focus — signals Cloudflare that
            # a real user is present, which speeds up Turnstile token generation significantly.
            session.click(200, 400)

            # Wait for textarea readiness, then focus and input text
            debug.log("[DeepInfra] Waiting for active textarea...")
            text_entered = False
            for _ in range(80):  # Up to 40 seconds
                try:
                    ready = session.evaluate_js(
                        """
                    (() => {
                        const ta = document.querySelector('textarea');
                        const ts = document.querySelector('[name=cf-turnstile-response]');
                        if (!ta) return 'no_textarea';
                        if (ta.disabled) return 'disabled';
                        if (!ts) return 'no_turnstile';
                        ta.click();
                        ta.focus();
                        ta.scrollIntoView({ block: 'center' });
                        return 'ready';
                    })()
                    """
                    )

                    if ready == "ready":
                        debug.log(
                            "[DeepInfra] Textarea and Turnstile found, focusing and entering text..."
                        )

                        # Retrieve textarea nodeId for native focusing
                        doc = session.call("DOM.getDocument")
                        root_id = doc["root"]["nodeId"]
                        textarea = session.call(
                            "DOM.querySelector", nodeId=root_id, selector="textarea"
                        )

                        # Native focus via CDP
                        session.call("DOM.focus", nodeId=textarea["nodeId"])

                        # Enter text via native CDP command
                        import random

                        test_prompt = random.choice(
                            [
                                "Hello",
                                "Hi",
                                "Hey there",
                                "Testing",
                                "Ping",
                                "What's up?",
                                "Can you hear me?",
                            ]
                        )
                        session.call("Input.insertText", text=test_prompt)

                        time.sleep(0.5)

                        # Simulate Enter keypress
                        session.call(
                            "Input.dispatchKeyEvent",
                            type="keyDown",
                            windowsVirtualKeyCode=13,
                            key="Enter",
                            code="Enter",
                            text="\r",
                            unmodifiedText="\r",
                        )
                        session.call(
                            "Input.dispatchKeyEvent",
                            type="keyUp",
                            windowsVirtualKeyCode=13,
                            key="Enter",
                            code="Enter",
                            text="\r",
                            unmodifiedText="\r",
                        )

                        text_entered = True
                        break
                except Exception:
                    pass
                time.sleep(0.5)

            if not text_entered:
                debug.log(
                    "[DeepInfra] Textarea/Turnstile not ready or failed to submit, retrying attempt..."
                )
                session.close()
                continue

            # Poll page for Turnstile token
            debug.log("[DeepInfra] Waiting for Cloudflare Turnstile solve...")
            token_js = "document.querySelector('[name=cf-turnstile-response]') ? document.querySelector('[name=cf-turnstile-response]').value : ''"
            token = ""
            for i in range(240):  # Up to 120 seconds per attempt
                try:
                    token = session.evaluate_js(token_js)
                    if token:
                        debug.log(f"[DeepInfra] Token generated on check {i+1}!")
                        return token
                except Exception:
                    pass
                time.sleep(0.5)

        except Exception as e:
            debug.log(f"[DeepInfra] Error on attempt {attempt + 1}: {e}")
        finally:
            session.close()

    return ""


async def get_turnstile_token_async(model: str = None) -> str:
    """Run the synchronous Turnstile solver in a thread pool executor."""
    if not model:
        model = DeepInfra.default_model
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(None, _get_turnstile_token_sync, model)


class DeepInfra(OpenaiTemplate):
    url = "https://deepinfra.com"
    login_url = "https://deepinfra.com/dash/api_keys"
    base_url = "https://api.deepinfra.com/v1/openai"

    working = True
    active_by_default = True

    default_model = "zai-org/GLM-5.2"

    @classmethod
    async def get_quota(cls, **kwargs):
        return {}

    @classmethod
    def get_models(cls, **kwargs):
        if not cls.models:
            url = "https://api.deepinfra.com/models/featured"
            response = requests.get(url, timeout=kwargs.get("timeout", 15))
            models = response.json()

            cls.models = {
                model["model_name"]: {"id": model["model_name"], **model}
                for model in models
                if model.get("type") == "text-generation"
                or model.get("reported_type") == "text-to-image"
            }
            cls.image_models = [
                model["model_name"]
                for model in models
                if model.get("reported_type") == "text-to-image"
            ]
            if cls.live == 0 and cls.models:
                cls.live += 1

        return cls.models

    @classmethod
    async def create_async_generator(
        cls, model, messages, api_key=None, headers=None, **kwargs
    ):
        if not api_key or not cls.is_provider_api_key(api_key):
            api_key = None # Ensure no API key is sent for web-page requests
            # Generate a Turnstile token for each request (required without an API key)
            token = await get_turnstile_token_async(model)
            if token:
                if headers is None:
                    headers = {}
                headers["X-DeepInfra-Turnstile"] = token
            else:
                raise ValueError(
                    "Failed to obtain Turnstile token for DeepInfra request."
                )

        async for chunk in super().create_async_generator(
            model, messages, api_key=api_key, headers=headers, **kwargs
        ):
            yield chunk

    @classmethod
    def get_headers(
        cls, stream: bool, api_key: str = None, headers: dict = None
    ) -> dict:
        if not api_key or not cls.is_provider_api_key(api_key):
            if headers is None:
                headers = {}
            headers["X-DeepInfra-Source"] = "web-page"
            headers["Origin"] = "https://deepinfra.com"
            headers["Referer"] = "https://deepinfra.com/"
            api_key = None
        return super().get_headers(stream, api_key, headers)