Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- from api import LineClient
- from api.services import ttypes
- import threading
- import traceback
- import livejson
- class Bot(threading.Thread):
- client = None
- unpaused = True
- alive = True
- invites = False
- def __init__(self, id=None, password=None, token=None, qr=False, debug=False, names=[]):
- super(Bot, self).__init__()
- if id is not None and password is not None:
- self.client = LineClient(id=id, password=password, qr=qr)
- elif token is not None:
- self.client = LineClient(token=token, qr=qr)
- else:
- self.client = LineClient()
- self.debug = debug
- self.bot_data = livejson.File("%s_data.json" % self.client.profile.mid)
- if "users" not in self.bot_data.keys():
- self.bot_data["users"] = []
- self.bot_data["users"].append(self.client.profile.mid)
- print("First time loading users, created list and added my own UID")
- self.names = names
- self.ops_dict = {26: self.operation_message_receive, 13: self.operation_invitation}
- self.message_type_dict = {0: self.process_message}
- self.command_dict = {"test": self.command_test}
- def run(self):
- while self.alive is True:
- while self.unpaused is True:
- for op in self.client.long_poll():
- if self.debug is True:
- print("[%s]:\t%s" % (ttypes.OpType._VALUES_TO_NAMES[op.type], op))
- if op.type in self.ops_dict.keys():
- self.ops_dict[op.type](op)
- else:
- # op unprocessed
- pass
- def operation_message_receive(self, op):
- if op.message.content_type in self.message_type_dict.keys():
- self.message_type_dict[op.message.content_type](op.message)
- else:
- # message unprocessed
- pass
- def operation_invitation(self, op):
- # handle invitations
- self.operation_invitation.append
- pass
- def command_test(self, message):
- message.reply_message("test")
- def process_message(self, message):
- # command detection (AUTIMM STYLE)
- if message.from_id in self.bot_data["users"]:
- # user detected
- message.array_original = message.text.split(" ")
- message.array_lower = message.text.lower().split(" ")
- message.array_length = len(message.array_lower)
- if message.array_lower[0] in self.names:
- # command detected
- if message.array_lower[1] in self.command_dict.keys():
- # valid command
- self.command_dict[message.array_lower[1]](message)
- else:
- message.reply_message(message.text)
- def pause(self):
- if self.unpaused is True:
- self.unpaused = False
- else:
- self.unpaused = True
- print("Bot running: %s" % self.unpaused)
- def kill(self):
- self.alive = False
- print("Killed bot thread")
- def main():
- try:
- b = Bot(debug=True, names=['meh'])
- b.start()
- except Exception as e:
- print("Failed to login: %s" % e)
- traceback.print_exc()
- while True:
- c = input("pause/kill")
- if c.lower() == "pause":
- b.pause()
- elif c.lower() == "kill":
- b.kill()
- b.join()
- exit()
- if __name__ == '__main__':
- main()
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement