# Copyright (c) 2023 - 2025, AG2ai, Inc., AG2ai open-source projects maintainers and core contributors
#
# SPDX-License-Identifier: Apache-2.0

from datetime import datetime
from typing import Annotated, Any, Union

from .....doc_utils import export_module
from .....import_utils import optional_import_block, require_optional_import
from .... import Tool
from ....dependency_injection import Depends, on

__all__ = ["TelegramRetrieveTool", "TelegramSendTool"]

with optional_import_block():
    from telethon import TelegramClient
    from telethon.tl.types import InputMessagesFilterEmpty, Message, PeerChannel, PeerChat, PeerUser

MAX_MESSAGE_LENGTH = 4096


@require_optional_import(["telethon", "telethon.tl.types"], "commsagent-telegram")
@export_module("autogen.tools.experimental")
class BaseTelegramTool:
    """Base class for Telegram tools containing shared functionality."""

    def __init__(self, api_id: str, api_hash: str, session_name: str) -> None:
        self._api_id = api_id
        self._api_hash = api_hash
        self._session_name = session_name

    def _get_client(self) -> "TelegramClient":  # type: ignore[no-any-unimported]
        """Get a fresh TelegramClient instance."""
        return TelegramClient(self._session_name, self._api_id, self._api_hash)

    @staticmethod
    def _get_peer_from_id(chat_id: str) -> Union["PeerChat", "PeerChannel", "PeerUser"]:  # type: ignore[no-any-unimported]
        """Convert a chat ID string to appropriate Peer type."""
        try:
            # Convert string to integer
            id_int = int(chat_id)

            # Channel/Supergroup: -100 prefix
            if str(chat_id).startswith("-100"):
                channel_id = int(str(chat_id)[4:])  # Remove -100 prefix
                return PeerChannel(channel_id)

            # Group: negative number without -100 prefix
            elif id_int < 0:
                group_id = -id_int  # Remove the negative sign
                return PeerChat(group_id)

            # User/Bot: positive number
            else:
                return PeerUser(id_int)

        except ValueError as e:
            raise ValueError(f"Invalid chat_id format: {chat_id}. Error: {str(e)}")

    async def _initialize_entity(self, client: "TelegramClient", chat_id: str) -> Any:  # type: ignore[no-any-unimported]
        """Initialize and cache the entity by trying different methods."""
        peer = self._get_peer_from_id(chat_id)

        try:
            # Try direct entity resolution first
            entity = await client.get_entity(peer)
            return entity
        except ValueError:
            try:
                # Get all dialogs (conversations)
                async for dialog in client.iter_dialogs():
                    # For users/bots, we need to find the dialog with the user
                    if (
                        isinstance(peer, PeerUser)
                        and dialog.entity.id == peer.user_id
                        or dialog.entity.id == getattr(peer, "channel_id", getattr(peer, "chat_id", None))
                    ):
                        return dialog.entity

                # If we get here, we didn't find the entity in dialogs
                raise ValueError(f"Could not find entity {chat_id} in dialogs")
            except Exception as e:
                raise ValueError(
                    f"Could not initialize entity for {chat_id}. "
                    f"Make sure you have access to this chat. Error: {str(e)}"
                )


@require_optional_import(["telethon"], "commsagent-telegram")
@export_module("autogen.tools.experimental")
class TelegramSendTool(BaseTelegramTool, Tool):
    """Sends a message to a Telegram channel, group, or user."""

    def __init__(self, *, api_id: str, api_hash: str, chat_id: str) -> None:
        """Initialize the TelegramSendTool.

        Args:
            api_id: Telegram API ID from https://my.telegram.org/apps.
            api_hash: Telegram API hash from https://my.telegram.org/apps.
            chat_id: The ID of the destination (Channel, Group, or User ID).
        """
        BaseTelegramTool.__init__(self, api_id, api_hash, "telegram_send_session")

        async def telegram_send_message(
            message: Annotated[str, "Message to send to the chat."],
            chat_id: Annotated[str, Depends(on(chat_id))],
        ) -> Any:
            """Sends a message to a Telegram chat.

            Args:
                message: The message to send.
                chat_id: The ID of the destination. (uses dependency injection)
            """
            try:
                client = self._get_client()
                async with client:
                    # Initialize and cache the entity
                    entity = await self._initialize_entity(client, chat_id)

                    if len(message) > MAX_MESSAGE_LENGTH:
                        chunks = [
                            message[i : i + (MAX_MESSAGE_LENGTH - 1)]
                            for i in range(0, len(message), (MAX_MESSAGE_LENGTH - 1))
                        ]
                        first_message: Message | None = None  # type: ignore[no-any-unimported]

                        for i, chunk in enumerate(chunks):
                            sent = await client.send_message(
                                entity=entity,
                                message=chunk,
                                parse_mode="html",
                                reply_to=first_message.id if first_message else None,
                            )

                            # Store the first message to chain replies
                            if i == 0:
                                first_message = sent
                                sent_message_id = str(sent.id)

                        return (
                            f"Message sent successfully ({len(chunks)} chunks, first ID: {sent_message_id}):\n{message}"
                        )
                    else:
                        sent = await client.send_message(entity=entity, message=message, parse_mode="html")
                        return f"Message sent successfully (ID: {sent.id}):\n{message}"

            except Exception as e:
                return f"Message send failed, exception: {str(e)}"

        Tool.__init__(
            self,
            name="telegram_send",
            description="Sends a message to a personal channel, bot channel, group, or channel.",
            func_or_tool=telegram_send_message,
        )


