如何将Twitter API返回的Python字典数据按列存储为目标格式CSV文件

问题原因 #

你当前的代码存在两个核心问题：

• 请求参数里重复定义了tweet.fields，后一次赋值会覆盖前一次，导致你只能拿到created_at字段，拿不到推文text内容
• 写入逻辑错误：你直接遍历了Twitter API返回的整个字典的键值对写入CSV，自然会得到键值逐行存储的结果。实际所有推文数据存放在返回结果的data字段中，是一个列表，每个元素对应一条完整推文。

修正步骤 #

1. 修正请求参数 #

把重复的tweet.fields合并，示例：

query_params = {
    'query': '(Johnson&Johnson) -is:retweet -is:verified -baby -lotion -shampoo',
    'tweet.fields': 'created_at,text', # 合并两个字段，用逗号分隔
    'start_time':'2021-01-20T00:00:01.000Z', 
    'end_time':'2021-02-17T23:30:00.000Z'
}

2. 替换CSV写入逻辑 #

你已经导入了pandas，用它处理最方便，直接把推文列表转成DataFrame后导出CSV即可，不需要自己遍历写行。

在main函数里，拿到返回结果之后加上导出逻辑即可：

def main():
    headers = create_headers(bearer_token)
    json_response = connect_to_endpoint(search_url, headers, query_params)
    # 不需要转成json字符串再load，直接用返回的字典即可
    y = json_response
    # 取推文列表
    tweet_list = y.get('data', [])
    if not tweet_list:
        print("未获取到推文数据")
        return
    # 导出为CSV，自动用字段做列名，每条推文占一行
    df = pd.DataFrame(tweet_list)
    df.to_csv('tweets.csv', index=False, encoding='utf-8-sig')

如果你不想用pandas，想用原生csv模块实现，就用csv.DictWriter：

def main():
    headers = create_headers(bearer_token)
    json_response = connect_to_endpoint(search_url, headers, query_params)
    y = json_response
    tweet_list = y.get('data', [])
    if not tweet_list:
        print("未获取到推文数据")
        return
    # 取所有字段名作为CSV列名
    field_names = tweet_list[0].keys()
    with open('tweets.csv', 'w', newline='', encoding='utf-8-sig') as f:
        writer = csv.DictWriter(f, fieldnames=field_names)
        writer.writeheader() # 写入表头
        writer.writerows(tweet_list) # 批量写入所有推文

完整调整后代码 #

import requests
import pandas as pd
import csv

bearer_token = "insert here"

search_url = "https://api.twitter.com/2/tweets/search/all"

# 修正后的请求参数
query_params = {
    'query': '(Johnson&Johnson) -is:retweet -is:verified -baby -lotion -shampoo',
    'tweet.fields': 'created_at,text',
    'start_time':'2021-01-20T00:00:01.000Z', 
    'end_time':'2021-02-17T23:30:00.000Z'
}

def create_headers(bearer_token):
    headers = {"Authorization": "Bearer {}".format(bearer_token)}
    return headers

def connect_to_endpoint(url, headers, params):
    response = requests.request("GET", search_url, headers=headers, params=params)
    print('请求状态码:', response.status_code)
    if response.status_code != 200:
        raise Exception(response.status_code, response.text)
    return response.json()

def main():
    headers = create_headers(bearer_token)
    json_response = connect_to_endpoint(search_url, headers, query_params)
    tweet_list = json_response.get('data', [])
    if not tweet_list:
        print("未获取到推文数据")
        return
    # pandas导出方式
    df = pd.DataFrame(tweet_list)
    df.to_csv('tweets.csv', index=False, encoding='utf-8-sig')
    # 原生csv模块导出方式（二选一即可）
    # field_names = tweet_list[0].keys()
    # with open('tweets.csv', 'w', newline='', encoding='utf-8-sig') as f:
    #     writer = csv.DictWriter(f, fieldnames=field_names)
    #     writer.writeheader()
    #     writer.writerows(tweet_list)

if __name__ == "__main__":
    main()

内容的提问来源于stack exchange，提问作者Poppy