karaolidis/fbchat
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #406 from carpedm20/refactor-model-parsing

Browse Source 
Refactor model parsing
This commit is contained in:
Mads Marquart
2019-04-25 21:06:26 +02:00
committed by GitHub
parent a71835a5b8 86a6e07804
commit 61502ed32a
21 changed files with 876 additions and 870 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
fbchat/__init__.py
View File

@@ -8,8 +8,9 @@ from __future__ import unicode_literals
# These imports are far too general, but they're needed for backwards compatbility.
from .utils import *
from .graphql import *
from .models import *
from ._graphql import graphql_queries_to_json, graphql_response_to_json, GraphQL
from ._client import Client
__title__ = "fbchat"

38
fbchat/_attachment.py
View File

@@ -2,6 +2,7 @@
from __future__ import unicode_literals
import attr
from . import _util
@attr.s(cmp=False)
@@ -46,3 +47,40 @@ class ShareAttachment(Attachment):
# Put here for backwards compatibility, so that the init argument order is preserved
uid = attr.ib(None)
@classmethod
def _from_graphql(cls, data):
from . import _file
url = data.get("url")
rtn = cls(
uid=data.get("deduplication_key"),
author=data["target"]["actors"][0]["id"]
if data["target"].get("actors")
else None,
url=url,
original_url=_util.get_url_parameter(url, "u")
if "/l.php?u=" in url
else url,
title=data["title_with_entities"].get("text"),
description=data["description"].get("text")
if data.get("description")
else None,
source=data["source"].get("text"),
attachments=[
_file.graphql_to_subattachment(attachment)
for attachment in data.get("subattachments")
],
)
media = data.get("media")
if media and media.get("image"):
image = media["image"]
rtn.image_url = image.get("uri")
rtn.original_image_url = (
_util.get_url_parameter(rtn.image_url, "url")
if "/safe_image.php" in rtn.image_url
else rtn.image_url
)
rtn.image_width = image.get("width")
rtn.image_height = image.get("height")
return rtn

224
fbchat/_client.py
View File

