简介：面向使用 Visual Studio 2008 开发气象信息应用的 C# 学习者，这套中国天气网天气数据获取示例围绕 API 调用、HTTP 请求与响应处理、JSON/XML 解析及窗体界面展示展开，适合需要参考完整实现流程的初中级开发者。压缩包共 34 个文件，以 cs 源码、config 配置、dll 与 exe 程序、resx 资源等为主，整体约 259KB，包含解决方案与 Windows Forms 工程结构，便于直接打开对照学习。已有 2314 人学习下载，可作为天气查询小工具的实践样本。读者可从中了解如何封装请求、注册并使用接口密钥、解析城市天气字段，并把温度、湿度、风向等信息呈现在文本框、标签或图表中，同时参考异常处理与 API 使用限制等排错思路，快速搭建可运行的天气数据获取原型。

1. 一个 9 位编码换一份天气数据，这笔账先算清楚

做运维大屏、农业墒情、物流时效预估，绕不开的一件事就是获取天气数据。商业气象接口按调用量计费，一个中小项目每天几千次请求，账单一出来就得重新评估方案；而公开气象站点本身就把实时温度、湿度、风向风速、多日预报挂在页面上，只是没给你一份带文档的 SDK。这份资源做的就是把这条路走通：给定一个 9 位城市编码，拼请求、处理编码、解析字段、落到库里，再用定时任务把它变成一条不断线的数据流。

它适合三类人：要做天气看板但没预算买商业接口的前后端；需要把天气当特征喂进模型的数据侧同学；想练手 HTTP 采集、字符编码处理、幂等入库的初学者。不适合的是对 SLA 有硬承诺的生产场景，公开站点的可用性、字段稳定性、更新时刻都没有任何承诺，这一点后面会反复提到。先想清楚你是「拿它做一个能自洽的看板」还是「拿它当交易系统的输入」，前者够用，后者必须再叠一层校验和第二数据源。

2. 三条链路先分清：实时观测、当日概要、多日预报各拿什么

同一套城市编码，能问出三种完全不同的数据。很多人上来就抓那个「看起来字段最全」的接口，结果发现里面没有逐时温度；或者只抓了多日预报，画不出一条今天下午的温度曲线。先把链路分层，后面所有代码和表结构才有依据。

2.1 接口分层与字段覆盖面

按更新频率和数据形态，公开气象站点一般提供四类内容，实际能用的主要是前三类：

选型上我一般这么定：做实时看板只要实时观测 + 多日预报两条；做冷热负荷预测需要逐小时；做农业积温统计反而要多日预报的历史快照，每天存一份，日积月累才有得算。别贪多，先把一条链路跑顺，再按需加第二条，因为限频和失败重试的压力是叠加的。

2.2 城市编码：9 位数字背后的层级和校验

这套体系里城市标识是 9 位纯数字，形如101010100。它不是随机号，前几位标的是站点类别和行政区归属，末两位区分同一城市下的不同区县。实操中三个结论最重要：编码必须当字符串处理，不要转 int；不同站点对同一城市的编码可能不完全一致；不是所有区县都有独立数据，末位换一位就可能拿不到返回。

所以编码不要硬编码在脚本里，维护一张city_code表，字段至少包含编码、中文名、上级编码、是否可用、最近一次成功时间。抓取失败时能回退到上级城市，而不是直接断档。

import re CITY_CODE_RE = re.compile(r"^\d{9}$") def normalize_city_code(raw: str) -> str: """把脏输入清洗成 9 位纯数字编码。 常见的三种脏数据： 1. 从网页地址栏复制来的 101010100.shtml 2. 带前导字母或空格的 ' cn101010100 ' 3. 被 Excel 当数字处理过、末尾补了 .0 的 '101010100.0' """ s = str(raw).strip() s = re.sub(r"\.(s?html?|0)$", "", s, flags=re.I) # 先剥后缀，再抽数字 s = re.sub(r"\D", "", s) # 只保留数字 if not CITY_CODE_RE.match(s): raise ValueError(f"城市编码不合法: {raw!r}") return s

顺序很关键：必须先剥后缀再抽数字，否则101010100.shtml抽完数字会变成 13 位。这里没有用int()转换，是因为一旦某类编码出现前导 0，int()会静默吃掉它，而这类错误在入库后才暴露，排查成本极高。校验函数只保证格式，一个编码是否真的可用，要靠一次真实请求来证明，不要靠编码规则去推断。

