438 lines
20 KiB
Python
438 lines
20 KiB
Python
import time
|
|
import logging, pprint
|
|
|
|
from xmpp_tg.utils import var_dump, display_tg_name, get_contact_jid, localtime
|
|
|
|
# modified by adnidor
|
|
|
|
class WrongNumberOfArgsError(Exception):
|
|
pass
|
|
|
|
class CommandHandler():
|
|
_on_connect = lambda: None
|
|
_helptext = "Available Commands:"
|
|
|
|
def _unknown_command_handler(self, *args, **kwargs):
|
|
return "Unknown command, for a list send !help"
|
|
|
|
def _min_args(self, num_args):
|
|
if len(self.arguments) < num_args:
|
|
raise WrongNumberOfArgsError("!{} needs at least {} arguments".format(self._command, num_args))
|
|
|
|
def __init__(self, msg):
|
|
self._command = msg["body"].split(" ")[0][1:]
|
|
self._handler = getattr(self, self._command, self._unknown_command_handler)
|
|
self.type = "groupchat" if msg["type"] == "groupchat" else "chat"
|
|
self.sender = msg["from"]
|
|
self.jid = msg["from"].bare
|
|
self.replyto = self.sender.full if self.type == "chat" else self.sender.bare
|
|
self.arguments = msg["body"].split(" ")[1:]
|
|
self.msg = msg
|
|
|
|
def _update(self, text):
|
|
xmpp.send_message(mto=self.replyto, mtype=self.type, mbody=text)
|
|
|
|
def debug(self, *args, **kwargs):
|
|
"""nolist Show debug info"""
|
|
return pprint.pformat(self.__dict__)
|
|
|
|
def help(self, *args, **kwargs):
|
|
"""List available commands"""
|
|
#taken from https://www.python.org/dev/peps/pep-0257/#handling-docstring-indentation
|
|
def trim(docstring):
|
|
if not docstring:
|
|
return ''
|
|
# Convert tabs to spaces (following the normal Python rules)
|
|
# and split into a list of lines:
|
|
lines = docstring.expandtabs().splitlines()
|
|
# Determine minimum indentation (first line doesn't count):
|
|
indent = 500
|
|
for line in lines[1:]:
|
|
stripped = line.lstrip()
|
|
if stripped:
|
|
indent = min(indent, len(line) - len(stripped))
|
|
# Remove indentation (first line is special):
|
|
trimmed = [lines[0].strip()]
|
|
if indent < 500:
|
|
for line in lines[1:]:
|
|
trimmed.append(line[indent:].rstrip())
|
|
# Strip off trailing and leading blank lines:
|
|
while trimmed and not trimmed[-1]:
|
|
trimmed.pop()
|
|
while trimmed and not trimmed[0]:
|
|
trimmed.pop(0)
|
|
# Return a single string:
|
|
return '\n'.join(trimmed)
|
|
|
|
if len(self.arguments) == 0:
|
|
methods = [func for func in dir(self) if not func.startswith("_") and callable(getattr(self, func)) and not isinstance(func, Exception)]
|
|
reply = self._helptext
|
|
for method in methods:
|
|
docstring = getattr(self, method).__doc__
|
|
if docstring is None:
|
|
docstring = "No description available"
|
|
if docstring.startswith("nolist "):
|
|
continue
|
|
reply += "\n!{} - {}".format(method, docstring.split("\n")[0])
|
|
return reply
|
|
else:
|
|
method = getattr(self,self.arguments[0])
|
|
reply = trim(method.__doc__)
|
|
return reply
|
|
|
|
class GateCommandHandler(CommandHandler):
|
|
_helptext = "Available Gateway commands:"
|
|
|
|
def configure(self, gate):
|
|
"""Get config/set config options"""
|
|
config_exclude = ['jid', 'tg_phone']
|
|
|
|
option = self.arguments[0] if len(self.arguments) >= 1 else None
|
|
value = self.arguments[1] if len(self.arguments) == 2 else None
|
|
|
|
if value is not None and option not in config_exclude:
|
|
gate.db_connection.execute("update accounts set {} = ? where jid = ?".format(option), (value, self.jid,))
|
|
gate.accounts[self.jid] = gate.db_connection.execute("SELECT * FROM accounts where jid = ?", (self.jid,)).fetchone()
|
|
|
|
message = "=== Your current configuration ===\n\n"
|
|
for param, value in gate.accounts[self.jid].items():
|
|
message = message + "<%s>: %s" % (param, value) + "\n"
|
|
message = message + "\nTo modify some option, please, send !configure param value"
|
|
return message
|
|
|
|
def login(self, gate): #1 arg
|
|
"""Initiates Telegram session
|
|
|
|
Usage: !login <phone number in international format>"""
|
|
self._min_args(1)
|
|
|
|
phone_no = self.arguments[0]
|
|
|
|
self._update("Please wait...")
|
|
gate.spawn_tg_client(self.jid, phone_no)
|
|
|
|
if gate.tg_connections[self.jid].is_user_authorized():
|
|
gate.send_presence(pto=jid, pfrom=gate.boundjid.bare, ptype='online', pstatus='connected')
|
|
return "You are already authenticated in Telegram"
|
|
else:
|
|
# remove old sessions for this JID #
|
|
gate.db_connection.execute("DELETE from accounts where jid = ?", (self.jid,))
|
|
gate.tg_connections[self.jid].send_code_request(phone_no)
|
|
gate._update('Gate is connected. Telegram should send SMS message to you.')
|
|
return 'Please enter one-time code via !code 12345.'
|
|
|
|
def code(self, gate):
|
|
"""nolist Enter confirmation code"""
|
|
self._min_args(1)
|
|
|
|
code = self.arguments[0]
|
|
jid = self.jid
|
|
|
|
if not gate.tg_connections[jid].is_user_authorized():
|
|
try:
|
|
self._update('Trying authenticate...')
|
|
gate.tg_connections[jid].sign_in(gate.tg_phones[jid], code)
|
|
except SessionPasswordNeededError:
|
|
self._update('Two-factor authentication detected.')
|
|
self._update('Please enter your password via !password abc123.')
|
|
return
|
|
|
|
if gate.tg_connections[jid].is_user_authorized():
|
|
gate.send_presence(pto=jid, pfrom=gate.boundjid.bare, ptype='online', pstatus='connected')
|
|
self._update('Authentication successful. Initiating Telegram...')
|
|
gate.db_connection.execute("INSERT INTO accounts(jid, tg_phone) VALUES(?, ?)", (jid, gate.tg_phones[jid],))
|
|
gate.accounts[jid] = gate.db_connection.execute("SELECT * FROM accounts where jid = ?", (jid,)).fetchone()
|
|
gate.init_tg(jid)
|
|
|
|
else:
|
|
return 'Authentication failed.'
|
|
else:
|
|
return 'You are already authenticated. Please use !logout before new login.'
|
|
|
|
def password(self, gate):
|
|
"""nolist Enter password"""
|
|
self._min_args(1)
|
|
|
|
password = self.arguments[1]
|
|
jid = gate.jidhndl
|
|
|
|
if not gate.tg_connections[jid].is_user_authorized():
|
|
self._update('Checking password...')
|
|
gate.tg_connections[jid].sign_in(password=password)
|
|
|
|
if gate.tg_connections[jid].is_user_authorized():
|
|
gate.send_presence(pto=jid, pfrom=gate.boundjid.bare, ptype='online', pstatus='connected')
|
|
self._update('Authentication successful. Initiating Telegram...')
|
|
gate.db_connection.execute("INSERT INTO accounts(jid, tg_phone) VALUES(?, ?)", (jid, gate.tg_phones[jid],))
|
|
gate.accounts[jid] = gate.db_connection.execute("SELECT * FROM accounts where jid = ?", (jid,)).fetchone()
|
|
gate.init_tg(jid)
|
|
|
|
else:
|
|
return 'Authentication failed.'
|
|
else:
|
|
return 'You are already authenticated. Please use !logout before new login.'
|
|
|
|
def reload_dialogs(self, gate):
|
|
"""Reload Telegram dialogs"""
|
|
if not gate.tg_connections[self.jid].is_user_authorized():
|
|
return "Error"
|
|
|
|
gate.tg_process_dialogs(self.jid)
|
|
return "Dialogs reloadad"
|
|
|
|
def logout(self, gate):
|
|
"""Deletes current Telegram session at Gateway"""
|
|
gate.tg_connections[self.jid].log_out()
|
|
gate.db_connection.execute("DELETE FROM accounts WHERE jid = ?", (self.jid,))
|
|
return 'Your Telegram session was deleted'
|
|
|
|
def add(self, gate): #1 arg
|
|
"""Add contact by nickname or t.me link"""
|
|
self._min_args(1)
|
|
|
|
argument = self.arguments[0]
|
|
jid = self.jid
|
|
|
|
result = gate.tg_connections[jid].get_entity(argument)
|
|
if type(result) == User:
|
|
tg_peer = InputPeerUser( result.id, result.access_hash )
|
|
result = gate.tg_connections[jid].invoke(SendMessageRequest(tg_peer, 'Hello! I just want to add you in my contact list.', generate_random_long()))
|
|
elif type(result) == Channel:
|
|
tg_peer = InputPeerChannel( result.id, result.access_hash )
|
|
gate.tg_connections[jid].invoke(JoinChannelRequest(InputPeerChannel(result.id, result.access_hash)))
|
|
else:
|
|
return 'Sorry, nothing found.'
|
|
|
|
gate.tg_process_dialogs(jid)
|
|
|
|
def join(self, gate): #1 arg
|
|
"""Join conference via invite link"""
|
|
self._min_args(1)
|
|
|
|
argument = self.arguments[0]
|
|
jid = self.jid
|
|
|
|
link = argument.split('/') # https://t.me/joinchat/HrCmckx_SkMbSGFLhXCvSg
|
|
gate.tg_connections[jid].invoke(ImportChatInviteRequest(link[4]))
|
|
time.sleep(1)
|
|
gate.tg_process_dialogs(jid)
|
|
|
|
def group(self, gate): #2 args
|
|
"""Creat a Group"""
|
|
self._min_args(2)
|
|
# group name? #
|
|
groupname = parsed[1]
|
|
|
|
# group users? #
|
|
groupuser = gate.tg_connections[jid].get_entity(parsed[2])
|
|
|
|
# we re ready to make group
|
|
gate.tg_connections[jid].invoke(CreateChatRequest([groupuser], groupname))
|
|
gate.tg_process_dialogs(jid)
|
|
|
|
|
|
def supergroup(self, gate): #1 arg
|
|
"""Create a channel"""
|
|
self._min_args(1)
|
|
|
|
groupname = self.arguments[0]
|
|
gate.tg_connections[self.jid].invoke(CreateChannelRequest(groupname, groupname, megagroup = True))
|
|
gate.tg_process_dialogs(self.jid)
|
|
|
|
channel = supergroup
|
|
|
|
def username(self, gate): #1 arg
|
|
"""Change your Telegram nickname"""
|
|
self._min_args(1)
|
|
|
|
username = self.arguments[0]
|
|
|
|
gate.tg_connections[self.jid].invoke(UpdateUsernameRequest(username))
|
|
|
|
def name(self, gate): #1 or 2 args
|
|
"""Change your Telegram display name"""
|
|
self._min_args(1)
|
|
|
|
firstname = self.arguments[0]
|
|
lastname = self.arguments[1] if len(self.arguments) > 1 else None
|
|
|
|
gate.tg_connections[self.jid].invoke(UpdateProfileRequest(first_name = firstname, last_name = lastname))
|
|
|
|
def about(self, gate): #>0 args
|
|
"""Change your Telegram 'about' text"""
|
|
self._min_args(1)
|
|
|
|
about = self.msg['body'][7:]
|
|
gate.tg_connections[self.jid].invoke(UpdateProfileRequest(about = about))
|
|
|
|
def import_contact(self, gate): #2 args
|
|
"""Add contact by phone number"""
|
|
self._min_args(2)
|
|
|
|
phone = self.arguments[0]
|
|
firstname = self.arguments[1]
|
|
lastname = self.arguments[2] if len(self.arguments) > 2 else None
|
|
|
|
contact = InputPhoneContact(client_id=generate_random_long(), phone=phone, first_name=firstname, last_name=lastname)
|
|
gate.tg_connections[self.jid].invoke(ImportContactsRequest([contact]))
|
|
gate.tg_process_dialogs(jid)
|
|
|
|
def roster(self, gate):
|
|
"""Send Telegram contact list back, with JIDs"""
|
|
response = "Telegram chats:\n"
|
|
for jid,tid in gate.contact_list[self.jid].items():
|
|
response += "{}: {}\n".format(tid, jid)
|
|
return response
|
|
|
|
def oldhelp(self, gate):
|
|
"""nolist Old !help message"""
|
|
return ('=== Available gateway commands ===:\n\n'
|
|
'!help - Displays this text\n'
|
|
'!login +123456789 - Initiates Telegram session\n'
|
|
'!code 12345 - Entering one-time code during auth\n'
|
|
'!password abc123 - Entering password during two-factor auth\n'
|
|
'!configure - Configure transport settings\n'
|
|
#'!list_sessions - List all created sessions at Telegram servers\n'
|
|
#'!delete_session 123 - Delete session\n'
|
|
'!logout - Deletes current Telegram session at gate\n'
|
|
'!reload_dialogs - Reloads dialogs list from Telegram\n\n'
|
|
|
|
'!add - Find and add Telegram contact. Any formats accepted (nickname or t.me link)\n\n'
|
|
|
|
'!join - Join Telegram conference via invite link \n\n'
|
|
|
|
'!import phone firstname lastname - Add Telegram contact with phone number \n\n'
|
|
|
|
'!group GroupName @InviteContact - Create a normal group\n'
|
|
'!supergroup SupergroupName - Create a supergroup\n'
|
|
'!channel ChannelName - Create a channel\n\n'
|
|
|
|
'!name first last - Change your name in Telegram\n'
|
|
'!about text - Change about text in Telegram\n'
|
|
'!username - Changes your @username in Telegram\n\n'
|
|
|
|
'!roster - Lists yout TG roster\n')
|
|
|
|
class ChatCommandHandler(CommandHandler):
|
|
_helptext = "Available Userchat commands:"
|
|
|
|
def __init__(self, msg):
|
|
super(ChatCommandHandler, self).__init__(msg)
|
|
if self._command.startswith("s/"):
|
|
self._handler = self._replace
|
|
self.tg_id = int(msg['to'].node[1:])
|
|
|
|
def block(self, gate):
|
|
"""Block this user"""
|
|
nickname = display_tg_name(gate.tg_dialogs[self.jid]['users'][self.tg_id])
|
|
gate.tg_connections[self.jid].invoke(BlockRequest(InputPeerUser(self.tg_id, gate.tg_dialogs[self.jid]['users'][self.tg_id].access_hash)))
|
|
gate.gate_reply_message(iq, 'User %s blacklisted!' % nickname)
|
|
|
|
def unblock(self, gate):
|
|
"""Unblock this user"""
|
|
nickname = display_tg_name(gate.tg_dialogs[self.jid]['users'][self.tg_id])
|
|
gate.tg_connections[self.jid].invoke(UnblockRequest(InputPeerUser(self.tg_id, gate.tg_dialogs[self.jid]['users'][self.tg_id].access_hash)))
|
|
gate.gate_reply_message(iq, 'User %s unblacklisted!' % nickname)
|
|
|
|
def remove(self, gate):
|
|
"""Remove User from Telegram contact list"""
|
|
peer = InputPeerUser(self.tg_id, gate.tg_dialogs[self.jid]['users'][self.tg_id].access_hash)
|
|
c_jid = get_contact_jid(gate.tg_dialogs[self.jid]['users'][self.tg_id], gate.boundjid.bare)
|
|
|
|
gate.tg_connections[self.jid].invoke(DeleteContactRequest(peer))
|
|
gate.tg_connections[self.jid].invoke(DeleteHistoryRequest(peer, max_id = 0, just_clear = None))
|
|
|
|
gate.send_presence(pto = self.jid, pfrom = c_jid, ptype ='unavailable')
|
|
gate.send_presence(pto = self.jid, pfrom = c_jid, ptype ='unsubscribed')
|
|
gate.send_presence(pto = self.jid, pfrom = c_jid, ptype ='unsubscribe')
|
|
|
|
def _replace(self, gate):
|
|
peer = InputPeerUser(hdnl.tg_id, gate.tg_dialogs[self.jid]['users'][self.tg_id].access_hash)
|
|
|
|
msg_id, edited = gate.edit_message(hdnl.jid, hdnl.tg_id, self.msg['body'])
|
|
if not edited: return
|
|
|
|
# and send it
|
|
if edited != '' and edited != ' ':
|
|
gate.tg_dialogs[self.jid]['messages'][self.tg_id]["body"] = edited
|
|
gate.tg_connections[self.jid].invoke(EditMessageRequest(peer, msg_id, message = edited))
|
|
else:
|
|
del(gate.tg_dialogs[self.jid]['messages'][self.tg_id])
|
|
gate.tg_connections[self.jid].invoke(DeleteMessagesRequest([msg_id], revoke = True))
|
|
|
|
class GroupchatCommandHandler(CommandHandler):
|
|
_helptext = "Available Groupchat commands:"
|
|
|
|
def __init__(self, msg):
|
|
super(GroupchatCommandHandler, self).__init__(msg)
|
|
if self._command.startswith("s/"):
|
|
self._handler = self._replace
|
|
self.tg_id = int(self.msg['to'].node[1:])
|
|
|
|
def leave(self, gate):
|
|
"""Leave group or channel"""
|
|
if self.tg_id in gate.tg_dialogs[self.jid]['supergroups']:
|
|
peer = InputPeerChannel(self.tg_id, gate.tg_dialogs[self.jid]['supergroups'][self.tg_id].access_hash)
|
|
gate.tg_connections[self.jid].invoke(LeaveChannelRequest(peer))
|
|
gate.tg_connections[self.jid].invoke(DeleteHistoryRequest(peer, max_id = 0, just_clear = None))
|
|
c_jid = get_contact_jid(gate.tg_dialogs[self.jid]['supergroups'][self.tg_id], gate.boundjid.bare)
|
|
|
|
gate.send_presence(pto =self.jid, pfrom = c_jid, ptype ='unavailable')
|
|
gate.send_presence(pto =self.jid, pfrom = c_jid, ptype ='unsubscribed')
|
|
gate.send_presence(pto =self.jid, pfrom = c_jid, ptype ='unsubscribe')
|
|
|
|
if self.tg_id in gate.tg_dialogs[self.jid]['groups']:
|
|
gate.tg_connections[self.jid].invoke(DeleteChatUserRequest(self.tg_id, gate.tg_connections[self.jid].me))
|
|
gate.tg_connections[self.jid].invoke(DeleteHistoryRequest(InputPeerChat(self.tg_id), max_id = 0, just_clear = None))
|
|
c_jid = get_contact_jid(gate.tg_dialogs[self.jid]['groups'][self.tg_id], gate.boundjid.bare)
|
|
|
|
gate.send_presence(pto =self.jid, pfrom = c_jid, ptype ='unavailable')
|
|
gate.send_presence(pto =self.jid, pfrom = c_jid, ptype ='unsubscribed')
|
|
gate.send_presence(pto =self.jid, pfrom = c_jid, ptype ='unsubscribe')
|
|
|
|
def invite(self, gate):
|
|
"""Invite user to group"""
|
|
if self.tg_id in gate.tg_dialogs[self.jid]['supergroups']:
|
|
invited_user = gate.tg_connections[[self.jid]].get_entity(parsed[1])
|
|
if type(invited_user) == User:
|
|
gate.tg_connections[[self.jid]].invoke(EditBannedRequest(InputPeerChannel(self.tg_id, gate.tg_dialogs[[self.jid]]['supergroups'][
|
|
self.tg_id].access_hash), invited_user, ChannelBannedRights(until_date=None, view_messages=False)))
|
|
gate.tg_connections[[self.jid]].invoke(InviteToChannelRequest(InputPeerChannel(self.tg_id, gate.tg_dialogs[[self.jid]]['supergroups'][
|
|
self.tg_id].access_hash), [invited_user]))
|
|
|
|
if self.tg_id in gate.tg_dialogs[[self.jid]]['groups']:
|
|
invited_user = gate.tg_connections[[self.jid]].get_entity(parsed[1])
|
|
if type(invited_user) == User:
|
|
gate.tg_connections[[self.jid]].invoke(AddChatUserRequest(self.tg_id, invited_user, 0))
|
|
|
|
def kick(hndl, self):
|
|
"""Kick user"""
|
|
tg_id = int(hndl.msg['to'].node[1:])
|
|
if tg_id in self.tg_dialogs[hndl.jid]['supergroups']:
|
|
kicked_user = self.tg_connections[hndl.jid].get_entity(parsed[1])
|
|
if type(kicked_user) == User:
|
|
self.tg_connections[hndl.jid].invoke(EditBannedRequest(InputPeerChannel(tg_id, self.tg_dialogs[hndl.jid]['supergroups'][tg_id].access_hash), kicked_user, ChannelBannedRights(until_date=None, view_messages=True)))
|
|
if tg_id in self.tg_dialogs[hndl.jid]['groups']:
|
|
kicked_user = self.tg_connections[hndl.jid].get_entity(parsed[1])
|
|
if type(kicked_user) == User:
|
|
self.tg_connections[hndl.jid].invoke(DeleteChatUserRequest(tg_id, kicked_user))
|
|
|
|
def _replace(hndl, self):
|
|
tg_id = int(hndl.msg['to'].node[1:])
|
|
peer = InputPeerChannel(tg_id, self.tg_dialogs[hndl.jid]['supergroups'][tg_id].access_hash) if tg_id in self.tg_dialogs[
|
|
hndl.jid]['supergroups'] else InputPeerChat(tg_id)
|
|
|
|
msg_id, edited = self.edit_message(hndl.jid, tg_id, iq['body'])
|
|
if not edited: return
|
|
|
|
# and send it
|
|
if edited != '' and edited != ' ':
|
|
self.tg_dialogs[hndl.jid]['messages'][tg_id]["body"] = edited
|
|
self.tg_connections[hndl.jid].invoke(EditMessageRequest(peer, msg_id, message = edited))
|
|
else:
|
|
del(self.tg_dialogs[hndl.jid]['messages'][tg_id])
|
|
if isinstance(peer, InputPeerChannel):
|
|
self.tg_connections[hndl.jid].invoke(DeleteMessagesChannel(peer, [msg_id]))
|
|
else:
|
|
self.tg_connections[hndl.jid].invoke(DeleteMessagesRequest([msg_id], revoke = True))
|