@require_optional_import(["telethon"], "commsagent-telegram")
@export_module("autogen.tools.experimental")
class TelegramRetrieveTool(BaseTelegramTool, Tool):
    """Retrieves messages from a Telegram channel."""

    def __init__(self, *, api_id: str, api_hash: str, chat_id: str) -> None:
        """Initialize the TelegramRetrieveTool.

        Args:
            api_id: Telegram API ID from https://my.telegram.org/apps.
            api_hash: Telegram API hash from https://my.telegram.org/apps.
            chat_id: The ID of the chat to retrieve messages from (Channel, Group, Bot Chat ID).
        """
        BaseTelegramTool.__init__(self, api_id, api_hash, "telegram_retrieve_session")
        self._chat_id = chat_id

        async def telegram_retrieve_messages(
            chat_id: Annotated[str, Depends(on(chat_id))],
            messages_since: Annotated[
                str | None,
                "Date to retrieve messages from (ISO format) OR message ID. If None, retrieves latest messages.",
            ] = None,
            maximum_messages: Annotated[
                int | None, "Maximum number of messages to retrieve. If None, retrieves all messages since date."
            ] = None,
            search: Annotated[str | None, "Optional string to search for in messages."] = None,
        ) -> Any:
            """Retrieves messages from a Telegram chat.

            Args:
                chat_id: The ID of the chat. (uses dependency injection)
                messages_since: ISO format date string OR message ID to retrieve messages from.
                maximum_messages: Maximum number of messages to retrieve.
                search: Optional string to search for in messages.
            """
            try:
                client = self._get_client()
                async with client:
                    # Initialize and cache the entity
                    entity = await self._initialize_entity(client, chat_id)

                    # Setup retrieval parameters
                    params = {
                        "entity": entity,
                        "limit": maximum_messages if maximum_messages else None,
                        "search": search if search else None,
                        "filter": InputMessagesFilterEmpty(),
                        "wait_time": None,  # No wait time between requests
                    }

                    # Handle messages_since parameter
                    if messages_since:
                        try:
                            # Try to parse as message ID first
                            msg_id = int(messages_since)
                            params["min_id"] = msg_id
                        except ValueError:
                            # Not a message ID, try as ISO date
                            try:
                                date = datetime.fromisoformat(messages_since.replace("Z", "+00:00"))
                                params["offset_date"] = date
                                params["reverse"] = (
                                    True  # Need this because the date gets messages before a certain date by default
                                )
                            except ValueError:
                                return {
                                    "error": "Invalid messages_since format. Please provide either a message ID or ISO format date (e.g., '2025-01-25T00:00:00Z')"
                                }

                    # Retrieve messages
                    messages = []
                    count = 0
                    # For bot users, we need to get both sent and received messages
                    if isinstance(self._get_peer_from_id(chat_id), PeerUser):
                        print(f"Retrieving messages for bot chat {chat_id}")

                    async for message in client.iter_messages(**params):
                        count += 1
                        messages.append({
                            "id": str(message.id),
                            "date": message.date.isoformat(),
                            "from_id": str(message.from_id) if message.from_id else None,
                            "text": message.text,
                            "reply_to_msg_id": str(message.reply_to_msg_id) if message.reply_to_msg_id else None,
                            "forward_from": str(message.forward.from_id) if message.forward else None,
                            "edit_date": message.edit_date.isoformat() if message.edit_date else None,
                            "media": bool(message.media),
                            "entities": [
                                {"type": e.__class__.__name__, "offset": e.offset, "length": e.length}
                                for e in message.entities
                            ]
                            if message.entities
                            else None,
                        })

                        # Check if we've hit the maximum
                        if maximum_messages and len(messages) >= maximum_messages:
                            break

                    return {
                        "message_count": len(messages),
                        "messages": messages,
                        "start_time": messages_since or "latest",
                    }

            except Exception as e:
                return f"Message retrieval failed, exception: {str(e)}"

        Tool.__init__(
            self,
            name="telegram_retrieve",
            description="Retrieves messages from a Telegram chat based on datetime/message ID and/or number of latest messages.",
            func_or_tool=telegram_retrieve_messages,
        )
