简易、强大的推特（Twitter）采集程序,支持元搜索,用户,粉丝,关注,发文,回复,评论等采集
from .twitter_crawler import set_proxy, set_cookie, common_crawler, user_crawler, search_crawler
from .twitter_utils import TwitterFilter

__version__ = '1.0.04'

easy_twitter_crawler/__init__.py
hanxinkong/easy-twitter-crawler
from .twitter_types import *
from .twitter_scraper import TwitterScraper
from .constants import *

_scraper = TwitterScraper()


def set_cookie(cookie=None):
    """
    :param cookie: cookie 字典；格式为{'d_c0': '.....'}；
    """
    if cookie is None:
        cookie = {}

    cookies = []
    for key in cookie.keys():
        value = cookie[key]
        cookie_str = f'{key}={value}'
        cookies.append(cookie_str)
    _scraper.default_headers['cookie'] = '; '.join(cookies)


def set_proxy(proxy: Optional[Dict[str, str]] = None):
    """
    设置代理；每次请求都切换一次代理才算最佳
    """
    _scraper.set_proxy(proxy)


def _set_timeout(timeout: int):
    assert isinstance(timeout, int), 'timeout值应为大于0的整数'
    if timeout < 5 or timeout < 0:
        timeout = DEFAULT_REQUESTS_TIMEOUT
    _scraper.set_timeout(timeout=timeout)


def search_crawler(key_word: Optional[str] = None,
                   count: Optional[int] = 0,
                   comment_count: Optional[int] = -1,
                   **kwargs) -> Union[Iterator[UserType],
Iterator[ArticleType],
Iterator[VideoType],
Iterator[PhotoType]]:
    """
    关键词搜索采集（...）
    :param key_word: 关键词
    :param kwargs:
    :param count: (int) 采集指定数量的。该值过大可能会导致多次请求。-1 不采集 0采集全部（默认） >0采集指定的数量
    :param comment_count: (int) 采集指定数量的评论。该值过大可能会导致多次请求。默认-1 不采集 0采集全部 >0采集指定的数量
    @ page_limit (int): 需要采集的页数, 默认为constants下的 DEFAULT_PAGE_LIMIT
    @ data_type:（str or list or set or tuple） 获取数据类型 可选择（热门：Top 用户：People 最新：Latest 视频：Videos 照片：Photos） 大小写均可，默认所有类型都会采集
    @ drill_down_count: (int) 下钻内容采集数量(问题下的回答)，，默认-1 不采集 0采集全部 >0采集指定的数量
    :return:
    """
    _set_timeout(kwargs.pop('timeout', DEFAULT_REQUESTS_TIMEOUT))
    _scraper.requests_kwargs['timeout'] = kwargs.pop('timeout', DEFAULT_REQUESTS_TIMEOUT)
    options: Union[Dict[str, Any], Set[str]] = kwargs.setdefault('options', {})
    if isinstance(options, set):
        options = {k: True for k in options}
    options.setdefault('key_word', key_word)
    options['drill_down_count'] = kwargs.pop('drill_down_count', -1)
    options['comment_count'] = comment_count
    data_types = kwargs.get('data_type', ALL_SEARCH_TYPE)
    kwargs['data_type'] = options['data_type'] = data_types
    kwargs['sort'] = options['sort'] = kwargs.get('sort', None)
    kwargs['time_interval'] = options['time_interval'] = kwargs.get('time_interval', None)
    kwargs['total_count'] = count

    data_types = [data_types] if isinstance(data_types, str) else data_types
    if isinstance(data_types, list):
        for data_type in data_types:
            kwargs['data_type'] = data_type.capitalize()
            for result in _scraper.search_crawler(key_word, **kwargs):
                yield result


def common_crawler(task_id: Union[str],
                   data_type: Optional[str] = None,
                   drill_down_count: Optional[int] = -1,
                   comment_count: Optional[int] = -1,
                   **kwargs) -> Union[Iterator[CommentType],
Iterator[ArticleType]]:
    """
    通用采集()
    :param task_id:
    :param data_type: 指定数据类型的采集 (article or comment)
    :param drill_down_count: (int) 下钻内容采集数量 (文章可下钻采集评论)，默认-1 不采集 0采集全部 >0采集指定的数量
    :param comment_count: (int) 采集指定数量的评论。该值过大可能会导致多次请求;默认-1 不采集 0采集全部 >0采集指定的数量
    :return:
    """
    _set_timeout(kwargs.pop('timeout', DEFAULT_REQUESTS_TIMEOUT))
    options: Union[Dict[str, Any], Set[str]] = kwargs.setdefault('options', {})
    if isinstance(options, set):
        options = {k: True for k in options}
    if data_type:
        options['data_type'] = data_type
    kwargs['pubdate_sort'] = kwargs.get('pubdate_sort', True)
    options['drill_down_count'] = kwargs['drill_down_count'] = drill_down_count
    options['comment_count'] = comment_count
    options['task_id'] = task_id
    options['start_time'] = kwargs.get('start_time', '')
    options['end_time'] = kwargs.get('end_time', '')
    kwargs['total_count'] = comment_count

    assert data_type in (ARTICLE, COMMENT), '匹配不到可以采集的数据类型，请校对data_type的值'

    if data_type == ARTICLE:
        return _scraper.article_crawler(article_id=task_id, **kwargs)
    elif data_type == COMMENT:
        return _scraper.comment_crawler(article_id=task_id, **kwargs)


def user_crawler(user_id: Union[str],
                 **kwargs) -> Iterator[UserType]:
    """
    账号采集
    :param user_id: (str) 账号id 如 https://twitter.com/phivolcs_dost中 phivolcs_dost中为user_id
                    数据api：

    @ followers_count： 是否采集该账号的关注列表(粉丝) -1不采集 默认。0采集全部（可能会导致多次请求）；>0将采集指定数量
    @ following_count： 是否采集该账号的好友列表（关注别人） -1不采集 默认。0采集全部（可能会导致多次请求）；>0将采集指定数量
    @ article_count： 是否采集该账号的文章列表 -1不采集 默认。0采集全部（可能会导致多次请求）；>0将采集指定数量
    @ drill_down_count: 是否向下钻取内容（）-1不采集 默认。0采集全部（可能会导致多次请求）；>0将采集指定数量
    """
    _set_timeout(kwargs.pop('timeout', DEFAULT_REQUESTS_TIMEOUT))
    options: Union[Dict[str, Any], Set[str]] = kwargs.setdefault('options', {})
    if isinstance(options, set):
        options = {k: True for k in options}
    options['followers_count'] = kwargs.pop('followers_count', -1)
    options['following_count'] = kwargs.pop('following_count', -1)
    options['article_count'] = kwargs.pop('article_count', -1)
    options['reply_count'] = kwargs.pop('reply_count', -1)
    options['drill_down_count'] = kwargs.pop('drill_down_count', -1)
    kwargs['sort'] = options['sort'] = kwargs.get('sort', 'created')
    options['user_id'] = user_id
    options['start_time'] = kwargs.get('start_time', '')
    options['end_time'] = kwargs.get('end_time', '')

    return _scraper.user_crawler(user_id, **kwargs)

easy_twitter_crawler/twitter_crawler/__init__.py
hanxinkong/easy-twitter-crawler
# 推特基础URL
BASE_URL = 'https://twitter.com/'
# 推特基础api
BASE_API = 'https://twitter.com/i/api/'

# 推特搜索页基础api (热门：Top 用户：People 最新：Latest 视频：Videos 照片：Photos)
# SEARCH_URL = 'https://twitter.com/i/api/graphql/GcjM7tlxA-EAM98COHsYwg/SearchTimeline?variables=%7B%22rawQuery%22%3A%22{key_word}%22%2C%22count%22%3A20%2C%22cursor%22%3A%22{cursor}%22%2C%22querySource%22%3A%22typed_query%22%2C%22product%22%3A%22{data_type}%22%7D%26features%3D%7B%22rweb_lists_timeline_redesign_enabled%22%3Atrue%2C%22responsive_web_graphql_exclude_directive_enabled%22%3Atrue%2C%22verified_phone_label_enabled%22%3Afalse%2C%22creator_subscriptions_tweet_preview_api_enabled%22%3Atrue%2C%22responsive_web_graphql_timeline_navigation_enabled%22%3Atrue%2C%22responsive_web_graphql_skip_user_profile_image_extensions_enabled%22%3Afalse%2C%22tweetypie_unmention_optimization_enabled%22%3Atrue%2C%22responsive_web_edit_tweet_api_enabled%22%3Atrue%2C%22graphql_is_translatable_rweb_tweet_is_translatable_enabled%22%3Atrue%2C%22view_counts_everywhere_api_enabled%22%3Atrue%2C%22longform_notetweets_consumption_enabled%22%3Atrue%2C%22tweet_awards_web_tipping_enabled%22%3Afalse%2C%22freedom_of_speech_not_reach_fetch_enabled%22%3Atrue%2C%22standardized_nudges_misinfo%22%3Atrue%2C%22tweet_with_visibility_results_prefer_gql_limited_actions_policy_enabled%22%3Atrue%2C%22longform_notetweets_rich_text_read_enabled%22%3Atrue%2C%22longform_notetweets_inline_media_enabled%22%3Atrue%2C%22responsive_web_enhance_cards_enabled%22%3Afalse%7D'
SEARCH_URL = 'https://twitter.com/i/api/graphql/GcjM7tlxA-EAM98COHsYwg/SearchTimeline?variables=%7B%22rawQuery%22%3A%22{key_word}%22%2C%22count%22%3A40%2C%22cursor%22%3A%22{cursor}%22%2C%22querySource%22%3A%22typed_query%22%2C%22product%22%3A%22{data_type}%22%7D&features=%7B%22rweb_lists_timeline_redesign_enabled%22%3Atrue%2C%22responsive_web_graphql_exclude_directive_enabled%22%3Atrue%2C%22verified_phone_label_enabled%22%3Afalse%2C%22creator_subscriptions_tweet_preview_api_enabled%22%3Atrue%2C%22responsive_web_graphql_timeline_navigation_enabled%22%3Atrue%2C%22responsive_web_graphql_skip_user_profile_image_extensions_enabled%22%3Afalse%2C%22tweetypie_unmention_optimization_enabled%22%3Atrue%2C%22responsive_web_edit_tweet_api_enabled%22%3Atrue%2C%22graphql_is_translatable_rweb_tweet_is_translatable_enabled%22%3Atrue%2C%22view_counts_everywhere_api_enabled%22%3Atrue%2C%22longform_notetweets_consumption_enabled%22%3Atrue%2C%22tweet_awards_web_tipping_enabled%22%3Afalse%2C%22freedom_of_speech_not_reach_fetch_enabled%22%3Atrue%2C%22standardized_nudges_misinfo%22%3Atrue%2C%22tweet_with_visibility_results_prefer_gql_limited_actions_policy_enabled%22%3Atrue%2C%22longform_notetweets_rich_text_read_enabled%22%3Atrue%2C%22longform_notetweets_inline_media_enabled%22%3Atrue%2C%22responsive_web_enhance_cards_enabled%22%3Afalse%7D'

# 用户url
USER_BASE_URL = 'https://twitter.com/{user_name}'

