path: root/external/messenger.py

#!/usr/bin/env python3

import sys
import json
import signal
import threading
import queue
import pickle
import time

import hashlib

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"
SEND = "send"
CLOSE = "close"

# external -> ezbr
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_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,
            }


# ---- 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 InitialSyncThread(threading.Thread):
    def __init__(self, client, bridge, threads, *args, **kwargs):
        super(InitialSyncThread, self).__init__(*args, **kwargs)

        self.client = client
        self.bridge = bridge
        self.threads = threads

    def run(self):
        sys.stderr.write("fb thread list: {}\n".format(self.threads))

        for thread in self.threads:
            sys.stderr.write("fb thread: {}\n".format(thread))
            if thread.type == ThreadType.GROUP:
                members = self.client.fetchAllUsersFromThreads([thread])

                self.bridge.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):
        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.bridge.getUserShortName(m) for m in who[:3]] + ["..."])
            else:
                room_info["name"] = ", ".join([self.bridge.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.bridge.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("fb thread member: {}\n".format(member))
            self.bridge.ensureJoined(self.bridge.getUserId(member), thread.uid)

    def backlog_room(self, thread):
        prev_last_seen = self.bridge.cache_get("last_seen_%s"%thread.uid)
        if prev_last_seen == "":
            messages = self.client.fetchThreadMessages(thread.uid, limit=100)
        else:
            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:
                        found = True
                        break
                    else:
                        messages.append(m)
        for m in reversed(messages):
            self.bridge.onMessage(thread_id=thread.uid,
                                  thread_type=thread.type,
                                  message_object=m)

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("Start client.listen()\n")
        self.client.listen()


# ---- MAIN LOOP THAT HANDLES REQUESTS FROM BRIDGE ----

class MessengerBridge:
    def __init__(self):
        self.rev_uid = {}
        self.uid_map = {}
        self.joined_map = {}

    def getUserId(self, user):
        retval = None
        if user.url is not None and not "?" in user.url:
            user_id = user.url.split("/")[-1]
            self.rev_uid[user_id] = user.uid
            retval = user_id
        else:
            retval = user.uid

        if user.uid not in self.uid_map:
            self.uid_map[user.uid] = retval

            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": self.getUserId(user),
                "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 in self.rev_uid:
            return self.rev_uid[user_id]
        else:
            return 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

        while self.keep_running:
            try:
                line = sys.stdin.readline()
            except KeyboardInterrupt:
                sys.stderr.write("(messenger) shutting down")
                self.close()
                time.sleep(5)
                sys.exit(0)

            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:
                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:
            client_file = "/tmp/fbclient_" + hashlib.sha224(cmd["data"]["email"].encode("utf-8")).hexdigest()

            try:
                f = open(client_file, "rb")
                self.client = pickle.load(f)
                f.close()
                sys.stderr.write("(python messenger) using previous client: {}\n".format(client_file))
            except:
                self.client = None

            if self.client is None:
                email, password = cmd["data"]["email"], cmd["data"]["password"]
                self.client = MessengerBridgeClient(email=email, password=password, max_tries=1)

            if not self.client.isLoggedIn():
                return {"_type": "ret_error", "error": "Unable to login (?)"}

            try:
                f = open(client_file, "wb")
                pickle.dump(self.client, f)
                f.close()
            except:
                pass

            self.client.setBridge(self)

            threads = self.client.fetchThreadList()
            # 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.GROUP:
                    self.getUserId(thread)

            InitialSyncThread(self.client, self, threads).start()
            ClientListenThread(self.client).start()

        elif ty == CLOSE:
            self.close()

        elif ty == GET_USER:
            userId = self.getUserIdFromUid(self.client.uid)
            return {"_type": REP_OK, "user": userId}

        elif ty == INVITE and cmd["room"] == "":
            return {"_type": REP_OK}

        elif ty == SEND:
            event = cmd["data"]
            if event["type"] in [EVENT_MESSAGE, EVENT_ACTION]:
                # TODO: attachments
                msg = Message(event["text"])
                if event["type"] == EVENT_ACTION:
                    msg.text = "* " + event["text"]

                if event["room"] != "":
                    msg_id = self.client.send(msg, thread_id=event["room"], thread_type=ThreadType.GROUP)
                elif event["recipient"] != "":
                    uid = self.revUserId(event["recipient"])
                    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})
        rep = q.get(block=True, timeout=30)
        del self.cache_gets[num]
        return rep

    def cache_put(self, key, value):
        self.write({"_type": CACHE_PUT, "key": key, "value": value})

    def ensureJoined(self, userId, room):
        key = "{}--{}".format(userId, room)
        if not key in self.joined_map:
            self.write({
                "_type": EVENT,
                "data": {
                    "type": EVENT_JOIN,
                    "author": userId,
                    "room": room,
                }
            })
            self.joined_map[key] = true

    def onMessage(self, thread_id, thread_type, message_object, **kwargs):
        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 = {
                "type": EVENT_MESSAGE,
                "author": author,
                "text": message_object.text,
                "attachments": []
            }
        for at in message_object.attachments:
            if isinstance(at, ImageAttachment):
                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) or isinstance(at, AudioAttachment):
                event["attachments"].append({
                    "filename": at.url.split("?")[0].split("/")[-1],
                    "url": at.url,
                })
            else:
                event["text"] += "\nUnhandled attachment: {}".format(at)

        if thread_type == ThreadType.GROUP:
            event["room"] = thread_id
            self.ensureJoined(author, thread_id)

        self.write({"_type": EVENT, "data": event})

        self.cache_put("last_seen_%s"%thread_id, message_object.uid)

    def onPeopleAdded(self, *args, **kwargs):
        pass
    def onPersonRemoved(self, *args, **kwargs):
        pass
    def onTitleChange(self, *args, **kwargs):
        pass

if __name__ == "__main__":
    bridge = MessengerBridge()
    bridge.run()