Python天气数据采集管道：编码映射、请求重试与SQLite落库

简介：这份压缩包围绕中国天气网天气数据获取，面向具备C#基础、使用VS2008与.NET Framework开发气象应用的开发者，提供可直接参考的Windows Forms示例工程。包内共有34个文件，以9个cs源码为主，并含sln解决方案、csproj项目配置、config配置、resx资源及exe、dll等程序集文件，整体约259KB，可执行程序便于查看运行效果。已有2314人学习，说明其在同类基础示例中有一定参考度。工程覆盖API接口调用、HTTP请求与响应处理、JSON或XML解析、密钥配置及界面数据显示等环节，并涉及城市ID、查询类型、实时天气或预报数据、温度湿度风向等字段提取，可帮助读者理解天气信息从请求到呈现的组织方式，并排查配置与解析问题。目录保留解决方案与项目配置，便于在VS2008中打开对照，适合作为天气数据采集入门与课程实践的小型样例。

1. 先别急着写爬虫：这份资源解决的是「获取天气数据」这件事

有个做社区生鲜排班的朋友，说排班表上要写未来三天天气，他原以为获取天气数据就是一次 HTTP 请求，两周后却被三件事拖住：城市名和接口要的编码对不上、返回字段悄悄改过两次、批量跑几百个城市直接被限流。真正麻烦从来不是发请求，而是把编码映射、请求、解析、落库、重试拼成能长期跑的东西。这份资源按这个思路拆开：一个轻量 Python 工程，请求层与解析层分离，城市编码单独维护成表，抓下来的数据同时写 CSV 和 SQLite，适合物流调度、门店补货、排班看板这类需要稳定拿天气的二次开发场景。只想查今天下不下雨的话，别用它。

2. 项目结构与技术选型：请求层、解析层、存储层怎么分工

2.1 目录结构与分层职责

先看仓库摊开后的样子。整个工程的目录按职责切，每个包只做一件事，接手上手会轻松很多：

weather-collector/ ├── config/ │ ├── settings.py # 接口地址、超时、并发、存储路径 │ └── cities.csv # 城市名 -> 城市编码映射表 ├── core/ │ ├── fetcher.py # 只负责发请求，返回原始文本 │ ├── parser.py # 只负责把文本变成结构化 dict │ ├── storage.py # 只负责写 CSV / SQLite │ ├── normalizer.py # 字段清洗、单位剥离、天气现象归一 │ └── pipeline.py # 把上面四件事串起来 ├── scripts/ │ ├── run_once.py # 单城市调试入口 │ └── run_batch.py # 批量入口 ├── data/ │ └── weather.db ├── logs/ ├── requirements.txt └── README.md

这里的分层逻辑值得说两句。fetcher.py拿到的永远是str，它不知道天气字段长什么样；parser.py拿到的永远是一段文本，它不关心这段文本是 HTTP 拿的还是从文件里读的；storage.py拿到的永远是dict列表，它不关心数据从哪来。这么切的好处很直接：上游接口改字段名，你只动parser.py；换成 PostgreSQL，你只动storage.py；想加个「从本地缓存读」的降级分支，只需要在pipeline.py里换个分支。

我一般会把normalizer.py单独拎出来，而不是塞进parser.py。原因是解析和清洗的变更频率完全不一样——解析跟着接口走，一年改几次；清洗跟着业务走，业务方今天说要数字温度、明天说要天气现象枚举值，改得很勤。混在一起，两个人都要抢同一个文件。

config/cities.csv单独成表也很关键。很多新手把城市编码硬编码在代码里，写成CITY_MAP = {"某城市": "101XXXXXX", ...}，一开始没感觉，等到要加一个区级编码，就得改代码、提交、发版。做成 CSV 之后，运营同学自己就能加行，代码侧只负责读。

2.2 技术选型：为什么是这几个依赖

依赖清单我一般不铺得太开，这个工程的requirements.txt大致是这样：

