SQLite3字符串日期列对比报错与查询无结果问题求解

• 逻辑顺序完全颠倒：先查询旧表最新日期，随后用to_sql的if_exists='replace'参数直接覆盖全表，之前查询的旧日期完全失效
• SQL语句拼接错误：调整后的代码没有加f-string标识，{data}没有被替换为实际的日期值，查询条件失效
• 没有处理首次运行的边界情况：第一次执行时表不存在或表内无数据，fetchone()返回None，直接取data[0]会抛出异常
• 字符串拼接SQL的风险：直接拼接日期字符串容易引发格式错误、SQL注入问题，应该使用参数化查询或者pandas内置过滤能力

import tweepy
import time
import pandas as pd
import sqlite3

con = sqlite3.connect('tweetScaping.db')
cur = con.cursor()
consumer_key = "**********************"
consumer_secret = "****"
access_token = "****-*****"
access_token_secret = "***************"
auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token, access_token_secret)
api = tweepy.API(auth, wait_on_rate_limit=True)

# 搜索关键词
text_query = 'bouygues telecom OR @bouyguestelecom OR #Bouygues'
count = 200

try:
    # 1. 爬取最新数据
    tweets = tweepy.Cursor(api.search_tweets, q=text_query + " -filter:retweets").items(count)
    tweets_list = [[tweet.created_at, tweet.id, tweet.text] for tweet in tweets]
    tweets_df = pd.DataFrame(tweets_list, columns=['tweet_created_at', 'tweet_id', 'tweet_text'])
    # 统一日期格式为字符串，避免sqlite类型转换问题
    tweets_df['tweet_created_at'] = tweets_df['tweet_created_at'].astype(str)

    # 2. 查询旧表最新日期，兼容首次运行
    latest_date = None
    # 先判断表是否存在
    cur.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='tweets_b_db'")
    if cur.fetchone() is not None:
        cur.execute("SELECT MAX(tweet_created_at) FROM tweets_b_db")
        res = cur.fetchone()
        if res[0] is not None:
            latest_date = res[0]

    # 3. 过滤出比最新日期大的新增数据
    if latest_date is not None:
        new_tweets = tweets_df[tweets_df['tweet_created_at'] > latest_date]
    else:
        new_tweets = tweets_df

    # 4. 追加写入数据库，不要用replace覆盖
    if not new_tweets.empty:
        new_tweets.to_sql(name='tweets_b_db', con=con, if_exists='append', index=False)
        print(f"新增{len(new_tweets)}条记录")
    else:
        print("无新增数据")

except BaseException as e:
    print('执行失败:', str(e))
    time.sleep(3)
finally:
    # 提交修改并关闭连接
    con.commit()
    con.close()

• 写入方式改为if_exists='append'，只追加新数据不覆盖历史数据，满足每小时增量爬取的需求
• 调整逻辑顺序：先爬新数据→再查历史最新日期→过滤新增数据→最后写入，避免查询的旧日期失效
• 增加表存在判断、空数据判断，兼容首次运行的场景
• 统一日期转为字符串格式存储，sqlite中ISO格式的日期字符串字典序和时间序完全一致，可以直接比较大小，不需要额外调用Datetime函数转换

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