# 个人基础信息URL
USER_BASE_API = 'https://twitter.com/i/api/graphql/rePnxwe9LZ51nQ7Sn_xN_A/UserByScreenName?variables=%7B%22screen_name%22%3A%22{user_name}%22%2C%22withSafetyModeUserFields%22%3Atrue%2C%22withSuperFollowsUserFields%22%3Atrue%7D&features=%7B%22responsive_web_twitter_blue_verified_badge_is_enabled%22%3Atrue%2C%22responsive_web_graphql_exclude_directive_enabled%22%3Afalse%2C%22verified_phone_label_enabled%22%3Afalse%2C%22responsive_web_graphql_skip_user_profile_image_extensions_enabled%22%3Afalse%2C%22responsive_web_graphql_timeline_navigation_enabled%22%3Atrue%7D'
# 用户推文
USER_ARTICLE_URL = 'https://api.twitter.com/graphql/sj-BEQ0Jq5AwrydqFstdvg/UserTweets?variables=%7B%22userId%22%3A%22{user_id}%22%2C%22count%22%3A40%2C%22{cursor}includePromotedContent%22%3Atrue%2C%22withQuickPromoteEligibilityTweetFields%22%3Atrue%2C%22withSuperFollowsUserFields%22%3Atrue%2C%22withDownvotePerspective%22%3Afalse%2C%22withReactionsMetadata%22%3Afalse%2C%22withReactionsPerspective%22%3Afalse%2C%22withSuperFollowsTweetFields%22%3Atrue%2C%22withVoice%22%3Atrue%2C%22withV2Timeline%22%3Atrue%7D&features=%7B%22responsive_web_twitter_blue_verified_badge_is_enabled%22%3Atrue%2C%22verified_phone_label_enabled%22%3Afalse%2C%22responsive_web_graphql_timeline_navigation_enabled%22%3Atrue%2C%22longform_notetweets_consumption_enabled%22%3Afalse%2C%22tweetypie_unmention_optimization_enabled%22%3Atrue%2C%22vibe_api_enabled%22%3Atrue%2C%22responsive_web_edit_tweet_api_enabled%22%3Atrue%2C%22graphql_is_translatable_rweb_tweet_is_translatable_enabled%22%3Atrue%2C%22view_counts_everywhere_api_enabled%22%3Atrue%2C%22standardized_nudges_misinfo%22%3Atrue%2C%22tweet_with_visibility_results_prefer_gql_limited_actions_policy_enabled%22%3Afalse%2C%22interactive_text_enabled%22%3Atrue%2C%22responsive_web_text_conversations_enabled%22%3Afalse%2C%22responsive_web_enhance_cards_enabled%22%3Afalse%7D'
# 用户回复
USER_REPLY_URL = 'https://twitter.com/i/api/graphql/1-5o8Qhfc2kWlu_2rWNcug/UserTweetsAndReplies?variables=%7B%22userId%22%3A%22{user_id}%22%2C%22count%22%3A20%2C%22cursor%22%3A%22{cursor}%22%2C%22includePromotedContent%22%3Atrue%2C%22withCommunity%22%3Atrue%2C%22withVoice%22%3Atrue%2C%22withV2Timeline%22%3Atrue%7D&features=%7B%22rweb_lists_timeline_redesign_enabled%22%3Afalse%2C%22responsive_web_graphql_exclude_directive_enabled%22%3Atrue%2C%22verified_phone_label_enabled%22%3Afalse%2C%22creator_subscriptions_tweet_preview_api_enabled%22%3Atrue%2C%22responsive_web_graphql_timeline_navigation_enabled%22%3Atrue%2C%22responsive_web_graphql_skip_user_profile_image_extensions_enabled%22%3Afalse%2C%22tweetypie_unmention_optimization_enabled%22%3Atrue%2C%22responsive_web_edit_tweet_api_enabled%22%3Atrue%2C%22graphql_is_translatable_rweb_tweet_is_translatable_enabled%22%3Atrue%2C%22view_counts_everywhere_api_enabled%22%3Atrue%2C%22longform_notetweets_consumption_enabled%22%3Atrue%2C%22responsive_web_twitter_article_tweet_consumption_enabled%22%3Afalse%2C%22tweet_awards_web_tipping_enabled%22%3Afalse%2C%22freedom_of_speech_not_reach_fetch_enabled%22%3Atrue%2C%22standardized_nudges_misinfo%22%3Atrue%2C%22tweet_with_visibility_results_prefer_gql_limited_actions_policy_enabled%22%3Atrue%2C%22longform_notetweets_rich_text_read_enabled%22%3Atrue%2C%22longform_notetweets_inline_media_enabled%22%3Atrue%2C%22responsive_web_media_download_video_enabled%22%3Afalse%2C%22responsive_web_enhance_cards_enabled%22%3Afalse%7D&fieldToggles=%7B%22withAuxiliaryUserLabels%22%3Afalse%2C%22withArticleRichContentState%22%3Afalse%7D'
# 用户媒体
USER_MEDIA_URL = ''
# 用户喜欢
USER_LIKES_URL = ''
# 用户粉丝
# USER_FOLLOWERS_API = 'https://api.twitter.com/graphql/6AB4F2ntV7x0Oad-FvRHbw/Followers?variables=%7B%22userId%22%3A%22{user_id}%22%2C%22count%22%3A20%2C{cursor}%22includePromotedContent%22%3Afalse%2C%22withSuperFollowsUserFields%22%3Atrue%2C%22withDownvotePerspective%22%3Afalse%2C%22withReactionsMetadata%22%3Afalse%2C%22withReactionsPerspective%22%3Afalse%2C%22withSuperFollowsTweetFields%22%3Atrue%7D&features=%7B%22responsive_web_twitter_blue_verified_badge_is_enabled%22%3Atrue%2C%22responsive_web_graphql_exclude_directive_enabled%22%3Afalse%2C%22verified_phone_label_enabled%22%3Afalse%2C%22responsive_web_graphql_timeline_navigation_enabled%22%3Atrue%2C%22responsive_web_graphql_skip_user_profile_image_extensions_enabled%22%3Afalse%2C%22tweetypie_unmention_optimization_enabled%22%3Atrue%2C%22vibe_api_enabled%22%3Atrue%2C%22responsive_web_edit_tweet_api_enabled%22%3Atrue%2C%22graphql_is_translatable_rweb_tweet_is_translatable_enabled%22%3Atrue%2C%22view_counts_everywhere_api_enabled%22%3Atrue%2C%22longform_notetweets_consumption_enabled%22%3Atrue%2C%22freedom_of_speech_not_reach_appeal_label_enabled%22%3Afalse%2C%22standardized_nudges_misinfo%22%3Atrue%2C%22tweet_with_visibility_results_prefer_gql_limited_actions_policy_enabled%22%3Afalse%2C%22interactive_text_enabled%22%3Atrue%2C%22responsive_web_text_conversations_enabled%22%3Afalse%2C%22responsive_web_enhance_cards_enabled%22%3Afalse%7D'
USER_FOLLOWERS_API = 'https://api.twitter.com/1.1/followers/list.json?include_can_media_tag=1&skip_status=1&cards_platform=Web-12&include_cards=1&include_ext_alt_text=True&include_quote_count=True&include_reply_count=1&include_entities=True&include_user_entities=True&send_error_codes=True&simple_quoted_tweet=True&include_tweet_replies=True&tweet_mode=extended&include_card_uri=True&include_rts=False&exclude_replies=False&trim_user=False&user_id={user_id}&count=200'

# 用户关注
# USER_FOLLOWING_API = 'https://api.twitter.com/graphql/n1HqDGhrfjoOv2VUOtjbtg/Following?variables=%7B%22userId%22%3A%22{user_id}%22%2C%22count%22%3A20%2C{cursor}%22includePromotedContent%22%3Afalse%2C%22withSuperFollowsUserFields%22%3Atrue%2C%22withDownvotePerspective%22%3Afalse%2C%22withReactionsMetadata%22%3Afalse%2C%22withReactionsPerspective%22%3Afalse%2C%22withSuperFollowsTweetFields%22%3Atrue%7D&features=%7B%22responsive_web_twitter_blue_verified_badge_is_enabled%22%3Atrue%2C%22responsive_web_graphql_exclude_directive_enabled%22%3Afalse%2C%22verified_phone_label_enabled%22%3Afalse%2C%22responsive_web_graphql_timeline_navigation_enabled%22%3Atrue%2C%22responsive_web_graphql_skip_user_profile_image_extensions_enabled%22%3Afalse%2C%22tweetypie_unmention_optimization_enabled%22%3Atrue%2C%22vibe_api_enabled%22%3Atrue%2C%22responsive_web_edit_tweet_api_enabled%22%3Atrue%2C%22graphql_is_translatable_rweb_tweet_is_translatable_enabled%22%3Atrue%2C%22view_counts_everywhere_api_enabled%22%3Atrue%2C%22longform_notetweets_consumption_enabled%22%3Atrue%2C%22freedom_of_speech_not_reach_appeal_label_enabled%22%3Afalse%2C%22standardized_nudges_misinfo%22%3Atrue%2C%22tweet_with_visibility_results_prefer_gql_limited_actions_policy_enabled%22%3Afalse%2C%22interactive_text_enabled%22%3Atrue%2C%22responsive_web_text_conversations_enabled%22%3Afalse%2C%22responsive_web_enhance_cards_enabled%22%3Afalse%7D'
USER_FOLLOWING_API = 'https://api.twitter.com/1.1/friends/list.json?include_can_media_tag=1&skip_status=1&cards_platform=Web-12&include_cards=1&include_ext_alt_text=True&include_quote_count=True&include_reply_count=1&include_entities=True&include_user_entities=True&send_error_codes=True&simple_quoted_tweet=True&include_tweet_replies=True&tweet_mode=extended&include_card_uri=True&include_rts=False&exclude_replies=False&trim_user=False&user_id={user_id}&count=200'

# 文章URL
ARTICLE_URL = 'https://twitter.com/{user_name}/status/{article_id}'
ARTICLE_API = 'https://twitter.com/i/api/graphql/2ICDjqPd81tulZcYrtpTuQ/TweetResultByRestId?variables=%7B%22tweetId%22%3A%22{article_id}%22%2C%22withCommunity%22%3Afalse%2C%22includePromotedContent%22%3Afalse%2C%22withVoice%22%3Afalse%7D&features=%7B%22creator_subscriptions_tweet_preview_api_enabled%22%3Atrue%2C%22tweetypie_unmention_optimization_enabled%22%3Atrue%2C%22responsive_web_edit_tweet_api_enabled%22%3Atrue%2C%22graphql_is_translatable_rweb_tweet_is_translatable_enabled%22%3Atrue%2C%22view_counts_everywhere_api_enabled%22%3Atrue%2C%22longform_notetweets_consumption_enabled%22%3Atrue%2C%22responsive_web_twitter_article_tweet_consumption_enabled%22%3Afalse%2C%22tweet_awards_web_tipping_enabled%22%3Afalse%2C%22freedom_of_speech_not_reach_fetch_enabled%22%3Atrue%2C%22standardized_nudges_misinfo%22%3Atrue%2C%22tweet_with_visibility_results_prefer_gql_limited_actions_policy_enabled%22%3Atrue%2C%22longform_notetweets_rich_text_read_enabled%22%3Atrue%2C%22longform_notetweets_inline_media_enabled%22%3Atrue%2C%22responsive_web_graphql_exclude_directive_enabled%22%3Atrue%2C%22verified_phone_label_enabled%22%3Afalse%2C%22responsive_web_media_download_video_enabled%22%3Afalse%2C%22responsive_web_graphql_skip_user_profile_image_extensions_enabled%22%3Afalse%2C%22responsive_web_graphql_timeline_navigation_enabled%22%3Atrue%2C%22responsive_web_enhance_cards_enabled%22%3Afalse%7D'