requests几乎是默认选择，但很多人只用它的get()，不用Session。这个项目里批量跑城市的时候，Session的连接复用能省一大截握手开销，几百个城市跑下来差别明显。重试策略我挂在HTTPAdapter上而不是自己写for循环，因为连接层重试能覆盖「TCP 握手失败」这类try/except抓不到的边缘情况。

存储这块，SQLite 是够的。天气数据本身结构简单、写入量不大（几百个城市 × 每天几次），SQLite 单文件、零部署、支持UPSERT，非常适合这种「跑完就想查」的场景。真有并发写入需求再换 PostgreSQL，接口不用变。

提示：pandas只在「一次性导出历史数据给业务方」时才值得引入。日常写入用sqlite3标准库，启动快、内存占用小，别为了一个to_csv把整个 pandas 拖进来。

2.3 一次抓取在四个模块里的数据流向

把一次完整抓取拆开看，数据是这样流动的：pipeline.py从cities.csv读出一个城市编码，交给fetcher.fetch(code)；fetcher 返回(raw_text, error)；没有 error 就交给parser.parse(raw_text)拿到原始 dict；再交给normalizer.clean(raw_dict)得到字段干净、单位统一的 dict；最后storage.save(records)落库。整条链路在pipeline.py里就是几十行：

# core/pipeline.py from core.fetcher import WeatherFetcher from core.parser import parse_weather from core.normalizer import clean_weather from core.storage import WeatherStorage def run_one(fetcher, storage, city_code, city_name): raw_text, err = fetcher.fetch(city_code) if err: # 失败不抛异常，交给上层收集失败清单，方便重跑 return {"city_code": city_code, "city_name": city_name, "ok": False, "err": err} parsed = parse_weather(raw_text, city_code) if parsed is None: return {"city_code": city_code, "city_name": city_name, "ok": False, "err": "parse_failed"} record = clean_weather(parsed, city_name) storage.save(record) return {"city_code": city_code, "city_name": city_name, "ok": True, "err": None} def run_batch(city_list, fetcher, storage, sleep_fn): results = [] for city in city_list: results.append(run_one(fetcher, storage, city["code"], city["name"])) sleep_fn() # 由调用方注入，方便测试时传空函数 return results

这段代码里有两个设计点。第一是失败不抛异常，而是返回一个带ok标志的字典。批量跑几百个城市，中间某一个失败很正常，如果直接抛异常，整个任务就断了；返回字典可以最后统一汇总失败清单，只重跑失败的那几个。第二是sleep_fn通过参数注入，而不是在函数里硬写time.sleep()。这样写单元测试的时候传一个lambda: None，几百个城市的测试用例一秒跑完，不用真等。

run_batch返回的results是后续所有排查工作的抓手。我在实际使用里会把它存成logs/run_2024xxxx.json，里面就三样东西：跑成功多少、失败的城市编码、失败原因。下次重跑时读这个文件，只跑失败的。

3. 核心链路拆解：从城市编码到天气数据落库

3.1 城市编码表：把城市名翻译成接口认识的 ID

上游气象接口基本都不认城市名，只认编码。这类公开气象数据站点的编码通常是 9 位数字，按层级递增分配：同一个城市下面会挂多个区级编码，区级编码往往是在市级编码基础上往后递增。这就带来一个隐性问题——如果你只按「城市名」去查，很容易漏掉区，或者把同一个城市的多个区当成不同城市重复抓。

cities.csv的结构建议做成三列：

city_code,city_name,parent_code 101XXXXXX,某市, 101XXXX0X,某市A区,101XXXXXX 101XXXX0Y,某市B区,101XXXXXX

parent_code为空表示这是主城市编码。抓取时只抓主城市（parent_code == ""），区级编码留作「需要更细粒度时再开」的备选。加载这张表的代码很短，但有几个细节：