2.3 请求骨架：会话复用、超时与重试

单次requests.get能跑通，不代表能跑一天。默认没有连接复用，没有超时，失败不重试，这三件事在几十个城市 × 每小时一轮的场景下会立刻变成事故。所以第一版脚本就要把Session、超时元组、重试策略这三样定下来。

import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry BASE = "https://<气象站点接口前缀>" # 站点地址抽成常量，换站点只改这一行 TIMEOUT = (3.05, 10) # (连接超时, 读超时)，秒 def build_session() -> requests.Session: s = requests.Session() retry = Retry( total=3, backoff_factor=1.5, # 间隔 1.5s、3s、6s status_forcelist=(429, 500, 502, 503, 504), allowed_methods=frozenset(["GET"]), ) s.mount("https://", HTTPAdapter(max_retries=retry, pool_maxsize=10)) s.headers.update({ "User-Agent": "Mozilla/5.0 (compatible; weather-collector/1.0)", "Accept": "application/json,text/plain,*/*", "Accept-Encoding": "gzip, deflate", # 显式声明，避免服务端返回 br "Referer": BASE + "/", }) return s

TIMEOUT用元组而不是单个数字，是因为连接超时和读超时的合理值差一个量级：连不上通常几秒内就有结论，而服务端把请求挂住慢慢吐数据的情况，读超时给 10 秒才合理。backoff_factor=1.5让重试间隔递增，避免服务端抖动时你的重试把对方压死，也避免自己一直踩着同样的节拍撞墙。Accept-Encoding显式写死，是因为有些库会自动协商 Brotli，而环境里没装解压依赖时会拿到一串乱码字节，排查起来非常反直觉。

Referer这一项不是可选项。同一个接口，带不带它决定了返回完整数据还是 403 或空壳，这条放在第 5 章细说。

3. 把响应抠成结构体：字段解析、编码处理与幂等入库

拿到响应只是第一步，真正吃掉时间的是「这个字符串到底怎么变成一行能查的数据」。这一步翻车的概率远高于请求本身，因为错误是静默的：字段错位、单位错、时间错，页面上看起来一切正常，等你要做同比的时候才发现全是垃圾。

3.1 字段逐项对照与单位统一

公开气象接口的字段名普遍很短，两三个字母，含义靠约定。常见的一组对照如下，注意单位只写这个接口里的实际单位，不同链路对同一个字段的定义可能不一样。

风向和风力的处理最容易被轻视。风向返回「东南风」这类文字，做风玫瑰图必须转角度，转换表自己维护一份就好，北=0、东北=45、东=90，依次类推，出现「无持续风向」时角度记 NULL 而不是 0——记 0 会让均值统计严重偏北。风力返回「3级」时不能直接float()，先用正则抠数字；返回3.4这类小数的，要确认它到底是等级还是 m/s，两者差了一个量级。

时间字段只有时分是最阴的一条。批处理时如果直接把它和当前日期拼，跨零点那一轮就会出现「昨天 23:50」被记成「今天 23:50」，整个时间序列断开。稳妥做法是：用请求发起时刻做基准日期，若解析出的时刻与基准相差超过 6 小时，判为跨天，日期回退一天。

3.2 编码与外壳：GBK、gzip、var data =

响应体有两层外壳要拆。第一层是字符编码：这类站点的响应头经常声明 UTF-8，实际内容却是 GBK 系编码，直接resp.text得到一片问号。第二层是 JSON 外壳：有的接口返回标准 JSON，有的在前面挂一句var data =之类的赋值语句，交给json.loads第一个字符就报错。

import json, re def decode_body(resp) -> str: """按 UTF-8 -> GB18030 顺序试解码。 gb18030 是 gbk/gb2312 的超集，用它兜底能覆盖绝大多数中文乱码， 而且不需要依赖 charset 探测库（短文本探测准确率很低）。 """ raw = resp.content for enc in ("utf-8", "gb18030"): try: return raw.decode(enc) except UnicodeDecodeError: continue return raw.decode("utf-8", errors="replace") def strip_wrapper(text: str) -> str: """剥掉 var data = / callback(...) 之类的壳，截出第一个 { 到最后一个 }。""" start, end = text.find("{"), text.rfind("}") if start == -1 or end == -1 or end <= start: raise ValueError("响应里没有 JSON 主体，大概率是错误页或空壳") return text[start:end + 1]