# 评论API
COMMENT_API = 'https://api.twitter.com/graphql/6lWNh96EXDJCXl05SAtn_g/TweetDetail?variables=%7B%22focalTweetId%22%3A%22{article_id}%22%2C%22{cursor}with_rux_injections%22%3Atrue%2C%22includePromotedContent%22%3Atrue%2C%22withCommunity%22%3Atrue%2C%22withQuickPromoteEligibilityTweetFields%22%3Atrue%2C%22withBirdwatchNotes%22%3Afalse%2C%22withSuperFollowsUserFields%22%3Atrue%2C%22withDownvotePerspective%22%3Afalse%2C%22withReactionsMetadata%22%3Afalse%2C%22withReactionsPerspective%22%3Afalse%2C%22withSuperFollowsTweetFields%22%3Atrue%2C%22withVoice%22%3Atrue%2C%22withV2Timeline%22%3Atrue%7D&features=%7B%22responsive_web_twitter_blue_verified_badge_is_enabled%22%3Atrue%2C%22verified_phone_label_enabled%22%3Afalse%2C%22responsive_web_graphql_timeline_navigation_enabled%22%3Atrue%2C%22longform_notetweets_consumption_enabled%22%3Afalse%2C%22tweetypie_unmention_optimization_enabled%22%3Atrue%2C%22vibe_api_enabled%22%3Atrue%2C%22responsive_web_edit_tweet_api_enabled%22%3Atrue%2C%22graphql_is_translatable_rweb_tweet_is_translatable_enabled%22%3Atrue%2C%22view_counts_everywhere_api_enabled%22%3Atrue%2C%22standardized_nudges_misinfo%22%3Atrue%2C%22tweet_with_visibility_results_prefer_gql_limited_actions_policy_enabled%22%3Afalse%2C%22interactive_text_enabled%22%3Atrue%2C%22responsive_web_text_conversations_enabled%22%3Afalse%2C%22responsive_web_enhance_cards_enabled%22%3Afalse%7D'

#  CURSOR拼接规则
JOIN_CURSOR = 'cursor":"{cursor}","referrer":"tweet","'
USER_ARTICLE_CURSOR = 'cursor":"{cursor}==","'
USER_REPLY_CURSOR = '{cursor}'
# USER_FOLLOWERS_CURSOR = '"cursor":"{cursor}",'
USER_FOLLOWERS_CURSOR = '{cursor}'
USER_FOLLOWING_CURSOR = USER_FOLLOWERS_CURSOR

#  文章类型 转推/引推/原发(None)
RETWEETED = 'retweet'
QUOTED = 'quoted'

# 数据类型
ARTICLE = 'article'
REPLY = 'reply'
USER = 'user'
COMMENT = 'comment'
FOLLOWERS = 'followers'
FOLLOWING = 'following'

# 搜索类型
TOP = 'Top'
PEOPLE = 'People'
LATEST = 'Latest'
VIDEOS = 'Videos'
PHOTOS = 'Photos'

ALL_SEARCH_TYPE = [
    TOP,
    PEOPLE,
    LATEST,
    VIDEOS,
    PHOTOS
]

# 协程数
ASYNC_COUNT = 1

#  翻页数
DEFAULT_PAGE_LIMIT = 30

# 采集评论数
DEFAULT_COMMENT_COUNT = 50

# 默认请求休眠时间
DEFAULT_REQUESTS_TIMEOUT = 10

# 默认请求等待时间
DEFAULT_REQUESTS_SLEEP = 0

easy_twitter_crawler/twitter_crawler/constants.py
hanxinkong/easy-twitter-crawler
"""
自定义异常类
"""
from enum import Enum


class ErrorCode(Enum):
    """异常错误码"""
    pass


class AccountErrorCode(ErrorCode):
    pass


class ProxyException(Exception):
    """
    代理异常
    """
    pass


class CaptchaException(Exception):
    """
    验证异常
    """
    pass


class UserException(Exception):
    """
    用户异常信息
    """

    def __init__(self, user, msg: str = ''):
        self.user = user
        self.msg = msg

    def __str__(self):
        return f'【{self.user}】: {self.msg}'


class UserLoginException(UserException):
    """用户登录异常"""

    def __str__(self):
        return f'【{self.user}】 : 该用户信息采集需要登录'


class UserLoseException(UserException):
    """账户不存在"""

    def __str__(self):
        return f'【{self.user}】: 该账号不存在'


class UserPrivateException(UserException):
    """用户私密"""

    def __str__(self):
        return f'【{self.user}】 : 该用户已设置为私密，无法采集'


class UserSuspendedException(UserException):
    """用户封禁"""

    def __str__(self):
        return f'【{self.user}】 : 该用户已被封禁，无法采集'


class ArticleException(Exception):
    """文章异常信息"""

    def __init__(self, article_id: str, msg: str = ''):
        self.article_id = article_id
        self.msg = msg

    def __str__(self):
        return f'【{self.article_id}】: {self.msg}'


class ArticleLoginException(ArticleException):
    """文章登录异常"""

    def __str__(self):
        return f'【{self.article_id}】: 该推文采集需要登录'

easy_twitter_crawler/twitter_crawler/exceptions.py
hanxinkong/easy-twitter-crawler
"""
推特清洗器
"""
import re

from easy_spider_tool import date_parse
from requests_html import HTML

from .exceptions import (ArticleLoginException, UserLoginException,
                         UserPrivateException, UserSuspendedException, UserException)
from .twitter_scraper import *
from .constants import *

from easy_twitter_crawler.twitter_utils.twitter_utils import (
    generating_page_link, jsonpath, regx_match,
    generating_page_links, quote,
    extract_error_info, extract_time
)


def extract_data(raw_html, options: Options, request_fn: RequestFunction, full_html=None) -> Union[
    Iterator[UserType],
    Iterator[ArticleType],
    Iterator[VideoType],
    Iterator[PhotoType],
    Iterator[CommentType]
]:
    return BaseExtractor(raw_html, options, request_fn, full_html).extract_data()


def extract_user(raw_html, options: Options, request_fn: RequestFunction, full_html=None) -> UserType:
    return UserExtractor(raw_html, options, request_fn, full_html).extract_data()


def extract_comment(raw_html, options: Options, request_fn: RequestFunction, full_html=None) -> CommentType:
    return CommentExtractor(raw_html, options, request_fn, full_html).extract_data()


def init_reaction():
    """
    互动量
    """
    return {
        'up_count': 0,  # 点赞量
        'cmt_count': 0,  # 评论量
        'read_count': 0,  # 阅读量
        'rtt_count': 0,  # 转发量
        'share_count': 0,  # 分享量
        'collect_count': 0,  # 收藏量
        'quote_count': 0,  # 引用量
        'play_count': 0  # 视频播放量
    }