# config/settings.py 中的加载函数 import csv from pathlib import Path def load_cities(path="config/cities.csv", only_top_level=True): cities = [] with open(path, newline="", encoding="utf-8-sig") as f: reader = csv.DictReader(f) for row in reader: code = (row.get("city_code") or "").strip() name = (row.get("city_name") or "").strip() parent = (row.get("parent_code") or "").strip() # 编码必须是纯数字且长度为 9，脏数据直接跳过并打日志 if not code.isdigit() or len(code) != 9: print(f"[跳过] 非法编码: {row}") continue if only_top_level and parent: continue cities.append({"code": code, "name": name}) return cities

参数说明：encoding="utf-8-sig"是为了兼容 Excel 保存 CSV 时加的 BOM 头，不加的话第一列列名会变成\ufeffcity_code，DictReader拿到的 key 就对不上，row.get("city_code")永远是None，这个坑后面还会细讲。only_top_level默认True，意思是只抓主城市；改成False就会连区级一起抓，数据量大概翻三到五倍。

编码校验那段别嫌麻烦。真实场景里这张表是人工维护的，一定会有多打一个空格、少写一位、粘进来一个全角字符的情况。校验放在加载阶段，比等请求打出去返回 400 再去查要省事得多。

3.2 请求构造：UA、Referer、超时与重试的写法

fetcher.py是整个工程里最容易出问题的地方，因为它是唯一和外网打交道的模块。公开气象接口对请求头的检查通常比想象中严格，默认的python-requests/2.xUA 基本会被直接拒掉。

# core/fetcher.py import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry class WeatherFetcher: def __init__(self, base_url, timeout=8, max_retry=3): self.base_url = base_url self.timeout = timeout self.session = requests.Session() # 连接层重试：只对 5xx 和连接错误生效，4xx 不重试（4xx 重试也没用） retry = Retry( total=max_retry, backoff_factor=0.8, # 0.8s -> 1.6s -> 3.2s status_forcelist=[500, 502, 503, 504], allowed_methods=["GET"], raise_on_status=False, ) adapter = HTTPAdapter(max_retries=retry, pool_connections=10, pool_maxsize=10) self.session.mount("https://", adapter) self.session.mount("http://", adapter) self.session.headers.update({ "User-Agent": ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/120.0 Safari/537.36"), "Accept": "application/json, text/plain, */*", "Accept-Encoding": "gzip, deflate", "Referer": base_url, "Connection": "keep-alive", }) def fetch(self, city_code): url = f"{self.base_url}/weather" try: resp = self.session.get(url, params={"code": city_code}, timeout=self.timeout) resp.raise_for_status() # 关键一行：不要依赖 requests 自动猜编码 resp.encoding = "utf-8" return resp.text, None except requests.RequestException as e: return None, f"{type(e).__name__}: {e}"

逐项说参数。timeout=8是连接和读取各自的上限，公开接口正常响应在 1 秒内，8 秒还没回来的基本可以判定为异常，设太大只会让批量任务卡死。backoff_factor=0.8控制的是指数退避的基数，实际等待是0.8 × 2^(n-1)，也就是 0.8、1.6、3.2 秒，这个量级刚好，间隔太短没意义，间隔太长批量跑不动。

status_forcelist只放 5xx 不放 429，是因为 429 通常意味着「你被限流了」，这时候重试只会加重限流，正确做法是停下来等更久，或者降低并发——这个逻辑挂在业务层更合适，不该塞进连接层重试。

raise_on_status=False配合raise_for_status()是个细节：让 urllib3 不因为状态码抛异常（避免重试逻辑和异常处理打架），由业务代码统一用raise_for_status()判断，异常都走同一个except。

resp.encoding = "utf-8"这行单独拎出来讲。requests在响应头没有charset时，会按ISO-8859-1猜测编码，中文全变乱码。手动指定之后resp.text才是对的。这一行不加，后面所有解析都会出问题，而且报错信息很难指向真正原因。

3.3 返回解析：从 JSON 里取需要的字段

拿到文本之后先别急着json.loads。这类公开接口返回有两种形态，一种是纯 JSON，另一种是带回调名的 JSONP，形如var data = {...};。如果直接json.loads会抛JSONDecodeError，而报错信息里只有「Expecting value: line 1 column 1」，新手很容易以为是网络问题。

