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

Merge pull request #499 from carpedm20/session-in-models

Browse Source 
Add ThreadABC helper, and move a bunch of methods out of Client
This commit is contained in:
Mads Marquart
2020-01-09 11:33:45 +01:00
parent d8d044f091 2ec0be9635
commit 942c3e5b70
26 changed files with 1011 additions and 1642 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
docs/api.rst
View File

@@ -17,8 +17,6 @@ Threads
-------
.. autoclass:: Thread()
.. autoclass:: ThreadType(Enum)
:undoc-members:
.. autoclass:: Page()
.. autoclass:: User()
.. autoclass:: Group()

26
docs/intro.rst
View File

@@ -44,11 +44,7 @@ When you're done using the client, and want to securely logout, use `Client.logo
Threads
-------
A thread can refer to two things: A Messenger group chat or a single Facebook user
`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
A thread can refer to two things: A Messenger group chat (`Group`) or a single Facebook user (`User`).
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`
@@ -68,13 +64,14 @@ The same method can be applied to some user accounts, though if they've set a cu
Here's an snippet showing the usage of thread IDs and thread types, where ``<user id>`` and ``<group id>``
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)
user.send(Message(text='<message>'))
group.send(Message(text='<message>'))
Some functions (e.g. `Client.change_thread_color`) don't require a thread type, so in these cases you just provide the thread ID::
Some functions 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>')
thread = fbchat.Thread(session=session, id="<user-or-group-id>")
thread.set_color(ThreadColor.BILOBA_FLOWER)
thread.set_color(ThreadColor.MESSENGER_BLUE)
.. _intro_message_ids:
@@ -89,7 +86,7 @@ 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)
message_id = thread.send(Message(text='message'))
client.react_to_message(message_id, MessageReaction.LOVE)
@@ -106,7 +103,8 @@ like adding users to and removing users from a group chat, logically only works
The simplest way of using ``fbchat`` is to send a message.
The following snippet will, as you've probably already figured out, send the message ``test message`` to your account::
message_id = client.send(Message(text='test message'), thread_id=session.user_id, thread_type=ThreadType.USER)
user = User(session=session, id=session.user_id)
message_id = user.send(Message(text='test message'))
You can see a full example showing all the possible thread interactions with ``fbchat`` by going to :ref:`examples`
@@ -173,7 +171,7 @@ meaning it will simply print information to the console when an event happens
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):
def on_message(self, mid, author_id, message_object, thread, ts, metadata, msg, **kwargs):
# Do something with message_object here
pass
@@ -182,7 +180,7 @@ The event actions can be changed by subclassing the `Client`, and then overwriti
**Notice:** The following snippet is as equally valid as the previous one::
class CustomClient(Client):
def on_message(self, message_object, author_id, thread_id, thread_type, **kwargs):
def on_message(self, message_object, author_id, thread, **kwargs):
# Do something with message_object here
pass

10
examples/basic_usage.py
View File

@@ -5,15 +5,11 @@ session = fbchat.Session.login("<email>", "<password>")
print("Own id: {}".format(sesion.user_id))
# Create helper client class
client = fbchat.Client(session)
# Create helper User class
user = fbchat.Thread(session=session, id=session.user_id)
# Send a message to yourself
client.send(
fbchat.Message(text="Hi me!"),
thread_id=session.user_id,
thread_type=fbchat.ThreadType.USER,
)
user.send(fbchat.Message(text="Hi me!"))
# Log the user out
session.logout()

10
examples/echobot.py
View File

@@ -2,15 +2,15 @@ import fbchat
# Subclass fbchat.Client and override required methods
class EchoBot(fbchat.Client):
def on_message(self, author_id, message_object, thread_id, thread_type, **kwargs):
self.mark_as_delivered(thread_id, message_object.id)
self.mark_as_read(thread_id)
def on_message(self, author_id, message_object, thread, **kwargs):
self.mark_as_delivered(thread.id, message_object.id)
self.mark_as_read(thread.id)
print("{} from {} in {}".format(message_object, thread_id, thread_type.name))
print("{} from {} in {}".format(message_object, thread))
# If you're not the author, echo
if author_id != self.session.user_id:
self.send(message_object, thread_id=thread_id, thread_type=thread_type)
thread.send(message_object)
session = fbchat.Session.login("<email>", "<password>")

16
examples/fetch.py
View File

@@ -39,8 +39,13 @@ threads += client.fetch_thread_list(offset=20, limit=10)
print("Threads: {}".format(threads))
# If we have a thread id, we can use `fetch_thread_info` to fetch a `Thread` object
thread = client.fetch_thread_info("<thread id>")["<thread id>"]
print("thread's name: {}".format(thread.name))
# Gets the last 10 messages sent to the thread
messages = client.fetch_thread_messages(thread_id="<thread id>", limit=10)
messages = thread.fetch_messages(limit=10)
# Since the message come in reversed order, reverse them
messages.reverse()
@@ -49,22 +54,15 @@ for message in messages:
print(message.text)
# If we have a thread id, we can use `fetch_thread_info` to fetch a `Thread` object
thread = client.fetch_thread_info("<thread id>")["<thread id>"]
print("thread's name: {}".format(thread.name))
print("thread's type: {}".format(thread.type))
# `search_for_threads` searches works like `search_for_users`, but gives us a list of threads instead
thread = client.search_for_threads("<name of thread>")[0]
print("thread's name: {}".format(thread.name))
print("thread's type: {}".format(thread.type))
# Here should be an example of `getUnread`
# Print image url for 20 last images from thread.
images = client.fetch_thread_images("<thread id>")
images = thread.fetch_images()
for image in itertools.islice(image, 20):
print(image.large_preview_url)

88
examples/interract.py
View File

