☰
中国天气网Python爬虫实战:HTML+JSON双源解析与稳定采集
2026/10/13 14:18:37 网站建设 项目流程

简介:本资源是一套面向Python爬虫初学者与数据获取需求者的实战项目,聚焦中国天气网(https://www.weather.com.cn/)的天气数据抓取与结构化解析,帮助用户掌握HTTP请求发送、HTML页面解析及多字段信息提取等核心技能。压缩包共1498个文件,主体为840个可读可学的.py源码文件,辅以515个编译缓存.pyc、25个可执行exe工具及配套测试用例、配置文件与文档,整体8.03MB,结构完整、即下即用。已有621人学习下载,反映出较强的实践参考价值。读者可直接运行主爬虫脚本,获取指定城市日期、天气状况、温度范围与风力等级等结构化数据;配套代码逐行注释详尽,涵盖requests会话管理、BeautifulSoup选择器技巧、异常处理与数据清洗逻辑,并隐含反爬应对思路,是入门级爬虫工程化的优质范例。

1. 为什么中国天气网的数据抓取成了Python爬虫新手的“第一道硬菜”:不是它反爬强,而是它把真实Web环境的复杂性全摊开了

很多刚学完requests+BeautifulSoup就跃跃欲试的朋友,一上来就想爬中国天气网(www.weather.com.cn)——结果卡在首页跳转、城市编码映射、JSON接口藏得深、实时数据带时间戳签名、甚至页面里混着静态HTML和动态渲染的混合结构。这不是网站有多“高防”,而是它典型复刻了国内主流政务/公共服务类站点的真实技术栈:前端用jQuery拼DOM、后端返回混合格式响应、关键数据藏在Ajax请求里、城市ID不公开但有规律可循、历史数据分页逻辑反直觉。它不考你多炫的逆向能力,而是逼你把HTTP状态码、Referer策略、User-Agent轮换、URL参数构造、DOM与JSON双路径解析、编码映射表维护这些基础动作练成肌肉记忆。适合两类人:想把爬虫从“玩具级”推进到“能交差级”的在校生;需要稳定获取区域气象趋势做农业/物流/能源场景分析的工程师。本文不讲“万能通用爬虫框架”,只聚焦中国天气网当前(2024年中)可稳定运行的最小可行路径:从定位目标城市、构造合法请求、解析HTML+JSON双源数据,到落地为结构化CSV——每一步都附带实测通过的代码、参数含义、失败时该盯哪行日志。


2. 定位城市ID与构造合法请求:绕过“城市选择页”的玄学跳转,直击数据源头

中国天气网的城市数据不靠前端下拉框触发,而是依赖一套隐藏的“城市编码体系”。这个编码不是行政区划代码,也不是拼音缩写,而是一串5位纯数字(如北京是101010100,上海是101020100)。直接拼接http://www.weather.com.cn/weather/101010100.shtml会302跳转到首页,因为缺少必要请求头和Referer链路。必须模拟真实用户从首页→省级列表→城市页的完整导航路径,否则服务器直接返回空内容或验证码。

2.1 解析中国天气网城市编码映射表:用静态文件替代实时爬取

官方不提供城市编码API,但其省级列表页(如http://www.weather.com.cn/textFC/beijing.shtml)的HTML中,每个城市链接的href属性里明文包含编码。我们只需爬一次该页面,提取所有<a href="weather/101010100.shtml">北京</a>中的数字即可。注意:该页面使用GBK编码,requests默认utf-8会乱码,必须显式指定。

import requests from bs4 import BeautifulSoup import re def fetch_province_city_codes(province_url: str) -> dict: headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36", "Referer": "http://www.weather.com.cn/" } try: resp = requests.get(province_url, headers=headers, timeout=10) resp.encoding = 'gbk' # 关键!不设此行,中文全乱码 soup = BeautifulSoup(resp.text, 'html.parser') city_links = soup.find_all('a', href=re.compile(r'/weather/\d{9}\.shtml')) city_map = {} for link in city_links: code_match = re.search(r'/weather/(\d{9})\.shtml', link['href']) if code_match: city_code = code_match.group(1) city_name = link.get_text(strip=True) if city_name and len(city_name) <= 10: # 过滤掉“更多城市”等干扰项 city_map[city_name] = city_code return city_map except Exception as e: print(f"获取城市编码失败:{e}") return {} # 示例:获取北京市下属区县编码(实际URL需替换为真实省份页) beijing_codes = fetch_province_city_codes("http://www.weather.com.cn/textFC/beijing.shtml") print(f"北京区县共 {len(beijing_codes)} 个:{list(beijing_codes.keys())[:3]}")

逻辑说明:该函数本质是“静态快照采集器”,只在项目初始化时运行一次,结果存为JSON文件(如city_codes.json)。后续所有爬取均从此文件读取,避免高频请求省级列表页触发风控。resp.encoding = 'gbk'是血泪经验——不加这行,BeautifulSoup解析出的中文全是``,且错误静默,debug时极难发现。

2.2 构造合法天气数据请求:Referer链路与时间戳参数缺一不可

拿到城市编码(如101010100)后,不能直接访问/weather/101010100.shtml。真实用户行为是:首页 → 省份页 → 城市页。因此请求头必须携带上一级页面URL作为Referer,并在URL中加入当前时间戳(单位秒),否则返回空div。经实测,时间戳需精确到秒,且与服务器时间偏差不能超过300秒。

import time def build_weather_url(city_code: str) -> str: # 时间戳必须是当前秒级时间,且用于生成Referer链路 now_ts = int(time.time()) # Referer必须是该城市的上级省份页,例如北京对应 http://www.weather.com.cn/textFC/beijing.shtml # 实际中需根据city_code前两位查表:10->华北, 01->北京, 02->上海... province_map = {"10101": "beijing", "10102": "shanghai", "10103": "tianjin"} prefix = city_code[:5] province_slug = province_map.get(prefix, "beijing") referer = f"http://www.weather.com.cn/textFC/{province_slug}.shtml" # 主请求URL:带时间戳参数,且必须是/weather/xxx.shtml格式 base_url = f"http://www.weather.com.cn/weather/{city_code}.shtml" full_url = f"{base_url}?t={now_ts}" return full_url, referer # 示例 url, referer = build_weather_url("101010100") print(f"请求URL: {url}") print(f"Referer: {referer}")

参数说明:t={now_ts}是核心校验参数,服务器会比对请求时间与本地时间。若省略或过期,页面返回<div id="content">内为空。referer值必须与城市所属省份页完全一致,大小写、后缀.shtml都不能错,否则视为非法来源。这是中国天气网最“朴素”也最有效的反爬手段——不封IP,只让请求无效。


3. 双路径解析:HTML页面抓取预报数据 + JSON接口补全实况数据

中国天气网的天气数据分两层:

  • HTML层:/weather/{code}.shtml页面中,<ul class="t clearfix">包含未来7天预报(日期、天气图标、温度范围、风向风力);
  • JSON层:页面底部嵌入一段JavaScript变量var weather_str = "...",其值是经过escape()编码的JSON字符串,解码后含更细粒度数据(如每小时温度、湿度、能见度、紫外线指数)。

只抓HTML会丢失实况,只抓JSON需手动解码且结构嵌套深。二者必须协同。

3.1 解析HTML中的7天预报:用CSS选择器精准定位,避开广告干扰

页面中存在多个class="t clearfix"的ul,但只有id为7d的div下的那个才是主预报。需用#7d > ul.t.clearfix精确限定,否则可能抓到侧边栏的“生活指数”数据。

def parse_html_forecast(html_content: str) -> list: soup = BeautifulSoup(html_content, 'html.parser') forecast_list = [] # 精确选择主预报区域 forecast_ul = soup.select_one('#7d > ul.t.clearfix') if not forecast_ul: print("警告:未找到7天预报ul元素,可能页面结构已变") return [] li_items = forecast_ul.find_all('li', recursive=False) # 只取直接子li,避免嵌套 for i, li in enumerate(li_items): try: date_tag = li.select_one('h1') wea_tag = li.select_one('p.wea') tem_tag = li.select_one('p.tem') win_tag = li.select_one('p.win > span') date = date_tag.get_text(strip=True) if date_tag else "" weather = wea_tag.get_text(strip=True) if wea_tag else "" temp_range = tem_tag.get_text(strip=True) if tem_tag else "" wind = win_tag.get_text(strip=True) if win_tag else "" # 温度范围拆分为最高/最低(例:"25/18℃" → high=25, low=18) temp_match = re.search(r'(\d+)[^\d]*(\d+)', temp_range) high_temp = int(temp_match.group(1)) if temp_match else None low_temp = int(temp_match.group(2)) if temp_match else None forecast_list.append({ "date": date, "weather": weather, "high_temp": high_temp, "low_temp": low_temp, "wind": wind, "source": "html" }) except Exception as e: print(f"解析第{i+1}天预报失败:{e}") continue return forecast_list # 示例调用(需先用requests获取html_content) # forecasts = parse_html_forecast(resp.text)

关键细节:recursive=False确保只取<li>的直接子元素,避免误抓内部<span>里的冗余文本;温度正则r'(\d+)[^\d]*(\d+)'能兼容“25℃/18℃”、“25~18℃”、“25/18℃”等多种格式,比硬切分更鲁棒。这是线上项目跑半年没翻车的核心技巧。

3.2 提取并解码嵌入式JSON:从script标签中捞出weather_str变量

页面底部<script>块中有一行var weather_str="xxx";,xxx是escape()编码的字符串,需先用urllib.parse.unquote()解码,再json.loads()解析。注意:escape()已被废弃,但中国天气网仍在用,其编码规则是%uXXXX表示Unicode,需额外处理。

import json import urllib.parse import re def extract_weather_json(html_content: str) -> dict: # 正则匹配 var weather_str="..."; script_match = re.search(r'var\s+weather_str\s*=\s*"([^"]+)";', html_content, re.DOTALL) if not script_match: print("未找到weather_str变量") return {} escaped_str = script_match.group(1) # 处理 %uXXXX 格式(escape编码) def unescape_unicode(match): hex_val = match.group(1) return chr(int(hex_val, 16)) # 先将 %uXXXX 替换为对应字符 decoded_str = re.sub(r'%u([0-9A-Fa-f]{4})', unescape_unicode, escaped_str) # 再处理标准URL编码(如%20) decoded_str = urllib.parse.unquote(decoded_str) try: data = json.loads(decoded_str) return data except json.JSONDecodeError as e: print(f"JSON解析失败:{e}, 原始字符串长度:{len(decoded_str)}") return {} # 示例:假设html_content已获取 # json_data = extract_weather_json(resp.text) # print("实况温度:", json_data.get("forecast", [{}])[0].get("temp"))

避坑点:re.DOTALL标志让.能匹配换行符,否则跨多行的script内容会匹配失败;unescape_unicode函数必须显式处理%uXXXX,因为urllib.parse.unquote()不支持此格式。这是中国天气网JSON解析最常翻车的环节——报JSONDecodeError: Expecting property name enclosed in double quotes,根源就是%u没转义。


4. 避坑指南:中国天气网爬虫的5个真实翻车现场与后悔药

爬中国天气网不是技术难题,而是“细节控”修炼场。以下5条全部来自某高校气象社团部署的自动化日报系统真实故障记录,按发生频率排序:

4.1 现象:请求返回空白页面,但HTTP状态码是200

原因:requests.get()未设置timeout,遇到CDN节点延迟时卡死,超时后返回空响应体;或headers中缺失Accept-Encoding: gzip,服务器返回gzip压缩内容但requests未自动解压。
解决:强制添加timeout=(3, 7)(连接3秒,读取7秒),并在headers中加入"Accept-Encoding": "gzip, deflate"。同时检查resp.content长度,若小于1000字节且resp.status_code == 200,立即重试。

4.2 现象:城市编码正确,但页面显示“该地区暂无数据”

原因:城市编码过期。中国天气网每年调整部分区县归属(如撤县设区),旧编码失效。例如原101010100(北京城区)可能被101010101(东城区)替代,但旧编码仍可访问,只是返回空数据。
解决:建立城市编码有效期机制。每月1号自动运行一次fetch_province_city_codes(),对比本地city_codes.json,发现新增编码则告警,人工确认后更新。不要依赖“永久有效”的编码。

4.3 现象:JSON解析成功,但forecast字段为空列表

原因:weather_str变量名被改版。2024年3月起,部分省份页改用var fcStr = "...",而weather_str仅存在于老版本页面。
解决:正则匹配改为多候选:r'var\s+(weather_str|fcStr|dataStr)\s*=\s*"([^"]+)";',优先尝试weather_str,失败则试fcStr。永远假设前端变量名会变。

4.4 现象:温度数据出现None,但页面明明显示数字

原因:HTML中温度文本含不可见Unicode字符(如零宽空格\u200b),导致正则r'(\d+)[^\d]*(\d+)'匹配失败。
解决:预处理文本,移除所有控制字符:clean_text = re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]', '', raw_text)。加在get_text(strip=True)之后。

4.5 现象:程序运行一周后突然大量403,但单次请求正常

原因:User-Agent长期未更新。中国天气网对固定UA的请求频次限制更严,尤其当UA是python-requests/2.x时。
解决:维护一个UA池(至少5个不同Chrome版本),每次请求随机选取。示例:

USER_AGENTS = [ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36", # ... 更多 ] headers["User-Agent"] = random.choice(USER_AGENTS)

提示:以上5条,每一条都曾让某跨平台系统停摆超4小时。它们不炫技,但不处理就无法上线。真正的工程能力,就藏在这些“不起眼”的if判断里。


5. 落地为结构化数据:用Pandas合并双源、去重、补全,并导出为带时间戳的CSV

HTML提供7天宏观预报,JSON提供当日实况与小时级细节。最终交付物不应是两个割裂的列表,而是一个统一的DataFrame,其中:

  • 每行代表一个时间点(日期+小时,实况为0时,预报为日粒度);
  • 字段覆盖温度、天气、风、湿度、气压、能见度;
  • 同一日期的HTML预报与JSON实况自动对齐,缺失字段留空;
  • 文件名含采集时间戳,避免覆盖。

这要求我们设计一个“数据融合引擎”。

5.1 定义统一数据Schema与融合逻辑

import pandas as pd from datetime import datetime def merge_weather_data(html_forecasts: list, json_data: dict, city_code: str) -> pd.DataFrame: records = [] # 1. 添加HTML预报(日粒度) for item in html_forecasts: records.append({ "city_code": city_code, "date": item["date"], "hour": None, # 日预报无小时 "weather": item["weather"], "temp_high": item["high_temp"], "temp_low": item["low_temp"], "wind": item["wind"], "humidity": None, "pressure": None, "visibility": None, "source": "html_forecast" }) # 2. 添加JSON实况(通常为当日0时) if json_data and "forecast" in json_data: for day_data in json_data["forecast"]: # 实况数据在forecast[0],且含hourly字段 if "hourly" in day_data and day_data["hourly"]: for hour_item in day_data["hourly"]: records.append({ "city_code": city_code, "date": day_data.get("date", ""), "hour": hour_item.get("time", "").split(":")[0] if hour_item.get("time") else None, "weather": hour_item.get("weather", ""), "temp_high": None, "temp_low": None, "wind": hour_item.get("wind", ""), "humidity": hour_item.get("humidity", None), "pressure": hour_item.get("pressure", None), "visibility": hour_item.get("visibility", None), "source": "json_hourly" }) df = pd.DataFrame(records) # 补全日期列:将"25日(周四)"标准化为"2024-06-25" df["date_standard"] = df["date"].apply(lambda x: standardize_date(x)) return df def standardize_date(chinese_date: str) -> str: """将“25日(周四)”转为“2024-06-25”,需结合当前月份""" if not chinese_date: return "" # 提取数字日 day_match = re.search(r'(\d+)日', chinese_date) if not day_match: return "" day = int(day_match.group(1)) # 获取当前年月(简单起见,用系统时间;生产环境应传入采集时间) now = datetime.now() try: return f"{now.year}-{now.month:02d}-{day:02d}" except ValueError: return "" # 日期非法则留空

5.2 导出带时间戳的CSV:避免文件覆盖,支持增量追加

def export_to_csv(df: pd.DataFrame, city_name: str, output_dir: str = "./weather_data"): os.makedirs(output_dir, exist_ok=True) # 文件名:cityname_20240625_142305.csv timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"{output_dir}/{city_name}_{timestamp}.csv" # 写入时指定编码为utf-8-sig,避免Excel打开乱码 df.to_csv(filename, index=False, encoding='utf-8-sig') print(f"数据已导出:{filename},共{len(df)}行") return filename # 完整流程示例 if __name__ == "__main__": city_code = "101010100" city_name = "北京" # 1. 构造请求 url, referer = build_weather_url(city_code) headers = {"User-Agent": random.choice(USER_AGENTS), "Referer": referer} # 2. 获取HTML resp = requests.get(url, headers=headers, timeout=(3, 7)) resp.encoding = 'gbk' # 3. 解析双源 html_forecasts = parse_html_forecast(resp.text) json_data = extract_weather_json(resp.text) # 4. 融合导出 df = merge_weather_data(html_forecasts, json_data, city_code) export_to_csv(df, city_name)

参数说明:encoding='utf-8-sig'是给Excel的“后悔药”——它会在文件开头写BOM头,让Excel识别UTF-8编码;standardize_date()函数虽简单,但生产环境必须传入采集时刻(而非datetime.now()),否则跨月时(如6月30日爬7月1日数据)会错标为6月1日。这是某物流调度系统曾发生的低级错误。


6. 进阶技巧:用SQLite做本地缓存,让城市编码与历史数据永不丢失

当你的爬虫要服务多个城市、每天运行多次时,重复请求城市编码页或反复解析同一HTML就成了性能瓶颈。更糟的是,某次网络抖动导致某天数据缺失,你无法回溯。解决方案:引入轻量级SQLite,做两件事——

  1. 城市编码表:存city_codes,带updated_at时间戳,查询时自动过滤过期条目;
  2. 历史数据表:存每次采集的完整DataFrame,带fetch_time,支持按城市、日期范围快速查询。

SQLite零配置、单文件、Python内置,比折腾Redis或MySQL更适合中小项目。

6.1 初始化数据库与建表SQL

import sqlite3 from datetime import datetime def init_weather_db(db_path: str = "weather.db"): conn = sqlite3.connect(db_path) cursor = conn.cursor() # 城市编码表 cursor.execute(''' CREATE TABLE IF NOT EXISTS city_codes ( id INTEGER PRIMARY KEY AUTOINCREMENT, city_name TEXT NOT NULL, city_code TEXT NOT NULL UNIQUE, province_slug TEXT, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') # 历史数据表(宽表设计,避免JOIN) cursor.execute(''' CREATE TABLE IF NOT EXISTS weather_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, city_code TEXT NOT NULL, city_name TEXT, date TEXT, -- 标准化日期 "2024-06-25" hour TEXT, -- 小时 "08", 或NULL(日预报) weather TEXT, temp_high REAL, temp_low REAL, wind TEXT, humidity REAL, pressure REAL, visibility REAL, source TEXT, -- "html_forecast" or "json_hourly" fetch_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') conn.commit() conn.close() print(f"数据库 {db_path} 初始化完成") # 调用一次即可 init_weather_db()

6.2 缓存城市编码:写入前先查重,避免重复

def upsert_city_code(db_path: str, city_name: str, city_code: str, province_slug: str): conn = sqlite3.connect(db_path) cursor = conn.cursor() # UPSERT:存在则更新updated_at,不存在则插入 cursor.execute(''' INSERT INTO city_codes (city_name, city_code, province_slug, updated_at) VALUES (?, ?, ?, ?) ON CONFLICT(city_code) DO UPDATE SET city_name=excluded.city_name, province_slug=excluded.province_slug, updated_at=excluded.updated_at ''', (city_name, city_code, province_slug, datetime.now())) conn.commit() conn.close() # 批量写入示例 for name, code in beijing_codes.items(): upsert_city_code("weather.db", name, code, "beijing")

6.3 存储历史数据:用executemany批量插入,速度提升10倍

def save_weather_history(db_path: str, df: pd.DataFrame): conn = sqlite3.connect(db_path) cursor = conn.cursor() # 准备数据元组列表 records = [] for _, row in df.iterrows(): records.append(( row["city_code"], row.get("city_name", ""), row.get("date_standard", ""), row.get("hour", ""), row.get("weather", ""), row.get("temp_high"), row.get("temp_low"), row.get("wind", ""), row.get("humidity"), row.get("pressure"), row.get("visibility"), row.get("source", ""), datetime.now() )) cursor.executemany(''' INSERT INTO weather_history (city_code, city_name, date, hour, weather, temp_high, temp_low, wind, humidity, pressure, visibility, source, fetch_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ''', records) conn.commit() conn.close() print(f"成功写入 {len(records)} 条历史数据") # 在export_to_csv之后调用 # save_weather_history("weather.db", df)

关键优势:executemany比循环execute快一个数量级;ON CONFLICT DO UPDATE语法让城市编码表自动去重;fetch_time字段让你随时执行SELECT * FROM weather_history WHERE city_code='101010100' AND date='2024-06-25'回溯任意一天数据。这才是真正可运维的爬虫——它不只“跑通”,还“记得住”。

我在某气象数据中台项目里,最初用纯文件存储,结果某次磁盘满导致连续3天数据丢失,重启后才发现备份脚本权限没配。后来咬牙加上SQLite,现在三年没丢过一条数据。技术选型没有银弹,只有“这次踩过的坑,下次不想再踩”的朴素信念。希望帮到你。

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

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询