class BaseExtractor:
    video_regex = re.compile(r'play_url\".*\"(http.*\.mp4\?.*=hw?)"')
    ele_regex = re.compile(r'<.*</.*>')

    def __init__(self, element, options, request_fn, full_html=None):
        self.element = element
        self.options = options
        self.request_fn = request_fn
        self._type = None
        self._content_html = None
        self.info = {}
        # 详情页
        self._detail_response = None

    @property
    def type(self):
        if self._type is not None:
            return self._type

        self._type = self.options.get('data_type', '')

        if any([
            self._type.capitalize() in (TOP, LATEST, VIDEOS, PHOTOS),
            self._type.lower() == REPLY,
            self._type.lower() == COMMENT
        ]):
            self._type = ARTICLE
        elif self._type.capitalize() == PEOPLE:
            self._type = USER

        return self._type.lower()

    def content_html(self, content=None):
        if self._content_html is not None:
            return self._content_html
        html = content if content else self.element.get('content', '')
        self._content_html = HTML(html=html)
        return self._content_html

    def detail_response(self):
        """
        详情页请求
        :param:
        :return:
        """
        if self._detail_response is not None:
            return self._detail_response
        self._detail_response = self.request_fn(self.info.get('source_url'))
        return self._detail_response

    def extract_data(self):
        """
        数据清洗入口
        :return:
        """
        # 推文/评论
        err_info = extract_error_info(self.element)
        msg = err_info['msg']

        article_id = self.extract_id()[f'{self.type}_id']

        if 'need to log in to Twitter' in msg:
            raise ArticleLoginException(article_id)
        elif '此推文不可用' in msg or 'This Tweet is unavailable' in msg:
            raise ArticleException(article_id, msg)
        elif '这条推文违反了 Twitter 规则' in msg:
            raise ArticleException(article_id, msg)

        # tweet_results = jsonpath(self.element, ['$...itemContent.tweet_results'], first=True, default={})
        # if not tweet_results:
        #     raise ArticleException(article_id, '推文不存在')

        methods = [
            self.extract_id,
            self.extract_pub_time,
            self.extract_author,
            self.extract_url,
            self.extract_pictures,
            self.extract_video_url,
            self.extract_play_duration,
            self.extract_init_reaction,
            self.extract_retweet,
            self.extract_quote,
            self.extract_embed_url,
            self.extract_language,
            self.extract_content,
        ]
        for method in methods:
            try:
                partial_info = method()
                if partial_info is None:
                    continue
                # logger.warning(f'method: {method.__name__} return: {partial_info}')
                self.info.update(partial_info)
            except Exception as ex:
                logger.debug(f'method: {method.__name__} error:{ex}')

        return self.info

    def extract_id(self) -> PartialType:
        # tweet-1671055903175372802
        data_id = ''  # jsonpath(self.options, '$.task_id', first=True, default='')
        if not data_id:
            data_id = jsonpath(
                self.element,
                [
                    "$..tweet_results.result.tweet.rest_id",
                    "$..itemContent.tweet_results.result.rest_id",
                    "$..entryId",
                    "$..rest_id"  # 新发文
                ],
                first=True,
                default=''
            )

        return {
            f'{self.type}_id': data_id.replace('tweet-', '')
        }

    def extract_pub_time(self) -> PartialType:
        """
        编辑时间
        :return:
        """
        pub_time = jsonpath(
            self.element,
            [
                "$..itemContent.tweet_results.result.legacy.created_at",
                "$..itemContent.tweet_results.result.tweet.legacy.created_at",
                "$..legacy.created_at"  # 新发文
            ],
            first=True,
            default=''
        )

        pub_time = extract_time(pub_time)

        return {
            'pub_time': pub_time
        }

    def extract_url(self) -> PartialType:
        """
        url 清洗
        """
        url = jsonpath(self.element, '$.url', first=True, default='')
        if not url:
            user_name = jsonpath(self.info, '$.author_info.user_name', first=True, default='')
            article_id = self.info.get(f'{self.type}_id', '')
            url = ARTICLE_URL.format(user_name=user_name, article_id=article_id) if all([user_name, article_id]) else ''

        return {
            'source_url': url
        }

    def extract_pictures(self) -> PartialType:
        """
        图片清洗
        """
        pic = jsonpath(
            self.element,
            [
                "$..itemContent.tweet_results.result.legacy.entities.media[*].media_url_https",
                "$..itemContent.tweet_results.result.tweet.legacy.entities.media[*].media_url_https",
                "$..legacy.entities.media[*].media_url_https"  # 新发文
            ],
            first=False,
            default=[]
        )
        pic = list(filter(lambda x: x and not "/video/" in x, pic))
        # pic = [i for i in pic if not "/video/" in i]
        return {
            'pictures': list(set(pic))
        }

    def extract_video_url(self) -> PartialType:
        """
        视频链接
        """
        video_url = jsonpath(
            self.element,
            [
                "$..itemContent.tweet_results.result.legacy.extended_entities.media[*].video_info.variants[?(@.content_type==\"video/mp4\")].url",
                "$..legacy.extended_entities.media[*].video_info.variants[?(@.content_type==\"video/mp4\")].url"
            ],
            first=False,
            default=[]
        )

        return {'video_url': list(set(video_url))}

    def extract_author(self, author_info=None) -> PartialType:
        """
        用户信息
        """
        user_id = jsonpath(
            self.element,
            [
                '$..itemContent.tweet_results.result..core.user_results.result.rest_id',
                '$..core.user_results.result.rest_id'  # 新发文
            ],
            first=True,
            default=''
        )
        user_name = jsonpath(
            self.element,
            [
                '$..itemContent.tweet_results.result..core.user_results.result.legacy.screen_name',
                '$..core.user_results.result.legacy.screen_name',  # 新发文
            ],
            first=True,
            default=''
        )

        user_url = USER_BASE_URL.format(user_name=user_name) if user_name else ''

        return {
            'author_info': {
                'user_id': user_id,
                'user_name': user_name,
                'user_url': user_url,
            }
        }

    def extract_init_reaction(self) -> PartialType:
        """
        统一解析互动量
        """

        init_ = init_reaction()
        init_.update({
            'up_count': jsonpath(
                self.element,
                [
                    '$..itemContent.tweet_results.result..legacy.favorite_count',
                    '$..legacy.favorite_count',  # 新发文
                ],
                first=True,
                default=0
            ),
            'cmt_count': jsonpath(
                self.element,
                [
                    '$..itemContent.tweet_results.result..legacy.reply_count',
                    '$..legacy.reply_count',  # 新发文
                ],
                first=True,
                default=0
            ),
            'read_count': int(jsonpath(
                self.element,
                [
                    '$..itemContent.tweet_results.result..views.count',
                    '$..views.count',
                ],
                first=True,
                default=0
            )),
            'rtt_count': jsonpath(
                self.element,
                [
                    '$..itemContent.tweet_results.result..legacy.retweet_count',
                    '$..legacy.retweet_count',
                ],
                first=True,
                default=0
            ),
            'share_count': 0,
            'collect_count': jsonpath(
                self.element,
                [
                    '$..legacy.bookmark_count',
                ],
                first=True,
                default=0
            ),
            'quote_count': jsonpath(
                self.element,
                [
                    '$..itemContent.tweet_results.result..legacy.quote_count',
                    '$..legacy.quote_count',
                ],
                first=True,
                default=0
            ),
            'play_count': jsonpath(
                self.element,
                [
                    '$..legacy.extended_entities.media[*].mediaStats.viewCount',
                ],
                first=True,
                default=0
            )
        })

        return init_

    def extract_play_duration(self) -> PartialType:
        """
        视频播放时长
        """
        play_duration = jsonpath(
            self.element,
            [
                '$..legacy.extended_entities.media[*].video_info.duration_millis',
            ],
            first=True,
            default=0
        )

        return {'play_duration': play_duration}

    def extract_retweet(self) -> PartialType:
        """
        转推信息
        """
        retweet_article_id = jsonpath(
            self.element,
            [
                "$..itemContent.tweet_results.result.legacy.retweeted_status_result.result.rest_id",
                "$..itemContent.tweet_results.result..legacy.retweeted_status_result.result.tweet.rest_id",
                "$..legacy.retweeted_status_result.result.rest_id",
            ],
            first=True,
            default=''
        )

        retweet_user_id = jsonpath(
            self.element,
            [
                "$..itemContent.tweet_results.result..legacy.retweeted_status_result.result.core.user_results.result.rest_id",
                "$..itemContent.tweet_results.result..tweet.legacy.retweeted_status_result.result.tweet.core.user_results.result.rest_id",
                "$..legacy.retweeted_status_result.result..core.user_results.result.rest_id",
            ],
            first=True,
            default=''
        )
        retweet_user_name = jsonpath(
            self.element,
            [
                "$..itemContent.tweet_results.result..legacy.retweeted_status_result.result.core.user_results.result.legacy.screen_name",
                "$..itemContent.tweet_results.result..tweet.legacy.retweeted_status_result.result.tweet.core.user_results.result.legacy.screen_name",
                "$..legacy.retweeted_status_result.result..core.user_results.result.legacy.screen_name",
            ],
            first=True,
            default=''
        )

        return {
            'is_retweet': bool(retweet_article_id),
            'retweet_info': {
                f'{self.type}_id': retweet_article_id,
                'retweet_author': {
                    'user_id': retweet_user_id,
                    'user_name': retweet_user_name,
                },
                'url': ARTICLE_URL.format(user_name=retweet_user_name, article_id=retweet_article_id) if all(
                    [retweet_article_id, retweet_user_name]) else ''
            }
        }

    def extract_quote(self) -> PartialType:
        """
        信息
        """
        quote_article_id = jsonpath(
            self.element,
            [
                "$..itemContent.tweet_results.result.quoted_status_result.result.rest_id",
                "$..itemContent.tweet_results.result..quoted_status_result.result.tweet.rest_id",
                "$..quoted_status_result.result.rest_id"
            ],
            first=True,
            default=''
        )

        quote_user_id = jsonpath(
            self.element,
            [
                "$..itemContent.tweet_results.result..quoted_status_result..user_results.result.rest_id",
                "$..quoted_status_result..user_results.result.rest_id",
            ],
            first=True,
            default=''
        )
        quote_user_name = jsonpath(
            self.element,
            [
                "$..itemContent.tweet_results.result..quoted_status_result..user_results.result.legacy.screen_name",
                "$..quoted_status_result..user_results.result.legacy.screen_name",
            ],
            first=True,
            default=''
        )

        return {
            'is_quote': bool(quote_article_id),
            'quote_info': {
                f'{self.type}_id': quote_article_id,
                'quote_author': {
                    'user_id': quote_user_id,
                    'user_name': quote_user_name,
                },
                'url': ARTICLE_URL.format(user_name=quote_user_name, article_id=quote_article_id) if all(
                    [quote_article_id, quote_user_name]) else ''
            }
        }

    def extract_embed_url(self) -> PartialType:
        """
        内嵌的url/本推文中提到的链接
        """
        short_links = jsonpath(
            self.element,
            [
                "$..itemContent.tweet_results.result..legacy.entities.urls[*].expanded_url",
                "$..itemContent.tweet_results.result..tweet.legacy.entities.urls[*].expanded_url",
                "$..legacy.entities.urls[*].expanded_url",
            ],
            first=False,
            default=[]
        )

        return {
            'embed_url': list(set(short_links))
        }

    def extract_language(self) -> PartialType:
        """
        语种
        """
        language = jsonpath(
            self.element,
            [
                "$..itemContent.tweet_results.result..legacy.lang",
                "$..itemContent.tweet_results.result..tweet.legacy.lang",
                "$..legacy.lang",
            ],
            first=True,
            default=''
        )

        return {
            'language': language
        }

    def extract_content(self, content=None) -> PartialType:
        content = jsonpath(self.element, '$.content', first=True) if content is None else content

        is_retweet = jsonpath(self.info, '$.is_retweet', first=True, default='')
        is_quote = jsonpath(self.info, '$.is_quote', first=True, default='')

        expr = [
            "$..itemContent.tweet_results.result..legacy.full_text",
            "$..itemContent.tweet_results.result..tweet.legacy.full_text",
            "$..legacy.full_text",
        ]

        if is_retweet:  # 转推完整文本
            expr = ["$..legacy.retweeted_status_result.result.legacy.full_text"]

        content = jsonpath(
            self.element,
            expr,
            first=True,
            default=''
        )

        if content and self.ele_regex.findall(content):
            contents = []
            for ele in self.content_html(content).find('p,h1,h2,h3,h4,h5,h6,pre>code,li'):
                if ele and ele.text:
                    if ele.text not in contents:
                        contents.append(ele.text)
            content = '\n'.join(contents)
            content = re.sub(r'<.*>', '', content)
        return {
            'content': content
        }

    def extract_meta_data(self, start_url, type_name, **kwargs) -> PartialType:
        """
        获取账号的文章,关注，粉丝数据
        :param start_url: 请求数据的接口
        :param type_name: 请求数据类型名称
        :return: 返回对应数据集合
        """
        pass


