591 lines
20 KiB
Python
Executable File
591 lines
20 KiB
Python
Executable File
#!/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 = "<userId>--<threadId>")
|
|
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()
|
|
|