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

Move send methods to ThreadABC

Browse Source
This commit is contained in:
Mads Marquart
2020-01-09 00:25:40 +01:00
parent aeca4865ae
commit 13aa1f5e5a
2 changed files with 123 additions and 283 deletions
Download Patch File Download Diff File Expand all files Collapse all files

281
fbchat/_client.py
View File

@@ -845,111 +845,6 @@ class Client:
SEND METHODS
"""
def _old_message(self, message):
return message if isinstance(message, Message) else Message(text=message)
def _do_send_request(self, data, get_thread_id=False):
"""Send the data to `SendURL`, and returns the message ID or None on failure."""
mid, thread_id = self._session._do_send_request(data)
if get_thread_id:
return mid, thread_id
else:
return mid
def send(self, message, thread_id=None, thread_type=ThreadType.USER):
"""Send message to a thread.
Args:
message (Message): Message to send
thread_id: User/Group ID to send to. See :ref:`intro_threads`
thread_type (ThreadType): See :ref:`intro_threads`
Returns:
:ref:`Message ID <intro_message_ids>` of the sent message
Raises:
FBchatException: If request failed
"""
thread = thread_type._to_class()(id=thread_id)
data = thread._to_send_data()
data.update(message._to_send_data())
return self._do_send_request(data)
def wave(self, wave_first=True, thread_id=None, thread_type=None):
"""Wave hello to a thread.
Args:
wave_first: Whether to wave first or wave back
thread_id: User/Group ID to send to. See :ref:`intro_threads`
thread_type (ThreadType): See :ref:`intro_threads`
Returns:
:ref:`Message ID <intro_message_ids>` of the sent message
Raises:
FBchatException: If request failed
"""
thread = thread_type._to_class()(id=thread_id)
data = thread._to_send_data()
data["action_type"] = "ma-type:user-generated-message"
data["lightweight_action_attachment[lwa_state]"] = (
"INITIATED" if wave_first else "RECIPROCATED"
)
data["lightweight_action_attachment[lwa_type]"] = "WAVE"
if thread_type == ThreadType.USER:
data["specific_to_list[0]"] = "fbid:{}".format(thread_id)
return self._do_send_request(data)
def quick_reply(self, quick_reply, payload=None, thread_id=None, thread_type=None):
"""Reply to chosen quick reply.
Args:
quick_reply (QuickReply): Quick reply to reply to
payload: Optional answer to the quick reply
thread_id: User/Group ID to send to. See :ref:`intro_threads`
thread_type (ThreadType): See :ref:`intro_threads`
Returns:
:ref:`Message ID <intro_message_ids>` of the sent message
Raises:
FBchatException: If request failed
"""
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, thread_id=thread_id, thread_type=thread_type
)
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 unsend(self, mid):
"""Unsend message by it's ID (removes it for everyone).
@@ -959,182 +854,6 @@ class Client:
data = {"message_id": mid}
j = self._payload_post("/messaging/unsend_message/?dpr=1", data)
def _send_location(
self, location, current=True, message=None, thread_id=None, thread_type=None
):
thread = thread_type._to_class()(id=thread_id)
data = thread._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]"] = location.latitude
data["location_attachment[coordinates][longitude]"] = location.longitude
data["location_attachment[is_current_location]"] = current
return self._do_send_request(data)
def send_location(self, location, message=None, thread_id=None, thread_type=None):
"""Send a given location to a thread as the user's current location.
Args:
location (LocationAttachment): Location to send
message (Message): Additional message
thread_id: User/Group ID to send to. See :ref:`intro_threads`
thread_type (ThreadType): See :ref:`intro_threads`
Returns:
:ref:`Message ID <intro_message_ids>` of the sent message
Raises:
FBchatException: If request failed
"""
self._send_location(
location=location,
current=True,
message=message,
thread_id=thread_id,
thread_type=thread_type,
)
def send_pinned_location(
self, location, message=None, thread_id=None, thread_type=None
):
"""Send a given location to a thread as a pinned location.
Args:
location (LocationAttachment): Location to send
message (Message): Additional message
thread_id: User/Group ID to send to. See :ref:`intro_threads`
thread_type (ThreadType): See :ref:`intro_threads`
Returns:
:ref:`Message ID <intro_message_ids>` of the sent message
Raises:
FBchatException: If request failed
"""
self._send_location(
location=location,
current=False,
message=message,
thread_id=thread_id,
thread_type=thread_type,
)
def _upload(self, files, voice_clip=False):
return self._session._upload(files, voice_clip=voice_clip)
def _send_files(
self, files, message=None, thread_id=None, thread_type=ThreadType.USER
):
"""Send files from file IDs to a thread.
`files` should be a list of tuples, with a file's ID and mimetype.
"""
thread = thread_type._to_class()(id=thread_id)
data = thread._to_send_data()
data.update(self._old_message(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._do_send_request(data)
def send_remote_files(
self, file_urls, message=None, thread_id=None, thread_type=ThreadType.USER
):
"""Send files from URLs to a thread.
Args:
file_urls: URLs of files to upload and send
message: Additional message
thread_id: User/Group ID to send to. See :ref:`intro_threads`
thread_type (ThreadType): See :ref:`intro_threads`
Returns:
:ref:`Message ID <intro_message_ids>` of the sent files
Raises:
FBchatException: If request failed
"""
file_urls = _util.require_list(file_urls)
files = self._upload(_util.get_files_from_urls(file_urls))
return self._send_files(
files=files, message=message, thread_id=thread_id, thread_type=thread_type
)
def send_local_files(
self, file_paths, message=None, thread_id=None, thread_type=ThreadType.USER
):
"""Send local files to a thread.
Args:
file_paths: Paths of files to upload and send
message: Additional message
thread_id: User/Group ID to send to. See :ref:`intro_threads`
thread_type (ThreadType): See :ref:`intro_threads`
Returns:
:ref:`Message ID <intro_message_ids>` of the sent files
Raises:
FBchatException: If request failed
"""
file_paths = _util.require_list(file_paths)
with _util.get_files_from_paths(file_paths) as x:
files = self._upload(x)
return self._send_files(
files=files, message=message, thread_id=thread_id, thread_type=thread_type
)
def send_remote_voice_clips(
self, clip_urls, message=None, thread_id=None, thread_type=ThreadType.USER
):
"""Send voice clips from URLs to a thread.
Args:
clip_urls: URLs of clips to upload and send
message: Additional message
thread_id: User/Group ID to send to. See :ref:`intro_threads`
thread_type (ThreadType): See :ref:`intro_threads`
Returns:
:ref:`Message ID <intro_message_ids>` of the sent files
Raises:
FBchatException: If request failed
"""
clip_urls = _util.require_list(clip_urls)
files = self._upload(_util.get_files_from_urls(clip_urls), voice_clip=True)
return self._send_files(
files=files, message=message, thread_id=thread_id, thread_type=thread_type
)
def send_local_voice_clips(
self, clip_paths, message=None, thread_id=None, thread_type=ThreadType.USER
):
"""Send local voice clips to a thread.
Args:
clip_paths: Paths of clips to upload and send
message: Additional message
thread_id: User/Group ID to send to. See :ref:`intro_threads`
thread_type (ThreadType): See :ref:`intro_threads`
Returns:
:ref:`Message ID <intro_message_ids>` of the sent files
Raises:
FBchatException: If request failed
"""
clip_paths = _util.require_list(clip_paths)
with _util.get_files_from_paths(clip_paths) as x:
files = self._upload(x, voice_clip=True)
return self._send_files(
files=files, message=message, thread_id=thread_id, thread_type=thread_type
)
def forward_attachment(self, attachment_id, thread_id=None):
"""Forward an attachment.

125
fbchat/_thread.py
View File

@@ -1,8 +1,8 @@
import abc
import attr
from ._core import attrs_default, Enum, Image
from . import _session
from typing import MutableMapping, Any
from . import _util, _session
from typing import MutableMapping, Any, Iterable, Tuple
class ThreadType(Enum):
@@ -92,6 +92,127 @@ class ThreadABC(metaclass=abc.ABCMeta):
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 thread_type == ThreadType.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 _forced_fetch(self, message_id: str) -> dict:
params = {
"thread_and_message_id": {"thread_id": self.id, "message_id": message_id}