class UserExtractor(BaseExtractor):
    """
    用户信息
    """

    def extract_data(self):
        user_id = jsonpath(self.options, '$.user_id', first=True, default='')

        # 私密账号
        if jsonpath(self.element, '$..__typename', first=True, default='') == 'UserUnavailable':
            reason = jsonpath(self.element, '$..reason', first=True, default='')
            if reason.lower() in 'Suspended'.lower():
                raise UserSuspendedException(user_id)

            raise UserPrivateException(user_id)
        # 用户不存在
        if jsonpath(self.element, '$..data', first=True, default={}):
            raise UserLoseException(user_id)

        # 账户冻结
        err_info = extract_error_info(self.element)
        msg = err_info['msg']
        if '冻结违反' in msg or 'violate the Twitter Rules' in msg:
            raise UserException(user_id, '该用户被冻结')

        methods = [
            self.extract_user,
        ]
        for method in methods:
            try:
                partial_info = method()
                if partial_info is None:
                    continue
                # logger.warning(f'method: {method.__name__} return: {partial_info}')
                self.info.update(partial_info)
            except Exception as ex:
                logger.debug(f'method: {method.__name__} error:{ex}')

        # ********* 用户文章列表采集 ********* #
        total_count = self.info['tweets_count']
        count = self.options.get('article_count', -1)
        if count > -1 and total_count > 0:
            total_count = count if 0 < count < total_count else total_count
            cursor = ''
            start_url = generating_page_link(USER_ARTICLE_URL, cursor=cursor, user_id=self.info['user_id'])
            result = self.extract_meta_data(
                start_url=start_url,
                type_name=ARTICLE,
                total_count=total_count,
                user_id=self.info['user_id']
            ) or {}

            self.info.update(result)

        # ********* 用户回复列表采集 ********* #
        total_count = self.info['tweets_count']
        count = self.options.get('reply_count', -1)
        if count > -1 and total_count > 0:
            total_count = count if 0 < count < total_count else total_count
            cursor = ''
            start_url = generating_page_link(USER_REPLY_URL, cursor=cursor, user_id=self.info['user_id'])
            result = self.extract_meta_data(
                start_url=start_url,
                type_name=REPLY,
                total_count=total_count,
                user_id=self.info['user_id']
            ) or {}

            self.info.update(result)

        # ********* 用户粉丝列表采集 ********* #
        total_count = self.info['followers_count']
        count = self.options.get('followers_count', -1)
        if count > -1 and total_count > 0:
            total_count = count if 0 < count < total_count else total_count
            cursor = ''
            start_url = generating_page_link(USER_FOLLOWERS_API, cursor=cursor, user_id=self.info['user_id'])
            result = self.extract_meta_data(
                start_url=start_url,
                type_name=FOLLOWERS,
                total_count=total_count,
                user_id=self.info['user_id']
            ) or {}
            self.info.update(result)

        # ********* 用户关注列表采集 ********* #
        total_count = self.info['following_count']
        count = self.options.get('following_count', -1)
        if count > -1 and total_count > 0:
            total_count = count if 0 < count < total_count else total_count
            cursor = ''
            start_url = generating_page_link(USER_FOLLOWING_API, cursor=cursor, user_id=self.info['user_id'])
            result = self.extract_meta_data(
                start_url=start_url,
                type_name=FOLLOWING,
                total_count=total_count,
                user_id=self.info['user_id']
            ) or {}
            self.info.update(result)

        return self.info

    def extract_user(self) -> PartialType:

        birthday = jsonpath(self.element, ['$.legacy_extended_profile.birthdate'], first=True, default={})

        birthday_year = str(birthday.get('year', ''))
        birthday_month = str(birthday.get('month', ''))
        birthday_day = str(birthday.get('day', ''))

        user_name = jsonpath(self.element, ['$..screen_name'], first=True, default='')
        join_date = jsonpath(self.element, ['$..created_at'], first=True, default='')
        return {
            'user_id': jsonpath(self.element, ['$.rest_id', '$.id_str'], first=True, default=''),
            'user_name': user_name,
            'display_name': jsonpath(self.element, ['$..name'], first=True, default=''),
            'user_url': f'https://twitter.com/{user_name}' if user_name else '',
            'profile_picture': jsonpath(self.element, ['$..profile_image_url_https'], first=True, default=''),
            'header_photo': jsonpath(self.element, ['$..profile_banner_url'], first=True, default=''),
            'bio': jsonpath(self.element, ['$..description'], first=True, default=''),
            'location': jsonpath(self.element, ['$..location'], first=True, default=''),
            'website_link': {
                'display_url': jsonpath(self.element, ['$..entities.url.urls[*].display_url'], first=True,
                                        default=''),
                'expanded_url': jsonpath(self.element, ['$..entities.url.urls[*].expanded_url'], first=True,
                                         default=''),
            },
            'birthday': {
                'birthday': '-'.join([birthday_year, birthday_month, birthday_day]).strip('-'),
                'visibility': birthday.get('visibility', '')
            },
            'followers_count': jsonpath(self.element, ['$..followers_count'], first=True, default=0),
            'following_count': jsonpath(self.element, ['$..friends_count'], first=True, default=0),
            'tweets_count': jsonpath(self.element, ['$..statuses_count'], first=True, default=0),
            'likes_count': jsonpath(self.element, ['$..favourites_count'], first=True, default=0),
            'media_count': jsonpath(self.element, ['$..media_count'], first=True, default=0),
            'listed_count': jsonpath(self.element, ['$..listed_count'], first=True, default=0),
            'join_date': extract_time(join_date),
            'auth': {
                'is_verified': jsonpath(self.element, '$..verified', first=True, default=False),
                'verified_type': jsonpath(self.element, '$..verified_type', first=True, default=''),
                'verification_info': jsonpath(self.element, '$.verification_info.reason.description.text', first=True,
                                              default=''),
            },
            'professional': {
                'professional_category': jsonpath(self.element, '$.professional.category[*].name', first=True,
                                                  default=''),
                'professional_type': jsonpath(self.element, '$.professional.professional_type', first=True, default=''),
            },
            'protected': jsonpath(self.element, '$..protected', first=True, default=False),
        }

    def extract_meta_data(self, start_url, type_name, **kwargs) -> PartialType:
        """
        获取账号的文章,动态等数据
        :param start_url: 请求数据的接口
        :param type_name: 请求数据类型名称
        :return: 返回对应数据集合
        """
        total_count = kwargs.get('total_count', 0)
        user_id = kwargs.pop('user_id', False)
        start_time = self.options.pop('start_time', False)
        end_time = self.options.pop('end_time', False)

        if start_time:
            start_time = date_parse(start_time, default_time_scheme=None)

        if type_name == REPLY:
            self.options.update({'data_type': ARTICLE})
        else:
            self.options.update({'data_type': type_name})

        if total_count == 0:
            return

        page_urls = generating_page_links(start_url, total_count)
        data_list = []

        # ---------仅能一页一页采集------------#
        article_ele_regex = r"^tweet-.*"
        expr = []
        last_cursor = None
        next_cursor = ''
        join_cursor = None

        base_url = None

        if type_name in (FOLLOWERS, FOLLOWING):
            expr = [  # 用户粉丝/关注
                '$.data.user.result.timeline.timeline.instructions[-1:].entries[*].content.itemContent.user_results.result',
                '$.users[*]'  # 开发者api
            ]
            base_url = USER_FOLLOWERS_API if type_name == FOLLOWERS else USER_FOLLOWING_API
            join_cursor = USER_FOLLOWERS_CURSOR if type_name == FOLLOWERS else USER_FOLLOWING_CURSOR
        elif type_name == ARTICLE:
            expr = [  # 用户文章
                '$..entries[*]'
            ]
            base_url = USER_ARTICLE_URL
            join_cursor = USER_ARTICLE_CURSOR
        elif type_name == REPLY:
            expr = [  # 用户回复
                '$..instructions[*]..entries..items..item..tweet_results.result'
            ]
            base_url = USER_REPLY_URL
            join_cursor = USER_REPLY_CURSOR

        for index, page_url in enumerate(page_urls, start=1):

            if next_cursor:
                next_cursor = quote(join_cursor.format(cursor=next_cursor))

            page_url = generating_page_link(base_url, cursor=next_cursor, user_id=user_id, data_type=type_name)

            logger.warning(f'[{index}] ===> request next cursor {next_cursor}')

            # logger.warning(f'request next url {page_url}')

            response = self.request_fn(page_url, **kwargs)

            # 翻页失败，停止继续翻页
            if response is None:
                break

            if response and response.status_code == 200:
                response_json = json.loads(response.text) if response else {}

                err_info = extract_error_info(response_json)

                # 需要登录
                if err_info['name'] == 'AuthorizationError':
                    raise UserLoginException(user_id)

                elements = jsonpath(
                    response_json,
                    expr,
                    first=False,
                    default=[]
                )

                # 后续没有文章了
                if len(elements) < 3:
                    break

                for element in elements:
                    if total_count <= len(data_list):
                        return {
                            type_name: data_list
                        }

                    if type_name == FOLLOWERS:
                        extractor = UserFriendExtractor(element, self.options, self.request_fn, full_html=None)
                    elif type_name == FOLLOWING:
                        extractor = UserFollowingExtractor(element, self.options, self.request_fn, full_html=None)
                    elif type_name == ARTICLE:
                        # 置顶推文
                        # 排除推荐
                        entry_id = jsonpath(element, ['$..entryId'], first=True, default='')
                        if not regx_match(article_ele_regex, entry_id, first=True, default=False):
                            continue
                        extractor = BaseExtractor(element, self.options, self.request_fn, full_html=None)
                    elif type_name == REPLY:
                        # 排除文章，只要回复
                        reply_type = jsonpath(element, ['$..in_reply_to_status_id_str'], first=True, default='')
                        if not reply_type:
                            continue
                        extractor = UserReplyExtractor(element, self.options, self.request_fn, full_html=None)
                    else:
                        extractor = BaseExtractor(element, self.options, self.request_fn, full_html=None)

                    result = extractor.extract_data()
                    # yield result
                    pub_time = result.get('pub_time', '')
                    # 根据时间过滤
                    if start_time and pub_time <= start_time:
                        return {
                            type_name: data_list
                        }
                    if end_time and pub_time >= end_time:
                        continue

                    data_list.append(result)

                next_cursor = extract_cursor(response_json)

                if type_name == ARTICLE or type_name == REPLY:
                    if next_cursor == '' or last_cursor == next_cursor:
                        break
                elif type_name in (FOLLOWERS, FOLLOWING):
                    if str(next_cursor) == str(0) or last_cursor == next_cursor:
                        break
                # elif type_name == FOLLOWERS:
                # node_type = jsonpath(response_json,
                #                      '$.data.user.result.timeline.timeline.instructions[*].entries..element',
                #                      first=True, default='')
                # if next_cursor == '' or node_type != 'user':
                #     break

                last_cursor = next_cursor

        return {type_name: data_list}


class UserArticleExtractor(UserExtractor):
    """
    用户文章
    """
    pass


class UserReplyExtractor(BaseExtractor):
    """
    用户回复
    """

    def __init__(self, element, options, request_fn, full_html=None):
        super().__init__(element, options, request_fn)

    def extract_data(self):
        """
        数据清洗入口
        :return:
        """

        self.info.update(**super().extract_data())
        methods = [
            self.extract_ref_article_id,
            self.extract_ref_author,
        ]

        for method in methods:
            try:
                partial_info = method()
                if partial_info is None:
                    continue
                # logger.warning(f'method: {method.__name__} return: {partial_info}')
                self.info.update(partial_info)
            except Exception as ex:
                logger.debug(f'method: {method.__name__} error:{ex}')

        return self.info

    def extract_ref_article_id(self) -> PartialType:
        """
        被评论文章id
        @return:
        """
        article_id = jsonpath(self.element, ['$..in_reply_to_status_id_str'], first=True, default='')

        return {'ref_article_id': article_id}

    def extract_ref_author(self) -> PartialType:
        """
        被评论文章id
        @return:
        """
        user_id = jsonpath(self.element, ['$..in_reply_to_user_id_str'], first=True, default='')
        user_name = jsonpath(self.element, ['$..in_reply_to_screen_name'], first=True, default='')

        user_url = USER_BASE_URL.format(user_name=user_name) if user_name else ''
        return {
            'ref_author_info': {
                'user_id': user_id,
                'user_name': user_name,
                'user_url': user_url,
            }
        }


class UserFriendExtractor(UserExtractor):
    """
    用户关注人或粉丝
    """

    def extract_data(self):
        return self.extract_user()


class UserFollowingExtractor(UserExtractor):
    """
    用户关注
    """

    def extract_data(self):
        return self.extract_user()