@@ -4,94 +4,64 @@ session = fbchat.Session.login("<email>", "<password>")
client = fbchat.Client(session)
thread_id = "1234567890"
thread_type = fbchat.ThreadType.GROUP
thread = User(session=session, id=session.user_id)
# thread = User(session=session, id="0987654321")
# thread = Group(session=session, id="1234567890")
# Will send a message to the thread
client.send(
fbchat.Message(text="<message>"), thread_id=thread_id, thread_type=thread_type
)
thread.send(fbchat.Message(text="<message>"))
# Will send the default `like` emoji
client.send(
fbchat.Message(emoji_size=fbchat.EmojiSize.LARGE),
thread_id=thread_id,
thread_type=thread_type,
)
thread.send(fbchat.Message(emoji_size=fbchat.EmojiSize.LARGE))
# Will send the emoji `👍`
client.send(
fbchat.Message(text="👍", emoji_size=fbchat.EmojiSize.LARGE),
thread_id=thread_id,
thread_type=thread_type,
)
thread.send(fbchat.Message(text="👍", emoji_size=fbchat.EmojiSize.LARGE))
# Will send the sticker with ID `767334476626295`
client.send(
fbchat.Message(sticker=fbchat.Sticker("767334476626295")),
thread_id=thread_id,
thread_type=thread_type,
)
thread.send(fbchat.Message(sticker=fbchat.Sticker("767334476626295")))
# Will send a message with a mention
client.send(
thread.send(
fbchat.Message(
text="This is a @mention",
mentions=[fbchat.Mention(thread_id, offset=10, length=8)],
),
thread_id=thread_id,
thread_type=thread_type,
)
)
# Will send the image located at `<image path>`
client.send_local_image(
"<image path>",
message=fbchat.Message(text="This is a local image"),
thread_id=thread_id,
thread_type=thread_type,
thread.send_local_image(
"<image path>", message=fbchat.Message(text="This is a local image")
)
# Will download the image at the URL `<image url>`, and then send it
client.send_remote_image(
"<image url>",
message=fbchat.Message(text="This is a remote image"),
thread_id=thread_id,
thread_type=thread_type,
thread.send_remote_image(
"<image url>", message=fbchat.Message(text="This is a remote image")
)
# Only do these actions if the thread is a group
if thread_type == fbchat.ThreadType.GROUP:
# Will remove the user with ID `<user id>` from the thread
client.remove_user_from_group("<user id>", thread_id=thread_id)
# Will add the user with ID `<user id>` to the thread
client.add_users_to_group("<user id>", thread_id=thread_id)
# Will add the users with IDs `<1st user id>`, `<2nd user id>` and `<3th user id>` to the thread
client.add_users_to_group(
["<1st user id>", "<2nd user id>", "<3rd user id>"], thread_id=thread_id
)
if isinstance(thread, fbchat.Group):
# Will remove the user with ID `<user id>` from the group
thread.remove_participant("<user id>")
# Will add the users with IDs `<1st user id>`, `<2nd user id>` and `<3th user id>` to the group
thread.add_participants(["<1st user id>", "<2nd user id>", "<3rd user id>"])
# Will change the title of the group to `<title>`
thread.change_title("<title>")
# Will change the nickname of the user `<user_id>` to `<new nickname>`
client.change_nickname(
"<new nickname>", "<user id>", thread_id=thread_id, thread_type=thread_type
)
thread.set_nickname(fbchat.User(session=session, id="<user id>"), "<new nickname>")
# Will change the title of the thread to `<title>`
client.change_thread_title("<title>", thread_id=thread_id, thread_type=thread_type)
# Will set the typing status of the thread to `TYPING`
client.set_typing_status(
fbchat.TypingStatus.TYPING, thread_id=thread_id, thread_type=thread_type
)
# Will set the typing status of the thread
thread.start_typing()
# Will change the thread color to `MESSENGER_BLUE`
client.change_thread_color(fbchat.ThreadColor.MESSENGER_BLUE, thread_id=thread_id)
thread.set_color(fbchat.ThreadColor.MESSENGER_BLUE)
# Will change the thread emoji to `👍`
client.change_thread_emoji("👍", thread_id=thread_id)
thread.set_emoji("👍")
# Will react to a message with a 😍 emoji
client.react_to_message("<message id>", fbchat.MessageReaction.LOVE)
# message = fbchat.Message(session=session, id="<message id>")
#
# # Will react to a message with a 😍 emoji
# message.react(fbchat.MessageReaction.LOVE)

43
examples/keepbot.py
View File

@@ -16,50 +16,48 @@ old_nicknames = {
class KeepBot(fbchat.Client):
def on_color_change(self, author_id, new_color, thread_id, thread_type, **kwargs):
if old_thread_id == thread_id and old_color != new_color:
def on_color_change(self, author_id, new_color, thread, **kwargs):
if old_thread_id == thread.id and old_color != new_color:
print(
"{} changed the thread color. It will be changed back".format(author_id)
)
self.change_thread_color(old_color, thread_id=thread_id)
thread.set_color(old_color)
def on_emoji_change(self, author_id, new_emoji, thread_id, thread_type, **kwargs):
if old_thread_id == thread_id and new_emoji != old_emoji:
def on_emoji_change(self, author_id, new_emoji, thread, **kwargs):
if old_thread_id == thread.id and new_emoji != old_emoji:
print(
"{} changed the thread emoji. It will be changed back".format(author_id)
)
self.change_thread_emoji(old_emoji, thread_id=thread_id)
thread.set_emoji(old_emoji)
def on_people_added(self, added_ids, author_id, thread_id, **kwargs):
if old_thread_id == thread_id and author_id != self.session.user_id:
def on_people_added(self, added_ids, author_id, thread, **kwargs):
if old_thread_id == thread.id and author_id != self.session.user_id:
print("{} got added. They will be removed".format(added_ids))
for added_id in added_ids:
self.remove_user_from_group(added_id, thread_id=thread_id)
thread.remove_participant(added_id)
def on_person_removed(self, removed_id, author_id, thread_id, **kwargs):
def on_person_removed(self, removed_id, author_id, thread, **kwargs):
# No point in trying to add ourself
if (
old_thread_id == thread_id
old_thread_id == thread.id
and removed_id != self.session.user_id
and author_id != self.session.user_id
):
print("{} got removed. They will be re-added".format(removed_id))
self.add_users_to_group(removed_id, thread_id=thread_id)
thread.add_participants(removed_id)
def on_title_change(self, author_id, new_title, thread_id, thread_type, **kwargs):
if old_thread_id == thread_id and old_title != new_title:
def on_title_change(self, author_id, new_title, thread, **kwargs):
if old_thread_id == thread.id and old_title != new_title:
print(
"{} changed the thread title. It will be changed back".format(author_id)
)
self.change_thread_title(
old_title, thread_id=thread_id, thread_type=thread_type
)
thread.set_title(old_title)
def on_nickname_change(
self, author_id, changed_for, new_nickname, thread_id, thread_type, **kwargs
self, author_id, changed_for, new_nickname, thread, **kwargs
):
if (
old_thread_id == thread_id
old_thread_id == thread.id
and changed_for in old_nicknames
and old_nicknames[changed_for] != new_nickname
):
@@ -68,11 +66,8 @@ class KeepBot(fbchat.Client):
author_id, changed_for
)
)
self.change_nickname(
old_nicknames[changed_for],
changed_for,
thread_id=thread_id,
thread_type=thread_type,
thread.set_nickname(
changed_for, old_nicknames[changed_for],
)

14
examples/removebot.py
View File

@@ -2,21 +2,17 @@ import fbchat
class RemoveBot(fbchat.Client):
def on_message(self, author_id, message_object, thread_id, thread_type, **kwargs):
def on_message(self, author_id, message_object, thread, **kwargs):
# We can only kick people from group chats, so no need to try if it's a user chat
if (
message_object.text == "Remove me!"
and thread_type == fbchat.ThreadType.GROUP
):
print("{} will be removed from {}".format(author_id, thread_id))
self.remove_user_from_group(author_id, thread_id=thread_id)
if message_object.text == "Remove me!" and isinstance(thread, fbchat.Group):
print("{} will be removed from {}".format(author_id, thread))
thread.remove_participant(author_id)
else:
# Sends the data to the inherited on_message, so that we can still see when a message is recieved
super(RemoveBot, self).on_message(
author_id=author_id,
message_object=message_object,
thread_id=thread_id,
thread_type=thread_type,
thread=thread,
**kwargs,
)

2
fbchat/__init__.py
View File

@@ -14,7 +14,7 @@ from . import _core, _util
from ._core import Image
from ._exception import FBchatException, FBchatFacebookError
from ._session import Session
from ._thread import ThreadType, ThreadLocation, ThreadColor, Thread
from ._thread import ThreadLocation, ThreadColor, ThreadABC, Thread
from ._user import TypingStatus, User, ActiveStatus
from ._group import Group
from ._page import Page

1568
fbchat/_client.py
View File

File diff suppressed because it is too large Load Diff

158
fbchat/_group.py
View File

@@ -1,15 +1,17 @@
import attr
from ._core import attrs_default, Image
from . import _util, _plan
from ._thread import ThreadType, Thread
from . import _util, _session, _plan, _thread, _user
from typing import Sequence, Iterable
@attrs_default
class Group(Thread):
"""Represents a Facebook group. Inherits `Thread`."""
type = ThreadType.GROUP
class Group(_thread.ThreadABC):
"""Represents a Facebook group. Implements `ThreadABC`."""
#: The session to use when making requests.
session = attr.ib(type=_session.Session)
#: The group's unique identifier.
id = attr.ib(converter=str)
#: The group's picture
photo = attr.ib(None)
#: The name of the group
@@ -37,8 +39,120 @@ class Group(Thread):
# Link for joining group
join_link = attr.ib(None)
def add_participants(self, user_ids: Iterable[str]):
"""Add users to the group.
Args:
user_ids: One or more user IDs to add
"""
data = self._to_send_data()
data["action_type"] = "ma-type:log-message"
data["log_message_type"] = "log:subscribe"
for i, user_id in enumerate(user_ids):
if user_id == self.session.user_id:
raise ValueError(
"Error when adding users: Cannot add self to group thread"
)
else:
data[
"log_message_data[added_participants][{}]".format(i)
] = "fbid:{}".format(user_id)
return self.session._do_send_request(data)
def remove_participant(self, user_id: str):
"""Remove user from the group.
Args:
user_id: User ID to remove
"""
data = {"uid": user_id, "tid": self.id}
j = self._payload_post("/chat/remove_participants/", data)
def _admin_status(self, user_ids: Iterable[str], status: bool):
data = {"add": admin, "thread_fbid": self.id}
for i, user_id in enumerate(user_ids):
data["admin_ids[{}]".format(i)] = str(user_id)
j = self.session._payload_post("/messaging/save_admins/?dpr=1", data)
def add_admins(self, user_ids: Iterable[str]):
"""Set specified users as group admins.
Args:
user_ids: One or more user IDs to set admin
"""
self._admin_status(user_ids, True)
def remove_admins(self, user_ids: Iterable[str]):
"""Remove admin status from specified users.
Args:
user_ids: One or more user IDs to remove admin
"""
self._admin_status(user_ids, False)
def set_title(self, title: str):
"""Change title of the group.
Args:
title: New title
"""
data = {"thread_name": title, "thread_id": self.id}
j = self.session._payload_post("/messaging/set_thread_name/?dpr=1", data)
def set_image(self, image_id: str):
"""Change the group image from an image id.
Args:
image_id: ID of uploaded image
"""
data = {"thread_image_id": image_id, "thread_id": self.id}
j = self.session._payload_post("/messaging/set_thread_image/?dpr=1", data)
def set_approval_mode(self, require_admin_approval: bool):
"""Change the group's approval mode.
Args:
require_admin_approval: True or False
"""
data = {"set_mode": int(require_admin_approval), "thread_fbid": thread_id}
j = self.session._payload_post("/messaging/set_approval_mode/?dpr=1", data)
def _users_approval(self, user_ids: Iterable[str], approve: bool):
data = {
"client_mutation_id": "0",
"actor_id": self.session.user_id,
"thread_fbid": self.id,
"user_ids": list(user_ids),
"response": "ACCEPT" if approve else "DENY",
"surface": "ADMIN_MODEL_APPROVAL_CENTER",
}
(j,) = self.session._graphql_requests(
_graphql.from_doc_id("1574519202665847", {"data": data})
)
def accept_users(self, user_ids: Iterable[str]):
"""Accept users to the group from the group's approval.
Args:
user_ids: One or more user IDs to accept
"""
self._users_approval(user_ids, True)
def deny_users(self, user_ids: Iterable[str]):
"""Deny users from joining the group.
Args:
user_ids: One or more user IDs to deny
"""
self._users_approval(user_ids, False)
@classmethod
def _from_graphql(cls, data):
def _from_graphql(cls, session, data):
if data.get("image") is None:
data["image"] = {}
c_info = cls._parse_customization_info(data)
@@ -52,6 +166,7 @@ class Group(Thread):
plan = _plan.Plan._from_graphql(data["event_reminders"]["nodes"][0])
return cls(
session=session,
id=data["thread_key"]["thread_fbid"],
participants=set(
[
@@ -82,3 +197,32 @@ class Group(Thread):
def _to_send_data(self):
return {"thread_fbid": self.id}
@attrs_default
class NewGroup(_thread.ThreadABC):
"""Helper class to create new groups.
TODO: Complete this!
Construct this class with the desired users, and call a method like `wave`, to...
"""
#: The session to use when making requests.
session = attr.ib(type=_session.Session)
#: The users that should be added to the group.
_users = attr.ib(type=Sequence[_user.User])
@property
def id(self):
raise NotImplementedError(
"The method you called is not supported on NewGroup objects."
" Please use the supported methods to create the group, before attempting"
" to call the method."
)
def _to_send_data(self) -> dict:
return {
"specific_to_list[{}]".format(i): "fbid:{}".format(user.id)
for i, user in enumerate(self._users)
}

60
fbchat/_message.py
View File

@@ -2,7 +2,8 @@ import attr
import json
from string import Formatter
from ._core import log, attrs_default, Enum
from . import _util, _attachment, _location, _file, _quick_reply, _sticker
from . import _util, _session, _attachment, _location, _file, _quick_reply, _sticker
from typing import Optional
class EmojiSize(Enum):
@@ -78,14 +79,18 @@ class Mention:
class Message:
"""Represents a Facebook message."""
# TODO: Make these fields required!
#: The session to use when making requests.
session = attr.ib(None, type=_session.Session)
#: The message ID
id = attr.ib(None, converter=str)
#: The actual message
text = attr.ib(None)
#: 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
id = attr.ib(None)
#: ID of the sender
author = attr.ib(None)
#: Datetime of when the message was sent
@@ -111,6 +116,38 @@ class Message:
#: Whether the message was forwarded
forwarded = attr.ib(False)
def unsend(self):
"""Unsend the message (removes it for everyone)."""
data = {"message_id": self.id}
j = self.session._payload_post("/messaging/unsend_message/?dpr=1", data)
def react(self, reaction: Optional[MessageReaction]):
"""React to the message, or removes reaction.
Args:
reaction: Reaction emoji to use, if None removes reaction
"""
data = {
"action": "ADD_REACTION" if reaction else "REMOVE_REACTION",
"client_mutation_id": "1",
"actor_id": self.session.user_id,
"message_id": self.id,
"reaction": reaction.value if reaction else None,
}
data = {"doc_id": 1491398900900362, "variables": json.dumps({"data": data})}
j = self.session._payload_post("/webgraphql/mutation", data)
_util.handle_graphql_errors(j)
@classmethod
def from_fetch(cls, thread, message_id: str) -> "Message":
"""Fetch `Message` object from the given message id.
Args:
message_id: Message ID to fetch from
"""
message_info = thread._forced_fetch(message_id).get("message")
return Message._from_graphql(thread.session, message_info)
@classmethod
def format_mentions(cls, text, *args, **kwargs):
"""Like `str.format`, but takes tuples with a thread id and text instead.
@@ -224,7 +261,7 @@ class Message:
return []
@classmethod
def _from_graphql(cls, data, read_receipts=None):
def _from_graphql(cls, session, data, read_receipts=None):
if data.get("message_sender") is None:
data["message_sender"] = {}
if data.get("message") is None:
@@ -250,12 +287,13 @@ class Message:
replied_to = cls._from_graphql(data["replied_to_message"]["message"])
return cls(
session=session,
id=str(data["message_id"]),
text=data["message"].get("text"),
mentions=[
Mention._from_range(m) for m in data["message"].get("ranges") or ()
],
emoji_size=EmojiSize._from_tags(tags),
id=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,
@@ -278,7 +316,7 @@ class Message:
)
@classmethod
def _from_reply(cls, data, replied_to=None):
def _from_reply(cls, session, data, replied_to=None):
tags = data["messageMetadata"].get("tags")
metadata = data.get("messageMetadata", {})
@@ -305,13 +343,14 @@ class Message:
)
return cls(
session=session,
id=metadata.get("messageId"),
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),
id=metadata.get("messageId"),
author=str(metadata.get("actorFbId")),
created_at=_util.millis_to_datetime(metadata.get("timestamp")),
sticker=sticker,
@@ -324,7 +363,9 @@ class Message:
)
@classmethod
def _from_pull(cls, data, mid=None, tags=None, author=None, created_at=None):
def _from_pull(
cls, session, data, mid=None, tags=None, author=None, created_at=None
):
mentions = []
if data.get("data") and data["data"].get("prng"):
try:
@@ -371,10 +412,11 @@ class Message:
)
return cls(
session=session,
id=mid,
text=data.get("body"),
mentions=mentions,
emoji_size=EmojiSize._from_tags(tags),
id=mid,
author=author,
created_at=created_at,
sticker=sticker,

19
fbchat/_page.py
View File

@@ -1,15 +1,16 @@
import attr
from ._core import attrs_default, Image
from . import _plan
from ._thread import ThreadType, Thread
from . import _session, _plan, _thread
@attrs_default
class Page(Thread):
"""Represents a Facebook page. Inherits `Thread`."""
type = ThreadType.PAGE
class Page(_thread.ThreadABC):
"""Represents a Facebook page. Implements `ThreadABC`."""
#: The session to use when making requests.
session = attr.ib(type=_session.Session)
#: The unique identifier of the page.
id = attr.ib(converter=str)
#: The page's picture
photo = attr.ib(None)
#: The name of the page
@@ -31,8 +32,11 @@ class Page(Thread):
#: The page's category
category = attr.ib(None)
def _to_send_data(self):
return {"other_user_fbid": self.id}
@classmethod
def _from_graphql(cls, data):
def _from_graphql(cls, session, data):
if data.get("profile_picture") is None:
data["profile_picture"] = {}
if data.get("city") is None:
@@ -42,6 +46,7 @@ class Page(Thread):
plan = _plan.Plan._from_graphql(data["event_reminders"]["nodes"][0])
return cls(
session=session,
id=data["id"],
url=data.get("url"),
city=data.get("city").get("name"),

484
fbchat/_thread.py
View File

@@ -1,26 +1,9 @@
import abc
import attr
import datetime
from ._core import attrs_default, Enum, Image
class ThreadType(Enum):
"""Used to specify what type of Facebook thread is being used.
See :ref:`intro_threads` for more info.
"""
USER = 1
GROUP = 2
PAGE = 3
def _to_class(self):
"""Convert this enum value to the corresponding class."""
from . import _user, _group, _page
return {
ThreadType.USER: _user.User,
ThreadType.GROUP: _group.Group,
ThreadType.PAGE: _page.Page,
}[self]
from . import _util, _exception, _session
from typing import MutableMapping, Any, Iterable, Tuple
class ThreadLocation(Enum):
@@ -67,17 +50,438 @@ class ThreadColor(Enum):
return cls._extend_if_invalid(value)
@attrs_default
class Thread:
"""Represents a Facebook thread."""
class ThreadABC(metaclass=abc.ABCMeta):
"""Implemented by thread-like classes.
#: The unique identifier of the thread.
id = attr.ib(converter=str)
#: Specifies the type of thread. Can be used a ``thread_type``. See :ref:`intro_threads` for more info
type = None
This is private to implement.
"""
@property
@abc.abstractmethod
def session(self) -> _session.Session:
"""The session to use when making requests."""
raise NotImplementedError
@property
@abc.abstractmethod
def id(self) -> str:
"""The unique identifier of the thread."""
raise NotImplementedError
@abc.abstractmethod
def _to_send_data(self) -> MutableMapping[str, str]:
raise NotImplementedError
def wave(self, first: bool = True) -> str:
"""Wave hello to the thread.
Args:
first: Whether to wave first or wave back
"""
data = self._to_send_data()
data["action_type"] = "ma-type:user-generated-message"
data["lightweight_action_attachment[lwa_state]"] = (
"INITIATED" if first else "RECIPROCATED"
)
data["lightweight_action_attachment[lwa_type]"] = "WAVE"
# TODO: This!
# if isinstance(self, _user.User):
# data["specific_to_list[0]"] = "fbid:{}".format(thread_id)
message_id, thread_id = self.session._do_send_request(data)
return message_id
def send(self, message) -> str:
"""Send message to the thread.
Args:
message (Message): Message to send
Returns:
:ref:`Message ID <intro_message_ids>` of the sent message
"""
data = self._to_send_data()
data.update(message._to_send_data())
return self.session._do_send_request(data)
def _send_location(self, current, latitude, longitude, message=None) -> str:
data = self._to_send_data()
if message is not None:
data.update(message._to_send_data())
data["action_type"] = "ma-type:user-generated-message"
data["location_attachment[coordinates][latitude]"] = latitude
data["location_attachment[coordinates][longitude]"] = longitude
data["location_attachment[is_current_location]"] = current
return self.session._do_send_request(data)
def send_location(self, latitude: float, longitude: float, message=None):
"""Send a given location to a thread as the user's current location.
Args:
latitude: The location latitude
longitude: The location longitude
message: Additional message
"""
self._send_location(
True, latitude=latitude, longitude=longitude, message=message,
)
def send_pinned_location(self, latitude: float, longitude: float, message=None):
"""Send a given location to a thread as a pinned location.
Args:
latitude: The location latitude
longitude: The location longitude
message: Additional message
"""
self._send_location(
False, latitude=latitude, longitude=longitude, message=message,
)
def send_files(self, files: Iterable[Tuple[str, str]], message):
"""Send files from file IDs to a thread.
`files` should be a list of tuples, with a file's ID and mimetype.
"""
data = self._to_send_data()
data.update(message._to_send_data())
data["action_type"] = "ma-type:user-generated-message"
data["has_attachment"] = True
for i, (file_id, mimetype) in enumerate(files):
data["{}s[{}]".format(_util.mimetype_to_key(mimetype), i)] = file_id
return self.session._do_send_request(data)
# TODO: This!
# def quick_reply(self, quick_reply, payload=None):
# """Reply to chosen quick reply.
#
# Args:
# quick_reply (QuickReply): Quick reply to reply to
# payload: Optional answer to the quick reply
# """
# if isinstance(quick_reply, QuickReplyText):
# 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 `LocationAttachment`")
# return self.send_location(payload)
# elif isinstance(quick_reply, QuickReplyEmail):
# 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):
# 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 search_messages(
self, query: str, offset: int = 0, limit: int = 5
) -> Iterable[str]:
"""Find and get message IDs by query.
Args:
query: Text to search for
offset (int): Number of messages to skip
limit (int): Max. number of messages to retrieve
Returns:
typing.Iterable: Found Message IDs
"""
# TODO: Return proper searchable iterator
data = {
"query": query,
"snippetOffset": offset,
"snippetLimit": limit,
"identifier": "thread_fbid",
"thread_fbid": self.id,
}
j = self.session._payload_post("/ajax/mercury/search_snippets.php?dpr=1", data)
result = j["search_snippets"][query]
snippets = result[self.id]["snippets"] if result.get(self.id) else []
for snippet in snippets:
yield snippet["message_id"]
def fetch_messages(self, limit: int = 20, before: datetime.datetime = None):
"""Fetch messages in a thread, ordered by most recent.
Args:
limit: Max. number of messages to retrieve
before: The point from which to retrieve messages
Returns:
list: `Message` objects
"""
# TODO: Return proper searchable iterator
params = {
"id": self.id,
"message_limit": limit,
"load_messages": True,
"load_read_receipts": True,
"before": _util.datetime_to_millis(before) if before else None,
}
(j,) = self.session._graphql_requests(
_graphql.from_doc_id("1860982147341344", params)
)
if j.get("message_thread") is None:
raise FBchatException("Could not fetch thread {}: {}".format(self.id, j))
read_receipts = j["message_thread"]["read_receipts"]["nodes"]
messages = [
Message._from_graphql(self.session, message, read_receipts)
for message in j["message_thread"]["messages"]["nodes"]
]
messages.reverse()
return messages
def fetch_images(self):
"""Fetch images/videos posted in the thread."""
# TODO: Return proper searchable iterator
data = {"id": self.id, "first": 48}
(j,) = self.session._graphql_requests(
_graphql.from_query_id("515216185516880", data)
)
while True:
try:
i = j[self.id]["message_shared_media"]["edges"][0]
except IndexError:
if j[self.id]["message_shared_media"]["page_info"].get("has_next_page"):
data["after"] = j[self.id]["message_shared_media"]["page_info"].get(
"end_cursor"
)
(j,) = self.session._graphql_requests(
_graphql.from_query_id("515216185516880", data)
)
continue
else:
break
if i["node"].get("__typename") == "MessageImage":
yield ImageAttachment._from_list(i)
elif i["node"].get("__typename") == "MessageVideo":
yield VideoAttachment._from_list(i)
else:
yield Attachment(id=i["node"].get("legacy_attachment_id"))
del j[self.id]["message_shared_media"]["edges"][0]
def set_nickname(self, user_id: str, nickname: str):
"""Change the nickname of a user in the thread.
Args:
user_id: User that will have their nickname changed
nickname: New nickname
"""
data = {
"nickname": nickname,
"participant_id": user_id,
"thread_or_other_fbid": self.id,
}
j = self.session._payload_post(
"/messaging/save_thread_nickname/?source=thread_settings&dpr=1", data
)
def set_color(self, color: ThreadColor):
"""Change thread color.
Args:
color: New thread color
"""
data = {
"color_choice": color.value if color != ThreadColor.MESSENGER_BLUE else "",
"thread_or_other_fbid": self.id,
}
j = self.session._payload_post(
"/messaging/save_thread_color/?source=thread_settings&dpr=1", data
)
def set_emoji(self, emoji: str):
"""Change thread color.
Args:
emoji: New thread emoji
"""
data = {"emoji_choice": emoji, "thread_or_other_fbid": self.id}
# While changing the emoji, the Facebook web client actually sends multiple
# different requests, though only this one is required to make the change.
j = self.session._payload_post(
"/messaging/save_thread_emoji/?source=thread_settings&dpr=1", data
)
def forward_attachment(self, attachment_id):
"""Forward an attachment.
Args:
attachment_id: Attachment ID to forward
"""
data = {
"attachment_id": attachment_id,
"recipient_map[{}]".format(_util.generate_offline_threading_id()): self.id,
}
j = self.session._payload_post("/mercury/attachments/forward/", data)
if not j.get("success"):
raise _exception.FBchatFacebookError(
"Failed forwarding attachment: {}".format(j["error"]),
fb_error_message=j["error"],
)
def _set_typing(self, typing):
data = {
"typ": "1" if typing else "0",
"thread": self.id,
# TODO: This
# "to": self.id if isinstance(self, _user.User) else "",
"source": "mercury-chat",
}
j = self.session._payload_post("/ajax/messaging/typ.php", data)
def start_typing(self):
"""Set the current user to start typing in the thread."""
self._set_typing(True)
def stop_typing(self):
"""Set the current user to stop typing in the thread."""
self._set_typing(False)
def create_plan(
self,
name: str,
at: datetime.datetime,
location_name: str = None,
location_id: str = None,
):
"""Create a new plan.
# TODO: Arguments
Args:
title: Name of the new plan
at: When the plan is for
"""
data = {
"event_type": "EVENT",
"event_time": _util.datetime_to_seconds(at),
"title": name,
"thread_id": self.id,
"location_id": location_id or "",
"location_name": location or "",
"acontext": ACONTEXT,
}
j = self.session._payload_post("/ajax/eventreminder/create", data)
if "error" in j:
raise _exception.FBchatFacebookError(
"Failed creating plan: {}".format(j["error"]),
fb_error_message=j["error"],
)
def create_poll(self, question: str, options=Iterable[Tuple[str, bool]]):
"""Create poll in a thread.
# TODO: Arguments
"""
# We're using ordered dictionaries, because the Facebook endpoint that parses
# the POST parameters is badly implemented, and deals with ordering the options
# wrongly. If you can find a way to fix this for the endpoint, or if you find
# another endpoint, please do suggest it ;)
data = OrderedDict([("question_text", question), ("target_id", self.id)])
for i, (text, vote) in enumerate(options):
data["option_text_array[{}]".format(i)] = text
data["option_is_selected_array[{}]".format(i)] = str(int(vote))
j = self.session._payload_post(
"/messaging/group_polling/create_poll/?dpr=1", data
)
if j.get("status") != "success":
raise _exception.FBchatFacebookError(
"Failed creating poll: {}".format(j.get("errorTitle")),
fb_error_message=j.get("errorMessage"),
)
def mute(self, duration: datetime.timedelta = None):
"""Mute the thread.
Args:
duration: Time to mute, use ``None`` to mute forever
"""
if duration is None:
setting = "-1"
else:
setting = str(_util.timedelta_to_seconds(duration))
data = {"mute_settings": setting, "thread_fbid": self.id}
j = self.session._payload_post(
"/ajax/mercury/change_mute_thread.php?dpr=1", data
)
def unmute(self):
"""Unmute the thread."""
return self.mute(datetime.timedelta(0))
def _mute_reactions(self, mode: bool):
data = {"reactions_mute_mode": "1" if mode else "0", "thread_fbid": self.id}
j = self.session._payload_post(
"/ajax/mercury/change_reactions_mute_thread/?dpr=1", data
)
def mute_reactions(self):
"""Mute thread reactions."""
self._mute_reactions(True)
def unmute_reactions(self):
"""Unmute thread reactions."""
self._mute_reactions(False)
def _mute_mentions(self, mode: bool):
data = {"mentions_mute_mode": "1" if mode else "0", "thread_fbid": self.id}
j = self.session._payload_post(
"/ajax/mercury/change_mentions_mute_thread/?dpr=1", data
)
def mute_mentions(self):
"""Mute thread mentions."""
self._mute_mentions(True)
def unmute_mentions(self):
"""Unmute thread mentions."""
self._mute_mentions(False)
def mark_as_spam(self):
"""Mark the thread as spam, and delete it."""
data = {"id": self.id}
j = self.session._payload_post("/ajax/mercury/mark_spam.php?dpr=1", data)
def _forced_fetch(self, message_id: str) -> dict:
params = {
"thread_and_message_id": {"thread_id": self.id, "message_id": message_id}
}
(j,) = self.session._graphql_requests(
_graphql.from_doc_id("1768656253222505", params)
)
return j
@staticmethod
def _parse_customization_info(data):
def _parse_customization_info(data: Any) -> MutableMapping[str, Any]:
if data is None or data.get("customization_info") is None:
return {}
info = data["customization_info"]
@@ -109,6 +513,24 @@ class Thread:
rtn["own_nickname"] = pc[1].get("nickname")
return rtn
@attrs_default
class Thread(ThreadABC):
"""Represents a Facebook thread, where the actual type is unknown.
Implements parts of `ThreadABC`, call the method to figure out if your use case is
supported. Otherwise, you'll have to use an `User`/`Group`/`Page` object.
Note: This list may change in minor versions!
"""
#: The session to use when making requests.
session = attr.ib(type=_session.Session)
#: The unique identifier of the thread.
id = attr.ib(converter=str)
def _to_send_data(self):
# TODO: Only implement this in subclasses
return {"other_user_fbid": self.id}
raise NotImplementedError(
"The method you called is not supported on raw Thread objects."
" Please use an appropriate User/Group/Page object instead!"
)

45
fbchat/_user.py
View File

@@ -1,7 +1,6 @@
import attr
from ._core import attrs_default, Enum, Image
from . import _util, _plan
from ._thread import ThreadType, Thread
from . import _util, _session, _plan, _thread
GENDERS = {
@@ -42,11 +41,13 @@ class TypingStatus(Enum):
@attrs_default
class User(Thread):
"""Represents a Facebook user. Inherits `Thread`."""
type = ThreadType.USER
class User(_thread.ThreadABC):
"""Represents a Facebook user. Implements `ThreadABC`."""
#: The session to use when making requests.
session = attr.ib(type=_session.Session)
#: The user's unique identifier.
id = attr.ib(converter=str)
#: The user's picture
photo = attr.ib(None)
#: The name of the user
@@ -78,8 +79,31 @@ class User(Thread):
#: The default emoji
emoji = attr.ib(None)
def _to_send_data(self):
return {"other_user_fbid": self.id}
def confirm_friend_request(self):
"""Confirm a friend request, adding the user to your friend list."""
data = {"to_friend": self.id, "action": "confirm"}
j = self.session._payload_post("/ajax/add_friend/action.php?dpr=1", data)
def remove_friend(self):
"""Remove the user from the client's friend list."""
data = {"uid": self.id}
j = self.session._payload_post("/ajax/profile/removefriendconfirm.php", data)
def block(self):
"""Block messages from the user."""
data = {"fbid": self.id}
j = self.session._payload_post("/messaging/block_messages/?dpr=1", data)
def unblock(self):
"""Unblock a previously blocked user."""
data = {"fbid": self.id}
j = self.session._payload_post("/messaging/unblock_messages/?dpr=1", data)
@classmethod
def _from_graphql(cls, data):
def _from_graphql(cls, session, data):
if data.get("profile_picture") is None:
data["profile_picture"] = {}
c_info = cls._parse_customization_info(data)
@@ -88,6 +112,7 @@ class User(Thread):
plan = _plan.Plan._from_graphql(data["event_reminders"]["nodes"][0])
return cls(
session=session,
id=data["id"],
url=data.get("url"),
first_name=data.get("first_name"),
@@ -106,7 +131,7 @@ class User(Thread):
)
@classmethod
def _from_thread_fetch(cls, data):
def _from_thread_fetch(cls, session, data):
if data.get("big_image_src") is None:
data["big_image_src"] = {}
c_info = cls._parse_customization_info(data)
@@ -133,6 +158,7 @@ class User(Thread):
plan = _plan.Plan._from_graphql(data["event_reminders"]["nodes"][0])
return cls(
session=session,
id=user["id"],
url=user.get("url"),
name=user.get("name"),
@@ -152,8 +178,9 @@ class User(Thread):
)
@classmethod
def _from_all_fetch(cls, data):
def _from_all_fetch(cls, session, data):
return cls(
session=session,
id=data["id"],
first_name=data.get("firstName"),
url=data.get("uri"),

23
tests/conftest.py
View File

@@ -3,19 +3,24 @@ import json
from utils import *
from contextlib import contextmanager
from fbchat import ThreadType, Message, Mention
from fbchat import Message, Mention
@pytest.fixture(scope="session")
def session():
return object() # TODO: Add a mocked session
@pytest.fixture(scope="session")
def user(client2):
return {"id": client2.id, "type": ThreadType.USER}
return {"id": client2.id, "type": None}
@pytest.fixture(scope="session")
def group(pytestconfig):
return {
"id": load_variable("group_id", pytestconfig.cache),
"type": ThreadType.GROUP,
"type": None,
}
@@ -24,11 +29,9 @@ def group(pytestconfig):
params=["user", "group", pytest.param("none", marks=[pytest.mark.xfail()])],
)
def thread(request, user, group):
return {
"user": user,
"group": group,
"none": {"id": "0", "type": ThreadType.GROUP},
}[request.param]
return {"user": user, "group": group, "none": {"id": "0", "type": None},}[
request.param
]
@pytest.fixture(scope="session")
@@ -98,9 +101,7 @@ def compare(client, thread):
def inner(caught_event, **kwargs):
d = {
"author_id": client.id,
"thread_id": client.id
if thread["type"] == ThreadType.USER
else thread["id"],
"thread_id": client.id if thread["type"] == None else thread["id"],
"thread_type": thread["type"],
}
d.update(kwargs)

3
tests/test_fetch.py
View File

@@ -1,7 +1,7 @@
import pytest
from os import path
from fbchat import ThreadType, Message, Mention, EmojiSize, Sticker
from fbchat import Message, Mention, EmojiSize, Sticker
from utils import subset, STICKER_LIST, EMOJI_LIST
pytestmark = pytest.mark.online
@@ -89,7 +89,6 @@ def test_fetch_info(client1, group):
assert info.name == "Mark Zuckerberg"
info = client1.fetch_group_info(group["id"])[group["id"]]
assert info.type == ThreadType.GROUP
def test_fetch_image_url(client):

5
tests/test_group.py
View File

@@ -1,7 +1,7 @@
from fbchat._group import Group
def test_group_from_graphql():
def test_group_from_graphql(session):
data = {
"name": "Group ABC",
"thread_key": {"thread_fbid": "11223344"},
@@ -26,6 +26,7 @@ def test_group_from_graphql():
"event_reminders": {"nodes": []},
}
assert Group(
session=session,
id="11223344",
photo=None,
name="Group ABC",
@@ -40,4 +41,4 @@ def test_group_from_graphql():
approval_mode=False,
approval_requests=set(),
join_link="",
) == Group._from_graphql(data)
) == Group._from_graphql(session, data)

5
tests/test_page.py
View File

@@ -2,7 +2,7 @@ import fbchat
from fbchat._page import Page
def test_page_from_graphql():
def test_page_from_graphql(session):
data = {
"id": "123456",
"name": "Some school",
@@ -12,10 +12,11 @@ def test_page_from_graphql():
"city": None,
}
assert Page(
session=session,
id="123456",
photo=fbchat.Image(url="https://scontent-arn2-1.xx.fbcdn.net/v/..."),
name="Some school",
url="https://www.facebook.com/some-school/",
city=None,
category="SCHOOL",
) == Page._from_graphql(data)
) == Page._from_graphql(session, data)

4
tests/test_plans.py
View File

@@ -1,6 +1,6 @@
import pytest
from fbchat import Plan, FBchatFacebookError, ThreadType
from fbchat import Plan, FBchatFacebookError
from utils import random_hex, subset
from time import time
@@ -93,7 +93,7 @@ def test_on_plan_ended(client, thread, catch_event, compare):
x.wait(180)
assert subset(
x.res,
thread_id=client.id if thread["type"] == ThreadType.USER else thread["id"],
thread_id=client.id if thread["type"] is None else thread["id"],
thread_type=thread["type"],
)

16
tests/test_polls.py
View File

@@ -1,6 +1,6 @@
import pytest
from fbchat import Poll, PollOption, ThreadType
from fbchat import Poll, PollOption
from utils import random_hex, subset
pytestmark = pytest.mark.online
@@ -49,12 +49,7 @@ def poll_data(request, client1, group, catch_event):
def test_create_poll(client1, group, catch_event, poll_data):
event, poll, _ = poll_data
assert subset(
event,
author_id=client1.id,
thread_id=group["id"],
thread_type=ThreadType.GROUP,
)
assert subset(event, author_id=client1.id, thread=group)
assert subset(
vars(event["poll"]), title=poll.title, options_count=len(poll.options)
)
@@ -88,12 +83,7 @@ def test_update_poll_vote(client1, group, catch_event, poll_data):
new_options=new_options,
)
assert subset(
x.res,
author_id=client1.id,
thread_id=group["id"],
thread_type=ThreadType.GROUP,
)
assert subset(x.res, author_id=client1.id, thread=group)
assert subset(
vars(x.res["poll"]), title=poll.title, options_count=len(options + new_options)
)

2
tests/test_search.py
View File

@@ -1,5 +1,4 @@
import pytest
from fbchat import ThreadType
pytestmark = pytest.mark.online
@@ -11,7 +10,6 @@ def test_search_for(client1):
u = users[0]
assert u.id == "4"
assert u.type == ThreadType.USER
assert u.photo[:4] == "http"
assert u.url[:4] == "http"
assert u.name == "Mark Zuckerberg"

21
tests/test_thread.py
View File

@@ -1,12 +1,6 @@
import pytest
import fbchat
from fbchat._thread import ThreadType, ThreadColor, Thread
def test_thread_type_to_class():
assert fbchat.User == ThreadType.USER._to_class()
assert fbchat.Group == ThreadType.GROUP._to_class()
assert fbchat.Page == ThreadType.PAGE._to_class()
from fbchat import ThreadColor, ThreadABC, Thread
def test_thread_color_from_graphql():
@@ -19,8 +13,8 @@ def test_thread_color_from_graphql():
def test_thread_parse_customization_info_empty():
assert {} == Thread._parse_customization_info(None)
assert {} == Thread._parse_customization_info({"customization_info": None})
assert {} == ThreadABC._parse_customization_info(None)
assert {} == ThreadABC._parse_customization_info({"customization_info": None})
def test_thread_parse_customization_info_group():
@@ -43,7 +37,7 @@ def test_thread_parse_customization_info_group():
"color": ThreadColor.BRILLIANT_ROSE,
"nicknames": {"123456789": "A", "987654321": "B"},
}
assert expected == Thread._parse_customization_info(data)
assert expected == ThreadABC._parse_customization_info(data)
def test_thread_parse_customization_info_user():
@@ -62,4 +56,9 @@ def test_thread_parse_customization_info_user():
# ... Other irrelevant fields
}
expected = {"emoji": None, "color": None, "own_nickname": "A", "nickname": "B"}
assert expected == Thread._parse_customization_info(data)
assert expected == ThreadABC._parse_customization_info(data)
def test_thread_create_and_implements_thread_abc(session):
thread = Thread(session=session, id="123")
assert thread._parse_customization_info