# core/parser.py import json import re _JSONP_PATTERN = re.compile(r"^[^(]*\((.*)\)[^)]*$", re.S) def _loads(raw_text): text = raw_text.strip() # 先尝试纯 JSON try: return json.loads(text) except json.JSONDecodeError: pass # 再尝试剥 JSONP 外壳 m = _JSONP_PATTERN.match(text) if m: return json.loads(m.group(1)) raise ValueError("既不是合法 JSON，也不是 JSONP") def parse_weather(raw_text, city_code): try: payload = _loads(raw_text) except (ValueError, json.JSONDecodeError): return None # 上游常见的包装层，data 可能不存在，用 or {} 兜底 data = payload.get("data") or payload.get("weatherinfo") or {} if not data: return None return { "city_code": city_code, "obs_time": data.get("obsTime") or data.get("time"), "temperature": data.get("temp"), "humidity": data.get("humidity"), "wind_dir": data.get("windDir"), "wind_power": data.get("windPower"), "weather_text": data.get("weather"), "pressure": data.get("pressure"), "visibility": data.get("vis"), "raw": json.dumps(payload, ensure_ascii=False), # 原始报文留底 }

这段的重点是raw字段。把整个原始报文原样存一份，看着浪费空间，实际上是给未来的自己留的后悔药——上游哪天加了个「空气质量」字段，你想回溯历史数据，有raw就能重新解析，没有就只能从今天开始攒。

payload.get("data") or payload.get("weatherinfo")这个写法是应对「同一接口不同版本字段名不一致」的常见做法。or的短路特性刚好能用：前一个取到空值就取后一个，不用写两个if。

3.4 字段清洗与落库：单位剥离、枚举归一、UPSERT

parser.py出来的数据还带着单位，温度可能是"23℃"、湿度可能是"56%"、风力可能是"3级"。这些字符串直接入库，后续做统计就全是麻烦。normalizer.py负责把它们变成数字：

# core/normalizer.py import re from datetime import datetime _NUM = re.compile(r"-?\d+(?:\.\d+)?") def _to_float(value): if value is None: return None if isinstance(value, (int, float)): return float(value) m = _NUM.search(str(value)) return float(m.group()) if m else None WEATHER_ENUM = { "晴": "clear", "多云": "cloudy", "阴": "overcast", "小雨": "rain_light", "中雨": "rain_moderate", "大雨": "rain_heavy", "雪": "snow", "雾": "fog", "霾": "haze", } def clean_weather(parsed, city_name): return { "city_code": parsed["city_code"], "city_name": city_name, "obs_time": parsed["obs_time"] or datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "temperature": _to_float(parsed["temperature"]), "humidity": _to_float(parsed["humidity"]), "wind_power": _to_float(parsed["wind_power"]), "wind_dir": parsed["wind_dir"], "weather_text": parsed["weather_text"], "weather_type": WEATHER_ENUM.get(parsed["weather_text"] or "", "unknown"), "raw": parsed["raw"], }

_NUM这个正则特意带了-?，因为北方冬季温度是负的，写成-8℃，只匹配\d+就会把负号吃掉，变成 8 度。这个 bug 特别隐蔽，跑了一个冬天才发现数据不对。

WEATHER_ENUM是个开放式映射表。上游返回的天气现象文本有很多种表述，比如「晴间多云」「局部多云」「阵雨转晴」，这些没法穷举，所以默认落到unknown，靠weather_text原始值兜底。等数据攒够了，再把高频的unknown补进映射表，比一开始就硬猜要靠谱。

落库用 SQLite 的UPSERT，天然解决重复抓取的问题：

