Telegram

TelegramReader

Bases: BaseReader

电报帖子/聊天消息/评论阅读器。

从Telegram频道或聊天中读取帖子/聊天消息/评论。

在使用Telegram的API之前，您需要获取自己的API ID和哈希值：

1. 使用要使用的开发者帐户的电话号码登录到您的Telegram帐户。
2. 点击API开发工具下方。
3. 将出现一个创建新应用程序窗口。填写您的应用程序详细信息。目前只有前两个字段（应用程序标题和简称）可以稍后更改，不需要输入任何URL。
4. 最后点击创建应用程序。请记住，您的API哈希是秘密的，Telegram不会让您撤销它。不要在任何地方发布它！

这个API ID和哈希是您的应用程序使用的，而不是您的电话号码。您可以使用这个API ID和哈希与任何电话号码一起使用。

Parameters:

Name	Type	Description	Default
session_name	str	要使用的会话文件的文件名（如果给定字符串，可以是完整路径），否则将使用Session实例。	required
api_id	int	您从https://my.telegram.org获取的API ID。	required
api_hash	str	您从https://my.telegram.org获取的API哈希。	required
phone_number	str	将发送验证码的电话号码。	required

Source code in llama_index/readers/telegram/base.py

class TelegramReader(BaseReader):
    """电报帖子/聊天消息/评论阅读器。

    从Telegram频道或聊天中读取帖子/聊天消息/评论。

    在使用Telegram的API之前，您需要获取自己的API ID和哈希值：

        1. 使用要使用的开发者帐户的电话号码登录到您的Telegram帐户。
        2. 点击API开发工具下方。
        3. 将出现一个创建新应用程序窗口。填写您的应用程序详细信息。目前只有前两个字段（应用程序标题和简称）可以稍后更改，不需要输入任何URL。
        4. 最后点击创建应用程序。请记住，您的API哈希是秘密的，Telegram不会让您撤销它。不要在任何地方发布它！

    这个API ID和哈希是您的应用程序使用的，而不是您的电话号码。您可以使用这个API ID和哈希与任何电话号码一起使用。

    Args:
        session_name (str): 要使用的会话文件的文件名（如果给定字符串，可以是完整路径），否则将使用Session实例。
        api_id (int): 您从https://my.telegram.org获取的API ID。
        api_hash (str): 您从https://my.telegram.org获取的API哈希。
        phone_number (str): 将发送验证码的电话号码。
"""

    def __init__(
        self,
        session_name: str,
        api_id: int,
        api_hash: str,
        phone_number: str,
    ) -> None:
        """使用参数进行初始化。"""
        super().__init__()
        self.session_name = session_name
        self.api_id = api_id
        self.api_hash = api_hash
        self.phone_number = phone_number
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.loop)

    def load_data(
        self,
        entity_name: str,
        post_id: Optional[int] = None,
        limit: Optional[int] = None,
        start_date: Optional[datetime.datetime] = None,
        end_date: Optional[datetime.datetime] = None,
    ) -> List[Document]:
        """从Telegram频道或聊天中加载帖子/聊天消息/评论。

由于Telethon是一个异步库，您需要等待协程函数才能运行它们（或者，运行循环直到它们完成）

Args:
    entity_name（str）：要检索消息历史记录的实体。
    post_id（int）：如果设置为帖子ID，则将返回回复此ID的评论。否则将获取帖子/聊天消息。
    limit（int）：要检索的消息数量。
    start_date（datetime.datetime）：时间段的开始日期。
    end_date（datetime.datetime）：时间段的结束日期。
"""
        return self.loop.run_until_complete(
            self._load_data(
                entity_name=entity_name,
                post_id=post_id,
                limit=limit,
                start_date=start_date,
                end_date=end_date,
            )
        )

    async def _load_data(
        self,
        entity_name: str,
        post_id: Optional[int] = None,
        limit: Optional[int] = None,
        start_date: Optional[datetime.datetime] = None,
        end_date: Optional[datetime.datetime] = None,
    ) -> List[Document]:
        """从Telegram频道或聊天中加载帖子/聊天消息/评论。

Args:
    entity_name（str）：要检索消息历史记录的实体。
    post_id（int）：如果设置为帖子ID，则将返回回复该ID的评论。否则将获取帖子/聊天消息。
    limit（int）：要检索的消息数量。
    start_date（datetime.datetime）：时间段的开始日期。
    end_date（datetime.datetime）：时间段的结束日期。
"""
        import telethon

        client = telethon.TelegramClient(self.session_name, self.api_id, self.api_hash)
        await client.start(phone=self.phone_number)

        results = []
        async with client:
            if end_date and start_date:
                # Asynchronously iterate over messages in between start_date and end_date
                async for message in client.iter_messages(
                    entity_name,
                    reply_to=post_id,
                    limit=limit,
                    offset_date=end_date,
                ):
                    if message.date < start_date:
                        break
                    if isinstance(message.text, str) and message.text != "":
                        results.append(Document(text=self._remove_links(message.text)))
            else:
                # Asynchronously iterate over messages
                async for message in client.iter_messages(
                    entity_name,
                    reply_to=post_id,
                    limit=limit,
                ):
                    if isinstance(message.text, str) and message.text != "":
                        results.append(Document(text=self._remove_links(message.text)))
        return results

    def _remove_links(self, string) -> str:
        """从给定的字符串中删除所有的URL，只保留基本域名。"""

        def replace_match(match):
            text = match.group(1)
            return text if text else ""

        url_pattern = r"https?://(?:www\.)?((?!www\.).)+?"
        return re.sub(url_pattern, replace_match, string)

load_data

load_data(
    entity_name: str,
    post_id: Optional[int] = None,
    limit: Optional[int] = None,
    start_date: Optional[datetime] = None,
    end_date: Optional[datetime] = None,
) -> List[Document]

从Telegram频道或聊天中加载帖子/聊天消息/评论。

由于Telethon是一个异步库，您需要等待协程函数才能运行它们（或者，运行循环直到它们完成）

Source code in llama_index/readers/telegram/base.py

    def load_data(
        self,
        entity_name: str,
        post_id: Optional[int] = None,
        limit: Optional[int] = None,
        start_date: Optional[datetime.datetime] = None,
        end_date: Optional[datetime.datetime] = None,
    ) -> List[Document]:
        """从Telegram频道或聊天中加载帖子/聊天消息/评论。

由于Telethon是一个异步库，您需要等待协程函数才能运行它们（或者，运行循环直到它们完成）

Args:
    entity_name（str）：要检索消息历史记录的实体。
    post_id（int）：如果设置为帖子ID，则将返回回复此ID的评论。否则将获取帖子/聊天消息。
    limit（int）：要检索的消息数量。
    start_date（datetime.datetime）：时间段的开始日期。
    end_date（datetime.datetime）：时间段的结束日期。
"""
        return self.loop.run_until_complete(
            self._load_data(
                entity_name=entity_name,
                post_id=post_id,
                limit=limit,
                start_date=start_date,
                end_date=end_date,
            )
        )