12
tests/test_thread_interraction.py
View File

@@ -1,6 +1,6 @@
import pytest
from fbchat import Message, ThreadType, FBchatFacebookError, TypingStatus, ThreadColor
from fbchat import Message, FBchatFacebookError, TypingStatus, ThreadColor
from utils import random_hex, subset
from os import path
@@ -42,14 +42,8 @@ def test_remove_from_and_add_admins_to_group(client1, client2, group, catch_even
def test_change_title(client1, group, catch_event):
title = random_hex()
with catch_event("on_title_change") as x:
client1.change_thread_title(title, group["id"], thread_type=ThreadType.GROUP)
assert subset(
x.res,
author_id=client1.id,
new_title=title,
thread_id=group["id"],
thread_type=ThreadType.GROUP,
)
client1.change_thread_title(title, group["id"])
assert subset(x.res, author_id=client1.id, new_title=title, thread=group)
def test_change_nickname(client, client_all, catch_event, compare):

15
tests/test_user.py
View File

@@ -4,7 +4,7 @@ import fbchat
from fbchat._user import User, ActiveStatus
def test_user_from_graphql():
def test_user_from_graphql(session):
data = {
"id": "1234",
"name": "Abc Def Ghi",
@@ -17,6 +17,7 @@ def test_user_from_graphql():
"viewer_affinity": 0.4560002,
}
assert User(
session=session,
id="1234",
photo=fbchat.Image(url="https://scontent-arn2-1.xx.fbcdn.net/v/..."),
name="Abc Def Ghi",
@@ -26,10 +27,10 @@ def test_user_from_graphql():
is_friend=True,
gender="female_singular",
affinity=0.4560002,
) == User._from_graphql(data)
) == User._from_graphql(session, data)
def test_user_from_thread_fetch():
def test_user_from_thread_fetch(session):
data = {
"thread_key": {"thread_fbid": None, "other_user_id": "1234"},
"name": None,
@@ -138,6 +139,7 @@ def test_user_from_thread_fetch():
"delivery_receipts": ...,
}
assert User(
session=session,
id="1234",
photo=fbchat.Image(url="https://scontent-arn2-1.xx.fbcdn.net/v/..."),
name="Abc Def Ghi",
@@ -152,10 +154,10 @@ def test_user_from_thread_fetch():
own_nickname="B",
color=None,
emoji=None,
) == User._from_thread_fetch(data)
) == User._from_thread_fetch(session, data)
def test_user_from_all_fetch():
def test_user_from_all_fetch(session):
data = {
"id": "1234",
"name": "Abc Def Ghi",
@@ -176,6 +178,7 @@ def test_user_from_all_fetch():
"is_blocked": False,
}
assert User(
session=session,
id="1234",
photo=fbchat.Image(url="https://scontent-arn2-1.xx.fbcdn.net/v/..."),
name="Abc Def Ghi",
@@ -183,7 +186,7 @@ def test_user_from_all_fetch():
first_name="Abc",
is_friend=True,
gender="female_singular",
) == User._from_all_fetch(data)
) == User._from_all_fetch(session, data)
@pytest.mark.skip(reason="can't gather test data, the pulling is broken")

2
tests/utils.py
View File

@@ -5,7 +5,7 @@ import pytest
from os import environ
from random import randrange
from contextlib import contextmanager
from fbchat import ThreadType, EmojiSize, FBchatFacebookError, Sticker, Client
from fbchat import EmojiSize, FBchatFacebookError, Sticker, Client
log = logging.getLogger("fbchat.tests").addHandler(logging.NullHandler())