简介围绕中国天气网天气数据获取的编程实例面向使用 Visual Studio 2008 与 .NET Framework 开发气象信息应用的初学者和进阶开发者重点解决接口调用、请求响应处理与返回数据解析等问题。压缩包共 34 个文件以 9 个 C# 源码文件为核心配合项目配置、资源文件、程序集及可执行文件等整体约 259KB目录保持完整工程结构便于直接打开对照学习。已有 2314 人学习说明该主题具有稳定的实践参考需求。资源以 Windows Forms 示例工程呈现涵盖中国天气网 API 参数组织、HttpWebRequest/HttpWebResponse 请求收发、API 密钥使用、JSON 或 XML 数据解析以及温度、湿度、风向等信息在界面控件中的展示并涉及请求频率限制和异常捕获等排错思路。对希望理解传统 .NET 桌面端天气采集流程的读者较有参考价值可作为气象数据采集与界面展示的入门样例。1. 从页面复制到接口取数获取天气数据的落地边界如果你只是偶尔查一次天气浏览器打开页面就够了但要把获取天气数据接进日报、告警或调度系统页面复制粘贴立刻失效。公开气象数据站点通常把实况、逐小时和预报拆在不同入口真正能落到工程里的是把城市键、请求参数、响应结构和入库键拆开做成可重试、可校验的获取天气数据流程。这套实现适合做运维看板、门店排班、校园巡检和物流时效分析的开发者一台能跑 Python 的机器就能复现不适合想绕过访问控制、批量抓取非公开数据的人。2. 数据源与请求参数把获取天气数据拆成可重试的三步公开气象数据站点的页面看起来简单落到脚本里往往要处理三件事选哪个粒度、请求带什么头、失败后怎么退。先把这三步拆开后面的城市编码、解析和入库才有地方挂。下面按我平时写采集脚本的顺序展开每一步都落到可运行的代码块参数可以照着改。2.1 先定数据粒度实况、逐小时还是预报数据粒度不是越多越好。实况字段少、更新快适合做告警和看板逐小时数据量大适合排班、调度和趋势判断预报带未来时间适合提前量决策但唯一键要同时考虑城市、预报发布时间和目标时间。常见做法是先选一个主粒度跑通闭环再把其他粒度做成旁路表不要一上来全量抓否则请求量和存储都会失控。数据粒度常见更新节奏典型字段建议入库键适合场景实况分钟级到小时级温度、湿度、风向、风速城市键 观测时间告警、实时看板逐小时小时级温度、降水、云量城市键 目标小时排班、调度预报每日多次最高温、最低温、天气现象城市键 发布时间 目标日期提前量决策选型时先问一句业务到底是要“现在怎么样”还是“接下来怎么样”。如果只是看板展示实况加缓存就够如果要做明日排班预报的目标日期必须进入唯一键否则同一天多次发布的数据会互相覆盖。我一般把观测时间、发布时间、目标时间分三个字段存不在代码里做隐式转换后面排查会省很多事。提示公开数据也要看站点的使用说明和频率限制能少请求就少请求不要抓取登录后或非公开的数据。2.2 请求头与会话哪些字段必须带哪些别乱带请求头不是越多越好。公开天气接口通常认 User-Agent、Accept 和 Referer 这几个字段用来识别调用方和页面来源不要伪造登录态、Cookie 或私有签名。用requests.Session()复用连接把超时拆成连接超时和读取超时避免一个慢响应把整个任务拖住。下面这段是通用骨架URL 用占位域名实际替换成你所用公开站点的接口地址。import requests SESSION requests.Session() SESSION.headers.update({ # 标识调用用途方便对方识别不要伪装成登录用户 User-Agent: WeatherFetch/1.0 (contact: opsexample.com), Accept: application/json,text/plain,*/*, # 部分公开页面会校验来源按站点公开要求填写 Referer: https://example.com/weather/, }) def fetch_weather(city_key: str, data_type: str observe) - requests.Response: url https://example.com/api/weather params { city: city_key, type: data_type, } # 连接 3.05 秒读取 8 秒可按网络情况调整 resp SESSION.get(url, paramsparams, timeout(3.05, 8)) resp.raise_for_status() return resp逻辑说明会话对象把请求头、连接池和重试策略集中在一处避免每个函数各写一套raise_for_status()只负责把 4xx/5xx 抛出来业务层再决定重试还是跳过。参数说明city_key是数据源能识别的城市键不是中文城市名data_type控制实况、逐小时或预报入口timeout用元组比单数字更可控连接阶段失败通常比读取阶段快。请求头里最容易踩坑的是 Referer。有的站点只校验域名前缀有的会校验完整路径还有的不校验先按页面公开请求抄一份再逐项删减验证不要一次性堆十几个头。User-Agent 写清用途和联系方式比伪装成浏览器更稳妥也方便出问题时对方定位。2.3 超时、重试与限流用退避代替硬扛失败重试最怕写成“立刻再试三次”。公开接口遇到 429 或 5xx 时短时间密集重试只会让情况更糟。常见做法是只对 GET 重试对 429、500、502、503、504 做指数退避并且把总次数压到 3 次以内业务层再加一层串行限流保证同一城市不会并发轰炸。下面这段用urllib3的Retry挂到会话上参数不复杂但足够应付大多数公开接口的瞬时抖动。from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry retry Retry( total3, # 最多重试 3 次含首次之外的重试 connect3, read2, backoff_factor0.8, # 退避间隔0.8s、1.6s、3.2s status_forcelist[429, 500, 502, 503, 504], allowed_methods[GET], respect_retry_after_headerTrue, ) adapter HTTPAdapter(max_retriesretry, pool_connections4, pool_maxsize4) SESSION.mount(https://, adapter) SESSION.mount(http://, adapter)逻辑说明Retry只处理连接失败、读取失败和指定状态码不处理 404 这类业务错误respect_retry_after_headerTrue会优先遵守服务端返回的等待时间。参数说明backoff_factor不要设得太小0.5 到 1 之间比较常见pool_maxsize不需要开大采集任务通常串行或小并发allowed_methods限定 GET避免误重试写操作。限流放在业务层更直观同一城市两次请求之间至少间隔一个固定秒数批量城市按队列逐个跑失败的城市进入下一轮而不是原地死循环。这样即使某个城市键配错也不会把整批任务拖垮。3. 城市编码与请求构造从城市名到数据源能识别的键城市名是最容易出问题的一层。页面上的“A市”“A 市”“A市区”在数据源里可能对应不同键直接拿中文名拼参数十有八九返回空数据。稳妥的做法是维护一张自己的城市编码表把别名、正式名、数据源代码分开存脚本只认编码表里的键。这一章把清洗、构造和响应识别串起来。3.1 城市名清洗与编码表落地清洗只做确定性的动作去首尾空格、去内部空白、统一全半角、去掉“市辖区”“城区”这类后缀。不要用模糊匹配猜城市猜错一次比报错更难查。下面这个函数只做基础归一化返回的还是中文名真正的映射交给编码表。import re def normalize_city(name: str) - str: if not name: return name name.strip() # 去掉所有空白字符包括全角空格 name re.sub(r\s, , name) # 只处理明确的后缀不要过度替换 for suffix in (市辖区, 城区): if name.endswith(suffix) and len(name) len(suffix): name name[: -len(suffix)] return name逻辑说明normalize_city不负责查表只把输入整理成稳定形式后续用别名表匹配。参数说明name来自表单、CSV 或页面选择框后缀列表按你的数据源实际情况增减不要把所有“市”字都删掉否则“A市”和“A”会被混为一谈。编码表建议用 CSV 维护字段至少包含alias、code、level、enabled。alias可以放多个别名用竖线分隔code是数据源识别的键enabled用来临时停用某个错误映射。加载时用utf-8-sig兼容带 BOM 的文件避免第一列读出来带不可见字符。import csv def load_city_map(path: str) - dict[str, str]: mapping: dict[str, str] {} with open(path, newline, encodingutf-8-sig) as f: for row in csv.DictReader(f): if row.get(enabled, 1).strip() ! 1: continue code (row.get(code) or ).strip() if not code: continue aliases (row.get(alias) or ).split(|) for alias in aliases: key normalize_city(alias) if key: mapping[key] code return mapping逻辑说明先过滤停用行再按别名逐个写入映射后写入的别名会覆盖先写入的所以表里不要放冲突别名。参数说明path是 CSV 路径alias用竖线分隔多个写法code必须非空enabled为1才生效。实际跑的时候先打印映射条数和前几条样例确认没有把“A市”映射到“B县”的键。字段含义示例注意alias城市别名多个用竖线分隔A市A市区code数据源城市键X1001以实际公开接口为准level行政层级city只作辅助不参与匹配enabled是否启用10 表示跳过3.2 构造请求URL、参数与签名边界请求构造要把“稳定参数”和“可变参数”分开。城市键、数据类型、时间范围属于可变参数固定的接口路径、版本号、来源标识放在配置里。公开接口通常不需要签名遇到需要私有签名或加密参数的接口先确认它是不是公开数据不是公开数据就不要碰不要逆向签名逻辑。下面这个函数只做参数拼装不做请求方便单测。from urllib.parse import urlencode def build_params(city_key: str, data_type: str, target_date: str | None None) - dict: params { city: city_key, type: data_type, format: json, } if data_type forecast and target_date: params[date] target_date return params def build_url(base_url: str, params: dict) - str: return f{base_url}?{urlencode(params)}逻辑说明build_params把业务参数收口避免在请求函数里散落字符串build_url只用于日志和调试真正请求还是把params交给requests让它处理编码。参数说明data_type与站点入口对应target_date只对预报类接口有意义格式按站点要求常见是YYYY-MM-DDformat不是所有站点都认按实际返回调整。注意把完整请求 URL 打进日志前先确认没有敏感参数公开接口也不要把内部令牌写进 URL。3.3 响应结构识别JSON、JSONP 与 HTML 的三种入口同一个站点不同入口可能返回不同结构接口返回 JSON老页面返回 JSONP降级页返回 HTML。解析器不能默认resp.json()一定成功。常见做法是先看Content-Type再看正文首字符最后才决定用哪种解析。下面这个函数把三种情况收在一起解析失败时保留原始文本片段方便排查。import json import re def detect_and_parse(resp) - dict: ctype (resp.headers.get(Content-Type) or ).lower() text resp.text.strip() if json in ctype and not text.startswith(((, [)): return {kind: json, data: resp.json()} # JSONPcallback({...}); m re.search(r^[^(]*\((.*)\)\s*;?\s*$, text, re.S) if m: return {kind: jsonp, data: json.loads(m.group(1))} if text.startswith(({, [)): return {kind: json, data: json.loads(text)} return {kind: html, data: {raw: text[:500]}}逻辑说明先按 Content-Type 走快路径再兜底识别 JSONP 和裸 JSON最后把 HTML 截断保留避免日志爆掉。参数说明resp是requests.ResponseJSONP 正则只取第一个括号对复杂的嵌套回调不要用这个函数硬解raw只留前 500 个字符够定位问题即可。识别出结构后业务层要按kind分流JSON 走字段映射JSONP 转成 JSON 后走同一套映射HTML 直接标记为异常并进入重放队列。这样后面字段漂移时至少能知道是解析层错了还是源站换了页面。4. 解析与落库把天气字段变成可查询的数据拿到响应只是第一步真正让获取天气数据变得可用的是字段映射和入库键。字段名要跟业务字段分开源字段改了只改映射表入库键要能区分“同一城市同一观测时间”的重复抓取不能让重跑产生重复行。下面按解析、落库、缓存放三块写。4.1 JSON 字段映射与缺失值处理字段映射最忌讳“源字段直接当表字段”。源站可能把温度叫temp也可能叫temperature还可能放在data.obs下面。常见做法是写一个显式映射函数输出固定结构的字典缺失值保留None不要用 0 填。温度 0 度和“没有数据”是两回事填 0 会让告警误判。def parse_observe(payload: dict) - dict: data payload.get(data) or {} obs data.get(obs) or data def pick(*keys): for key in keys: if key in obs and obs[key] not in (, None): return obs[key] return None return { city_key: pick(cityKey, city_code, cityId), obs_time: pick(obsTime, observe_time, time), temp: pick(temp, temperature), humidity: pick(humidity, rh), wind_dir: pick(windDir, wind_direction), wind_speed: pick(windSpeed, wind_speed), }逻辑说明pick按候选键顺序取第一个非空值兼容源字段改名data.get(obs) or data兼容字段嵌套一层和没嵌套两种情况。参数说明payload是解析后的 JSON 字典候选键按你实际见到的字段增减返回结构固定后面入库只认这几个键。源字段候选目标字段类型缺失处理cityKey / city_codecity_key字符串为空则丢弃该条obsTime / observe_timeobs_time字符串为空则丢弃该条temp / temperaturetemp浮点数保留 Nonehumidity / rhhumidity浮点数保留 NonewindDir / wind_directionwind_dir字符串保留 NonewindSpeed / wind_speedwind_speed浮点数保留 None映射完成后先做一次断言city_key和obs_time不能为空。这两个字段是入库键缺一个就无法判断重复。其他字段可以空空值在展示层做“暂无数据”处理不要在下游用 0 代替。4.2 增量写入与唯一键设计入库键建议用(city_key, obs_time)。obs_time是观测时间不是抓取时间抓取时间单独放fetched_at只用于排查和统计。每次抓取用INSERT ... ON CONFLICT DO UPDATE重跑不会产生重复行只会更新同一条记录。下面这段 SQLite DDL 和 upsert 可以直接放进脚本。import sqlite3 DDL CREATE TABLE IF NOT EXISTS weather_obs ( city_key TEXT NOT NULL, obs_time TEXT NOT NULL, temp REAL, humidity REAL, wind_dir TEXT, wind_speed REAL, fetched_at TEXT NOT NULL, content_hash TEXT NOT NULL, PRIMARY KEY (city_key, obs_time) ); def init_db(conn: sqlite3.Connection) - None: conn.execute(DDL) conn.commit() def save_obs(conn: sqlite3.Connection, row: dict) - None: conn.execute( INSERT INTO weather_obs( city_key, obs_time, temp, humidity, wind_dir, wind_speed, fetched_at, content_hash ) VALUES( :city_key, :obs_time, :temp, :humidity, :wind_dir, :wind_speed, :fetched_at, :content_hash ) ON CONFLICT(city_key, obs_time) DO UPDATE SET temp excluded.temp, humidity excluded.humidity, wind_dir excluded.wind_dir, wind_speed excluded.wind_speed, fetched_at excluded.fetched_at, content_hash excluded.content_hash , row)逻辑说明主键约束保证同一城市同一观测时间只有一行excluded表示本次准备插入的值冲突时用它覆盖旧值。参数说明row必须包含全部占位符字段fetched_at用 UTC 时间字符串content_hash用于判断内容是否真的变化下一节展开。如果业务需要保留历史版本不要在主表上做覆盖另建一张weather_obs_history每次抓取都插入主表只保留最新。两张表共用同一套解析结果写入时先写历史再 upsert 主表重跑也不会丢版本。4.3 缓存与失败重放缓存分两层短周期内避免重复请求失败后能重放。短缓存用内存字典或 Redis 都行键用city_key data_type值存响应摘要和过期时间失败重放把原始请求参数和响应片段落一张fetch_failed表下一轮优先跑失败队列。内容指纹用标准化 JSON 的 SHA-256字段顺序变化不会影响结果。import hashlib import json from datetime import datetime, timezone def content_hash(payload: dict) - str: body json.dumps(payload, ensure_asciiFalse, sort_keysTrue, separators(,, :)) return hashlib.sha256(body.encode(utf-8)).hexdigest() def now_utc() - str: return datetime.now(timezone.utc).strftime(%Y-%m-%dT%H:%M:%SZ)逻辑说明sort_keysTrue让字典键顺序固定separators去掉多余空格保证同一内容得到同一指纹。参数说明payload建议传解析后的业务字段不要传整个原始响应否则时间戳或请求 ID 变化会导致指纹永远不同now_utc只用于fetched_at不参与业务时间。失败重放表至少存city_key、data_type、params_json、last_error、retry_count、next_retry_at。重试次数超过 3 次就标记为人工处理不要无限重放。这样一套跑下来获取天气数据的链路就有了解析、入库和回捞三个落点后面排查也有黑匣子可看。5. 避坑与排查获取天气数据最常见的五类翻车这一类脚本跑通一次不难难的是连续跑一周不出怪数据。下面五条是我踩过的坑按“现象 → 原因 → 解决”写每条都能直接对照日志排查。5.1 空值不是接口坏了编码与城市键先查现象接口返回 200data是空对象或空数组日志里看不出异常。原因通常不是接口挂了而是城市键不对中文名没归一化、编码表里别名冲突、把“A市”传成了“A市区”。解决顺序是先用编码表打印实际请求的city_key再拿这个键手工请求一次最后对比页面选择框里的值和表里的值。我一般会在请求前加一条断言city_key必须匹配^[A-Z0-9_]$这类格式格式不对直接跳过并记失败队列避免把空数据写进表。5.2 状态码 200 但内容不对响应类型与重定向现象resp.json()报解析错误或者解析出来字段全是None。原因可能是请求被重定向到 HTML 说明页也可能入口返回了 JSONP甚至返回了风控提示页。解决方法是先看Content-Type和正文前 80 个字符再用 3.3 节的detect_and_parse分流重定向次数用allow_redirectsFalse单独观察一次确认跳转目标是不是公开页面。不要一上来就加verifyFalse或关闭重定向那会把真正的问题盖住。5.3 频控与封禁别把重试写成攻击现象前几次正常后面连续 429 或 403换城市也不恢复。原因多半是重试太密或并发太高触发了站点的频率限制。解决方法是把重试退避拉开批量任务改成串行给每个城市加最小间隔已经拿到缓存的数据先不请求失败队列放到下一轮。血泪经验是不要把Retry.total设成 10 以上也不要在 429 后立刻换 User-Agent 继续打先降速比换头有用。5.4 时间与时区观测时间别直接当入库时间现象看板上温度曲线在凌晨出现断崖或者同一条记录反复更新。原因可能是obs_time带时区、带毫秒或者直接把抓取时间当成了观测时间。解决方法是统一存 UTC展示层再转本地时区obs_time解析失败就丢弃不要用now()兜底。入库键只用观测时间抓取时间放fetched_at这样重跑不会改变业务时间。5.5 字段漂移解析器不能只认一种结构现象昨天还能跑的脚本今天报KeyError或字段全空。原因通常是源站调整了字段名或嵌套层级。解决方法是解析器用候选键列表输出结构固定同时把原始响应摘要写入日志字段缺失时能对比前后差异。我一般会在每天第一轮跑完后统计空值率某个字段连续两轮空值率超过阈值就发告警而不是等到看板画不出来才发现。6. 进阶把获取天气数据做成可校验的增量任务6.1 双源校验与阈值告警单源数据出问题时最怕整条链路静默。常见做法是接第二个公开气象数据源做交叉校验不要求字段完全一致只比对温度、湿度、风速的偏差。下面这张表是我一般会设的检查项阈值按业务容忍度改不要照搬。校验项建议阈值处理温度偏差超过 5 摄氏度标记可疑保留双源原始值湿度偏差超过 25 个百分点标记可疑下一轮复查风速偏差超过 8 米/秒标记可疑人工确认空值率单城市连续 2 轮超过 50%发告警并暂停该城市6.2 调度与幂等重跑不重复写入调度用系统定时任务就够关键是幂等。每次任务只处理“当前小时未成功”的城市成功标记写回队列表重跑时先查(city_key, obs_time)是否存在存在就走更新不存在才插入。命令行入口可以写成这样参数从环境变量或配置文件读不要把城市列表写死在代码里。# 示例每 10 分钟跑一轮失败队列优先 */10 * * * * cd /opt/weather-fetch /usr/bin/python3 run_once.py --queue failed --limit 50 logs/fetch.log 21逻辑说明--queue failed表示先跑失败队列--limit 50控制单轮请求量避免一次拉太多触发频控。参数说明路径和 Python 解释器按实际环境替换日志追加到文件方便后面按城市键检索。6.3 一个具体技巧用内容指纹判断是否真的更新很多公开接口会在响应里带请求时间或随机串直接比对整个响应会误判“每次都更新”。我一般只对解析后的业务字段做指纹指纹相同就不写库只更新fetched_at。这样既减少写入也能让“数据到底有没有变”这件事有据可查。def should_write(old_hash: str | None, new_hash: str) - bool: # 没有旧记录或内容真的变了才写业务字段 return old_hash is None or old_hash ! new_hash逻辑说明old_hash从表里按(city_key, obs_time)查查不到就是Nonenew_hash由 4.3 节的content_hash生成。参数说明指纹只覆盖业务字段不要包含fetched_at和请求 ID。从那以后我每次接入新的公开气象数据源都会先把城市编码表、字段映射表和失败队列表跑一遍再开定时任务这三张表不清后面全是后悔药。希望帮到你。本文还有配套的精品资源点击获取 
长河君 · 2026/11/2 13:25:47 · 阅读 1.2k 获取天气数据：可重试采集与幂等增量入库
简介围绕中国天气网天气数据获取的编程实例面向使用 Visual Studio 2008 与 .NET Framework 开发气象信息应用的初学者和进阶开发者重点解决接口调用、请求响应处理与返回数据解析等问题。压缩包共 34 个文件以 9 个 C# 源码文件为核心配合项目配置、资源文件、程序集及可执行文件等整体约 259KB目录保持完整工程结构便于直接打开对照学习。已有 2314 人学习说明该主题具有稳定的实践参考需求。资源以 Windows Forms 示例工程呈现涵盖中国天气网 API 参数组织、HttpWebRequest/HttpWebResponse 请求收发、API 密钥使用、JSON 或 XML 数据解析以及温度、湿度、风向等信息在界面控件中的展示并涉及请求频率限制和异常捕获等排错思路。对希望理解传统 .NET 桌面端天气采集流程的读者较有参考价值可作为气象数据采集与界面展示的入门样例。1. 从页面复制到接口取数获取天气数据的落地边界如果你只是偶尔查一次天气浏览器打开页面就够了但要把获取天气数据接进日报、告警或调度系统页面复制粘贴立刻失效。公开气象数据站点通常把实况、逐小时和预报拆在不同入口真正能落到工程里的是把城市键、请求参数、响应结构和入库键拆开做成可重试、可校验的获取天气数据流程。这套实现适合做运维看板、门店排班、校园巡检和物流时效分析的开发者一台能跑 Python 的机器就能复现不适合想绕过访问控制、批量抓取非公开数据的人。2. 数据源与请求参数把获取天气数据拆成可重试的三步公开气象数据站点的页面看起来简单落到脚本里往往要处理三件事选哪个粒度、请求带什么头、失败后怎么退。先把这三步拆开后面的城市编码、解析和入库才有地方挂。下面按我平时写采集脚本的顺序展开每一步都落到可运行的代码块参数可以照着改。2.1 先定数据粒度实况、逐小时还是预报数据粒度不是越多越好。实况字段少、更新快适合做告警和看板逐小时数据量大适合排班、调度和趋势判断预报带未来时间适合提前量决策但唯一键要同时考虑城市、预报发布时间和目标时间。常见做法是先选一个主粒度跑通闭环再把其他粒度做成旁路表不要一上来全量抓否则请求量和存储都会失控。数据粒度常见更新节奏典型字段建议入库键适合场景实况分钟级到小时级温度、湿度、风向、风速城市键 观测时间告警、实时看板逐小时小时级温度、降水、云量城市键 目标小时排班、调度预报每日多次最高温、最低温、天气现象城市键 发布时间 目标日期提前量决策选型时先问一句业务到底是要“现在怎么样”还是“接下来怎么样”。如果只是看板展示实况加缓存就够如果要做明日排班预报的目标日期必须进入唯一键否则同一天多次发布的数据会互相覆盖。我一般把观测时间、发布时间、目标时间分三个字段存不在代码里做隐式转换后面排查会省很多事。提示公开数据也要看站点的使用说明和频率限制能少请求就少请求不要抓取登录后或非公开的数据。2.2 请求头与会话哪些字段必须带哪些别乱带请求头不是越多越好。公开天气接口通常认 User-Agent、Accept 和 Referer 这几个字段用来识别调用方和页面来源不要伪造登录态、Cookie 或私有签名。用requests.Session()复用连接把超时拆成连接超时和读取超时避免一个慢响应把整个任务拖住。下面这段是通用骨架URL 用占位域名实际替换成你所用公开站点的接口地址。import requests SESSION requests.Session() SESSION.headers.update({ # 标识调用用途方便对方识别不要伪装成登录用户 User-Agent: WeatherFetch/1.0 (contact: opsexample.com), Accept: application/json,text/plain,*/*, # 部分公开页面会校验来源按站点公开要求填写 Referer: https://example.com/weather/, }) def fetch_weather(city_key: str, data_type: str observe) - requests.Response: url https://example.com/api/weather params { city: city_key, type: data_type, } # 连接 3.05 秒读取 8 秒可按网络情况调整 resp SESSION.get(url, paramsparams, timeout(3.05, 8)) resp.raise_for_status() return resp逻辑说明会话对象把请求头、连接池和重试策略集中在一处避免每个函数各写一套raise_for_status()只负责把 4xx/5xx 抛出来业务层再决定重试还是跳过。参数说明city_key是数据源能识别的城市键不是中文城市名data_type控制实况、逐小时或预报入口timeout用元组比单数字更可控连接阶段失败通常比读取阶段快。请求头里最容易踩坑的是 Referer。有的站点只校验域名前缀有的会校验完整路径还有的不校验先按页面公开请求抄一份再逐项删减验证不要一次性堆十几个头。User-Agent 写清用途和联系方式比伪装成浏览器更稳妥也方便出问题时对方定位。2.3 超时、重试与限流用退避代替硬扛失败重试最怕写成“立刻再试三次”。公开接口遇到 429 或 5xx 时短时间密集重试只会让情况更糟。常见做法是只对 GET 重试对 429、500、502、503、504 做指数退避并且把总次数压到 3 次以内业务层再加一层串行限流保证同一城市不会并发轰炸。下面这段用urllib3的Retry挂到会话上参数不复杂但足够应付大多数公开接口的瞬时抖动。from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry retry Retry( total3, # 最多重试 3 次含首次之外的重试 connect3, read2, backoff_factor0.8, # 退避间隔0.8s、1.6s、3.2s status_forcelist[429, 500, 502, 503, 504], allowed_methods[GET], respect_retry_after_headerTrue, ) adapter HTTPAdapter(max_retriesretry, pool_connections4, pool_maxsize4) SESSION.mount(https://, adapter) SESSION.mount(http://, adapter)逻辑说明Retry只处理连接失败、读取失败和指定状态码不处理 404 这类业务错误respect_retry_after_headerTrue会优先遵守服务端返回的等待时间。参数说明backoff_factor不要设得太小0.5 到 1 之间比较常见pool_maxsize不需要开大采集任务通常串行或小并发allowed_methods限定 GET避免误重试写操作。限流放在业务层更直观同一城市两次请求之间至少间隔一个固定秒数批量城市按队列逐个跑失败的城市进入下一轮而不是原地死循环。这样即使某个城市键配错也不会把整批任务拖垮。3. 城市编码与请求构造从城市名到数据源能识别的键城市名是最容易出问题的一层。页面上的“A市”“A 市”“A市区”在数据源里可能对应不同键直接拿中文名拼参数十有八九返回空数据。稳妥的做法是维护一张自己的城市编码表把别名、正式名、数据源代码分开存脚本只认编码表里的键。这一章把清洗、构造和响应识别串起来。3.1 城市名清洗与编码表落地清洗只做确定性的动作去首尾空格、去内部空白、统一全半角、去掉“市辖区”“城区”这类后缀。不要用模糊匹配猜城市猜错一次比报错更难查。下面这个函数只做基础归一化返回的还是中文名真正的映射交给编码表。import re def normalize_city(name: str) - str: if not name: return name name.strip() # 去掉所有空白字符包括全角空格 name re.sub(r\s, , name) # 只处理明确的后缀不要过度替换 for suffix in (市辖区, 城区): if name.endswith(suffix) and len(name) len(suffix): name name[: -len(suffix)] return name逻辑说明normalize_city不负责查表只把输入整理成稳定形式后续用别名表匹配。参数说明name来自表单、CSV 或页面选择框后缀列表按你的数据源实际情况增减不要把所有“市”字都删掉否则“A市”和“A”会被混为一谈。编码表建议用 CSV 维护字段至少包含alias、code、level、enabled。alias可以放多个别名用竖线分隔code是数据源识别的键enabled用来临时停用某个错误映射。加载时用utf-8-sig兼容带 BOM 的文件避免第一列读出来带不可见字符。import csv def load_city_map(path: str) - dict[str, str]: mapping: dict[str, str] {} with open(path, newline, encodingutf-8-sig) as f: for row in csv.DictReader(f): if row.get(enabled, 1).strip() ! 1: continue code (row.get(code) or ).strip() if not code: continue aliases (row.get(alias) or ).split(|) for alias in aliases: key normalize_city(alias) if key: mapping[key] code return mapping逻辑说明先过滤停用行再按别名逐个写入映射后写入的别名会覆盖先写入的所以表里不要放冲突别名。参数说明path是 CSV 路径alias用竖线分隔多个写法code必须非空enabled为1才生效。实际跑的时候先打印映射条数和前几条样例确认没有把“A市”映射到“B县”的键。字段含义示例注意alias城市别名多个用竖线分隔A市A市区code数据源城市键X1001以实际公开接口为准level行政层级city只作辅助不参与匹配enabled是否启用10 表示跳过3.2 构造请求URL、参数与签名边界请求构造要把“稳定参数”和“可变参数”分开。城市键、数据类型、时间范围属于可变参数固定的接口路径、版本号、来源标识放在配置里。公开接口通常不需要签名遇到需要私有签名或加密参数的接口先确认它是不是公开数据不是公开数据就不要碰不要逆向签名逻辑。下面这个函数只做参数拼装不做请求方便单测。from urllib.parse import urlencode def build_params(city_key: str, data_type: str, target_date: str | None None) - dict: params { city: city_key, type: data_type, format: json, } if data_type forecast and target_date: params[date] target_date return params def build_url(base_url: str, params: dict) - str: return f{base_url}?{urlencode(params)}逻辑说明build_params把业务参数收口避免在请求函数里散落字符串build_url只用于日志和调试真正请求还是把params交给requests让它处理编码。参数说明data_type与站点入口对应target_date只对预报类接口有意义格式按站点要求常见是YYYY-MM-DDformat不是所有站点都认按实际返回调整。注意把完整请求 URL 打进日志前先确认没有敏感参数公开接口也不要把内部令牌写进 URL。3.3 响应结构识别JSON、JSONP 与 HTML 的三种入口同一个站点不同入口可能返回不同结构接口返回 JSON老页面返回 JSONP降级页返回 HTML。解析器不能默认resp.json()一定成功。常见做法是先看Content-Type再看正文首字符最后才决定用哪种解析。下面这个函数把三种情况收在一起解析失败时保留原始文本片段方便排查。import json import re def detect_and_parse(resp) - dict: ctype (resp.headers.get(Content-Type) or ).lower() text resp.text.strip() if json in ctype and not text.startswith(((, [)): return {kind: json, data: resp.json()} # JSONPcallback({...}); m re.search(r^[^(]*\((.*)\)\s*;?\s*$, text, re.S) if m: return {kind: jsonp, data: json.loads(m.group(1))} if text.startswith(({, [)): return {kind: json, data: json.loads(text)} return {kind: html, data: {raw: text[:500]}}逻辑说明先按 Content-Type 走快路径再兜底识别 JSONP 和裸 JSON最后把 HTML 截断保留避免日志爆掉。参数说明resp是requests.ResponseJSONP 正则只取第一个括号对复杂的嵌套回调不要用这个函数硬解raw只留前 500 个字符够定位问题即可。识别出结构后业务层要按kind分流JSON 走字段映射JSONP 转成 JSON 后走同一套映射HTML 直接标记为异常并进入重放队列。这样后面字段漂移时至少能知道是解析层错了还是源站换了页面。4. 解析与落库把天气字段变成可查询的数据拿到响应只是第一步真正让获取天气数据变得可用的是字段映射和入库键。字段名要跟业务字段分开源字段改了只改映射表入库键要能区分“同一城市同一观测时间”的重复抓取不能让重跑产生重复行。下面按解析、落库、缓存放三块写。4.1 JSON 字段映射与缺失值处理字段映射最忌讳“源字段直接当表字段”。源站可能把温度叫temp也可能叫temperature还可能放在data.obs下面。常见做法是写一个显式映射函数输出固定结构的字典缺失值保留None不要用 0 填。温度 0 度和“没有数据”是两回事填 0 会让告警误判。def parse_observe(payload: dict) - dict: data payload.get(data) or {} obs data.get(obs) or data def pick(*keys): for key in keys: if key in obs and obs[key] not in (, None): return obs[key] return None return { city_key: pick(cityKey, city_code, cityId), obs_time: pick(obsTime, observe_time, time), temp: pick(temp, temperature), humidity: pick(humidity, rh), wind_dir: pick(windDir, wind_direction), wind_speed: pick(windSpeed, wind_speed), }逻辑说明pick按候选键顺序取第一个非空值兼容源字段改名data.get(obs) or data兼容字段嵌套一层和没嵌套两种情况。参数说明payload是解析后的 JSON 字典候选键按你实际见到的字段增减返回结构固定后面入库只认这几个键。源字段候选目标字段类型缺失处理cityKey / city_codecity_key字符串为空则丢弃该条obsTime / observe_timeobs_time字符串为空则丢弃该条temp / temperaturetemp浮点数保留 Nonehumidity / rhhumidity浮点数保留 NonewindDir / wind_directionwind_dir字符串保留 NonewindSpeed / wind_speedwind_speed浮点数保留 None映射完成后先做一次断言city_key和obs_time不能为空。这两个字段是入库键缺一个就无法判断重复。其他字段可以空空值在展示层做“暂无数据”处理不要在下游用 0 代替。4.2 增量写入与唯一键设计入库键建议用(city_key, obs_time)。obs_time是观测时间不是抓取时间抓取时间单独放fetched_at只用于排查和统计。每次抓取用INSERT ... ON CONFLICT DO UPDATE重跑不会产生重复行只会更新同一条记录。下面这段 SQLite DDL 和 upsert 可以直接放进脚本。import sqlite3 DDL CREATE TABLE IF NOT EXISTS weather_obs ( city_key TEXT NOT NULL, obs_time TEXT NOT NULL, temp REAL, humidity REAL, wind_dir TEXT, wind_speed REAL, fetched_at TEXT NOT NULL, content_hash TEXT NOT NULL, PRIMARY KEY (city_key, obs_time) ); def init_db(conn: sqlite3.Connection) - None: conn.execute(DDL) conn.commit() def save_obs(conn: sqlite3.Connection, row: dict) - None: conn.execute( INSERT INTO weather_obs( city_key, obs_time, temp, humidity, wind_dir, wind_speed, fetched_at, content_hash ) VALUES( :city_key, :obs_time, :temp, :humidity, :wind_dir, :wind_speed, :fetched_at, :content_hash ) ON CONFLICT(city_key, obs_time) DO UPDATE SET temp excluded.temp, humidity excluded.humidity, wind_dir excluded.wind_dir, wind_speed excluded.wind_speed, fetched_at excluded.fetched_at, content_hash excluded.content_hash , row)逻辑说明主键约束保证同一城市同一观测时间只有一行excluded表示本次准备插入的值冲突时用它覆盖旧值。参数说明row必须包含全部占位符字段fetched_at用 UTC 时间字符串content_hash用于判断内容是否真的变化下一节展开。如果业务需要保留历史版本不要在主表上做覆盖另建一张weather_obs_history每次抓取都插入主表只保留最新。两张表共用同一套解析结果写入时先写历史再 upsert 主表重跑也不会丢版本。4.3 缓存与失败重放缓存分两层短周期内避免重复请求失败后能重放。短缓存用内存字典或 Redis 都行键用city_key data_type值存响应摘要和过期时间失败重放把原始请求参数和响应片段落一张fetch_failed表下一轮优先跑失败队列。内容指纹用标准化 JSON 的 SHA-256字段顺序变化不会影响结果。import hashlib import json from datetime import datetime, timezone def content_hash(payload: dict) - str: body json.dumps(payload, ensure_asciiFalse, sort_keysTrue, separators(,, :)) return hashlib.sha256(body.encode(utf-8)).hexdigest() def now_utc() - str: return datetime.now(timezone.utc).strftime(%Y-%m-%dT%H:%M:%SZ)逻辑说明sort_keysTrue让字典键顺序固定separators去掉多余空格保证同一内容得到同一指纹。参数说明payload建议传解析后的业务字段不要传整个原始响应否则时间戳或请求 ID 变化会导致指纹永远不同now_utc只用于fetched_at不参与业务时间。失败重放表至少存city_key、data_type、params_json、last_error、retry_count、next_retry_at。重试次数超过 3 次就标记为人工处理不要无限重放。这样一套跑下来获取天气数据的链路就有了解析、入库和回捞三个落点后面排查也有黑匣子可看。5. 避坑与排查获取天气数据最常见的五类翻车这一类脚本跑通一次不难难的是连续跑一周不出怪数据。下面五条是我踩过的坑按“现象 → 原因 → 解决”写每条都能直接对照日志排查。5.1 空值不是接口坏了编码与城市键先查现象接口返回 200data是空对象或空数组日志里看不出异常。原因通常不是接口挂了而是城市键不对中文名没归一化、编码表里别名冲突、把“A市”传成了“A市区”。解决顺序是先用编码表打印实际请求的city_key再拿这个键手工请求一次最后对比页面选择框里的值和表里的值。我一般会在请求前加一条断言city_key必须匹配^[A-Z0-9_]$这类格式格式不对直接跳过并记失败队列避免把空数据写进表。5.2 状态码 200 但内容不对响应类型与重定向现象resp.json()报解析错误或者解析出来字段全是None。原因可能是请求被重定向到 HTML 说明页也可能入口返回了 JSONP甚至返回了风控提示页。解决方法是先看Content-Type和正文前 80 个字符再用 3.3 节的detect_and_parse分流重定向次数用allow_redirectsFalse单独观察一次确认跳转目标是不是公开页面。不要一上来就加verifyFalse或关闭重定向那会把真正的问题盖住。5.3 频控与封禁别把重试写成攻击现象前几次正常后面连续 429 或 403换城市也不恢复。原因多半是重试太密或并发太高触发了站点的频率限制。解决方法是把重试退避拉开批量任务改成串行给每个城市加最小间隔已经拿到缓存的数据先不请求失败队列放到下一轮。血泪经验是不要把Retry.total设成 10 以上也不要在 429 后立刻换 User-Agent 继续打先降速比换头有用。5.4 时间与时区观测时间别直接当入库时间现象看板上温度曲线在凌晨出现断崖或者同一条记录反复更新。原因可能是obs_time带时区、带毫秒或者直接把抓取时间当成了观测时间。解决方法是统一存 UTC展示层再转本地时区obs_time解析失败就丢弃不要用now()兜底。入库键只用观测时间抓取时间放fetched_at这样重跑不会改变业务时间。5.5 字段漂移解析器不能只认一种结构现象昨天还能跑的脚本今天报KeyError或字段全空。原因通常是源站调整了字段名或嵌套层级。解决方法是解析器用候选键列表输出结构固定同时把原始响应摘要写入日志字段缺失时能对比前后差异。我一般会在每天第一轮跑完后统计空值率某个字段连续两轮空值率超过阈值就发告警而不是等到看板画不出来才发现。6. 进阶把获取天气数据做成可校验的增量任务6.1 双源校验与阈值告警单源数据出问题时最怕整条链路静默。常见做法是接第二个公开气象数据源做交叉校验不要求字段完全一致只比对温度、湿度、风速的偏差。下面这张表是我一般会设的检查项阈值按业务容忍度改不要照搬。校验项建议阈值处理温度偏差超过 5 摄氏度标记可疑保留双源原始值湿度偏差超过 25 个百分点标记可疑下一轮复查风速偏差超过 8 米/秒标记可疑人工确认空值率单城市连续 2 轮超过 50%发告警并暂停该城市6.2 调度与幂等重跑不重复写入调度用系统定时任务就够关键是幂等。每次任务只处理“当前小时未成功”的城市成功标记写回队列表重跑时先查(city_key, obs_time)是否存在存在就走更新不存在才插入。命令行入口可以写成这样参数从环境变量或配置文件读不要把城市列表写死在代码里。# 示例每 10 分钟跑一轮失败队列优先 */10 * * * * cd /opt/weather-fetch /usr/bin/python3 run_once.py --queue failed --limit 50 logs/fetch.log 21逻辑说明--queue failed表示先跑失败队列--limit 50控制单轮请求量避免一次拉太多触发频控。参数说明路径和 Python 解释器按实际环境替换日志追加到文件方便后面按城市键检索。6.3 一个具体技巧用内容指纹判断是否真的更新很多公开接口会在响应里带请求时间或随机串直接比对整个响应会误判“每次都更新”。我一般只对解析后的业务字段做指纹指纹相同就不写库只更新fetched_at。这样既减少写入也能让“数据到底有没有变”这件事有据可查。def should_write(old_hash: str | None, new_hash: str) - bool: # 没有旧记录或内容真的变了才写业务字段 return old_hash is None or old_hash ! new_hash逻辑说明old_hash从表里按(city_key, obs_time)查查不到就是Nonenew_hash由 4.3 节的content_hash生成。参数说明指纹只覆盖业务字段不要包含fetched_at和请求 ID。从那以后我每次接入新的公开气象数据源都会先把城市编码表、字段映射表和失败队列表跑一遍再开定时任务这三张表不清后面全是后悔药。希望帮到你。本文还有配套的精品资源点击获取