如何用Python自动化批量请求里斯本城市开放传感器JSON数据

批量查询实现方案 #

你可以通过以下方式实现自动化批量查询，同时兼顾服务端友好性和容错能力：

实现思路

• 批量生成0001到0080格式的站点编号，统一管理查询起止时间
• 增加请求间隔避免对服务端造成过大压力，符合非商业使用的规范
• 增加异常捕获和重试逻辑，单站点请求失败不会中断整体任务
• 自动合并所有站点的返回数据为统一DataFrame，可直接导出为文件备用

完整实现代码

# -*- coding: utf-8 -*-
import pandas as pd 
import urllib.request, json 
import time
from flatten_json import flatten

# 配置参数，可根据需求修改
START_DATE = "202104010000"
END_DATE = "202108310000"
SITE_COUNT = 80
REQUEST_INTERVAL = 1.5  # 两次请求的间隔秒数，可根据实际情况调整
MAX_RETRY = 3  # 单站点请求失败后的最大重试次数

# 存储所有站点的数据
all_data = []

for site_num in range(1, SITE_COUNT + 1):
    # 格式化站点编号为4位补零格式，如0001、0080
    site_id = f"{site_num:04d}"
    url = f"http://opendata-cml.qart.pt/measurements/RULAEQ{site_id}?startDate={START_DATE}&endDate={END_DATE}"
    
    # 重试逻辑
    retry_count = 0
    success = False
    while retry_count < MAX_RETRY and not success:
        try:
            with urllib.request.urlopen(url, timeout=10) as resp:
                data = json.loads(resp.read().decode())
            # 扁平化JSON并转为DataFrame
            data_flattened = [flatten(d) for d in data]
            df = pd.DataFrame(data_flattened)
            # 新增站点编号列，方便后续区分不同站点的数据
            df["site_id"] = site_id
            all_data.append(df)
            print(f"站点{site_id}数据获取成功，共{len(df)}条记录")
            success = True
        except Exception as e:
            retry_count += 1
            print(f"站点{site_id}第{retry_count}次请求失败，错误信息：{str(e)}")
            time.sleep(2)
    
    # 请求间隔
    time.sleep(REQUEST_INTERVAL)

# 合并所有站点的数据
total_df = pd.concat(all_data, ignore_index=True)
# 可选：导出为CSV文件
total_df.to_csv("lisbon_noise_data_202104-202108.csv", index=False, encoding="utf-8-sig")
print(f"所有数据获取完成，总记录数：{len(total_df)}")

优化说明

如果后续需要查询更多站点或者更大时间范围的数据，可以改用aiohttp实现异步请求进一步提升效率，不过对于当前80个站点的量级，上述同步方案已经足够稳定高效，且逻辑简单易于维护。

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