gpt4free

from __future__ import annotations

import asyncio
import base64
import hashlib
import json
import os
import random
import re
import time
import uuid
from copy import copy
from typing import (
    AsyncIterator,
    Iterator,
    Optional,
    Generator,
    Dict,
    Union,
    List,
    Any,
    AsyncGenerator,
    Set,
)

from ...requests.curl_cffi import AsyncSession

try:
    import zendriver as nodriver

    has_nodriver = True
except ImportError:
    has_nodriver = False

from ..base_provider import AsyncAuthedProvider, ProviderModelMixin
from ...typing import AsyncResult, Messages, Cookies, MediaListType
from ...requests.raise_for_status import raise_for_status
from ...requests import StreamSession
from ...requests import get_nodriver_session
from ...image import ImageRequest, to_image, to_bytes, detect_file_type
from ...errors import MissingAuthError, NoValidHarFileError, ModelNotFoundError
from ...providers.response import (
    JsonConversation,
    FinishReason,
    SynthesizeData,
    AuthResult,
    ImageResponse,
    ImagePreview,
    ResponseType,
    JsonRequest,
    format_link,
)
from ...providers.response import TitleGeneration, RequestLogin, Reasoning
from ...tools.media import merge_media
from ..helper import format_cookies, format_media_prompt, to_string
from ..openai.models import (
    default_model,
    default_image_model,
    models,
    image_models,
    text_models,
    model_aliases,
)
from ..openai.har_file import get_request_config
from ..openai.har_file import (
    RequestConfig,
    arkReq,
    arkose_url,
    start_url,
    conversation_url,
    backend_url,
    prepare_url,
    backend_anon_url,
)
from ..openai.proofofwork import generate_proof_token
from ..openai.new import get_requirements_token, get_config
from ... import debug

DEFAULT_HEADERS = {
    "accept": "*/*",
    "accept-encoding": "gzip, deflate, br, zstd",
    "accept-language": "en-US,en;q=0.8",
    "referer": "https://chatgpt.com/",
    "sec-ch-ua": '"Google Chrome";v="131", "Chromium";v="131", "Not_A Brand";v="24"',
    "sec-ch-ua-mobile": "?0",
    "sec-ch-ua-platform": '"Windows"',
    "sec-fetch-dest": "empty",
    "sec-fetch-mode": "cors",
    "sec-fetch-site": "same-origin",
    "sec-gpc": "1",
    "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",
}

INIT_HEADERS = {
    "accept": "*/*",
    "accept-language": "en-US,en;q=0.8",
    "cache-control": "no-cache",
    "pragma": "no-cache",
    "priority": "u=0, i",
    "sec-ch-ua": '"Google Chrome";v="131", "Chromium";v="131", "Not_A Brand";v="24"',
    "sec-ch-ua-arch": '"arm"',
    "sec-ch-ua-bitness": '"64"',
    "sec-ch-ua-mobile": "?0",
    "sec-ch-ua-model": '""',
    "sec-ch-ua-platform": '"Windows"',
    "sec-ch-ua-platform-version": '"14.4.0"',
    "sec-fetch-dest": "document",
    "sec-fetch-mode": "navigate",
    "sec-fetch-site": "none",
    "sec-fetch-user": "?1",
    "upgrade-insecure-requests": "1",
    "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",
}

UPLOAD_HEADERS = {
    "accept": "application/json, text/plain, */*",
    "accept-language": "en-US,en;q=0.8",
    "referer": "https://chatgpt.com/",
    "priority": "u=1, i",
    "sec-ch-ua": '"Google Chrome";v="131", "Chromium";v="131", "Not_A Brand";v="24"',
    "sec-ch-ua-mobile": "?0",
    "sec-ch-ua-platform": '"macOS"',
    "sec-fetch-dest": "empty",
    "sec-fetch-mode": "cors",
    "sec-fetch-site": "cross-site",
    "x-ms-blob-type": "BlockBlob",
    "x-ms-version": "2020-04-08",
    "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",
}

ImagesCache: Dict[str, dict] = {}


