Skip to content

Fetch msg

TelegramDialog

Bases: BaseModel

channel_name 

channel_name: str

channel_id 

channel_id: int

TelegramMessage

Bases: BaseModel

url 

url: str

text 

text: str | None | Any

TelegramManager

Bases: BaseModel

Methods:

Name Description
get_personal_info
get_channel_names
get_channel_messages
get_all_messages

client 

client: TelegramClient = Field(
    default=None,
    title="Telegram Client",
    description="Telethon client for interacting with Telegram API, it will be initialized after the model is created.",
)

get_personal_info 

get_personal_info() -> User
Source code in src/fetch_msg.py 
44
45
46
47
48
async def get_personal_info(self) -> User:
    me = await self.client.get_me()
    if not isinstance(me, User):
        raise ValueError("You are not logged in.")
    return me

get_channel_names 

get_channel_names() -> list[TelegramDialog]
Source code in src/fetch_msg.py 
50
51
52
53
54
55
56
async def get_channel_names(self) -> list[TelegramDialog]:
    result = []
    async for dialog in self.client.iter_dialogs():
        if isinstance(dialog, Dialog):
            telegram_dialog = TelegramDialog(channel_name=dialog.name, channel_id=dialog.id)
            result.append(telegram_dialog)
    return result

get_channel_messages 

get_channel_messages(channel_name: str) -> list[TelegramMessage]
Source code in src/fetch_msg.py 
58
59
60
61
62
63
64
65
66
async def get_channel_messages(self, channel_name: str) -> list[TelegramMessage]:
    channel = await self.client.get_entity(channel_name)
    result = []
    async for message in self.client.iter_messages(channel):
        if isinstance(message, Message) and (message.photo or message.video):
            url = f"https://t.me/c/{channel.id}/{message.id}"
            logfire.info("Found media message", url=url)
            result.append(TelegramMessage(url=url, text=message.text))
    return result

get_all_messages 

get_all_messages() -> None
Source code in src/fetch_msg.py 
68
69
70
71
72
73
74
75
76
77
78
79
async def get_all_messages(self) -> None:
    await self.client.start()
    me = await self.get_personal_info()
    logfire.info("Logged in as", phone=me.phone)

    channels = await self.get_channel_names()
    for channel in channels:
        messages = await self.get_channel_messages(channel_name=channel.channel_name)
        for message in messages:
            logfire.info("Fetched messages", **message.model_dump())
            break
        break