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

Refactor message sending

Browse Source
This commit is contained in:
Mads Marquart
2020-01-09 16:15:52 +01:00
parent 6d6f779d26
commit a1b3fd3ffa
8 changed files with 131 additions and 181 deletions
Download Patch File Download Diff File Expand all files Collapse all files

123
fbchat/_thread.py
View File

@@ -73,6 +73,17 @@ class ThreadABC(metaclass=abc.ABCMeta):
def _to_send_data(self) -> MutableMapping[str, str]:
raise NotImplementedError
# Note:
# You can go out of Facebook's spec with `self.session._do_send_request`!
#
# A few examples:
# - You can send a sticker and an emoji at the same time
# - You can wave, send a sticker and text at the same time
# - You can reply to a message with a sticker
#
# We won't support those use cases, it'll make for a confusing API!
# If we absolutely need to in the future, we can always add extra functionality
def wave(self, first: bool = True) -> str:
"""Wave hello to the thread.
@@ -85,73 +96,127 @@ class ThreadABC(metaclass=abc.ABCMeta):
"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.
def send_text(
self,
text: str,
mentions: Iterable["_message.Mention"] = None,
files: Iterable[Tuple[str, str]] = None,
reply_to_id: str = None,
) -> str:
"""Send a message to the thread.
Args:
message (Message): Message to send
text: Text to send
mentions: Optional mentions
files: Optional tuples, each containing an uploaded file's ID and mimetype
reply_to_id: Optional message to reply to
Returns:
:ref:`Message ID <intro_message_ids>` of the sent message
"""
data = self._to_send_data()
data.update(message._to_send_data())
data["action_type"] = "ma-type:user-generated-message"
if text is None: # To support `send_files`
data["body"] = text
for i, mention in enumerate(mentions or ()):
data.update(mention._to_send_data(i))
if files:
data["has_attachment"] = True
for i, (file_id, mimetype) in enumerate(files or ()):
data["{}s[{}]".format(_util.mimetype_to_key(mimetype), i)] = file_id
if reply_to_id:
data["replied_to_message_id"] = reply_to_id
return self.session._do_send_request(data)
def _send_location(self, current, latitude, longitude, message=None) -> str:
def send_emoji(self, emoji: str, size: "_message.EmojiSize") -> str:
"""Send an emoji to the thread.
Args:
emoji: The emoji to send
size: The size of the emoji
Returns:
:ref:`Message ID <intro_message_ids>` of the sent message
"""
data = self._to_send_data()
data["action_type"] = "ma-type:user-generated-message"
data["body"] = emoji
data["tags[0]"] = "hot_emoji_size:{}".format(size.name.lower())
return self.session._do_send_request(data)
def send_sticker(self, sticker_id: str) -> str:
"""Send a sticker to the thread.
Args:
sticker_id: ID of the sticker to send
Returns:
:ref:`Message ID <intro_message_ids>` of the sent message
"""
data = self._to_send_data()
data["action_type"] = "ma-type:user-generated-message"
data["sticker_id"] = sticker_id
return self.session._do_send_request(data)
def _send_location(self, current, latitude, longitude) -> 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):
def send_location(self, latitude: float, longitude: float):
"""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,
)
self._send_location(True, latitude=latitude, longitude=longitude)
def send_pinned_location(self, latitude: float, longitude: float, message=None):
def send_pinned_location(self, latitude: float, longitude: float):
"""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,
)
self._send_location(False, latitude=latitude, longitude=longitude)
def send_files(self, files: Iterable[Tuple[str, str]], message):
def send_files(self, files: Iterable[Tuple[str, str]]):
"""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
return self.send_text(text=None, files=files)
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)
# xmd = {"quick_replies": []}
# for quick_reply in quick_replies:
# # TODO: Move this to `_quick_reply.py`
# q = dict()
# q["content_type"] = quick_reply._type
# q["payload"] = quick_reply.payload
# q["external_payload"] = quick_reply.external_payload
# q["data"] = quick_reply.data
# if quick_reply.is_response:
# q["ignore_for_webhook"] = False
# if isinstance(quick_reply, _quick_reply.QuickReplyText):
# q["title"] = quick_reply.title
# if not isinstance(quick_reply, _quick_reply.QuickReplyLocation):
# q["image_url"] = quick_reply.image_url
# xmd["quick_replies"].append(q)
# if len(quick_replies) == 1 and quick_replies[0].is_response:
# xmd["quick_replies"] = xmd["quick_replies"][0]
# data["platform_xmd"] = json.dumps(xmd)
# TODO: This!
# def quick_reply(self, quick_reply, payload=None):