简介：面向Python爬虫初学者以及需要获取实时天气数据的开发者，这份实战项目围绕中国天气网（weather.com.cn）展开，演示如何用requests发送HTTP请求、配合BeautifulSoup解析HTML，提取日期、天气状况、温度与风力等字段，涵盖网页抓取、数据提取到结果整理的完整链路，可作为爬虫入门练手，也可在数据分析之前采集数据。压缩包共1498个文件，约8.03MB，主体是840个py源码与515个pyc字节码，另含exe可执行文件、testcase测试用例、txt说明以及wheel、installer等依赖安装组件，并附带虚拟环境相关文件，能省去部分环境配置的麻烦。已有622人学习。配套的完整代码示例与逐段解析，可帮助读者理解请求头构造、页面结构定位与字段清洗思路，并迁移到其他城市或其他页面结构的抓取任务中。

1. 从一次 403 说起：天气数据抓取到底难在哪

给内部看板补天气字段那次，我信心满满地requests.get了气象站点的首页，拿回来的是一个几乎没数据的壳页面，连刷几次直接 403。后来才想明白：这类站点的实况观测数据根本不在 HTML 里，而在浏览器 Network 面板那几条 XHR 请求上——带 Referer、带毫秒时间戳、响应体是一段var dataSK = {...}的赋值语句而不是纯 JSON。这就是 Python 爬虫抓天气数据最容易翻车的地方：目标不是“网页”，而是网页背后那条接口。这篇按接口定位、请求构造、解析入库、调度自动化、避坑排查的顺序走一遍，适合需要每天拿一批城市实况做看板、做分析、做补数的从业者，也适合刚学完 requests 想找个真实目标练手的人。

2. 接口定位与请求构造：把浏览器发的那一份请求原样搬下来

抓天气数据这件事，九成的失败发生在写代码之前——没找对请求。先花十分钟在浏览器里把“人是怎么拿到数据的”看清楚，后面写代码就是抄作业。

2.1 Network 面板里那几条请求，先分清哪条是真数据

按 F12 打开开发者工具，切到 Network 面板，勾上 Fetch/XHR，然后刷新页面。这时候会看到几条请求，通常分三类：页面本身的 HTML 文档、静态资源（js/css/图片）、以及带城市编码的数据接口。前两类直接忽略，重点看第三类——点开 Response 标签，如果看到的是结构化的键值对，或者var dataSK = {...}这种赋值语句，那就是目标。

我的习惯是先在过滤框里敲城市编码或weather、sk这类关键词缩小范围，再逐条点开看。找到之后要记三样东西：完整的 Request URL（包括协议和路径）、Request Headers 里的 Referer 与 User-Agent、Query String Parameters 里的全部参数。参数里经常有一个_=1730000000000这样的毫秒时间戳，它的作用是破浏览器缓存，服务端一般不校验具体值，但少了它有些接口会返回旧数据。

提示：不要一上来就在请求上右键 Copy as cURL 然后丢给在线转换工具。先人工看一眼哪些参数是必需的、哪些是可选的，否则很容易把一串一次性 Cookie 固化进代码，第二天全部失效。

还有一点值得说：同一个页面可能同时存在“实时实况”和“七天预报”两条接口，路径前缀不同、返回结构也不同。实况接口通常按小时更新，预报接口一天更新几次，抓取频率要分开设定。混在一起按同一节奏跑，既浪费请求数，也容易触发频控。

2.2 requests 复刻请求：headers 里哪几个字段是硬门槛

找到接口后，把它翻译成 requests 代码。核心是把路径、查询参数、请求头三部分对齐。下面这段是最小可用版本，我在多个气象类接口上都用过类似结构。

import time import requests CITY_CODE = "101010100" # 占位示例，换成目标城市的 9 位编码 def fetch_realtime(city_code: str, timeout: int = 10) -> str: """抓取单城市实时实况，返回原始响应文本""" ts = int(time.time() * 1000) # 毫秒时间戳，用于绕过缓存 url = f"https://weather.example.com/api/sk/{city_code}" headers = { # UA 缺了会被部分站点直接判定为非浏览器流量 "User-Agent": ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/120.0.0.0 Safari/537.36"), # Referer 表示“我是从站点页面点进来的”，很多接口会校验 "Referer": "https://weather.example.com/", "Accept": "application/json, text/javascript, */*; q=0.01", } params = {"_": ts} resp = requests.get(url, headers=headers, params=params, timeout=timeout) resp.raise_for_status() # 4xx/5xx 直接抛，别把错误页当数据 resp.encoding = resp.apparent_encoding # 有的返回 utf-8，有的返回 gbk return resp.text

