简介基于Visual Studio 2008开发的中国天气网数据获取工程源码专为.NET初学者和需要集成天气信息的桌面应用开发者准备。源码演示了从API调用、HTTP请求封装、响应读取到结果解析的完整处理链路覆盖实时天气、预报查询等常见需求同时包含API密钥配置管理和基本的异常捕获逻辑可直接编译运行或迁移到个人项目中二次开发。压缩包共34个文件包含9个C#源文件、sln与csproj工程配置、exe可执行程序、DLL依赖库、resx界面资源及资源文件等包体仅259KB结构紧凑适合按文件顺序逐段研读。已有2310人学习下载。通过阅读代码可重点掌握HttpWebRequest构造请求、HttpWebResponse接收响应、JavaScriptSerializer解析JSON、XDocument解析XML等关键写法也能熟悉VS2008中Windows窗体与后台逻辑的关联方式为后续对接其他Web API接口打下扎实基础是一份少而精的练手资料。1. 中国天气网获取天气数据不花钱不注册把官方数据抓回本地做天气类自动化的人都会遇到同一个问题数据从哪拿。早期我图省事直接接第三方天气 API注册、密钥、免费额度、字段限制折腾了一圈最后还是回到了中国天气网。中国天气网获取天气数据这条路最大的价值不是界面好看而是它有覆盖国内两千多个市县、无需注册就能访问的公开接口字段也贴合国内用户习惯。本文要把这条链路完整拆开城市代码怎么来、接口怎么选、JS 变量怎么解析、数据怎么落库、定时任务怎么稳定跑以及我踩过的几个坑。适合想自己搭天气采集、做定时提醒或本地可视化的开发者预算为零只需要一台能跑 Python 的机器。2. 先摸清中国天气网的数据出口三个常用接口与城市代码规则中国天气网没有一页一页写好的开放平台文档数据的真实出口都藏在网页背后。直接去抓 HTML 页面是下策页面结构改一次你的正则就废一次正确做法是先用浏览器开发者工具定位到真正返回数据的接口再决定用哪个。2.1 网页和接口是两回事开发者工具里看真实请求打开中国天气网首页按 F12 进入 Network 面板刷新页面然后筛选 XHR 或 Fetch 请求。你会发现浏览器自己请求了一堆.html后缀的地址响应体却根本不是 HTML而是以var开头的 JS 变量。这就是中国天气网接口的一个特点真正的数据不是 JSON是 JS 变量赋值代码。以北京为例页面加载后会看到一个形如http://d1.weather.com.cn/weather_index/101010100.html?_17...的请求。点开响应能看到var cityDZ 北京;、var dataSK {...};这样的内容。我们做采集目标就是这些.html接口而不是网页正文。用开发者工具定位接口已经是老生常谈了但我想强调一个习惯把 Network 面板里筛选条件保留为d1.weather.com.cn或data/cityinfo。中国天气网页面会附带很多统计类、广告类请求不按域名过滤会看得头皮发麻。看到目标请求之后右键复制为 cURL里面带了浏览器自动补全的Referer、User-Agent等信息后面写 Python 请求头可以直接照抄。还有一个容易被忽略的点?_17...这一串看起来像路径的一部分其实是个时间戳参数作用类似于时间戳防缓存。带上它可以避免命中中间层缓存导致拿到好几小时前的旧数据。后面写脚本时我会保留这个参数。2.2 三个常用接口城市列表、老版天气、新版实时根据我日常抓取的经验中国天气网最常用的接口分三类覆盖“找城市 → 拿基础天气 → 拿实时详细数据”三个环节。第一类是城市代码接口域名是www.weather.com.cn/data/city3jdata/返回标准 JSON。省份、市级、区县三层各有一个地址递归请求就能拼出全国城市清单。第二类是偏老的天气接口形如www.weather.com.cn/data/cityinfo/{cityid}.html返回 JSON里面带城市名、今天的高低温、天气现象和发布时间。第三类是新版实时接口形如d1.weather.com.cn/weather_index/{cityid}.html返回 JS 变量里面既有实时温度、风向、湿度也有生活指数和一段预报数据信息量比老接口大很多。三者不是简单的替换关系。老接口简单、返回干净适合快速验证接口通不通新接口信息量大是正式采集的首选城市代码接口则是地基。我一般这样选跑通流程时用老接口因为一眼能看懂返回正式建数据管线时用新版接口因为字段更全、覆盖面更广。接口用途URL 形态返回格式请求头要求省份与城市列表data/city3jdata/china.html等JSON无明显要求老版城市天气data/cityinfo/101010100.htmlJSON无明显要求新版实时天气d1.weather.com.cn/weather_index/101010100.htmlJS 变量需带 Referer 与 User-Agent需要特别说明的是新版接口的请求头里必须带Referer: http://www.weather.com.cn/否则大概率返回 403。这个 Referer 就像是一句“我从官网页面过来的”是这套接口最低限度的身份声明。我在踩坑阶段因为漏掉它连续被挡了好几次当时的报错信息又只有一串状态码排查起来特别憋屈。2.3 城市代码的层级规则省、市、区县怎么拼城市 ID 是整套采集流程的地基。中国天气网的城市 ID 是 9 位数字开头三位固定是101后面每一位对应行政区划层级。拿101010100举例前三位101表示中国接着01是北京再往后01加00是北京城区。把规则理解透就不需要人工去记几千个 ID而是按“省 → 市 → 区县”的层级自动生成。代码层面常见做法是直接用官方接口一层层取先请求china.html拿到所有省份代码再请求provshi/{prov}.html拿到省下面的城市或区县代码。这里有一个很典型的坑直辖市和省辖市会跳级省下面直接就是区县没有“市”这一层。所以不要把“三层结构”写死在代码里否则北京、上海这类城市会直接抓空。另一个细节是接口返回的 JSON 里键和值都是中文环境下直接可见的字符串比如{10101: 北京}。抓回来之后建议立刻存成一份本地 JSON 文件后续定时任务直接从本地读不要每次跑都重新打城市列表接口。城市 ID 又不会天天变省一次请求还能降低被限流的概率。3. 用 Python 跑通最小采集链路城市代码、请求头、JS 解析与落库这章进入正题我会把一个最小可运行的采集链路写出来再逐段解释为什么这么写。环境只需要 Python 3.8 以上加一个requests库。下面的代码不是一次性玩具而是我平时用的简化版本每个部分都能复用。3.1 抓城市代码表从 china.html 递归到区县第一步是拿到城市清单。我会用requests.Session建一个长连接会话统一设置User-Agent然后写三个函数分别请求省份、市、区县三个层级。用 Session 的原因很实际接口虽然免费但频繁建立新连接对服务端和自己都没有好处连接复用是基本礼貌也能减少被断连的概率。import requests session requests.Session() session.headers.update({ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) }) base http://www.weather.com.cn/data/city3jdata def get_provinces(): return session.get(f{base}/china.html).json() def get_cities(prov_code): return session.get(f{base}/provshi/{prov_code}.html).json() def get_districts(city_code): return session.get(f{base}/city/{city_code}.html).json()逻辑说明get_provinces()返回的字典形如{10101: 北京}键是省份代码值是中文省名。get_cities()接收省份代码返回该省下的城市对于直辖市这一层返回的可能直接是区县。get_districts()接收一个完整城市代码返回下一级区县。参数说明prov_code是省份代码形如10101city_code是城市或区县代码形如101010100。如果某个城市没有下一级get_districts()会请求一个不存在的地址返回的不是 JSONjson()就会抛异常这是正常的需要捕捉。递归遍历时我的做法是对每一层做异常兜底取不到下层就用当前层代码兜底def walk_all(): for prov_code, prov_name in get_provinces().items(): cities get_cities(prov_code) for city_code, city_name in cities.items(): try: dists get_districts(city_code) for dist_code, dist_name in dists.items(): yield prov_name, city_name, dist_name, dist_code except (requests.RequestException, ValueError, KeyError): yield prov_name, city_name, city_name, city_code代码里yield是生成器写法每产出一条就是一行“省份、城市、区县、区县代码”的记录。ValueError对应json()解析失败KeyError应对返回内容结构变化。如果你只关心地级市可以在dist_code分支里直接跳过或者在兜底位置用city_code作为最终采集单位。跑完整个递归全国城市大概几千条。我一般会把结果写进citys.json后续所有脚本从本地文件读取避免每次启动都对城市列表接口发起几十次请求。3.2 实时天气请求与解析带 Referer 的 JS 变量怎么拆拿到城市 ID 之后就可以请求新版实时接口了。这个接口返回的不是 JSON而是 JS 变量赋值代码里面有一个var dataSK {...};实时天气就藏在这个对象里。我处理这类“黑匣子”响应时习惯用正则先把大括号内容抠出来再交给json.loads解析。import re import json import time import requests city_id 101010100 ts int(time.time() * 1000) url fhttp://d1.weather.com.cn/weather_index/{city_id}.html?_{ts} headers { Referer: http://www.weather.com.cn/, User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64), } resp requests.get(url, headersheaders, timeout5) text resp.content.decode(utf-8, errorsignore) match re.search(rvar dataSK (\{.*?\});, text) if not match: raise RuntimeError(fno dataSK found: {text[:200]}) data_sk json.loads(match.group(1)) print(data_sk)这段代码重点有两个。第一是headers里的Referer没有它基本会得到 403这是这套接口最明显的门槛。第二是resp.content.decode(utf-8, errorsignore)先拿字节再自己指定编码不依赖requests的编码猜测避免部分网络环境下中文乱码。正则部分\{.*?\}里的.*?是非贪婪匹配遇到第一个};就会停下来。这很重要因为dataSK后面往往紧跟着dataZS等更大段的 JS 变量贪婪匹配会把后面一大段内容全部吞进去导致json.loads直接失败。如果正则匹配不到说明接口结构可能变了我把响应前 200 个字符打印出来方便判断是变量改名还是被限流。这里还要说明dataSK里常见的字段temp是实时温度WD是风向WS是风力等级SD是相对湿度time是观测时间cityname是城市名。不同时期返回的字段会有细微出入所以解析时我建议全部用.get()而不是下标访问缺字段时给默认值不要因为一个字段不存在就让整个任务中断。3.3 存成 CSV 还是 SQLite我的字段映射方案数据解析完只是临时变量下一步是落盘。如果只是跑几天的试验CSV 够用要长期积累做趋势分析SQLite 更稳。我的习惯是先用 CSV 验证字段没问题再切 SQLite。先看 CSV 版本。核心是控制好表头只写一次避免重复跑时出现多行列名。import os import csv from datetime import datetime row { city_id: city_id, city_name: data_sk.get(cityname, ), temp: data_sk.get(temp, ), wind_dir: data_sk.get(WD, ), wind_level: data_sk.get(WS, ), humidity: data_sk.get(SD, ), observed_at: data_sk.get(time, ), fetched_at: datetime.now().isoformat(timespecseconds), } csv_path weather.csv need_header not os.path.exists(csv_path) or os.path.getsize(csv_path) 0 with open(csv_path, a, newline, encodingutf-8) as f: writer csv.DictWriter(f, fieldnameslist(row.keys())) if need_header: writer.writeheader() writer.writerow(row)逻辑说明row字典里的键就是最终字段data_sk.get(cityname, )表示取不到就填空字符串。need_header判断文件是否存在或为空只有两种情况才写表头否则DictWriter会把字典的键作为表头重复写入后续用 pandas 读的时候第一行全是列名非常难受。参数说明newline是 Python 写 CSV 的标准写法避免空行encodingutf-8保证中文能正常写入timespecseconds让时间戳不带毫秒看起来清爽一点。如果你打算积累几个月的数据建议直接用 SQLite。建表和写入语句放到第 5 章讲因为那里要顺便讲清楚主键和幂等设计放在一起更顺。4. 避坑指南中国天气网数据采集的 5 个高频坑与排查方法这套接口的反爬强度其实不算硬真正的坑都藏在细节里。下面这几条是我实际采集过程中翻过车的地方按“现象 → 原因 → 解决”写清楚希望能帮你节省排查时间。4.1 现象请求返回 403 或“系统繁忙”现象脚本第一次跑得好好的过了一阵突然连续 403或者返回一段 HTML 页面里面写着“系统繁忙”。原因新版接口要求请求头里带Referer: http://www.weather.com.cn/。另外它对访问频率很敏感短时间内反复请求同一个城市 ID 会触发限流某些云服务器出口 IP 也会被误伤。解决先把Referer和User-Agent补全。然后给每次请求之间加至少 2 秒的间隔不要 for 循环里无脑连续打。还要用requests.Session保持连接减少握手次数。实在不行就在错误日志里记下状态码和响应前 200 个字符看到底是限流还是被防火墙拦了。4.2 现象中文乱码或出现大量空字符现象解析出来的城市名、天气现象变成å—之类的乱码或者某些字段变成空字符串。原因部分老接口在部分网络环境下返回的是 GBK 编码而requests.text会按响应头猜测编码猜错就乱码。新版接口也有概率在中间层缓存里被转成旧编码虽然概率低但遇到一次就够头疼。解决统一用resp.content.decode(utf-8, errorsignore)这种方式先拿字节再自己指定编码不依赖resp.text的自动猜测。如果发现老接口乱码改成resp.content.decode(gbk, errorsignore)再试。判断标准很简单能解析出cityname就是编码对了解析出中文偏旁就是编码没对上。4.3 现象夜间或偏远站点取到的字段是空的现象城市的温度、湿度都有值但风向是空或者整个dataSK里只有城市名其他字段全缺。原因中国天气网的实时观测依赖各地气象站点数据小城市、夜间某些时段、站点维护时上游可能根本没发布数据。这不是脚本问题是上游数据缺失。解决解析时用data_sk.get(WD, )这类写法字段缺失就给空字符串不让脚本崩溃。如果连temp都是空这轮数据直接跳过不落库并记一条警告日志。后续做历史分析时把空值占比算出来超过一定比例就换一个邻近城市 ID别硬等。4.4 现象定时任务跑了一段时间后突然失效现象脚本单独运行时正常放进 cron 后执行结果永远是空或者返回的dataSK一直是同一个时间戳。原因一种情况是接口把变量名改了或加了一层包装正则var dataSK匹配不上了另一种是本地网络环境或中间缓存命中拿到的永远是旧内容。解决给脚本加一个debug开关当正则匹配不到时把text[:300]写进日志。接口结构变了就改匹配模式例如把正则应宽松一些写成rvar\sdataSK\s*\s*(\{.*?\})\s*;覆盖变量名后有空格或换行的情况。时间戳参数?_建议每次都重新生成不要写死否则很容易命中缓存。4.5 现象明明设置 30 分钟跑一次实际一天只跑了一次现象cron 日志显示命令执行了但采集数据的时间间隔远大于预期甚至一天只有一两条记录。原因很多情况是系统休眠、网络唤醒延迟或 Python 环境问题。笔记本插电休眠后 cron 不会准时唤醒另外python3命令在不同的 PATH 下可能指向不同版本缺了requests库直接抛错但 cron 的日志默认只发邮件你不看邮件就完全没感知。解决在 cron 里写死 Python 的绝对路径比如/usr/bin/python3不要用裸python3。脚本开头做一次简单网络探测连不上就 sleep 30 秒再重试。把系统休眠策略改成接电源时不休眠。这类问题最难查因为表面看没有任何报错只是不干活。5. 把单次脚本变成数据管线定时调度、幂等写入与指数退避重试单次脚本能跑通只是第一步让它长期稳定跑才是重点。这一章的思路是定时触发、写入幂等、失败重试、数据自检四个环节串在一起才算一条合格的数据管线。5.1 定时调度cron 与 APScheduler 两种方式最简单的定时方式是系统 cron。以 30 分钟一次的频率为例在 crontab 里写*/30 * * * * cd /opt/weather /usr/bin/python3 fetch_weather.py run.log 21逻辑说明cd /opt/weather先切到项目目录确保脚本里的相对路径稳定/usr/bin/python3是绝对路径避免 cron 环境 PATH 不同导致找不到 Python run.log 21把标准输出和错误输出都追加到日志文件不然脚本报错时你连原因都看不到。如果你不想依赖系统 crontab也可以用 APScheduler 在 Python 里自管调度from apscheduler.schedulers.blocking import BlockingScheduler def run_once(): # 这里调用你自己的采集主函数 pass scheduler BlockingScheduler() scheduler.add_job(run_once, interval, minutes30, jitter60) scheduler.start()参数说明minutes30表示每 30 分钟执行一次jitter60是随机抖动 60 秒让任务不会每次都在整点整分触发也避免多台机器同时打接口造成瞬时压力。BlockingScheduler会占住一个进程适合跑在专门的采集机上。两种方式的区别在于cron 依赖操作系统进程挂了没人管APScheduler 至少能把异常抛到日志里配合后续的重试逻辑更容易定位问题。5.2 幂等写入主键设对重复跑不出脏数据定时任务跑久了最怕同一个observed_at被写入两遍导致后续分析时同一个时间点出现多条记录。解决方案是让写入操作幂等同一城市、同一观测时间重复写入只覆盖旧值不新增。SQLite 里用INSERT OR REPLACE实现最省事。import sqlite3 conn sqlite3.connect(weather.db) conn.execute( CREATE TABLE IF NOT EXISTS weather_obs ( city_id TEXT, city_name TEXT, temp TEXT, wind_dir TEXT, wind_level TEXT, humidity TEXT, observed_at TEXT, fetched_at TEXT, PRIMARY KEY (city_id, observed_at) ) ) conn.execute( INSERT OR REPLACE INTO weather_obs VALUES (?, ?, ?, ?, ?, ?, ?, ?), (row[city_id], row[city_name], row[temp], row[wind_dir], row[wind_level], row[humidity], row[observed_at], row[fetched_at]) ) conn.commit()逻辑说明CREATE TABLE IF NOT EXISTS保证表存在重复执行不会报错。PRIMARY KEY (city_id, observed_at)是这里最关键的设计只要是同一个城市、同一个观测时间写入就只更新那一行。fetched_at会跟着更新刚好能用来判断这个时刻的数据是几点抓到的。场景说明当你手动补跑、网络重试、或者改了脚本重复执行同一批城市时幂等写入能避免所有脏数据问题。不夸张地说这一个主键设置省掉了我后续大量的清洗时间。5.3 失败重试指数退避与错误日志网络请求总有抖动重试要写但别写成死循环。常见做法是指数退避第一次失败等 1 秒第二次等 3 秒第三次等 7 秒最多重试三次就放弃把这次失败记进日志下一轮定时任务自然会再试。import time import requests def fetch_weather_index(city_id, max_retries3): url fhttp://d1.weather.com.cn/weather_index/{city_id}.html headers { Referer: http://www.weather.com.cn/, User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64), } for attempt in range(max_retries): try: resp requests.get(url, headersheaders, timeout5) if resp.status_code 200: return resp print(fretry {attempt}: status{resp.status_code}) except requests.RequestException as e: print(fretry {attempt}: {e}) time.sleep(2 ** attempt 1) return None参数说明2 ** attempt 1在 attempt 为 0、1、2 时分别生成 2、3、5 秒的等待时间既不会立刻重试也不会把重试时间拉太长。timeout5防止某个城市接口卡住整个任务。日志里把status_code或异常信息打出来后面排查才能有据可查靠猜的话基本是浪费时间。5.4 数据健康检查一个简单的字段波动校验定时任务跑起来之后最怕的不是报错而是接口悄悄返回错数据。每天采集的数据量一大肉眼根本看不过来。我一般会在写入前加一道基础校验温度必须在 -60 到 60 之间湿度在 0 到 100 之间观测时间必须是最近 2 小时以内。不满足就跳过并单独写进warn.log。def is_row_ok(row): try: temp float(row[temp]) hum float(row[humidity].rstrip(%)) except ValueError: return False if temp -60 or temp 60: return False if hum 0 or hum 100: return False return True逻辑说明humidity字段带%符号先rstrip(%)再转float转不了就是异常数据。温度范围和湿度范围是常识阈值能挡住大部分意外错误。观测时间校验我建议对比row[observed_at]和当前时间超过 2 小时说明接口可能返回了过期缓存这种数据留着只会污染历史记录。这一节和第 3 章形成闭环数据采集到最后不是写进库里就完事而是让它变成一个能自检、能报警的小系统。有了这道关卡定时任务才算真正能脱离人工盯梢。6. 进阶用法把天气数据接进自己的提醒与可视化数据管线稳定之后下一步就是让数据产生实际价值。我见过最实用的两个方向一个是推送提醒一个是本地可视化。推送可以用企业微信机器人或钉钉机器人原理一致都是往 webhook 地址 POST 一段 JSON本地可视化可以用 SQLite 加一个几十行代码的 Flask 页面跑在局域网里手机也能看。先说推送。每天早晚各一次把当前温度和天气现象拼成一句话发到手机比打开天气 App 翻半天更快。核心代码就三行webhook_url https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key你的key payload {msgtype: text, text: {content: f{row[city_name]} 当前 {row[temp]}℃{row[wind_dir]}{row[wind_level]}}} requests.post(webhook_url, jsonpayload, timeout5)这类通知服务都需要你在对应后台创建机器人拿到专属 webhook 地址不同平台对 JSON 格式的要求略有差异但思路一致。我自己的习惯是把推送频率限制在早晚各一次避免用力过猛造成打扰。另一个习惯是给可视化页面加一层简单缓存每天早上从 SQLite 读一次当天数据生成一张温度曲线图页面访问时直接返回图片或 JSON不实时查库。说实话中国天气网的接口并不是最好看的字段结构偶尔也会让你想摔键盘免费的东西确实有这样的代价。但用久了你会发现它覆盖广、稳定性高改动频率也远低于那些价格一天一个样的商业天气 API。只要把城市代码、字段兜底和幂等入库这三件事做好这个方向就能长期跑下去甚至能积累出属于自己的一份气象历史数据。希望这些实践和踩坑笔记能帮到你少走一点我走过的弯路。本文还有配套的精品资源点击获取 
长河君 · 2026/10/25 10:35:50 · 阅读 1.2k 中国天气网数据采集实战：免费接口解析与Python自动化抓取
简介基于Visual Studio 2008开发的中国天气网数据获取工程源码专为.NET初学者和需要集成天气信息的桌面应用开发者准备。源码演示了从API调用、HTTP请求封装、响应读取到结果解析的完整处理链路覆盖实时天气、预报查询等常见需求同时包含API密钥配置管理和基本的异常捕获逻辑可直接编译运行或迁移到个人项目中二次开发。压缩包共34个文件包含9个C#源文件、sln与csproj工程配置、exe可执行程序、DLL依赖库、resx界面资源及资源文件等包体仅259KB结构紧凑适合按文件顺序逐段研读。已有2310人学习下载。通过阅读代码可重点掌握HttpWebRequest构造请求、HttpWebResponse接收响应、JavaScriptSerializer解析JSON、XDocument解析XML等关键写法也能熟悉VS2008中Windows窗体与后台逻辑的关联方式为后续对接其他Web API接口打下扎实基础是一份少而精的练手资料。1. 中国天气网获取天气数据不花钱不注册把官方数据抓回本地做天气类自动化的人都会遇到同一个问题数据从哪拿。早期我图省事直接接第三方天气 API注册、密钥、免费额度、字段限制折腾了一圈最后还是回到了中国天气网。中国天气网获取天气数据这条路最大的价值不是界面好看而是它有覆盖国内两千多个市县、无需注册就能访问的公开接口字段也贴合国内用户习惯。本文要把这条链路完整拆开城市代码怎么来、接口怎么选、JS 变量怎么解析、数据怎么落库、定时任务怎么稳定跑以及我踩过的几个坑。适合想自己搭天气采集、做定时提醒或本地可视化的开发者预算为零只需要一台能跑 Python 的机器。2. 先摸清中国天气网的数据出口三个常用接口与城市代码规则中国天气网没有一页一页写好的开放平台文档数据的真实出口都藏在网页背后。直接去抓 HTML 页面是下策页面结构改一次你的正则就废一次正确做法是先用浏览器开发者工具定位到真正返回数据的接口再决定用哪个。2.1 网页和接口是两回事开发者工具里看真实请求打开中国天气网首页按 F12 进入 Network 面板刷新页面然后筛选 XHR 或 Fetch 请求。你会发现浏览器自己请求了一堆.html后缀的地址响应体却根本不是 HTML而是以var开头的 JS 变量。这就是中国天气网接口的一个特点真正的数据不是 JSON是 JS 变量赋值代码。以北京为例页面加载后会看到一个形如http://d1.weather.com.cn/weather_index/101010100.html?_17...的请求。点开响应能看到var cityDZ 北京;、var dataSK {...};这样的内容。我们做采集目标就是这些.html接口而不是网页正文。用开发者工具定位接口已经是老生常谈了但我想强调一个习惯把 Network 面板里筛选条件保留为d1.weather.com.cn或data/cityinfo。中国天气网页面会附带很多统计类、广告类请求不按域名过滤会看得头皮发麻。看到目标请求之后右键复制为 cURL里面带了浏览器自动补全的Referer、User-Agent等信息后面写 Python 请求头可以直接照抄。还有一个容易被忽略的点?_17...这一串看起来像路径的一部分其实是个时间戳参数作用类似于时间戳防缓存。带上它可以避免命中中间层缓存导致拿到好几小时前的旧数据。后面写脚本时我会保留这个参数。2.2 三个常用接口城市列表、老版天气、新版实时根据我日常抓取的经验中国天气网最常用的接口分三类覆盖“找城市 → 拿基础天气 → 拿实时详细数据”三个环节。第一类是城市代码接口域名是www.weather.com.cn/data/city3jdata/返回标准 JSON。省份、市级、区县三层各有一个地址递归请求就能拼出全国城市清单。第二类是偏老的天气接口形如www.weather.com.cn/data/cityinfo/{cityid}.html返回 JSON里面带城市名、今天的高低温、天气现象和发布时间。第三类是新版实时接口形如d1.weather.com.cn/weather_index/{cityid}.html返回 JS 变量里面既有实时温度、风向、湿度也有生活指数和一段预报数据信息量比老接口大很多。三者不是简单的替换关系。老接口简单、返回干净适合快速验证接口通不通新接口信息量大是正式采集的首选城市代码接口则是地基。我一般这样选跑通流程时用老接口因为一眼能看懂返回正式建数据管线时用新版接口因为字段更全、覆盖面更广。接口用途URL 形态返回格式请求头要求省份与城市列表data/city3jdata/china.html等JSON无明显要求老版城市天气data/cityinfo/101010100.htmlJSON无明显要求新版实时天气d1.weather.com.cn/weather_index/101010100.htmlJS 变量需带 Referer 与 User-Agent需要特别说明的是新版接口的请求头里必须带Referer: http://www.weather.com.cn/否则大概率返回 403。这个 Referer 就像是一句“我从官网页面过来的”是这套接口最低限度的身份声明。我在踩坑阶段因为漏掉它连续被挡了好几次当时的报错信息又只有一串状态码排查起来特别憋屈。2.3 城市代码的层级规则省、市、区县怎么拼城市 ID 是整套采集流程的地基。中国天气网的城市 ID 是 9 位数字开头三位固定是101后面每一位对应行政区划层级。拿101010100举例前三位101表示中国接着01是北京再往后01加00是北京城区。把规则理解透就不需要人工去记几千个 ID而是按“省 → 市 → 区县”的层级自动生成。代码层面常见做法是直接用官方接口一层层取先请求china.html拿到所有省份代码再请求provshi/{prov}.html拿到省下面的城市或区县代码。这里有一个很典型的坑直辖市和省辖市会跳级省下面直接就是区县没有“市”这一层。所以不要把“三层结构”写死在代码里否则北京、上海这类城市会直接抓空。另一个细节是接口返回的 JSON 里键和值都是中文环境下直接可见的字符串比如{10101: 北京}。抓回来之后建议立刻存成一份本地 JSON 文件后续定时任务直接从本地读不要每次跑都重新打城市列表接口。城市 ID 又不会天天变省一次请求还能降低被限流的概率。3. 用 Python 跑通最小采集链路城市代码、请求头、JS 解析与落库这章进入正题我会把一个最小可运行的采集链路写出来再逐段解释为什么这么写。环境只需要 Python 3.8 以上加一个requests库。下面的代码不是一次性玩具而是我平时用的简化版本每个部分都能复用。3.1 抓城市代码表从 china.html 递归到区县第一步是拿到城市清单。我会用requests.Session建一个长连接会话统一设置User-Agent然后写三个函数分别请求省份、市、区县三个层级。用 Session 的原因很实际接口虽然免费但频繁建立新连接对服务端和自己都没有好处连接复用是基本礼貌也能减少被断连的概率。import requests session requests.Session() session.headers.update({ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) }) base http://www.weather.com.cn/data/city3jdata def get_provinces(): return session.get(f{base}/china.html).json() def get_cities(prov_code): return session.get(f{base}/provshi/{prov_code}.html).json() def get_districts(city_code): return session.get(f{base}/city/{city_code}.html).json()逻辑说明get_provinces()返回的字典形如{10101: 北京}键是省份代码值是中文省名。get_cities()接收省份代码返回该省下的城市对于直辖市这一层返回的可能直接是区县。get_districts()接收一个完整城市代码返回下一级区县。参数说明prov_code是省份代码形如10101city_code是城市或区县代码形如101010100。如果某个城市没有下一级get_districts()会请求一个不存在的地址返回的不是 JSONjson()就会抛异常这是正常的需要捕捉。递归遍历时我的做法是对每一层做异常兜底取不到下层就用当前层代码兜底def walk_all(): for prov_code, prov_name in get_provinces().items(): cities get_cities(prov_code) for city_code, city_name in cities.items(): try: dists get_districts(city_code) for dist_code, dist_name in dists.items(): yield prov_name, city_name, dist_name, dist_code except (requests.RequestException, ValueError, KeyError): yield prov_name, city_name, city_name, city_code代码里yield是生成器写法每产出一条就是一行“省份、城市、区县、区县代码”的记录。ValueError对应json()解析失败KeyError应对返回内容结构变化。如果你只关心地级市可以在dist_code分支里直接跳过或者在兜底位置用city_code作为最终采集单位。跑完整个递归全国城市大概几千条。我一般会把结果写进citys.json后续所有脚本从本地文件读取避免每次启动都对城市列表接口发起几十次请求。3.2 实时天气请求与解析带 Referer 的 JS 变量怎么拆拿到城市 ID 之后就可以请求新版实时接口了。这个接口返回的不是 JSON而是 JS 变量赋值代码里面有一个var dataSK {...};实时天气就藏在这个对象里。我处理这类“黑匣子”响应时习惯用正则先把大括号内容抠出来再交给json.loads解析。import re import json import time import requests city_id 101010100 ts int(time.time() * 1000) url fhttp://d1.weather.com.cn/weather_index/{city_id}.html?_{ts} headers { Referer: http://www.weather.com.cn/, User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64), } resp requests.get(url, headersheaders, timeout5) text resp.content.decode(utf-8, errorsignore) match re.search(rvar dataSK (\{.*?\});, text) if not match: raise RuntimeError(fno dataSK found: {text[:200]}) data_sk json.loads(match.group(1)) print(data_sk)这段代码重点有两个。第一是headers里的Referer没有它基本会得到 403这是这套接口最明显的门槛。第二是resp.content.decode(utf-8, errorsignore)先拿字节再自己指定编码不依赖requests的编码猜测避免部分网络环境下中文乱码。正则部分\{.*?\}里的.*?是非贪婪匹配遇到第一个};就会停下来。这很重要因为dataSK后面往往紧跟着dataZS等更大段的 JS 变量贪婪匹配会把后面一大段内容全部吞进去导致json.loads直接失败。如果正则匹配不到说明接口结构可能变了我把响应前 200 个字符打印出来方便判断是变量改名还是被限流。这里还要说明dataSK里常见的字段temp是实时温度WD是风向WS是风力等级SD是相对湿度time是观测时间cityname是城市名。不同时期返回的字段会有细微出入所以解析时我建议全部用.get()而不是下标访问缺字段时给默认值不要因为一个字段不存在就让整个任务中断。3.3 存成 CSV 还是 SQLite我的字段映射方案数据解析完只是临时变量下一步是落盘。如果只是跑几天的试验CSV 够用要长期积累做趋势分析SQLite 更稳。我的习惯是先用 CSV 验证字段没问题再切 SQLite。先看 CSV 版本。核心是控制好表头只写一次避免重复跑时出现多行列名。import os import csv from datetime import datetime row { city_id: city_id, city_name: data_sk.get(cityname, ), temp: data_sk.get(temp, ), wind_dir: data_sk.get(WD, ), wind_level: data_sk.get(WS, ), humidity: data_sk.get(SD, ), observed_at: data_sk.get(time, ), fetched_at: datetime.now().isoformat(timespecseconds), } csv_path weather.csv need_header not os.path.exists(csv_path) or os.path.getsize(csv_path) 0 with open(csv_path, a, newline, encodingutf-8) as f: writer csv.DictWriter(f, fieldnameslist(row.keys())) if need_header: writer.writeheader() writer.writerow(row)逻辑说明row字典里的键就是最终字段data_sk.get(cityname, )表示取不到就填空字符串。need_header判断文件是否存在或为空只有两种情况才写表头否则DictWriter会把字典的键作为表头重复写入后续用 pandas 读的时候第一行全是列名非常难受。参数说明newline是 Python 写 CSV 的标准写法避免空行encodingutf-8保证中文能正常写入timespecseconds让时间戳不带毫秒看起来清爽一点。如果你打算积累几个月的数据建议直接用 SQLite。建表和写入语句放到第 5 章讲因为那里要顺便讲清楚主键和幂等设计放在一起更顺。4. 避坑指南中国天气网数据采集的 5 个高频坑与排查方法这套接口的反爬强度其实不算硬真正的坑都藏在细节里。下面这几条是我实际采集过程中翻过车的地方按“现象 → 原因 → 解决”写清楚希望能帮你节省排查时间。4.1 现象请求返回 403 或“系统繁忙”现象脚本第一次跑得好好的过了一阵突然连续 403或者返回一段 HTML 页面里面写着“系统繁忙”。原因新版接口要求请求头里带Referer: http://www.weather.com.cn/。另外它对访问频率很敏感短时间内反复请求同一个城市 ID 会触发限流某些云服务器出口 IP 也会被误伤。解决先把Referer和User-Agent补全。然后给每次请求之间加至少 2 秒的间隔不要 for 循环里无脑连续打。还要用requests.Session保持连接减少握手次数。实在不行就在错误日志里记下状态码和响应前 200 个字符看到底是限流还是被防火墙拦了。4.2 现象中文乱码或出现大量空字符现象解析出来的城市名、天气现象变成å—之类的乱码或者某些字段变成空字符串。原因部分老接口在部分网络环境下返回的是 GBK 编码而requests.text会按响应头猜测编码猜错就乱码。新版接口也有概率在中间层缓存里被转成旧编码虽然概率低但遇到一次就够头疼。解决统一用resp.content.decode(utf-8, errorsignore)这种方式先拿字节再自己指定编码不依赖resp.text的自动猜测。如果发现老接口乱码改成resp.content.decode(gbk, errorsignore)再试。判断标准很简单能解析出cityname就是编码对了解析出中文偏旁就是编码没对上。4.3 现象夜间或偏远站点取到的字段是空的现象城市的温度、湿度都有值但风向是空或者整个dataSK里只有城市名其他字段全缺。原因中国天气网的实时观测依赖各地气象站点数据小城市、夜间某些时段、站点维护时上游可能根本没发布数据。这不是脚本问题是上游数据缺失。解决解析时用data_sk.get(WD, )这类写法字段缺失就给空字符串不让脚本崩溃。如果连temp都是空这轮数据直接跳过不落库并记一条警告日志。后续做历史分析时把空值占比算出来超过一定比例就换一个邻近城市 ID别硬等。4.4 现象定时任务跑了一段时间后突然失效现象脚本单独运行时正常放进 cron 后执行结果永远是空或者返回的dataSK一直是同一个时间戳。原因一种情况是接口把变量名改了或加了一层包装正则var dataSK匹配不上了另一种是本地网络环境或中间缓存命中拿到的永远是旧内容。解决给脚本加一个debug开关当正则匹配不到时把text[:300]写进日志。接口结构变了就改匹配模式例如把正则应宽松一些写成rvar\sdataSK\s*\s*(\{.*?\})\s*;覆盖变量名后有空格或换行的情况。时间戳参数?_建议每次都重新生成不要写死否则很容易命中缓存。4.5 现象明明设置 30 分钟跑一次实际一天只跑了一次现象cron 日志显示命令执行了但采集数据的时间间隔远大于预期甚至一天只有一两条记录。原因很多情况是系统休眠、网络唤醒延迟或 Python 环境问题。笔记本插电休眠后 cron 不会准时唤醒另外python3命令在不同的 PATH 下可能指向不同版本缺了requests库直接抛错但 cron 的日志默认只发邮件你不看邮件就完全没感知。解决在 cron 里写死 Python 的绝对路径比如/usr/bin/python3不要用裸python3。脚本开头做一次简单网络探测连不上就 sleep 30 秒再重试。把系统休眠策略改成接电源时不休眠。这类问题最难查因为表面看没有任何报错只是不干活。5. 把单次脚本变成数据管线定时调度、幂等写入与指数退避重试单次脚本能跑通只是第一步让它长期稳定跑才是重点。这一章的思路是定时触发、写入幂等、失败重试、数据自检四个环节串在一起才算一条合格的数据管线。5.1 定时调度cron 与 APScheduler 两种方式最简单的定时方式是系统 cron。以 30 分钟一次的频率为例在 crontab 里写*/30 * * * * cd /opt/weather /usr/bin/python3 fetch_weather.py run.log 21逻辑说明cd /opt/weather先切到项目目录确保脚本里的相对路径稳定/usr/bin/python3是绝对路径避免 cron 环境 PATH 不同导致找不到 Python run.log 21把标准输出和错误输出都追加到日志文件不然脚本报错时你连原因都看不到。如果你不想依赖系统 crontab也可以用 APScheduler 在 Python 里自管调度from apscheduler.schedulers.blocking import BlockingScheduler def run_once(): # 这里调用你自己的采集主函数 pass scheduler BlockingScheduler() scheduler.add_job(run_once, interval, minutes30, jitter60) scheduler.start()参数说明minutes30表示每 30 分钟执行一次jitter60是随机抖动 60 秒让任务不会每次都在整点整分触发也避免多台机器同时打接口造成瞬时压力。BlockingScheduler会占住一个进程适合跑在专门的采集机上。两种方式的区别在于cron 依赖操作系统进程挂了没人管APScheduler 至少能把异常抛到日志里配合后续的重试逻辑更容易定位问题。5.2 幂等写入主键设对重复跑不出脏数据定时任务跑久了最怕同一个observed_at被写入两遍导致后续分析时同一个时间点出现多条记录。解决方案是让写入操作幂等同一城市、同一观测时间重复写入只覆盖旧值不新增。SQLite 里用INSERT OR REPLACE实现最省事。import sqlite3 conn sqlite3.connect(weather.db) conn.execute( CREATE TABLE IF NOT EXISTS weather_obs ( city_id TEXT, city_name TEXT, temp TEXT, wind_dir TEXT, wind_level TEXT, humidity TEXT, observed_at TEXT, fetched_at TEXT, PRIMARY KEY (city_id, observed_at) ) ) conn.execute( INSERT OR REPLACE INTO weather_obs VALUES (?, ?, ?, ?, ?, ?, ?, ?), (row[city_id], row[city_name], row[temp], row[wind_dir], row[wind_level], row[humidity], row[observed_at], row[fetched_at]) ) conn.commit()逻辑说明CREATE TABLE IF NOT EXISTS保证表存在重复执行不会报错。PRIMARY KEY (city_id, observed_at)是这里最关键的设计只要是同一个城市、同一个观测时间写入就只更新那一行。fetched_at会跟着更新刚好能用来判断这个时刻的数据是几点抓到的。场景说明当你手动补跑、网络重试、或者改了脚本重复执行同一批城市时幂等写入能避免所有脏数据问题。不夸张地说这一个主键设置省掉了我后续大量的清洗时间。5.3 失败重试指数退避与错误日志网络请求总有抖动重试要写但别写成死循环。常见做法是指数退避第一次失败等 1 秒第二次等 3 秒第三次等 7 秒最多重试三次就放弃把这次失败记进日志下一轮定时任务自然会再试。import time import requests def fetch_weather_index(city_id, max_retries3): url fhttp://d1.weather.com.cn/weather_index/{city_id}.html headers { Referer: http://www.weather.com.cn/, User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64), } for attempt in range(max_retries): try: resp requests.get(url, headersheaders, timeout5) if resp.status_code 200: return resp print(fretry {attempt}: status{resp.status_code}) except requests.RequestException as e: print(fretry {attempt}: {e}) time.sleep(2 ** attempt 1) return None参数说明2 ** attempt 1在 attempt 为 0、1、2 时分别生成 2、3、5 秒的等待时间既不会立刻重试也不会把重试时间拉太长。timeout5防止某个城市接口卡住整个任务。日志里把status_code或异常信息打出来后面排查才能有据可查靠猜的话基本是浪费时间。5.4 数据健康检查一个简单的字段波动校验定时任务跑起来之后最怕的不是报错而是接口悄悄返回错数据。每天采集的数据量一大肉眼根本看不过来。我一般会在写入前加一道基础校验温度必须在 -60 到 60 之间湿度在 0 到 100 之间观测时间必须是最近 2 小时以内。不满足就跳过并单独写进warn.log。def is_row_ok(row): try: temp float(row[temp]) hum float(row[humidity].rstrip(%)) except ValueError: return False if temp -60 or temp 60: return False if hum 0 or hum 100: return False return True逻辑说明humidity字段带%符号先rstrip(%)再转float转不了就是异常数据。温度范围和湿度范围是常识阈值能挡住大部分意外错误。观测时间校验我建议对比row[observed_at]和当前时间超过 2 小时说明接口可能返回了过期缓存这种数据留着只会污染历史记录。这一节和第 3 章形成闭环数据采集到最后不是写进库里就完事而是让它变成一个能自检、能报警的小系统。有了这道关卡定时任务才算真正能脱离人工盯梢。6. 进阶用法把天气数据接进自己的提醒与可视化数据管线稳定之后下一步就是让数据产生实际价值。我见过最实用的两个方向一个是推送提醒一个是本地可视化。推送可以用企业微信机器人或钉钉机器人原理一致都是往 webhook 地址 POST 一段 JSON本地可视化可以用 SQLite 加一个几十行代码的 Flask 页面跑在局域网里手机也能看。先说推送。每天早晚各一次把当前温度和天气现象拼成一句话发到手机比打开天气 App 翻半天更快。核心代码就三行webhook_url https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key你的key payload {msgtype: text, text: {content: f{row[city_name]} 当前 {row[temp]}℃{row[wind_dir]}{row[wind_level]}}} requests.post(webhook_url, jsonpayload, timeout5)这类通知服务都需要你在对应后台创建机器人拿到专属 webhook 地址不同平台对 JSON 格式的要求略有差异但思路一致。我自己的习惯是把推送频率限制在早晚各一次避免用力过猛造成打扰。另一个习惯是给可视化页面加一层简单缓存每天早上从 SQLite 读一次当天数据生成一张温度曲线图页面访问时直接返回图片或 JSON不实时查库。说实话中国天气网的接口并不是最好看的字段结构偶尔也会让你想摔键盘免费的东西确实有这样的代价。但用久了你会发现它覆盖广、稳定性高改动频率也远低于那些价格一天一个样的商业天气 API。只要把城市代码、字段兜底和幂等入库这三件事做好这个方向就能长期跑下去甚至能积累出属于自己的一份气象历史数据。希望这些实践和踩坑笔记能帮到你少走一点我走过的弯路。本文还有配套的精品资源点击获取