# core/storage.py import sqlite3, json DDL = """ CREATE TABLE IF NOT EXISTS weather_obs ( city_code TEXT NOT NULL, obs_time TEXT NOT NULL, city_name TEXT, temperature REAL, humidity REAL, wind_power REAL, wind_dir TEXT, weather_text TEXT, weather_type TEXT, raw TEXT, created_at TEXT DEFAULT (datetime('now', 'localtime')), PRIMARY KEY (city_code, obs_time) ); CREATE INDEX IF NOT EXISTS idx_obs_time ON weather_obs(obs_time); """ class WeatherStorage: def __init__(self, db_path="data/weather.db"): self.conn = sqlite3.connect(db_path) self.conn.executescript(DDL) def save(self, record): cols = ["city_code", "obs_time", "city_name", "temperature", "humidity", "wind_power", "wind_dir", "weather_text", "weather_type", "raw"] sql = f""" INSERT INTO weather_obs ({','.join(cols)}) VALUES ({','.join(['?'] * len(cols))}) ON CONFLICT(city_code, obs_time) DO UPDATE SET temperature = excluded.temperature, humidity = excluded.humidity, wind_power = excluded.wind_power, weather_text = excluded.weather_text, weather_type = excluded.weather_type, raw = excluded.raw; """ self.conn.execute(sql, [record.get(c) for c in cols]) self.conn.commit()

主键选(city_code, obs_time)是这套设计里最关键的一个决定。同一个城市同一个观测时间点只会有一条记录，重复跑同一批城市时不会堆数据，而是用最新一次抓取的结果覆盖。raw字段也在DO UPDATE SET里，保证「接口改了字段名、你想回溯重解析」时拿到的是最新报文。

idx_obs_time索引是给查询用的。业务方最常见的查询是「某时间段内所有城市的温度」，按obs_time走索引比全表扫快一个量级，这个索引成本很低，直接建上。

4. 跑起来：环境准备、配置项与三种运行方式

4.1 环境与依赖安装

Python 版本建议 3.9 以上，低于这个版本sqlite3对UPSERT语法的支持虽然也有（3.24 起支持），但一些字符串方法的差异会让人摸不着头脑。虚拟环境用venv就够，工程不大，没必要上更重的工具。

# 1. 创建并激活虚拟环境 python -m venv .venv source .venv/bin/activate # Windows 用 .venv\Scripts\activate # 2. 装依赖（只有必需项） pip install -r requirements.txt # 3. 建目录（首次运行需要） mkdir -p data logs # 4. 冒烟测试：只跑一个城市，确认链路通 python -m scripts.run_once --city-code 101XXXXXX

run_once.py是入口脚本，我在里面加了个冒烟模式：只抓一个城市、只打印不落库，方便第一次确认网络和解析是否正常：

# scripts/run_once.py import argparse from config.settings import load_cities, BASE_URL, DB_PATH from core.fetcher import WeatherFetcher from core.parser import parse_weather from core.normalizer import clean_weather from core.storage import WeatherStorage from core.pipeline import run_one def main(): ap = argparse.ArgumentParser() ap.add_argument("--city-code", help="指定城市编码，不传则取 cities.csv 第一个") ap.add_argument("--dry-run", action="store_true", help="只打印不落库") args = ap.parse_args() cities = load_cities() target = next((c for c in cities if c["code"] == args.city_code), cities[0]) fetcher = WeatherFetcher(BASE_URL) raw, err = fetcher.fetch(target["code"]) if err: print(f"[失败] {target['name']} -> {err}") return parsed = parse_weather(raw, target["code"]) record = clean_weather(parsed, target["name"]) print(f"[成功] {record['city_name']} {record['obs_time']} " f"{record['temperature']}℃ {record['weather_text']}") if not args.dry_run: WeatherStorage(DB_PATH).save(record) if __name__ == "__main__": main()

参数说明：--dry-run用来隔离「请求和解析」跟「写库」这两步。第一次跑的时候用它，能快速判断问题出在网络、解析还是存储。--city-code不传的话默认取cities.csv的第一行，这样python -m scripts.run_once直接就能跑，不用记编码。

4.2 配置项逐项说明

配置集中在config/settings.py，用一个文件管住所有可变项，改环境只改这里：

SLEEP_RANGE用随机区间而不是固定值，是很有意的一个细节。固定time.sleep(0.5)会让请求呈现完全规律的周期，反而不像正常流量；随机区间更自然，也能顺便打散瞬时并发。

4.3 三种运行方式：单城市、批量、定时

三种模式的适用场景不一样。单城市用于调试和验证，改完parser.py就拿一个城市跑一遍。批量用于日常采集，跑完全部城市。常驻定时用于长期部署，进程不退出，按间隔触发。

批量入口：

# 抓所有主城市，失败清单写到 logs/failed.json python -m scripts.run_batch --all --sleep 0.5 # 只重跑上次失败的城市 python -m scripts.run_batch --retry-failed logs/failed.json

--retry-failed对应run_batch.py里的一个分支：读 JSON 失败清单，只跑里面失败的编码。这个功能在批量任务里几乎是必需的，因为一次跑 300 个城市，总有几个会因为临时的网络抖动失败，全量重跑既慢又容易触发限流。

定时模式可以用apscheduler，也可以直接用系统的 crontab。我的建议是：能用 crontab 就别上调度库。一个run_batch.py加一行 crontab，比引入调度框架、维护常驻进程、处理进程崩溃要轻得多：

# 每天 07:05 和 13:05 各跑一次，输出重定向到日志 5 7,13 * * * cd /path/to/weather-collector && \ .venv/bin/python -m scripts.run_batch --all >> logs/cron.log 2>&1

只有在需要「进程内缓存、动态调整间隔、任务间传递状态」的时候，才值得上apscheduler。多数天气采集场景，一天跑几次，crontab 完全够用。

5. 避坑与排查：403、乱码、字段漂移这些坑我都踩过

5.1 现象：前几十个城市正常，后面全返回 403

现象：批量跑到第 30 个城市左右，后续全部返回 403，单独重跑那一个城市又是 200。

原因：上游对单位时间内的请求数做了限制，而且限制是按会话/来源维度的滑动窗口，前面的请求把额度耗掉了。很多人第一反应是加长 sleep，但如果并发维度也开着，加 sleep 没用。

解决：把SLEEP_RANGE调到(1.0, 2.0)，同时确认批量任务是串行的、没有开多线程。如果确实需要并发，把并发数压到 2 以内，并且并发之间也要共享一个限速器。另外检查User-Agent有没有被配置覆盖回默认值——有些工程的requests.get()在某个分支里没走Session，UA 就变回python-requests/x.x，一样会被拒。

5.2 现象：CSV 里的中文城市名全是乱码

现象：SQLite 里查出来正常，但用 Excel 打开导出的 CSV，中文全是「鏌愬煄甯」。

原因：两个层面的编码问题叠在一起。一是响应解析时resp.encoding没设，中文在入库前就已经坏了；二是写 CSV 时用了encoding="utf-8"，而 Excel 在 Windows 下默认按 GBK 解读无 BOM 的 UTF-8 文件。

解决：解析侧显式resp.encoding = "utf-8"；导出 CSV 侧用encoding="utf-8-sig"，让 Python 在文件头写入 BOM，Excel 就能正确识别。读cities.csv时同样要用utf-8-sig，否则列名会带上不可见的 BOM 字符。

5.3 现象：某天开始某几个字段永远是 None

现象：温度、湿度正常，但wind_dir、pressure从某天起全部是None，日志里没有任何报错。

原因：上游接口调整了字段名，比如从windDir改成wind_direction。因为代码里用的是data.get("windDir")，取不到就返回None，不会抛异常，所以静默失败。

解决：在parser.py里加一层「字段存在性检查」，关键字段缺失时记一条 WARN 日志，而不是静默None。更彻底的做法是保留raw字段（前面设计里已经有了），事后可以从历史报文里重新解析。发现字段漂移后，把新老字段名都加进get的候选列表，用or连起来兜一段时间。

5.4 现象：批量跑到一半进程无声退出

现象：跑 200 个城市，跑到 130 左右进程直接没了，没有任何 traceback，logs里最后一行的下一条就是空。

