Python日期比较功能实现及代码优化咨询

嘿，作为Python新手能写出这样的实用脚本已经超棒了！针对你想要实现的「追踪上次检查日期、只输出新比赛」的需求，咱们来梳理下更合理的实现思路，顺便给你的代码做些优化～

关于「上次检查日期」的存储方案 #

你提到的全局字典思路在脚本运行时没问题，但脚本结束后数据就会丢失，下次重启又得重新记录——直接把日期存在CSV里才是最优解：毕竟CSV本身就是你的赛事数据源，给每行加个上次检查日期字段，读取时加载成可操作的对象，处理完再写回CSV，既能持久化数据，又能和现有工作流完美衔接。

核心实现逻辑 #

1. 读取CSV并解析日期：把每行的赛事名称、数据库链接、上次检查日期（转成datetime对象）存成结构化数据（比如字典列表），方便后续对比。
2. 逐个处理赛事：下载对应SQLite数据库后，先查询赛事的最新比赛结束时间，再对比上次检查日期，只提取之后的新比赛。
3. 更新并保存日期：如果有新比赛，就把CSV里的日期更新为数据库中最新的比赛结束时间；就算没有新比赛，也可以更新为当前时间（或者保持最新比赛时间，看你需求）。
4. 写回CSV：把处理后的所有赛事数据重新写入CSV，确保下次运行能读取到最新的检查日期。

重构后的完整代码 #

import csv
import sys
import urllib.request
import sqlite3
import datetime
from pathlib import Path

# 初始化路径（用pathlib替代os.path，跨平台更友好）
datapath = Path(sys.argv[0]).parent / "data"
datapath.mkdir(exist_ok=True)  # 自动创建data文件夹（如果不存在）
datafile = datapath / "sqlcompetitions.csv"

def load_competitions(csv_file):
    """加载CSV中的赛事数据，解析上次检查日期为datetime对象"""
    competitions = []
    try:
        with open(csv_file, newline='', encoding='utf-8') as f:
            # 定义CSV字段名：赛事名称、数据库链接、上次检查日期
            reader = csv.DictReader(f, fieldnames=["name", "url", "last_checked"])
            next(reader)  # 跳过表头（如果你的CSV原来没有表头，这行可以删掉）
            
            for row in reader:
                # 处理空日期：如果是第一次运行，设为一个极早的时间（确保能提取所有历史比赛）
                last_checked_str = row["last_checked"].strip()
                if last_checked_str:
                    last_checked = datetime.datetime.fromisoformat(last_checked_str)
                else:
                    last_checked = datetime.datetime.min
                
                competitions.append({
                    "name": row["name"].strip(),
                    "url": row["url"].strip(),
                    "last_checked": last_checked
                })
    except FileNotFoundError:
        print(f"❌ 错误：未找到文件 {csv_file}")
        sys.exit(1)
    return competitions

def process_single_competition(competition):
    """处理单个赛事：下载数据库、提取新比赛、更新检查日期"""
    name = competition["name"]
    url = competition["url"]
    last_checked = competition["last_checked"]
    db_path = datapath / f"{name}.sqlite"

    print(f"\n🔍 正在处理赛事：{name}")

    # 下载最新SQLite数据库
    try:
        urllib.request.urlretrieve(url, db_path)
        print(f"✅ 已下载{name}的最新数据库")
    except Exception as e:
        print(f"❌ 下载{name}数据库失败：{str(e)}")
        return competition  # 下载失败，返回原数据不更新

    # 连接数据库并提取新比赛
    conn = None
    try:
        conn = sqlite3.connect(db_path)
        cursor = conn.cursor()

        # 获取赛事最新的比赛结束时间
        cursor.execute("SELECT MAX(finished) FROM leaguematches")
        latest_finished_str = cursor.fetchone()[0]
        
        if not latest_finished_str:
            print(f"ℹ️ {name}暂无比赛数据")
            # 无比赛时更新检查日期为当前时间
            competition["last_checked"] = datetime.datetime.now()
            return competition

        # 转换数据库时间为datetime对象（假设数据库中finished字段是ISO格式字符串，如'2024-05-20 14:30:00'）
        latest_finished = datetime.datetime.fromisoformat(latest_finished_str)

        # 查询上次检查之后的新比赛
        cursor.execute("""
            SELECT finished, tvhome, idracehome, coachhome, teamhome, scorehome, 
                   tvaway, idraceaway, coachaway, teamaway, scoreaway 
            FROM leaguematches 
            WHERE finished > ?
            ORDER BY finished ASC
        """, (last_checked.isoformat(),))

        new_matches = cursor.fetchall()
        if new_matches:
            print(f"🎉 找到{len(new_matches)}场新比赛：")
            for row in new_matches:
                print(f"DATA: {row[0]}; TVHOME: {row[1]}; IDRACEHOME: {row[2]}; COACHHOME: {row[3]}; TEAMHOME: {row[4]}; SCOREHOME: {row[5]}; TVAWAY: {row[6]}; IDRACEAWAY: {row[7]}; COACHAWAY: {row[8]}; TEAMAWAY: {row[9]}; SCOREAWAY: {row[10]};")
            # 更新检查日期为最新比赛的结束时间
            competition["last_checked"] = latest_finished
        else:
            print(f"ℹ️ 自{last_checked.strftime('%Y-%m-%d %H:%M')}以来，{name}暂无新比赛")
            # 无新比赛时，更新检查日期为当前时间（或保持latest_finished，按需调整）
            competition["last_checked"] = datetime.datetime.now()

    except Exception as e:
        print(f"❌ 处理{name}数据库失败：{str(e)}")
    finally:
        if conn:
            conn.close()

    return competition

def save_competitions(csv_file, competitions):
    """将更新后的赛事数据写回CSV"""
    with open(csv_file, 'w', newline='', encoding='utf-8') as f:
        writer = csv.DictWriter(f, fieldnames=["name", "url", "last_checked"])
        writer.writeheader()  # 写入表头
        for comp in competitions:
            writer.writerow({
                "name": comp["name"],
                "url": comp["url"],
                "last_checked": comp["last_checked"].isoformat()
            })
    print(f"\n✅ 赛事数据已更新并保存到 {csv_file}")

def main():
    competitions = load_competitions(datafile)
    if not competitions:
        print("ℹ️ 没有可处理的赛事")
        return
    
    # 批量处理所有赛事
    updated_competitions = [process_single_competition(comp) for comp in competitions]
    
    # 保存更新结果
    save_competitions(datafile, updated_competitions)

if __name__ == "__main__":
    main()

额外的代码优化建议 #

1. 用pathlib替代os.path：跨平台兼容性更好，代码更简洁，不用手动拼接路径分隔符。
2. 细化异常处理：每个步骤（下载、数据库操作）单独捕获异常，避免一个赛事失败导致整个脚本崩溃。
3. 去掉不必要的time.sleep(1)：打印比赛信息不需要延迟，会拖慢脚本运行效率。
4. 参数化SQL查询：用WHERE finished > ?这种参数化写法，既能避免SQL注入（虽然是本地数据库，但养成好习惯），又能避免字符串拼接的错误。
5. 函数职责单一化：每个函数只做一件事（加载数据、处理单个赛事、保存数据），代码可读性和可维护性更强。

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