yannik/tg4xmpp
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

rewrite process_chat_user_command

Browse Source
This commit is contained in:
adnidor
2019-02-27 00:55:52 +01:00
parent 4f07b24c6d
commit 8bdad9d0f0
1 changed files with 144 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files

253
xmpp_tg/xmpp.py
View File

@@ -262,75 +262,6 @@ class XMPPTelegram(ComponentXMPP):
self.send_presence(pto=jid, pfrom=self.boundjid.bare, ptype='unavailable')
sys.exit(0)
class MessageHandler():
_unknown_command_handler = lambda self, self2: "Unknown command, for a list send '!help'"
_on_connect = lambda: None
class WrongNumberOfArgsError(Exception):
pass
def _min_args(self, num_args):
if len(self.arguments) < num_args:
raise self.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):
"""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))]
reply = "Available commands:"
for method in methods:
docstring = getattr(self, method).__doc__
if docstring is None:
docstring = "No description available"
reply += "\n"+method+" ("+docstring.split("\n")[0]+")"
return reply
else:
method = getattr(self,self.arguments[0])
reply = trim(method.__doc__)
return reply
class GateMessageHandler(MessageHandler):
def configure(hndl, self):
@@ -601,11 +532,78 @@ class XMPPTelegram(ComponentXMPP):
self.gate_reply_message(msg, reply)
#msg.reply(reply).send()
def process_chat_user_command(self, iq):
parsed = iq['body'].split(' ')
jid = iq['from'].bare
class ChatCommandHandler(MessageHandler):
if parsed[0] == '!help':
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(hndl, self):
nickname = display_tg_name(self.tg_dialogs[hndl.jid]['users'][hndl.tg_id])
self.tg_connections[hndl.jid].invoke(BlockRequest( InputPeerUser(hndl.tg_id, self.tg_dialogs[jid]['users'][hndl.tg_id].access_hash) ) )
self.gate_reply_message(iq, 'User %s blacklisted!' % nickname)
def unblock(hndl, self):
nickname = display_tg_name(self.tg_dialogs[hndl.jid]['users'][hndl.tg_id])
self.tg_connections[hndl.jid].invoke(UnblockRequest( InputPeerUser(hndl.tg_id, self.tg_dialogs[jid]['users'][hndl.tg_id].access_hash) ) )
self.gate_reply_message(iq, 'User %s unblacklisted!' % nickname)
def remove(hndl, self):
peer = InputPeerUser(hndl.tg_id, self.tg_dialogs[hndl.jid]['users'][hndl.tg_id].access_hash)
c_jid = get_contact_jid(self.tg_dialogs[hndl.jid]['users'][hndl.tg_id], self.boundjid.bare)
self.tg_connections[hndl.jid].invoke( DeleteContactRequest(peer) )
self.tg_connections[hndl.jid].invoke( DeleteHistoryRequest( peer, max_id = 0, just_clear = None ) )
self.send_presence(pto = hndl.jid, pfrom = c_jid, ptype = 'unavailable')
self.send_presence(pto = hndl.jid, pfrom = c_jid, ptype = 'unsubscribed')
self.send_presence(pto = hndl.jid, pfrom = c_jid, ptype = 'unsubscribe')
def _replace(hndl, self):
peer = InputPeerUser(hdnl.tg_id, self.tg_dialogs[hndl.jid]['users'][hndl.tg_id].access_hash)
msg_id, edited = self.edit_message(hdnl.jid, hdnl.tg_id, hndl.msg['body'])
if not edited: return
# and send it
if edited != '' and edited != ' ':
self.tg_dialogs[hndl.jid]['messages'][hndl.tg_id]["body"] = edited
self.tg_connections[hndl.jid].invoke( EditMessageRequest(peer, msg_id, message = edited) )
else:
del(self.tg_dialogs[hndl.jid]['messages'][hndl.tg_id])
self.tg_connections[hndl.jid].invoke( DeleteMessagesRequest([msg_id], revoke = True) )
def process_chat_user_command(self, iq):
logging.info("received command "+str(msg["body"])+" from "+str(msg["from"])+" for "+str(msg["to"]))
is_command = msg["body"].startswith("!") and msg["body"][1] != "_"
if is_command:
command = msg["body"].split(" ")[0][1:]
handler = self.ChatCommandHandler(msg)._handler
try:
reply = str(handler(self))
except Exception as e:
if self.config["debug"]:
reply = "******* DEBUG MODE ACTIVE *********\n"
reply += "An Exception occured while executing this command:\n"
reply += traceback.format_exc()
else:
if isinstance(e, NotAuthorizedError):
reply = str(e)
elif isinstance(e, self.MessageHandler.WrongNumberOfArgsError):
reply = str(e)
else:
logging.error("Exception in command from {}, command was '{}'".format(msg["from"],msg["body"]))
traceback.print_exc()
reply = "Internal error, please contact Sysadmin"
if reply is not None:
self.gate_reply_message(msg, reply)
#msg.reply(reply).send()
if False:
self.gate_reply_message(iq, '=== Available dialog commands ===:\n\n'
'!help - Displays this text\n'
@@ -614,42 +612,6 @@ class XMPPTelegram(ComponentXMPP):
'!unblock - Unblacklists current user\n'
'!remove - Removes history and contact from your contact list\n'
)
elif parsed[0] == '!block':
tg_id = int(iq['to'].node[1:])
nickname = display_tg_name(self.tg_dialogs[jid]['users'][tg_id])
self.tg_connections[jid].invoke(BlockRequest( InputPeerUser(tg_id, self.tg_dialogs[jid]['users'][tg_id].access_hash) ) )
self.gate_reply_message(iq, 'User %s blacklisted!' % nickname)
elif parsed[0] == '!unblock':
tg_id = int(iq['to'].node[1:])
nickname = display_tg_name(self.tg_dialogs[jid]['users'][tg_id])
self.tg_connections[jid].invoke(UnblockRequest( InputPeerUser(tg_id, self.tg_dialogs[jid]['users'][tg_id].access_hash) ) )
self.gate_reply_message(iq, 'User %s unblacklisted!' % nickname)
elif parsed[0] == '!remove':
tg_id = int(iq['to'].node[1:])
peer = InputPeerUser(tg_id, self.tg_dialogs[jid]['users'][tg_id].access_hash)
c_jid = get_contact_jid(self.tg_dialogs[jid]['users'][tg_id], self.boundjid.bare)
self.tg_connections[jid].invoke( DeleteContactRequest(peer) )
self.tg_connections[jid].invoke( DeleteHistoryRequest( peer, max_id = 0, just_clear = None ) )
self.send_presence(pto = jid, pfrom = c_jid, ptype = 'unavailable')
self.send_presence(pto = jid, pfrom = c_jid, ptype = 'unsubscribed')
self.send_presence(pto = jid, pfrom = c_jid, ptype = 'unsubscribe')
elif iq['body'].startswith('!s/'):
tg_id = int(iq['to'].node[1:])
peer = InputPeerUser(tg_id, self.tg_dialogs[jid]['users'][tg_id].access_hash)
msg_id, edited = self.edit_message(jid, tg_id, iq['body'])
if not edited: return
# and send it
if edited != '' and edited != ' ':
self.tg_dialogs[jid]['messages'][tg_id]["body"] = edited
self.tg_connections[jid].invoke( EditMessageRequest(peer, msg_id, message = edited) )
else:
del(self.tg_dialogs[jid]['messages'][tg_id])
self.tg_connections[jid].invoke( DeleteMessagesRequest([msg_id], revoke = True) )
def process_chat_group_command(self, iq):
@@ -1001,3 +963,76 @@ class XMPPTelegram(ComponentXMPP):
conn.execute("CREATE TABLE IF NOT EXISTS accounts(jid VARCHAR(255), tg_phone VARCHAR(25), use_roster_exchange BOOLEAN default false, keep_online BOOLEAN default false, status_update_interval INTEGER default 30, status_xa_interval INTEGER default 24, enable_avatars BOOLEAN default false)")
return conn
class MessageHandler():
_on_connect = lambda: None
def _unknown_command_handler(self, *args, **kwargs):
return "Unknown command, for a list send !help"
class WrongNumberOfArgsError(Exception):
pass
def _min_args(self, num_args):
if len(self.arguments) < num_args:
raise self.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):
"""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))]
reply = "Available commands:"
for method in methods:
docstring = getattr(self, method).__doc__
if docstring is None:
docstring = "No description available"
reply += "\n"+method+" ("+docstring.split("\n")[0]+")"
return reply
else:
method = getattr(self,self.arguments[0])
reply = trim(method.__doc__)
return reply