逻辑说明：时间戳参数保证每次请求的 URL 唯一，避免中间层返回缓存；raise_for_status()把状态码异常提前暴露，否则后面解析会拿到一段 HTML 错误页，报错信息完全对不上；编码交给apparent_encoding自动判断，比硬写utf-8稳。

参数说明：timeout=10是连接加读取的总超时，城市多的时候别设太大，否则一个卡住的请求会拖垮整批；headers里 UA 和 Referer 是两个硬门槛，缺 UA 常见结果是 403，缺 Referer 常见结果是返回空壳或者默认城市的天气；params里的_只是缓存破坏参数，值对不对无所谓，但必须存在。

如果实测发现必须带 Cookie，别急着把浏览器里那一串复制死。更稳的做法是先用一个会话对象访问一次首页，让服务端把 Cookie 下发到requests.Session()里，后续请求复用同一个 session。这样 Cookie 过期时下一次会话会自然刷新，不用手工替换。

2.3 城市编码：9 位编码怎么拆、批量列表从哪来

气象类站点的城市编码基本是一套 9 位数字体系，结构上分三段：前三位标识省级区域，中间三位标识地级区域，后三位标识县级区域。实际抓取时不需要理解每一段的含义，只要保证拿到的编码是站点认可的值即可。编码错了的典型表现是接口返回 200，但内容里城市名跟你预期的对不上，或者温度字段直接为空。

批量城市的来源有两个：一是页面上那个城市选择下拉框，展开后 DOM 里通常带着>import csv def load_city_codes(path: str = "cities.csv") -> list[tuple[str, str]]: """读取城市编码表，返回 [(code, name), ...] CSV 至少包含 code、name 两列，允许有表头 """ result = [] with open(path, newline="", encoding="utf-8-sig") as f: reader = csv.DictReader(f) for row in reader: code = (row.get("code") or "").strip() name = (row.get("name") or "").strip() if len(code) == 9 and code.isdigit(): # 只收合法长度的纯数字编码 result.append((code, name)) return result

逻辑说明：用utf-8-sig读取是为了兼容用表格软件另存出来的 CSV，那种文件开头常带一个不可见 BOM，用普通utf-8读会让第一列的列名变成\ufeffcode，然后所有行都取不到值。长度和纯数字校验是廉价但有效的脏数据过滤。

参数说明：path指向编码表；函数返回列表而不是字典，是因为有的站点允许同一城市名对应多个编码（比如城区和郊区），字典会覆盖。真要按名字查，落地时再单独建索引。

注意：编码表不要写死在代码里。站点调整行政区划或者新增站点时，改 CSV 比改代码安全得多，也不会因为一次改动把历史数据的主键搞乱。

2.4 先看清楚响应结构再写解析

很多人拿到响应文本就直接json.loads，然后被Expecting value: line 1 column 1打回来。原因就是响应体并不是纯 JSON。写解析之前，先把原始响应打印出来看前两百个字符，确认它到底是纯 JSON、JSONP 包裹、还是赋值语句。

这张表值得先抄下来。天气数据最恶心的地方不是拿不到，而是字段格式不统一：同一个接口不同城市返回的类型都可能不一样，有的城市温度是"23"，有的带单位。解析层必须按“字符串进、清洗后出”的思路写，不能假设类型。

3. 解析与入库：JSON 拍平、HTML 兜底、SQLite 增量

拿到原始文本只是原料，真正能用的是结构化的一行记录。这一层要解决三件事：把嵌套结构拍平、把脏值洗干净、把重复抓取写成幂等。

3.1 剥掉 JSONP 外壳，再做字段拍平

先处理外壳。如果响应是var dataSK = {...};这种形式，用正则把花括号部分抠出来再交给json.loads。这一步不能省，也不能用字符串切片硬截，因为不同接口的前缀长度不一样。

