尧图网站建设 
简介本资源是一套面向Python数据采集与环境数据分析初学者及从业者的实战工具包解决从中国环境监测总站CNEMC稳定获取实时空气质量六参数PM2.5、PM10、SO₂、NO₂、CO、O₃的工程化需求。资源包含3371个文件主体为3354个按小时命名的CSV数据文件如2022-03-16T19.csv覆盖逐小时浓度均值与8小时滑动平均值等关键指标辅以11个Python脚本含爬取、清洗、校验逻辑、3个Jupyter Notebook含可视化与延时说明分析、1份README.md文档及配置文件结构清晰、开箱即用。压缩包大小为407.68MB已累计被1941人学习下载。读者可直接复用数据采集框架、理解CNEMC数据发布机制如半小时延滞、小时均值计算规则、验证字段完整性并基于真实历史数据开展趋势分析、模型训练或教学演示。 用Python抓取中国环境总站CNEMC的实时空气质量数据这件事听起来很简单网上随手一搜就是一堆教程。但真上手你就会发现老文章的接口早就失效了要么城市编码对不上要么返回一堆你看不懂的字段。我自己前前后后踩了不少坑干脆把一套能跑的完整方案整理出来从找接口、发请求、解析数据到落库一次讲清楚。这篇文章适合刚接触Python爬虫的人也适合想拿空气质量数据做分析、做可视化或者做毕业设计的同学。1. 动手之前先搞懂CNEMC的实时数据到底藏在哪1.1 从网页到接口数据是怎么来的中国环境监测总站CNEMC做过一个“全国城市空气质量实时发布平台”网页上能看到每个城市的实时AQI、PM2.5、PM10、二氧化硫、二氧化氮、臭氧、一氧化碳这些指标。页面上的数据不是凭空生成的它背后一定有一个接口在给前端喂数据。我们写Python去抓数据本质上是模拟浏览器去请求这个接口然后把返回的JSON解析出来。我早期犯过一个错误直接拿requests去请求网页本身结果发现页面是服务端渲染加动态加载的拉下来一堆HTML标签真正有用的数据全在XHR请求里。所以正确思路是先打开浏览器开发者工具F12切到Network面板勾选XHR然后刷新页面看哪些请求返回的是JSON。CNEMC平台的接口有个特点返回体经常是数组套对象的结构顶层就是列表里面每个元素对应一个城市或一个站点的监测数据字段名基本是英文缩写一眼能看出来。有人可能会问直接用selenium模拟浏览器行不行当然行但没必要。selenium要起浏览器、要装驱动、跑起来还吃内存放在定时任务里容易崩。我们这里只抓JSONrequests就够了轻量、可控、出错还好排查。1.2 一份实时数据的字段字典第一次拿到返回数据你可能有点懵满屏的“AQI”、“PM2.5”、“PrimaryPollutant”不知道哪些该留哪些该扔。我整理了一份常用字段说明在做数据清洗之前最好先把这个搞清楚。字段含义常见单位City城市名称字符串Station监测站点名称字符串AQI空气质量指数无量纲PM2.5细颗粒物μg/m³PM10可吸入颗粒物μg/m³SO2二氧化硫μg/m³NO2二氧化氮μg/m³O3臭氧μg/m³CO一氧化碳mg/m³Quality空气质量等级优/良/轻度污染等PrimaryPollutant首要污染物字符串TimePoint数据时间时间字符串注意CO的单位和其他不一样是mg/m³不是μg/m³换算的时候要小心。我见过有人把CO和PM2.5直接放在一张图里对比结果曲线完全没法看就是因为单位没归一。另外PrimaryPollutant如果接口返回的是“NA”或者“-”表示当前没有首要污染物这种情况通常对应空气质量为优。1.3 环境准备装好这三个库就够了这个项目不需要复杂的环境Python 3.8以上的版本就行。依赖库严格来说只有一个硬需求requests。如果后面要做数据处理和可视化再补pandas、matplotlib。当然如果你用的是Anaconda这几个库大概率都装好了。pip install requests pip install pandas pip install matplotlib为什么我不推荐一开始就上scrapy这样的重型框架因为单次抓取、单城市数据用requests是性价比最高的选择代码量少逻辑直观。scrapy适合大规模分布式采集对空气质量这种小体量数据属于杀鸡用牛刀。等你的需求真到了要并发抓几百个城市、还要做增量更新和数据去重的时候再考虑框架不迟。2. 核心链路用requests把实时数据拿回来2.1 抓包确认请求地址与参数接下来是整篇文章最关键的环节——找到真实接口。CNEMC这个平台的历史版本不少接口地址变过几次我这次实测用的是一套比较稳定的地址。打开空气质量发布平台页面在Network里找XHR请求正常情况下能看到一个包含“API”字样的请求点开它的Headers可以看到完整的请求URL和参数。我实测的地址是这样的http://106.37.208.243:8000/AirQuality/API/Now/GetAQIData请求参数里通常包含city和typecity是城市拼音小写比如beijing、shanghai、guangzhoutype是数据类型实时报一般是HOUR日报是DAY。如果你在页面上选择了一个城市接口会返回该城市当天的实时AQI数据。还有一个常用接口是获取城市内所有站点的数据http://106.37.208.243:8000/AirQuality/API/Now/CityStation/beijing这个接口返回的是北京各监测子站比如万寿西宫、奥体中心等的实时数据。要注意如果你在抓包时发现接口地址和我写的不完全一样别慌很可能是平台更新了版本。你只要把Network里最新的URL复制下来替换掉代码里的BASE_URL就行整体逻辑一样。2.2 请求头、时间参数与城市编码三个关键细节第一个细节是请求头。直接裸请求很可能被拒至少要带上User-Agent让他看起来像个正常浏览器。有些版本的接口还会校验Referer少了Referer会返回空数据或者直接403。我习惯在代码里定义一个HEADERS字典把常用的UA、Referer、Accept都写上一劳永逸。第二个细节是时间参数。部分接口支持按时间范围查询参数名可能是startTime、endTime格式是“yyyy-MM-dd HH:mm:ss”。实时数据接口一般不需要传时间默认返回最新时次但如果发现返回的数据是空数组你可以试着显式传一个当前时间比如用datetime模块生成from datetime import datetime now_str datetime.now().strftime(%Y-%m-%d %H:%M:%S)第三个细节是城市编码。我这套接口用的是城市拼音但有些别的平台用的是行政区划编码比如110000表示北京。如果你抓到的接口用的是编码手头又只有城市名单就得多做一步编码映射。我的建议是维护一个城市拼音字典或者直接从接口返回里提取城市列表先抓一次全量数据看看响应的City字段都长什么样再决定要不要走映射。2.3 写一个稳定的请求函数下面给出一段可以直接跑的示例代码我加了一些异常处理避免因为网络抖动或接口返回异常导致整个脚本崩掉。import requests from datetime import datetime import time HEADERS { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36, Referer: http://106.37.208.243:8000/, Accept: application/json, text/javascript, */*; q0.01, } BASE_URL http://106.37.208.243:8000/AirQuality/API/Now/GetAQIData def fetch_realtime_aqi(city: str beijing, retry: int 3): params {city: city, type: HOUR} for attempt in range(retry): try: resp requests.get(BASE_URL, paramsparams, headersHEADERS, timeout10) resp.raise_for_status() data resp.json() if not data: print(f[{datetime.now()}] {city} 返回空数据可能接口参数有误) return [] return data except requests.exceptions.Timeout: print(f[{datetime.now()}] {city} 请求超时第 {attempt 1} 次重试) time.sleep(2) except requests.exceptions.HTTPError as e: print(f[{datetime.now()}] {city} HTTP 错误: {e}) if attempt retry - 1: raise time.sleep(2) except requests.exceptions.JSONDecodeError: print(f[{datetime.now()}] {city} 返回内容不是合法 JSON) time.sleep(2) return []这里我用了一个retry参数默认重试3次每次失败后睡2秒。实测下来对于空气质量这种官方接口重试机制非常有用因为服务端偶尔会抽风但不至于长时间不可用。timeout设成10秒也是经验值太短容易误判超时太长会影响定时任务的整体节奏。3. 数据到手之后解析、清洗与落库实操3.1 JSON解析与异常值处理接口返回的JSON格式一般是列表套字典所以resp.json()拿到手以后直接遍历就行。但这里有个坑不同城市返回的字段可能不一样有的城市空气质量好PrimaryPollutant就会是“NA”有的站点可能某个污染物数据缺失返回“-”或者空字符串。我的处理方式是先把数据转成统一的字典结构再交给pandas处理。这里的关键在于对缺失值的统一判断我写了一个小函数把各种异常值全部转成None后续用pandas处理就会方便很多。import pandas as pd def normalize_value(value): if value in (NA, -, , None): return None try: return float(value) except (ValueError, TypeError): return None def parse_aqi_data(data): rows [] for item in data: rows.append({ city: item.get(City), time: item.get(TimePoint), aqi: normalize_value(item.get(AQI)), pm25: normalize_value(item.get(PM2.5)), pm10: normalize_value(item.get(PM10)), so2: normalize_value(item.get(SO2)), no2: normalize_value(item.get(NO2)), o3: normalize_value(item.get(O3)), co: normalize_value(item.get(CO)), quality: item.get(Quality), primary_pollutant: item.get(PrimaryPollutant), }) return pd.DataFrame(rows)把异常值统一成None是为了后面mean()、dropna()这些操作能正常发挥作用。真实数据里经常会有某个站点某小时数据缺席的情况如果你不做清洗直接画图折线图上会出现一个很扎眼的缺口不知道的人还以为是采集程序出Bug了。3.2 单位换算与AQI等级自动判定拿到字段以后如果只是展示那原始单位就够了。但如果你想做多城市对比或者把数据入库做长期分析建议先统一单位。CO这里特别容易搞混它在接口里通常是mg/m³如果要在和PM2.5的同一张图里展示乘1000转成μg/m³更合理。AQI等级也可以用代码自动判定防止接口没返回Quality字段时抓瞎。按照中国标准AQI分六级AQI区间等级0-50优51-100良101-150轻度污染151-200中度污染201-300重度污染大于300严重污染我在做数据可视化时经常需要自己写一个判定函数因为有时候接口返回的Quality字段是中文有时候又是英文规范不一致。自己定一套判别逻辑最简单也最可控def aqi_level(aqi): if aqi is None: return 无数据 aqi int(aqi) if aqi 50: return 优 elif aqi 100: return 良 elif aqi 150: return 轻度污染 elif aqi 200: return 中度污染 elif aqi 300: return 重度污染 else: return 严重污染3.3 增量写入CSV与数据库的思路数据量小的时候CSV是最省事的存储方案。我习惯用pandas的to_csv加上modea追加写入同时把encoding设成utf-8-sig这样用Excel打开也不会乱码。注意追加写入时不要每次写表头否则文件里会有一堆重复的列名。import os CSV_FILE aqi_history.csv def save_to_csv(df, file_pathCSV_FILE): if df.empty: return header not os.path.exists(file_path) or os.path.getsize(file_path) 0 df.to_csv(file_path, modea, headerheader, indexFalse, encodingutf-8-sig)如果你是做长期采集建议别只存CSV最好同步写一份到SQLite或者MySQL里方便按时间范围查询。SQLite不需要额外服务Python自带sqlite3模块几十行代码就能建表插入数据。存数据库的另一个好处是可以做去重避免定时任务重复抓取时产生大量重复记录。最简单的去重策略是拿citytimepoint做唯一键插入前先查一下有没有。4. 踩坑实录这些坑我替你先踩了4.1 访问被拒绝八成是UA和Referer在拦你最常遇到的现象是浏览器里打开接口地址好好的Python一请求就返回403或者返回空列表。我一开始也很纳闷后来把浏览器开发者工具里的请求头完整复制到代码里问题立刻消失。其实服务端主要就是看User-Agent和Referer。解决办法很简单把代码里的HEADERS补全Referer填平台首页地址就行。另外不要短时间内疯狂请求官方接口虽然没明确写限制但你几秒钟刷几百次被封IP也是早晚的事。我自己的经验是抓实时数据的话五分钟刷一次就够了。空气质量数据本身不是秒级变化你刷得再频繁拿到的东西也差不多反而增加被封风险。4.2 返回“暂无数据”怎么办有时候接口通了但返回的数组是空的。这个问题的原因五花八门我遇到过三种第一个是城市编码不对比如你传了citybeijing但接口实际期望的是“北京”或者反过来。我一般会先用接口文档或者抓包确认一下。第二个是type参数不对实时数据有时用“HOUR”有时用“REAL”版本不一样。第三个是时间参数问题如果传了startTime和endTime要确保时间格式和时区正确。官方平台用的是北京时间如果你的服务器是UTC时区记得加8小时否则查出来的永远是“未来”或者“过去”的数据。我的排查习惯是遇到空数组先手动把URL复制到浏览器里访问一下看看到底返回什么。浏览器里能看到数据说明接口和参数没问题问题出在代码的headers或者编码上浏览器里也是空数组那就要检查参数了。4.3 定时任务挂机采集的稳定性设计如果打算让脚本长期定时跑稳定性比功能本身更重要。我一共踩过两次比较大的坑。第一次是脚本跑了一个月突然某天接口返回的数据格式变了原来能解析的字段找不到了整个任务直接抛异常后面的几十次采集全部失败。后来我在解析函数外面套了一层try/except即便某次数据格式异常也能记录日志继续跑下去不至于全盘崩掉。第二次是网络问题服务器偶尔会丢包或者DNS解析变慢。解决方案就是前面代码里的重试机制多加一个指数退避。第一次失败等2秒第二次等4秒第三次等8秒最多重试三次再不行就用日志记录下来留给人工排查。用cron调度的时候注意把脚本的日志重定向到文件里*/5 * * * * cd /path/to/project python fetch_aqi.py logs/fetch.log 21这样即使出错你也能从日志里看到完整堆栈而不是深夜被一条“脚本挂了”的短信吵醒。4.4 常见问题速查表现象可能原因解决方案403 Forbidden缺少User-Agent或Referer补全请求头模拟浏览器访问返回空列表城市编码或type参数有误浏览器直接访问接口确认参数JSON解析异常接口返回了HTML错误页打印resp.text排查增加重试字段全是None字段名对不上先打印原始JSON确认字段名中文乱码编码设置不对设置resp.encodingutf-8采集一段时间后停止IP被限制或异常未捕获降低频率、增加异常捕获和日志追加写入重复表头to_csv每次都写header判断文件是否存在仅首次写表头这些坑其实都很小但每一个都足以让你在深夜怀疑人生。尤其是“浏览器能打开、Python不行”的这类问题本质上是你的请求头还不够像浏览器把Headers里的关键字段都带上就好。我在实际使用中还有一个习惯就是每次采集完之后顺手把数据按城市拆分存成独立文件这样后续做单城市分析的时候不用反复过滤。另外CNEMC的接口偶尔会有数据延迟刚刚整点的时候抓可能拿到的还是上一小时的数据这时候不用慌过几分钟再抓一次就正常了。如果你要做实时推送提醒建议在逻辑里加一个“数据时间”判断别拿旧数据当真数据。这套方案跑起来以后后面能玩的花样就多了。比如把历史数据积累一段时间做一个空气质量趋势分析看看哪些城市的季节性规律最明显也可以做个简单的定时提醒当某个城市的AQI超过阈值时自动给你发邮件。数据源本身是公开的采集门槛也不高真正值钱的是你拿这些数据做出的分析和判断。本文还有配套的精品资源点击获取 Python爬虫实战：抓取CNEMC实时空气质量数据全攻略
简介本资源是一套面向Python数据采集与环境数据分析初学者及从业者的实战工具包解决从中国环境监测总站CNEMC稳定获取实时空气质量六参数PM2.5、PM10、SO₂、NO₂、CO、O₃的工程化需求。资源包含3371个文件主体为3354个按小时命名的CSV数据文件如2022-03-16T19.csv覆盖逐小时浓度均值与8小时滑动平均值等关键指标辅以11个Python脚本含爬取、清洗、校验逻辑、3个Jupyter Notebook含可视化与延时说明分析、1份README.md文档及配置文件结构清晰、开箱即用。压缩包大小为407.68MB已累计被1941人学习下载。读者可直接复用数据采集框架、理解CNEMC数据发布机制如半小时延滞、小时均值计算规则、验证字段完整性并基于真实历史数据开展趋势分析、模型训练或教学演示。 用Python抓取中国环境总站CNEMC的实时空气质量数据这件事听起来很简单网上随手一搜就是一堆教程。但真上手你就会发现老文章的接口早就失效了要么城市编码对不上要么返回一堆你看不懂的字段。我自己前前后后踩了不少坑干脆把一套能跑的完整方案整理出来从找接口、发请求、解析数据到落库一次讲清楚。这篇文章适合刚接触Python爬虫的人也适合想拿空气质量数据做分析、做可视化或者做毕业设计的同学。1. 动手之前先搞懂CNEMC的实时数据到底藏在哪1.1 从网页到接口数据是怎么来的中国环境监测总站CNEMC做过一个“全国城市空气质量实时发布平台”网页上能看到每个城市的实时AQI、PM2.5、PM10、二氧化硫、二氧化氮、臭氧、一氧化碳这些指标。页面上的数据不是凭空生成的它背后一定有一个接口在给前端喂数据。我们写Python去抓数据本质上是模拟浏览器去请求这个接口然后把返回的JSON解析出来。我早期犯过一个错误直接拿requests去请求网页本身结果发现页面是服务端渲染加动态加载的拉下来一堆HTML标签真正有用的数据全在XHR请求里。所以正确思路是先打开浏览器开发者工具F12切到Network面板勾选XHR然后刷新页面看哪些请求返回的是JSON。CNEMC平台的接口有个特点返回体经常是数组套对象的结构顶层就是列表里面每个元素对应一个城市或一个站点的监测数据字段名基本是英文缩写一眼能看出来。有人可能会问直接用selenium模拟浏览器行不行当然行但没必要。selenium要起浏览器、要装驱动、跑起来还吃内存放在定时任务里容易崩。我们这里只抓JSONrequests就够了轻量、可控、出错还好排查。1.2 一份实时数据的字段字典第一次拿到返回数据你可能有点懵满屏的“AQI”、“PM2.5”、“PrimaryPollutant”不知道哪些该留哪些该扔。我整理了一份常用字段说明在做数据清洗之前最好先把这个搞清楚。字段含义常见单位City城市名称字符串Station监测站点名称字符串AQI空气质量指数无量纲PM2.5细颗粒物μg/m³PM10可吸入颗粒物μg/m³SO2二氧化硫μg/m³NO2二氧化氮μg/m³O3臭氧μg/m³CO一氧化碳mg/m³Quality空气质量等级优/良/轻度污染等PrimaryPollutant首要污染物字符串TimePoint数据时间时间字符串注意CO的单位和其他不一样是mg/m³不是μg/m³换算的时候要小心。我见过有人把CO和PM2.5直接放在一张图里对比结果曲线完全没法看就是因为单位没归一。另外PrimaryPollutant如果接口返回的是“NA”或者“-”表示当前没有首要污染物这种情况通常对应空气质量为优。1.3 环境准备装好这三个库就够了这个项目不需要复杂的环境Python 3.8以上的版本就行。依赖库严格来说只有一个硬需求requests。如果后面要做数据处理和可视化再补pandas、matplotlib。当然如果你用的是Anaconda这几个库大概率都装好了。pip install requests pip install pandas pip install matplotlib为什么我不推荐一开始就上scrapy这样的重型框架因为单次抓取、单城市数据用requests是性价比最高的选择代码量少逻辑直观。scrapy适合大规模分布式采集对空气质量这种小体量数据属于杀鸡用牛刀。等你的需求真到了要并发抓几百个城市、还要做增量更新和数据去重的时候再考虑框架不迟。2. 核心链路用requests把实时数据拿回来2.1 抓包确认请求地址与参数接下来是整篇文章最关键的环节——找到真实接口。CNEMC这个平台的历史版本不少接口地址变过几次我这次实测用的是一套比较稳定的地址。打开空气质量发布平台页面在Network里找XHR请求正常情况下能看到一个包含“API”字样的请求点开它的Headers可以看到完整的请求URL和参数。我实测的地址是这样的http://106.37.208.243:8000/AirQuality/API/Now/GetAQIData请求参数里通常包含city和typecity是城市拼音小写比如beijing、shanghai、guangzhoutype是数据类型实时报一般是HOUR日报是DAY。如果你在页面上选择了一个城市接口会返回该城市当天的实时AQI数据。还有一个常用接口是获取城市内所有站点的数据http://106.37.208.243:8000/AirQuality/API/Now/CityStation/beijing这个接口返回的是北京各监测子站比如万寿西宫、奥体中心等的实时数据。要注意如果你在抓包时发现接口地址和我写的不完全一样别慌很可能是平台更新了版本。你只要把Network里最新的URL复制下来替换掉代码里的BASE_URL就行整体逻辑一样。2.2 请求头、时间参数与城市编码三个关键细节第一个细节是请求头。直接裸请求很可能被拒至少要带上User-Agent让他看起来像个正常浏览器。有些版本的接口还会校验Referer少了Referer会返回空数据或者直接403。我习惯在代码里定义一个HEADERS字典把常用的UA、Referer、Accept都写上一劳永逸。第二个细节是时间参数。部分接口支持按时间范围查询参数名可能是startTime、endTime格式是“yyyy-MM-dd HH:mm:ss”。实时数据接口一般不需要传时间默认返回最新时次但如果发现返回的数据是空数组你可以试着显式传一个当前时间比如用datetime模块生成from datetime import datetime now_str datetime.now().strftime(%Y-%m-%d %H:%M:%S)第三个细节是城市编码。我这套接口用的是城市拼音但有些别的平台用的是行政区划编码比如110000表示北京。如果你抓到的接口用的是编码手头又只有城市名单就得多做一步编码映射。我的建议是维护一个城市拼音字典或者直接从接口返回里提取城市列表先抓一次全量数据看看响应的City字段都长什么样再决定要不要走映射。2.3 写一个稳定的请求函数下面给出一段可以直接跑的示例代码我加了一些异常处理避免因为网络抖动或接口返回异常导致整个脚本崩掉。import requests from datetime import datetime import time HEADERS { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36, Referer: http://106.37.208.243:8000/, Accept: application/json, text/javascript, */*; q0.01, } BASE_URL http://106.37.208.243:8000/AirQuality/API/Now/GetAQIData def fetch_realtime_aqi(city: str beijing, retry: int 3): params {city: city, type: HOUR} for attempt in range(retry): try: resp requests.get(BASE_URL, paramsparams, headersHEADERS, timeout10) resp.raise_for_status() data resp.json() if not data: print(f[{datetime.now()}] {city} 返回空数据可能接口参数有误) return [] return data except requests.exceptions.Timeout: print(f[{datetime.now()}] {city} 请求超时第 {attempt 1} 次重试) time.sleep(2) except requests.exceptions.HTTPError as e: print(f[{datetime.now()}] {city} HTTP 错误: {e}) if attempt retry - 1: raise time.sleep(2) except requests.exceptions.JSONDecodeError: print(f[{datetime.now()}] {city} 返回内容不是合法 JSON) time.sleep(2) return []这里我用了一个retry参数默认重试3次每次失败后睡2秒。实测下来对于空气质量这种官方接口重试机制非常有用因为服务端偶尔会抽风但不至于长时间不可用。timeout设成10秒也是经验值太短容易误判超时太长会影响定时任务的整体节奏。3. 数据到手之后解析、清洗与落库实操3.1 JSON解析与异常值处理接口返回的JSON格式一般是列表套字典所以resp.json()拿到手以后直接遍历就行。但这里有个坑不同城市返回的字段可能不一样有的城市空气质量好PrimaryPollutant就会是“NA”有的站点可能某个污染物数据缺失返回“-”或者空字符串。我的处理方式是先把数据转成统一的字典结构再交给pandas处理。这里的关键在于对缺失值的统一判断我写了一个小函数把各种异常值全部转成None后续用pandas处理就会方便很多。import pandas as pd def normalize_value(value): if value in (NA, -, , None): return None try: return float(value) except (ValueError, TypeError): return None def parse_aqi_data(data): rows [] for item in data: rows.append({ city: item.get(City), time: item.get(TimePoint), aqi: normalize_value(item.get(AQI)), pm25: normalize_value(item.get(PM2.5)), pm10: normalize_value(item.get(PM10)), so2: normalize_value(item.get(SO2)), no2: normalize_value(item.get(NO2)), o3: normalize_value(item.get(O3)), co: normalize_value(item.get(CO)), quality: item.get(Quality), primary_pollutant: item.get(PrimaryPollutant), }) return pd.DataFrame(rows)把异常值统一成None是为了后面mean()、dropna()这些操作能正常发挥作用。真实数据里经常会有某个站点某小时数据缺席的情况如果你不做清洗直接画图折线图上会出现一个很扎眼的缺口不知道的人还以为是采集程序出Bug了。3.2 单位换算与AQI等级自动判定拿到字段以后如果只是展示那原始单位就够了。但如果你想做多城市对比或者把数据入库做长期分析建议先统一单位。CO这里特别容易搞混它在接口里通常是mg/m³如果要在和PM2.5的同一张图里展示乘1000转成μg/m³更合理。AQI等级也可以用代码自动判定防止接口没返回Quality字段时抓瞎。按照中国标准AQI分六级AQI区间等级0-50优51-100良101-150轻度污染151-200中度污染201-300重度污染大于300严重污染我在做数据可视化时经常需要自己写一个判定函数因为有时候接口返回的Quality字段是中文有时候又是英文规范不一致。自己定一套判别逻辑最简单也最可控def aqi_level(aqi): if aqi is None: return 无数据 aqi int(aqi) if aqi 50: return 优 elif aqi 100: return 良 elif aqi 150: return 轻度污染 elif aqi 200: return 中度污染 elif aqi 300: return 重度污染 else: return 严重污染3.3 增量写入CSV与数据库的思路数据量小的时候CSV是最省事的存储方案。我习惯用pandas的to_csv加上modea追加写入同时把encoding设成utf-8-sig这样用Excel打开也不会乱码。注意追加写入时不要每次写表头否则文件里会有一堆重复的列名。import os CSV_FILE aqi_history.csv def save_to_csv(df, file_pathCSV_FILE): if df.empty: return header not os.path.exists(file_path) or os.path.getsize(file_path) 0 df.to_csv(file_path, modea, headerheader, indexFalse, encodingutf-8-sig)如果你是做长期采集建议别只存CSV最好同步写一份到SQLite或者MySQL里方便按时间范围查询。SQLite不需要额外服务Python自带sqlite3模块几十行代码就能建表插入数据。存数据库的另一个好处是可以做去重避免定时任务重复抓取时产生大量重复记录。最简单的去重策略是拿citytimepoint做唯一键插入前先查一下有没有。4. 踩坑实录这些坑我替你先踩了4.1 访问被拒绝八成是UA和Referer在拦你最常遇到的现象是浏览器里打开接口地址好好的Python一请求就返回403或者返回空列表。我一开始也很纳闷后来把浏览器开发者工具里的请求头完整复制到代码里问题立刻消失。其实服务端主要就是看User-Agent和Referer。解决办法很简单把代码里的HEADERS补全Referer填平台首页地址就行。另外不要短时间内疯狂请求官方接口虽然没明确写限制但你几秒钟刷几百次被封IP也是早晚的事。我自己的经验是抓实时数据的话五分钟刷一次就够了。空气质量数据本身不是秒级变化你刷得再频繁拿到的东西也差不多反而增加被封风险。4.2 返回“暂无数据”怎么办有时候接口通了但返回的数组是空的。这个问题的原因五花八门我遇到过三种第一个是城市编码不对比如你传了citybeijing但接口实际期望的是“北京”或者反过来。我一般会先用接口文档或者抓包确认一下。第二个是type参数不对实时数据有时用“HOUR”有时用“REAL”版本不一样。第三个是时间参数问题如果传了startTime和endTime要确保时间格式和时区正确。官方平台用的是北京时间如果你的服务器是UTC时区记得加8小时否则查出来的永远是“未来”或者“过去”的数据。我的排查习惯是遇到空数组先手动把URL复制到浏览器里访问一下看看到底返回什么。浏览器里能看到数据说明接口和参数没问题问题出在代码的headers或者编码上浏览器里也是空数组那就要检查参数了。4.3 定时任务挂机采集的稳定性设计如果打算让脚本长期定时跑稳定性比功能本身更重要。我一共踩过两次比较大的坑。第一次是脚本跑了一个月突然某天接口返回的数据格式变了原来能解析的字段找不到了整个任务直接抛异常后面的几十次采集全部失败。后来我在解析函数外面套了一层try/except即便某次数据格式异常也能记录日志继续跑下去不至于全盘崩掉。第二次是网络问题服务器偶尔会丢包或者DNS解析变慢。解决方案就是前面代码里的重试机制多加一个指数退避。第一次失败等2秒第二次等4秒第三次等8秒最多重试三次再不行就用日志记录下来留给人工排查。用cron调度的时候注意把脚本的日志重定向到文件里*/5 * * * * cd /path/to/project python fetch_aqi.py logs/fetch.log 21这样即使出错你也能从日志里看到完整堆栈而不是深夜被一条“脚本挂了”的短信吵醒。4.4 常见问题速查表现象可能原因解决方案403 Forbidden缺少User-Agent或Referer补全请求头模拟浏览器访问返回空列表城市编码或type参数有误浏览器直接访问接口确认参数JSON解析异常接口返回了HTML错误页打印resp.text排查增加重试字段全是None字段名对不上先打印原始JSON确认字段名中文乱码编码设置不对设置resp.encodingutf-8采集一段时间后停止IP被限制或异常未捕获降低频率、增加异常捕获和日志追加写入重复表头to_csv每次都写header判断文件是否存在仅首次写表头这些坑其实都很小但每一个都足以让你在深夜怀疑人生。尤其是“浏览器能打开、Python不行”的这类问题本质上是你的请求头还不够像浏览器把Headers里的关键字段都带上就好。我在实际使用中还有一个习惯就是每次采集完之后顺手把数据按城市拆分存成独立文件这样后续做单城市分析的时候不用反复过滤。另外CNEMC的接口偶尔会有数据延迟刚刚整点的时候抓可能拿到的还是上一小时的数据这时候不用慌过几分钟再抓一次就正常了。如果你要做实时推送提醒建议在逻辑里加一个“数据时间”判断别拿旧数据当真数据。这套方案跑起来以后后面能玩的花样就多了。比如把历史数据积累一段时间做一个空气质量趋势分析看看哪些城市的季节性规律最明显也可以做个简单的定时提醒当某个城市的AQI超过阈值时自动给你发邮件。数据源本身是公开的采集门槛也不高真正值钱的是你拿这些数据做出的分析和判断。本文还有配套的精品资源点击获取