#!/usr/bin/env python3 import sys import json import signal import threading import queue import pickle import time import traceback from urllib.parse import unquote as UrlUnquote import base64 import getpass import zlib import fbchat from fbchat.models import * # ---- MESSAGE TYPES ---- # ezbr -> external CONFIGURE = "configure" GET_USER = "get_user" SET_USER_INFO = "set_user_info" SET_ROOM_INFO = "set_room_info" JOIN = "join" INVITE = "invite" LEAVE = "leave" SEARCH = "search" SEND = "send" CLOSE = "close" # external -> ezbr SAVE_CONFIG = "save_config" JOINED = "joined" LEFT = "left" USER_INFO_UPDATED = "user_info_updated" ROOM_INFO_UPDATED = "room_info_updated" EVENT = "event" CACHE_PUT = "cache_put" CACHE_GET = "cache_get" # reply messages # ezbr -> external: all must wait for a reply! # external -> ezbr: only CACHE_GET produces a reply REP_OK = "rep_ok" REP_SEARCH_RESULTS = "rep_search_results" REP_ERROR = "rep_error" # Event types EVENT_JOIN = "join" EVENT_LEAVE = "leave" EVENT_MESSAGE = "message" EVENT_ACTION = "action" def mediaObjectOfURL(url): return { "filename": url.split("?")[0].split("/")[-1], "url": url, } def stripFbLinkPrefix(url): PREFIX = "https://l.facebook.com/l.php?u=" if url[:len(PREFIX)] == PREFIX: return UrlUnquote(url[len(PREFIX):].split('&')[0]) else: return url # ---- MESSENGER CLIENT CLASS THAT HANDLES EVENTS ---- class MessengerBridgeClient(fbchat.Client): def __init__(self, *args, **kwargs): super(MessengerBridgeClient, self).__init__(*args, **kwargs) self.bridge = None def setBridge(self, bridge): self.bridge = bridge ## Redirect all interesting events to Bridge def onMessage(self, *args, **kwargs): self.bridge.onMessage(*args, **kwargs) def onPeopleAdded(self, *args, **kwargs): self.bridge.onPeopleAdded(*args, **kwargs) def onPersonRemoved(self, *args, **kwargs): self.bridge.onPersonRemoved(*args, **kwargs) def onTitleChange(self, *args, **kwargs): self.bridge.onTitleChange(*args, **kwargs) # ---- SEPARATE THREADS FOR INITIAL SYNC & CLIENT LISTEN ---- class SyncerThread(threading.Thread): def __init__(self, bridge, thread_queue, *args, **kwargs): super(SyncerThread, self).__init__(*args, **kwargs) self.bridge = bridge self.thread_queue = thread_queue def run(self): while True: thread = self.thread_queue.get(block=True) sys.stderr.write("(python) fb thread: {}\n".format(thread)) self.bridge.setup_joined_thread(thread) class ClientListenThread(threading.Thread): def __init__(self, client, *args, **kwargs): super(ClientListenThread, self).__init__(*args, **kwargs) self.client = client def run(self): sys.stderr.write("(python messenger) Start client.listen()\n") self.client.listen() # ---- MAIN LOOP THAT HANDLES REQUESTS FROM BRIDGE ---- class MessengerBridge: def __init__(self): self.init_backlog_length = 100 # We cache maps between two kinds of identifiers: # - facebook uids of users # - identifiers for the bridge, which are the username when defined (otherwise equal to above) # Generally speaking, the first is referred to as uid whereas the second is just id # THESE MAPS SHOULD NOT BE USED DIRECTLY, instead functions getUserId, getUserIdFromUid and revUserId should be used self.uid_map = {} # map from fb user uid to bridge id self.rev_uid = {} # map fro bridge id to fb user uid # caches the room we (the user of the bridge) have joined (map keys = room uid) self.my_joined_rooms = {} # caches for the people that are in rooms so that we don't send JOINED every time (map keys = "--") self.others_joined_map = {} def getUserId(self, user): retval = None if user.url is not None and not "?" in user.url: retval = user.url.split("/")[-1] else: retval = user.uid if user.uid not in self.uid_map: self.uid_map[user.uid] = retval self.rev_uid[retval] = user.uid user_info = { "display_name": user.name, } if user.photo is not None: user_info["avatar"] = mediaObjectOfURL(user.photo) self.write({ "_type": USER_INFO_UPDATED, "user": retval, "data": user_info, }) return retval def getUserIdFromUid(self, uid): if uid in self.uid_map: return self.uid_map[uid] else: user = self.client.fetchUserInfo(uid)[uid] return self.getUserId(user) def revUserId(self, user_id): if user_id not in self.rev_uid: for user in self.client.searchForUsers(user_id): self.getUserId(user) if user_id not in self.rev_uid: raise ValueError("User not found: {}".format(user_id)) return self.rev_uid[user_id] def getUserShortName(self, user): if user.first_name != None: return user.first_name else: return user.name def run(self): self.client = None self.keep_running = True self.cache_gets = {} self.num = 0 self.my_user_id = "" while self.keep_running: try: line = sys.stdin.readline() except KeyboardInterrupt: sys.stderr.write("(python messenger) shutting down") self.close() break sys.stderr.write("(python) reading {}\n".format(line.strip())) cmd = json.loads(line) try: rep = self.handle_cmd(cmd) if rep is None: rep = {} if "_type" not in rep: rep["_type"] = REP_OK except Exception as e: sys.stderr.write("(python) {}\n".format(traceback.format_exc())) rep = { "_type": REP_ERROR, "error": "{}".format(e) } rep["_id"] = cmd["_id"] self.write(rep) def write(self, msg): msgstr = json.dumps(msg) sys.stderr.write("(python) writing {}\n".format(msgstr)) sys.stdout.write(msgstr + "\n") sys.stdout.flush() def handle_cmd(self, cmd): ty = cmd["_type"] if ty == CONFIGURE: self.init_backlog_length = int(cmd["data"]["initial_backlog"]) has_pickle = "client_pickle" in cmd["data"] and len(cmd["data"]["client_pickle"]) > 0 if has_pickle: data = base64.b64decode(cmd["data"]["client_pickle"]) data = zlib.decompress(data) self.client = pickle.loads(data) else: email, password = cmd["data"]["email"], cmd["data"]["password"] self.client = MessengerBridgeClient(email=email, password=password, max_tries=1) ## TODO: save client in new client_pickle config value if not self.client.isLoggedIn(): return {"_type": REP_ERROR, "error": "Unable to login (invalid pickle?)"} if not has_pickle: new_config = cmd["data"] data = pickle.dumps(self.client) data = zlib.compress(data) new_config["client_pickle"] = base64.b64encode(data).decode('ascii') self.write({"_type": SAVE_CONFIG, "data": new_config}) self.client.setBridge(self) self.my_user_id = self.getUserIdFromUid(self.client.uid) threads = self.client.fetchThreadList(limit=10) # ensure we have a correct mapping for bridged user IDs to fb uids # (this should be fast) for thread in threads: if thread.type == ThreadType.USER: self.getUserId(thread) self.sync_thread_queue = queue.Queue(100) SyncerThread(self, self.sync_thread_queue).start() for thread in reversed(threads): self.sync_thread_queue.put(thread) ClientListenThread(self.client).start() elif ty == CLOSE: self.close() elif ty == GET_USER: return {"_type": REP_OK, "user": self.my_user_id} elif ty == JOIN: self.ensure_i_joined(cmd["room"]) elif ty == LEAVE: thread_id = cmd["room"] self.client.removeUserFromGroup(self.client.uid, thread_id) if thread_id in self.my_joined_rooms: del self.my_joined_rooms[thread_id] elif ty == INVITE: if cmd["room"] != "": uid = self.revUserId(cmd["user"]) self.client.addUsersToGroup([uid], cmd["room"]) elif ty == SEARCH: users = self.client.searchForUsers(cmd["data"]) rep = [] for user in users: rep.append({ "id": self.getUserId(user), "display_name": user.name, }) return {"_type": REP_SEARCH_RESULTS, "data": rep} elif ty == SEND: event = cmd["data"] if event["type"] in [EVENT_MESSAGE, EVENT_ACTION]: attachments = [] if "attachments" in event and isinstance(event["attachments"], list): for at in event["attachments"]: if "url" in at: attachments.append(at["url"]) else: # TODO sys.stdout.write("Unhandled: attachment without URL") msg = Message(event["text"]) if event["type"] == EVENT_ACTION: msg.text = "* " + event["text"] if event["room"] != "": if len(attachments) > 0: msg_id = self.client.sendRemoteFiles(attachments, message=msg, thread_id=event["room"], thread_type=ThreadType.GROUP) else: msg_id = self.client.send(msg, thread_id=event["room"], thread_type=ThreadType.GROUP) elif event["recipient"] != "": uid = self.revUserId(event["recipient"]) sys.stderr.write("(python) Sending to {}\n".format(uid)) if len(attachments) > 0: msg_id = self.client.sendRemoteFiles(attachments, message=msg, thread_id=uid, thread_type=ThreadType.USER) else: msg_id = self.client.send(msg, thread_id=uid, thread_type=ThreadType.USER) else: return {"_type": REP_ERROR, "error": "Invalid message"} return {"_type": REP_OK, "event_id": msg_id} elif ty == REP_OK and cmd["_id"] in self.cache_gets: self.cache_gets[cmd["_id"]].put(cmd["value"]) else: return {"_type": REP_ERROR, "error": "Not implemented"} def close(self): self.keep_running = False self.client.stopListening() def cache_get(self, key): self.num += 1 num = self.num q = queue.Queue(1) self.cache_gets[num] = q self.write({"_type": CACHE_GET, "_id": num, "key": key}) try: rep = q.get(block=True, timeout=30) except queue.Empty: rep = "" del self.cache_gets[num] return rep def cache_put(self, key, value): self.write({"_type": CACHE_PUT, "key": key, "value": value}) # ---- Info sync ---- def ensure_i_joined(self, thread_id): if thread_id not in self.my_joined_rooms: self.my_joined_rooms[thread_id] = True thread = self.client.fetchThreadInfo(thread_id)[thread_id] self.sync_thread_queue.put(thread) def setup_joined_thread(self, thread): sys.stderr.write("(python) setup_joined_thread {}".format(thread)) if thread.type == ThreadType.GROUP: members = self.client.fetchAllUsersFromThreads([thread]) self.write({ "_type": JOINED, "room": thread.uid, }) self.send_room_info(thread, members) self.send_room_members(thread, members) self.backlog_room(thread) def send_room_info(self, thread, members): members.sort(key=lambda m: m.uid) room_info = {} if thread.name is not None: room_info["name"] = thread.name else: who = [m for m in members if m.uid != self.client.uid] if len(who) > 3: room_info["name"] = ", ".join([self.getUserShortName(m) for m in who[:3]] + ["..."]) else: room_info["name"] = ", ".join([self.getUserShortName(m) for m in who]) if thread.photo is not None: room_info["picture"] = mediaObjectOfURL(thread.photo) else: for m in members: if m.uid != self.client.uid and m.photo is not None: room_info["picture"] = mediaObjectOfURL(m.photo) break self.write({ "_type": ROOM_INFO_UPDATED, "room": thread.uid, "data": room_info, }) def send_room_members(self, thread, members): for member in members: sys.stderr.write("(python) fb thread member: {}\n".format(member)) self.ensureJoined(self.getUserId(member), thread.uid) def backlog_room(self, thread): prev_last_seen = self.cache_get("last_seen_%s"%thread.uid) if prev_last_seen == "": prev_last_seen = None messages = [] found = False while not found: before = None if len(messages) > 0: before = messages[-1].timestamp page = self.client.fetchThreadMessages(thread.uid, before=before, limit=20) for m in page: if m.uid == prev_last_seen or len(messages) > self.init_backlog_length: found = True break else: messages.append(m) for m in reversed(messages): if m.text is None: m.text = "" m.text = "[{}] {}".format( time.strftime("%Y-%m-%d %H:%M %Z", time.localtime(float(m.timestamp)/1000)).strip(), m.text) self.onMessage(thread_id=thread.uid, thread_type=thread.type, message_object=m) def ensureJoined(self, userId, room): key = "{}--{}".format(userId, room) if not key in self.others_joined_map: self.write({ "_type": EVENT, "data": { "type": EVENT_JOIN, "author": userId, "room": room, } }) self.others_joined_map[key] = True # ---- Event handlers ---- def onMessage(self, thread_id, thread_type, message_object, **kwargs): self.ensure_i_joined(thread_id) if message_object.author == self.client.uid: # Ignore our own messages return sys.stderr.write("(python messenger) Got message: {}\n".format(message_object)) author = self.getUserIdFromUid(message_object.author) event = { "id": message_object.uid, "type": EVENT_MESSAGE, "author": author, "text": message_object.text, "attachments": [] } if event["text"] is None: event["text"] = "" for at in message_object.attachments: if isinstance(at, ImageAttachment): try: full_url = self.client.fetchImageUrl(at.uid) except: time.sleep(1) full_url = self.client.fetchImageUrl(at.uid) event["attachments"].append({ "filename": full_url.split("?")[0].split("/")[-1], "url": full_url, "image_size": { "width": at.width, "height": at.height, }, }) elif isinstance(at, FileAttachment): url = stripFbLinkPrefix(at.url) event["attachments"].append({ "filename": at.name, "url": url, }) elif isinstance(at, AudioAttachment): url = stripFbLinkPrefix(at.url) event["attachments"].append({ "filename": at.filename, "url": url, }) elif isinstance(at, ShareAttachment): event["text"] += "\n{}\n{}".format(at.description, at.url) else: event["text"] += "\nUnhandled attachment: {}".format(at) if isinstance(message_object.sticker, Sticker): stk = message_object.sticker event["attachments"].append({ "filename": stk.label, "url": stk.url, "image_size": { "width": stk.width, "height": stk.height, }, }) if thread_type == ThreadType.GROUP: event["room"] = thread_id self.ensureJoined(author, thread_id) if event["text"] != "" or len(event["attachments"]) > 0: self.write({"_type": EVENT, "data": event}) self.cache_put("last_seen_%s"%thread_id, message_object.uid) def onPeopleAdded(self, added_ids, thread_id, *args, **kwargs): for user_id in added_ids: if user_id == self.client.uid: self.ensure_i_joined(thread_id) else: self.ensureJoined(self.getUserIdFromUid(user_id), thread_id) def onPersonRemoved(self, removed_id, thread_id, *args, **kwargs): if removed_id == self.client.uid: self.write({ "_type": LEFT, "room": thread_id, }) if thread_id in self.my_joined_rooms: del self.my_joined_rooms[thread_id] else: userId = self.getUserIdFromUid(removed_id), self.write({ "_type": EVENT, "data": { "type": EVENT_JOIN, "author": userId, "room": thread_id, } }) map_key = "{}--{}".format(userId, thread_id) if map_key in self.others_joined_map: del self.others_joined_map[map_key] def onTitleChange(self, author_id, new_title, thread_id, thread_type, *args, **kwargs): self.ensure_i_joined(thread_id) if thread_type == ThreadType.GROUP: self.write({ "_type": ROOM_INFO_UPDATED, "room": thread_id, "data": {"name": new_title}, }) # ---- CLI ---- def createClientPickle(): email = input("Email address of Facebook account: ") password = getpass.getpass() client = MessengerBridgeClient(email, password, max_tries=1) if not client.isLoggedIn(): print("Could not log in (why???)") print("Still creating pickle though, maybe it will work after login was authorized?") print("") data = pickle.dumps(client) data = zlib.compress(data) data = base64.b64encode(data).decode('ascii') print(data) if __name__ == "__main__": if "create_client_pickle" in sys.argv: createClientPickle() else: bridge = MessengerBridge() bridge.run()