import json import re def strip_jsonp(raw: str) -> str: """从 var xxx = {...}; 这类响应里抠出 JSON 主体""" text = raw.strip() if text.startswith("{"): return text # 已经是纯 JSON，直接返回 m = re.search(r"=\s*(\{.*\})", text, re.S) return m.group(1) if m else text def parse_realtime(raw: str, city_code: str) -> dict: payload = json.loads(strip_jsonp(raw)) data = payload.get("weatherinfo") or payload # 有的接口多包一层 return { "city_code": city_code, "temp": to_float(data.get("temp")), "humidity": to_float(str(data.get("SD", "")).rstrip("%")), "wind_dir": (data.get("WD") or "").strip(), "wind_level": (data.get("WS") or "").strip(), "obs_time": (data.get("time") or "").strip(), } def to_float(v) -> float | None: """把 '23'、'23℃'、'' 统一转成 float 或 None""" if v is None: return None s = re.sub(r"[^\d.\-]", "", str(v)) try: return float(s) except ValueError: return None

逻辑说明：strip_jsonp先判断是不是纯 JSON，是就不动，避免正则误伤；payload.get("weatherinfo") or payload兼容“字段直接平铺”和“多包一层”两种返回。to_float统一清洗，把所有非数字字符剔掉再转，转不了返回None而不是抛异常——单条脏数据不应该让整批任务挂掉。

参数说明：返回的字典键名建议跟数据库列名保持一致，能省掉一层映射。obs_time只保留原始时分字符串，日期在入库时补，不要在解析层拼日期，否则跨零点的那条数据会归到错误的一天。

提示：解析函数不要顺手做写库。分成“抓取 → 解析 → 入库”三段，任何一段出问题都能单独重跑，不用重新请求一遍接口。

3.2 接口不可用时，用 HTML 表格兜底

再稳的接口也有挂的时候。真正要长期跑的服务，得准备一条 HTML 兜底路径：访问城市详情页，解析页面里的观测表格。这条路拿到的字段少一些，但胜在不依赖 XHR 接口。

from bs4 import BeautifulSoup def parse_html_table(html: str) -> dict: """从详情页表格里抽实时观测，字段缺失返回 None""" soup = BeautifulSoup(html, "lxml") table = soup.find("table", class_="b") # 以实际 class 为准 if not table: return {} out = {} for tr in table.find_all("tr"): cells = [td.get_text(strip=True) for td in tr.find_all("td")] if len(cells) >= 2: out[cells[0]] = cells[1] # 左列是字段名，右列是值 return out

逻辑说明：用lxml解析器比默认的html.parser快，容错也更好；按“左列字段名、右列值”的方式遍历，比按下标定位td更抗页面微调，只要表格的字段顺序没大改就还能用。

参数说明：class_="b"只是示意，实际要用开发者工具确认表格的 class 或 id；get_text(strip=True)去掉单元格里的空白和换行。兜底路径拿到的字段名是中文，写库前要做一次中文到列名的映射，映射表固定下来，别在循环里现拼。

3.3 SQLite 建表与幂等写入：主键选错等于白干

存储我一般选 SQLite，单文件、零依赖、够用。关键在主键设计——用自增 ID 做主键，重复抓取就会把表撑成两倍；正确做法是把“城市编码 + 观测日期 + 观测时刻”做成联合唯一键，写入用 upsert。

import sqlite3 def init_db(db_path: str = "weather.db") -> sqlite3.Connection: conn = sqlite3.connect(db_path) conn.execute(""" CREATE TABLE IF NOT EXISTS weather_obs ( city_code TEXT NOT NULL, obs_date TEXT NOT NULL, obs_time TEXT NOT NULL, temp REAL, humidity REAL, wind_dir TEXT, wind_level TEXT, updated_at TEXT DEFAULT (datetime('now', 'localtime')), PRIMARY KEY (city_code, obs_date, obs_time) ) """) conn.execute("CREATE INDEX IF NOT EXISTS idx_obs_date ON weather_obs(obs_date)") return conn def upsert_obs(conn: sqlite3.Connection, row: dict, obs_date: str) -> None: conn.execute(""" INSERT INTO weather_obs (city_code, obs_date, obs_time, temp, humidity, wind_dir, wind_level) VALUES (:city_code, :obs_date, :obs_time, :temp, :humidity, :wind_dir, :wind_level) ON CONFLICT(city_code, obs_date, obs_time) DO UPDATE SET temp = excluded.temp, humidity = excluded.humidity, wind_dir = excluded.wind_dir, wind_level = excluded.wind_level, updated_at = datetime('now', 'localtime') """, {**row, "obs_date": obs_date}) conn.commit()