class CommentExtractor(BaseExtractor):
    """
    发文评论
    """

    def __init__(self, element, options, request_fn, full_html=None):
        super().__init__(element, options, request_fn)

    def extract_data(self):
        """
        数据清洗入口
        :return:
        """

        self.info.update(**super().extract_data())
        methods = [
            self.extract_ref_article_id,
            self.extract_ref_author,
            self.extract_ref_source_url,
            self.extract_root_article_id,
            self.extract_root_author,
            self.extract_root_source_url,
        ]

        for method in methods:
            try:
                partial_info = method()
                if partial_info is None:
                    continue
                # logger.warning(f'method: {method.__name__} return: {partial_info}')
                self.info.update(partial_info)
            except Exception as ex:
                logger.debug(f'method: {method.__name__} error:{ex}')

        return self.info

    def extract_ref_article_id(self) -> PartialType:
        """
        被评论文章id
        @return:
        """
        article_id = jsonpath(self.element, ['$..in_reply_to_status_id_str'], first=True, default='')

        return {'ref_article_id': article_id}

    def extract_ref_author(self) -> PartialType:
        """
        被评论文章作者信息
        @return:
        """
        user_id = jsonpath(self.element, ['$..in_reply_to_user_id_str'], first=True, default='')
        user_name = jsonpath(self.element, ['$..in_reply_to_screen_name'], first=True, default='')

        user_url = USER_BASE_URL.format(user_name=user_name) if user_name else ''
        return {
            'ref_author_info': {
                'user_id': user_id,
                'user_name': user_name,
                'user_url': user_url,
            }
        }

    def extract_ref_source_url(self) -> PartialType:
        """
        被评论文章url
        @return:
        """
        user_name = jsonpath(self.info, '$.ref_author_info.user_name', first=True, default='')
        article_id = jsonpath(self.info, '$.ref_article_id', first=True, default='')
        url = ARTICLE_URL.format(user_name=user_name, article_id=article_id) if all([user_name, article_id]) else ''

        return {'ref_source_url': url}

    def extract_root_article_id(self) -> PartialType:
        """
        根文章id
        @return:
        """
        article_id = jsonpath(self.element, ['$..conversation_id_str'], first=True, default='')

        return {'root_article_id': article_id}

    def extract_root_author(self) -> PartialType:
        """
        根文章用户信息
        @return:
        """
        user_id = jsonpath(self.element, ['$..entities.user_mentions[-1:].id_str'], first=True, default='')
        user_name = jsonpath(self.element, ['$..entities.user_mentions[-1:].screen_name'], first=True, default='')
        user_url = USER_BASE_URL.format(user_name=user_name) if user_name else ''

        return {
            'root_author_info': {
                'user_id': user_id,
                'user_name': user_name,
                'user_url': user_url,
            }
        }

    def extract_root_source_url(self) -> PartialType:
        """
        根文章url
        @return:
        """
        user_name = jsonpath(self.info, '$.root_author_info.user_name', first=True, default='')
        article_id = jsonpath(self.info, '$.root_article_id', first=True, default='')
        url = ARTICLE_URL.format(user_name=user_name, article_id=article_id) if all([user_name, article_id]) else ''

        return {'root_source_url': url}

easy_twitter_crawler/twitter_crawler/extractors.py
hanxinkong/easy-twitter-crawler
"""
页面迭代器
"""
import json
from loguru import logger

from easy_twitter_crawler.twitter_crawler.exceptions import ArticleException, UserLoseException
from easy_twitter_crawler.twitter_crawler.twitter_types import *
from easy_twitter_crawler.twitter_crawler.constants import *
from easy_twitter_crawler.twitter_utils.twitter_utils import (
    quote, jsonpath, extract_cursor, find_element,
    for_to_regx_match, regx_match, generating_page_link
)


def iter_search_pages(key_word: str, request_fn: RequestFunction, **kwargs) -> Iterator[Page]:
    """
    搜索
    :return:
    """
    start_url = kwargs.pop('start_url', None)
    data_type = kwargs.pop('data_type', '')

    kwargs.update({'key_word': key_word, 'data_type': data_type})

    if not start_url:
        key_word = key_word
        start_url = SEARCH_URL.format(key_word=key_word, data_type=data_type, cursor='')
    return generic_iter_pages(start_url, PageParser, request_fn, **kwargs)


def iter_article_pages(article_id: str, request_fn: RequestFunction, **kwargs) -> Iterator[Page]:
    start_url = kwargs.pop('start_url', None)
    if not start_url:
        start_url = ARTICLE_API.format(article_id=article_id, cursor='')
        kwargs.update({'article_id': article_id})
    return generic_iter_pages(start_url, ArticlePageParser, request_fn, **kwargs)


def iter_user_pages(user_id: str, request_fn: RequestFunction, **kwargs) -> Iterator[Page]:
    start_url = kwargs.pop('start_url', None)
    if not start_url:
        start_url = USER_BASE_API.format(user_name=f'{user_id}')
    return generic_iter_pages(start_url, UserPageParser, request_fn, **kwargs)


def iter_comment_pages(article_id: str, request_fn: RequestFunction, **kwargs) -> Iterator[Page]:
    start_url = kwargs.pop('start_url', None)
    if not start_url:
        start_url = COMMENT_API.format(article_id=article_id, cursor='')
        kwargs.update({'article_id': article_id})
    return generic_iter_pages(start_url, ArticleCommentPageParser, request_fn, **kwargs)


def generic_iter_pages(start_url, page_parser_cls, request_fn, **kwargs) -> Iterator[Page]:
    next_url = start_url

    while next_url:
        try:
            response = request_fn(next_url, **kwargs)
        except Exception as e:
            response = None
            logger.error(f'error: {e}')
        parser = page_parser_cls(response)
        page = parser.get_pages(**kwargs)
        yield page
        next_page_info = parser.get_next_page(**kwargs)
        next_url = next_page_info.get('next_url')

        if next_url:
            logger.warning(f'request next url {next_url}')
        else:
            logger.warning('last page')
            return


class PageParser:
    """
    json数据清洗
    """
    article_ele_regex = r"^tweet-.*"
    promoted_article_ele_regex = r"^promoted-tweet.*"
    user_ele_regex = r"^user-.*"

    def __init__(self, response):
        self.response = response
        self.html = None
        self.json_data = None
        self._parse()
        self.data_length = 0
        self.last_cursor = ''

    def _parse(self):
        # jsons = []
        assert self.response is not None, 'response is null'
        self.json_data = json.loads(self.response.text)
        self.html = self.response.html

    def get_raw_page(self) -> RawPage:
        return self.html

    def get_next_page(self, **kwargs) -> Dict[str, Union[str, bool]]:
        assert self.json_data is not None, 'json_data is null'
        next_cursor = extract_cursor(self.json_data)

        key_word = kwargs.pop('key_word', '')
        data_type = kwargs.pop('data_type', '')

        next_url = None

        if next_cursor and self.last_cursor != next_cursor and self.data_length > 2:
            self.last_cursor = next_cursor
            next_url = generating_page_link(SEARCH_URL, key_word=key_word, data_type=data_type, cursor=next_cursor)

        return {'next_cursor': next_cursor, 'next_url': next_url}

    def get_pages(self, **kwargs) -> Page:
        assert self.json_data is not None, 'json_data is null'
        data_list = None

        data = jsonpath(
            self.json_data,
            [
                '$.data.search_by_raw_query.search_timeline.timeline.instructions[*].entries[*]',
                # 搜索-用户
                # '$.data.search_by_raw_query.search_timeline.timeline.instructions[*].entries[*].content.itemContent.user_results.result'
            ],
            first=False,
            default=[]
        )
        self.data_length = len(data)

        matches = jsonpath(data, '$..entryId', first=False, default=[])
        # 是否为用户类型
        user_type = False
        if len(matches) > 0:
            user_type = matches[0].startswith('user-')

        matched_values = [value for value in matches if
                          for_to_regx_match(
                              [self.article_ele_regex, self.promoted_article_ele_regex, self.user_ele_regex],
                              value,
                              first=True,
                              default=False
                          )]

        data = [find_element(data, i) for i in matched_values]

        if user_type:
            data = jsonpath(data, '$..content.itemContent.user_results.result', first=False, default=[])

        if not data_list:
            data_list = data

        assert data_list is not None
        return data_list


class VideoPageParser(PageParser):
    """
    视频json数据清洗
    """
    pass


class ArticlePageParser(PageParser):
    """
    文章json数据清洗
    """

    def get_pages(self, **kwargs) -> Page:
        assert self.json_data is not None, 'json_data is null'
        data_list = None
        article_id = kwargs.pop('article_id', '')

        data = jsonpath(
            self.json_data,
            [
                f"$..instructions[0].entries[?(@.entryId=='tweet-{article_id}')]",
                "$.data.tweetResult.result"
            ],
            first=True,
            default={}
        )

        if not data:
            raise ArticleException(article_id, '推文不存在')

        reason = jsonpath(data, '$..__typename', first=True, default='')
        if 'TweetUnavailable' in reason:
            raise ArticleException(article_id, '推文不可用,可能是受保护的')

        if not data_list:
            data_list = [data]
        assert data_list is not None
        return data_list

    def get_next_page(self, **kwargs) -> Dict[str, Union[str, bool]]:
        assert self.json_data is not None, 'json_data is null'

        return {'next_cursor': '', 'next_url': ''}


class UserPageParser(PageParser):
    """
    用户json数据清洗
    """

    def get_pages(self, **kwargs) -> Page:
        assert self.json_data is not None, 'json_data is null'

        data_list = None
        data = jsonpath(
            self.json_data,
            [
                '$.data.user.result',
            ],
            first=True,
            default={}
        )

        if not data:
            # 用户不存在
            user_id = jsonpath(kwargs, '$.options.user_id', first=True, default='')
            raise UserLoseException(user=user_id)

        if not data_list:
            data_list = [data]
        assert data_list is not None
        return data_list


class UserArticlePageParser(PageParser):
    """
    用户的文章json数据清洗
    """
    pass


class ArticleCommentPageParser(PageParser):
    """
    用户的文章评论json数据清洗
    """
    cmt_ele_regex = r'^conversationthread(?!.*showmore)'
    showmore_ele_regex = r'^cursor-showmorethreads-.*'

    def get_pages(self, **kwargs) -> Page:
        assert self.json_data is not None, 'json_data is null'
        data_list = None
        comment_elements = []

        article_id = kwargs.pop('article_id', '')

        data = jsonpath(
            self.json_data,
            '$..instructions[*].entries[*]',
            first=False,
            default=[]
        )

        # -----------------提取包含评论的节点---------------------#
        article_element = find_element(data, f'tweet-{article_id}', first=False, default={})

        if bool(jsonpath(
                article_element,
                '$..itemContent.tweet_results.result.legacy.in_reply_to_status_id_str',
                first=True,
                default=''
        )):
            comment_elements.extend(article_element)

        matches = jsonpath(data, '$..entryId', first=False, default=[])
        # 是否有显示更多评论
        show_more_matched = [value for value in matches if
                             regx_match(self.showmore_ele_regex, value, first=True, default=False)]

        matched_values = [value for value in matches if
                          regx_match(self.cmt_ele_regex, value, first=True, default=False)]

        data = [find_element(data, i) for i in matched_values]
        comment_elements.extend(jsonpath(
            data,
            '$.*.content.items[*]',
            first=False,
            default=[]
        ))
        # 二次处理无效数据
        matches = jsonpath(comment_elements, '$..entryId', first=False, default=[])
        matched_values = [value for value in matches if
                          regx_match(self.cmt_ele_regex, value, first=True, default=False)]

        comment_elements = [find_element(data, i) for i in matched_values]

        if not comment_elements and not show_more_matched:
            raise ArticleException(article_id, '回复不存在')

        if not data_list:
            data_list = comment_elements
        assert data_list is not None
        return data_list

    def get_next_page(self, **kwargs) -> Dict[str, Union[str, bool]]:
        assert self.json_data is not None, 'json_data is null'
        next_cursor = extract_cursor(self.json_data)

        article_id = kwargs.pop('article_id', '')

        next_url = None
        if next_cursor and self.last_cursor != next_cursor:
            self.last_cursor = next_cursor
            join_cursor = quote(JOIN_CURSOR.format(cursor=next_cursor))
            # print(next_cursor)
            next_url = generating_page_link(COMMENT_API, article_id=article_id, cursor=join_cursor)

        return {'next_cursor': next_cursor, 'next_url': next_url}

