定时每5秒下载URL数据并保存至CSV的技术问题求助

现有代码的核心问题 #

1. 文件操作错误：使用wb二进制写入模式会每次清空文件，无法实现追加；且直接写入字典对象不符合CSV格式规范。
2. 依赖缺失：未导入time模块，执行time.sleep()会抛出异常。
3. 非法调用：data()是错误写法，data是请求返回的字典对象，并非可调用函数。
4. 循环阻塞：time.sleep(60)会让程序每分钟才检查一次定时任务，完全无法实现每5秒执行一次的需求。
5. 表头逻辑缺失：没有实现首次写入表头、后续仅追加数据行的逻辑。

修正后的代码 #

import schedule
import requests
import csv
import time
from pathlib import Path

CSV_FILE = 'cam_data.csv'
TARGET_URL = '你的目标URL'  # 替换为实际请求URL
RUN_DURATION = 60  # 总运行时长（1分钟）

def fetch_and_save_data():
    # 请求并解析数据
    try:
        resp = requests.get(TARGET_URL)
        resp.raise_for_status()  # 捕获HTTP请求错误
        data = resp.json()
    except Exception as e:
        print(f"数据请求失败: {e}")
        return

    # 处理JSON中的null值，转换为空字符串保证CSV格式合法
    processed_data = {k: v if v is not None else '' for k, v in data.items()}

    # 判断文件是否存在，决定是否写入表头
    file_exists = Path(CSV_FILE).exists()

    # 追加写入CSV
    with open(CSV_FILE, 'a', newline='', encoding='utf-8') as f:
        writer = csv.DictWriter(f, fieldnames=processed_data.keys())
        if not file_exists:
            writer.writeheader()  # 首次运行写入表头
        writer.writerow(processed_data)  # 追加新数据行
    print(f"数据已写入: {processed_data}")

# 设置每5秒执行一次采集任务
schedule.every(5).seconds.do(fetch_and_save_data)

# 控制任务运行1分钟后停止
start_time = time.time()
while time.time() - start_time < RUN_DURATION:
    schedule.run_pending()
    time.sleep(1)  # 每秒检查一次任务，避免阻塞

print("1分钟数据采集任务已完成")

关键改进说明 #

• 使用csv.DictWriter处理CSV写入，自动匹配表头与数据列，无需手动拼接字符串
• 采用a追加模式打开文件，配合newline=''避免Windows系统下出现空行
• 主动处理JSON中的null值，转换为空字符串保证CSV格式有效性
• 通过Path判断文件是否存在，实现首次运行自动写入表头的逻辑
• 主循环控制总运行时长为1分钟，每秒检查一次定时任务，确保任务按时触发
• 添加异常捕获，处理请求失败的异常情况

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