class OpenaiChat(AsyncAuthedProvider, ProviderModelMixin):
    """A class for creating and managing conversations with OpenAI chat service"""

    label = "OpenAI ChatGPT"
    url = "https://chatgpt.com"
    working = True
    active_by_default = True
    use_nodriver = True
    image_cache = True
    supports_gpt_4 = True
    supports_message_history = True
    supports_system_message = True
    default_model = default_model
    default_image_model = default_image_model
    image_models = image_models
    vision_models = text_models
    models = models
    model_aliases = model_aliases
    synthesize_content_type = "audio/aac"
    request_config = RequestConfig()
    quota_url = "https://chatgpt.com/backend-api/me"

    _api_key: str = None
    _headers: dict = None
    _cookies: Cookies = None
    _expires: int = None

    @classmethod
    async def get_quota(cls, **kwargs):
        auth = cls.get_auth_result()
        async with StreamSession(
            cookies=auth.cookies, headers=auth.headers, impersonate="chrome"
        ) as session:
            async with session.get(cls.quota_url) as response:
                user = await response.json()
                return {"id": user.get("id"), "name": user.get("name")}

    @classmethod
    async def login(cls, **kwargs):
        cache_file = cls.get_cache_file()
        async for chunk in cls.on_auth_async(**kwargs):
            if isinstance(chunk, AuthResult):
                cls.write_cache_file(cache_file, chunk)
                return

    @classmethod
    async def on_auth_async(cls, proxy: str = None, **kwargs) -> AsyncIterator:
        async for chunk in cls.login_generator(proxy=proxy):
            yield chunk
        yield AuthResult(
            api_key=cls._api_key,
            cookies=cls._cookies or cls.request_config.cookies or {},
            headers=cls._headers
            or cls.request_config.headers
            or cls.get_default_headers(),
            expires=cls._expires,
            proof_token=cls.request_config.proof_token,
            turnstile_token=cls.request_config.turnstile_token,
        )

    @classmethod
    async def upload_files(
        cls,
        session: StreamSession,
        auth_result: AuthResult,
        media: MediaListType,
    ) -> List[ImageRequest]:
        """
        Upload an image to the service and get the download URL

        Args:
            session: The StreamSession object to use for requests
            headers: The headers to include in the requests
            media: The files to upload, either a PIL Image object or a bytes object

        Returns:
            An ImageRequest object that contains the download URL, file name, and other data
        """

        async def upload_file(file, image_name=None) -> ImageRequest:
            debug.log(f"Uploading file: {image_name}")
            file_data = {}

            data_bytes = to_bytes(file)
            # Check Cache
            hasher = hashlib.md5()
            hasher.update(data_bytes)
            image_hash = hasher.hexdigest()
            cache_file = ImagesCache.get(image_hash)
            if cls.image_cache and cache_file:
                debug.log("Using cached image")
                return ImageRequest(cache_file)
            extension, mime_type = detect_file_type(data_bytes)
            if "image" in mime_type:
                # Convert the image to a PIL Image object
                file = to_image(data_bytes)
                use_case = "multimodal"
                file_data.update({"height": file.height, "width": file.width})
            else:
                use_case = "my_files"
            image_name = (
                f"file-{len(data_bytes)}{extension}"
                if image_name is None
                else image_name
            )
            data = {
                "file_name": image_name,
                "file_size": len(data_bytes),
                "use_case": use_case,
            }
            # Post the image data to the service and get the image data
            async with session.post(
                f"{cls.url}/backend-api/files", json=data, headers=cls._headers
            ) as response:
                cls._update_request_args(auth_result, session)
                await raise_for_status(response, "Create file failed")
                file_data.update(
                    {
                        **data,
                        **await response.json(),
                        "mime_type": mime_type,
                        "extension": extension,
                    }
                )
            # Put the image bytes to the upload URL and check the status
            await asyncio.sleep(1)
            async with session.put(
                file_data["upload_url"],
                data=data_bytes,
                headers={
                    **UPLOAD_HEADERS,
                    "Content-Type": file_data["mime_type"],
                    "x-ms-blob-type": "BlockBlob",
                    "x-ms-version": "2020-04-08",
                    "Origin": "https://chatgpt.com",
                },
            ) as response:
                await raise_for_status(response)
            # Post the file ID to the service and get the download URL
            async with session.post(
                f"{cls.url}/backend-api/files/{file_data['file_id']}/uploaded",
                json={},
                headers=auth_result.headers,
            ) as response:
                cls._update_request_args(auth_result, session)
                await raise_for_status(response, "Get download url failed")
                uploaded_data = await response.json()
                file_data["download_url"] = uploaded_data["download_url"]
            ImagesCache[image_hash] = file_data.copy()
            return ImageRequest(file_data)

        medias: List["ImageRequest"] = []
        for item in media:
            item = item if isinstance(item, tuple) else (item,)
            __uploaded_media = await upload_file(*item)
            medias.append(__uploaded_media)
        return medias

    @classmethod
    def create_messages(
        cls,
        messages: Messages,
        image_requests: ImageRequest = None,
        system_hints: list = None,
    ):
        """
        Create a list of messages for the user input

        Args:
            prompt: The user input as a string
            image_response: The image response object, if any

        Returns:
            A list of messages with the user input and the image, if any
        """
        # merged_messages = []
        # last_message = None
        # for message in messages:
        #     current_message = last_message
        #     if current_message is not None:
        #         if current_message["role"] == message["role"]:
        #             current_message["content"] += "\n" + message["content"]
        #         else:
        #             merged_messages.append(current_message)
        #             last_message = message.copy()
        #     else:
        #         last_message = message.copy()
        # if last_message is not None:
        #     merged_messages.append(last_message)

        messages = [
            {
                "id": str(uuid.uuid4()),
                "author": {"role": message["role"]},
                "content": {
                    "content_type": "text",
                    "parts": [to_string(message["content"])],
                },
                "metadata": {
                    "serialization_metadata": {"custom_symbol_offsets": []},
                    **({"system_hints": system_hints} if system_hints else {}),
                },
                "create_time": time.time(),
            }
            for message in messages
        ]
        # Check if there is an image response
        if image_requests:
            # Change content in last user message
            messages[-1]["content"] = {
                "content_type": "multimodal_text",
                "parts": [
                    *[
                        {
                            "asset_pointer": f"file-service://{image_request.get('file_id')}",
                            "height": image_request.get("height"),
                            "size_bytes": image_request.get("file_size"),
                            "width": image_request.get("width"),
                        }
                        for image_request in image_requests
                        # Add For Images Only
                        if image_request.get("use_case") == "multimodal"
                    ],
                    messages[-1]["content"]["parts"][0],
                ],
            }
            # Add the metadata object with the attachments
            messages[-1]["metadata"] = {
                "attachments": [
                    {
                        "id": image_request.get("file_id"),
                        "mimeType": image_request.get("mime_type"),
                        "name": image_request.get("file_name"),
                        "size": image_request.get("file_size"),
                        **(
                            {
                                "height": image_request.get("height"),
                                "width": image_request.get("width"),
                            }
                            if image_request.get("use_case") == "multimodal"
                            else {}
                        ),
                    }
                    for image_request in image_requests
                ]
            }
        return messages

    @classmethod
    async def get_generated_image(
        cls,
        session: StreamSession,
        auth_result: AuthResult,
        element: Union[dict, str],
        prompt: str = None,
        conversation_id: str = None,
        status: Optional[str] = None,
    ) -> ImagePreview | ImageResponse | None:
        download_urls = []
        is_sediment = False
        if prompt is None:
            try:
                prompt = element["metadata"]["dalle"]["prompt"]
            except KeyError:
                pass
        if "asset_pointer" in element:
            element = element["asset_pointer"]
        if isinstance(element, str) and element.startswith("file-service://"):
            element = element.split("file-service://", 1)[-1]
        elif isinstance(element, str) and element.startswith("sediment://"):
            is_sediment = True
            element = element.split("sediment://")[-1]
        else:
            raise RuntimeError(f"Invalid image element: {element}")
        if is_sediment:
            url = f"{cls.url}/backend-api/conversation/{conversation_id}/attachment/{element}/download"
        else:
            url = f"{cls.url}/backend-api/files/{element}/download"
        try:
            async with session.get(url, headers=auth_result.headers) as response:
                cls._update_request_args(auth_result, session)
                await raise_for_status(response)
                data = await response.json()
                download_url = data.get("download_url")
                if download_url is not None:
                    download_urls.append(download_url)
                    debug.log(f"OpenaiChat: Found image: {download_url}")
                else:
                    debug.log("OpenaiChat: No download URL found in response: ", data)
        except Exception as e:
            debug.error("OpenaiChat: Download image failed")
            debug.error(e)
        if download_urls:
            # status = None, finished_successfully
            if is_sediment and status != "finished_successfully":
                return ImagePreview(
                    download_urls,
                    prompt,
                    {"status": status, "headers": auth_result.headers},
                )
            else:
                return ImageResponse(
                    download_urls,
                    prompt,
                    {"status": status, "headers": auth_result.headers},
                )

    @classmethod
    async def create_authed(
        cls,
        model: str,
        messages: Messages,
        auth_result: AuthResult,
        proxy: str = None,
        timeout: int = 360,
        auto_continue: bool = False,
        action: Optional[str] = None,
        conversation: Conversation = None,
        media: MediaListType = None,
        return_conversation: bool = True,
        web_search: bool = False,
        prompt: str = None,
        conversation_mode: Optional[dict] = None,
        temporary: Optional[bool] = None,
        conversation_id: Optional[str] = None,
        reasoning_effort: Optional[str] = None,
        **kwargs,
    ) -> AsyncResult:
        """
        Create an asynchronous generator for the conversation.

        Args:
            model (str): The model name.
            messages (Messages): The list of previous messages.
            proxy (str): Proxy to use for requests.
            timeout (int): Timeout for requests.
            api_key (str): Access token for authentication.
            auto_continue (bool): Flag to automatically continue the conversation.
            action (str): Type of action ('next', 'continue', 'variant').
            media (MediaListType): Images to include in the conversation.
            return_conversation (bool): Flag to include response fields in the output.
            **kwargs: Additional keyword arguments.

        Yields:
            AsyncResult: Asynchronous results from the generator.

        Raises:
            RuntimeError: If an error occurs during processing.
        """
        if temporary is None:
            temporary = action is not None and conversation_id is None
        if action is None:
            action = "next"
        async with StreamSession(
            proxy=proxy, impersonate="chrome", timeout=timeout
        ) as session:
            image_requests = None
            media = merge_media(media, messages)
            if not cls.needs_auth and not media:
                if cls._headers is None:
                    cls._create_request_args(cls._cookies)
                    async with session.get(cls.url, headers=INIT_HEADERS) as response:
                        cls._update_request_args(auth_result, session)
                        await raise_for_status(response)
            else:
                if cls._headers is None and getattr(auth_result, "cookies", None):
                    cls._create_request_args(auth_result.cookies, auth_result.headers)
                if not cls._set_api_key(getattr(auth_result, "api_key", None)):
                    raise MissingAuthError("Access token is not valid")
                async with session.get(cls.url, headers=cls._headers) as response:
                    cls._update_request_args(auth_result, session)
                    await raise_for_status(response)

                # try:
                image_requests = await cls.upload_files(session, auth_result, media)
                # except Exception as e:
                #     debug.error("OpenaiChat: Upload image failed")
                #     debug.error(e)
            try:
                model = cls.get_model(model)
            except ModelNotFoundError:
                pass
            image_model = False
            if model in cls.image_models:
                image_model = True
                model = cls.default_model
            if conversation is None:
                conversation = Conversation(
                    None,
                    str(uuid.uuid4()),
                    getattr(auth_result, "cookies", {}).get("oai-did"),
                )
            else:
                conversation = copy(conversation)

            if conversation_mode is None:
                _gizmo_id = kwargs.get("gizmo_id")
                if _gizmo_id:
                    conversation_mode = {
                        "kind": "gizmo_interaction",
                        "gizmo_id": _gizmo_id,
                    }
                else:
                    conversation_mode = {"kind": "primary_assistant"}

            if getattr(auth_result, "cookies", {}).get("oai-did") != getattr(
                conversation, "user_id", None
            ):
                conversation = Conversation(None, str(uuid.uuid4()))
            if cls._api_key is None:
                auto_continue = False
            conversation.finish_reason = None
            sources = OpenAISources([])
            references = ContentReferences()
            system_hints = ["picture_v2"] if image_model else []
            if reasoning_effort == "high":
                system_hints.append("reason")
            if web_search:
                system_hints.append("search")
            while conversation.finish_reason is None:
                conduit_token = None
                if cls._api_key is not None:
                    data = {
                        "action": "next",
                        "fork_from_shared_post": False,
                        "parent_message_id": conversation.message_id,
                        "model": model,
                        "timezone_offset_min": -120,
                        "timezone": "Europe/Berlin",
                        "conversation_mode": conversation_mode,
                        "system_hints": system_hints,
                        "supports_buffering": True,
                        "supported_encodings": ["v1"],
                    }
                    if temporary:
                        data["history_and_training_disabled"] = True
                    if conversation.conversation_id is not None and not temporary:
                        data["conversation_id"] = conversation.conversation_id
                    async with session.post(
                        prepare_url, json=data, headers=cls._headers
                    ) as response:
                        await raise_for_status(response)
                        conduit_token = (await response.json())["conduit_token"]
                async with session.post(
                    f"{cls.url}/backend-anon/sentinel/chat-requirements"
                    if cls._api_key is None
                    else f"{cls.url}/backend-api/sentinel/chat-requirements",
                    json={
                        "p": None
                        if not getattr(auth_result, "proof_token", None)
                        else get_requirements_token(
                            getattr(auth_result, "proof_token", None)
                        )
                    },
                    headers=cls._headers,
                ) as response:
                    if response.status in (401, 403):
                        raise MissingAuthError(f"Response status: {response.status}")
                    else:
                        cls._update_request_args(auth_result, session)
                    await raise_for_status(response)
                    chat_requirements = await response.json()
                    need_turnstile = chat_requirements.get("turnstile", {}).get(
                        "required", False
                    )
                    need_arkose = chat_requirements.get("arkose", {}).get(
                        "required", False
                    )
                    chat_token = chat_requirements.get("token")

                    # if need_arkose and cls.request_config.arkose_token is None:
                #     await get_request_config(proxy)
                #     cls._create_request_args(auth_result.cookies, auth_result.headers)
                #     cls._set_api_key(auth_result.access_token)
                #     if auth_result.arkose_token is None:
                #         raise MissingAuthError("No arkose token found in .har file")
                if "proofofwork" in chat_requirements:
                    user_agent = getattr(auth_result, "headers", {}).get("user-agent")
                    proof_token = getattr(auth_result, "proof_token", None)
                    if proof_token is None:
                        auth_result.proof_token = get_config(user_agent)
                    proofofwork = generate_proof_token(
                        **chat_requirements["proofofwork"],
                        user_agent=user_agent,
                        proof_token=proof_token,
                    )
                # [debug.log(text) for text in (
                # f"Arkose: {'False' if not need_arkose else auth_result.arkose_token[:12]+'...'}",
                # f"Proofofwork: {'False' if proofofwork is None else proofofwork[:12]+'...'}",
                # f"AccessToken: {'False' if cls._api_key is None else cls._api_key[:12]+'...'}",
                # )]
                data = {
                    "action": "next",
                    "parent_message_id": conversation.message_id,
                    "model": model,
                    "timezone_offset_min": -120,
                    "timezone": "Europe/Berlin",
                    "conversation_mode": conversation_mode,
                    "enable_message_followups": True,
                    "system_hints": system_hints,
                    "supports_buffering": True,
                    "supported_encodings": ["v1"],
                    "client_contextual_info": {
                        "is_dark_mode": False,
                        "time_since_loaded": random.randint(20, 500),
                        "page_height": 578,
                        "page_width": 1850,
                        "pixel_ratio": 1,
                        "screen_height": 1080,
                        "screen_width": 1920,
                    },
                    "paragen_cot_summary_display_override": "allow",
                }
                if temporary:
                    data["history_and_training_disabled"] = True

                if conversation.conversation_id is not None and not temporary:
                    data["conversation_id"] = conversation.conversation_id
                    debug.log(
                        f"OpenaiChat: Use conversation: {conversation.conversation_id}"
                    )
                prompt = conversation.prompt = format_media_prompt(messages, prompt)
                if action != "continue":
                    data["parent_message_id"] = getattr(
                        conversation, "parent_message_id", conversation.message_id
                    )
                    conversation.parent_message_id = None
                    new_messages = messages
                    if conversation.conversation_id is not None:
                        new_messages = []
                        for message in messages:
                            if message.get("role") == "assistant":
                                new_messages = []
                            else:
                                new_messages.append(message)
                    data["messages"] = cls.create_messages(
                        new_messages, image_requests, ["search"] if web_search else None
                    )
                yield JsonRequest.from_dict(data)
                headers = {
                    **cls._headers,
                    "accept": "text/event-stream",
                    "content-type": "application/json",
                    "openai-sentinel-chat-requirements-token": chat_token,
                    **(
                        {}
                        if conduit_token is None
                        else {"x-conduit-token": conduit_token}
                    ),
                }
                # if cls.request_config.arkose_token:
                #    headers["openai-sentinel-arkose-token"] = cls.request_config.arkose_token
                if proofofwork is not None:
                    headers["openai-sentinel-proof-token"] = proofofwork
                if (
                    need_turnstile
                    and getattr(auth_result, "turnstile_token", None) is not None
                ):
                    headers[
                        "openai-sentinel-turnstile-token"
                    ] = auth_result.turnstile_token
                async with session.post(
                    backend_anon_url if cls._api_key is None else backend_url,
                    json=data,
                    headers=headers,
                ) as response:
                    cls._update_request_args(auth_result, session)
                    if response.status in (401, 403, 429, 500):
                        raise MissingAuthError("Access token is not valid")
                    elif response.status == 422:
                        raise RuntimeError((await response.json()), data)
                    await raise_for_status(response)
                    buffer = ""
                    matches = []
                    async for line in response.iter_lines():
                        pattern = re.compile(r"file-service://[\w-]+")
                        for match in pattern.finditer(line.decode(errors="ignore")):
                            if match.group(0) in matches:
                                continue
                            matches.append(match.group(0))
                            generated_image = await cls.get_generated_image(
                                session, auth_result, match.group(0), prompt
                            )
                            if generated_image is not None:
                                yield generated_image
                        async for chunk in cls.iter_messages_line(
                            session,
                            auth_result,
                            line,
                            conversation,
                            sources,
                            references,
                        ):
                            if isinstance(chunk, str):
                                chunk = (
                                    chunk.replace("\ue203", "")
                                    .replace("\ue204", "")
                                    .replace("\ue206", "")
                                )
                                buffer += chunk
                                if buffer.find("\ue200") != -1:
                                    if buffer.find("\ue201") != -1:

                                        def sequence_replacer(match):
                                            def citation_replacer(match: re.Match[str]):
                                                ref_type = match.group(1)
                                                ref_index = int(match.group(2))
                                                if (
                                                    (
                                                        ref_type == "image"
                                                        and is_image_embedding
                                                    )
                                                    or is_video_embedding
                                                    or ref_type == "forecast"
                                                ):
                                                    reference = (
                                                        references.get_reference(
                                                            {
                                                                "ref_index": ref_index,
                                                                "ref_type": ref_type,
                                                            }
                                                        )
                                                    )
                                                    if not reference:
                                                        return ""

                                                    if ref_type == "forecast":
                                                        if reference.get("alt"):
                                                            return reference.get("alt")
                                                        if reference.get("prompt_text"):
                                                            return reference.get(
                                                                "prompt_text"
                                                            )

                                                    if (
                                                        is_image_embedding
                                                        and reference.get(
                                                            "content_url", ""
                                                        )
                                                    ):
                                                        return f"![{reference.get('title', '')}]({reference.get('content_url')})"

                                                    if is_video_embedding:
                                                        if reference.get(
                                                            "url", ""
                                                        ) and reference.get(
                                                            "thumbnail_url", ""
                                                        ):
                                                            return f"[![{reference.get('title', '')}]({reference['thumbnail_url']})]({reference['url']})"
                                                        video_match = re.match(
                                                            r"video\n(.*?)\nturn[0-9]+",
                                                            match.group(0),
                                                        )
                                                        if video_match:
                                                            return video_match.group(1)
                                                    return ""

                                                source_index = sources.get_index(
                                                    {
                                                        "ref_index": ref_index,
                                                        "ref_type": ref_type,
                                                    }
                                                )
                                                if (
                                                    source_index is not None
                                                    and len(sources.list) > source_index
                                                ):
                                                    link = sources.list[source_index][
                                                        "url"
                                                    ]
                                                    return f"[[{source_index + 1}]]({link})"
                                                return f""

                                            def products_replacer(match: re.Match[str]):
                                                try:
                                                    products_data = json.loads(
                                                        match.group(1)
                                                    )
                                                    products_str = ""
                                                    for idx, _ in enumerate(
                                                        products_data.get(
                                                            "selections", []
                                                        )
                                                        or []
                                                    ):
                                                        name = products_data.get(
                                                            "selections", []
                                                        )[idx][1]
                                                        tags = products_data.get(
                                                            "tags", []
                                                        )[idx]
                                                        products_str += (
                                                            f"{name} - {tags}\n\n"
                                                        )

                                                    return products_str
                                                except Exception:
                                                    return ""

                                            sequence_content = match.group(1)
                                            sequence_content = (
                                                sequence_content.replace("\ue200", "")
                                                .replace("\ue202", "\n")
                                                .replace("\ue201", "")
                                            )
                                            sequence_content = sequence_content.replace(
                                                "navlist\n", "#### "
                                            )

                                            # Handle search, news, view and image citations
                                            is_image_embedding = (
                                                sequence_content.startswith("i\nturn")
                                            )
                                            is_video_embedding = (
                                                sequence_content.startswith("video\n")
                                            )
                                            sequence_content = re.sub(
                                                r"(?:cite\nturn[0-9]+|forecast\nturn[0-9]+|video\n.*?\nturn[0-9]+|i?\n?turn[0-9]+)(search|news|view|image|forecast)(\d+)",
                                                citation_replacer,
                                                sequence_content,
                                            )
                                            sequence_content = re.sub(
                                                r"products\n(.*)",
                                                products_replacer,
                                                sequence_content,
                                            )
                                            sequence_content = re.sub(
                                                r'product_entity\n\[".*","(.*)"\]',
                                                lambda x: x.group(1),
                                                sequence_content,
                                            )
                                            return sequence_content

                                        # process only completed sequences and do not touch start of next not completed sequence
                                        buffer = re.sub(
                                            r"\ue200(.*?)\ue201",
                                            sequence_replacer,
                                            buffer,
                                            flags=re.DOTALL,
                                        )

                                        if (
                                            buffer.find("\ue200") != -1
                                        ):  # still have uncompleted sequence
                                            continue
                                    else:
                                        # do not yield to consume rest part of special sequence
                                        continue

                                yield buffer
                                buffer = ""
                            else:
                                yield chunk
                        if conversation.finish_reason is not None:
                            break
                    if buffer:
                        yield buffer
                if sources.list:
                    yield sources
                if conversation.generated_images:
                    yield ImageResponse(
                        conversation.generated_images.urls,
                        conversation.prompt,
                        {"headers": auth_result.headers},
                    )
                    conversation.generated_images = None
                conversation.prompt = None
                if return_conversation:
                    yield conversation
                if auth_result.api_key is not None:
                    yield SynthesizeData(
                        cls.__name__,
                        {
                            "conversation_id": conversation.conversation_id,
                            "message_id": conversation.message_id,
                            "voice": "maple",
                        },
                    )
                if auto_continue and conversation.finish_reason == "max_tokens":
                    conversation.finish_reason = None
                    action = "continue"
                    await asyncio.sleep(5)
                else:
                    break

            if conversation.task and kwargs.get("wait_media", True):
                async for _m in cls.wss_media(
                    session, conversation, auth_result.headers, auth_result
                ):
                    yield _m
            # if kwargs.get("wait_media"):
            #     async for _m in cls.wait_media(session, conversation, headers, auth_result):
            #         yield _m

            yield FinishReason(conversation.finish_reason)

    @classmethod
    async def wss_media(
        cls,
        _session,
        conversation: Conversation,
        headers: Dict[str, str],
        auth_result: AuthResult,
        timeout: Optional[int] = 20,
    ):
        seen_assets: Set[str] = set()
        async with AsyncSession(
            timeout=timeout,
            impersonate="chrome",
            headers=headers,
            cookies=auth_result.cookies,
        ) as session:
            response = await session.get(
                "https://chatgpt.com/backend-api/celsius/ws/user",
                headers=headers,
            )
            response.raise_for_status()
            websocket_url = response.json().get("websocket_url")
            started = False
            wss = await session.ws_connect(websocket_url, timeout=3)
            while not wss.closed:
                try:
                    last_msg = await wss.recv_json(
                        timeout=60 if not started else timeout
                    )
                except Exception:
                    break
                conversation_id = conversation.task.get("conversation_id")
                message_id = conversation.task.get("message", {}).get("id")
                if (
                    isinstance(last_msg, dict)
                    and last_msg.get("type") == "conversation-update"
                ):
                    if (
                        last_msg.get("payload", {}).get("conversation_id")
                        != conversation_id
                    ):
                        continue

                    message = (
                        last_msg.get("payload", {})
                        .get("update_content", {})
                        .get("message", {})
                    )
                    if message.get("id") != message_id:
                        continue

                    # if last_msg.get("payload", {}).get("update_type") == 'async-task-start':
                    #     started = True
                    started = True
                    if (
                        last_msg.get("payload", {}).get("update_type")
                        == "async-task-update-message"
                    ):
                        status = message.get("status")
                        parts = message.get("content").get("parts") or []
                        for part in parts:
                            if part.get("content_type") != "image_asset_pointer":
                                continue
                            asset = part.get("asset_pointer")
                            if not asset or asset in seen_assets:
                                continue
                            seen_assets.add(asset)
                            generated_images = await cls.get_generated_image(
                                _session,
                                auth_result,
                                asset,
                                conversation.prompt or "",
                                conversation.conversation_id,
                                status,
                            )
                            if generated_images is not None:
                                yield generated_images
                        if message.get("status") == "finished_successfully":
                            await wss.close()
                            return

    @classmethod
    async def wait_media(
        cls,
        session,
        conversation,
        headers: Dict[str, str],
        auth_result: AuthResult,
        poll_interval: int = 10,
        timeout: Optional[int] = None,
    ) -> AsyncGenerator[Any, None]:
        start_time = asyncio.get_event_loop().time()
        seen_assets: Set[str] = set()
        running = True
        has_image_task = False
        generation_started = False

        while running:
            if timeout is not None:
                elapsed = asyncio.get_event_loop().time() - start_time
                if elapsed > timeout:
                    return
            # https://chatgpt.com/backend-api/tasks
            async with session.get(
                f"https://chatgpt.com/backend-api/conversation/{conversation.conversation_id}",
                headers=headers,
            ) as response:
                await raise_for_status(response)
                data = await response.json()

            mapping = data.get("mapping") or {}
            if not mapping:
                return

            last_node = list(mapping.values())[-1] or {}
            last_message = last_node.get("message") or {}
            metadata = last_message.get("metadata") or {}
            status = last_message.get("status")
            image_task_id = metadata.get("image_gen_task_id")
            if not has_image_task and not image_task_id:
                return

            if image_task_id and not has_image_task:
                debug.log(f"OpenaiChat: Wait Task: {image_task_id}")
                has_image_task = True
            if status == "in_progress":
                generation_started = True
            elif generation_started and status == "finished_successfully":
                running = False
            if generation_started:
                content = last_message.get("content") or {}
                parts = content.get("parts") or []
                for part in parts:
                    if part.get("content_type") != "image_asset_pointer":
                        continue
                    asset = part.get("asset_pointer")
                    if not asset or asset in seen_assets:
                        continue
                    seen_assets.add(asset)
                    generated_images = await cls.get_generated_image(
                        session,
                        auth_result,
                        asset,
                        conversation.prompt or metadata.get("async_task_title") or "",
                        conversation.conversation_id,
                        status,
                    )
                    if generated_images is not None:
                        yield generated_images
            if generation_started and status == "finished_successfully":
                return
            await asyncio.sleep(poll_interval)

    @classmethod
    async def iter_messages_line(
        cls,
        session: StreamSession,
        auth_result: AuthResult,
        line: bytes,
        fields: Conversation,
        sources: OpenAISources,
        references: ContentReferences,
    ) -> AsyncIterator:
        if not line.startswith(b"data: "):
            return
        elif line.startswith(b"data: [DONE]"):
            return
        try:
            line = json.loads(line[6:])
        except Exception:
            return
        if not isinstance(line, dict):
            return
        if "type" in line:
            if line["type"] == "title_generation":
                yield TitleGeneration(line["title"])
        fields.p = line.get("p", fields.p)
        if fields.p is not None and fields.p.startswith("/message/content/thoughts"):
            if fields.p.endswith("/content"):
                if fields.thoughts_summary:
                    yield Reasoning(token="", status=fields.thoughts_summary)
                    fields.thoughts_summary = ""
                yield Reasoning(token=line.get("v"))
            elif fields.p.endswith("/summary"):
                fields.thoughts_summary += line.get("v")
            return
        if "v" in line:
            v = line.get("v")
            if isinstance(v, str) and fields.recipient == "all":
                if fields.p == "/message/metadata/refresh_key_info":
                    yield ""
                elif "p" not in line or line.get("p") == "/message/content/parts/0":
                    yield Reasoning(token=v) if fields.is_thinking else v
            elif isinstance(v, list):
                buffer = ""
                for m in v:
                    if (
                        m.get("p") == "/message/content/parts/0"
                        and fields.recipient == "all"
                    ):
                        buffer += m.get("v")
                    elif m.get("p") == "/message/metadata/image_gen_title":
                        fields.prompt = m.get("v")
                    elif m.get("p") == "/message/content/parts/0/asset_pointer":
                        status = next(
                            filter(lambda x: x.get("p") == "/message/status", v), {}
                        ).get("v", None)
                        generated_images = (
                            fields.generated_images
                        ) = await cls.get_generated_image(
                            session,
                            auth_result,
                            m.get("v"),
                            fields.prompt,
                            fields.conversation_id,
                            status,
                        )
                        if generated_images is not None:
                            if buffer:
                                yield buffer
                            yield generated_images
                    elif m.get("p") == "/message/metadata/search_result_groups":
                        for entry in [p.get("entries") for p in m.get("v")]:
                            for link in entry:
                                sources.add_source(link)
                    elif m.get(
                        "p"
                    ) == "/message/metadata/content_references" and not isinstance(
                        m.get("v"), int
                    ):
                        for entry in m.get("v"):
                            for link in entry.get("sources", []):
                                sources.add_source(link)
                            for link in entry.get("items", []):
                                sources.add_source(link)
                            for link in entry.get("fallback_items", []) or []:
                                sources.add_source(link)
                            if m.get("o", None) == "append":
                                references.add_reference(entry)
                    elif m.get("p") and re.match(
                        r"^/message/metadata/content_references/\d+$", m.get("p")
                    ):
                        if "url" in m.get("v") or "link" in m.get("v"):
                            sources.add_source(m.get("v"))
                        for link in m.get("v").get("fallback_items", []) or []:
                            sources.add_source(link)

                        match = re.match(
                            r"^/message/metadata/content_references/(\d+)$", m.get("p")
                        )
                        if (
                            match
                            and m.get("o") == "append"
                            and isinstance(m.get("v"), dict)
                        ):
                            idx = int(match.group(1))
                            references.merge_reference(idx, m.get("v"))
                    elif (
                        m.get("p")
                        and re.match(
                            r"^/message/metadata/content_references/\d+/fallback_items$",
                            m.get("p"),
                        )
                        and isinstance(m.get("v"), list)
                    ):
                        for link in m.get("v", []) or []:
                            sources.add_source(link)
                    elif (
                        m.get("p")
                        and re.match(
                            r"^/message/metadata/content_references/\d+/items$",
                            m.get("p"),
                        )
                        and isinstance(m.get("v"), list)
                    ):
                        for link in m.get("v", []) or []:
                            sources.add_source(link)
                    elif (
                        m.get("p")
                        and re.match(
                            r"^/message/metadata/content_references/\d+/refs$",
                            m.get("p"),
                        )
                        and isinstance(m.get("v"), list)
                    ):
                        match = re.match(
                            r"^/message/metadata/content_references/(\d+)/refs$",
                            m.get("p"),
                        )
                        if match:
                            idx = int(match.group(1))
                            references.update_reference(
                                idx, m.get("o"), "refs", m.get("v")
                            )
                    elif (
                        m.get("p")
                        and re.match(
                            r"^/message/metadata/content_references/\d+/alt$",
                            m.get("p"),
                        )
                        and isinstance(m.get("v"), list)
                    ):
                        match = re.match(
                            r"^/message/metadata/content_references/(\d+)/alt$",
                            m.get("p"),
                        )
                        if match:
                            idx = int(match.group(1))
                            references.update_reference(
                                idx, m.get("o"), "alt", m.get("v")
                            )
                    elif (
                        m.get("p")
                        and re.match(
                            r"^/message/metadata/content_references/\d+/prompt_text$",
                            m.get("p"),
                        )
                        and isinstance(m.get("v"), list)
                    ):
                        match = re.match(
                            r"^/message/metadata/content_references/(\d+)/prompt_text$",
                            m.get("p"),
                        )
                        if match:
                            idx = int(match.group(1))
                            references.update_reference(
                                idx, m.get("o"), "prompt_text", m.get("v")
                            )
                    elif (
                        m.get("p")
                        and re.match(
                            r"^/message/metadata/content_references/\d+/refs/\d+$",
                            m.get("p"),
                        )
                        and isinstance(m.get("v"), dict)
                    ):
                        match = re.match(
                            r"^/message/metadata/content_references/(\d+)/refs/(\d+)$",
                            m.get("p"),
                        )
                        if match:
                            reference_idx = int(match.group(1))
                            ref_idx = int(match.group(2))
                            references.update_reference(
                                reference_idx, m.get("o"), "refs", m.get("v"), ref_idx
                            )
                    elif (
                        m.get("p")
                        and re.match(
                            r"^/message/metadata/content_references/\d+/images$",
                            m.get("p"),
                        )
                        and isinstance(m.get("v"), list)
                    ):
                        match = re.match(
                            r"^/message/metadata/content_references/(\d+)/images$",
                            m.get("p"),
                        )
                        if match:
                            idx = int(match.group(1))
                            references.update_reference(
                                idx, m.get("o"), "images", m.get("v")
                            )
                    elif m.get("p") == "/message/metadata/finished_text":
                        fields.is_thinking = False
                        if buffer:
                            yield buffer
                        yield Reasoning(status=m.get("v"))
                    elif (
                        m.get("p") == "/message/metadata" and fields.recipient == "all"
                    ):
                        fields.finish_reason = (
                            m.get("v", {}).get("finish_details", {}).get("type")
                        )
                        break

                yield buffer
            elif isinstance(v, dict):
                if fields.conversation_id is None:
                    fields.conversation_id = v.get("conversation_id")
                    debug.log(f"OpenaiChat: New conversation: {fields.conversation_id}")
                m = v.get("message", {})
                fields.recipient = m.get("recipient", fields.recipient)
                if fields.recipient == "all":
                    c = m.get("content", {})
                    if (
                        c.get("content_type") == "text"
                        and m.get("author", {}).get("role") == "tool"
                        and "initial_text" in m.get("metadata", {})
                    ):
                        fields.is_thinking = True
                        yield Reasoning(
                            status=m.get("metadata", {}).get("initial_text")
                        )
                    # if c.get("content_type") == "multimodal_text":
                    #    for part in c.get("parts"):
                    #        if isinstance(part, dict) and part.get("content_type") == "image_asset_pointer":
                    #            yield await cls.get_generated_image(session, auth_result, part, fields.prompt, fields.conversation_id)
                    if m.get("author", {}).get("role") == "assistant":
                        if fields.parent_message_id is None:
                            fields.parent_message_id = v.get("message", {}).get("id")
                        fields.message_id = v.get("message", {}).get("id")
                    if m.get("status") == "finished_successfully" and m.get(
                        "metadata", {}
                    ).get("image_gen_task_id"):
                        fields.task = v
            return
        if "error" in line and line.get("error"):
            raise RuntimeError(line.get("error"))

    @classmethod
    async def synthesize(cls, params: dict) -> AsyncIterator[bytes]:
        async with StreamSession(impersonate="chrome", timeout=0) as session:
            async with session.get(
                f"{cls.url}/backend-api/synthesize", params=params, headers=cls._headers
            ) as response:
                await raise_for_status(response)
                async for chunk in response.iter_content():
                    yield chunk

    @classmethod
    async def login_generator(
        cls,
        proxy: str = None,
        api_key: str = None,
        proof_token: str = None,
        cookies: Cookies = None,
        headers: dict = None,
        **kwargs,
    ) -> AsyncIterator:
        if cls._expires is not None and (cls._expires - 60 * 10) < time.time():
            cls._headers = cls._api_key = None
        if cls._headers is None or headers is not None:
            cls._headers = {} if headers is None else headers
        if proof_token is not None:
            cls.request_config.proof_token = proof_token
        if cookies is not None:
            cls.request_config.cookies = cookies
        if api_key is not None:
            cls._create_request_args(
                cls.request_config.cookies, cls.request_config.headers
            )
            cls._set_api_key(api_key)
        else:
            try:
                cls.request_config = await get_request_config(cls.request_config, proxy)
                if cls.request_config is None:
                    cls.request_config = RequestConfig()
                cls._create_request_args(
                    cls.request_config.cookies, cls.request_config.headers
                )
                if cls.needs_auth and cls.request_config.access_token is None:
                    raise NoValidHarFileError(f"Missing access token")
                if not cls._set_api_key(cls.request_config.access_token):
                    raise NoValidHarFileError(
                        f"Access token is not valid: {cls.request_config.access_token}"
                    )
            except NoValidHarFileError:
                if has_nodriver:
                    if cls.request_config.access_token is None:
                        yield RequestLogin(
                            cls.label, os.environ.get("G4F_LOGIN_URL", "")
                        )
                        await cls.nodriver_auth(proxy)
                else:
                    raise

    @classmethod
    async def nodriver_auth(cls, proxy: str = None):
        async with get_nodriver_session(proxy=proxy) as browser:
            page = await browser.get(cls.url)

            def on_request(event: nodriver.cdp.network.RequestWillBeSent, page=None):
                if not hasattr(event, "request"):
                    return
                if event.request.url == start_url or event.request.url.startswith(
                    conversation_url
                ):
                    if cls.request_config.headers is None:
                        cls.request_config.headers = {}
                    for key, value in event.request.headers.items():
                        cls.request_config.headers[key.lower()] = value
                elif event.request.url in (backend_url, backend_anon_url, prepare_url):
                    if "OpenAI-Sentinel-Proof-Token" in event.request.headers:
                        cls.request_config.proof_token = json.loads(
                            base64.b64decode(
                                event.request.headers["OpenAI-Sentinel-Proof-Token"]
                                .split("gAAAAAB", 1)[-1]
                                .split("~")[0]
                                .encode()
                            ).decode()
                        )
                    if "OpenAI-Sentinel-Turnstile-Token" in event.request.headers:
                        cls.request_config.turnstile_token = event.request.headers[
                            "OpenAI-Sentinel-Turnstile-Token"
                        ]
                    if "Authorization" in event.request.headers:
                        cls._api_key = event.request.headers["Authorization"].split()[
                            -1
                        ]
                elif event.request.url == arkose_url:
                    cls.request_config.arkose_request = arkReq(
                        arkURL=event.request.url,
                        arkBx=None,
                        arkHeader=event.request.headers,
                        arkBody=event.request.post_data,
                        userAgent=event.request.headers.get("User-Agent"),
                    )

            await page.send(nodriver.cdp.network.enable())
            page.add_handler(nodriver.cdp.network.RequestWillBeSent, on_request)
            await page.reload()
            user_agent = await page.evaluate(
                "window.navigator.userAgent", return_by_value=True
            )
            debug.log(f"OpenaiChat: User-Agent: {user_agent}")
            for _ in range(3):
                try:
                    if cls.needs_auth:
                        try:
                            await page.select(
                                '[data-testid="accounts-profile-button"]', 300
                            )
                        except TimeoutError:
                            continue
                    try:
                        textarea = await page.select("#prompt-textarea", 300)
                        await textarea.send_keys("Hello")
                        await asyncio.sleep(1)
                    except TimeoutError:
                        continue
                except nodriver.core.connection.ProtocolException:
                    continue
                break
            try:
                button = await page.select('[data-testid="send-button"]')
                await button.click()
                debug.log("OpenaiChat: 'Hello' sended")
            except TimeoutError:
                pass
            while True:
                body = await page.evaluate(
                    "JSON.stringify(window.__remixContext)", return_by_value=True
                )
                if hasattr(body, "value"):
                    body = body.value
                if body:
                    match = re.search(r'"accessToken":"(.+?)"', body)
                    if match:
                        cls._api_key = match.group(1)
                        break
                if cls._api_key is not None or not cls.needs_auth:
                    break
                await asyncio.sleep(1)
                debug.log("OpenaiChat: Waiting for access token...")
            debug.log(
                f"OpenaiChat: Access token: {'False' if cls._api_key is None else cls._api_key[:12] + '...'}"
            )
            # while True:
            #    if cls.request_config.proof_token:
            #        break
            #    await asyncio.sleep(1)
            #    debug.log("OpenaiChat: Waiting for proof token...")
            # debug.log(f"OpenaiChat: Proof token: Yes")
            cls.request_config.data_build = await page.evaluate(
                "document.documentElement.getAttribute('data-build')"
            )
            cls.request_config.cookies = await page.send(get_cookies([cls.url]))
            await page.close()
            cls._create_request_args(
                cls.request_config.cookies,
                cls.request_config.headers,
                user_agent=user_agent,
            )
            cls._set_api_key(cls._api_key)
            debug.log(f"OpenaiChat: Sleep 10s")
            await asyncio.sleep(10)

    @staticmethod
    def get_default_headers() -> Dict[str, str]:
        return {
            **DEFAULT_HEADERS,
            "content-type": "application/json",
        }

    @classmethod
    def _create_request_args(
        cls, cookies: Cookies = None, headers: dict = None, user_agent: str = None
    ):
        cls._headers = cls.get_default_headers() if headers is None else headers
        if user_agent is not None:
            cls._headers["user-agent"] = user_agent
        cls._cookies = {} if cookies is None else cookies
        cls._update_cookie_header()

    @classmethod
    def _update_request_args(cls, auth_result: AuthResult, session: StreamSession):
        if hasattr(auth_result, "cookies"):
            for c in (
                session.cookie_jar
                if hasattr(session, "cookie_jar")
                else session.cookies.jar
            ):
                auth_result.cookies[getattr(c, "key", getattr(c, "name", ""))] = c.value
            cls._cookies = auth_result.cookies
        cls._update_cookie_header()

    @classmethod
    def _set_api_key(cls, api_key: str):
        cls._api_key = api_key
        if api_key:
            exp = api_key.split(".")[1]
            exp = (exp + "=" * (4 - len(exp) % 4)).encode()
            cls._expires = json.loads(base64.b64decode(exp)).get("exp")
            debug.log(
                f"OpenaiChat: API key expires at\n {cls._expires} we have:\n {time.time()}"
            )
            if time.time() > cls._expires:
                debug.log(f"OpenaiChat: API key is expired")
                return False
            else:
                cls._headers["authorization"] = f"Bearer {api_key}"
                return True
        return True

    @classmethod
    def _update_cookie_header(cls):
        if cls._cookies:
            cls._headers["cookie"] = format_cookies(cls._cookies)


