mirror of
https://github.com/RYGhub/royalnet.git
synced 2024-11-23 19:44:20 +00:00
97 lines
3.1 KiB
Python
97 lines
3.1 KiB
Python
from __future__ import annotations
|
|
import royalnet.royaltyping as t
|
|
|
|
import royalnet.engineer.bullet.casing as c
|
|
import royalnet.engineer.bullet.contents as co
|
|
import telethon as tt
|
|
import telethon.tl.types as tlt
|
|
import telethon.tl.custom as tlc
|
|
import async_property as ap
|
|
import datetime
|
|
from royalnet_telethon.formatting import tg_html_format
|
|
|
|
|
|
class TelegramMessage(co.Message):
|
|
def __init__(self, msg: tlc.Message):
|
|
super().__init__()
|
|
self._msg: tlc.Message = msg
|
|
|
|
def __hash__(self) -> int:
|
|
return self._msg.id
|
|
|
|
@ap.async_property
|
|
async def text(self) -> t.Optional[str]:
|
|
return self._msg.text
|
|
|
|
@ap.async_property
|
|
async def timestamp(self) -> t.Optional[datetime.datetime]:
|
|
return max(self._msg.date, self._msg.edit_date)
|
|
|
|
@ap.async_property
|
|
async def channel(self) -> t.Optional[TelegramChannel]:
|
|
channel: t.Union[tlt.Chat, tlt.User, tlt.Channel] = await self._msg.get_chat()
|
|
return TelegramChannel(channel=channel, client=self._msg.client)
|
|
|
|
@ap.async_property
|
|
async def sender(self) -> t.Optional[TelegramUser]:
|
|
sender: tlt.User = await self._msg.get_sender()
|
|
return TelegramUser(user=sender, client=self._msg.client)
|
|
|
|
async def reply(self, *,
|
|
text: str = None,
|
|
files: t.List[t.BinaryIO] = None) -> t.Optional[TelegramMessage]:
|
|
sent = await self._msg.reply(message=tg_html_format(text) if text else None, file=files, parse_mode="HTML")
|
|
return TelegramMessage(msg=sent)
|
|
|
|
|
|
class TelegramChannel(co.Channel):
|
|
def __init__(self, channel: t.Union[tlt.Chat, tlt.User, tlt.Channel], client: tt.TelegramClient):
|
|
super().__init__()
|
|
self._channel: t.Union[tlt.Chat, tlt.User, tlt.Channel] = channel
|
|
self._client: tt.TelegramClient = client
|
|
|
|
def __hash__(self):
|
|
return self._channel.id
|
|
|
|
@ap.async_property
|
|
async def name(self) -> t.Optional[str]:
|
|
return self._channel.title
|
|
|
|
async def send_message(self, *,
|
|
text: str = None,
|
|
files: t.List[t.BinaryIO] = None) -> t.Optional[TelegramMessage]:
|
|
sent = await self._client.send_message(
|
|
self._channel,
|
|
message=tg_html_format(text) if text else None,
|
|
file=files,
|
|
parse_mode="HTML"
|
|
)
|
|
return TelegramMessage(msg=sent)
|
|
|
|
|
|
class TelegramUser(co.User):
|
|
def __init__(self, user: tlt.User, client: tt.TelegramClient):
|
|
super().__init__()
|
|
self._user: tlt.User = user
|
|
self._client: tt.TelegramClient = client
|
|
|
|
def __hash__(self):
|
|
return self._user.id
|
|
|
|
@ap.async_property
|
|
async def name(self) -> t.Optional[str]:
|
|
if self._user.username:
|
|
return f"{self._user.username}"
|
|
elif self._user.last_name:
|
|
return f"{self._user.first_name} {self._user.last_name}"
|
|
return f"{self._user.first_name}"
|
|
|
|
async def slide(self) -> TelegramChannel:
|
|
return TelegramChannel(channel=self._user, client=self._client)
|
|
|
|
|
|
__all__ = (
|
|
"TelegramMessage",
|
|
"TelegramChannel",
|
|
"TelegramUser"
|
|
)
|