stdplugins/info.py

@@ -10,7 +10,7 @@ from telethon.tl.types import MessageEntityPre
 from telethon.tl.tlobject import TLObject
 import datetime
 
-PRINTABLE_SET = set(string.printable.encode())
+PRINTABLE_SET = set(bytes(string.printable, 'ascii'))
 STR_LEN_MAX = 256
 BYTE_LEN_MAX = 64
 
@@ -19,7 +19,7 @@ def parse_pre(text):
     text = text.strip()
     return (
         text,
-        [MessageEntityPre(offset=0, length=len(add_surrogate(text)), language='')]
+        [MessageEntityPre(offset=0, length=len(add_surrogate(text)), language='potato')]
     )
 
 
@@ -33,55 +33,55 @@ def yaml_format(obj, indent=0):
         obj = obj.to_dict()
 
     if isinstance(obj, dict):
-        if not obj:
-            return 'dict:'
         result.append(obj.get('_', 'dict') + ':')
-        items = obj.items()
-        has_multiple_items = len(items) > 2
-        if has_multiple_items:
-            result.append('\n')
-        indent += 2
-        for k, v in items:
-            if k == '_' or v is None:
-                continue
-            formatted = yaml_format(v, indent)
-            if not formatted.strip():
-                continue
-            result.append(' ' * (indent if has_multiple_items else 1))
-            result.append(f'{k}: {formatted}')
-            result.append('\n')
-        result.pop()
-        indent -= 2
-        result.append(' ' * indent)
+        if obj:
+            items = obj.items()
+            has_multiple_items = len(items) > 2
+            if has_multiple_items:
+                result.append('\n')
+            indent += 2
+            for k, v in items:
+                if k == '_' or v is None:
+                    continue
+                formatted = yaml_format(v, indent)
+                if not formatted.strip():
+                    continue
+                result.append(' ' * (indent if has_multiple_items else 1))
+                result.append(f'{k}: {formatted}')
+                result.append('\n')
+            result.pop()
+            indent -= 2
+            result.append(' ' * indent)
     elif isinstance(obj, str):
         # truncate long strings and display elipsis
-        result = repr(obj[:STR_LEN_MAX])
+        result.append(repr(obj[:STR_LEN_MAX]))
         if len(obj) > STR_LEN_MAX:
-            result += '…'
-        return result
+            result.append('…')
     elif isinstance(obj, bytes):
         # repr() bytes if it's printable, hex like "FF EE BB" otherwise
         if all(c in PRINTABLE_SET for c in obj):
-            return repr(obj)
+            result.append(repr(obj))
         else:
-            return ('<…>' if len(obj) > BYTE_LEN_MAX else
-                    ' '.join(f'{b:02X}' for b in obj))
+            if len(obj) > BYTE_LEN_MAX:
+                result.append('<…>')
+            else:
+                result.append(' '.join(f'{b:02X}' for b in obj))
     elif isinstance(obj, datetime.datetime):
         # ISO-8601 without timezone offset (telethon dates are always UTC)
-        return obj.strftime('%Y-%m-%d %H:%M:%S')
+        result.append(obj.strftime('%Y-%m-%d %H:%M:%S'))
     elif hasattr(obj, '__iter__'):
         # display iterables one after another at the base indentation level
         result.append('\n')
         indent += 2
         for x in obj:
-
-            result.append(f"{' ' * indent}- {yaml_format(x, indent + 2)}")
+            result.append(' ' * indent)
+            result.append(yaml_format(x, indent))
             result.append('\n')
         result.pop()
         indent -= 2
         result.append(' ' * indent)
     else:
-        return repr(obj)
+        result.append(repr(obj))
 
     return ''.join(result)
 

stdplugins/kbass_core.py

@@ -1,46 +0,0 @@
-"""
-Contains code used by other kbass_* plugins
-"""
-from uniborg import util
-
-
-def self_reply_cmd(borg, pattern):
-    def wrapper(function):
-        @borg.on(util.admin_cmd(pattern))
-        async def wrapped(event, *args, **kwargs):
-            await event.delete()
-            target = await util.get_target_message(borg, event)
-            if not target:
-                return
-            return await function(event, target, *args, **kwargs)
-        return wrapped
-    return wrapper
-
-
-def self_reply_selector(borg, cmd):
-    def wrapper(function):
-        @borg.on(util.admin_cmd(cmd + r"( [+-]?\d+)?$"))
-        async def wrapped(event, *args, **kwargs):
-            await event.delete()
-            reply = await event.get_reply_message()
-            if not reply:
-                return
-            num_offset = int(event.pattern_match.group(1) or 0)
-            reverse = num_offset > 0
-
-            targets = [reply] if reverse else []
-            targets.extend(await borg.get_messages(
-                await event.get_input_chat(),
-                limit=abs(num_offset),
-                offset_id=reply.id,
-                reverse=reverse
-            ))
-            if not reverse:
-                # reverse the list because we want things to always be in
-                # history order
-                targets.reverse()
-                targets.append(reply)
-
-            return await function(event, targets, num_offset, *args, **kwargs)
-        return wrapped
-    return wrapper