class Conversation(JsonConversation):
    """
    Class to encapsulate response fields.
    """

    def __init__(
        self,
        conversation_id: str = None,
        message_id: str = None,
        user_id: str = None,
        finish_reason: str = None,
        parent_message_id: str = None,
        is_thinking: bool = False,
    ):
        self.conversation_id = conversation_id
        self.message_id = message_id
        self.finish_reason = finish_reason
        self.recipient = "all"
        self.parent_message_id = (
            message_id if parent_message_id is None else parent_message_id
        )
        self.user_id = user_id
        self.is_thinking = is_thinking
        self.p = None
        self.thoughts_summary = ""
        self.prompt = None
        self.generated_images: ImagePreview = None
        self.task: dict = None


def get_cookies(
    urls: Optional[Iterator[str]] = None,
) -> Generator[Dict, Dict, Dict[str, str]]:
    params = {}
    if urls is not None:
        params["urls"] = [i for i in urls]
    cmd_dict = {
        "method": "Network.getCookies",
        "params": params,
    }
    json = yield cmd_dict
    return {c["name"]: c["value"] for c in json["cookies"]} if "cookies" in json else {}


class OpenAISources(ResponseType):
    list: List[Dict[str, str]]

    def __init__(self, sources: List[Dict[str, str]]) -> None:
        """Initialize with a list of source dictionaries."""
        self.list = []
        for source in sources:
            self.add_source(source)

    def add_source(self, source: Union[Dict[str, str], str]) -> None:
        """Add a source to the list, cleaning the URL if necessary."""
        source = source if isinstance(source, dict) else {"url": source}
        url = source.get("url", source.get("link", None))
        if not url:
            return

        url = re.sub(r"[&?]utm_source=.+", "", url)
        source["url"] = url

        ref_info = self.get_ref_info(source)
        if ref_info:
            existing_source, idx = self.find_by_ref_info(ref_info)
            if existing_source and idx is not None:
                self.list[idx] = source
                return

        existing_source, idx = self.find_by_url(source["url"])
        if existing_source and idx is not None:
            self.list[idx] = source
            return

        self.list.append(source)

    def __str__(self) -> str:
        """Return formatted sources as a string."""
        if not self.list:
            return ""
        return "\n\n\n\n" + (
            "\n>\n".join(
                [
                    f"> [{idx + 1}] {format_link(link['url'], link.get('title', ''))}"
                    for idx, link in enumerate(self.list)
                ]
            )
        )

    def get_ref_info(self, source: Dict[str, str]) -> dict[str, str | int] | None:
        ref_index = source.get("ref_id", {}).get("ref_index", None)
        ref_type = source.get("ref_id", {}).get("ref_type", None)
        if isinstance(ref_index, int):
            return {
                "ref_index": ref_index,
                "ref_type": ref_type,
            }

        for ref_info in source.get("refs") or []:
            ref_index = ref_info.get("ref_index", None)
            ref_type = ref_info.get("ref_type", None)
            if isinstance(ref_index, int):
                return {
                    "ref_index": ref_index,
                    "ref_type": ref_type,
                }

        return None

    def find_by_ref_info(self, ref_info: dict[str, str | int]):
        for idx, source in enumerate(self.list):
            source_ref_info = self.get_ref_info(source)
            if (
                source_ref_info
                and source_ref_info["ref_index"] == ref_info["ref_index"]
                and source_ref_info["ref_type"] == ref_info["ref_type"]
            ):
                return source, idx

        return None, None

    def find_by_url(self, url: str):
        for idx, source in enumerate(self.list):
            if source["url"] == url:
                return source, idx
        return None, None

    def get_index(self, ref_info: dict[str, str | int]) -> int | None:
        _, index = self.find_by_ref_info(ref_info)
        if index is not None:
            return index

        return None