easy_twitter_crawler/twitter_crawler/page_iterators.py
hanxinkong/easy-twitter-crawler
import time
import itertools
import asyncio
from requests_html import HTMLSession, AsyncHTMLSession
from functools import partial
from easy_twitter_crawler.twitter_utils.twitter_utils import get_useragent, get_headers
from .page_iterators import *
from .constants import *
from .twitter_types import *
from .extractors import extract_data, extract_user, extract_comment


class TwitterScraper:
    """
    推特采集
    """
    default_headers = {
        "user-agent": get_useragent(),
    }

    def __init__(self, session=None, async_session=None, requests_kwargs=None):
        if session is None:
            session = HTMLSession()
            session.headers.update(self.default_headers)
        if requests_kwargs is None:
            requests_kwargs = {}
        if async_session is None:
            async_session = AsyncHTMLSession(workers=ASYNC_COUNT)
            async_session.headers.update(self.default_headers)
        self.async_session = async_session
        self.session = session
        self.requests_kwargs = requests_kwargs

    def set_proxy(self, proxy: Optional[Dict[str, str]] = None):
        """
        设置代理
        :param proxy: proxy = {'http': 'http://ip:port', 'https': 'http://ip:port'}
        :return:
        """
        proxies = {
            'proxies': proxy
        }
        self.requests_kwargs.update(proxies)

    def set_timeout(self, timeout: int):
        """
        设置请求超时 单位秒
        """
        self.requests_kwargs['timeout'] = timeout

    def search_crawler(self, key_word: Union[str], **kwargs) -> Union[Iterator[UserType],
    Iterator[ArticleType],
    Iterator[VideoType],
    Iterator[PhotoType]]:
        """
        通过关键词对检索结果进行采集
        :param key_word: 需要采集的关键词
        :return:
        """
        kwargs['scraper'] = self
        iter_search_pages_fn = partial(iter_search_pages, key_word=key_word, request_fn=self.send, **kwargs)
        data_type = kwargs.get('data_type', '')

        if data_type == PEOPLE:
            extractor = extract_user
        else:
            extractor = extract_data

        return self._generic_crawler(extractor, iter_search_pages_fn, **kwargs)

    def article_crawler(self, article_id: Union[str], **kwargs) -> Iterator[ArticleType]:
        """
        通过文章id采集文章页数据
        """
        kwargs['scraper'] = self
        iter_article_pages_fn = partial(iter_article_pages, article_id=article_id, request_fn=self.send, **kwargs)
        return self._generic_crawler(extract_data, iter_article_pages_fn, **kwargs)

    def user_crawler(self, user_id: Union[str], **kwargs) -> Iterator[UserType]:
        """
        通过账号id采集个人主页数据
        """
        kwargs['scraper'] = self
        iter_user_page_fn = partial(iter_user_pages, user_id=user_id, request_fn=self.send, **kwargs)
        return self._generic_crawler(extract_user, iter_user_page_fn, **kwargs)

    def comment_crawler(self, article_id: Union[str], **kwargs) -> Iterator[UserType]:
        """
        通过文章id采集文章页评论
        """
        kwargs['scraper'] = self
        iter_comment_page_fn = partial(iter_comment_pages, article_id=article_id, request_fn=self.send, **kwargs)
        return self._generic_crawler(extract_comment, iter_comment_page_fn, **kwargs)

    def send(self, url, **kwargs):
        if not url:
            logger.error('url is null')
        method = kwargs.get('method', 'GET')
        return self.post(url, **kwargs) if method == 'POST' else self.get(url, **kwargs)

    def get(self, url, **kwargs):
        """
        请求方法
        """
        assert url is not None, 'url is null'
        if isinstance(url, str):
            cookie = self.default_headers.get('cookie')
            self.default_headers.update(get_headers(cookie) or {})
            self.session.headers.update(self.default_headers)

            retry_limit = 5
            response = None
            for retry in range(1, retry_limit + 1):
                try:
                    time.sleep(DEFAULT_REQUESTS_SLEEP)
                    response = self.session.get(url, **self.requests_kwargs)
                    response.raise_for_status()
                    return response
                except Exception as e:
                    if retry < retry_limit:
                        sleep_time = retry * 3
                        logger.debug(f'重连第{retry}次，休眠{sleep_time}秒, 异常：{e}')
                        time.sleep(sleep_time)

            assert response is not None, f'重新请求{retry_limit}次， response为空'

        if isinstance(url, list):  # 使用协程请求
            return self.generic_response(url, **kwargs)

    def generic_response(self, urls, **kwargs):
        urls = [urls[i: i + ASYNC_COUNT] for i in range(0, len(urls), ASYNC_COUNT)]
        for sub_urls in urls:
            tasks = [lambda url=url: self.async_get(url, **kwargs) for url in sub_urls]
            results = self.async_session.run(*tasks)
            yield results

    async def async_get(self, url, **kwargs):
        if self.default_headers.get('cookie', False):
            self.default_headers.update(get_headers(self.default_headers.get('cookie')) or {})
        self.async_session.headers.update(self.default_headers)
        try:
            response = await self.async_session.get(url, **self.requests_kwargs)
            if response and response.status_code != 200:
                logger.error(f'request url: {url}, response code: {response.status_code}')
            await asyncio.sleep(2)
            return response
        except Exception as e:
            logger.error(f'error: {e}')

    def post(self, url, **kwargs):
        pass

    def _generic_crawler(self,
                         extract_fn,
                         iter_pages_fn,
                         options=None,
                         **kwargs):
        """
        中转函数
        @extract_fn 数据清洗方法
        @iter_pages_fn 页面处理方法
        @options 参数
        """
        page_limit = kwargs.get('page_limit') if kwargs.get('page_limit', 0) else DEFAULT_PAGE_LIMIT
        counter = itertools.count(0) if page_limit is None else range(page_limit)
        if options is None:
            options = {}
        elif isinstance(options, set):
            options = {k: True for k in options}
        total_count = kwargs.get('total_count', 0)
        count = 0

        for i, page in zip(counter, iter_pages_fn()):
            for element in page:
                if 0 < total_count <= count:
                    return None

                count += 1
                try:
                    info = extract_fn(element, options=options, request_fn=self.send)
                    yield info
                except Exception as e:
                    raise
                    # logger.error(e)

easy_twitter_crawler/twitter_crawler/twitter_scraper.py
hanxinkong/easy-twitter-crawler
from typing import Any, Callable, Dict, Iterable, Tuple, List, Union, Optional, Set, Iterator
from requests_html import Element
from requests import Response

URL = str
Options = Dict[str, Any]

Keyword = Dict[str, List]
CommentType = Dict[str, Any]

# 搜索类型
UserType = Dict[str, Any]
ArticleType = Dict[str, Any]
VideoType = Dict[str, Any]
PhotoType = Dict[str, Any]

#
RawPage = Element
RawPost = Element
Page = Iterable[RawPost]
Credentials = Tuple[str, str]
RequestFunction = Callable[[URL], Response]
PartialType = Optional[Dict[str, Any]]

easy_twitter_crawler/twitter_crawler/twitter_types.py
hanxinkong/easy-twitter-crawler
from easy_twitter_crawler.twitter_utils import twitter_utils
from .twitter_filter import TwitterFilter


def get_proxy():
    return twitter_utils.get_proxy()


def get_headers(cookie: str):
    """获取头"""
    return twitter_utils.get_headers(cookie)

easy_twitter_crawler/twitter_utils/__init__.py
hanxinkong/easy-twitter-crawler
import inspect
from typing import Union


class TwitterFilter:
    __word_category_format = {
        'exact': '"{}"',
        'filter_any': '({})',  # 原为any 关键词冲突
        'exclude': '-{}',
        'tab': '(#{})',
        'lang': 'lang:{}',
    }
    __account_category_format = {
        'filter_from': 'from:{}',  # 原为from 后加()
        'to': 'to:{}',  # 后加()
        '@': '@:{}',  # 后加()
    }
    __filter_category_format = {
        'only_replies': 'filter:replies',
        'only_links': 'filter:links',
        'exclude_replies': '-filter:replies',
        'exclude_links': '-filter:links',
    }
    __interact_category_format = {
        'min_replies': 'min_replies:{}',
        'min_faves': 'min_faves:{}',
        'min_retweets': 'min_retweets:{}',
        'min_read': 'min_retweets:{}',
    }
    __date_category_format = {
        'since': 'since:{}',  # since:2010-06-13
        'until': 'until:{}'
    }

    def __init__(self, key_word: str):
        self.key_word = key_word
        self.__fill_keyword = None
        self.__word_category_dataset = []
        self.__account_category_dataset = []
        self.__filter_category_dataset = []
        self.__interact_category_dataset = []
        self.__date_category_dataset = []

    def filter_join(self) -> str:
        """过滤条件拼接"""

        filter_join_dateset = [
            *self.__word_category_dataset,
            *self.__account_category_dataset,
            *self.__filter_category_dataset,
            *self.__interact_category_dataset,
            *self.__date_category_dataset,
        ]

        if self.__fill_keyword is None:
            filter_join_dateset.insert(0, self.key_word)

        return ' '.join(filter_join_dateset)

    def word_category(self, exact: str = None, filter_any: str = None,
                      exclude: str = None,
                      tab: str = None, lang: str = None):
        """词语过滤器"""
        frame = inspect.currentframe()
        args, _, _, values = inspect.getargvalues(frame)
        args.remove('self')

        values = dict(values)
        for arg in args:
            if values[arg] is not None:  # 代表有key，但是没值
                if values[arg] == '':
                    self.__fill_keyword = False
                    values[arg] = [self.key_word]
                elif not values[arg].startswith('"'):
                    values[arg] = values[arg].split()
                else:
                    values[arg] = [values[arg]]

                for i in values[arg]:
                    key_join = self.__word_category_format[arg].format(i)
                    if key_join:
                        self.__word_category_dataset.append(key_join)

        return self.__word_category_dataset

    def account_category(self, filter_from: str = None, to: str = None,
                         at: str = None):
        """账户过滤器"""

        frame = inspect.currentframe()
        args, _, _, values = inspect.getargvalues(frame)
        args.remove('self')

        values = dict(values)
        for arg in args:
            if values[arg] is not None:
                if not values[arg].startswith('"'):
                    values[arg] = values[arg].split()
                else:
                    values[arg] = [values[arg]]

                key_join = ' OR '.join([self.__account_category_format[arg].format(i) for i in values[arg]])

                if key_join:
                    self.__account_category_dataset.append('(' + key_join + ')')

        return self.__account_category_dataset

    def filter_category(self, only_replies: str = None, only_links: str = None, exclude_replies: str = None,
                        exclude_links: str = None):
        """过滤过滤器"""

        frame = inspect.currentframe()
        args, _, _, values = inspect.getargvalues(frame)
        args.remove('self')

        values = dict(values)
        for arg in args:
            if values[arg] is not None:
                key_join = self.__filter_category_format[arg].format(values[arg])
                self.__filter_category_dataset.append(key_join)

        return self.__filter_category_dataset

    def interact_category(self, min_replies: Union[str, int] = None, min_faves: Union[str, int] = None,
                          min_retweets: Union[str, int] = None, **kwargs):
        """互动量过滤器"""

        frame = inspect.currentframe()
        args, _, _, values = inspect.getargvalues(frame)
        args.remove('self')

        values = dict(values)
        for arg in args:
            if values[arg] is not None:
                if values[arg] != '' and arg in self.__interact_category_format:
                    key_join = self.__interact_category_format[arg].format(values[arg])
                    self.__interact_category_dataset.append(key_join)

        return self.__interact_category_dataset

    def extend_interact_category(self, min_read: Union[str, int] = None):
        """扩展互动量过滤器"""
        return min_read

    def date_category(self, since: str = None, until: str = None):
        """日期过滤器"""

        frame = inspect.currentframe()
        args, _, _, values = inspect.getargvalues(frame)
        args.remove('self')

        values = dict(values)
        for arg in args:
            if values[arg] is not None:
                if values[arg] != '':
                    key_join = self.__date_category_format[arg].format(values[arg])
                    self.__date_category_dataset.append(key_join)

        return self.__date_category_dataset

