企业官网建设流程全解析

简介：围绕中国天气网天气数据获取的编程实例，面向使用 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: ops@example.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, params=params, 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( total=3, # 最多重试 3 次，含首次之外的重试 connect=3, read=2, backoff_factor=0.8, # 退避间隔：0.8s、1.6s、3.2s status_forcelist=[429, 500, 502, 503, 504], allowed_methods=["GET"], respect_retry_after_header=True, ) adapter = HTTPAdapter(max_retries=retry, pool_connections=4, pool_maxsize=4) SESSION.mount("https://", adapter) SESSION.mount("http://", adapter)

逻辑说明：Retry只处理连接失败、读取失败和指定状态码，不处理 404 这类业务错误；respect_retry_after_header=True会优先遵守服务端返回的等待时间。参数说明：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="", encoding="utf-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县”的键。

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-DD；format不是所有站点都认，按实际返回调整。

注意：把完整请求 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()} # JSONP：callback({...}); 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.Response；JSONP 正则只取第一个括号对，复杂的嵌套回调不要用这个函数硬解；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 字典；候选键按你实际见到的字段增减；返回结构固定，后面入库只认这几个键。

映射完成后先做一次断言：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_ascii=False, sort_keys=True, 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_keys=True让字典键顺序固定，separators去掉多余空格，保证同一内容得到同一指纹。参数说明：payload建议传解析后的业务字段，不要传整个原始响应，否则时间戳或请求 ID 变化会导致指纹永远不同；now_utc只用于fetched_at，不参与业务时间。

失败重放表至少存city_key、data_type、params_json、last_error、retry_count、next_retry_at。重试次数超过 3 次就标记为人工处理，不要无限重放。这样一套跑下来，获取天气数据的链路就有了解析、入库和回捞三个落点，后面排查也有黑匣子可看。

5. 避坑与排查：获取天气数据最常见的五类翻车

这一类脚本跑通一次不难，难的是连续跑一周不出怪数据。下面五条是我踩过的坑，按“现象 → 原因 → 解决”写，每条都能直接对照日志排查。

5.1 空值不是接口坏了：编码与城市键先查

现象：接口返回 200，data是空对象或空数组，日志里看不出异常。原因通常不是接口挂了，而是城市键不对：中文名没归一化、编码表里别名冲突、把“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_redirects=False单独观察一次，确认跳转目标是不是公开页面。不要一上来就加verify=False或关闭重定向，那会把真正的问题盖住。

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 双源校验与阈值告警

单源数据出问题时，最怕整条链路静默。常见做法是接第二个公开气象数据源做交叉校验，不要求字段完全一致，只比对温度、湿度、风速的偏差。下面这张表是我一般会设的检查项，阈值按业务容忍度改，不要照搬。

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 2>&1

逻辑说明：--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)查，查不到就是None；new_hash由 4.3 节的content_hash生成。参数说明：指纹只覆盖业务字段，不要包含fetched_at和请求 ID。从那以后我每次接入新的公开气象数据源，都会先把城市编码表、字段映射表和失败队列表跑一遍，再开定时任务；这三张表不清，后面全是后悔药。希望帮到你。

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