Make Client.search_for_X more forwards compatible

This commit is contained in:
Mads Marquart
2020-01-14 22:06:12 +01:00
parent 22dcf6d69a
commit 2b45fdbc8a

View File

@@ -3,7 +3,7 @@ import time
import requests import requests
from ._core import log from ._core import log
from . import _util, _graphql, _session, _poll, _user, _thread, _message from . import _util, _graphql, _session, _poll, _user, _page, _group, _thread, _message
from ._exception import FBchatException, FBchatFacebookError from ._exception import FBchatException, FBchatFacebookError
from ._thread import ThreadLocation from ._thread import ThreadLocation
@@ -78,7 +78,7 @@ class Client:
users.append(_user.UserData._from_all_fetch(self.session, data)) users.append(_user.UserData._from_all_fetch(self.session, data))
return users return users
def search_for_users(self, name, limit=10): def search_for_users(self, name: str, limit: int) -> Iterable[_user.UserData]:
"""Find and get users by their name. """Find and get users by their name.
Args: Args:
@@ -87,92 +87,71 @@ class Client:
Returns: Returns:
list: `User` objects, ordered by relevance list: `User` objects, ordered by relevance
Raises:
FBchatException: If request failed
""" """
params = {"search": name, "limit": limit} params = {"search": name, "limit": limit}
(j,) = self.session._graphql_requests( (j,) = self.session._graphql_requests(
_graphql.from_query(_graphql.SEARCH_USER, params) _graphql.from_query(_graphql.SEARCH_USER, params)
) )
return [ return (
UserData._from_graphql(self.session, node) UserData._from_graphql(self.session, node)
for node in j[name]["users"]["nodes"] for node in j[name]["users"]["nodes"]
] )
def search_for_pages(self, name, limit=10): def search_for_pages(self, name: str, limit: int) -> Iterable[_page.PageData]:
"""Find and get pages by their name. """Find and get pages by their name.
Args: Args:
name: Name of the page name: Name of the page
limit: The max. amount of pages to fetch
Returns:
list: `Page` objects, ordered by relevance
Raises:
FBchatException: If request failed
""" """
params = {"search": name, "limit": limit} params = {"search": name, "limit": limit}
(j,) = self.session._graphql_requests( (j,) = self.session._graphql_requests(
_graphql.from_query(_graphql.SEARCH_PAGE, params) _graphql.from_query(_graphql.SEARCH_PAGE, params)
) )
return [ return (
PageData._from_graphql(self.session, node) PageData._from_graphql(self.session, node)
for node in j[name]["pages"]["nodes"] for node in j[name]["pages"]["nodes"]
] )
def search_for_groups(self, name, limit=10): def search_for_groups(self, name: str, limit: int) -> Iterable[_group.GroupData]:
"""Find and get group threads by their name. """Find and get group threads by their name.
Args: Args:
name: Name of the group thread name: Name of the group thread
limit: The max. amount of groups to fetch limit: The max. amount of groups to fetch
Returns:
list: `Group` objects, ordered by relevance
Raises:
FBchatException: If request failed
""" """
params = {"search": name, "limit": limit} params = {"search": name, "limit": limit}
(j,) = self.session._graphql_requests( (j,) = self.session._graphql_requests(
_graphql.from_query(_graphql.SEARCH_GROUP, params) _graphql.from_query(_graphql.SEARCH_GROUP, params)
) )
return [ return (
GroupData._from_graphql(self.session, node) GroupData._from_graphql(self.session, node)
for node in j["viewer"]["groups"]["nodes"] for node in j["viewer"]["groups"]["nodes"]
] )
def search_for_threads(self, name, limit=10): def search_for_threads(self, name: str, limit: int) -> Iterable[_thread.ThreadABC]:
"""Find and get threads by their name. """Find and get threads by their name.
Args: Args:
name: Name of the thread name: Name of the thread
limit: The max. amount of groups to fetch limit: The max. amount of threads to fetch
Returns:
list: `User`, `Group` and `Page` objects, ordered by relevance
Raises:
FBchatException: If request failed
""" """
params = {"search": name, "limit": limit} params = {"search": name, "limit": limit}
(j,) = self.session._graphql_requests( (j,) = self.session._graphql_requests(
_graphql.from_query(_graphql.SEARCH_THREAD, params) _graphql.from_query(_graphql.SEARCH_THREAD, params)
) )
rtn = []
for node in j[name]["threads"]["nodes"]: for node in j[name]["threads"]["nodes"]:
if node["__typename"] == "User": if node["__typename"] == "User":
rtn.append(UserData._from_graphql(self.session, node)) yield UserData._from_graphql(self.session, node)
elif node["__typename"] == "MessageThread": elif node["__typename"] == "MessageThread":
# MessageThread => Group thread # MessageThread => Group thread
rtn.append(GroupData._from_graphql(self.session, node)) yield GroupData._from_graphql(self.session, node)
elif node["__typename"] == "Page": elif node["__typename"] == "Page":
rtn.append(PageData._from_graphql(self.session, node)) yield PageData._from_graphql(self.session, node)
elif node["__typename"] == "Group": elif node["__typename"] == "Group":
# We don't handle Facebook "Groups" # We don't handle Facebook "Groups"
pass pass
@@ -181,8 +160,6 @@ class Client:
"Unknown type {} in {}".format(repr(node["__typename"]), node) "Unknown type {} in {}".format(repr(node["__typename"]), node)
) )
return rtn
def _search_messages(self, query, offset, limit): def _search_messages(self, query, offset, limit):
data = {"query": query, "offset": offset, "limit": limit} data = {"query": query, "offset": offset, "limit": limit}
j = self.session._payload_post("/ajax/mercury/search_snippets.php?dpr=1", data) j = self.session._payload_post("/ajax/mercury/search_snippets.php?dpr=1", data)