stdplugins/kbass_edit.py

@@ -1,43 +0,0 @@
-"""
-Reply to a message with .e to make a draft of it (with '\n.e' appended)
-Reply to your own message with <text>.e to edit the message to <text>
-if <text> is ".", the message is edited to empty/deleted
-"""
-import asyncio
-
-from telethon.errors import MessageEmptyError
-from telethon.tl.functions.messages import SaveDraftRequest
-from telethon.tl.functions.messages import EditMessageRequest
-
-from stdplugins.kbass_core import self_reply_cmd
-
-
-@self_reply_cmd(borg, r"^\.e$")
-async def on_edit_start(event, target):
-    await asyncio.sleep(3)  # tdesktop doesn't sync drafts when the window is active
-    await borg(SaveDraftRequest(
-        peer=await event.get_input_chat(),
-        message=(target.message or '.') + '\n.e',
-        entities=target.entities,
-        no_webpage=not target.media,
-        reply_to_msg_id=target.id
-    ))
-
-
-@self_reply_cmd(borg, r'(?ms)^(.+)\.e$')
-async def on_edit_end(event, target):
-    text = event.pattern_match.group(1)
-    message = event.message.message[:-2]
-    if message.strip() == '.':
-        message = ''
-    try:
-        await borg(EditMessageRequest(
-            peer=await event.get_input_chat(),
-            id=target.id,
-            no_webpage=not target.media,
-            message=message,
-            entities=event.message.entities
-        ))
-    except MessageEmptyError:
-        # Can't make text message empty, so delete it
-        await borg.delete_messages(chat, target)

stdplugins/kbass_file_to_photo.py

@@ -1,34 +0,0 @@
-"""
-Reply to a file with .f to send it as a photo
-"""
-from io import BytesIO
-
-from stdplugins.kbass_core import self_reply_cmd
-from telethon import types
-from telethon.errors import PhotoInvalidDimensionsError
-
-
-@self_reply_cmd(borg, r"^\.f$")
-async def on_file_to_photo(event, target):
-    try:
-        image = target.media.document
-    except AttributeError:
-        return
-    if not image.mime_type.startswith('image/'):
-        return  # This isn't an image
-    if image.mime_type == 'image/webp':
-        return  # Telegram doesn't let you directly send stickers as photos
-    if image.size > 10 * 1024 * 1024:
-        return  # We'd get PhotoSaveFileInvalidError otherwise
-
-    file = await borg.download_media(target, file=BytesIO())
-    file.seek(0)
-    img = await borg.upload_file(file)
-
-    try:
-        await event.respond(
-            reply_to=target,
-            file=types.InputMediaUploadedPhoto(img)
-        )
-    except PhotoInvalidDimensionsError:
-        return

stdplugins/kbass_save.py

@@ -1,15 +0,0 @@
-"""
-Reply to a message with .s <n> to forward <n> messages from that point to your
-saved messages (negative values of <n> go backwards in history)
-"""
-import asyncio
-
-from stdplugins.kbass_core import self_reply_selector
-
-
-@self_reply_selector(borg, r'\.s')
-async def on_save(event, targets, num_offset):
-    await borg.forward_messages('me', targets)
-    msg = await event.respond(f'Saved {abs(num_offset) + 1} messages!')
-    await asyncio.sleep(3)
-    await borg.delete_messages(msg.to_id, msg)

stdplugins/kbass_toggle_preview.py

@@ -1,25 +0,0 @@
-"""
-Reply to a message with .p <y/n> to toggle the webpage preview of a message
-"""
-from telethon.errors import MessageNotModifiedError
-from telethon.tl.functions.messages import EditMessageRequest
-
-from stdplugins.kbass_core import self_reply_cmd
-
-
-@self_reply_cmd(borg, r"^\.p(?: ?)([yn])?$")
-async def on_edit_preview(event, target):
-    preview = event.pattern_match.group(1) == 'y'
-    if not event.pattern_match.group(1):
-        preview = not bool(target.media)
-    try:
-        await borg(EditMessageRequest(
-            peer=await event.get_input_chat(),
-            id=target.id,
-            no_webpage=not preview,
-            message=target.message,
-            entities=target.entities
-        ))
-    except MessageNotModifiedError:
-        # There was no preview to modify
-        pass

