ryg/royalnet
ryg/royalnet
1
Fork 0
mirror of https://github.com/RYGhub/royalnet.git synced 2024-11-23 19:44:20 +00:00
Code Activity

🔧 Make bullet fields private

Browse source
This commit is contained in:
Steffo 2021-04-03 19:10:38 +02:00
parent 14c94c7aa2
commit 3ab014cc6f
Signed by: steffo
GPG key ID: 6965406171929D01
4 changed files with 43 additions and 43 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
pyproject.toml
View file

@ -1,6 +1,6 @@
[tool.poetry] [tool.poetry]
name = "royalnet-telethon" name = "royalnet-telethon"
version = "0.1.6" version = "0.2.0"
description = "A Telethon-based frontend for the royalnet.engineer module." description = "A Telethon-based frontend for the royalnet.engineer module."
authors = ["Stefano Pigozzi <me@steffo.eu>"] authors = ["Stefano Pigozzi <me@steffo.eu>"]
license = "AGPL-3.0-or-later" license = "AGPL-3.0-or-later"

46
royalnet_telethon/bullet/contents/__init__.py
View file

@ -13,75 +13,75 @@ import datetime
class TelegramMessage(co.Message): class TelegramMessage(co.Message):
def __init__(self, msg: tlc.Message): def __init__(self, msg: tlc.Message):
super().__init__() super().__init__()
self.msg: tlc.Message = msg self._msg: tlc.Message = msg
def __hash__(self) -> int: def __hash__(self) -> int:
return self.msg.id return self._msg.id
@ap.async_property @ap.async_property
async def text(self) -> t.Optional[str]: async def text(self) -> t.Optional[str]:
return self.msg.text return self._msg.text
@ap.async_property @ap.async_property
async def timestamp(self) -> t.Optional[datetime.datetime]: async def timestamp(self) -> t.Optional[datetime.datetime]:
return max(self.msg.date, self.msg.edit_date) return max(self._msg.date, self._msg.edit_date)
@ap.async_property @ap.async_property
async def channel(self) -> t.Optional[TelegramChannel]: async def channel(self) -> t.Optional[TelegramChannel]:
channel: t.Union[tlt.Chat, tlt.User, tlt.Channel] = await self.msg.get_chat() channel: t.Union[tlt.Chat, tlt.User, tlt.Channel] = await self._msg.get_chat()
return TelegramChannel(channel=channel, client=self.msg.client) return TelegramChannel(channel=channel, client=self._msg.client)
@ap.async_property @ap.async_property
async def sender(self) -> t.Optional[TelegramUser]: async def sender(self) -> t.Optional[TelegramUser]:
sender: tlt.User = await self.msg.get_sender() sender: tlt.User = await self._msg.get_sender()
return TelegramUser(user=sender, client=self.msg.client) return TelegramUser(user=sender, client=self._msg.client)
async def reply(self, *, async def reply(self, *,
text: str = None, text: str = None,
files: t.List[t.BinaryIO] = None) -> t.Optional[TelegramMessage]: files: t.List[t.BinaryIO] = None) -> t.Optional[TelegramMessage]:
sent = await self.msg.reply(message=text, file=files) sent = await self._msg.reply(message=text, file=files)
return TelegramMessage(msg=sent) return TelegramMessage(msg=sent)
class TelegramChannel(co.Channel): class TelegramChannel(co.Channel):
def __init__(self, channel: t.Union[tlt.Chat, tlt.User, tlt.Channel], client: tt.TelegramClient): def __init__(self, channel: t.Union[tlt.Chat, tlt.User, tlt.Channel], client: tt.TelegramClient):
super().__init__() super().__init__()
self.channel: t.Union[tlt.Chat, tlt.User, tlt.Channel] = channel self._channel: t.Union[tlt.Chat, tlt.User, tlt.Channel] = channel
self.client: tt.TelegramClient = client self._client: tt.TelegramClient = client
def __hash__(self): def __hash__(self):
return self.channel.id return self._channel.id
@ap.async_property @ap.async_property
async def name(self) -> t.Optional[str]: async def name(self) -> t.Optional[str]:
return self.channel.title return self._channel.title
async def send_message(self, *, async def send_message(self, *,
text: str = None, text: str = None,
files: t.List[t.BinaryIO] = None) -> t.Optional[TelegramMessage]: files: t.List[t.BinaryIO] = None) -> t.Optional[TelegramMessage]:
sent = await self.client.send_message(self.channel, message=text, file=files) sent = await self._client.send_message(self._channel, message=text, file=files)
return TelegramMessage(msg=sent) return TelegramMessage(msg=sent)
class TelegramUser(co.User): class TelegramUser(co.User):
def __init__(self, user: tlt.User, client: tt.TelegramClient): def __init__(self, user: tlt.User, client: tt.TelegramClient):
super().__init__() super().__init__()
self.user: tlt.User = user self._user: tlt.User = user
self.client: tt.TelegramClient = client self._client: tt.TelegramClient = client
def __hash__(self): def __hash__(self):
return self.user.id return self._user.id
@ap.async_property @ap.async_property
async def name(self) -> t.Optional[str]: async def name(self) -> t.Optional[str]:
if self.user.username: if self._user.username:
return f"{self.user.username}" return f"{self._user.username}"
elif self.user.last_name: elif self._user.last_name:
return f"{self.user.first_name} {self.user.last_name}" return f"{self._user.first_name} {self._user.last_name}"
return f"{self.user.first_name}" return f"{self._user.first_name}"
async def slide(self) -> TelegramChannel: async def slide(self) -> TelegramChannel:
return TelegramChannel(channel=self.user, client=self.client) return TelegramChannel(channel=self._user, client=self._client)
__all__ = ( __all__ = (

18
royalnet_telethon/bullet/projectiles/message.py
View file

@ -7,40 +7,40 @@ from ..contents.__init__ import TelegramMessage
class TelegramMessageReceived(p.MessageReceived): class TelegramMessageReceived(p.MessageReceived):
def __init__(self, event: tlc.Message): def __init__(self, event: tlc.Message):
super().__init__() super().__init__()
self.event: tlc.Message = event self._event: tlc.Message = event
def __hash__(self) -> int: def __hash__(self) -> int:
return self.event.id return self._event.id
@ap.async_cached_property @ap.async_cached_property
async def message(self) -> TelegramMessage: async def message(self) -> TelegramMessage:
return TelegramMessage(msg=self.event) return TelegramMessage(msg=self._event)
class TelegramMessageEdited(p.MessageEdited): class TelegramMessageEdited(p.MessageEdited):
def __init__(self, event: tlc.Message): def __init__(self, event: tlc.Message):
super().__init__() super().__init__()
self.event: tlc.Message = event self._event: tlc.Message = event
def __hash__(self) -> int: def __hash__(self) -> int:
return self.event.id return self._event.id
@ap.async_cached_property @ap.async_cached_property
async def message(self) -> TelegramMessage: async def message(self) -> TelegramMessage:
return TelegramMessage(msg=self.event) return TelegramMessage(msg=self._event)
class TelegramMessageDeleted(p.MessageDeleted): class TelegramMessageDeleted(p.MessageDeleted):
def __init__(self, event: tlc.Message): def __init__(self, event: tlc.Message):
super().__init__() super().__init__()
self.event: tlc.Message = event self._event: tlc.Message = event
def __hash__(self) -> int: def __hash__(self) -> int:
return self.event.id return self._event.id
@ap.async_cached_property @ap.async_cached_property
async def message(self) -> TelegramMessage: async def message(self) -> TelegramMessage:
return TelegramMessage(msg=self.event) return TelegramMessage(msg=self._event)
__all__ = ( __all__ = (

20
royalnet_telethon/pda.py
View file

@ -54,13 +54,13 @@ class TelethonPDA:
self.conversations: t.List[engi.Conversation] = [] self.conversations: t.List[engi.Conversation] = []
""" """
A :class:`list` of conversations to run before a new event is :meth:`.put` in a A :class:`list` of conversations to run before a new _event is :meth:`.put` in a
:class:`~royalnet.engineer.dispenser.Dispenser`. :class:`~royalnet.engineer.dispenser.Dispenser`.
""" """
self.client: tt.TelegramClient = tt.TelegramClient("bot", api_id=tg_api_id, api_hash=tg_api_hash) self.client: tt.TelegramClient = tt.TelegramClient("bot", api_id=tg_api_id, api_hash=tg_api_hash)
""" """
The :mod:`telethon` Telegram client that this PDA will use to interface with Telegram. The :mod:`telethon` Telegram _client that this PDA will use to interface with Telegram.
""" """
self._register_events() self._register_events()
@ -76,12 +76,12 @@ class TelethonPDA:
self.client.add_event_handler(callback=self._message_new, event=tt.events.NewMessage()) self.client.add_event_handler(callback=self._message_new, event=tt.events.NewMessage())
self.client.add_event_handler(callback=self._message_edit, event=tt.events.MessageEdited()) self.client.add_event_handler(callback=self._message_edit, event=tt.events.MessageEdited())
self.client.add_event_handler(callback=self._message_delete, event=tt.events.MessageDeleted()) self.client.add_event_handler(callback=self._message_delete, event=tt.events.MessageDeleted())
# self.client.add_event_handler(callback=self._message_read, event=tt.events.MessageRead()) # self._client.add_event_handler(callback=self._message_read, _event=tt.events.MessageRead())
# self.client.add_event_handler(callback=self._chat_action, event=tt.events.ChatAction()) # self._client.add_event_handler(callback=self._chat_action, _event=tt.events.ChatAction())
# self.client.add_event_handler(callback=self._user_update, event=tt.events.UserUpdate()) # self._client.add_event_handler(callback=self._user_update, _event=tt.events.UserUpdate())
# self.client.add_event_handler(callback=self._callback_query, event=tt.events.CallbackQuery()) # self._client.add_event_handler(callback=self._callback_query, _event=tt.events.CallbackQuery())
# self.client.add_event_handler(callback=self._inline_query, event=tt.events.InlineQuery()) # self._client.add_event_handler(callback=self._inline_query, _event=tt.events.InlineQuery())
# self.client.add_event_handler(callback=self._album, event=tt.events.Album()) # self._client.add_event_handler(callback=self._album, _event=tt.events.Album())
def _determine_key(self, event: tlc.message.Message): def _determine_key(self, event: tlc.message.Message):
if self.mode == TelethonPDAMode.GLOBAL: if self.mode == TelethonPDAMode.GLOBAL:
@ -186,13 +186,13 @@ class TelethonPDA:
log.debug(f"Creating run task for: {conversation!r}") log.debug(f"Creating run task for: {conversation!r}")
loop.create_task(dispenser.run(conversation, _pda=self), name=f"{repr(conversation)}") loop.create_task(dispenser.run(conversation, _pda=self), name=f"{repr(conversation)}")
log.debug("Running a event loop cycle...") log.debug("Running a _event loop cycle...")
await asyncio.sleep(0) await asyncio.sleep(0)
log.debug(f"Putting projectile {proj!r} in dispenser {dispenser!r}...") log.debug(f"Putting projectile {proj!r} in dispenser {dispenser!r}...")
await dispenser.put(proj) await dispenser.put(proj)
log.debug("Awaiting another event loop cycle...") log.debug("Awaiting another _event loop cycle...")
await asyncio.sleep(0) await asyncio.sleep(0)