先解码再截壳，顺序不能反。字节层面直接找{会撞上被 gzip 压缩过的内容，或撞上 GBK 双字节里的某个位；先转成字符串，截取才是安全的。strip_wrapper用find和rfind而不是正则，是因为正则贪婪匹配在嵌套花括号时容易吃错边界，而截第一个开括号到最后一个闭括号在单对象响应里永远正确。

外壳剥完还要加一层防御：json.loads报错时把原始文本落盘，文件名带城市编码和时间戳，保留最近几百个。这一步看着多余，但错误页、限频页、维护公告页长得都像 JSON 的开头，不存原文你根本不知道当时拿回了什么。

3.3 建表与 UPSERT

入库的核心诉求只有一条：同一条数据重复写多少遍，结果都一样。公开站点的观测值会在同一小时内被修正，重跑脚本也很常见，所以唯一键必须落在「城市 + 观测时刻」上，写入用INSERT ... ON DUPLICATE KEY UPDATE。

CREATE TABLE weather_obs ( id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT, city_code CHAR(9) NOT NULL, obs_time DATETIME NOT NULL, temp_c DECIMAL(4,1) NULL, humidity TINYINT UNSIGNED NULL, wind_dir VARCHAR(16) NULL, wind_deg SMALLINT UNSIGNED NULL, wind_speed_ms DECIMAL(4,1) NULL, pressure_hpa DECIMAL(6,1) NULL, visibility_km DECIMAL(4,1) NULL, raw_json JSON NULL, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (id), UNIQUE KEY uk_city_time (city_code, obs_time), KEY idx_time (obs_time) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; INSERT INTO weather_obs (city_code, obs_time, temp_c, humidity, wind_dir, wind_deg, wind_speed_ms, pressure_hpa, visibility_km, raw_json) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE temp_c = VALUES(temp_c), humidity = VALUES(humidity), wind_dir = VALUES(wind_dir), wind_deg = VALUES(wind_deg), wind_speed_ms = VALUES(wind_speed_ms), pressure_hpa = VALUES(pressure_hpa), visibility_km = VALUES(visibility_km), raw_json = VALUES(raw_json);

几个字段类型是有意这么定的。city_code用CHAR(9)而不是INT，因为编码有前导位语义，数字类型会丢前导 0；humidity用TINYINT UNSIGNED，0～100 刚好够，比DECIMAL省一半空间；raw_json单独留一列存原文，解析逻辑改版之后可以离线重跑，不用回头再抓一遍——这就是我的后悔药，抓过的数据永远别再抓第二次。

obs_time建普通索引而不是放进联合索引的尾部，是因为绝大多数查询是「某城市某时间段」，联合唯一键(city_code, obs_time)已经覆盖了单城市查询，但跨城市的时序报表需要单独的idx_time。索引不是越多越好，但这两个是查得出的。

4. 从单次脚本到常驻采集器：调度、限频与容错

脚本能跑通和能连跑一个月，中间隔着一整套工程化处理。这一章的东西不加，前面全白搭——因为数据断档往往不是「程序崩了」，而是「程序还活着，但已经三天没抓到东西了」。

4.1 调度节奏与抖动

调度只定一个原则：抓取频率由数据源的实际更新频率决定，不由你的焦虑决定。实时观测大约每小时更新一次，那就每 10～15 分钟抓一轮，目的是「尽量贴近更新时刻」，而不是「抓得越多越准」。抓得再密，值没变，你只是白送请求。

第二个原则是抖动。整齐划一的节拍器式请求，对服务端是最不友好的形态，也最容易在运维修正的时候被一起收走。

import time, random, logging from datetime import datetime def collect_round(session, city_codes, save): ok = fail = 0 for code in city_codes: try: text = fetch(session, "/obs", {"city": code}) record = to_record(code, parse_payload(text)) save(record) # 内部走 UPSERT ok += 1 except Exception as exc: fail += 1 logging.warning("collect_failed city=%s err=%r", code, exc) finally: # 城市之间随机间隔 0.8~2.0 秒，city 数量多时整体节奏自然摊开 time.sleep(random.uniform(0.8, 2.0)) logging.info("round_done ok=%d fail=%d at=%s", ok, fail, datetime.now().isoformat()) return ok, fail

finally里 sleep 是有意的：不管成功失败都等，否则一旦某个编码持续失败，重试会变成零间隔的连续打击，几秒钟内就能把自己的 IP 送上黑名单。round_done这条日志单独打印，是给后面的心跳检测留的锚点——没有这行，你连「上一轮是什么时候跑的」都答不上来。

4.2 失败分级与降级策略

失败要分三级处理，不能一把except全兜住然后 pass。

第一级是网络类，连接超时、读超时、5xx。这类交给Retry自动处理，外层只记录不告警，因为偶发一两次属于常态。

第二级是协议类，403、429、返回空壳或错误页。这类说明你和对方的关系出问题了，重试没用，要继续跑就必须降速或换策略，所以应该触发一个全局冷却：暂停本轮剩余城市，睡 5～10 分钟，下一轮减半频率。

第三级是解析类，JSON 报错、字段缺失、单位异常。这类最危险，因为程序不崩，日志里只有一行 warning，很容易被忽略。做法是把它和「上一轮成功时间」绑起来：连续 3 轮解析失败，就把该城市在city_code表里标成不可用，同时发一条告警。

降级的目标不是「永远有数据」，而是「永远知道数据是不是新鲜的」。我的做法是每次成功写入顺手更新一张collect_status表，记录城市编码、最后成功时间、连续失败次数。看板读数据之前先看一眼last_success，超过阈值就打上「数据延迟」的水印，比拿旧数据冒充实时数据诚实得多。

4.3 日志与数据质量闸门

日志按行输出结构化内容，字段固定为：城市编码、接口路径、状态码、耗时毫秒、重试次数、解析结果。不固定字段的日志在一周之后就是纯噪音，翻起来比重新抓一遍还累。

质量闸门放在入库前，用几条硬规则挡住明显异常的值：气温在 -60～60 之外直接丢；湿度不在 0～100 之间直接丢；风速大于 150 m/s 直接丢；obs_time早于上一轮成功时间的，判为重复观测，走 UPSERT 覆盖即可。这些阈值不是气象学结论，是脏数据过滤器，宁可漏掉一条真数据，也别让一条离谱值污染整条曲线。

再补一条很多采集脚本没有的检查：obs_time的间隔分布。正常情况下相邻两条记录间隔在 5～70 分钟之间，如果出现大量 0 秒间隔，说明时间拼接出了问题（都拼成了同一时刻）；如果出现大段 3 小时以上的空洞，说明中间有人静默失败。

5. 避坑与排查：五条返工记录

这一章是我在类似项目上真实返工过的几处，按「现象、原因、解决」写，能对号入座就别自己再撞一遍。

5.1 请求层的三个坑

坑一：本地跑得好好的，上了服务器就 403。现象是同一份代码，本机返回完整数据，部署到机房 IP 之后稳定 403，换回本机又正常。原因是服务端对来源做了识别，缺Referer或者 UA 太像脚本，来自机房网段的请求更容易被拦。解决办法不是加代理绕，而是老实把请求头补齐：User-Agent写清楚（带上自用标识比伪装浏览器更合适），Referer指向站点自身的首页路径，同时把并发降到单线程、城市之间留间隔。补齐之后大部分 403 会消失，剩下的基本是频率问题。

坑二：跑了半天，失败率高得莫名其妙。现象是Retry全开的情况下仍然有 20% 以上的失败，日志里全是ReadTimeout。原因是默认超时。requests不传timeout就是无限等，一个挂住的连接会把整个线程占死，后面的城市全排队；传了但只传一个数字，往往把连接超时和读超时设成了同一个值。解决是按(3.05, 10)这种元组分开设，并且把pool_maxsize调到并发数以上，避免连接池排队。

坑三：某一个城市编码一直失败，其他都正常。现象是 99% 的城市成功，个别编码永远拿不到返回。原因有两层：一是这个编码本身没有独立数据（末位对应的是很小的区县），二是编码格式看着对但不在覆盖范围内。解决是在city_code表里加supported标记和parent_code，抓取失败时自动回退到上级城市，并在结果里标注「数据来自上级城市」，避免下游把市区天气当成郊区天气用。

5.2 解析与存储的两个坑

坑四：中文全是问号，但代码没报错。现象是城市名、风向全变成???，其他字段正常。原因是响应头声明的编码和实际内容不一致，resp.text按错的编码解了。解决是绕开resp.text，直接拿resp.content按 UTF-8 → GB18030 顺序试解码，都失败再errors="replace"。顺便提醒一句，别依赖自动编码探测，短中文文本的探测准确率低得离谱，探测出的编码时对时错，属于典型玄学。

坑五：JSON 解析偶发失败，重跑又好了。现象是每隔几天出现一批JSONDecodeError，重跑同样城市就成功。原因不是数据有问题，是那几次拿到的是限频页或维护公告页，长得也像一段 HTML/JSON 混合体，截壳逻辑截出了半截对象。解决是三步：解析前检查Content-Type和响应长度，明显不对直接判失败；strip_wrapper之后先用json.loads试一次，失败就把原文按city_code_时间戳.txt落盘；连续失败超过阈值触发冷却。存原文这一步非常关键，没有它，你只能在「偶发」这个词里打转。

6. 交叉校验与增量落地：让这份数据能被信任

抓下来只是原料，能被信任才算成品。我一般不在采集端做复杂校验，而是在采集之后单独跑一遍校验脚本，把可疑记录标出来而不是删掉——删掉就再也查不清到底是源的问题还是我的问题。

6.1 五条能挡住大部分错误的校验规则

前两条挡脏值，第三条挡乱序，第四条挡断档，第五条挡映射表写错。四条里最值钱的是第四条，因为断档是唯一一种「程序不报错、数据也不脏，但结论全错」的故障形态。

from datetime import timedelta def audit(city_code, rows, max_gap_min=90): """rows 需按 obs_time 升序。返回可疑记录与断档区间。""" suspects, gaps = [], [] prev = None for r in rows: if r["temp_c"] is not None and not (-60 <= r["temp_c"] <= 60): suspects.append((r["obs_time"], "temp_out_of_range")) if r["humidity"] is not None and not (0 <= r["humidity"] <= 100): suspects.append((r["obs_time"], "humidity_out_of_range")) if prev is not None: gap = r["obs_time"] - prev if gap <= timedelta(0): suspects.append((r["obs_time"], "time_not_increasing")) elif gap > timedelta(minutes=max_gap_min): gaps.append((prev, r["obs_time"], gap)) prev = r["obs_time"] return suspects, gaps

max_gap_min=90这个阈值是按小时级更新频率定的，取值比正常间隔留了一倍余量，太紧会天天误报，太松则挡不住真实断档。函数只返回可疑清单，不改数据，理由是校验规则本身也会错，先观察两周再决定要不要自动清理。

6.2 只落变更，把库压下来

跑满一个月之后你会发现，绝大多数写入都是重复的：温度没变、湿度没变，只是你又抓了一次。这时候加一层变更检测，能省下大量存储和后续扫描时间。

MUTABLE = ("temp_c", "humidity", "wind_dir", "wind_deg", "wind_speed_ms", "pressure_hpa", "visibility_km") def is_changed(new, old, eps=0.05): """数值字段允许 eps 容差，避免 23.40 与 23.399999 被判为变更。""" for key in MUTABLE: a, b = new.get(key), old.get(key) if isinstance(a, (int, float)) and isinstance(b, (int, float)): if abs(a - b) > eps: return True elif a != b: return True return False

容差eps=0.05是针对一位小数的温度、湿度定的，浮点比较不做容差迟早会出现「每次都在变」的假象，把增量表写成全量表。变更检测放在写入之前，先查同城市最近一条，is_changed为假就只更新collect_status的心跳时间，不落新行。

最后说一个习惯。从那以后我每次接新的公开数据源，都强制先跑一遍「三天数据回放」：把三天里每一轮的原始响应全存下来，离线重放解析和新表结构，确认字段没有中途变形、时间没有跳变、断档在可接受范围内，才让它接生产。这套流程看着慢，但比起上线一周后发现整月数据全是错位再回滚，快到不知道哪里去了。希望帮到你。

本文还有配套的精品资源，点击获取