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

Merge pull request #489 from carpedm20/model-changes

Browse Source 
Model changes
This commit is contained in:
Mads Marquart
2019-12-11 16:32:28 +01:00
committed by GitHub
parent 074c271fb8 d1fbf0ba0a
commit 6c82e4d966
29 changed files with 690 additions and 725 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
docs/intro.rst
View File

@@ -13,8 +13,8 @@ You should also make sure that the file's access control is appropriately restri
Logging In
----------
Simply create an instance of :class:`Client`. If you have two factor authentication enabled, type the code in the terminal prompt
(If you want to supply the code in another fashion, overwrite :func:`Client.on_2fa_code`)::
Simply create an instance of `Client`. If you have two factor authentication enabled, type the code in the terminal prompt
(If you want to supply the code in another fashion, overwrite `Client.on_2fa_code`)::
from fbchat import Client
from fbchat.models import *
@@ -26,15 +26,15 @@ Replace ``<email>`` and ``<password>`` with your email and password respectively
For ease of use then most of the code snippets in this document will assume you've already completed the login process
Though the second line, ``from fbchat.models import *``, is not strictly necessary here, later code snippets will assume you've done this
If you want to change how verbose ``fbchat`` is, change the logging level (in :class:`Client`)
If you want to change how verbose ``fbchat`` is, change the logging level (in `Client`)
Throughout your code, if you want to check whether you are still logged in, use :func:`Client.is_logged_in`.
An example would be to login again if you've been logged out, using :func:`Client.login`::
Throughout your code, if you want to check whether you are still logged in, use `Client.is_logged_in`.
An example would be to login again if you've been logged out, using `Client.login`::
if not client.is_logged_in():
client.login('<email>', '<password>')
When you're done using the client, and want to securely logout, use :func:`Client.logout`::
When you're done using the client, and want to securely logout, use `Client.logout`::
client.logout()
@@ -46,14 +46,14 @@ Threads
A thread can refer to two things: A Messenger group chat or a single Facebook user
:class:`ThreadType` is an enumerator with two values: ``USER`` and ``GROUP``.
`ThreadType` is an enumerator with two values: ``USER`` and ``GROUP``.
These will specify whether the thread is a single user chat or a group chat.
This is required for many of ``fbchat``'s functions, since Facebook differentiates between these two internally
Searching for group chats and finding their ID can be done via. :func:`Client.search_for_groups`,
and searching for users is possible via. :func:`Client.search_for_users`. See :ref:`intro_fetching`
Searching for group chats and finding their ID can be done via. `Client.search_for_groups`,
and searching for users is possible via. `Client.search_for_users`. See :ref:`intro_fetching`
You can get your own user ID by using :any:`Client.uid`
You can get your own user ID by using `Client.uid`
Getting the ID of a group chat is fairly trivial otherwise, since you only need to navigate to `<https://www.facebook.com/messages/>`_,
click on the group you want to find the ID of, and then read the id from the address bar.
@@ -71,7 +71,7 @@ corresponds to the ID of a single user, and the ID of a group respectively::
client.send(Message(text='<message>'), thread_id='<user id>', thread_type=ThreadType.USER)
client.send(Message(text='<message>'), thread_id='<group id>', thread_type=ThreadType.GROUP)
Some functions (e.g. :func:`Client.change_thread_color`) don't require a thread type, so in these cases you just provide the thread ID::
Some functions (e.g. `Client.change_thread_color`) don't require a thread type, so in these cases you just provide the thread ID::
client.change_thread_color(ThreadColor.BILOBA_FLOWER, thread_id='<user id>')
client.change_thread_color(ThreadColor.MESSENGER_BLUE, thread_id='<group id>')
@@ -85,8 +85,8 @@ Message IDs
Every message you send on Facebook has a unique ID, and every action you do in a thread,
like changing a nickname or adding a person, has a unique ID too.
Some of ``fbchat``'s functions require these ID's, like :func:`Client.react_to_message`,
and some of then provide this ID, like :func:`Client.send`.
Some of ``fbchat``'s functions require these ID's, like `Client.react_to_message`,
and some of then provide this ID, like `Client.send`.
This snippet shows how to send a message, and then use the returned ID to react to that message with a 😍 emoji::
message_id = client.send(Message(text='message'), thread_id=thread_id, thread_type=thread_type)
@@ -118,7 +118,7 @@ Fetching Information
You can use ``fbchat`` to fetch basic information like user names, profile pictures, thread names and user IDs
You can retrieve a user's ID with :func:`Client.search_for_users`.
You can retrieve a user's ID with `Client.search_for_users`.
The following snippet will search for users by their name, take the first (and most likely) user, and then get their user ID from the result::
users = client.search_for_users('<name of user>')
@@ -140,11 +140,11 @@ Sessions
``fbchat`` provides functions to retrieve and set the session cookies.
This will enable you to store the session cookies in a separate file, so that you don't have to login each time you start your script.
Use :func:`Client.get_gession` to retrieve the cookies::
Use `Client.get_gession` to retrieve the cookies::
session_cookies = client.get_gession()
Then you can use :func:`Client.set_gession`::
Then you can use `Client.set_gession`::
client.set_gession(session_cookies)
@@ -162,7 +162,7 @@ Or you can set the ``session_cookies`` on your initial login.
Listening & Events
------------------
To use the listening functions ``fbchat`` offers (like :func:`Client.listen`),
To use the listening functions ``fbchat`` offers (like `Client.listen`),
you have to define what should be executed when certain events happen.
By default, (most) events will just be a `logging.info` statement,
meaning it will simply print information to the console when an event happens
@@ -170,7 +170,7 @@ meaning it will simply print information to the console when an event happens
.. note::
You can identify the event methods by their ``on`` prefix, e.g. ``on_message``
The event actions can be changed by subclassing the :class:`Client`, and then overwriting the event methods::
The event actions can be changed by subclassing the `Client`, and then overwriting the event methods::
class CustomClient(Client):
def on_message(self, mid, author_id, message_object, thread_id, thread_type, ts, metadata, msg, **kwargs):

1
fbchat/__init__.py
View File

@@ -11,6 +11,7 @@ _logging.getLogger(__name__).addHandler(_logging.NullHandler())
# The order of these is somewhat significant, e.g. User has to be imported after Thread!
from . import _core, _util
from ._core import Image
from ._exception import FBchatException, FBchatFacebookError
from ._thread import ThreadType, ThreadLocation, ThreadColor, Thread
from ._user import TypingStatus, User, ActiveStatus

47
fbchat/_attachment.py
View File

