wah.su/stickerbridge
1
0
Fork 0
mirror of https://github.com/wah-su/stickerbridge.git synced 2025-04-05 15:54:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

feat #11: allow to manually specify import_name

Browse source
This commit is contained in:
poly000 2023-07-24 14:59:36 +08:00
parent cf8e8ff2a1
commit beea91a27e
No known key found for this signature in database
2 changed files with 25 additions and 17 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

36
stickerbridge/bot_commands.py
View file

@ -6,7 +6,13 @@ from telegram_exporter import TelegramExporter
class Command: class Command:
def __init__(self, client: AsyncClient, room: MatrixRoom, command: str, tg_exporter: TelegramExporter): def __init__(
self,
client: AsyncClient,
room: MatrixRoom,
command: str,
tg_exporter: TelegramExporter,
):
self.client = client self.client = client
self.room = room self.room = room
self.command = command.lower() self.command = command.lower()
@ -26,12 +32,11 @@ class Command:
"I am the bot that imports stickers from Telegram and upload them to Matrix rooms\n\n" "I am the bot that imports stickers from Telegram and upload them to Matrix rooms\n\n"
"List of commands:\n" "List of commands:\n"
"help - Show this help message.\n" "help - Show this help message.\n"
"import <pack_name> - Use this to import Telegram stickers from given link" "import <pack_name> [import_name] - Use this to import Telegram stickers from given link. import_name is pack_name if not provided"
) )
await send_text_to_room(self.client, self.room.room_id, text) await send_text_to_room(self.client, self.room.room_id, text)
async def _import_stickerpack(self): async def _import_stickerpack(self):
if not self.args: if not self.args:
text = ( text = (
"You need to enter stickerpack name.\n" "You need to enter stickerpack name.\n"
@ -41,26 +46,29 @@ class Command:
return return
pack_name = self.args[0] pack_name = self.args[0]
import_name = self.args[1] if len(self.args) > 1 else pack_name
reuploader = MatrixReuploader(self.client, self.room, exporter=self.tg_exporter) reuploader = MatrixReuploader(self.client, self.room, exporter=self.tg_exporter)
async for status in reuploader.import_stickerset_to_room(pack_name): async for status in reuploader.import_stickerset_to_room(
pack_name, import_name
):
switch = { switch = {
MatrixReuploader.STATUS_DOWNLOADING: f'Downloading stickerpack {pack_name}...', MatrixReuploader.STATUS_DOWNLOADING: f"Downloading stickerpack {pack_name}...",
MatrixReuploader.STATUS_UPLOADING: f'Uploading stickerpack {pack_name}...', MatrixReuploader.STATUS_UPLOADING: f"Uploading stickerpack {pack_name}...",
MatrixReuploader.STATUS_UPDATING_ROOM_STATE: f'Updating room state...', MatrixReuploader.STATUS_UPDATING_ROOM_STATE: f"Updating room state...",
MatrixReuploader.STATUS_OK: 'Done', MatrixReuploader.STATUS_OK: "Done",
MatrixReuploader.STATUS_NO_PERMISSION: ( MatrixReuploader.STATUS_NO_PERMISSION: (
'I do not have permissions to create any stickerpack in this room\n' "I do not have permissions to create any stickerpack in this room\n"
'Please, give me mod 🙏' "Please, give me mod 🙏"
), ),
MatrixReuploader.STATUS_PACK_EXISTS: ( MatrixReuploader.STATUS_PACK_EXISTS: (
f"Stickerpack '{pack_name}' already exists.\n" f"Stickerpack '{pack_name}' already exists.\n"
'Please delete it first.' "Please delete it first."
), ),
MatrixReuploader.STATUS_PACK_EMPTY: ( MatrixReuploader.STATUS_PACK_EMPTY: (
f'Warning: Telegram pack {pack_name} find out empty or not existing.' f"Warning: Telegram pack {pack_name} find out empty or not existing."
) ),
} }
text = switch.get(status, 'Warning: Unknown status') text = switch.get(status, "Warning: Unknown status")
await send_text_to_room(self.client, self.room.room_id, text) await send_text_to_room(self.client, self.room.room_id, text)
async def _unknown_command(self): async def _unknown_command(self):

6
stickerbridge/matrix_reuploader.py
View file

@ -32,18 +32,18 @@ class MatrixReuploader:
async def _has_permission_to_upload(self) -> bool: async def _has_permission_to_upload(self) -> bool:
return await has_permission(self.client, self.room.room_id, 'state_default') return await has_permission(self.client, self.room.room_id, 'state_default')
async def import_stickerset_to_room(self, pack_name: str): async def import_stickerset_to_room(self, pack_name: str, import_name: str):
if not await self._has_permission_to_upload(): if not await self._has_permission_to_upload():
yield self.STATUS_NO_PERMISSION yield self.STATUS_NO_PERMISSION
return return
stickerset = MatrixStickerset(pack_name) stickerset = MatrixStickerset(import_name)
if await is_stickerpack_existing(self.client, self.room.room_id, stickerset.name()): if await is_stickerpack_existing(self.client, self.room.room_id, stickerset.name()):
yield self.STATUS_PACK_EXISTS yield self.STATUS_PACK_EXISTS
return return
yield self.STATUS_DOWNLOADING yield self.STATUS_DOWNLOADING
converted_stickerset = await self.exporter.get_stickerset(stickerset.name()) converted_stickerset = await self.exporter.get_stickerset(pack_name)
yield self.STATUS_UPLOADING yield self.STATUS_UPLOADING
for sticker in converted_stickerset: for sticker in converted_stickerset:
with tempfile.NamedTemporaryFile('w+b') as file: with tempfile.NamedTemporaryFile('w+b') as file: