简介这份压缩包围绕中国天气网天气数据获取面向具备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大致是这样依赖用途是否必需备注requestsHTTP 请求必需配合 urllib3 的 Retry 做重试urllib3连接层重试策略必需requests 自带无需单独装pandasCSV 批量导出可选可选数据量小的场景用标准库 csv 就够apscheduler定时调度可选只在常驻进程模式下需要loguru结构化日志可选换标准 logging 也行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区,101XXXXXXparent_code为空表示这是主城市编码。抓取时只抓主城市parent_code 区级编码留作「需要更细粒度时再开」的备选。加载这张表的代码很短但有几个细节# config/settings.py 中的加载函数 import csv from pathlib import Path def load_cities(pathconfig/cities.csv, only_top_levelTrue): cities [] with open(path, newline, encodingutf-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参数说明encodingutf-8-sig是为了兼容 Excel 保存 CSV 时加的 BOM 头不加的话第一列列名会变成\ufeffcity_codeDictReader拿到的 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, timeout8, max_retry3): self.base_url base_url self.timeout timeout self.session requests.Session() # 连接层重试只对 5xx 和连接错误生效4xx 不重试4xx 重试也没用 retry Retry( totalmax_retry, backoff_factor0.8, # 0.8s - 1.6s - 3.2s status_forcelist[500, 502, 503, 504], allowed_methods[GET], raise_on_statusFalse, ) adapter HTTPAdapter(max_retriesretry, pool_connections10, pool_maxsize10) 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}, timeoutself.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}逐项说参数。timeout8是连接和读取各自的上限公开接口正常响应在 1 秒内8 秒还没回来的基本可以判定为异常设太大只会让批量任务卡死。backoff_factor0.8控制的是指数退避的基数实际等待是0.8 × 2^(n-1)也就是 0.8、1.6、3.2 秒这个量级刚好间隔太短没意义间隔太长批量跑不动。status_forcelist只放 5xx 不放 429是因为 429 通常意味着「你被限流了」这时候重试只会加重限流正确做法是停下来等更久或者降低并发——这个逻辑挂在业务层更合适不该塞进连接层重试。raise_on_statusFalse配合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_asciiFalse), # 原始报文留底 }这段的重点是raw字段。把整个原始报文原样存一份看着浪费空间实际上是给未来的自己留的后悔药——上游哪天加了个「空气质量」字段你想回溯历史数据有raw就能重新解析没有就只能从今天开始攒。payload.get(data) or payload.get(weatherinfo)这个写法是应对「同一接口不同版本字段名不一致」的常见做法。or的短路特性刚好能用前一个取到空值就取后一个不用写两个if。3.4 字段清洗与落库单位剥离、枚举归一、UPSERTparser.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_pathdata/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 101XXXXXXrun_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, actionstore_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用一个文件管住所有可变项改环境只改这里配置项默认值说明调优建议BASE_URL实际接口地址上游气象接口根地址换数据源只改这一处TIMEOUT8 秒单次请求超时网络差可调到 12别超过 15MAX_RETRY35xx 重试次数线上建议 2避免拖长任务SLEEP_RANGE(0.3, 0.8)城市间随机间隔秒被限流时调到 (1.0, 2.0)DB_PATHdata/weather.dbSQLite 路径大表场景建议加索引文件分离CSV_OUTPUT_DIRdata/csvCSV 落盘目录按天分目录避免单目录文件过多LOG_DIRlogs运行日志与失败清单建议按月轮转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 21只有在需要「进程内缓存、动态调整间隔、任务间传递状态」的时候才值得上apscheduler。多数天气采集场景一天跑几次crontab 完全够用。5. 避坑与排查403、乱码、字段漂移这些坑我都踩过5.1 现象前几十个城市正常后面全返回 403现象批量跑到第 30 个城市左右后续全部返回 403单独重跑那一个城市又是 200。原因上游对单位时间内的请求数做了限制而且限制是按会话/来源维度的滑动窗口前面的请求把额度耗掉了。很多人第一反应是加长 sleep但如果并发维度也开着加 sleep 没用。解决把SLEEP_RANGE调到(1.0, 2.0)同时确认批量任务是串行的、没有开多线程。如果确实需要并发把并发数压到 2 以内并且并发之间也要共享一个限速器。另外检查User-Agent有没有被配置覆盖回默认值——有些工程的requests.get()在某个分支里没走SessionUA 就变回python-requests/x.x一样会被拒。5.2 现象CSV 里的中文城市名全是乱码现象SQLite 里查出来正常但用 Excel 打开导出的 CSV中文全是「鏌愬煄甯」。原因两个层面的编码问题叠在一起。一是响应解析时resp.encoding没设中文在入库前就已经坏了二是写 CSV 时用了encodingutf-8而 Excel 在 Windows 下默认按 GBK 解读无 BOM 的 UTF-8 文件。解决解析侧显式resp.encoding utf-8导出 CSV 侧用encodingutf-8-sig让 Python 在文件头写入 BOMExcel 就能正确识别。读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 左右进程直接没了没有任何 tracebacklogs里最后一行的下一条就是空。原因多半是被系统 OOM killer 干掉了。常见触发点是每次请求都新建Session连接池不释放或者把所有原始报文攒在内存里最后一次性写文件。解决Session在 fetcher 初始化时建一次全程复用结果边跑边写不要攒到最后。如果确实需要收集全部结果写到临时文件而不是内存列表。排查时可以看系统日志里的 OOM 记录或者在脚本里加resource.getrusage()定期打印内存占用。5.5 现象同一天的数据在库里出现两条现象查询某城市某天的数据发现同一个观测时间点有两条记录字段值略有差异。原因表建的时候没设联合主键或者主键设了但obs_time的格式不统一一个是2024-05-01 08:00:00一个是2024-05-01T08:00:00SQLite 认为是两个不同的值。解决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 个城市对账」的流程哪怕只改了一行正则。希望帮到你。本文还有配套的精品资源点击获取