逻辑说明：联合主键保证同一城市同一时刻只有一条记录；ON CONFLICT ... DO UPDATE让“重跑”变成“修正”，晚到的修正值会覆盖早先的不准数据，比先删后插安全。单独给obs_date建索引，是因为查历史时基本都按日期范围扫。

参数说明：obs_date由调用方按抓取时的本地日期生成，跨零点抓取要注意用接口返回的时刻判断归属日，不要用当前时间一刀切。conn.commit()每条提交一次，城市量上百时建议改成每批提交一次，事务开销能降一个量级。

4. 调度与自动化：把「跑通一次」变成「跑一个月」

单次抓取跑通不算本事，能稳定跑一个月才算。调度层要解决三个现实问题：什么时候抓、失败了怎么办、进程怎么活。

4.1 抓取窗口与增量策略

实况数据一般每小时更新一次，整点后几分钟数据才落库。我的经验是整点后第 5 分钟开始抓，连续抓 20 分钟，中间每轮间隔几十秒。这样即使某一轮失败，后面几轮还有机会补上，不至于留下整点空缺。

增量策略不要用“查数据库最后一条时间”来判断，那个逻辑在城市多、抓取乱序的情况下会漏。更稳的做法是每次抓当前时刻前后两个整点窗口的数据，靠 upsert 去重，写多几次没有副作用，漏了就真没了。

注意：不要把所有城市卡在同一秒发出去。几百个城市瞬间并发，触发频控是小事，被临时封掉出口才是大事。

4.2 限速、重试与失败落盘

重试要带退避，不要固定间隔连打。失败的任务必须落盘，不能只打日志——日志会被轮转掉，落盘的失败清单可以第二天单独补跑。

import json import random import time from pathlib import Path FAIL_LOG = Path("failed_tasks.jsonl") def fetch_with_retry(city_code: str, retries: int = 3, base: float = 1.5) -> str | None: """失败重试，仍失败则落盘并返回 None""" for attempt in range(1, retries + 1): try: return fetch_realtime(city_code) except Exception as exc: if attempt == retries: with FAIL_LOG.open("a", encoding="utf-8") as f: f.write(json.dumps( {"city": city_code, "err": str(exc), "ts": time.time()}, ensure_ascii=False) + "\n") return None # 指数退避 + 随机抖动，避免多城市同步重试 time.sleep(base ** attempt + random.uniform(0, 0.5)) return None

逻辑说明：base ** attempt是指数退避，第一次等约 1.5 秒，第二次约 2.25 秒；加随机抖动是为了打散多城市的重试节奏，防止它们在同一秒集体重发再次触发限流。最后一次失败写 JSONL，一行一个任务，方便用脚本批量重放。

参数说明：retries=3对偶发网络抖动够用，接口整体故障时重试再多也没意义，交给下一轮调度更划算。FAIL_LOG建议按天切分，文件大了重放脚本读取会变慢。

4.3 APScheduler 常驻进程与 crontab 两种挂法

小规模、单机跑，我倾向用 APScheduler 起一个常驻进程，逻辑都在 Python 里，调试方便。

from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger def job_hourly(): for code, _name in load_city_codes("cities.csv"): raw = fetch_with_retry(code) if raw: row = parse_realtime(raw, code) upsert_obs(conn, row, obs_date=time.strftime("%Y-%m-%d")) time.sleep(random.uniform(0.8, 1.6)) # 城市之间限速 sched = BlockingScheduler(timezone="Asia/Shanghai") sched.add_job(job_hourly, CronTrigger(minute="5-25/10")) # 每小时 5/15/25 分 sched.start()

逻辑说明：CronTrigger(minute="5-25/10")表示每小时的第 5、15、25 分钟各跑一次，三轮补抓；城市之间sleep随机 0.8 到 1.6 秒，几百个城市跑完也就十几分钟，不会形成突发流量。

参数说明：timezone必须显式指定，容器里默认是 UTC，不写会导致抓取窗口整体偏移八小时，这个坑非常隐蔽。如果机器上已经有 cron 体系，也可以直接写 crontab 调脚本，省一个常驻进程，代价是状态和日志要自己管。

5. 避坑排查：五个让我白跑一整晚的问题

5.1 请求层：状态码 200，内容是壳

现象：脚本不报错，解析出来全是None，抓了一晚上一条有效数据都没有。 原因：接口对请求头做了校验，缺 Referer 或者 UA 不匹配时返回的不是 403，而是 200 加一个结构相同但字段全空的模板。 解决：解析前先做一次内容有效性判断，比如温度字段为空就直接判定本轮失败并记录原始响应前 300 字符。定位时把resp.request.headers打出来，跟浏览器里那条请求逐项对比，差异一眼就能看出来。

