Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: add muted/blocked users api #180

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
68 changes: 68 additions & 0 deletions twikit/client/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -2401,6 +2401,40 @@ async def unblock_user(self, user_id: str) -> User:
response, _ = await self.v11.destroy_blocks(user_id)
return User(self, build_user_data(response))

async def get_blocked_users(
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suggestion: Consider refactoring get_blocked_users and get_muted_users to reduce duplication

The get_blocked_users and get_muted_users methods are very similar. Consider creating a generic method for fetching user lists and then implement these as thin wrappers around it. This would improve maintainability and reduce the chance of inconsistencies.

async def _get_user_list(self, endpoint: str, count: int = 20, cursor: str | None = None) -> Result[User]:
    # Implementation here

async def get_blocked_users(self, count: int = 20, cursor: str | None = None) -> Result[User]:
    return await self._get_user_list('blocks', count, cursor)

async def get_muted_users(self, count: int = 20, cursor: str | None = None) -> Result[User]:
    return await self._get_user_list('mutes', count, cursor)

self, count: int = 20, cursor: str | None = None
) -> Result[User]:
"""
Retrieves blocked users.

Parameters
----------
count : :class:`int`, default=20
The number of blocked users to retrieve.
cursor : :class:`str`, default=None
The cursor for pagination.

Returns
-------
Result[:class:`User`]
A Result object containing a list of User objects representing
blocked users.

Examples
--------
>>> blocked_users = await client.get_blocked_users()
>>> for user in blocked_users:
... print(user)
<User id="...">
<User id="...">
...
...
"""
user_id = await self.user_id()
return await self._get_user_friendship(
user_id, count, self.gql.blocked_accounts_all, cursor
)

async def mute_user(self, user_id: str) -> User:
"""
Mutes a user.
Expand Down Expand Up @@ -2443,6 +2477,40 @@ async def unmute_user(self, user_id: str) -> User:
response, _ = await self.v11.destroy_mutes(user_id)
return User(self, build_user_data(response))

async def get_muted_users(
self, count: int = 20, cursor: str | None = None
) -> Result[User]:
"""
Retrieves muted users.

Parameters
----------
count : :class:`int`, default=20
The number of muted users to retrieve.
cursor : :class:`str`, default=None
The cursor for pagination.

Returns
-------
Result[:class:`User`]
A Result object containing a list of User objects representing
muted users.

Examples
--------
>>> muted_users = await client.get_muted_users()
>>> for user in muted_users:
... print(user)
<User id="...">
<User id="...">
...
...
"""
user_id = await self.user_id()
return await self._get_user_friendship(
user_id, count, self.gql.muted_accounts, cursor
)

async def get_trends(
self,
category: Literal['trending', 'for-you', 'news', 'sports', 'entertainment'],
Expand Down
25 changes: 25 additions & 0 deletions twikit/client/gql.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
from typing import TYPE_CHECKING

from ..constants import (
BLOCKED_ACCOUNTS_ALL_FEATURES,
BOOKMARK_FOLDER_TIMELINE_FEATURES,
COMMUNITY_NOTE_FEATURES,
COMMUNITY_TWEETS_FEATURES,
Expand Down Expand Up @@ -96,6 +97,8 @@ def url(path):
MEMBERS_SLICE_TIMELINE_QUERY = url('KDAssJ5lafCy-asH4wm1dw/membersSliceTimeline_Query')
MODERATORS_SLICE_TIMELINE_QUERY = url('9KI_r8e-tgp3--N5SZYVjg/moderatorsSliceTimeline_Query')
COMMUNITY_TWEET_SEARCH_MODULE_QUERY = url('5341rmzzvdjqfmPKfoHUBw/CommunityTweetSearchModuleQuery')
BLOCKED_ACCOUNTS_ALL = url("ugCclQ08T0qMYjS3SYvMdQ/BlockedAccountsAll")
MUTED_ACCOUNTS = url("J0tjDZrm9M6UYRoPtXcvhg/MutedAccounts")


class GQLClient:
Expand Down Expand Up @@ -690,3 +693,25 @@ async def tweet_result_by_rest_id(self, tweet_id):
return await self.gql_get(
Endpoint.TWEET_RESULT_BY_REST_ID, variables, TWEET_RESULT_BY_REST_ID_FEATURES, extra_params=params
)

async def blocked_accounts_all(self, user_id, count, cursor):
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suggestion: Standardize coding style across similar methods

The blocked_accounts_all and muted_accounts methods are quite similar, but use slightly different styles for defining the variables dictionary. Consider standardizing the coding style across these methods for better consistency and readability.

Suggested change
async def blocked_accounts_all(self, user_id, count, cursor):
async def blocked_accounts_all(self, user_id: str, count: int, cursor: Optional[str] = None) -> Dict:
variables = {
"userId": user_id,
"count": count,
"cursor": cursor,
}

variables = {
"count": count,
"includePromotedContent": False,
"withSafetyModeUserFields": False,
}
if cursor is not None:
variables['cursor'] = cursor

return await self.gql_get(
Endpoint.BLOCKED_ACCOUNTS_ALL, variables, BLOCKED_ACCOUNTS_ALL_FEATURES
)

async def muted_accounts(self, user_id, count, cursor):
variables={"count":count,"includePromotedContent":False}
if cursor is not None:
variables['cursor'] = cursor

return await self.gql_get(
Endpoint.MUTED_ACCOUNTS, variables, BLOCKED_ACCOUNTS_ALL_FEATURES
)
26 changes: 26 additions & 0 deletions twikit/constants.py
Original file line number Diff line number Diff line change
Expand Up @@ -225,3 +225,29 @@
'longform_notetweets_inline_media_enabled': True,
'responsive_web_enhance_cards_enabled': False
}

BLOCKED_ACCOUNTS_ALL_FEATURES = {
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

issue (complexity): Consider using a single dictionary for feature flags.

The introduction of the BLOCKED_ACCOUNTS_ALL_FEATURES dictionary increases complexity due to duplication of many entries from the existing feature flags dictionary. This duplication can lead to maintenance challenges and potential inconsistencies if updates are needed. To simplify, consider using a single dictionary for feature flags and introducing logic to handle context-specific differences. This approach will reduce duplication, improve maintainability, and enhance readability.

"rweb_tipjar_consumption_enabled": True,
"responsive_web_graphql_exclude_directive_enabled": True,
"verified_phone_label_enabled": False,
"creator_subscriptions_tweet_preview_api_enabled": True,
"responsive_web_graphql_timeline_navigation_enabled": True,
"responsive_web_graphql_skip_user_profile_image_extensions_enabled": False,
"communities_web_enable_tweet_community_results_fetch": True,
"c9s_tweet_anatomy_moderator_badge_enabled": True,
"articles_preview_enabled": True,
"responsive_web_edit_tweet_api_enabled": True,
"graphql_is_translatable_rweb_tweet_is_translatable_enabled": True,
"view_counts_everywhere_api_enabled": True,
"longform_notetweets_consumption_enabled": True,
"responsive_web_twitter_article_tweet_consumption_enabled": True,
"tweet_awards_web_tipping_enabled": False,
"creator_subscriptions_quote_tweet_preview_enabled": False,
"freedom_of_speech_not_reach_fetch_enabled": True,
"standardized_nudges_misinfo": True,
"tweet_with_visibility_results_prefer_gql_limited_actions_policy_enabled": True,
"rweb_video_timestamps_enabled": True,
"longform_notetweets_rich_text_read_enabled": True,
"longform_notetweets_inline_media_enabled": True,
"responsive_web_enhance_cards_enabled": False,
}