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.tl.functions.messages import SendReactionRequest 11 from telethon import types as tgty 12 from telethon.errors.rpcerrorlist import MessageNotModifiedError, MessageAuthorRequiredError, ReactionInvalidError 13 14 from utils import command, HELP 15 from emoji2emoticon import emo_inv 16 17 class exclam(command): 18 def __init__(self, telegram): 19 self.commands = \ 20 { # Command Handler Arguments Min Max Maxsplit 21 '!del': (self.handle_command_del, 1, 1, -1), 22 '!ed': (self.handle_command_ed, 2, 2, 2), 23 '!fwd': (self.handle_command_fwd, 2, 2, -1), 24 '!re': (self.handle_command_re, 2, 2, 2), 25 '!react': (self.handle_command_react, 2, 2, -1), 26 '!reupl': (self.handle_command_reupl, 2, 3, 3), 27 '!upl': (self.handle_command_upl, 1, 2, 2), 28 } 29 self.tg = telegram 30 self.irc = telegram.irc 31 self.tmp_ircnick = None 32 self.tmp_telegram_id = None 33 self.tmp_tg_msg = None 34 35 async def command(self, message, telegram_id, user): 36 self.tmp_telegram_id = telegram_id 37 res = await self.parse_command(message, nick=None) 38 if isinstance(res, tuple): 39 await self.irc.send_msg(self.irc.service_user, None, res[0], user) 40 res = False 41 return res, self.tmp_tg_msg 42 43 async def check_msg(self, cid): 44 id = self.tg.mid.id_to_num_offset(self.tmp_telegram_id, cid) 45 if id is None or id < -2147483648 or id > 2147483647: 46 chk_msg = None 47 else: 48 chk_msg = await self.tg.telegram_client.get_messages(entity=self.tmp_telegram_id, ids=id) 49 return id, chk_msg 50 51 async def handle_command_re(self, cid=None, msg=None, help=None): 52 if not help: 53 id, chk_msg = await self.check_msg(cid) 54 if chk_msg is not None: 55 self.tmp_tg_msg = await self.tg.telegram_client.send_message(self.tmp_telegram_id, msg, reply_to=id) 56 reply = True 57 else: 58 reply = ('!re: Unknown message to reply',) 59 else: # HELP.brief or HELP.desc (first line) 60 reply = (' !re Reply to a message',) 61 if help == HELP.desc: # rest of HELP.desc 62 reply += \ 63 ( 64 ' !re <compact_id> <message>', 65 'Reply with <message> to a message with <compact_id> on current', 66 'channel/chat.', 67 ) 68 return reply 69 70 async def handle_command_ed(self, cid=None, new_msg=None, help=None): 71 if not help: 72 id, ed_msg = await self.check_msg(cid) 73 if ed_msg is not None: 74 try: 75 self.tmp_tg_msg = await self.tg.telegram_client.edit_message(ed_msg, new_msg) 76 except MessageNotModifiedError: 77 self.tmp_tg_msg = ed_msg 78 reply = True 79 except MessageAuthorRequiredError: 80 reply = ('!ed: Not the author of the message to edit',) 81 else: 82 reply = True 83 else: 84 reply = ('Unknown message to edit',) 85 else: # HELP.brief or HELP.desc (first line) 86 reply = (' !ed Edit a message',) 87 if help == HELP.desc: # rest of HELP.desc 88 reply += \ 89 ( 90 ' !ed <compact_id> <new_message>', 91 'Edit a message with <compact_id> on current channel/chat,', 92 '<new_message> replaces the current message.', 93 ) 94 return reply 95 96 async def handle_command_del(self, cid=None, help=None): 97 if not help: 98 id, del_msg = await self.check_msg(cid) 99 if del_msg is not None: 100 deleted = await self.tg.telegram_client.delete_messages(self.tmp_telegram_id, del_msg) 101 if deleted[0].pts_count == 0: 102 reply = ('!del: Not possible to delete',) 103 else: 104 self.tmp_tg_msg = None 105 reply = None 106 else: 107 reply = ('Unknown message to delete',) 108 else: # HELP.brief or HELP.desc (first line) 109 reply = (' !del Delete a message',) 110 if help == HELP.desc: # rest of HELP.desc 111 reply += \ 112 ( 113 ' !del <compact_id>', 114 'Delete a message with <compact_id> on current channel/chat' 115 ) 116 return reply 117 118 async def handle_command_fwd(self, cid=None, chat=None, help=None): 119 if not help: 120 id, chk_msg = await self.check_msg(cid) 121 if chk_msg is not None: 122 async def send_fwd(tgt_ent, id): 123 from_ent = await self.tg.telegram_client.get_entity(self.tmp_telegram_id) 124 self.tmp_tg_msg = await self.tg.telegram_client.forward_messages(tgt_ent, id, from_ent) 125 return self.tmp_tg_msg 126 127 tgt = chat.lower() 128 if tgt in self.irc.iid_to_tid: 129 tgt_ent = await self.tg.telegram_client.get_entity(self.irc.iid_to_tid[tgt]) 130 msg = await send_fwd(tgt_ent, id) 131 # echo fwded message 132 await self.tg.handle_telegram_message(event=None, message=msg) 133 reply = True 134 elif tgt in (u.irc_nick.lower() for u in self.irc.users.values() if u.stream): 135 tgt_ent = await self.tg.telegram_client.get_me() 136 await send_fwd(tgt_ent, id) 137 reply = True 138 else: 139 reply = ('!fwd: Unknown chat to forward',) 140 else: 141 reply = ('Unknown message to forward',) 142 else: # HELP.brief or HELP.desc (first line) 143 reply = (' !fwd Forward a message',) 144 if help == HELP.desc: # rest of HELP.desc 145 reply += \ 146 ( 147 ' !fwd <compact_id> <chat>', 148 'Forward a message with <compact_id> to <chat> channel/chat.' 149 ) 150 return reply 151 152 async def handle_command_upl(self, file=None, caption=None, help=None, re_id=None): 153 if not help: 154 try: 155 if file[:8] == 'https://' or file[:7] == 'http://': 156 file_path = file 157 else: 158 file_path = os.path.join(self.tg.telegram_upload_dir, file) 159 self.tmp_tg_msg = await self.tg.telegram_client.send_file(self.tmp_telegram_id, file_path, caption=caption, reply_to=re_id) 160 reply = True 161 except: 162 cmd = '!reupl' if re_id else '!upl' 163 reply = ('{}: Error uploading'.format(cmd),) 164 else: # HELP.brief or HELP.desc (first line) 165 reply = (' !upl Upload a file to current channel/chat',) 166 if help == HELP.desc: # rest of HELP.desc 167 reply += \ 168 ( 169 ' !upl <file name/URL> [<optional caption>]', 170 'Upload the file referenced by <file name/URL> to current', 171 'channel/chat, the file must be present in "upload"', 172 'irgramd local directory or be an external HTTP/HTTPS URL.', 173 ) 174 return reply 175 176 async def handle_command_reupl(self, cid=None, file=None, caption=None, help=None): 177 if not help: 178 id, chk_msg = await self.check_msg(cid) 179 if chk_msg is not None: 180 reply = await self.handle_command_upl(file, caption, re_id=id) 181 else: 182 reply = ('!reupl: Unknown message to reply',) 183 else: # HELP.brief or HELP.desc (first line) 184 reply = (' !reupl Reply to a message with an upload',) 185 if help == HELP.desc: # rest of HELP.desc 186 reply += \ 187 ( 188 ' !reupl <compact_id> <file name/URL> [<optional caption>]', 189 'Reply with the upload of <file name/URL> to a message with', 190 '<compact_id> on current channel/chat. The file must be', 191 'present in "upload" irgramd local directory or be an external', 192 'HTTP/HTTPS URL.', 193 ) 194 return reply 195 196 async def handle_command_react(self, cid=None, act=None, help=None): 197 if not help: 198 id, chk_msg = await self.check_msg(cid) 199 if chk_msg is not None: 200 if act in emo_inv: 201 utf8_emo = emo_inv[act] 202 reaction = [ tgty.ReactionEmoji(emoticon=utf8_emo) ] if utf8_emo else None 203 try: 204 update = await self.tg.telegram_client(SendReactionRequest(self.tmp_telegram_id, id, reaction=reaction)) 205 except ReactionInvalidError: 206 reply = ('!react: Reaction not allowed',) 207 else: 208 self.tmp_tg_msg = update.updates[0].message 209 reply = True 210 else: 211 reply = ('!react: Unknown reaction',) 212 else: 213 reply = ('!react: Unknown message to react',) 214 else: # HELP.brief or HELP.desc (first line) 215 reply = (' !react React to a message',) 216 if help == HELP.desc: # rest of HELP.desc 217 reply += \ 218 ( 219 ' !react <compact_id> <emoticon reaction>|-', 220 'React with <emoticon reaction> to a message with <compact_id>,', 221 'irgramd will translate emoticon to closest emoji.', 222 'Use - to remove a previous reaction.', 223 ) 224 return reply