1 # irgramd: IRC-Telegram gateway 2 # exclam.py: IRC exclamation command handlers 3 # 4 # Copyright (c) 2023, 2024 E. Bosch <presidev@AT@gmail.com> 5 # 6 # Use of this source code is governed by a MIT style license that 7 # can be found in the LICENSE file included in this project. 8 9 import os 10 from telethon.errors.rpcerrorlist import MessageNotModifiedError, MessageAuthorRequiredError 11 12 from utils import command, HELP 13 14 class exclam(command): 15 def __init__(self, telegram): 16 self.commands = \ 17 { # Command Handler Arguments Min Max Maxsplit 18 '!re': (self.handle_command_re, 2, 2, 2), 19 '!ed': (self.handle_command_ed, 2, 2, 2), 20 '!del': (self.handle_command_del, 1, 1, -1), 21 '!fwd': (self.handle_command_fwd, 2, 2, -1), 22 '!upl': (self.handle_command_upl, 1, 2, 2), 23 } 24 self.tg = telegram 25 self.irc = telegram.irc 26 self.tmp_ircnick = None 27 self.tmp_telegram_id = None 28 self.tmp_tg_msg = None 29 30 async def command(self, message, telegram_id, user): 31 self.tmp_telegram_id = telegram_id 32 res = await self.parse_command(message, nick=None) 33 if isinstance(res, tuple): 34 await self.irc.send_msg(self.irc.service_user, None, res[0], user) 35 res = False 36 return res, self.tmp_tg_msg 37 38 async def check_msg(self, cid): 39 id = self.tg.mid.id_to_num_offset(self.tmp_telegram_id, cid) 40 if id is None: 41 chk_msg = None 42 else: 43 chk_msg = await self.tg.telegram_client.get_messages(entity=self.tmp_telegram_id, ids=id) 44 return id, chk_msg 45 46 async def handle_command_re(self, cid=None, msg=None, help=None): 47 if not help: 48 id, chk_msg = await self.check_msg(cid) 49 if chk_msg is not None: 50 self.tmp_tg_msg = await self.tg.telegram_client.send_message(self.tmp_telegram_id, msg, reply_to=id) 51 reply = True 52 else: 53 reply = ('!re: Unknown message to reply',) 54 else: # HELP.brief or HELP.desc (first line) 55 reply = (' !re Reply to a message',) 56 if help == HELP.desc: # rest of HELP.desc 57 reply += \ 58 ( 59 ' !re <compact_id> <message>', 60 'Reply with <message> to a message with <compact_id> on current', 61 'channel/chat.' 62 ) 63 return reply 64 65 async def handle_command_ed(self, cid=None, new_msg=None, help=None): 66 if not help: 67 id, ed_msg = await self.check_msg(cid) 68 if ed_msg is not None: 69 try: 70 self.tmp_tg_msg = await self.tg.telegram_client.edit_message(ed_msg, new_msg) 71 except MessageNotModifiedError: 72 self.tmp_tg_msg = ed_msg 73 reply = True 74 except MessageAuthorRequiredError: 75 reply = ('!ed: Not the author of the message to edit',) 76 else: 77 reply = True 78 else: 79 reply = ('Unknown message to edit',) 80 else: # HELP.brief or HELP.desc (first line) 81 reply = (' !ed Edit a message',) 82 if help == HELP.desc: # rest of HELP.desc 83 reply += \ 84 ( 85 ' !ed <compact_id> <new_message>', 86 'Edit a message with <compact_id> on current channel/chat,', 87 '<new_message> replaces the current message.' 88 ) 89 return reply 90 91 async def handle_command_del(self, cid=None, help=None): 92 if not help: 93 id, del_msg = await self.check_msg(cid) 94 if del_msg is not None: 95 deleted = await self.tg.telegram_client.delete_messages(self.tmp_telegram_id, del_msg) 96 if deleted[0].pts_count == 0: 97 reply = ('!del: Not possible to delete',) 98 else: 99 self.tmp_tg_msg = None 100 reply = None 101 else: 102 reply = ('Unknown message to delete',) 103 else: # HELP.brief or HELP.desc (first line) 104 reply = (' !del Delete a message',) 105 if help == HELP.desc: # rest of HELP.desc 106 reply += \ 107 ( 108 ' !del <compact_id>', 109 'Delete a message with <compact_id> on current channel/chat' 110 ) 111 return reply 112 113 async def handle_command_fwd(self, cid=None, chat=None, help=None): 114 if not help: 115 id, chk_msg = await self.check_msg(cid) 116 if chk_msg is not None: 117 async def send_fwd(tgt_ent, id): 118 from_ent = await self.tg.telegram_client.get_entity(self.tmp_telegram_id) 119 self.tmp_tg_msg = await self.tg.telegram_client.forward_messages(tgt_ent, id, from_ent) 120 return self.tmp_tg_msg 121 122 tgt = chat.lower() 123 if tgt in self.irc.iid_to_tid: 124 tgt_ent = await self.tg.telegram_client.get_entity(self.irc.iid_to_tid[tgt]) 125 msg = await send_fwd(tgt_ent, id) 126 # echo fwded message 127 await self.tg.handle_telegram_message(event=None, message=msg) 128 reply = True 129 elif tgt in (u.irc_nick.lower() for u in self.irc.users.values() if u.stream): 130 tgt_ent = await self.tg.telegram_client.get_me() 131 await send_fwd(tgt_ent, id) 132 reply = True 133 else: 134 reply = ('!fwd: Unknown chat to forward',) 135 else: 136 reply = ('Unknown message to forward',) 137 else: # HELP.brief or HELP.desc (first line) 138 reply = (' !fwd Forward a message',) 139 if help == HELP.desc: # rest of HELP.desc 140 reply += \ 141 ( 142 ' !fwd <compact_id> <chat>', 143 'Forward a message with <compact_id> to <chat> channel/chat.' 144 ) 145 return reply 146 147 async def handle_command_upl(self, file=None, caption=None, help=None): 148 if not help: 149 try: 150 if file[:8] == 'https://' or file[:7] == 'http://': 151 file_path = file 152 else: 153 file_path = os.path.join(self.tg.telegram_upload_dir, file) 154 self.tmp_tg_msg = await self.tg.telegram_client.send_file(self.tmp_telegram_id, file_path, caption=caption) 155 reply = True 156 except: 157 reply = ('!upl: Error uploading',) 158 else: # HELP.brief or HELP.desc (first line) 159 reply = (' !upl Upload a file to current channel/chat',) 160 if help == HELP.desc: # rest of HELP.desc 161 reply += \ 162 ( 163 ' !upl <file name/URL> [<optional caption>]', 164 'Upload the file referenced by <file name/URL> to current', 165 'channel/chat, the file must be present in "upload"', 166 'irgramd local directory or be an external HTTP/HTTPS URL.' 167 ) 168 return reply