stdplugins/kbass_yank.py

@@ -1,28 +0,0 @@
-"""
-Like save but makes a draft in your chat of all the messages concatenated with
-two newlines
-"""
-import html
-
-from telethon.tl.functions.messages import SaveDraftRequest
-from telethon.extensions import html as thtml
-
-from stdplugins.kbass_core import self_reply_selector
-
-
-def get_message_html(message):
-    if message.action:
-        return html.escape(str(message.action))
-    return thtml.unparse(message.message, message.entities)
-
-
-@self_reply_selector(borg, r'\.y')
-async def on_yank(event, targets, num_offset):
-    message = '\n\n'.join(get_message_html(target) for target in targets)
-    message, entities = thtml.parse(message)
-    await borg(SaveDraftRequest(
-        peer='me',
-        message=message,
-        entities=entities,
-        no_webpage=True,
-    ))

stdplugins/ninja.py

@@ -5,15 +5,21 @@
 import asyncio
 
 from telethon import events
-from telethon.tl.types import InputPeerSelf
 import telethon.utils
 
 from uniborg import util
 
 
+async def get_target_message(event):
+    if event.is_reply and (await event.get_reply_message()).from_id == borg.uid:
+        return await event.get_reply_message()
+    async for message in borg.iter_messages(
+            await event.get_input_chat(), limit=20):
+        if message.out:
+            return message
+
+
 async def await_read(chat, message):
-    if isinstance(chat, InputPeerSelf):
-        return
     chat = telethon.utils.get_peer_id(chat)
 
     async def read_filter(read_event):
@@ -37,7 +43,7 @@ async def delete(event):
         text = event.pattern_match.group(2)
         if not text:
             return
-    target = await util.get_target_message(borg, event)
+    target = await get_target_message(event)
     if target:
         chat = await event.get_input_chat()
         await await_read(chat, target)

stdplugins/sp_prog_rename.py

@@ -3,7 +3,6 @@ import re
 
 from telethon import events
 from telethon.tl.functions.channels import EditTitleRequest
-from telethon.errors.rpcerrorlist import ChatNotModifiedError
 
 MULTI_EDIT_TIMEOUT = 80
 REVERT_TIMEOUT = 2 * 60 * 60
@@ -27,12 +26,9 @@ async def edit_title(title):
     global prog_tech_channel
     if prog_tech_channel is None:
         prog_tech_channel = await borg.get_entity(CHANNEL_ID)
-    try:
-        await borg(EditTitleRequest(
-            channel=prog_tech_channel, title=title
-        ))
-    except ChatNotModifiedError:
-        pass  # Everything is ok
+    await borg(EditTitleRequest(
+        channel=prog_tech_channel, title=title
+    ))
 
 
 async def wait_for_delete(deleted_fut, timeout):

stdplugins/superblock.py

@@ -1,9 +1,5 @@
-# This Source Code Form is subject to the terms of the Mozilla Public
-# License, v. 2.0. If a copy of the MPL was not distributed with this
-# file, You can obtain one at http://mozilla.org/MPL/2.0/.
-
 import asyncio
-
+import time
 from telethon import events
 import telethon.tl.functions as tlf
 from telethon.tl.types import InputPeerChannel, UpdateUserBlocked
@@ -12,7 +8,6 @@ from telethon.tl.functions.contacts import GetBlockedRequest
 
 # How often to fetch the full list of blocked users
 REFETCH_TIME = 60
-
 blocked_user_ids = set()
 
 

uniborg/util.py

@@ -28,12 +28,3 @@ async def is_read(borg, entity, message, is_out=None):
     dialog = (await borg(GetPeerDialogsRequest([entity]))).dialogs[0]
     max_id = dialog.read_outbox_max_id if is_out else dialog.read_inbox_max_id
     return message_id <= max_id
-
-
-async def get_target_message(borg, event):
-    if event.is_reply and (await event.get_reply_message()).from_id == borg.uid:
-        return await event.get_reply_message()
-    async for message in borg.iter_messages(
-            await event.get_input_chat(), limit=20):
-        if message.out:
-            return message