如何基于Tweepy动态扩展SQLite表并插入推特JSON数据

针对你用tweepy.StreamingClient采集推特数据时遇到的data字段结构动态变化问题，这里提供一套从初始窄表动态扩展列并插入数据的实现方案，核心是扁平化嵌套字段+动态检查并添加列+参数化插入。

1. 扁平化嵌套JSON字段 #

先把推特返回的嵌套data结构转成扁平的键值对，方便后续数据库操作。比如将geo.place_id转为geo_place_id，geo.coordinates.coordinates转为geo_coordinates_coordinates：

def flatten_json(data, parent_key='', sep='_'):
    items = {}
    for k, v in data.items():
        new_key = f"{parent_key}{sep}{k}" if parent_key else k
        if isinstance(v, dict):
            items.update(flatten_json(v, new_key, sep=sep))
        else:
            items[new_key] = v
    return items

2. 动态检查表结构并添加缺失列 #

每次插入数据前，先获取当前表的所有列，对比扁平化后的字段，将缺失的字段作为新列添加到表中（统一用TEXT类型兼容所有数据类型）：

import sqlite3

def get_table_columns(conn, table_name):
    cursor = conn.cursor()
    cursor.execute(f"PRAGMA table_info({table_name})")
    columns = [row[1] for row in cursor.fetchall()]
    cursor.close()
    return columns

def add_missing_columns(conn, table_name, flattened_data):
    current_cols = get_table_columns(conn, table_name)
    cursor = conn.cursor()
    for field in flattened_data.keys():
        if field not in current_cols:
            cursor.execute(f"ALTER TABLE {table_name} ADD COLUMN {field} TEXT")
            current_cols.append(field)
    conn.commit()
    cursor.close()

3. 动态插入数据 #

用参数化查询构建插入语句，避免SQL注入，同时适配动态变化的字段：

def insert_tweet_data(conn, table_name, flattened_data):
    fields = flattened_data.keys()
    placeholders = ', '.join(['?'] * len(fields))
    sql = f"INSERT INTO {table_name} ({', '.join(fields)}) VALUES ({placeholders})"
    cursor = conn.cursor()
    cursor.execute(sql, tuple(flattened_data.values()))
    conn.commit()
    cursor.close()

4. 整合到流处理回调中 #

在自定义的StreamingClient子类中，把上述逻辑整合到on_response方法里：

class TweetStream(tweepy.StreamingClient):
    def __init__(self, bearer_token, db_conn, table_name):
        super().__init__(bearer_token)
        self.db_conn = db_conn
        self.table_name = table_name

    def on_response(self, response):
        if not response.data:
            return
        # 只处理data字段，忽略includes中的内容
        flattened_data = flatten_json(response.data.data)
        # 添加缺失列
        add_missing_columns(self.db_conn, self.table_name, flattened_data)
        # 插入数据
        insert_tweet_data(self.db_conn, self.table_name, flattened_data)

# 初始化数据库与表（初始表至少要有一个主键列，用推特的id避免重复）
db_conn = sqlite3.connect('twitter_data.db')
cursor = db_conn.cursor()
cursor.execute("CREATE TABLE IF NOT EXISTS tweets (id TEXT PRIMARY KEY)")
db_conn.commit()
cursor.close()

# 启动流采集
stream = TweetStream('你的Bearer Token', db_conn, 'tweets')
# 添加流规则（示例：采集包含特定关键词的推文）
stream.add_rules(tweepy.StreamRule('python'))
stream.filter()

关键注意事项 #

• 性能优化：高频流场景下，单条数据检查列会有损耗，可以攒10-100条数据后，合并所有字段再统一添加列，批量插入。
• 主键冲突：推特的id是全局唯一的，初始表设置id为主键，避免重复插入相同推文。
• 数据类型：用TEXT类型兼容所有推特数据，如果需要特定类型（比如数字），可以在扁平化时判断字段类型，动态指定列类型，但会增加复杂度，研究场景下TEXT足够。

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