原因：多半是被系统 OOM killer 干掉了。常见触发点是每次请求都新建Session，连接池不释放；或者把所有原始报文攒在内存里，最后一次性写文件。

解决：Session在 fetcher 初始化时建一次，全程复用；结果边跑边写，不要攒到最后。如果确实需要收集全部结果，写到临时文件而不是内存列表。排查时可以看系统日志里的 OOM 记录，或者在脚本里加resource.getrusage()定期打印内存占用。

5.5 现象：同一天的数据在库里出现两条

现象：查询某城市某天的数据，发现同一个观测时间点有两条记录，字段值略有差异。

原因：表建的时候没设联合主键，或者主键设了但obs_time的格式不统一（一个是2024-05-01 08:00:00，一个是2024-05-01T08:00:00），SQLite 认为是两个不同的值。

解决：obs_time在normalizer.py里统一格式化，把T分隔符、时区后缀全部规整成%Y-%m-%d %H:%M:%S。主键必须是(city_code, obs_time)联合主键，且UPSERT的ON CONFLICT子句要和主键完全对应。已经脏了的历史数据，用一条DELETE加重新插入的方式清理一次。

注意：这五条里，第 3 条和第 5 条是最容易被忽略的——它们不报错，只是让数据慢慢变脏。加一条「关键字段非空率」的每日巡检，比事后翻数据要省事得多。

6. 二次开发：把单次抓取改成能长期跑的数据管道

单次能跑通只是起点，真正要上生产，还得补三块：调度、失败隔离、存储可替换。这三块的改造量都不大，但顺序很重要——先做失败隔离，再做调度，最后才动存储。

先做失败隔离，是因为批量任务里失败是常态。我一般会把run_batch的返回结果落成结构化 JSON，字段就三个：ok_count、failed（城市编码列表）、reasons（编码到原因的映射）。每次跑完把这份文件覆盖写，同时保留最近七天的历史。这样出问题时不用猜，直接看哪几个城市、什么原因。

再做调度。这里有个容易翻车的地方：不要一上来就写常驻进程加while True: sleep()。这种写法一旦某次请求卡住，整个循环就停了，而且没有任何告警。稳妥的做法是 crontab 触发短命进程，每次跑完就退出，进程状态由系统托管。真要常驻，至少给单次任务加一个总超时（比如 15 分钟），超时就主动退出，让下一轮重新开始。

最后动存储。把WeatherStorage抽成一个基类，SqliteStorage和PostgresStorage各自实现：

# core/storage.py 抽象层 from abc import ABC, abstractmethod class BaseStorage(ABC): @abstractmethod def save(self, record: dict) -> None: ... @abstractmethod def query(self, city_code: str, start: str, end: str) -> list: ... class SqliteStorage(BaseStorage): def save(self, record): ... def query(self, city_code, start, end): cur = self.conn.execute( "SELECT * FROM weather_obs WHERE city_code=? AND obs_time BETWEEN ? AND ? " "ORDER BY obs_time", (city_code, start, end)) return cur.fetchall()

改造之后，pipeline.py里只依赖BaseStorage，换后端时上层一行不用改。这个抽象层还有个附带好处：写测试时可以用一个InMemoryStorage，断言逻辑直接检查内存里的列表，不用碰真实数据库。

验证改造有没有把数据搞坏，我的做法是每天跑一次「对账」：随机抽 10 个城市，用改造后的代码和改造前的旧脚本各跑一遍，比对temperature、humidity、weather_text三个字段是否一致。连续对七天没差异，才算改造完成。这一步看着笨，但真能拦住「字段映射改漏一个」这类问题。

这套东西我是踩过坑之后才理清的。早些年我图快，直接把解析逻辑写进批量循环里，结果上游改字段之后，我花了一整晚从日志里翻哪几个城市受影响，最后还是靠raw字段重新解析才救回来。从那以后我每次动数据管道，都强制走一遍「抽 10 个城市对账」的流程，哪怕只改了一行正则。希望帮到你。

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