@@ -1,8 +1,9 @@
import attr
from ._core import attrs_default, Image
from . import _util
@attr.s
@attrs_default
class Attachment:
"""Represents a Facebook attachment."""
@@ -10,12 +11,12 @@ class Attachment:
uid = attr.ib(None)
@attr.s
@attrs_default
class UnsentMessage(Attachment):
"""Represents an unsent message attachment."""
@attr.s
@attrs_default
class ShareAttachment(Attachment):
"""Represents a shared item (e.g. URL) attachment."""
@@ -31,26 +32,30 @@ class ShareAttachment(Attachment):
description = attr.ib(None)
#: Name of the source
source = attr.ib(None)
#: URL of the attachment image
image_url = attr.ib(None)
#: The attached image
image = attr.ib(None)
#: URL of the original image if Facebook uses ``safe_image``
original_image_url = attr.ib(None)
#: Width of the image
image_width = attr.ib(None)
#: Height of the image
image_height = attr.ib(None)
#: List of additional attachments
attachments = attr.ib(factory=list, converter=lambda x: [] if x is None else x)
# Put here for backwards compatibility, so that the init argument order is preserved
uid = attr.ib(None)
attachments = attr.ib(factory=list)
@classmethod
def _from_graphql(cls, data):
from . import _file
image = None
original_image_url = None
media = data.get("media")
if media and media.get("image"):
image = Image._from_uri(media["image"])
original_image_url = (
_util.get_url_parameter(image.url, "url")
if "/safe_image.php" in image.url
else image.url
)
url = data.get("url")
rtn = cls(
return cls(
uid=data.get("deduplication_key"),
author=data["target"]["actors"][0]["id"]
if data["target"].get("actors")
@@ -64,20 +69,10 @@ class ShareAttachment(Attachment):
if data.get("description")
else None,
source=data["source"].get("text") if data.get("source") else None,
image=image,
original_image_url=original_image_url,
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

114
fbchat/_client.py
View File

@@ -38,9 +38,9 @@ ACONTEXT = {
class Client:
"""A client for the Facebook Chat (Messenger).
This is the main class of ``fbchat``, which contains all the methods you use to
interact with Facebook. You can extend this class, and overwrite the ``on`` methods,
to provide custom event handling (mainly useful while listening).
This is the main class, which contains all the methods you use to interact with
Facebook. You can extend this class, and overwrite the ``on`` methods, to provide
custom event handling (mainly useful while listening).
"""
@property
@@ -215,7 +215,7 @@ class Client:
limit: The max. amount of threads to fetch (default all threads)
Returns:
list: :class:`Thread` objects
list: `Thread` objects
Raises:
FBchatException: If request failed
@@ -266,7 +266,7 @@ class Client:
threads: Thread: List of threads to check for users
Returns:
list: :class:`User` objects
list: `User` objects
Raises:
FBchatException: If request failed
@@ -292,7 +292,7 @@ class Client:
"""Fetch all users the client is currently chatting with.
Returns:
list: :class:`User` objects
list: `User` objects
Raises:
FBchatException: If request failed
@@ -317,7 +317,7 @@ class Client:
limit: The max. amount of users to fetch
Returns:
list: :class:`User` objects, ordered by relevance
list: `User` objects, ordered by relevance
Raises:
FBchatException: If request failed
@@ -334,7 +334,7 @@ class Client:
name: Name of the page
Returns:
list: :class:`Page` objects, ordered by relevance
list: `Page` objects, ordered by relevance
Raises:
FBchatException: If request failed
@@ -352,7 +352,7 @@ class Client:
limit: The max. amount of groups to fetch
Returns:
list: :class:`Group` objects, ordered by relevance
list: `Group` objects, ordered by relevance
Raises:
FBchatException: If request failed
@@ -370,7 +370,7 @@ class Client:
limit: The max. amount of groups to fetch
Returns:
list: :class:`User`, :class:`Group` and :class:`Page` objects, ordered by relevance
list: `User`, `Group` and `Page` objects, ordered by relevance
Raises:
FBchatException: If request failed
@@ -441,7 +441,7 @@ class Client:
thread_id: User/Group ID to search in. See :ref:`intro_threads`
Returns:
typing.Iterable: Found :class:`Message` objects
typing.Iterable: Found `Message` objects
Raises:
FBchatException: If request failed
@@ -457,7 +457,7 @@ class Client:
Args:
query: Text to search for
fetch_messages: Whether to fetch :class:`Message` objects or IDs only
fetch_messages: Whether to fetch `Message` objects or IDs only
thread_limit (int): Max. number of threads to retrieve
message_limit (int): Max. number of messages to retrieve
@@ -531,7 +531,7 @@ class Client:
user_ids: One or more user ID(s) to query
Returns:
dict: :class:`User` objects, labeled by their ID
dict: `User` objects, labeled by their ID
Raises:
FBchatException: If request failed
@@ -556,7 +556,7 @@ class Client:
page_ids: One or more page ID(s) to query
Returns:
dict: :class:`Page` objects, labeled by their ID
dict: `Page` objects, labeled by their ID
Raises:
FBchatException: If request failed
@@ -578,7 +578,7 @@ class Client:
group_ids: One or more group ID(s) to query
Returns:
dict: :class:`Group` objects, labeled by their ID
dict: `Group` objects, labeled by their ID
Raises:
FBchatException: If request failed
@@ -603,7 +603,7 @@ class Client:
thread_ids: One or more thread ID(s) to query
Returns:
dict: :class:`Thread` objects, labeled by their ID
dict: `Thread` objects, labeled by their ID
Raises:
FBchatException: If request failed
@@ -669,7 +669,7 @@ class Client:
before (datetime.datetime): The point from which to retrieve messages
Returns:
list: :class:`Message` objects
list: `Message` objects
Raises:
FBchatException: If request failed
@@ -686,22 +686,14 @@ class Client:
if j.get("message_thread") is None:
raise FBchatException("Could not fetch thread {}: {}".format(thread_id, j))
read_receipts = j["message_thread"]["read_receipts"]["nodes"]
messages = [
Message._from_graphql(message)
Message._from_graphql(message, read_receipts)
for message in j["message_thread"]["messages"]["nodes"]
]
messages.reverse()
read_receipts = j["message_thread"]["read_receipts"]["nodes"]
for message in messages:
for receipt in read_receipts:
if (
_util.millis_to_datetime(int(receipt["watermark"]))
>= message.created_at
):
message.read_by.append(receipt["actor"]["id"])
return messages
def fetch_thread_list(
@@ -715,7 +707,7 @@ class Client:
before (datetime.datetime): The point from which to retrieve threads
Returns:
list: :class:`Thread` objects
list: `Thread` objects
Raises:
FBchatException: If request failed
@@ -814,7 +806,7 @@ class Client:
thread_id: User/Group ID to get message info from. See :ref:`intro_threads`
Returns:
Message: :class:`Message` object
Message: `Message` object
Raises:
FBchatException: If request failed
@@ -845,7 +837,7 @@ class Client:
plan_id: Plan ID to fetch from
Returns:
Plan: :class:`Plan` object
Plan: `Plan` object
Raises:
FBchatException: If request failed
@@ -901,7 +893,7 @@ class Client:
thread_id: ID of the thread
Returns:
typing.Iterable: :class:`ImageAttachment` or :class:`VideoAttachment`
typing.Iterable: `ImageAttachment` or `VideoAttachment`
"""
data = {"id": thread_id, "first": 48}
thread_id = str(thread_id)
@@ -964,7 +956,7 @@ class Client:
Raises:
FBchatException: If request failed
"""
thread = thread_type._to_class()(thread_id)
thread = thread_type._to_class()(uid=thread_id)
data = thread._to_send_data()
data.update(message._to_send_data())
return self._do_send_request(data)
@@ -983,7 +975,7 @@ class Client:
Raises:
FBchatException: If request failed
"""
thread = thread_type._to_class()(thread_id)
thread = thread_type._to_class()(uid=thread_id)
data = thread._to_send_data()
data["action_type"] = "ma-type:user-generated-message"
data["lightweight_action_attachment[lwa_state]"] = (
@@ -1009,31 +1001,40 @@ class Client:
Raises:
FBchatException: If request failed
"""
quick_reply.is_response = True
if isinstance(quick_reply, QuickReplyText):
return self.send(
Message(text=quick_reply.title, quick_replies=[quick_reply])
new = QuickReplyText(
payload=quick_reply.payload,
external_payload=quick_reply.external_payload,
data=quick_reply.data,
is_response=True,
title=quick_reply.title,
image_url=quick_reply.image_url,
)
return self.send(Message(text=quick_reply.title, quick_replies=[new]))
elif isinstance(quick_reply, QuickReplyLocation):
if not isinstance(payload, LocationAttachment):
raise TypeError(
"Payload must be an instance of `fbchat.LocationAttachment`"
)
raise TypeError("Payload must be an instance of `LocationAttachment`")
return self.send_location(
payload, thread_id=thread_id, thread_type=thread_type
)
elif isinstance(quick_reply, QuickReplyEmail):
if not payload:
payload = self.get_emails()[0]
quick_reply.external_payload = quick_reply.payload
quick_reply.payload = payload
return self.send(Message(text=payload, quick_replies=[quick_reply]))
new = QuickReplyEmail(
payload=payload if payload else self.get_emails()[0],
external_payload=quick_reply.payload,
data=quick_reply.data,
is_response=True,
image_url=quick_reply.image_url,
)
return self.send(Message(text=payload, quick_replies=[new]))
elif isinstance(quick_reply, QuickReplyPhoneNumber):
if not payload:
payload = self.get_phone_numbers()[0]
quick_reply.external_payload = quick_reply.payload
quick_reply.payload = payload
return self.send(Message(text=payload, quick_replies=[quick_reply]))
new = QuickReplyPhoneNumber(
payload=payload if payload else self.get_phone_numbers()[0],
external_payload=quick_reply.payload,
data=quick_reply.data,
is_response=True,
image_url=quick_reply.image_url,
)
return self.send(Message(text=payload, quick_replies=[new]))
def unsend(self, mid):
"""Unsend message by it's ID (removes it for everyone).
@@ -1047,7 +1048,7 @@ class Client:
def _send_location(
self, location, current=True, message=None, thread_id=None, thread_type=None
):
thread = thread_type._to_class()(thread_id)
thread = thread_type._to_class()(uid=thread_id)
data = thread._to_send_data()
if message is not None:
data.update(message._to_send_data())
@@ -1115,7 +1116,7 @@ class Client:
`files` should be a list of tuples, with a file's ID and mimetype.
"""
thread = thread_type._to_class()(thread_id)
thread = thread_type._to_class()(uid=thread_id)
data = thread._to_send_data()
data.update(self._old_message(message)._to_send_data())
data["action_type"] = "ma-type:user-generated-message"
@@ -1281,7 +1282,7 @@ class Client:
Raises:
FBchatException: If request failed
"""
data = Group(thread_id)._to_send_data()
data = Group(uid=thread_id)._to_send_data()
data["action_type"] = "ma-type:log-message"
data["log_message_type"] = "log:subscribe"
@@ -2533,9 +2534,8 @@ class Client:
i = d["deltaMessageReply"]
metadata = i["message"]["messageMetadata"]
thread_id, thread_type = get_thread_id_and_thread_type(metadata)
message = Message._from_reply(i["message"])
message.replied_to = Message._from_reply(i["repliedToMessage"])
message.reply_to_id = message.replied_to.uid
replied_to = Message._from_reply(i["repliedToMessage"])
message = Message._from_reply(i["message"], replied_to)
self.on_message(
mid=message.uid,
author_id=message.author,
@@ -3653,7 +3653,7 @@ class Client:
"""Called when the client is listening and client receives information about friend active status.
Args:
statuses (dict): Dictionary with user IDs as keys and :class:`ActiveStatus` as values
statuses (dict): Dictionary with user IDs as keys and `ActiveStatus` as values
msg: A full set of the data received
"""
log.debug("Buddylist overlay received: {}".format(statuses))

53
fbchat/_core.py
View File

@@ -1,11 +1,19 @@
import sys
import attr
import logging
import aenum
log = logging.getLogger("fbchat")
# Enable kw_only if the python version supports it
kw_only = sys.version_info[:2] > (3, 5)
#: Default attrs settings for classes
attrs_default = attr.s(slots=True, kw_only=kw_only)
class Enum(aenum.Enum):
"""Used internally by ``fbchat`` to support enumerations"""
"""Used internally to support enumerations"""
def __repr__(self):
# For documentation:
@@ -21,3 +29,46 @@ class Enum(aenum.Enum):
)
aenum.extend_enum(cls, "UNKNOWN_{}".format(value).upper(), value)
return cls(value)
# Frozen, so that it can be used in sets
@attr.s(frozen=True, slots=True, kw_only=kw_only)
class Image:
#: URL to the image
url = attr.ib(type=str)
#: Width of the image
width = attr.ib(None, type=int)
#: Height of the image
height = attr.ib(None, type=int)
@classmethod
def _from_uri(cls, data):
return cls(
url=data["uri"],
width=int(data["width"]) if data.get("width") else None,
height=int(data["height"]) if data.get("height") else None,
)
@classmethod
def _from_url(cls, data):
return cls(
url=data["url"],
width=int(data["width"]) if data.get("width") else None,
height=int(data["height"]) if data.get("height") else None,
)
@classmethod
def _from_uri_or_none(cls, data):
if data is None:
return None
if data.get("uri") is None:
return None
return cls._from_uri(data)
@classmethod
def _from_url_or_none(cls, data):
if data is None:
return None
if data.get("url") is None:
return None
return cls._from_url(data)

42
fbchat/_exception.py
View File

@@ -1,32 +1,32 @@
import attr
# Not frozen, since that doesn't work in PyPy
attrs_exception = attr.s(slots=True, auto_exc=True)
@attrs_exception
class FBchatException(Exception):
"""Custom exception thrown by ``fbchat``.
All exceptions in the ``fbchat`` module inherits this.
All exceptions in the module inherits this.
"""
message = attr.ib()
@attrs_exception
class FBchatFacebookError(FBchatException):
"""Raised when Facebook returns an error."""
#: The error code that Facebook returned
fb_error_code = None
fb_error_code = attr.ib(None)
#: The error message that Facebook returned (In the user's own language)
fb_error_message = None
fb_error_message = attr.ib(None)
#: The status code that was sent in the HTTP response (e.g. 404) (Usually only set if not successful, aka. not 200)
request_status_code = None
def __init__(
self,
message,
fb_error_code=None,
fb_error_message=None,
request_status_code=None,
):
super(FBchatFacebookError, self).__init__(message)
"""Thrown by ``fbchat`` when Facebook returns an error"""
self.fb_error_code = str(fb_error_code)
self.fb_error_message = fb_error_message
self.request_status_code = request_status_code
request_status_code = attr.ib(None)
@attrs_exception
class FBchatInvalidParameters(FBchatFacebookError):
"""Raised by Facebook if:
@@ -36,17 +36,19 @@ class FBchatInvalidParameters(FBchatFacebookError):
"""
@attrs_exception
class FBchatNotLoggedIn(FBchatFacebookError):
"""Raised by Facebook if the client has been logged out."""
fb_error_code = "1357001"
fb_error_code = attr.ib("1357001")
@attrs_exception
class FBchatPleaseRefresh(FBchatFacebookError):
"""Raised by Facebook if the client has been inactive for too long.
This error usually happens after 1-2 days of inactivity.
"""
fb_error_code = "1357004"
fb_error_message = "Please try closing and re-opening your browser window."
fb_error_code = attr.ib("1357004")
fb_error_message = attr.ib("Please try closing and re-opening your browser window.")

202
fbchat/_file.py
View File

@@ -1,9 +1,10 @@
import attr
from ._core import attrs_default, Image
from . import _util
from ._attachment import Attachment
@attr.s
@attrs_default
class FileAttachment(Attachment):
"""Represents a file that has been sent as a Facebook attachment."""
@@ -16,20 +17,18 @@ class FileAttachment(Attachment):
#: Whether Facebook determines that this file may be harmful
is_malicious = attr.ib(None)
# Put here for backwards compatibility, so that the init argument order is preserved
uid = attr.ib(None)
@classmethod
def _from_graphql(cls, data):
def _from_graphql(cls, data, size=None):
return cls(
url=data.get("url"),
size=size,
name=data.get("filename"),
is_malicious=data.get("is_malicious"),
uid=data.get("message_file_fbid"),
)
@attr.s
@attrs_default
class AudioAttachment(Attachment):
"""Represents an audio file that has been sent as a Facebook attachment."""
@@ -42,9 +41,6 @@ class AudioAttachment(Attachment):
#: Audio type
audio_type = attr.ib(None)
# 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(
@@ -55,7 +51,7 @@ class AudioAttachment(Attachment):
)
@attr.s(init=False)
@attrs_default
class ImageAttachment(Attachment):
"""Represents an image that has been sent as a Facebook attachment.
@@ -69,104 +65,49 @@ class ImageAttachment(Attachment):
width = attr.ib(None, converter=lambda x: None if x is None else int(x))
#: Height of original image
height = attr.ib(None, converter=lambda x: None if x is None else int(x))
#: Whether the image is animated
is_animated = attr.ib(None)
#: URL to a thumbnail of the image
thumbnail_url = attr.ib(None)
#: URL to a medium preview of the image
preview_url = attr.ib(None)
#: Width of the medium preview image
preview_width = attr.ib(None)
#: Height of the medium preview image
preview_height = attr.ib(None)
#: URL to a large preview of the image
large_preview_url = attr.ib(None)
#: Width of the large preview image
large_preview_width = attr.ib(None)
#: Height of the large preview image
large_preview_height = attr.ib(None)
#: URL to an animated preview of the image (e.g. for GIFs)
animated_preview_url = attr.ib(None)
#: Width of the animated preview image
animated_preview_width = attr.ib(None)
#: Height of the animated preview image
animated_preview_height = attr.ib(None)
def __init__(
self,
original_extension=None,
width=None,
height=None,
is_animated=None,
thumbnail_url=None,
preview=None,
large_preview=None,
animated_preview=None,
**kwargs
):
super(ImageAttachment, self).__init__(**kwargs)
self.original_extension = original_extension
if width is not None:
width = int(width)
self.width = width
if height is not None:
height = int(height)
self.height = height
self.is_animated = is_animated
self.thumbnail_url = thumbnail_url
if preview is None:
preview = {}
self.preview_url = preview.get("uri")
self.preview_width = preview.get("width")
self.preview_height = preview.get("height")
if large_preview is None:
large_preview = {}
self.large_preview_url = large_preview.get("uri")
self.large_preview_width = large_preview.get("width")
self.large_preview_height = large_preview.get("height")
if animated_preview is None:
animated_preview = {}
self.animated_preview_url = animated_preview.get("uri")
self.animated_preview_width = animated_preview.get("width")
self.animated_preview_height = animated_preview.get("height")
#: A set, containing variously sized / various types of previews of the image
previews = attr.ib(factory=set)
@classmethod
def _from_graphql(cls, data):
previews = {
Image._from_uri_or_none(data.get("thumbnail")),
Image._from_uri_or_none(data.get("preview") or data.get("preview_image")),
Image._from_uri_or_none(data.get("large_preview")),
Image._from_uri_or_none(data.get("animated_image")),
}
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"),
previews={p for p in previews if p},
uid=data.get("legacy_attachment_id"),
)
@classmethod
def _from_list(cls, data):
data = data["node"]
previews = {
Image._from_uri_or_none(data["image"]),
Image._from_uri(data["image1"]),
Image._from_uri(data["image2"]),
}
return cls(
width=data["original_dimensions"].get("x"),
height=data["original_dimensions"].get("y"),
thumbnail_url=data["image"].get("uri"),
large_preview=data["image2"],
preview=data["image1"],
previews={p for p in previews if p},
uid=data["legacy_attachment_id"],
)
@attr.s(init=False)
@attrs_default
class VideoAttachment(Attachment):
"""Represents a video that has been sent as a Facebook attachment."""
@@ -180,111 +121,66 @@ class VideoAttachment(Attachment):
duration = attr.ib(None)
#: URL to very compressed preview video
preview_url = attr.ib(None)
#: URL to a small preview image of the video
small_image_url = attr.ib(None)
#: Width of the small preview image
small_image_width = attr.ib(None)
#: Height of the small preview image
small_image_height = attr.ib(None)
#: URL to a medium preview image of the video
medium_image_url = attr.ib(None)
#: Width of the medium preview image
medium_image_width = attr.ib(None)
#: Height of the medium preview image
medium_image_height = attr.ib(None)
#: URL to a large preview image of the video
large_image_url = attr.ib(None)
#: Width of the large preview image
large_image_width = attr.ib(None)
#: Height of the large preview image
large_image_height = attr.ib(None)
def __init__(
self,
size=None,
width=None,
height=None,
duration=None,
preview_url=None,
small_image=None,
medium_image=None,
large_image=None,
**kwargs
):
super(VideoAttachment, self).__init__(**kwargs)
self.size = size
self.width = width
self.height = height
self.duration = duration
self.preview_url = preview_url
if small_image is None:
small_image = {}
self.small_image_url = small_image.get("uri")
self.small_image_width = small_image.get("width")
self.small_image_height = small_image.get("height")
if medium_image is None:
medium_image = {}
self.medium_image_url = medium_image.get("uri")
self.medium_image_width = medium_image.get("width")
self.medium_image_height = medium_image.get("height")
if large_image is None:
large_image = {}
self.large_image_url = large_image.get("uri")
self.large_image_width = large_image.get("width")
self.large_image_height = large_image.get("height")
#: A set, containing variously sized previews of the video
previews = attr.ib(factory=set)
@classmethod
def _from_graphql(cls, data):
def _from_graphql(cls, data, size=None):
previews = {
Image._from_uri_or_none(data.get("chat_image")),
Image._from_uri_or_none(data.get("inbox_image")),
Image._from_uri_or_none(data.get("large_image")),
}
return cls(
size=size,
width=data.get("original_dimensions", {}).get("width"),
height=data.get("original_dimensions", {}).get("height"),
duration=_util.millis_to_timedelta(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"),
previews={p for p in previews if p},
uid=data.get("legacy_attachment_id"),
)
@classmethod
def _from_subattachment(cls, data):
media = data["media"]
image = Image._from_uri_or_none(media.get("image"))
return cls(
duration=_util.millis_to_timedelta(media.get("playable_duration_in_ms")),
preview_url=media.get("playable_url"),
medium_image=media.get("image"),
previews={image} if image else {},
uid=data["target"].get("video_id"),
)
@classmethod
def _from_list(cls, data):
data = data["node"]
previews = {
Image._from_uri(data["image"]),
Image._from_uri(data["image1"]),
Image._from_uri(data["image2"]),
}
return cls(
width=data["original_dimensions"].get("x"),
height=data["original_dimensions"].get("y"),
small_image=data["image"],
medium_image=data["image1"],
large_image=data["image2"],
previews=previews,
uid=data["legacy_attachment_id"],
)
def graphql_to_attachment(data):
def graphql_to_attachment(data, size=None):
_type = data["__typename"]
if _type in ["MessageImage", "MessageAnimatedImage"]:
return ImageAttachment._from_graphql(data)
elif _type == "MessageVideo":
return VideoAttachment._from_graphql(data)
return VideoAttachment._from_graphql(data, size=size)
elif _type == "MessageAudio":
return AudioAttachment._from_graphql(data)
elif _type == "MessageFile":
return FileAttachment._from_graphql(data)
return FileAttachment._from_graphql(data, size=size)
return Attachment(uid=data.get("legacy_attachment_id"))

19
fbchat/_group.py
View File

@@ -1,30 +1,29 @@
import attr
from ._core import attrs_default, Image
from . import _util, _plan
from ._thread import ThreadType, Thread
@attr.s
@attrs_default
class Group(Thread):
"""Represents a Facebook group. Inherits `Thread`."""
type = ThreadType.GROUP
#: Unique list (set) of the group thread's participant user IDs
participants = attr.ib(factory=set, converter=lambda x: set() if x is None else x)
participants = attr.ib(factory=set)
#: A dictionary, containing user nicknames mapped to their IDs
nicknames = attr.ib(factory=dict, converter=lambda x: {} if x is None else x)
#: A :class:`ThreadColor`. The groups's message color
nicknames = attr.ib(factory=dict)
#: A `ThreadColor`. The groups's message color
color = attr.ib(None)
#: The groups's default emoji
emoji = attr.ib(None)
# Set containing user IDs of thread admins
admins = attr.ib(factory=set, converter=lambda x: set() if x is None else x)
admins = attr.ib(factory=set)
# True if users need approval to join
approval_mode = attr.ib(None)
# Set containing user IDs requesting to join
approval_requests = attr.ib(
factory=set, converter=lambda x: set() if x is None else x
)
approval_requests = attr.ib(factory=set)
# Link for joining group
join_link = attr.ib(None)
@@ -43,7 +42,7 @@ class Group(Thread):
plan = _plan.Plan._from_graphql(data["event_reminders"]["nodes"][0])
return cls(
data["thread_key"]["thread_fbid"],
uid=data["thread_key"]["thread_fbid"],
participants=set(
[
node["messaging_actor"]["id"]
@@ -64,7 +63,7 @@ class Group(Thread):
if data.get("group_approval_queue")
else None,
join_link=data["joinable_mode"].get("link"),
photo=data["image"].get("uri"),
photo=Image._from_uri_or_none(data["image"]),
name=data.get("name"),
message_count=data.get("messages_count"),
last_active=last_active,

51
fbchat/_location.py
View File

@@ -1,9 +1,10 @@
import attr
from ._core import attrs_default, Image
from ._attachment import Attachment
from . import _util
@attr.s
@attrs_default
class LocationAttachment(Attachment):
"""Represents a user location.
@@ -14,20 +15,13 @@ class LocationAttachment(Attachment):
latitude = attr.ib(None)
#: Longitude of the location
longitude = attr.ib(None)
#: URL of image showing the map of the location
image_url = attr.ib(None, init=False)
#: Width of the image
image_width = attr.ib(None, init=False)
#: Height of the image
image_height = attr.ib(None, init=False)
#: Image showing the map of the location
image = attr.ib(None)
#: URL to Bing maps with the location
url = attr.ib(None, init=False)
url = attr.ib(None)
# Address of the location
address = attr.ib(None)
# 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")
@@ -37,23 +31,20 @@ class LocationAttachment(Attachment):
address = None
except ValueError:
latitude, longitude = None, None
rtn = cls(
return cls(
uid=int(data["deduplication_key"]),
latitude=latitude,
longitude=longitude,
image=Image._from_uri_or_none(data["media"].get("image"))
if data.get("media")
else None,
url=url,
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
@attrs_default
class LiveLocationAttachment(LocationAttachment):
"""Represents a live user location."""
@@ -82,7 +73,13 @@ class LiveLocationAttachment(LocationAttachment):
@classmethod
def _from_graphql(cls, data):
target = data["target"]
rtn = cls(
image = None
media = data.get("media")
if media and media.get("image"):
image = Image._from_uri(media["image"])
return cls(
uid=int(target["live_location_id"]),
latitude=target["coordinate"]["latitude"]
if target.get("coordinate")
@@ -90,15 +87,9 @@ class LiveLocationAttachment(LocationAttachment):
longitude=target["coordinate"]["longitude"]
if target.get("coordinate")
else None,
image=image,
url=data.get("url"),
name=data["title_with_entities"]["text"],
expires_at=_util.seconds_to_datetime(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

280
fbchat/_message.py
View File

@@ -1,7 +1,7 @@
import attr
import json
from string import Formatter
from ._core import log, Enum
from ._core import log, attrs_default, Enum
from . import _util, _attachment, _location, _file, _quick_reply, _sticker
@@ -42,7 +42,7 @@ class MessageReaction(Enum):
NO = "👎"
@attr.s
@attrs_default
class Mention:
"""Represents a ``@mention``."""
@@ -53,43 +53,63 @@ class Mention:
#: The length of the mention
length = attr.ib(10)
@classmethod
def _from_range(cls, data):
return cls(
thread_id=data.get("entity", {}).get("id"),
offset=data.get("offset"),
length=data.get("length"),
)
@attr.s
@classmethod
def _from_prng(cls, data):
return cls(thread_id=data.get("i"), offset=data.get("o"), length=data.get("l"))
def _to_send_data(self, i):
return {
"profile_xmd[{}][id]".format(i): self.thread_id,
"profile_xmd[{}][offset]".format(i): self.offset,
"profile_xmd[{}][length]".format(i): self.length,
"profile_xmd[{}][type]".format(i): "p",
}
@attrs_default
class Message:
"""Represents a Facebook message."""
#: The actual message
text = attr.ib(None)
#: A list of :class:`Mention` objects
mentions = attr.ib(factory=list, converter=lambda x: [] if x is None else x)
#: A :class:`EmojiSize`. Size of a sent emoji
#: A list of `Mention` objects
mentions = attr.ib(factory=list)
#: A `EmojiSize`. Size of a sent emoji
emoji_size = attr.ib(None)
#: The message ID
uid = attr.ib(None, init=False)
uid = attr.ib(None)
#: ID of the sender
author = attr.ib(None, init=False)
author = attr.ib(None)
#: Datetime of when the message was sent
created_at = attr.ib(None, init=False)
created_at = attr.ib(None)
#: Whether the message is read
is_read = attr.ib(None, init=False)
#: A list of people IDs who read the message, works only with :func:`fbchat.Client.fetch_thread_messages`
read_by = attr.ib(factory=list, init=False)
#: A dictionary with user's IDs as keys, and their :class:`MessageReaction` as values
reactions = attr.ib(factory=dict, init=False)
#: A :class:`Sticker`
is_read = attr.ib(None)
#: A list of people IDs who read the message, works only with `Client.fetch_thread_messages`
read_by = attr.ib(factory=list)
#: A dictionary with user's IDs as keys, and their `MessageReaction` as values
reactions = attr.ib(factory=dict)
#: A `Sticker`
sticker = attr.ib(None)
#: A list of attachments
attachments = attr.ib(factory=list, converter=lambda x: [] if x is None else x)
#: A list of :class:`QuickReply`
quick_replies = attr.ib(factory=list, converter=lambda x: [] if x is None else x)
attachments = attr.ib(factory=list)
#: A list of `QuickReply`
quick_replies = attr.ib(factory=list)
#: Whether the message is unsent (deleted for everyone)
unsent = attr.ib(False, init=False)
unsent = attr.ib(False)
#: Message ID you want to reply to
reply_to_id = attr.ib(None)
#: Replied message
replied_to = attr.ib(None, init=False)
replied_to = attr.ib(None)
#: Whether the message was forwarded
forwarded = attr.ib(False, init=False)
forwarded = attr.ib(False)
@classmethod
def format_mentions(cls, text, *args, **kwargs):
@@ -139,8 +159,7 @@ class Message:
)
offset += len(name)
message = cls(text=result, mentions=mentions)
return message
return cls(text=result, mentions=mentions)
@staticmethod
def _get_forwarded_from_tags(tags):
@@ -158,10 +177,7 @@ class Message:
data["body"] = self.text
for i, mention in enumerate(self.mentions):
data["profile_xmd[{}][id]".format(i)] = mention.thread_id
data["profile_xmd[{}][offset]".format(i)] = mention.offset
data["profile_xmd[{}][length]".format(i)] = mention.length
data["profile_xmd[{}][type]".format(i)] = "p"
data.update(mention._to_send_data(i))
if self.emoji_size:
if self.text:
@@ -197,99 +213,82 @@ class Message:
return data
@staticmethod
def _parse_quick_replies(data):
if data:
data = json.loads(data).get("quick_replies")
if isinstance(data, list):
return [_quick_reply.graphql_to_quick_reply(q) for q in data]
elif isinstance(data, dict):
return [_quick_reply.graphql_to_quick_reply(data, is_response=True)]
return []
@classmethod
def _from_graphql(cls, data):
def _from_graphql(cls, data, read_receipts=None):
if data.get("message_sender") is None:
data["message_sender"] = {}
if data.get("message") is None:
data["message"] = {}
tags = data.get("tags_list")
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(tags),
sticker=_sticker.Sticker._from_graphql(data.get("sticker")),
)
rtn.forwarded = cls._get_forwarded_from_tags(tags)
rtn.uid = str(data["message_id"])
rtn.author = str(data["message_sender"]["id"])
rtn.created_at = _util.millis_to_datetime(int(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 = [
created_at = _util.millis_to_datetime(int(data.get("timestamp_precise")))
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)
for attachment in data["blob_attachments"] or ()
]
unsent = False
if data.get("extensible_attachment") is not None:
attachment = graphql_to_extensible_attachment(data["extensible_attachment"])
if isinstance(attachment, _attachment.UnsentMessage):
rtn.unsent = True
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
attachments.append(attachment)
@classmethod
def _from_reply(cls, data):
tags = data["messageMetadata"].get("tags")
rtn = cls(
text=data.get("body"),
replied_to = None
if data.get("replied_to_message"):
replied_to = cls._from_graphql(data["replied_to_message"]["message"])
return cls(
text=data["message"].get("text"),
mentions=[
Mention(m.get("i"), offset=m.get("o"), length=m.get("l"))
for m in json.loads(data.get("data", {}).get("prng", "[]"))
Mention._from_range(m) for m in data["message"].get("ranges") or ()
],
emoji_size=EmojiSize._from_tags(tags),
uid=str(data["message_id"]),
author=str(data["message_sender"]["id"]),
created_at=created_at,
is_read=not data["unread"] if data.get("unread") is not None else None,
read_by=[
receipt["actor"]["id"]
for receipt in read_receipts or ()
if _util.millis_to_datetime(int(receipt["watermark"])) >= created_at
],
reactions={
str(r["user"]["id"]): MessageReaction._extend_if_invalid(r["reaction"])
for r in data["message_reactions"]
},
sticker=_sticker.Sticker._from_graphql(data.get("sticker")),
attachments=attachments,
quick_replies=cls._parse_quick_replies(data.get("platform_xmd_encoded")),
unsent=unsent,
reply_to_id=replied_to.uid if replied_to else None,
replied_to=replied_to,
forwarded=cls._get_forwarded_from_tags(tags),
)
@classmethod
def _from_reply(cls, data, replied_to=None):
tags = data["messageMetadata"].get("tags")
metadata = data.get("messageMetadata", {})
rtn.forwarded = cls._get_forwarded_from_tags(tags)
rtn.uid = metadata.get("messageId")
rtn.author = str(metadata.get("actorFbId"))
rtn.created_at = _util.millis_to_datetime(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"]:
attachments = []
unsent = False
sticker = None
for attachment in data.get("attachments") or ():
attachment = json.loads(attachment["mercuryJSON"])
if attachment.get("blob_attachment"):
rtn.attachments.append(
attachments.append(
_file.graphql_to_attachment(attachment["blob_attachment"])
)
if attachment.get("extensible_attachment"):
@@ -297,57 +296,61 @@ class Message:
attachment["extensible_attachment"]
)
if isinstance(extensible_attachment, _attachment.UnsentMessage):
rtn.unsent = True
unsent = True
else:
rtn.attachments.append(extensible_attachment)
attachments.append(extensible_attachment)
if attachment.get("sticker_attachment"):
rtn.sticker = _sticker.Sticker._from_graphql(
sticker = _sticker.Sticker._from_graphql(
attachment["sticker_attachment"]
)
return rtn
return cls(
text=data.get("body"),
mentions=[
Mention._from_prng(m)
for m in _util.parse_json(data.get("data", {}).get("prng", "[]"))
],
emoji_size=EmojiSize._from_tags(tags),
uid=metadata.get("messageId"),
author=str(metadata.get("actorFbId")),
created_at=_util.millis_to_datetime(metadata.get("timestamp")),
sticker=sticker,
attachments=attachments,
quick_replies=cls._parse_quick_replies(data.get("platform_xmd_encoded")),
unsent=unsent,
reply_to_id=replied_to.uid if replied_to else None,
replied_to=replied_to,
forwarded=cls._get_forwarded_from_tags(tags),
)
@classmethod
def _from_pull(cls, data, mid=None, tags=None, author=None, created_at=None):
rtn = cls(text=data.get("body"))
rtn.uid = mid
rtn.author = author
rtn.created_at = created_at
mentions = []
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"])
mentions = [
Mention._from_prng(m)
for m in _util.parse_json(data["data"]["prng"])
]
except Exception:
log.exception("An exception occured while reading attachments")
if data.get("attachments"):
attachments = []
unsent = False
sticker = None
try:
for a in data["attachments"]:
for a in data.get("attachments") or ():
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"]
mercury["blob_attachment"], a["fileSize"]
)
if attach_type in [
"MessageFile",
"MessageVideo",
"MessageAudio",
]:
# TODO: Add more data here for audio files
attachment.size = int(a["fileSize"])
rtn.attachments.append(attachment)
attachments.append(attachment)
elif mercury.get("sticker_attachment"):
rtn.sticker = _sticker.Sticker._from_graphql(
sticker = _sticker.Sticker._from_graphql(
mercury["sticker_attachment"]
)
@@ -356,9 +359,9 @@ class Message:
mercury["extensible_attachment"]
)
if isinstance(attachment, _attachment.UnsentMessage):
rtn.unsent = True
unsent = True
elif attachment:
rtn.attachments.append(attachment)
attachments.append(attachment)
except Exception:
log.exception(
@@ -367,9 +370,18 @@ class Message:
)
)
rtn.emoji_size = EmojiSize._from_tags(tags)
rtn.forwarded = cls._get_forwarded_from_tags(tags)
return rtn
return cls(
text=data.get("body"),
mentions=mentions,
emoji_size=EmojiSize._from_tags(tags),
uid=mid,
author=author,
created_at=created_at,
sticker=sticker,
attachments=attachments,
unsent=unsent,
forwarded=cls._get_forwarded_from_tags(tags),
)
def graphql_to_extensible_attachment(data):

7
fbchat/_page.py
View File

@@ -1,9 +1,10 @@
import attr
from ._core import attrs_default, Image
from . import _plan
from ._thread import ThreadType, Thread
@attr.s
@attrs_default
class Page(Thread):
"""Represents a Facebook page. Inherits `Thread`."""
@@ -31,11 +32,11 @@ class Page(Thread):
plan = _plan.Plan._from_graphql(data["event_reminders"]["nodes"][0])
return cls(
data["id"],
uid=data["id"],
url=data.get("url"),
city=data.get("city").get("name"),
category=data.get("category_type"),
photo=data["profile_picture"].get("uri"),
photo=Image._from_uri(data["profile_picture"]),
name=data.get("name"),
message_count=data.get("messages_count"),
plan=plan,

47
fbchat/_plan.py
View File

@@ -1,6 +1,6 @@
import attr
import json
from ._core import Enum
from ._core import attrs_default, Enum
from . import _util
@@ -10,24 +10,24 @@ class GuestStatus(Enum):
DECLINED = 3
@attr.s
@attrs_default
class Plan:
"""Represents a plan."""
#: ID of the plan
uid = attr.ib(None, init=False)
#: Plan time (datetime), only precise down to the minute
time = attr.ib()
#: Plan title
title = attr.ib()
#: ID of the plan
uid = attr.ib(None)
#: Plan location name
location = attr.ib(None, converter=lambda x: x or "")
#: Plan location ID
location_id = attr.ib(None, converter=lambda x: x or "")
#: ID of the plan creator
author_id = attr.ib(None, init=False)
author_id = attr.ib(None)
#: Dictionary of `User` IDs mapped to their `GuestStatus`
guests = attr.ib(None, init=False)
guests = attr.ib(None)
@property
def going(self):
@@ -58,44 +58,41 @@ class Plan:
@classmethod
def _from_pull(cls, data):
rtn = cls(
return cls(
uid=data.get("event_id"),
time=_util.seconds_to_datetime(int(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 = {
author_id=data.get("event_creator_id"),
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(
return cls(
uid=data.get("oid"),
time=_util.seconds_to_datetime(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,
author_id=data.get("creator_id"),
guests={id_: GuestStatus[s] for id_, s in data["event_members"].items()},
)
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(
return cls(
uid=data.get("id"),
time=_util.seconds_to_datetime(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 = {
author_id=data["lightweight_event_creator"].get("id"),
guests={
x["node"]["id"]: GuestStatus[x["guest_list_state"]]
for x in data["event_reminder_members"]["edges"]
}
return rtn
},
)

7
fbchat/_poll.py
View File

@@ -1,13 +1,14 @@
import attr
from ._core import attrs_default
@attr.s
@attrs_default
class Poll:
"""Represents a poll."""
#: Title of the poll
title = attr.ib()
#: List of :class:`PollOption`, can be fetched with :func:`fbchat.Client.fetch_poll_options`
#: List of `PollOption`, can be fetched with `Client.fetch_poll_options`
options = attr.ib()
#: Options count
options_count = attr.ib(None)
@@ -24,7 +25,7 @@ class Poll:
)
@attr.s
@attrs_default
class PollOption:
"""Represents a poll option."""

13
fbchat/_quick_reply.py
View File

@@ -1,22 +1,23 @@
import attr
from ._core import attrs_default
from ._attachment import Attachment
@attr.s
@attrs_default
class QuickReply:
"""Represents a quick reply."""
#: Payload of the quick reply
payload = attr.ib(None)
#: External payload for responses
external_payload = attr.ib(None, init=False)
external_payload = attr.ib(None)
#: Additional data
data = attr.ib(None)
#: Whether it's a response for a quick reply
is_response = attr.ib(False)
@attr.s
@attrs_default
class QuickReplyText(QuickReply):
"""Represents a text quick reply."""
@@ -28,7 +29,7 @@ class QuickReplyText(QuickReply):
_type = "text"
@attr.s
@attrs_default
class QuickReplyLocation(QuickReply):
"""Represents a location quick reply (Doesn't work on mobile)."""
@@ -36,7 +37,7 @@ class QuickReplyLocation(QuickReply):
_type = "location"
@attr.s
@attrs_default
class QuickReplyPhoneNumber(QuickReply):
"""Represents a phone number quick reply (Doesn't work on mobile)."""
@@ -46,7 +47,7 @@ class QuickReplyPhoneNumber(QuickReply):
_type = "user_phone_number"
@attr.s
@attrs_default
class QuickReplyEmail(QuickReply):
"""Represents an email quick reply (Doesn't work on mobile)."""

4
fbchat/_state.py
View File

@@ -5,7 +5,7 @@ import requests
import random
import urllib.parse
from ._core import log
from ._core import log, attrs_default
from . import _graphql, _util, _exception
FB_DTSG_REGEX = re.compile(r'name="fb_dtsg" value="(.*?)"')
@@ -98,7 +98,7 @@ def _2fa_helper(session, code, r):
return r
@attr.s(slots=True) # TODO i Python 3: Add kw_only=True
@attrs_default
class State:
"""Stores and manages state required for most Facebook requests."""

44
fbchat/_sticker.py
View File

@@ -1,8 +1,9 @@
import attr
from ._core import attrs_default, Image
from ._attachment import Attachment
@attr.s
@attrs_default
class Sticker(Attachment):
"""Represents a Facebook sticker that has been sent to a thread as an attachment."""
@@ -23,12 +24,8 @@ class Sticker(Attachment):
#: The frame rate the spritemap is intended to be played in
frame_rate = attr.ib(None)
#: URL to the sticker's image
url = attr.ib(None)
#: Width of the sticker
width = attr.ib(None)
#: Height of the sticker
height = attr.ib(None)
#: The sticker's image
image = attr.ib(None)
#: The sticker's label/name
label = attr.ib(None)
@@ -36,19 +33,20 @@ class Sticker(Attachment):
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
return cls(
uid=data["id"],
pack=data["pack"].get("id") if data.get("pack") else None,
is_animated=bool(data.get("sprite_image")),
medium_sprite_image=data["sprite_image"].get("uri")
if data.get("sprite_image")
else None,
large_sprite_image=data["sprite_image_2x"].get("uri")
if data.get("sprite_image_2x")
else None,
frames_per_row=data.get("frames_per_row"),
frames_per_col=data.get("frames_per_column"),
frame_rate=data.get("frame_rate"),
image=Image._from_url_or_none(data),
label=data["label"] if data.get("label") else None,
)

8
fbchat/_thread.py
View File

@@ -1,5 +1,5 @@
import attr
from ._core import Enum
from ._core import attrs_default, Enum, Image
class ThreadType(Enum):
@@ -67,7 +67,7 @@ class ThreadColor(Enum):
return cls._extend_if_invalid(value)
@attr.s
@attrs_default
class Thread:
"""Represents a Facebook thread."""
@@ -75,7 +75,7 @@ class Thread:
uid = attr.ib(converter=str)
#: Specifies the type of thread. Can be used a ``thread_type``. See :ref:`intro_threads` for more info
type = None
#: A URL to the thread's picture
#: The thread's picture
photo = attr.ib(None)
#: The name of the thread
name = attr.ib(None)
@@ -83,7 +83,7 @@ class Thread:
last_active = attr.ib(None)
#: Number of messages in the thread
message_count = attr.ib(None)
#: Set :class:`Plan`
#: Set `Plan`
plan = attr.ib(None)
@staticmethod

18
fbchat/_user.py
View File

@@ -1,5 +1,5 @@
import attr
from ._core import Enum
from ._core import attrs_default, Enum, Image
from . import _util, _plan
from ._thread import ThreadType, Thread
@@ -41,7 +41,7 @@ class TypingStatus(Enum):
TYPING = 1
@attr.s
@attrs_default
class User(Thread):
"""Represents a Facebook user. Inherits `Thread`."""
@@ -63,7 +63,7 @@ class User(Thread):
nickname = attr.ib(None)
#: The clients nickname, as seen by the user
own_nickname = attr.ib(None)
#: A :class:`ThreadColor`. The message color
#: A `ThreadColor`. The message color
color = attr.ib(None)
#: The default emoji
emoji = attr.ib(None)
@@ -78,7 +78,7 @@ class User(Thread):
plan = _plan.Plan._from_graphql(data["event_reminders"]["nodes"][0])
return cls(
data["id"],
uid=data["id"],
url=data.get("url"),
first_name=data.get("first_name"),
last_name=data.get("last_name"),
@@ -89,7 +89,7 @@ class User(Thread):
color=c_info.get("color"),
emoji=c_info.get("emoji"),
own_nickname=c_info.get("own_nickname"),
photo=data["profile_picture"].get("uri"),
photo=Image._from_uri(data["profile_picture"]),
name=data.get("name"),
message_count=data.get("messages_count"),
plan=plan,
@@ -123,7 +123,7 @@ class User(Thread):
plan = _plan.Plan._from_graphql(data["event_reminders"]["nodes"][0])
return cls(
user["id"],
uid=user["id"],
url=user.get("url"),
name=user.get("name"),
first_name=first_name,
@@ -135,7 +135,7 @@ class User(Thread):
color=c_info.get("color"),
emoji=c_info.get("emoji"),
own_nickname=c_info.get("own_nickname"),
photo=user["big_image_src"].get("uri"),
photo=Image._from_uri(user["big_image_src"]),
message_count=data.get("messages_count"),
last_active=last_active,
plan=plan,
@@ -144,10 +144,10 @@ class User(Thread):
@classmethod
def _from_all_fetch(cls, data):
return cls(
data["id"],
uid=data["id"],
first_name=data.get("firstName"),
url=data.get("uri"),
photo=data.get("thumbSrc"),
photo=Image(url=data.get("thumbSrc")),
name=data.get("name"),
is_friend=data.get("is_friend"),
gender=GENDERS.get(data.get("gender")),

2
pyproject.toml
View File

@@ -14,7 +14,7 @@ maintainer-email = "madsmtm@gmail.com"
home-page = "https://github.com/carpedm20/fbchat/"
requires = [
"aenum~=2.0",
"attrs>=18.2",
"attrs>=19.1",
"requests~=2.19",
"beautifulsoup4~=4.0",
]

46
tests/test_attachment.py
View File

@@ -72,10 +72,8 @@ def test_share_from_graphql_link():
title="a.com",
description="",
source="a.com",
image_url=None,
image=None,
original_image_url=None,
image_width=None,
image_height=None,
attachments=[],
uid="ee.mid.$xyz",
) == ShareAttachment._from_graphql(data)
@@ -125,10 +123,10 @@ def test_share_from_graphql_link_with_image():
" Share photos and videos, send messages and get updates."
),
source=None,
image_url="https://www.facebook.com/rsrc.php/v3/x.png",
image=fbchat.Image(
url="https://www.facebook.com/rsrc.php/v3/x.png", width=325, height=325
),
original_image_url="https://www.facebook.com/rsrc.php/v3/x.png",
image_width=325,
image_height=325,
attachments=[],
uid="deadbeef123",
) == ShareAttachment._from_graphql(data)
@@ -187,14 +185,14 @@ def test_share_from_graphql_video():
" Subscribe to the official Rick As..."
),
source="youtube.com",
image_url=(
"https://external-arn2-1.xx.fbcdn.net/safe_image.php?d=xyz123"
image=fbchat.Image(
url="https://external-arn2-1.xx.fbcdn.net/safe_image.php?d=xyz123"
"&w=960&h=540&url=https%3A%2F%2Fi.ytimg.com%2Fvi%2FdQw4w9WgXcQ"
"%2Fmaxresdefault.jpg&sx=0&sy=0&sw=1280&sh=720&_nc_hash=abc123"
"%2Fmaxresdefault.jpg&sx=0&sy=0&sw=1280&sh=720&_nc_hash=abc123",
width=960,
height=540,
),
original_image_url="https://i.ytimg.com/vi/dQw4w9WgXcQ/maxresdefault.jpg",
image_width=960,
image_height=540,
attachments=[],
uid="ee.mid.$gAAT4Sw1WSGhzQ9uRWVtEpZHZ8ZPV",
) == ShareAttachment._from_graphql(data)
@@ -310,10 +308,12 @@ def test_share_with_image_subattachment():
title="",
description="Abc",
source="Def",
image_url="https://scontent-arn2-1.xx.fbcdn.net/v/t1.0-9/1.jpg",
image=fbchat.Image(
url="https://scontent-arn2-1.xx.fbcdn.net/v/t1.0-9/1.jpg",
width=720,
height=960,
),
original_image_url="https://scontent-arn2-1.xx.fbcdn.net/v/t1.0-9/1.jpg",
image_width=720,
image_height=960,
attachments=[None],
uid="deadbeef123",
) == ShareAttachment._from_graphql(data)
@@ -436,19 +436,23 @@ def test_share_with_video_subattachment():
title="",
description="Abc",
source="Def",
image_url="https://scontent-arn2-1.xx.fbcdn.net/v/t15.5256-10/p180x540/1.jpg",
image=fbchat.Image(
url="https://scontent-arn2-1.xx.fbcdn.net/v/t15.5256-10/p180x540/1.jpg",
width=960,
height=540,
),
original_image_url="https://scontent-arn2-1.xx.fbcdn.net/v/t15.5256-10/p180x540/1.jpg",
image_width=960,
image_height=540,
attachments=[
fbchat.VideoAttachment(
uid="2222",
duration=datetime.timedelta(seconds=24, microseconds=469000),
preview_url="https://video-arn2-1.xx.fbcdn.net/v/t42.9040-2/vid.mp4",
medium_image={
"uri": "https://scontent-arn2-1.xx.fbcdn.net/v/t15.5256-10/p180x540/1.jpg",
"width": 960,
"height": 540,
previews={
fbchat.Image(
url="https://scontent-arn2-1.xx.fbcdn.net/v/t15.5256-10/p180x540/1.jpg",
width=960,
height=540,
)
},
)
],

116
tests/test_file.py
View File

@@ -33,16 +33,18 @@ def test_imageattachment_from_list():
uid="1234",
width=2833,
height=1367,
thumbnail_url="https://scontent-arn2-1.xx.fbcdn.net/v/s261x260/1.jpg",
preview={
"uri": "https://scontent-arn2-1.xx.fbcdn.net/v/2.jpg",
"width": 960,
"height": 463,
},
large_preview={
"uri": "https://scontent-arn2-1.xx.fbcdn.net/v/s2048x2048/3.jpg",
"width": 2048,
"height": 988,
previews={
fbchat.Image(url="https://scontent-arn2-1.xx.fbcdn.net/v/s261x260/1.jpg"),
fbchat.Image(
url="https://scontent-arn2-1.xx.fbcdn.net/v/2.jpg",
width=960,
height=463,
),
fbchat.Image(
url="https://scontent-arn2-1.xx.fbcdn.net/v/s2048x2048/3.jpg",
width=2048,
height=988,
),
},
) == ImageAttachment._from_list({"node": data})
@@ -71,18 +73,20 @@ def test_videoattachment_from_list():
uid="1234",
width=640,
height=368,
small_image={
"uri": "https://scontent-arn2-1.xx.fbcdn.net/v/t15.3394-10/p261x260/1.jpg"
},
medium_image={
"uri": "https://scontent-arn2-1.xx.fbcdn.net/v/t15.3394-10/2.jpg",
"width": 640,
"height": 368,
},
large_image={
"uri": "https://scontent-arn2-1.xx.fbcdn.net/v/t15.3394-10/3.jpg",
"width": 640,
"height": 368,
previews={
fbchat.Image(
url="https://scontent-arn2-1.xx.fbcdn.net/v/t15.3394-10/p261x260/1.jpg"
),
fbchat.Image(
url="https://scontent-arn2-1.xx.fbcdn.net/v/t15.3394-10/2.jpg",
width=640,
height=368,
),
fbchat.Image(
url="https://scontent-arn2-1.xx.fbcdn.net/v/t15.3394-10/3.jpg",
width=640,
height=368,
),
},
) == VideoAttachment._from_list({"node": data})
@@ -152,11 +156,11 @@ def test_graphql_to_attachment_image1():
"width": 128,
},
"large_preview": {
"uri": "https://scontent-arn2-1.xx.fbcdn.net/v/2.png",
"uri": "https://scontent-arn2-1.xx.fbcdn.net/v/1.png",
"height": 128,
"width": 128,
},
"thumbnail": {"uri": "https://scontent-arn2-1.xx.fbcdn.net/v/p50x50/3.png"},
"thumbnail": {"uri": "https://scontent-arn2-1.xx.fbcdn.net/v/p50x50/2.png"},
"photo_encodings": [],
"legacy_attachment_id": "1234",
"original_dimensions": {"x": 128, "y": 128},
@@ -170,16 +174,13 @@ def test_graphql_to_attachment_image1():
width=None,
height=None,
is_animated=False,
thumbnail_url="https://scontent-arn2-1.xx.fbcdn.net/v/p50x50/3.png",
preview={
"uri": "https://scontent-arn2-1.xx.fbcdn.net/v/1.png",
"width": 128,
"height": 128,
},
large_preview={
"uri": "https://scontent-arn2-1.xx.fbcdn.net/v/2.png",
"width": 128,
"height": 128,
previews={
fbchat.Image(url="https://scontent-arn2-1.xx.fbcdn.net/v/p50x50/2.png"),
fbchat.Image(
url="https://scontent-arn2-1.xx.fbcdn.net/v/1.png",
width=128,
height=128,
),
},
) == graphql_to_attachment(data)
@@ -209,11 +210,8 @@ def test_graphql_to_attachment_image2():
width=None,
height=None,
is_animated=True,
preview={"uri": "https://cdn.fbsbx.com/v/1.gif", "width": 128, "height": 128},
animated_preview={
"uri": "https://cdn.fbsbx.com/v/1.gif",
"width": 128,
"height": 128,
previews={
fbchat.Image(url="https://cdn.fbsbx.com/v/1.gif", width=128, height=128)
},
) == graphql_to_attachment(data)
@@ -251,20 +249,22 @@ def test_graphql_to_attachment_video():
height=None,
duration=datetime.timedelta(seconds=6),
preview_url="https://video-arn2-1.xx.fbcdn.net/v/video-4321.mp4",
small_image={
"uri": "https://scontent-arn2-1.xx.fbcdn.net/v/s168x128/1.jpg",
"width": 168,
"height": 96,
},
medium_image={
"uri": "https://scontent-arn2-1.xx.fbcdn.net/v/p261x260/3.jpg",
"width": 452,
"height": 260,
},
large_image={
"uri": "https://scontent-arn2-1.xx.fbcdn.net/v/2.jpg",
"width": 640,
"height": 368,
previews={
fbchat.Image(
url="https://scontent-arn2-1.xx.fbcdn.net/v/s168x128/1.jpg",
width=168,
height=96,
),
fbchat.Image(
url="https://scontent-arn2-1.xx.fbcdn.net/v/p261x260/3.jpg",
width=452,
height=260,
),
fbchat.Image(
url="https://scontent-arn2-1.xx.fbcdn.net/v/2.jpg",
width=640,
height=368,
),
},
) == graphql_to_attachment(data)
@@ -350,9 +350,11 @@ def test_graphql_to_subattachment_video():
uid="1234",
duration=datetime.timedelta(seconds=24, microseconds=469000),
preview_url="https://video-arn2-1.xx.fbcdn.net/v/t42.9040-2/vid.mp4",
medium_image={
"uri": "https://scontent-arn2-1.xx.fbcdn.net/v/t15.5256-10/p180x540/1.jpg",
"width": 960,
"height": 540,
previews={
fbchat.Image(
url="https://scontent-arn2-1.xx.fbcdn.net/v/t15.5256-10/p180x540/1.jpg",
width=960,
height=540,
)
},
) == graphql_to_subattachment(data)

25
tests/test_location.py
View File

@@ -1,5 +1,6 @@
import pytest
import datetime
import fbchat
from fbchat._location import LocationAttachment, LiveLocationAttachment
@@ -33,12 +34,17 @@ def test_location_attachment_from_graphql():
"target": {"__typename": "MessageLocation"},
"subattachments": [],
}
expected = LocationAttachment(latitude=55.4, longitude=12.4322, uid=400828513928715)
expected.image_url = "https://external-arn2-1.xx.fbcdn.net/static_map.php?v=1020&osm_provider=2&size=545x280&zoom=15&markers=55.40000000%2C12.43220000&language=en"
expected.image_width = 545
expected.image_height = 280
expected.url = "https://l.facebook.com/l.php?u=https%3A%2F%2Fwww.bing.com%2Fmaps%2Fdefault.aspx%3Fv%3D2%26pc%3DFACEBK%26mid%3D8100%26where1%3D55.4%252C%2B12.4322%26FORM%3DFBKPL1%26mkt%3Den-GB&h=a&s=1"
assert expected == LocationAttachment._from_graphql(data)
assert LocationAttachment(
uid=400828513928715,
latitude=55.4,
longitude=12.4322,
image=fbchat.Image(
url="https://external-arn2-1.xx.fbcdn.net/static_map.php?v=1020&osm_provider=2&size=545x280&zoom=15&markers=55.40000000%2C12.43220000&language=en",
width=545,
height=280,
),
url="https://l.facebook.com/l.php?u=https%3A%2F%2Fwww.bing.com%2Fmaps%2Fdefault.aspx%3Fv%3D2%26pc%3DFACEBK%26mid%3D8100%26where1%3D55.4%252C%2B12.4322%26FORM%3DFBKPL1%26mkt%3Den-GB&h=a&s=1",
) == LocationAttachment._from_graphql(data)
@pytest.mark.skip(reason="need to gather test data")
@@ -73,16 +79,15 @@ def test_live_location_from_graphql_expired():
},
"subattachments": [],
}
expected = LiveLocationAttachment(
assert LiveLocationAttachment(
uid=2254535444791641,
name="Location-sharing ended",
expires_at=datetime.datetime(
2019, 1, 4, 18, 25, 45, tzinfo=datetime.timezone.utc
),
is_expired=True,
)
expected.url = "https://www.facebook.com/"
assert expected == LiveLocationAttachment._from_graphql(data)
url="https://www.facebook.com/",
) == LiveLocationAttachment._from_graphql(data)
@pytest.mark.skip(reason="need to gather test data")

3
tests/test_page.py
View File

@@ -1,3 +1,4 @@
import fbchat
from fbchat._page import Page
@@ -12,7 +13,7 @@ def test_page_from_graphql():
}
assert Page(
uid="123456",
photo="https://scontent-arn2-1.xx.fbcdn.net/v/...",
photo=fbchat.Image(url="https://scontent-arn2-1.xx.fbcdn.net/v/..."),
name="Some school",
url="https://www.facebook.com/some-school/",
city=None,

48
tests/test_plan.py
View File

@@ -3,13 +3,16 @@ from fbchat._plan import GuestStatus, Plan
def test_plan_properties():
plan = Plan(time=..., title=...)
plan.guests = {
plan = Plan(
time=...,
title=...,
guests={
"1234": GuestStatus.INVITED,
"2345": GuestStatus.INVITED,
"3456": GuestStatus.GOING,
"4567": GuestStatus.DECLINED,
}
},
)
assert set(plan.invited) == {"1234", "2345"}
assert plan.going == ["3456"]
assert plan.declined == ["4567"]
@@ -32,19 +35,18 @@ def test_plan_from_pull():
'{"guest_list_state":"GOING","node":{"id":"4567"}}]'
),
}
plan = Plan(
assert Plan(
uid="1111",
time=datetime.datetime(2017, 7, 14, 2, 40, tzinfo=datetime.timezone.utc),
title="abc",
)
plan.uid = "1111"
plan.author_id = "1234"
plan.guests = {
author_id="1234",
guests={
"1234": GuestStatus.INVITED,
"2356": GuestStatus.INVITED,
"3456": GuestStatus.DECLINED,
"4567": GuestStatus.GOING,
}
assert plan == Plan._from_pull(data)
},
) == Plan._from_pull(data)
def test_plan_from_fetch():
@@ -90,21 +92,20 @@ def test_plan_from_fetch():
"4567": "GOING",
},
}
plan = Plan(
assert Plan(
uid=1111,
time=datetime.datetime(2017, 7, 14, 2, 40, tzinfo=datetime.timezone.utc),
title="abc",
location="",
location_id="",
)
plan.uid = 1111
plan.author_id = 1234
plan.guests = {
author_id=1234,
guests={
"1234": GuestStatus.INVITED,
"2356": GuestStatus.INVITED,
"3456": GuestStatus.DECLINED,
"4567": GuestStatus.GOING,
}
assert plan == Plan._from_fetch(data)
},
) == Plan._from_fetch(data)
def test_plan_from_graphql():
@@ -133,18 +134,17 @@ def test_plan_from_graphql():
]
},
}
plan = Plan(
assert Plan(
time=datetime.datetime(2017, 7, 14, 2, 40, tzinfo=datetime.timezone.utc),
title="abc",
location="",
location_id="",
)
plan.uid = "1111"
plan.author_id = "1234"
plan.guests = {
uid="1111",
author_id="1234",
guests={
"1234": GuestStatus.INVITED,
"2356": GuestStatus.INVITED,
"3456": GuestStatus.DECLINED,
"4567": GuestStatus.GOING,
}
assert plan == Plan._from_graphql(data)
},
) == Plan._from_graphql(data)

6
tests/test_plans.py
View File

@@ -10,12 +10,12 @@ pytestmark = pytest.mark.online
@pytest.fixture(
scope="module",
params=[
Plan(int(time()) + 100, random_hex()),
Plan(time=int(time()) + 100, title=random_hex()),
pytest.param(
Plan(int(time()), random_hex()),
Plan(time=int(time()), title=random_hex()),
marks=[pytest.mark.xfail(raises=FBchatFacebookError)],
),
pytest.param(Plan(0, None), marks=[pytest.mark.xfail()]),
pytest.param(Plan(time=0, title=None), marks=[pytest.mark.xfail()]),
],
)
def plan_data(request, client, user, thread, catch_event, compare):

20
tests/test_polls.py
View File

@@ -13,26 +13,26 @@ pytestmark = pytest.mark.online
Poll(
title=random_hex(),
options=[
PollOption(random_hex(), vote=True),
PollOption(random_hex(), vote=True),
PollOption(text=random_hex(), vote=True),
PollOption(text=random_hex(), vote=True),
],
),
Poll(
title=random_hex(),
options=[
PollOption(random_hex(), vote=False),
PollOption(random_hex(), vote=False),
PollOption(text=random_hex(), vote=False),
PollOption(text=random_hex(), vote=False),
],
),
Poll(
title=random_hex(),
options=[
PollOption(random_hex(), vote=True),
PollOption(random_hex(), vote=True),
PollOption(random_hex(), vote=False),
PollOption(random_hex(), vote=False),
PollOption(random_hex()),
PollOption(random_hex()),
PollOption(text=random_hex(), vote=True),
PollOption(text=random_hex(), vote=True),
PollOption(text=random_hex(), vote=False),
PollOption(text=random_hex(), vote=False),
PollOption(text=random_hex()),
PollOption(text=random_hex()),
],
),
pytest.param(

13
tests/test_sticker.py
View File

@@ -1,4 +1,5 @@
import pytest
import fbchat
from fbchat._sticker import Sticker
@@ -15,14 +16,14 @@ def test_from_graphql_normal():
uid="369239383222810",
pack="227877430692340",
is_animated=False,
medium_sprite_image=None,
large_sprite_image=None,
frames_per_row=None,
frames_per_col=None,
frame_rate=None,
frames_per_row=1,
frames_per_col=1,
frame_rate=83,
image=fbchat.Image(
url="https://scontent-arn2-1.xx.fbcdn.net/v/redacted.png",
width=274,
height=274,
),
label="Like, thumbs up",
) == Sticker._from_graphql(
{
@@ -54,9 +55,11 @@ def test_from_graphql_animated():
frames_per_row=2,
frames_per_col=2,
frame_rate=142,
image=fbchat.Image(
url="https://scontent-arn2-1.fbcdn.net/v/redacted1.png",
width=240,
height=293,
),
label="Love, cat with heart",
) == Sticker._from_graphql(
{

7
tests/test_user.py
View File

@@ -1,5 +1,6 @@
import pytest
import datetime
import fbchat
from fbchat._user import User, ActiveStatus
@@ -17,7 +18,7 @@ def test_user_from_graphql():
}
assert User(
uid="1234",
photo="https://scontent-arn2-1.xx.fbcdn.net/v/...",
photo=fbchat.Image(url="https://scontent-arn2-1.xx.fbcdn.net/v/..."),
name="Abc Def Ghi",
url="https://www.facebook.com/profile.php?id=1234",
first_name="Abc",
@@ -137,7 +138,7 @@ def test_user_from_thread_fetch():
}
assert User(
uid="1234",
photo="https://scontent-arn2-1.xx.fbcdn.net/v/...",
photo=fbchat.Image(url="https://scontent-arn2-1.xx.fbcdn.net/v/..."),
name="Abc Def Ghi",
last_active=datetime.datetime(2017, 7, 14, 2, 40, tzinfo=datetime.timezone.utc),
message_count=1111,
@@ -175,7 +176,7 @@ def test_user_from_all_fetch():
}
assert User(
uid="1234",
photo="https://scontent-arn2-1.xx.fbcdn.net/v/...",
photo=fbchat.Image(url="https://scontent-arn2-1.xx.fbcdn.net/v/..."),
name="Abc Def Ghi",
url="https://www.facebook.com/profile.php?id=1234",
first_name="Abc",

10
tests/utils.py
View File

@@ -22,9 +22,13 @@ EMOJI_LIST = [
]
STICKER_LIST = [
Sticker("767334476626295"),
pytest.param(Sticker("0"), marks=[pytest.mark.xfail(raises=FBchatFacebookError)]),
pytest.param(Sticker(None), marks=[pytest.mark.xfail(raises=FBchatFacebookError)]),
Sticker(uid="767334476626295"),
pytest.param(
Sticker(uid="0"), marks=[pytest.mark.xfail(raises=FBchatFacebookError)]
),
pytest.param(
Sticker(uid=None), marks=[pytest.mark.xfail(raises=FBchatFacebookError)]
),
]
TEXT_LIST = [