easy_twitter_crawler/twitter_utils/twitter_filter.py
hanxinkong/easy-twitter-crawler
""""
推特工具类
"""
import datetime
import re
import requests
import time
import urllib3
from loguru import logger
from math import ceil
from my_fake_useragent import UserAgent
from urllib.parse import unquote, urljoin, quote
from easy_spider_tool import jsonpath, cookie_to_dic, clear_value, regx_match, for_to_regx_match, date_parse
from requests import RequestException

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

from easy_twitter_crawler.twitter_crawler.twitter_types import *

bearer = 'Bearer AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs%3D1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA'


def get_proxy():
    """
    获取代理
    """
    proxy_meta = {
        'http': 'http://127.0.0.1:10808',
        'https': 'http://127.0.0.1:10808'
    }
    return proxy_meta


def get_headers(cookie: str):
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/112.0.0.0 Safari/537.36 Edg/112.0.1722.68',
        'content-type': 'application/json',
        'x-twitter-active-user': 'yes',
        # 'x-twitter-auth-type': 'OAuth2Session',
        'authorization': bearer,
    }

    # 无需登录
    if not cookie:
        headers.update({
            'x-guest-token': gen_guest_token()
        })
    else:
        headers.update({
            'x-csrf-token': cookie_to_dic(cookie)['ct0'],
            'cookie': cookie,
        })
    return headers


def extract_time(date_obj: str) -> str:
    """
    时间处理
    """
    if not date_obj:
        return ''

    date_obj = (date_parse(
        date_obj.replace('+0000', ''),
        is_fmt=False,
        default_time_scheme=None,
        # timezone='Asia/Shanghai'
    ) + datetime.timedelta(hours=8)).strftime("%Y-%m-%d %H:%M:%S")

    return date_obj


def extract_error_info(json_data: Dict) -> PartialType:
    """
    处理错误消息
    @param json_data:
    @return:
    """

    return {
        'msg': jsonpath(
            json_data,
            [
                '$.errors..message',
                # 推文不可用/推文需要登录
                '$..itemContent.tweet_results.result.tombstone..text.text',
                '$..text.text',
                # 用户报错信息
                '$.data.user.result..text',
            ],
            first=True,
            default=''
        ),
        'name': jsonpath(
            json_data,
            '$.errors..name',
            first=True,
            default=''
        ),
        'type': jsonpath(
            json_data,
            '$.errors..kind',
            first=True,
            default=''
        ),
        'code': jsonpath(
            json_data,
            '$.errors..code',
            first=True,
            default=''
        ),
    }


def extract_cursor(json_data: Dict) -> str:
    """
    处理翻页cursor
    @param json_data:
    @return:
    """
    expr = [
        # 文章评论
        "$.data..instructions[0].entries.[?(@.cursorType=='Bottom'||@.cursorType=='ShowMoreThreads'||@.cursorType=='ShowMoreThreadsPrompt')].value",
        # 用户粉丝/关注
        "$.data.user.result.timeline.timeline.instructions[*].entries.[?(@.cursorType=='Bottom')].value",
        # 用户文章
        "$..entries.[?(@.cursorType=='Bottom')].value",
        # 搜索
        "$.data.search_by_raw_query.search_timeline.timeline.instructions..[?(@.cursorType=='Bottom')].value"
    ]

    cursor = jsonpath(
        json_data,
        expr,
        first=True,
        default=''
    )
    return cursor


def generating_page_link(base_url: str, cursor: str = None, **kwargs) -> Optional[str]:
    """
    根据cursor生成链接
    """
    page_url = None
    article_id = kwargs.pop('article_id', False)
    user_id = kwargs.pop('user_id', False)
    key_word = kwargs.pop('key_word', False)
    data_type = kwargs.pop('data_type', False)

    # 文章页
    if article_id:
        page_url = base_url.format(article_id=article_id, cursor=cursor)

    # 用户页
    if user_id:
        page_url = base_url.format(user_id=user_id, cursor=cursor)
    # 粉丝
    if data_type in ('followers', 'following'):
        if user_id and cursor:
            page_url = base_url.format(user_id=user_id) + f'&cursor={cursor}'

    # 搜索页
    if key_word:
        data_type = kwargs.pop('data_type', '')
        page_url = base_url.format(key_word=key_word, data_type=data_type, cursor=cursor)

    return page_url


def generating_page_links(base_url, total_num=50, limit=20):
    """
    根据总数及每页显示的个数生成下一页的urls
    @param base_url: 基础的url
    @param total_num: 数据总数 默认50
    @param limit: 每页展示的个数 默认20
    """
    page_urls = []
    for i in range(ceil(total_num / limit)):
        page_urls.append(re.sub(rf'start=\d+&count=\d+', f'start={i * limit}&count={limit}', base_url))
    return page_urls


def find_element(element, entity_id: str, key: str = 'entryId', first=True, default=None) -> Union[
    List[Dict[str, any]], Dict[str, any]]:
    if default is None:
        default = '' if first is True else []
    return jsonpath(element, f'$.[?(@["{key}"]=="{entity_id}")]', first=first, default=default)


def get_useragent():
    """
    随机获取useragent
    Returns: useragent
    """
    return UserAgent(phone=False, family='chrome', os_family='windows').random()


def gen_guest_token():
    """获取guest_token参数"""
    url_base = f'https://api.twitter.com/1.1/guest/activate.json'
    headers_ua = {
        'authorization': bearer,
        'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.0.0 Safari/537.36',
    }
    response = request('post', url_base, headers=headers_ua, proxies=get_proxy())
    if response is None:
        return
    res_json = response.json()
    guest_token = res_json.get('guest_token')
    return guest_token


class MaxRequestRetryCount(Exception):
    pass


def request(method: str, url: str, headers: Dict[str, str] = None, retry_num: int = 3, **kwargs) -> Response:
    retry_count = 0
    while retry_num >= retry_count:
        try:
            return requests.request(method.lower(), url, headers=headers, verify=False, **kwargs)

        except RequestException as rex:
            logger.error(
                f'request exception: {rex}' + f', kwargs: {kwargs}, next retry ...' if retry_count < retry_num else 'exit retry')
            retry_count += 1

    raise MaxRequestRetryCount(f'retry count: {retry_count}')

easy_twitter_crawler/twitter_utils/twitter_utils.py
hanxinkong/easy-twitter-crawler
# from gevent import monkey
# monkey.patch_all()
from easy_spider_tool import cookie_to_dic, format_json
from easy_twitter_crawler import set_proxy, set_cookie, common_crawler, user_crawler, search_crawler, TwitterFilter

proxy = {
    'http': 'http://127.0.0.1:10808',
    'https': 'http://127.0.0.1:10808'
}
# cookie = 'auth_token=686fa28f49400698820d0a3c344c51e3e44af73a; ct0=5bed99b7faad9dcc742eda564ddbcf37777f8794abd6d4d736919234440be2172da1e9a9fc48bb068db1951d1748ba5467db2bc3e768f122794265da0a9fa6135b4ef40763e7fd91f730d0bb1298136b'
cookie = ''
set_proxy(proxy)
set_cookie(cookie_to_dic(cookie))
num = 0

# 关键词采集使用案例（对关键词指定条件采集10条数据）
# key_word = 'elonmusk'
#
# twitter_filter = TwitterFilter(key_word)
# twitter_filter.word_category(lang='zh')
# twitter_filter.account_category(filter_from=None, to=None, at=None)
# twitter_filter.filter_category(only_replies=None, only_links=None, exclude_replies=None, exclude_links=None)
# twitter_filter.interact_category(min_replies=None, min_faves=None, min_retweets=None)
# twitter_filter.date_category(since='', until='')
# key_word = twitter_filter.filter_join()
#
# print(key_word)

# for info in search_crawler(
#         key_word,
#         data_type='Top',
#         count=10,
# ):
#     set_proxy(proxy)
#     set_cookie(cookie_to_dic(cookie))
#     num += 1
#     print(format_json(info))

# 用户信息采集使用案例（采集该用户信息及10条文章，10条回复，10个粉丝信息，10个关注信息）
for info in user_crawler(
        'elonmusk',
        article_count=10,
        reply_count=10,
        following_count=10,
        followers_count=10,
        # start_time='2023-07-20 00:00:00',
        # end_time='2023-07-27 00:00:00',
):
    set_proxy(proxy)
    set_cookie(cookie_to_dic(cookie))
    num += 1
    print(format_json(info))
    print(f"文章数：{len(info.get('article', []))}")
    print(f"粉丝数：{len(info.get('followers', []))}")
    print(f"关注数：{len(info.get('following', []))}")
    print(f"回复数：{len(info.get('reply', []))}")

# 通用采集使用案例（已知文章id，采集此文章信息）
# for info in common_crawler(
#         '1684447438864785409',
#         data_type='article',
#         # comment_count=1
# ):
#     set_proxy(proxy)
#     set_cookie(cookie_to_dic(cookie))
#     num += 1
#     print(format_json(info))
#
# print(num)

run/run.py
hanxinkong/easy-twitter-crawler
from os import path as os_path

from setuptools import setup, find_packages

from easy_twitter_crawler import __version__

this_directory = os_path.abspath(os_path.dirname(__file__))


# 读取文件内容
def read_file(filename):
    with open(os_path.join(this_directory, filename), encoding='utf-8') as f:
        long_description = f.read()
    return long_description


# 获取依赖
def read_requirements(filename):
    return [line.strip() for line in read_file(filename).splitlines()
            if not line.startswith('#')]


if __name__ == '__main__':
    setup(
        name='easy_twitter_crawler',  # 包名
        python_requires='>=3.6.8',  # python环境
        version=__version__,  # 包的版本
        description="简易、强大的推特（Twitter）采集程序,支持元搜索,用户,粉丝,关注,发文,回复,评论等采集",  # 包简介，显示在PyPI上
        long_description=read_file('README.md'),  # 读取的Readme文档内容
        long_description_content_type="text/markdown",  # 指定包文档格式为markdown
        author="hanxinkong",  # 作者相关信息
        author_email='xinkonghan@gmail.com',
        url='https://easy-twitter-crawler.xink.top/',
        packages=find_packages(),
        install_requires=read_requirements('requirements.txt'),  # 指定需要安装的依赖
        license="MIT",
        keywords=['easy', 'twitter', 'crawler', 'spider'],
    )
setup.py
hanxinkong/easy-twitter-crawler