@@ -1,6 +1,6 @@
# -*- coding: UTF-8 -*-
from __future__ import unicode_literals
import requests
import urllib
from uuid import uuid1
@@ -10,7 +10,7 @@ from mimetypes import guess_type
from collections import OrderedDict
from ._util import *
from .models import *
from .graphql import *
from ._graphql import graphql_queries_to_json, graphql_response_to_json, GraphQL
import time
import json
@@ -680,24 +680,12 @@ class Client(object):
raise FBchatException("Missing payload while fetching users: {}".format(j))
users = []
for key in j["payload"]:
k = j["payload"][key]
if k["type"] in ["user", "friend"]:
if k["id"] in ["0", 0]:
for data in j["payload"].values():
if data["type"] in ["user", "friend"]:
if data["id"] in ["0", 0]:
# Skip invalid users
pass
users.append(
User(
k["id"],
first_name=k.get("firstName"),
url=k.get("uri"),
photo=k.get("thumbSrc"),
name=k.get("name"),
is_friend=k.get("is_friend"),
gender=GENDERS.get(k.get("gender")),
)
)
continue
users.append(User._from_all_fetch(data))
return users
def searchForUsers(self, name, limit=10):
@@ -713,7 +701,7 @@ class Client(object):
params = {"search": name, "limit": limit}
j = self.graphql_request(GraphQL(query=GraphQL.SEARCH_USER, params=params))
return [graphql_to_user(node) for node in j[name]["users"]["nodes"]]
return [User._from_graphql(node) for node in j[name]["users"]["nodes"]]
def searchForPages(self, name, limit=10):
"""
@@ -727,7 +715,7 @@ class Client(object):
params = {"search": name, "limit": limit}
j = self.graphql_request(GraphQL(query=GraphQL.SEARCH_PAGE, params=params))
return [graphql_to_page(node) for node in j[name]["pages"]["nodes"]]
return [Page._from_graphql(node) for node in j[name]["pages"]["nodes"]]
def searchForGroups(self, name, limit=10):
"""
@@ -742,7 +730,7 @@ class Client(object):
params = {"search": name, "limit": limit}
j = self.graphql_request(GraphQL(query=GraphQL.SEARCH_GROUP, params=params))
return [graphql_to_group(node) for node in j["viewer"]["groups"]["nodes"]]
return [Group._from_graphql(node) for node in j["viewer"]["groups"]["nodes"]]
def searchForThreads(self, name, limit=10):
"""
@@ -760,12 +748,12 @@ class Client(object):
rtn = []
for node in j[name]["threads"]["nodes"]:
if node["__typename"] == "User":
rtn.append(graphql_to_user(node))
rtn.append(User._from_graphql(node))
elif node["__typename"] == "MessageThread":
# MessageThread => Group thread
rtn.append(graphql_to_group(node))
rtn.append(Group._from_graphql(node))
elif node["__typename"] == "Page":
rtn.append(graphql_to_page(node))
rtn.append(Page._from_graphql(node))
elif node["__typename"] == "Group":
# We don't handle Facebook "Groups"
pass
@@ -1008,16 +996,16 @@ class Client(object):
entry = entry["message_thread"]
if entry.get("thread_type") == "GROUP":
_id = entry["thread_key"]["thread_fbid"]
rtn[_id] = graphql_to_group(entry)
rtn[_id] = Group._from_graphql(entry)
elif entry.get("thread_type") == "ONE_TO_ONE":
_id = entry["thread_key"]["other_user_id"]
if pages_and_users.get(_id) is None:
raise FBchatException("Could not fetch thread {}".format(_id))
entry.update(pages_and_users[_id])
if entry["type"] == ThreadType.USER:
rtn[_id] = graphql_to_user(entry)
rtn[_id] = User._from_graphql(entry)
else:
rtn[_id] = graphql_to_page(entry)
rtn[_id] = Page._from_graphql(entry)
else:
raise FBchatException(
"{} had an unknown thread type: {}".format(thread_ids[i], entry)
@@ -1053,7 +1041,7 @@ class Client(object):
raise FBchatException("Could not fetch thread {}: {}".format(thread_id, j))
messages = [
graphql_to_message(message)
Message._from_graphql(message)
for message in j["message_thread"]["messages"]["nodes"]
]
messages.reverse()
@@ -1061,8 +1049,6 @@ class Client(object):
read_receipts = j["message_thread"]["read_receipts"]["nodes"]
for message in messages:
if message.replied_to:
message.reply_to_id = message.replied_to.uid
for receipt in read_receipts:
if int(receipt["watermark"]) >= int(message.timestamp):
message.read_by.append(receipt["actor"]["id"])
@@ -1108,9 +1094,18 @@ class Client(object):
}
j = self.graphql_request(GraphQL(doc_id="1349387578499440", params=params))
return [
graphql_to_thread(node) for node in j["viewer"]["message_threads"]["nodes"]
]
rtn = []
for node in j["viewer"]["message_threads"]["nodes"]:
_type = node.get("thread_type")
if _type == "GROUP":
rtn.append(Group._from_graphql(node))
elif _type == "ONE_TO_ONE":
rtn.append(User._from_thread_fetch(node))
else:
raise FBchatException(
"Unknown thread type: {}, with data: {}".format(_type, node)
)
return rtn
def fetchUnread(self):
"""
@@ -1180,10 +1175,7 @@ class Client(object):
"""
thread_id, thread_type = self._getThread(thread_id, None)
message_info = self._forcedFetch(thread_id, mid).get("message")
message = graphql_to_message(message_info)
if message.replied_to:
message.reply_to_id = message.replied_to.uid
return message
return Message._from_graphql(message_info)
def fetchPollOptions(self, poll_id):
"""
@@ -1197,7 +1189,7 @@ class Client(object):
j = self._post(
self.req_url.GET_POLL_OPTIONS, data, fix_request=True, as_json=True
)
return [graphql_to_poll_option(m) for m in j["payload"]]
return [PollOption._from_graphql(m) for m in j["payload"]]
def fetchPlanInfo(self, plan_id):
"""
@@ -1210,7 +1202,7 @@ class Client(object):
"""
data = {"event_reminder_id": plan_id}
j = self._post(self.req_url.PLAN_INFO, data, fix_request=True, as_json=True)
return graphql_to_plan(j["payload"])
return Plan._from_fetch(j["payload"])
def _getPrivateData(self):
j = self.graphql_request(GraphQL(doc_id="1868889766468115"))
@@ -2470,7 +2462,7 @@ class Client(object):
# Color change
elif delta_type == "change_thread_theme":
new_color = graphql_color_to_enum(delta["untypedData"]["theme_color"])
new_color = ThreadColor._from_graphql(delta["untypedData"]["theme_color"])
thread_id, thread_type = getThreadIdAndThreadType(metadata)
self.onColorChange(
mid=mid,
@@ -2729,7 +2721,7 @@ class Client(object):
thread_id, thread_type = getThreadIdAndThreadType(metadata)
event_type = delta["untypedData"]["event_type"]
poll_json = json.loads(delta["untypedData"]["question_json"])
poll = graphql_to_poll(poll_json)
poll = Poll._from_graphql(poll_json)
if event_type == "question_creation":
# User created group poll
self.onPollCreated(
@@ -2762,10 +2754,9 @@ class Client(object):
# Plan created
elif delta_type == "lightweight_event_create":
thread_id, thread_type = getThreadIdAndThreadType(metadata)
plan = graphql_to_plan(delta["untypedData"])
self.onPlanCreated(
mid=mid,
plan=plan,
plan=Plan._from_pull(delta["untypedData"]),
author_id=author_id,
thread_id=thread_id,
thread_type=thread_type,
@@ -2777,10 +2768,9 @@ class Client(object):
# Plan ended
elif delta_type == "lightweight_event_notify":
thread_id, thread_type = getThreadIdAndThreadType(metadata)
plan = graphql_to_plan(delta["untypedData"])
self.onPlanEnded(
mid=mid,
plan=plan,
plan=Plan._from_pull(delta["untypedData"]),
thread_id=thread_id,
thread_type=thread_type,
ts=ts,
@@ -2791,10 +2781,9 @@ class Client(object):
# Plan edited
elif delta_type == "lightweight_event_update":
thread_id, thread_type = getThreadIdAndThreadType(metadata)
plan = graphql_to_plan(delta["untypedData"])
self.onPlanEdited(
mid=mid,
plan=plan,
plan=Plan._from_pull(delta["untypedData"]),
author_id=author_id,
thread_id=thread_id,
thread_type=thread_type,
@@ -2806,10 +2795,9 @@ class Client(object):
# Plan deleted
elif delta_type == "lightweight_event_delete":
thread_id, thread_type = getThreadIdAndThreadType(metadata)
plan = graphql_to_plan(delta["untypedData"])
self.onPlanDeleted(
mid=mid,
plan=plan,
plan=Plan._from_pull(delta["untypedData"]),
author_id=author_id,
thread_id=thread_id,
thread_type=thread_type,
@@ -2821,11 +2809,10 @@ class Client(object):
# Plan participation change
elif delta_type == "lightweight_event_rsvp":
thread_id, thread_type = getThreadIdAndThreadType(metadata)
plan = graphql_to_plan(delta["untypedData"])
take_part = delta["untypedData"]["guest_status"] == "GOING"
self.onPlanParticipation(
mid=mid,
plan=plan,
plan=Plan._from_pull(delta["untypedData"]),
take_part=take_part,
author_id=author_id,
thread_id=thread_id,
@@ -2903,7 +2890,7 @@ class Client(object):
for l in i["messageLiveLocations"]:
mid = l["messageId"]
author_id = str(l["senderId"])
location = graphql_to_live_location(l)
location = LiveLocationAttachment._from_pull(l)
self.onLiveLocation(
mid=mid,
location=location,
@@ -2934,8 +2921,8 @@ class Client(object):
i = d["deltaMessageReply"]
metadata = i["message"]["messageMetadata"]
thread_id, thread_type = getThreadIdAndThreadType(metadata)
message = graphql_to_message_reply(i["message"])
message.replied_to = graphql_to_message_reply(i["repliedToMessage"])
message = Message._from_reply(i["message"])
message.replied_to = Message._from_reply(i["repliedToMessage"])
message.reply_to_id = message.replied_to.uid
self.onMessage(
mid=message.uid,
@@ -2951,83 +2938,14 @@ class Client(object):
# New message
elif delta.get("class") == "NewMessage":
mentions = []
if delta.get("data") and delta["data"].get("prng"):
try:
mentions = [
Mention(
str(mention.get("i")),
offset=mention.get("o"),
length=mention.get("l"),
)
for mention in parse_json(delta["data"]["prng"])
]
except Exception:
log.exception("An exception occured while reading attachments")
sticker = None
attachments = []
unsent = False
if delta.get("attachments"):
try:
for a in delta["attachments"]:
mercury = a["mercury"]
if mercury.get("blob_attachment"):
image_metadata = a.get("imageMetadata", {})
attach_type = mercury["blob_attachment"]["__typename"]
attachment = graphql_to_attachment(
mercury["blob_attachment"]
)
if attach_type in [
"MessageFile",
"MessageVideo",
"MessageAudio",
]:
# TODO: Add more data here for audio files
attachment.size = int(a["fileSize"])
attachments.append(attachment)
elif mercury.get("sticker_attachment"):
sticker = graphql_to_sticker(mercury["sticker_attachment"])
elif mercury.get("extensible_attachment"):
attachment = graphql_to_extensible_attachment(
mercury["extensible_attachment"]
)
if isinstance(attachment, UnsentMessage):
unsent = True
elif attachment:
attachments.append(attachment)
except Exception:
log.exception(
"An exception occured while reading attachments: {}".format(
delta["attachments"]
)
)
if metadata and metadata.get("tags"):
emoji_size = get_emojisize_from_tags(metadata.get("tags"))
message = Message(
text=delta.get("body"),
mentions=mentions,
emoji_size=emoji_size,
sticker=sticker,
attachments=attachments,
)
message.uid = mid
message.author = author_id
message.timestamp = ts
# message.reactions = {}
message.unsent = unsent
thread_id, thread_type = getThreadIdAndThreadType(metadata)
self.onMessage(
mid=mid,
author_id=author_id,
message=delta.get("body", ""),
message_object=message,
message_object=Message._from_pull(
delta, tags=metadata.get("tags"), author=author_id, timestamp=ts
),
thread_id=thread_id,
thread_type=thread_type,
ts=ts,
@@ -3112,53 +3030,25 @@ class Client(object):
# Chat timestamp
elif mtype == "chatproxy-presence":
buddylist = dict()
for _id in m.get("buddyList", {}):
payload = m["buddyList"][_id]
statuses = dict()
for id_, data in m.get("buddyList", {}).items():
statuses[id_] = ActiveStatus._from_chatproxy_presence(id_, data)
self._buddylist[id_] = statuses[id_]
last_active = payload.get("lat")
active = payload.get("p") in [2, 3]
in_game = int(_id) in m.get("gamers", {})
buddylist[_id] = last_active
if self._buddylist.get(_id):
self._buddylist[_id].last_active = last_active
self._buddylist[_id].active = active
self._buddylist[_id].in_game = in_game
else:
self._buddylist[_id] = ActiveStatus(
active=active, last_active=last_active, in_game=in_game
)
self.onChatTimestamp(buddylist=buddylist, msg=m)
self.onChatTimestamp(buddylist=statuses, msg=m)
# Buddylist overlay
elif mtype == "buddylist_overlay":
statuses = dict()
for _id in m.get("overlay", {}):
payload = m["overlay"][_id]
for id_, data in m.get("overlay", {}).items():
old_in_game = None
if id_ in self._buddylist:
old_in_game = self._buddylist[id_].in_game
last_active = payload.get("la")
active = payload.get("a") in [2, 3]
in_game = (
self._buddylist[_id].in_game
if self._buddylist.get(_id)
else False
statuses[id_] = ActiveStatus._from_buddylist_overlay(
data, old_in_game
)
status = ActiveStatus(
active=active, last_active=last_active, in_game=in_game
)
if self._buddylist.get(_id):
self._buddylist[_id].last_active = last_active
self._buddylist[_id].active = active
self._buddylist[_id].in_game = in_game
else:
self._buddylist[_id] = status
statuses[_id] = status
self._buddylist[id_] = statuses[id_]
self.onBuddylistOverlay(statuses=statuses, msg=m)

14
fbchat/_core.py
View File

@@ -1,8 +1,11 @@
# -*- coding: UTF-8 -*-
from __future__ import unicode_literals
import logging
import aenum
log = logging.getLogger("client")
class Enum(aenum.Enum):
"""Used internally by fbchat to support enumerations"""
@@ -10,3 +13,14 @@ class Enum(aenum.Enum):
def __repr__(self):
# For documentation:
return "{}.{}".format(type(self).__name__, self.name)
@classmethod
def _extend_if_invalid(cls, value):
try:
return cls(value)
except ValueError:
log.warning(
"Failed parsing {.__name__}({!r}). Extending enum.".format(cls, value)
)
aenum.extend_enum(cls, "UNKNOWN_{}".format(value).upper(), value)
return cls(value)

78
fbchat/_file.py
View File

@@ -21,6 +21,15 @@ class FileAttachment(Attachment):
# Put here for backwards compatibility, so that the init argument order is preserved
uid = attr.ib(None)
@classmethod
def _from_graphql(cls, data):
return cls(
url=data.get("url"),
name=data.get("filename"),
is_malicious=data.get("is_malicious"),
uid=data.get("message_file_fbid"),
)
@attr.s(cmp=False)
class AudioAttachment(Attachment):
@@ -38,6 +47,15 @@ class AudioAttachment(Attachment):
# Put here for backwards compatibility, so that the init argument order is preserved
uid = attr.ib(None)
@classmethod
def _from_graphql(cls, data):
return cls(
filename=data.get("filename"),
url=data.get("playable_url"),
duration=data.get("playable_duration_in_ms"),
audio_type=data.get("audio_type"),
)
@attr.s(cmp=False, init=False)
class ImageAttachment(Attachment):
@@ -122,6 +140,21 @@ class ImageAttachment(Attachment):
self.animated_preview_width = animated_preview.get("width")
self.animated_preview_height = animated_preview.get("height")
@classmethod
def _from_graphql(cls, data):
return cls(
original_extension=data.get("original_extension")
or (data["filename"].split("-")[0] if data.get("filename") else None),
width=data.get("original_dimensions", {}).get("width"),
height=data.get("original_dimensions", {}).get("height"),
is_animated=data["__typename"] == "MessageAnimatedImage",
thumbnail_url=data.get("thumbnail", {}).get("uri"),
preview=data.get("preview") or data.get("preview_image"),
large_preview=data.get("large_preview"),
animated_preview=data.get("animated_image"),
uid=data.get("legacy_attachment_id"),
)
@attr.s(cmp=False, init=False)
class VideoAttachment(Attachment):
@@ -195,3 +228,48 @@ class VideoAttachment(Attachment):
self.large_image_url = large_image.get("uri")
self.large_image_width = large_image.get("width")
self.large_image_height = large_image.get("height")
@classmethod
def _from_graphql(cls, data):
return cls(
width=data.get("original_dimensions", {}).get("width"),
height=data.get("original_dimensions", {}).get("height"),
duration=data.get("playable_duration_in_ms"),
preview_url=data.get("playable_url"),
small_image=data.get("chat_image"),
medium_image=data.get("inbox_image"),
large_image=data.get("large_image"),
uid=data.get("legacy_attachment_id"),
)
@classmethod
def _from_subattachment(cls, data):
media = data["media"]
return cls(
duration=media.get("playable_duration_in_ms"),
preview_url=media.get("playable_url"),
medium_image=media.get("image"),
uid=data["target"].get("video_id"),
)
def graphql_to_attachment(data):
_type = data["__typename"]
if _type in ["MessageImage", "MessageAnimatedImage"]:
return ImageAttachment._from_graphql(data)
elif _type == "MessageVideo":
return VideoAttachment._from_graphql(data)
elif _type == "MessageAudio":
return AudioAttachment._from_graphql(data)
elif _type == "MessageFile":
return FileAttachment._from_graphql(data)
return Attachment(uid=data.get("legacy_attachment_id"))
def graphql_to_subattachment(data):
_type = data["target"]["__typename"]
if _type == "Video":
return VideoAttachment._from_subattachment(data)
return None

591
fbchat/_graphql.py
View File

@@ -1,10 +1,10 @@
# -*- coding: UTF-8 -*-
from __future__ import unicode_literals
import json
import re
from .models import *
from ._util import *
from . import _util
from ._exception import FBchatException, FBchatUserError
# Shameless copy from https://stackoverflow.com/a/8730674
FLAGS = re.VERBOSE | re.MULTILINE | re.DOTALL
@@ -27,583 +27,6 @@ class ConcatJSONDecoder(json.JSONDecoder):
# End shameless copy
def graphql_color_to_enum(color):
if color is None:
return None
if not color:
return ThreadColor.MESSENGER_BLUE
color = color[2:] # Strip the alpha value
color_value = "#{}".format(color.lower())
return enum_extend_if_invalid(ThreadColor, color_value)
def get_customization_info(thread):
if thread is None or thread.get("customization_info") is None:
return {}
info = thread["customization_info"]
rtn = {
"emoji": info.get("emoji"),
"color": graphql_color_to_enum(info.get("outgoing_bubble_color")),
}
if (
thread.get("thread_type") == "GROUP"
or thread.get("is_group_thread")
or thread.get("thread_key", {}).get("thread_fbid")
):
rtn["nicknames"] = {}
for k in info.get("participant_customizations", []):
rtn["nicknames"][k["participant_id"]] = k.get("nickname")
elif info.get("participant_customizations"):
uid = thread.get("thread_key", {}).get("other_user_id") or thread.get("id")
pc = info["participant_customizations"]
if len(pc) > 0:
if pc[0].get("participant_id") == uid:
rtn["nickname"] = pc[0].get("nickname")
else:
rtn["own_nickname"] = pc[0].get("nickname")
if len(pc) > 1:
if pc[1].get("participant_id") == uid:
rtn["nickname"] = pc[1].get("nickname")
else:
rtn["own_nickname"] = pc[1].get("nickname")
return rtn
def graphql_to_sticker(s):
if not s:
return None
sticker = Sticker(uid=s["id"])
if s.get("pack"):
sticker.pack = s["pack"].get("id")
if s.get("sprite_image"):
sticker.is_animated = True
sticker.medium_sprite_image = s["sprite_image"].get("uri")
sticker.large_sprite_image = s["sprite_image_2x"].get("uri")
sticker.frames_per_row = s.get("frames_per_row")
sticker.frames_per_col = s.get("frames_per_column")
sticker.frame_rate = s.get("frame_rate")
sticker.url = s.get("url")
sticker.width = s.get("width")
sticker.height = s.get("height")
if s.get("label"):
sticker.label = s["label"]
return sticker
def graphql_to_attachment(a):
_type = a["__typename"]
if _type in ["MessageImage", "MessageAnimatedImage"]:
return ImageAttachment(
original_extension=a.get("original_extension")
or (a["filename"].split("-")[0] if a.get("filename") else None),
width=a.get("original_dimensions", {}).get("width"),
height=a.get("original_dimensions", {}).get("height"),
is_animated=_type == "MessageAnimatedImage",
thumbnail_url=a.get("thumbnail", {}).get("uri"),
preview=a.get("preview") or a.get("preview_image"),
large_preview=a.get("large_preview"),
animated_preview=a.get("animated_image"),
uid=a.get("legacy_attachment_id"),
)
elif _type == "MessageVideo":
return VideoAttachment(
width=a.get("original_dimensions", {}).get("width"),
height=a.get("original_dimensions", {}).get("height"),
duration=a.get("playable_duration_in_ms"),
preview_url=a.get("playable_url"),
small_image=a.get("chat_image"),
medium_image=a.get("inbox_image"),
large_image=a.get("large_image"),
uid=a.get("legacy_attachment_id"),
)
elif _type == "MessageAudio":
return AudioAttachment(
filename=a.get("filename"),
url=a.get("playable_url"),
duration=a.get("playable_duration_in_ms"),
audio_type=a.get("audio_type"),
)
elif _type == "MessageFile":
return FileAttachment(
url=a.get("url"),
name=a.get("filename"),
is_malicious=a.get("is_malicious"),
uid=a.get("message_file_fbid"),
)
else:
return Attachment(uid=a.get("legacy_attachment_id"))
def graphql_to_extensible_attachment(a):
story = a.get("story_attachment")
if story:
target = story.get("target")
if target:
_type = target["__typename"]
if _type == "MessageLocation":
url = story.get("url")
address = get_url_parameter(get_url_parameter(url, "u"), "where1")
try:
latitude, longitude = [float(x) for x in address.split(", ")]
address = None
except ValueError:
latitude, longitude = None, None
rtn = LocationAttachment(
uid=int(story["deduplication_key"]),
latitude=latitude,
longitude=longitude,
address=address,
)
media = story.get("media")
if media and media.get("image"):
image = media["image"]
rtn.image_url = image.get("uri")
rtn.image_width = image.get("width")
rtn.image_height = image.get("height")
rtn.url = url
return rtn
elif _type == "MessageLiveLocation":
rtn = LiveLocationAttachment(
uid=int(story["target"]["live_location_id"]),
latitude=story["target"]["coordinate"]["latitude"]
if story["target"].get("coordinate")
else None,
longitude=story["target"]["coordinate"]["longitude"]
if story["target"].get("coordinate")
else None,
name=story["title_with_entities"]["text"],
expiration_time=story["target"].get("expiration_time"),
is_expired=story["target"].get("is_expired"),
)
media = story.get("media")
if media and media.get("image"):
image = media["image"]
rtn.image_url = image.get("uri")
rtn.image_width = image.get("width")
rtn.image_height = image.get("height")
rtn.url = story.get("url")
return rtn
elif _type in ["ExternalUrl", "Story"]:
url = story.get("url")
rtn = ShareAttachment(
uid=a.get("legacy_attachment_id"),
author=story["target"]["actors"][0]["id"]
if story["target"].get("actors")
else None,
url=url,
original_url=get_url_parameter(url, "u")
if "/l.php?u=" in url
else url,
title=story["title_with_entities"].get("text"),
description=story["description"].get("text")
if story.get("description")
else None,
source=story["source"].get("text"),
attachments=[
graphql_to_subattachment(attachment)
for attachment in story.get("subattachments")
],
)
media = story.get("media")
if media and media.get("image"):
image = media["image"]
rtn.image_url = image.get("uri")
rtn.original_image_url = (
get_url_parameter(rtn.image_url, "url")
if "/safe_image.php" in rtn.image_url
else rtn.image_url
)
rtn.image_width = image.get("width")
rtn.image_height = image.get("height")
return rtn
else:
return UnsentMessage(uid=a.get("legacy_attachment_id"))
def graphql_to_subattachment(a):
_type = a["target"]["__typename"]
if _type == "Video":
media = a["media"]
return VideoAttachment(
duration=media.get("playable_duration_in_ms"),
preview_url=media.get("playable_url"),
medium_image=media.get("image"),
uid=a["target"].get("video_id"),
)
def graphql_to_live_location(a):
return LiveLocationAttachment(
uid=a["id"],
latitude=a["coordinate"]["latitude"] / (10 ** 8)
if not a.get("stopReason")
else None,
longitude=a["coordinate"]["longitude"] / (10 ** 8)
if not a.get("stopReason")
else None,
name=a.get("locationTitle"),
expiration_time=a["expirationTime"],
is_expired=bool(a.get("stopReason")),
)
def graphql_to_poll(a):
rtn = Poll(
title=a.get("title") if a.get("title") else a.get("text"),
options=[graphql_to_poll_option(m) for m in a.get("options")],
)
rtn.uid = int(a["id"])
rtn.options_count = a.get("total_count")
return rtn
def graphql_to_poll_option(a):
if a.get("viewer_has_voted") is None:
vote = None
elif isinstance(a["viewer_has_voted"], bool):
vote = a["viewer_has_voted"]
else:
vote = a["viewer_has_voted"] == "true"
rtn = PollOption(text=a.get("text"), vote=vote)
rtn.uid = int(a["id"])
rtn.voters = (
[m.get("node").get("id") for m in a.get("voters").get("edges")]
if isinstance(a.get("voters"), dict)
else a.get("voters")
)
rtn.votes_count = (
a.get("voters").get("count")
if isinstance(a.get("voters"), dict)
else a.get("total_count")
)
return rtn
def graphql_to_plan(a):
if a.get("event_members"):
rtn = Plan(
time=a.get("event_time"),
title=a.get("title"),
location=a.get("location_name"),
)
if a.get("location_id") != 0:
rtn.location_id = str(a.get("location_id"))
rtn.uid = a.get("oid")
rtn.author_id = a.get("creator_id")
guests = a.get("event_members")
rtn.going = [uid for uid in guests if guests[uid] == "GOING"]
rtn.declined = [uid for uid in guests if guests[uid] == "DECLINED"]
rtn.invited = [uid for uid in guests if guests[uid] == "INVITED"]
return rtn
elif a.get("id") is None:
rtn = Plan(
time=a.get("event_time"),
title=a.get("event_title"),
location=a.get("event_location_name"),
location_id=a.get("event_location_id"),
)
rtn.uid = a.get("event_id")
rtn.author_id = a.get("event_creator_id")
guests = json.loads(a.get("guest_state_list"))
else:
rtn = Plan(
time=a.get("time"),
title=a.get("event_title"),
location=a.get("location_name"),
)
rtn.uid = a.get("id")
rtn.author_id = a.get("lightweight_event_creator").get("id")
guests = a.get("event_reminder_members").get("edges")
rtn.going = [
m.get("node").get("id") for m in guests if m.get("guest_list_state") == "GOING"
]
rtn.declined = [
m.get("node").get("id")
for m in guests
if m.get("guest_list_state") == "DECLINED"
]
rtn.invited = [
m.get("node").get("id")
for m in guests
if m.get("guest_list_state") == "INVITED"
]
return rtn
def graphql_to_quick_reply(q, is_response=False):
data = dict()
_type = q.get("content_type").lower()
if q.get("payload"):
data["payload"] = q["payload"]
if q.get("data"):
data["data"] = q["data"]
if q.get("image_url") and _type is not QuickReplyLocation._type:
data["image_url"] = q["image_url"]
data["is_response"] = is_response
if _type == QuickReplyText._type:
if q.get("title") is not None:
data["title"] = q["title"]
rtn = QuickReplyText(**data)
elif _type == QuickReplyLocation._type:
rtn = QuickReplyLocation(**data)
elif _type == QuickReplyPhoneNumber._type:
rtn = QuickReplyPhoneNumber(**data)
elif _type == QuickReplyEmail._type:
rtn = QuickReplyEmail(**data)
return rtn
def graphql_to_message(message):
if message.get("message_sender") is None:
message["message_sender"] = {}
if message.get("message") is None:
message["message"] = {}
rtn = Message(
text=message.get("message").get("text"),
mentions=[
Mention(
m.get("entity", {}).get("id"),
offset=m.get("offset"),
length=m.get("length"),
)
for m in message.get("message").get("ranges", [])
],
emoji_size=get_emojisize_from_tags(message.get("tags_list")),
sticker=graphql_to_sticker(message.get("sticker")),
)
rtn.uid = str(message.get("message_id"))
rtn.author = str(message.get("message_sender").get("id"))
rtn.timestamp = message.get("timestamp_precise")
rtn.unsent = False
if message.get("unread") is not None:
rtn.is_read = not message["unread"]
rtn.reactions = {
str(r["user"]["id"]): enum_extend_if_invalid(MessageReaction, r["reaction"])
for r in message.get("message_reactions")
}
if message.get("blob_attachments") is not None:
rtn.attachments = [
graphql_to_attachment(attachment)
for attachment in message["blob_attachments"]
]
if message.get("platform_xmd_encoded"):
quick_replies = json.loads(message["platform_xmd_encoded"]).get("quick_replies")
if isinstance(quick_replies, list):
rtn.quick_replies = [graphql_to_quick_reply(q) for q in quick_replies]
elif isinstance(quick_replies, dict):
rtn.quick_replies = [
graphql_to_quick_reply(quick_replies, is_response=True)
]
if message.get("extensible_attachment") is not None:
attachment = graphql_to_extensible_attachment(message["extensible_attachment"])
if isinstance(attachment, UnsentMessage):
rtn.unsent = True
elif attachment:
rtn.attachments.append(attachment)
if message.get("replied_to_message") is not None:
rtn.replied_to = graphql_to_message(message["replied_to_message"]["message"])
return rtn
def graphql_to_message_reply(message):
rtn = Message(
text=message.get("body"),
mentions=[
Mention(m.get("i"), offset=m.get("o"), length=m.get("l"))
for m in json.loads(message.get("data", {}).get("prng", "[]"))
],
emoji_size=get_emojisize_from_tags(message["messageMetadata"].get("tags")),
)
metadata = message.get("messageMetadata", {})
rtn.uid = metadata.get("messageId")
rtn.author = str(metadata.get("actorFbId"))
rtn.timestamp = metadata.get("timestamp")
rtn.unsent = False
if message.get("data", {}).get("platform_xmd"):
quick_replies = json.loads(message["data"]["platform_xmd"]).get("quick_replies")
if isinstance(quick_replies, list):
rtn.quick_replies = [graphql_to_quick_reply(q) for q in quick_replies]
elif isinstance(quick_replies, dict):
rtn.quick_replies = [
graphql_to_quick_reply(quick_replies, is_response=True)
]
if message.get("attachments") is not None:
for attachment in message["attachments"]:
attachment = json.loads(attachment["mercuryJSON"])
if attachment.get("blob_attachment"):
rtn.attachments.append(
graphql_to_attachment(attachment["blob_attachment"])
)
if attachment.get("extensible_attachment"):
extensible_attachment = graphql_to_extensible_attachment(
attachment["extensible_attachment"]
)
if isinstance(extensible_attachment, UnsentMessage):
rtn.unsent = True
else:
rtn.attachments.append(extensible_attachment)
if attachment.get("sticker_attachment"):
rtn.sticker = graphql_to_sticker(attachment["sticker_attachment"])
return rtn
def graphql_to_user(user):
if user.get("profile_picture") is None:
user["profile_picture"] = {}
c_info = get_customization_info(user)
plan = None
if user.get("event_reminders"):
plan = (
graphql_to_plan(user["event_reminders"]["nodes"][0])
if user["event_reminders"].get("nodes")
else None
)
return User(
user["id"],
url=user.get("url"),
first_name=user.get("first_name"),
last_name=user.get("last_name"),
is_friend=user.get("is_viewer_friend"),
gender=GENDERS.get(user.get("gender")),
affinity=user.get("affinity"),
nickname=c_info.get("nickname"),
color=c_info.get("color"),
emoji=c_info.get("emoji"),
own_nickname=c_info.get("own_nickname"),
photo=user["profile_picture"].get("uri"),
name=user.get("name"),
message_count=user.get("messages_count"),
plan=plan,
)
def graphql_to_thread(thread):
if thread["thread_type"] == "GROUP":
return graphql_to_group(thread)
elif thread["thread_type"] == "ONE_TO_ONE":
if thread.get("big_image_src") is None:
thread["big_image_src"] = {}
c_info = get_customization_info(thread)
participants = [
node["messaging_actor"] for node in thread["all_participants"]["nodes"]
]
user = next(
p for p in participants if p["id"] == thread["thread_key"]["other_user_id"]
)
last_message_timestamp = None
if "last_message" in thread:
last_message_timestamp = thread["last_message"]["nodes"][0][
"timestamp_precise"
]
first_name = user.get("short_name")
if first_name is None:
last_name = None
else:
last_name = user.get("name").split(first_name, 1).pop().strip()
plan = None
if thread.get("event_reminders"):
plan = (
graphql_to_plan(thread["event_reminders"]["nodes"][0])
if thread["event_reminders"].get("nodes")
else None
)
return User(
user["id"],
url=user.get("url"),
name=user.get("name"),
first_name=first_name,
last_name=last_name,
is_friend=user.get("is_viewer_friend"),
gender=GENDERS.get(user.get("gender")),
affinity=user.get("affinity"),
nickname=c_info.get("nickname"),
color=c_info.get("color"),
emoji=c_info.get("emoji"),
own_nickname=c_info.get("own_nickname"),
photo=user["big_image_src"].get("uri"),
message_count=thread.get("messages_count"),
last_message_timestamp=last_message_timestamp,
plan=plan,
)
else:
raise FBchatException(
"Unknown thread type: {}, with data: {}".format(
thread.get("thread_type"), thread
)
)
def graphql_to_group(group):
if group.get("image") is None:
group["image"] = {}
c_info = get_customization_info(group)
last_message_timestamp = None
if "last_message" in group:
last_message_timestamp = group["last_message"]["nodes"][0]["timestamp_precise"]
plan = None
if group.get("event_reminders"):
plan = (
graphql_to_plan(group["event_reminders"]["nodes"][0])
if group["event_reminders"].get("nodes")
else None
)
return Group(
group["thread_key"]["thread_fbid"],
participants=set(
[
node["messaging_actor"]["id"]
for node in group["all_participants"]["nodes"]
]
),
nicknames=c_info.get("nicknames"),
color=c_info.get("color"),
emoji=c_info.get("emoji"),
admins=set([node.get("id") for node in group.get("thread_admins")]),
approval_mode=bool(group.get("approval_mode"))
if group.get("approval_mode") is not None
else None,
approval_requests=set(
node["requester"]["id"] for node in group["group_approval_queue"]["nodes"]
)
if group.get("group_approval_queue")
else None,
join_link=group["joinable_mode"].get("link"),
photo=group["image"].get("uri"),
name=group.get("name"),
message_count=group.get("messages_count"),
last_message_timestamp=last_message_timestamp,
plan=plan,
)
def graphql_to_page(page):
if page.get("profile_picture") is None:
page["profile_picture"] = {}
if page.get("city") is None:
page["city"] = {}
plan = None
if page.get("event_reminders"):
plan = (
graphql_to_plan(page["event_reminders"]["nodes"][0])
if page["event_reminders"].get("nodes")
else None
)
return Page(
page["id"],
url=page.get("url"),
city=page.get("city").get("name"),
category=page.get("category_type"),
photo=page["profile_picture"].get("uri"),
name=page.get("name"),
message_count=page.get("messages_count"),
plan=plan,
)
def graphql_queries_to_json(*queries):
"""
Queries should be a list of GraphQL objects
@@ -615,7 +38,7 @@ def graphql_queries_to_json(*queries):
def graphql_response_to_json(content):
content = strip_to_json(content) # Usually only needed in some error cases
content = _util.strip_to_json(content) # Usually only needed in some error cases
try:
j = json.loads(content, cls=ConcatJSONDecoder)
except Exception:
@@ -626,15 +49,15 @@ def graphql_response_to_json(content):
if "error_results" in x:
del rtn[-1]
continue
check_json(x)
_util.check_json(x)
[(key, value)] = x.items()
check_json(value)
_util.check_json(value)
if "response" in value:
rtn[int(key[1:])] = value["response"]
else:
rtn[int(key[1:])] = value["data"]
log.debug(rtn)
_util.log.debug(rtn)
return rtn

44
fbchat/_group.py
View File

@@ -2,6 +2,7 @@
from __future__ import unicode_literals
import attr
from . import _plan
from ._thread import ThreadType, Thread
@@ -60,6 +61,49 @@ class Group(Thread):
self.approval_requests = approval_requests
self.join_link = join_link
@classmethod
def _from_graphql(cls, data):
if data.get("image") is None:
data["image"] = {}
c_info = cls._parse_customization_info(data)
last_message_timestamp = None
if "last_message" in data:
last_message_timestamp = data["last_message"]["nodes"][0][
"timestamp_precise"
]
plan = None
if data.get("event_reminders") and data["event_reminders"].get("nodes"):
plan = _plan.Plan._from_graphql(data["event_reminders"]["nodes"][0])
return cls(
data["thread_key"]["thread_fbid"],
participants=set(
[
node["messaging_actor"]["id"]
for node in data["all_participants"]["nodes"]
]
),
nicknames=c_info.get("nicknames"),
color=c_info.get("color"),
emoji=c_info.get("emoji"),
admins=set([node.get("id") for node in data.get("thread_admins")]),
approval_mode=bool(data.get("approval_mode"))
if data.get("approval_mode") is not None
else None,
approval_requests=set(
node["requester"]["id"]
for node in data["group_approval_queue"]["nodes"]
)
if data.get("group_approval_queue")
else None,
join_link=data["joinable_mode"].get("link"),
photo=data["image"].get("uri"),
name=data.get("name"),
message_count=data.get("messages_count"),
last_message_timestamp=last_message_timestamp,
plan=plan,
)
@attr.s(cmp=False, init=False)
class Room(Group):

64
fbchat/_location.py
View File

@@ -3,6 +3,7 @@ from __future__ import unicode_literals
import attr
from ._attachment import Attachment
from . import _util
@attr.s(cmp=False)
@@ -30,6 +31,30 @@ class LocationAttachment(Attachment):
# Put here for backwards compatibility, so that the init argument order is preserved
uid = attr.ib(None)
@classmethod
def _from_graphql(cls, data):
url = data.get("url")
address = _util.get_url_parameter(_util.get_url_parameter(url, "u"), "where1")
try:
latitude, longitude = [float(x) for x in address.split(", ")]
address = None
except ValueError:
latitude, longitude = None, None
rtn = cls(
uid=int(data["deduplication_key"]),
latitude=latitude,
longitude=longitude,
address=address,
)
media = data.get("media")
if media and media.get("image"):
image = media["image"]
rtn.image_url = image.get("uri")
rtn.image_width = image.get("width")
rtn.image_height = image.get("height")
rtn.url = url
return rtn
@attr.s(cmp=False, init=False)
class LiveLocationAttachment(LocationAttachment):
@@ -46,3 +71,42 @@ class LiveLocationAttachment(LocationAttachment):
super(LiveLocationAttachment, self).__init__(**kwargs)
self.expiration_time = expiration_time
self.is_expired = is_expired
@classmethod
def _from_pull(cls, data):
return cls(
uid=data["id"],
latitude=data["coordinate"]["latitude"] / (10 ** 8)
if not data.get("stopReason")
else None,
longitude=data["coordinate"]["longitude"] / (10 ** 8)
if not data.get("stopReason")
else None,
name=data.get("locationTitle"),
expiration_time=data["expirationTime"],
is_expired=bool(data.get("stopReason")),
)
@classmethod
def _from_graphql(cls, data):
target = data["target"]
rtn = cls(
uid=int(target["live_location_id"]),
latitude=target["coordinate"]["latitude"]
if target.get("coordinate")
else None,
longitude=target["coordinate"]["longitude"]
if target.get("coordinate")
else None,
name=data["title_with_entities"]["text"],
expiration_time=target.get("expiration_time"),
is_expired=target.get("is_expired"),
)
media = data.get("media")
if media and media.get("image"):
image = media["image"]
rtn.image_url = image.get("uri")
rtn.image_width = image.get("width")
rtn.image_height = image.get("height")
rtn.url = data.get("url")
return rtn

208
fbchat/_message.py
View File

@@ -2,7 +2,9 @@
from __future__ import unicode_literals
import attr
import json
from string import Formatter
from . import _util, _attachment, _location, _file, _quick_reply, _sticker
from ._core import Enum
@@ -13,6 +15,22 @@ class EmojiSize(Enum):
MEDIUM = "369239343222814"
SMALL = "369239263222822"
@classmethod
def _from_tags(cls, tags):
string_to_emojisize = {
"large": cls.LARGE,
"medium": cls.MEDIUM,
"small": cls.SMALL,
"l": cls.LARGE,
"m": cls.MEDIUM,
"s": cls.SMALL,
}
for tag in tags or ():
data = tag.split(":", maxsplit=1)
if len(data) > 1 and data[0] == "hot_emoji_size":
return string_to_emojisize.get(data[1])
return None
class MessageReaction(Enum):
"""Used to specify a message reaction"""
@@ -125,3 +143,193 @@ class Message(object):
message = cls(text=result, mentions=mentions)
return message
@classmethod
def _from_graphql(cls, data):
if data.get("message_sender") is None:
data["message_sender"] = {}
if data.get("message") is None:
data["message"] = {}
rtn = cls(
text=data["message"].get("text"),
mentions=[
Mention(
m.get("entity", {}).get("id"),
offset=m.get("offset"),
length=m.get("length"),
)
for m in data["message"].get("ranges") or ()
],
emoji_size=EmojiSize._from_tags(data.get("tags_list")),
sticker=_sticker.Sticker._from_graphql(data.get("sticker")),
)
rtn.uid = str(data["message_id"])
rtn.author = str(data["message_sender"]["id"])
rtn.timestamp = data.get("timestamp_precise")
rtn.unsent = False
if data.get("unread") is not None:
rtn.is_read = not data["unread"]
rtn.reactions = {
str(r["user"]["id"]): MessageReaction._extend_if_invalid(r["reaction"])
for r in data["message_reactions"]
}
if data.get("blob_attachments") is not None:
rtn.attachments = [
_file.graphql_to_attachment(attachment)
for attachment in data["blob_attachments"]
]
if data.get("platform_xmd_encoded"):
quick_replies = json.loads(data["platform_xmd_encoded"]).get(
"quick_replies"
)
if isinstance(quick_replies, list):
rtn.quick_replies = [
_quick_reply.graphql_to_quick_reply(q) for q in quick_replies
]
elif isinstance(quick_replies, dict):
rtn.quick_replies = [
_quick_reply.graphql_to_quick_reply(quick_replies, is_response=True)
]
if data.get("extensible_attachment") is not None:
attachment = graphql_to_extensible_attachment(data["extensible_attachment"])
if isinstance(attachment, _attachment.UnsentMessage):
rtn.unsent = True
elif attachment:
rtn.attachments.append(attachment)
if data.get("replied_to_message") is not None:
rtn.replied_to = cls._from_graphql(data["replied_to_message"]["message"])
rtn.reply_to_id = rtn.replied_to.uid
return rtn
@classmethod
def _from_reply(cls, data):
rtn = cls(
text=data.get("body"),
mentions=[
Mention(m.get("i"), offset=m.get("o"), length=m.get("l"))
for m in json.loads(data.get("data", {}).get("prng", "[]"))
],
emoji_size=EmojiSize._from_tags(data["messageMetadata"].get("tags")),
)
metadata = data.get("messageMetadata", {})
rtn.uid = metadata.get("messageId")
rtn.author = str(metadata.get("actorFbId"))
rtn.timestamp = metadata.get("timestamp")
rtn.unsent = False
if data.get("data", {}).get("platform_xmd"):
quick_replies = json.loads(data["data"]["platform_xmd"]).get(
"quick_replies"
)
if isinstance(quick_replies, list):
rtn.quick_replies = [
_quick_reply.graphql_to_quick_reply(q) for q in quick_replies
]
elif isinstance(quick_replies, dict):
rtn.quick_replies = [
_quick_reply.graphql_to_quick_reply(quick_replies, is_response=True)
]
if data.get("attachments") is not None:
for attachment in data["attachments"]:
attachment = json.loads(attachment["mercuryJSON"])
if attachment.get("blob_attachment"):
rtn.attachments.append(
_file.graphql_to_attachment(attachment["blob_attachment"])
)
if attachment.get("extensible_attachment"):
extensible_attachment = graphql_to_extensible_attachment(
attachment["extensible_attachment"]
)
if isinstance(extensible_attachment, _attachment.UnsentMessage):
rtn.unsent = True
else:
rtn.attachments.append(extensible_attachment)
if attachment.get("sticker_attachment"):
rtn.sticker = _sticker.Sticker._from_graphql(
attachment["sticker_attachment"]
)
return rtn
@classmethod
def _from_pull(cls, data, mid=None, tags=None, author=None, timestamp=None):
rtn = cls(text=data.get("body"))
rtn.uid = mid
rtn.author = author
rtn.timestamp = timestamp
if data.get("data") and data["data"].get("prng"):
try:
rtn.mentions = [
Mention(
str(mention.get("i")),
offset=mention.get("o"),
length=mention.get("l"),
)
for mention in _util.parse_json(data["data"]["prng"])
]
except Exception:
_util.log.exception("An exception occured while reading attachments")
if data.get("attachments"):
try:
for a in data["attachments"]:
mercury = a["mercury"]
if mercury.get("blob_attachment"):
image_metadata = a.get("imageMetadata", {})
attach_type = mercury["blob_attachment"]["__typename"]
attachment = _file.graphql_to_attachment(
mercury["blob_attachment"]
)
if attach_type in [
"MessageFile",
"MessageVideo",
"MessageAudio",
]:
# TODO: Add more data here for audio files
attachment.size = int(a["fileSize"])
rtn.attachments.append(attachment)
elif mercury.get("sticker_attachment"):
rtn.sticker = _sticker.Sticker._from_graphql(
mercury["sticker_attachment"]
)
elif mercury.get("extensible_attachment"):
attachment = graphql_to_extensible_attachment(
mercury["extensible_attachment"]
)
if isinstance(attachment, _attachment.UnsentMessage):
rtn.unsent = True
elif attachment:
rtn.attachments.append(attachment)
except Exception:
_util.log.exception(
"An exception occured while reading attachments: {}".format(
data["attachments"]
)
)
rtn.emoji_size = EmojiSize._from_tags(tags)
return rtn
def graphql_to_extensible_attachment(data):
story = data.get("story_attachment")
if not story:
return None
target = story.get("target")
if not target:
return _attachment.UnsentMessage(uid=data.get("legacy_attachment_id"))
_type = target["__typename"]
if _type == "MessageLocation":
return _location.LocationAttachment._from_graphql(story)
elif _type == "MessageLiveLocation":
return _location.LiveLocationAttachment._from_graphql(story)
elif _type in ["ExternalUrl", "Story"]:
return _attachment.ShareAttachment._from_graphql(story)
return None

22
fbchat/_page.py
View File

@@ -2,6 +2,7 @@
from __future__ import unicode_literals
import attr
from . import _plan
from ._thread import ThreadType, Thread
@@ -36,3 +37,24 @@ class Page(Thread):
self.likes = likes
self.sub_title = sub_title
self.category = category
@classmethod
def _from_graphql(cls, data):
if data.get("profile_picture") is None:
data["profile_picture"] = {}
if data.get("city") is None:
data["city"] = {}
plan = None
if data.get("event_reminders") and data["event_reminders"].get("nodes"):
plan = _plan.Plan._from_graphql(data["event_reminders"]["nodes"][0])
return cls(
data["id"],
url=data.get("url"),
city=data.get("city").get("name"),
category=data.get("category_type"),
photo=data["profile_picture"].get("uri"),
name=data.get("name"),
message_count=data.get("messages_count"),
plan=plan,
)

87
fbchat/_plan.py
View File

@@ -2,6 +2,14 @@
from __future__ import unicode_literals
import attr
import json
from ._core import Enum
class GuestStatus(Enum):
INVITED = 1
GOING = 2
DECLINED = 3
@attr.s(cmp=False)
@@ -20,9 +28,76 @@ class Plan(object):
location_id = attr.ib(None, converter=lambda x: x or "")
#: ID of the plan creator
author_id = attr.ib(None, init=False)
#: List of the people IDs who will take part in the plan
going = attr.ib(factory=list, init=False)
#: List of the people IDs who won't take part in the plan
declined = attr.ib(factory=list, init=False)
#: List of the people IDs who are invited to the plan
invited = attr.ib(factory=list, init=False)
#: Dict of `User` IDs mapped to their `GuestStatus`
guests = attr.ib(None, init=False)
@property
def going(self):
"""List of the `User` IDs who will take part in the plan."""
return [
id_
for id_, status in (self.guests or {}).items()
if status is GuestStatus.GOING
]
@property
def declined(self):
"""List of the `User` IDs who won't take part in the plan."""
return [
id_
for id_, status in (self.guests or {}).items()
if status is GuestStatus.DECLINED
]
@property
def invited(self):
"""List of the `User` IDs who are invited to the plan."""
return [
id_
for id_, status in (self.guests or {}).items()
if status is GuestStatus.INVITED
]
@classmethod
def _from_pull(cls, data):
rtn = cls(
time=data.get("event_time"),
title=data.get("event_title"),
location=data.get("event_location_name"),
location_id=data.get("event_location_id"),
)
rtn.uid = data.get("event_id")
rtn.author_id = data.get("event_creator_id")
rtn.guests = {
x["node"]["id"]: GuestStatus[x["guest_list_state"]]
for x in json.loads(data["guest_state_list"])
}
return rtn
@classmethod
def _from_fetch(cls, data):
rtn = cls(
time=data.get("event_time"),
title=data.get("title"),
location=data.get("location_name"),
location_id=str(data["location_id"]) if data.get("location_id") else None,
)
rtn.uid = data.get("oid")
rtn.author_id = data.get("creator_id")
rtn.guests = {id_: GuestStatus[s] for id_, s in data["event_members"].items()}
return rtn
@classmethod
def _from_graphql(cls, data):
rtn = cls(
time=data.get("time"),
title=data.get("event_title"),
location=data.get("location_name"),
)
rtn.uid = data.get("id")
rtn.author_id = data["lightweight_event_creator"].get("id")
rtn.guests = {
x["node"]["id"]: GuestStatus[x["guest_list_state"]]
for x in data["event_reminder_members"]["edges"]
}
return rtn

47
fbchat/_poll.py
View File

@@ -8,27 +8,60 @@ import attr
class Poll(object):
"""Represents a poll"""
#: ID of the poll
uid = attr.ib(None, init=False)
#: Title of the poll
title = attr.ib()
#: List of :class:`PollOption`, can be fetched with :func:`fbchat.Client.fetchPollOptions`
options = attr.ib()
#: Options count
options_count = attr.ib(None, init=False)
options_count = attr.ib(None)
#: ID of the poll
uid = attr.ib(None)
@classmethod
def _from_graphql(cls, data):
return cls(
uid=int(data["id"]),
title=data.get("title") if data.get("title") else data.get("text"),
options=[PollOption._from_graphql(m) for m in data.get("options")],
options_count=data.get("total_count"),
)
@attr.s(cmp=False)
class PollOption(object):
"""Represents a poll option"""
#: ID of the poll option
uid = attr.ib(None, init=False)
#: Text of the poll option
text = attr.ib()
#: Whether vote when creating or client voted
vote = attr.ib(False)
#: ID of the users who voted for this poll option
voters = attr.ib(None, init=False)
voters = attr.ib(None)
#: Votes count
votes_count = attr.ib(None, init=False)
votes_count = attr.ib(None)
#: ID of the poll option
uid = attr.ib(None)
@classmethod
def _from_graphql(cls, data):
if data.get("viewer_has_voted") is None:
vote = None
elif isinstance(data["viewer_has_voted"], bool):
vote = data["viewer_has_voted"]
else:
vote = data["viewer_has_voted"] == "true"
return cls(
uid=int(data["id"]),
text=data.get("text"),
vote=vote,
voters=(
[m.get("node").get("id") for m in data.get("voters").get("edges")]
if isinstance(data.get("voters"), dict)
else data.get("voters")
),
votes_count=(
data.get("voters").get("count")
if isinstance(data.get("voters"), dict)
else data.get("total_count")
),
)

23
fbchat/_quick_reply.py
View File

@@ -74,3 +74,26 @@ class QuickReplyEmail(QuickReply):
def __init__(self, image_url=None, **kwargs):
super(QuickReplyEmail, self).__init__(**kwargs)
self.image_url = image_url
def graphql_to_quick_reply(q, is_response=False):
data = dict()
_type = q.get("content_type").lower()
if q.get("payload"):
data["payload"] = q["payload"]
if q.get("data"):
data["data"] = q["data"]
if q.get("image_url") and _type is not QuickReplyLocation._type:
data["image_url"] = q["image_url"]
data["is_response"] = is_response
if _type == QuickReplyText._type:
if q.get("title") is not None:
data["title"] = q["title"]
rtn = QuickReplyText(**data)
elif _type == QuickReplyLocation._type:
rtn = QuickReplyLocation(**data)
elif _type == QuickReplyPhoneNumber._type:
rtn = QuickReplyPhoneNumber(**data)
elif _type == QuickReplyEmail._type:
rtn = QuickReplyEmail(**data)
return rtn

21
fbchat/_sticker.py
View File

@@ -37,3 +37,24 @@ class Sticker(Attachment):
def __init__(self, uid=None):
super(Sticker, self).__init__(uid=uid)
@classmethod
def _from_graphql(cls, data):
if not data:
return None
self = cls(uid=data["id"])
if data.get("pack"):
self.pack = data["pack"].get("id")
if data.get("sprite_image"):
self.is_animated = True
self.medium_sprite_image = data["sprite_image"].get("uri")
self.large_sprite_image = data["sprite_image_2x"].get("uri")
self.frames_per_row = data.get("frames_per_row")
self.frames_per_col = data.get("frames_per_column")
self.frame_rate = data.get("frame_rate")
self.url = data.get("url")
self.width = data.get("width")
self.height = data.get("height")
if data.get("label"):
self.label = data["label"]
return self

43
fbchat/_thread.py
View File

@@ -47,6 +47,16 @@ class ThreadColor(Enum):
DARK_TANGERINE = "#ff9c19"
BRIGHT_TURQUOISE = "#0edcde"
@classmethod
def _from_graphql(cls, color):
if color is None:
return None
if not color:
return cls.MESSENGER_BLUE
color = color[2:] # Strip the alpha value
value = "#{}".format(color.lower())
return cls._extend_if_invalid(value)
@attr.s(cmp=False, init=False)
class Thread(object):
@@ -84,3 +94,36 @@ class Thread(object):
self.last_message_timestamp = last_message_timestamp
self.message_count = message_count
self.plan = plan
@staticmethod
def _parse_customization_info(data):
if data is None or data.get("customization_info") is None:
return {}
info = data["customization_info"]
rtn = {
"emoji": info.get("emoji"),
"color": ThreadColor._from_graphql(info.get("outgoing_bubble_color")),
}
if (
data.get("thread_type") == "GROUP"
or data.get("is_group_thread")
or data.get("thread_key", {}).get("thread_fbid")
):
rtn["nicknames"] = {}
for k in info.get("participant_customizations", []):
rtn["nicknames"][k["participant_id"]] = k.get("nickname")
elif info.get("participant_customizations"):
uid = data.get("thread_key", {}).get("other_user_id") or data.get("id")
pc = info["participant_customizations"]
if len(pc) > 0:
if pc[0].get("participant_id") == uid:
rtn["nickname"] = pc[0].get("nickname")
else:
rtn["own_nickname"] = pc[0].get("nickname")
if len(pc) > 1:
if pc[1].get("participant_id") == uid:
rtn["nickname"] = pc[1].get("nickname")
else:
rtn["own_nickname"] = pc[1].get("nickname")
return rtn

132
fbchat/_user.py
View File

@@ -3,9 +3,40 @@ from __future__ import unicode_literals
import attr
from ._core import Enum
from . import _plan
from ._thread import ThreadType, Thread
GENDERS = {
# For standard requests
0: "unknown",
1: "female_singular",
2: "male_singular",
3: "female_singular_guess",
4: "male_singular_guess",
5: "mixed",
6: "neuter_singular",
7: "unknown_singular",
8: "female_plural",
9: "male_plural",
10: "neuter_plural",
11: "unknown_plural",
# For graphql requests
"UNKNOWN": "unknown",
"FEMALE": "female_singular",
"MALE": "male_singular",
# '': 'female_singular_guess',
# '': 'male_singular_guess',
# '': 'mixed',
"NEUTER": "neuter_singular",
# '': 'unknown_singular',
# '': 'female_plural',
# '': 'male_plural',
# '': 'neuter_plural',
# '': 'unknown_plural',
}
class TypingStatus(Enum):
"""Used to specify whether the user is typing or has stopped typing"""
@@ -65,6 +96,91 @@ class User(Thread):
self.color = color
self.emoji = emoji
@classmethod
def _from_graphql(cls, data):
if data.get("profile_picture") is None:
data["profile_picture"] = {}
c_info = cls._parse_customization_info(data)
plan = None
if data.get("event_reminders") and data["event_reminders"].get("nodes"):
plan = _plan.Plan._from_graphql(data["event_reminders"]["nodes"][0])
return cls(
data["id"],
url=data.get("url"),
first_name=data.get("first_name"),
last_name=data.get("last_name"),
is_friend=data.get("is_viewer_friend"),
gender=GENDERS.get(data.get("gender")),
affinity=data.get("affinity"),
nickname=c_info.get("nickname"),
color=c_info.get("color"),
emoji=c_info.get("emoji"),
own_nickname=c_info.get("own_nickname"),
photo=data["profile_picture"].get("uri"),
name=data.get("name"),
message_count=data.get("messages_count"),
plan=plan,
)
@classmethod
def _from_thread_fetch(cls, data):
if data.get("big_image_src") is None:
data["big_image_src"] = {}
c_info = cls._parse_customization_info(data)
participants = [
node["messaging_actor"] for node in data["all_participants"]["nodes"]
]
user = next(
p for p in participants if p["id"] == data["thread_key"]["other_user_id"]
)
last_message_timestamp = None
if "last_message" in data:
last_message_timestamp = data["last_message"]["nodes"][0][
"timestamp_precise"
]
first_name = user.get("short_name")
if first_name is None:
last_name = None
else:
last_name = user.get("name").split(first_name, 1).pop().strip()
plan = None
if data.get("event_reminders") and data["event_reminders"].get("nodes"):
plan = _plan.Plan._from_graphql(data["event_reminders"]["nodes"][0])
return cls(
user["id"],
url=user.get("url"),
name=user.get("name"),
first_name=first_name,
last_name=last_name,
is_friend=user.get("is_viewer_friend"),
gender=GENDERS.get(user.get("gender")),
affinity=user.get("affinity"),
nickname=c_info.get("nickname"),
color=c_info.get("color"),
emoji=c_info.get("emoji"),
own_nickname=c_info.get("own_nickname"),
photo=user["big_image_src"].get("uri"),
message_count=data.get("messages_count"),
last_message_timestamp=last_message_timestamp,
plan=plan,
)
@classmethod
def _from_all_fetch(cls, data):
return cls(
data["id"],
first_name=data.get("firstName"),
url=data.get("uri"),
photo=data.get("thumbSrc"),
name=data.get("name"),
is_friend=data.get("is_friend"),
gender=GENDERS.get(data.get("gender")),
)
@attr.s(cmp=False)
class ActiveStatus(object):
@@ -74,3 +190,19 @@ class ActiveStatus(object):
last_active = attr.ib(None)
#: Whether the user is playing Messenger game now
in_game = attr.ib(None)
@classmethod
def _from_chatproxy_presence(cls, id_, data):
return cls(
active=data["p"] in [2, 3] if "p" in data else None,
last_active=data.get("lat"),
in_game=int(id_) in data.get("gamers", {}),
)
@classmethod
def _from_buddylist_overlay(cls, data, in_game=None):
return cls(
active=data["a"] in [2, 3] if "a" in data else None,
last_active=data.get("la"),
in_game=None,
)

69
fbchat/_util.py
View File

@@ -11,8 +11,7 @@ from os.path import basename
import warnings
import logging
import requests
import aenum
from .models import *
from ._exception import FBchatException, FBchatFacebookError
try:
from urllib.parse import urlencode, parse_qs, urlparse
@@ -47,45 +46,6 @@ USER_AGENTS = [
"Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/536.6 (KHTML, like Gecko) Chrome/20.0.1092.0 Safari/536.6",
]
LIKES = {
"large": EmojiSize.LARGE,
"medium": EmojiSize.MEDIUM,
"small": EmojiSize.SMALL,
"l": EmojiSize.LARGE,
"m": EmojiSize.MEDIUM,
"s": EmojiSize.SMALL,
}
GENDERS = {
# For standard requests
0: "unknown",
1: "female_singular",
2: "male_singular",
3: "female_singular_guess",
4: "male_singular_guess",
5: "mixed",
6: "neuter_singular",
7: "unknown_singular",
8: "female_plural",
9: "male_plural",
10: "neuter_plural",
11: "unknown_plural",
# For graphql requests
"UNKNOWN": "unknown",
"FEMALE": "female_singular",
"MALE": "male_singular",
# '': 'female_singular_guess',
# '': 'male_singular_guess',
# '': 'mixed',
"NEUTER": "neuter_singular",
# '': 'unknown_singular',
# '': 'female_plural',
# '': 'male_plural',
# '': 'neuter_plural',
# '': 'unknown_plural',
}
class ReqUrl(object):
"""A class containing all urls used by `fbchat`"""
@@ -295,20 +255,6 @@ def get_jsmods_require(j, index):
return None
def get_emojisize_from_tags(tags):
if tags is None:
return None
tmp = [tag for tag in tags if tag.startswith("hot_emoji_size:")]
if len(tmp) > 0:
try:
return LIKES[tmp[0].split(":")[1]]
except (KeyError, IndexError):
log.exception(
"Could not determine emoji size from {} - {}".format(tags, tmp)
)
return None
def require_list(list_):
if isinstance(list_, list):
return set(list_)
@@ -355,19 +301,6 @@ def get_files_from_paths(filenames):
fp.close()
def enum_extend_if_invalid(enumeration, value):
try:
return enumeration(value)
except ValueError:
log.warning(
"Failed parsing {.__name__}({!r}). Extending enum.".format(
enumeration, value
)
)
aenum.extend_enum(enumeration, "UNKNOWN_{}".format(value).upper(), value)
return enumeration(value)
def get_url_parameters(url, *args):
params = parse_qs(urlparse(url).query)
return [params[arg][0] for arg in args if params.get(arg)]

1
fbchat/client.py
View File

@@ -4,5 +4,4 @@ from __future__ import unicode_literals
from .utils import *
from .models import *
from .graphql import *
from ._client import Client

31
fbchat/graphql.py
View File

@@ -1,31 +0,0 @@
# -*- coding: UTF-8 -*-
"""This file is here to maintain backwards compatability."""
from __future__ import unicode_literals
from .models import *
from .utils import *
from ._graphql import (
FLAGS,
WHITESPACE,
ConcatJSONDecoder,
graphql_color_to_enum,
get_customization_info,
graphql_to_sticker,
graphql_to_attachment,
graphql_to_extensible_attachment,
graphql_to_subattachment,
graphql_to_live_location,
graphql_to_poll,
graphql_to_poll_option,
graphql_to_plan,
graphql_to_quick_reply,
graphql_to_message,
graphql_to_message_reply,
graphql_to_user,
graphql_to_thread,
graphql_to_group,
graphql_to_page,
graphql_queries_to_json,
graphql_response_to_json,
GraphQL,
)

2
fbchat/models.py
View File

@@ -26,4 +26,4 @@ from ._quick_reply import (
QuickReplyEmail,
)
from ._poll import Poll, PollOption
from ._plan import Plan
from ._plan import GuestStatus, Plan

4
fbchat/utils.py
View File

@@ -7,8 +7,6 @@ from ._util import (
log,
handler,
USER_AGENTS,
LIKES,
GENDERS,
ReqUrl,
facebookEncoding,
now,
@@ -25,12 +23,10 @@ from ._util import (
check_json,
check_request,
get_jsmods_require,
get_emojisize_from_tags,
require_list,
mimetype_to_key,
get_files_from_urls,
get_files_from_paths,
enum_extend_if_invalid,
get_url_parameters,
get_url_parameter,
)