5.2 编码层：中文城市名变成问号

现象：入库后城市名和风向字段显示为????或乱码。 原因：响应是 gbk 编码，而 requests 在没有显式声明时按 ISO-8859-1 猜，apparent_encoding在短文本上也会猜错。 解决：先看响应头里的Content-Type，有charset就按它来；没有就手动试gbk，判断标准是中文能正常显示。我在解析函数入口加了一句长度判断：如果解出来的字符串里有连续问号，换编码再解一次。

5.3 解析层：字段时有时无，float 直接炸

现象：批量跑的时候偶尔抛ValueError: could not convert string to float。 原因：个别城市返回的字段是空串或者带单位，甚至整个键都不存在，代码里直接float(data["temp"])就会中断整批。 解决：所有数值字段统一走清洗函数，取不到就落None，不要用try/except包住整个循环——那样你永远不知道是哪条数据出的问题。清洗函数里保留原始字符串到日志，方便回头核对。

5.4 存储层：重复跑一次，表里多出一倍

现象：补跑当天的任务之后，同一个城市同一时刻出现了两条记录。 原因：表用的是自增主键，没有对业务字段建唯一约束，INSERT每次都成功。 解决：改成联合主键加ON CONFLICT DO UPDATE。已经建好的表需要迁移：新建一张正确结构的表，把旧数据按业务键去重后导进去，再改名替换。迁移前记得备份.db文件，这类操作没有后悔药。

5.5 调度层：进程还活着，数据不更新

现象：ps能看到进程，日志停在昨天，数据库没有新数据。 原因：某个请求卡在无超时的连接上，BlockingScheduler的单线程执行器被占死，后续任务全部排队。 解决：所有网络请求强制带timeout；把执行器改成线程池并设max_instances=1、coalesce=True，任务堆积时只跑最新一次，不补跑历史积压。再加一个心跳：每次任务结束写一条时间戳到单独的表，超过两小时没更新就告警。

6. 数据校验与长期可维护：三个自检脚本

跑了几个月之后我发现，真正让这套东西活下来的不是抓取代码，而是校验。下面这张表是我现在每次上线前都会过一遍的规则。

落成一个可执行的检查脚本，逻辑很简单，但能把“数据悄悄错了两周没人发现”这类问题挡在门外。

import sqlite3 def sanity_check(db_path: str = "weather.db", report_date: str | None = None): conn = sqlite3.connect(db_path) day = report_date or __import__("time").strftime("%Y-%m-%d") rows = conn.execute(""" SELECT city_code, obs_time, temp, humidity FROM weather_obs WHERE obs_date = ? ORDER BY city_code, obs_time """, (day,)).fetchall() bad_temp = [r for r in rows if r[2] is not None and not (-60 <= r[2] <= 60)] bad_hum = [r for r in rows if r[3] is not None and not (0 <= r[3] <= 100)] # 按城市检查时刻间隔，缺口超过 90 分钟视为断档 gaps = [] prev = {} for code, t, *_ in rows: if code in prev: h1, m1 = map(int, prev[code].split(":")) h2, m2 = map(int, t.split(":")) if (h2 * 60 + m2) - (h1 * 60 + m1) > 90: gaps.append((code, prev[code], t)) prev[code] = t return {"bad_temp": len(bad_temp), "bad_hum": len(bad_hum), "gaps": gaps[:20]}

逻辑说明：一次扫描当天全部记录，分别统计越界值和时刻断档。断档检查只看相邻两条的间隔，不要求整点严格对齐——气象站本身也有延迟上报的情况，卡太死会天天误报。

参数说明：report_date不传就查当天；gaps只返回前 20 条，避免一个城市整天没数据时把返回撑爆。这个脚本我挂在每天凌晨跑，结果写进一张qc_log表，连续两天异常就人工介入。

说个血泪教训：早期我图省事，抓完直接入库，没有任何校验，结果有一次接口字段顺序变了，温度写进了湿度列，看板上出现了 96% 的“气温”，两天后才被业务方发现。从那以后我每次改解析代码，都强制先跑一遍校验脚本再合入，不管改动多小。数据这类东西，错得安静，比错得响亮可怕得多。希望帮到你。

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