Crypto期权宝典

import requestsimport zstandard as zstdimport jsonimport timeimport randomfrom requests.adapters import HTTPAdapterfrom requests.packages.urllib3.util.retry import Retry
# 禁用警告（如果你决定禁用SSL验证）import urllib3urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
def get_tweets(user_id, cursor=None, session=None):    url = "https://x.com/i/api/graphql/V7H0Ap3_Hh2FyS75OCDO3Q/UserTweets"        variables = {        "userId": user_id,        "count": 100,        "includePromotedContent": True,        "withQuickPromoteEligibilityTweetFields": True,        "withVoice": True,        "withV2Timeline": True    }    if cursor:        variables["cursor"] = cursor        params = {        "variables": json.dumps(variables),        "features": '{"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,"tweetypie_unmention_optimization_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}',        "fieldToggles": '{"withArticlePlainText":false}'    }        headers = {        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36 Edg/126.0.0.0",        "Accept": "*/*",        "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6",        "Accept-Encoding": "gzip, deflate, br, zstd",        "Referer": "https://x.com/Amy6Tina",        "Content-Type": "application/json",        "Authorization": "Bearer AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs%3D1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA",        "X-Csrf-Token": "60da60018c82ae37e1f3cd267d2b8cd24889e76c8b705898578702b82247fcad8b45477f9a0ff9d39775fc56289ff869108f0e3bc73a4a2a7d5fcec74e6551542d0c12fed80d3c03feda9ed1547bdf1c",        "X-Twitter-Auth-Type": "OAuth2Session",        "X-Twitter-Client-Language": "zh-cn",        "X-Twitter-Active-User": "yes"    }        cookies = {        "auth_token": "bd75016648bcfb3f0dc654142b6e37ee8c3b24bb",        "ct0": "60da60018c82ae37e1f3cd267d2b8cd24889e76c8b705898578702b82247fcad8b45477f9a0ff9d39775fc56289ff869108f0e3bc73a4a2a7d5fcec74e6551542d0c12fed80d3c03feda9ed1547bdf1c"    }        try:        response = session.get(url, params=params, headers=headers, cookies=cookies, verify=False)        response.raise_for_status()    except requests.RequestException as e:        if response.status_code == 429:            wait_time = int(response.headers.get('Retry-After', 60))            print(f"Rate limited. Waiting for {wait_time} seconds.")            time.sleep(wait_time)            return None, cursor  # Return None for tweets, but keep the cursor        print(f"Error: {e}")        return None, None
    try:        # 尝试解压缩响应内容        dctx = zstd.ZstdDecompressor()        decompressed_data = dctx.decompress(response.content)    except zstd.ZstdError as e:        print(f"Zstandard decompression error: {e}")        print(f"Response content type: {response.headers.get('Content-Type')}")        print(f"Response content length: {len(response.content)}")        print(f"First 100 bytes of response: {response.content[:100]}")        return None, None
    try:        # 解析 JSON        json_data = json.loads(decompressed_data)    except json.JSONDecodeError as e:        print(f"JSON decode error: {e}")        print(f"Decompressed data: {decompressed_data[:1000]}...")  # 打印前1000个字符        return None, None
    tweets = []    next_cursor = None        for instruction in json_data['data']['user']['result']['timeline_v2']['timeline']['instructions']:        if instruction['type'] == 'TimelineAddEntries':            for entry in instruction['entries']:                if 'itemContent' in entry['content']:                    if 'tweet_results' in entry['content']['itemContent']:                        tweet = entry['content']['itemContent']['tweet_results']['result']['legacy']['full_text']                        tweets.append(tweet)                elif entry['content']['entryType'] == 'TimelineTimelineCursor' and entry['content']['cursorType'] == 'Bottom':                    next_cursor = entry['content']['value']        return tweets, next_cursor
def get_all_tweets(user_id):    all_tweets = []    cursor = None        # 创建一个session对象，设置重试策略    session = requests.Session()    retries = Retry(total=5, backoff_factor=1, status_forcelist=[500, 502, 503, 504])    session.mount('https://', HTTPAdapter(max_retries=retries))        while True:        tweets, next_cursor = get_tweets(user_id, cursor, session)        if tweets is None and next_cursor is None:            break        elif tweets is None:            # If we hit a rate limit, wait and then continue with the same cursor            time.sleep(random.uniform(30, 60))  # Wait between 30 to 60 seconds            continue                all_tweets.extend(tweets)        print(f"Fetched {len(tweets)} tweets. Total: {len(all_tweets)}")                if not next_cursor:            break                cursor = next_cursor        time.sleep(random.uniform(2, 5))  # Random delay between 2 to 5 seconds        return all_tweets
# 使用示例user_id = "1453699244116774920"  # Amy6Tina 的用户IDall_tweets = get_all_tweets(user_id)
print(f"Total tweets fetched: {len(all_tweets)}")for tweet in all_tweets[:10]:  # 打印前10条推文    print(tweet)    print("---")
# 可以将所有推文保存到文件with open('all_tweets.txt', 'w', encoding='utf-8') as f:    for tweet in all_tweets:        f.write(tweet + '\n\n')