class ContentReferences:
    def __init__(self) -> None:
        self.list: List[Dict[str, Any]] = []

    def add_reference(self, reference_part: dict) -> None:
        self.list.append(reference_part)

    def merge_reference(self, idx: int, reference_part: dict):
        while len(self.list) <= idx:
            self.list.append({})

        self.list[idx] = {**self.list[idx], **reference_part}

    def update_reference(
        self, idx: int, operation: str, field: str, value: Any, ref_idx=None
    ) -> None:
        while len(self.list) <= idx:
            self.list.append({})

        if operation == "append" or operation == "add":
            if not isinstance(self.list[idx].get(field, None), list):
                self.list[idx][field] = []
            if isinstance(value, list):
                self.list[idx][field].extend(value)
            else:
                self.list[idx][field].append(value)

        if operation == "replace" and ref_idx is not None:
            if field == "refs" and not isinstance(
                self.list[idx].get(field, None), list
            ):
                self.list[idx][field] = []

            if isinstance(self.list[idx][field], list):
                if len(self.list[idx][field]) <= ref_idx:
                    self.list[idx][field].append(value)
                else:
                    self.list[idx][field][ref_idx] = value
            else:
                self.list[idx][field] = value

    def get_ref_info(
        self, source: Dict[str, str], target_ref_info: Dict[str, Union[str, int]]
    ) -> dict[str, str | int] | None:
        for idx, ref_info in enumerate(source.get("refs", [])) or []:
            if not isinstance(ref_info, dict):
                continue

            ref_index = ref_info.get("ref_index", None)
            ref_type = ref_info.get("ref_type", None)
            if isinstance(ref_index, int) and isinstance(ref_type, str):
                if not target_ref_info or (
                    target_ref_info["ref_index"] == ref_index
                    and target_ref_info["ref_type"] == ref_type
                ):
                    return {"ref_index": ref_index, "ref_type": ref_type, "idx": idx}

        return None

    def get_reference(self, ref_info: Dict[str, Union[str, int]]) -> Any:
        for reference in self.list:
            reference_ref_info = self.get_ref_info(reference, ref_info)

            if (
                not reference_ref_info
                or reference_ref_info["ref_index"] != ref_info["ref_index"]
                or reference_ref_info["ref_type"] != ref_info["ref_type"]
            ):
                continue

            if ref_info["ref_type"] != "image":
                return reference

            images = reference.get("images", [])
            if isinstance(images, list) and len(images) > reference_ref_info["idx"]:
                return images[reference_ref_info["idx"]]

        return None