从零构建专业天气数据爬虫：以天气网为例详解表单提交与模拟查询全流程

一、爬虫项目背景与目标

在数据驱动的时代，天气数据作为基础的环境信息，在农业预测、旅游规划、能源管理、历史事件回溯分析等领域具有重要价值。然而，主流天气网站通常仅提供有限的历史数据免费查询，且往往需要用户手动选择日期并点击查询按钮才能获取。对于需要批量获取长时间序列数据的场景，人工操作显然不可行。因此，开发一个能够自动提交表单、模拟查询行为、精确提取历史天气数据的爬虫，成为数据采集工程师的必备技能。

本文将以中国知名天气数据平台“天气网”（https://www.tianqi.com） 为目标，系统讲解如何利用Python爬虫技术，实现从日期选择、表单提交、响应解析到数据持久化的完整流程。本项目的核心难点在于：处理动态表单参数、维持会话状态、应对反爬虫机制、解析非结构化HTML文本。我们将采用最新稳定的技术栈，包括requests、BeautifulSoup4、pandas以及fake_useragent，提供可直接投入生产的代码。

核心功能点：

1. 自动构造目标城市的天气查询URL

2. 模拟浏览器提交年份和月份参数

3. 解析返回的HTML表格，提取日期、最高温度、最低温度、天气状况

4. 支持多日期范围批量爬取，输出结构化数据（CSV/Excel）

通过本文，您将掌握表单提交类爬虫的标准开发范式，并能轻松扩展到其他类似结构的网站。

目录

一、爬虫项目背景与目标

二、技术选型与环境搭建

2.1 为什么选择这些技术库？

2.2 环境配置步骤

2.3 目标网站分析

2.4 反爬虫机制识别

三、核心爬虫架构设计

3.1 整体流程图

3.2 模块划分

四、详细代码实现

4.1 网络请求模块（带重试和伪装）

4.2 表单参数提取器

4.3 天气数据解析器

4.4 主爬虫逻辑（表单提交模拟）

4.5 数据导出模块

4.6 异常处理与日志增强

五、完整运行示例与参数调优

六、高级优化与生产级改进

6.1 分布式爬取支持（Redis Queue）

6.2 代理IP池集成

6.3 数据清洗进阶

6.4 增量爬取与去重

七、常见问题与解决方案

Q1: 网站返回403 Forbidden

Q2: 表单提交后返回的仍是原页面（未查询成功）

Q3: 数据表格结构动态加载（Ajax）

Q4: 网站使用JavaScript加密参数

八、法律与伦理注意事项

九、扩展：从表单提交到API爬取

十、总结与进一步学习

二、技术选型与环境搭建

2.1 为什么选择这些技术库？

2.2 环境配置步骤

建议使用虚拟环境隔离依赖：

bash

# 创建Python 3.10+虚拟环境 python -m venv weather_spider_env source weather_spider_env/bin/activate # Linux/Mac # 或 weather_spider_env\Scripts\activate # Windows # 安装依赖 pip install requests beautifulsoup4 fake_useragent pandas retrying lxml

验证安装：

python

import sys print(sys.version) import requests print(requests.__version__) # 应输出 2.31.x 及以上

2.3 目标网站分析

目标URL模式（以北京为例）：https://www.tianqi.com/lishi/beijing/{year}{month}.html

• beijing为城市拼音

• {year}{month}为6位数字，如202401代表2024年1月

关键发现：

• 该网站历史数据按月份分页存储，直接访问构造的URL即可获取整月数据，无需额外的POST请求

• 但部分类似网站（如 weather.com）需要提交表单参数。为展示“表单提交”技术点，我们将模拟访问带有月份选择器的实际交互流程——即先发送GET获取页面，解析隐藏的表单token，再POST请求提交查询。

2.4 反爬虫机制识别

使用开发者工具（F12）分析请求特征：

1. User-Agent校验：服务器检测请求头中的UA，非浏览器标识返回403

2. Referer检查：部分页面要求来源页为本站

3. Cookie依赖：首次访问需设置会话Cookie

4. 请求频率限制：同IP短时间内请求过多会触发封禁

5. 动态Token：表单中包含一次性token参数

针对上述机制，我们的应对策略：

• 使用fake_useragent伪造浏览器UA

• 正确设置Referer和Origin头

• 使用requests.Session()自动管理Cookie

• 添加随机延迟（1~3秒）

• 从页面中提取token动态填充表单

三、核心爬虫架构设计

3.1 整体流程图

text

开始 → 用户输入城市拼音 → 生成目标URL → 发起GET请求获取主页面 → 提取表单隐藏参数（token等）→ 构造POST数据（年份、月份）→ 发送POST请求获取结果页面 → 解析HTML表格 → 清洗数据 → 循环下个月 → 最终合并导出CSV

3.2 模块划分

1. NetworkManager：负责HTTP请求，处理重试、头信息、代理

2. FormExtractor：从初始页面解析表单字段

3. DataParser：使用BeautifulSoup解析天气表格

4. DataExporter：将爬取结果保存为CSV/Excel

5. Scheduler：控制请求频率和日期范围循环

四、详细代码实现

4.1 网络请求模块（带重试和伪装）

python

import requests from fake_useragent import UserAgent from retrying import retry import time import random class WeatherHTTPClient: """封装HTTP请求，支持会话保持和自动重试""" def __init__(self, retry_max_attempts=3, timeout=10): self.session = requests.Session() self.timeout = timeout self.retry_attempts = retry_max_attempts self.ua = UserAgent() # 设置默认请求头 self.session.headers.update({ 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8', 'Accept-Language': 'zh-CN,zh;q=0.8,zh-TW;q=0.7,zh-HK;q=0.5,en-US;q=0.3,en;q=0.2', 'Accept-Encoding': 'gzip, deflate, br', 'Connection': 'keep-alive', 'Upgrade-Insecure-Requests': '1', }) def _get_random_headers(self, referer=None): """生成随机User-Agent并可选设置Referer""" headers = { 'User-Agent': self.ua.random, } if referer: headers['Referer'] = referer return headers @retry(stop_max_attempt_number=3, wait_fixed=2000) def get(self, url, params=None, referer=None): """带重试的GET请求""" headers = self._get_random_headers(referer) response = self.session.get( url, params=params, headers=headers, timeout=self.timeout ) response.raise_for_status() # 部分网站使用GBK编码，需自动检测 if 'charset=gbk' in response.text.lower() or 'charset=gb2312' in response.text.lower(): response.encoding = 'gbk' else: response.encoding = response.apparent_encoding return response @retry(stop_max_attempt_number=3, wait_fixed=2000) def post(self, url, data=None, referer=None): """带重试的POST请求""" headers = self._get_random_headers(referer) response = self.session.post( url, data=data, headers=headers, timeout=self.timeout ) response.raise_for_status() response.encoding = response.apparent_encoding return response def close(self): self.session.close()

4.2 表单参数提取器

很多天气网站使用隐藏input传递CSRF token或会话标识。我们需要从HTML中提取这些参数。

python

from bs4 import BeautifulSoup class FormParameterExtractor: """从页面中提取表单提交所需的参数""" @staticmethod def extract_hidden_inputs(html_content, form_id=None): """ 提取表单中的所有隐藏输入字段 :param html_content: HTML字符串 :param form_id: 可选，指定form的id或class :return: dict {name: value} """ soup = BeautifulSoup(html_content, 'lxml') # 定位表单 if form_id: form = soup.find('form', {'id': form_id}) or soup.find('form', {'class': form_id}) else: form = soup.find('form') # 取第一个表单 if not form: return {} hidden_inputs = {} for inp in form.find_all('input', {'type': 'hidden'}): name = inp.get('name') value = inp.get('value', '') if name: hidden_inputs[name] = value # 某些网站使用input的type不是hidden但仍需提交，如submit按钮的name # 可根据实际情况扩展 return hidden_inputs @staticmethod def extract_token(html_content, token_names=None): """ 提取CSRF token，常见名称: csrf_token, _token, authenticity_token """ if token_names is None: token_names = ['csrf_token', '_token', 'authenticity_token', 'csrfmiddlewaretoken'] soup = BeautifulSoup(html_content, 'lxml') for name in token_names: # 查找meta标签 meta = soup.find('meta', {'name': name}) if meta and meta.get('content'): return meta['content'] # 查找隐藏input inp = soup.find('input', {'name': name}) if inp and inp.get('value'): return inp['value'] return None

4.3 天气数据解析器

解析历史天气表格是关键步骤。以天气网为例，每个月的页面中包含一个表格，列通常包括：日期、最高温度、最低温度、天气、风向等。

python

class WeatherDataParser: """从HTML表格中提取天气记录""" @staticmethod def parse_monthly_weather(html_content, year, month): """ 解析月份天气页面 返回: list of dict """ soup = BeautifulSoup(html_content, 'lxml') # 查找天气表格 - 根据目标网站结构调整选择器 # 方式1: 通过class定位 table = soup.find('table', class_='history-table') # 方式2: 如果没有class，找包含'日期'文本的表格 if not table: for t in soup.find_all('table'): if '日期' in t.get_text(): table = t break if not table: raise ValueError("未找到天气数据表格，网站结构可能已变更") # 提取表头 headers = [] thead = table.find('thead') if thead: headers = [th.get_text(strip=True) for th in thead.find_all('th')] else: # 尝试从第一行获取 first_row = table.find('tr') headers = [td.get_text(strip=True) for td in first_row.find_all(['th', 'td'])] # 映射列索引（根据常见列名） col_mapping = {} for idx, col_name in enumerate(headers): if '日期' in col_name or 'date' in col_name.lower(): col_mapping['date'] = idx elif '最高' in col_name or 'max' in col_name.lower(): col_mapping['temp_max'] = idx elif '最低' in col_name or 'min' in col_name.lower(): col_mapping['temp_min'] = idx elif '天气' in col_name or 'condition' in col_name.lower(): col_mapping['condition'] = idx elif '风向' in col_name or 'wind' in col_name.lower(): col_mapping['wind'] = idx # 提取数据行 records = [] tbody = table.find('tbody') or table for row in tbody.find_all('tr'): cells = row.find_all(['td', 'th']) if len(cells) < len(headers): continue row_data = {} # 日期 if 'date' in col_mapping: date_str = cells[col_mapping['date']].get_text(strip=True) # 构造完整日期: 2024-01-01 # 假设页面中日期格式为'01'或'1日' day = ''.join(filter(str.isdigit, date_str)) if day: row_data['date'] = f"{year}-{month:02d}-{int(day):02d}" # 温度提取（去除'℃'等字符） if 'temp_max' in col_mapping: max_temp = cells[col_mapping['temp_max']].get_text(strip=True) row_data['temp_max'] = WeatherDataParser._extract_temperature(max_temp) if 'temp_min' in col_mapping: min_temp = cells[col_mapping['temp_min']].get_text(strip=True) row_data['temp_min'] = WeatherDataParser._extract_temperature(min_temp) if 'condition' in col_mapping: row_data['weather_condition'] = cells[col_mapping['condition']].get_text(strip=True) if 'wind' in col_mapping: row_data['wind'] = cells[col_mapping['wind']].get_text(strip=True) # 只有包含有效日期才添加 if row_data.get('date'): records.append(row_data) return records @staticmethod def _extract_temperature(temp_str): """从类似'15℃'或'-5°C'的字符串中提取整数温度""" import re match = re.search(r'(-?\d+)', temp_str) if match: return int(match.group(1)) return None

4.4 主爬虫逻辑（表单提交模拟）

以下代码演示完整流程：访问包含日期选择器的页面 → 提取表单参数 → 提交POST请求 → 解析响应。

python

import logging from datetime import datetime, timedelta import time import random class WeatherHistorySpider: def __init__(self, city_pinyin): self.city = city_pinyin self.http_client = WeatherHTTPClient() self.form_extractor = FormParameterExtractor() self.parser = WeatherDataParser() self.base_url = f"https://www.tianqi.com/lishi/{city_pinyin}/" self.query_url = "https://www.tianqi.com/lishi/{}/query" # 假设的提交端点 logging.basicConfig(level=logging.INFO) self.logger = logging.getLogger(__name__) def get_initial_page(self): """获取包含查询表单的初始页面""" self.logger.info(f"获取初始页面: {self.base_url}") response = self.http_client.get(self.base_url) return response.text def extract_form_parameters(self, html): """从初始页面解析表单所需的所有参数""" params = {} # 提取隐藏字段 hidden = self.form_extractor.extract_hidden_inputs(html) params.update(hidden) # 提取token token = self.form_extractor.extract_token(html) if token: params['csrf_token'] = token # 有些网站需要提交__VIEWSTATE等ASP.NET参数 viewstate = self.form_extractor.extract_hidden_inputs(html, form_id='aspnetForm') params.update(viewstate) return params def submit_query(self, year, month, form_params): """ 提交表单查询指定月份的天气 注意：这里可能需要根据实际网站的请求方式调整（GET或POST） """ # 构造POST数据 post_data = { 'year': str(year), 'month': str(month), **form_params } # 提交URL（部分网站直接使用原页面URL处理POST） submit_url = f"{self.base_url}query" self.logger.info(f"查询 {year}-{month:02d}") response = self.http_client.post(submit_url, data=post_data, referer=self.base_url) return response.text def crawl_month(self, year, month): """爬取单月数据（组合方式：直接构造URL模式）""" # 方法A: 若网站支持直接按月URL访问，则使用此方式（更稳定） direct_url = f"https://www.tianqi.com/lishi/{self.city}/{year}{month:02d}.html" self.logger.info(f"直接访问: {direct_url}") try: resp = self.http_client.get(direct_url, referer=self.base_url) records = self.parser.parse_monthly_weather(resp.text, year, month) return records except Exception as e: self.logger.error(f"直接访问失败: {e}") # 回退到表单提交方式 return self.crawl_month_via_form(year, month) def crawl_month_via_form(self, year, month): """方法B: 模拟表单提交查询""" # 第一步：获取初始页面 init_html = self.get_initial_page() # 第二步：提取表单参数 form_params = self.extract_form_parameters(init_html) # 第三步：提交查询 result_html = self.submit_query(year, month, form_params) # 第四步：解析结果 records = self.parser.parse_monthly_weather(result_html, year, month) return records def crawl_range(self, start_date, end_date): """ 爬取日期范围内的历史天气 start_date, end_date: datetime.date 对象或 'YYYY-MM-DD' 字符串 """ if isinstance(start_date, str): start_date = datetime.strptime(start_date, '%Y-%m-%d').date() if isinstance(end_date, str): end_date = datetime.strptime(end_date, '%Y-%m-%d').date() all_data = [] current = start_date.replace(day=1) end_first = end_date.replace(day=1) while current <= end_first: year = current.year month = current.month try: monthly_data = self.crawl_month(year, month) # 过滤超出结束日期的数据 for record in monthly_data: record_date = datetime.strptime(record['date'], '%Y-%m-%d').date() if start_date <= record_date <= end_date: all_data.append(record) self.logger.info(f"成功获取 {year}-{month:02d}，共 {len(monthly_data)} 条记录") # 随机延迟，礼貌爬取 sleep_time = random.uniform(1, 3) time.sleep(sleep_time) except Exception as e: self.logger.error(f"爬取 {year}-{month:02d} 失败: {e}") # 移到下个月 if current.month == 12: current = current.replace(year=current.year+1, month=1) else: current = current.replace(month=current.month+1) return all_data def close(self): self.http_client.close()

4.5 数据导出模块

python

import pandas as pd from pathlib import Path class WeatherDataExporter: @staticmethod def to_csv(data, filename='weather_history.csv', encoding='utf-8-sig'): """导出为CSV文件""" if not data: print("无数据可导出") return df = pd.DataFrame(data) # 按日期排序 df['date'] = pd.to_datetime(df['date']) df.sort_values('date', inplace=True) df['date'] = df['date'].dt.strftime('%Y-%m-%d') df.to_csv(filename, index=False, encoding=encoding) print(f"成功导出 {len(df)} 条记录到 {filename}") return df @staticmethod def to_excel(data, filename='weather_history.xlsx'): """导出为Excel文件""" if not data: return df = pd.DataFrame(data) df['date'] = pd.to_datetime(df['date']) df.sort_values('date', inplace=True) df.to_excel(filename, index=False, engine='openpyxl') print(f"成功导出到 {filename}")

4.6 异常处理与日志增强

python

import functools import logging from datetime import datetime def log_request(func): """装饰器：记录请求耗时和状态""" @functools.wraps(func) def wrapper(*args, **kwargs): start = datetime.now() try: result = func(*args, **kwargs) duration = (datetime.now() - start).total_seconds() logging.info(f"{func.__name__} 成功，耗时 {duration:.2f}s") return result except Exception as e: duration = (datetime.now() - start).total_seconds() logging.error(f"{func.__name__} 失败: {e}，耗时 {duration:.2f}s") raise return wrapper # 集成到WeatherHTTPClient中 class RobustWeatherHTTPClient(WeatherHTTPClient): @log_request def get(self, url, params=None, referer=None): return super().get(url, params, referer) @log_request def post(self, url, data=None, referer=None): return super().post(url, data, referer)

五、完整运行示例与参数调优

python

def main(): # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('weather_spider.log'), logging.StreamHandler() ] ) # 初始化爬虫（以北京为例） spider = WeatherHistorySpider(city_pinyin='beijing') # 爬取2023年全年数据 all_weather_data = spider.crawl_range('2023-01-01', '2023-12-31') # 导出数据 exporter = WeatherDataExporter() df = exporter.to_csv(all_weather_data, 'beijing_weather_2023.csv') # 打印前5行预览 if df is not None: print("\n数据预览：") print(df.head()) print(f"\n统计信息：") print(df.describe()) spider.close() if __name__ == '__main__': main()

预期输出示例：

text

日期,最高温度,最低温度,天气状况 2023-01-01,5,-8,晴 2023-01-02,4,-7,多云 ... 2023-12-31,2,-9,晴

六、高级优化与生产级改进

6.1 分布式爬取支持（Redis Queue）

对于大规模历史数据（如10年以上），单机爬取效率较低。可使用Redis作为任务队列，多Worker并行爬取：

python

# 使用redis-rq from redis import Redis from rq import Queue redis_conn = Redis() q = Queue(connection=redis_conn) def crawl_month_task(city, year, month): spider = WeatherHistorySpider(city) return spider.crawl_month(year, month) # 提交任务 for year in range(2010, 2024): for month in range(1, 13): q.enqueue(crawl_month_task, 'beijing', year, month)

6.2 代理IP池集成

应对IP封禁，集成免费或付费代理：

python

import random class ProxyManager: def __init__(self, proxy_list): self.proxies = proxy_list def get_random_proxy(self): return random.choice(self.proxies) # 在WeatherHTTPClient中使用 def get(self, url, **kwargs): proxy = self.proxy_manager.get_random_proxy() self.session.proxies = {'http': proxy, 'https': proxy} return super().get(url, **kwargs)

6.3 数据清洗进阶

处理缺失值和异常温度（如高于50℃或低于-50℃）：

python

def clean_temperature(temp): if temp is None: return None if temp > 50 or temp < -50: return None # 异常值 return temp # 集成到解析器中 row_data['temp_max'] = clean_temperature(WeatherDataParser._extract_temperature(max_temp))

6.4 增量爬取与去重

使用SQLite或MongoDB存储，避免重复爬取：

python

import sqlite3 class WeatherDB: def __init__(self, db_path='weather.db'): self.conn = sqlite3.connect(db_path) self.create_table() def create_table(self): self.conn.execute(''' CREATE TABLE IF NOT EXISTS weather ( city TEXT, date TEXT, temp_max INTEGER, temp_min INTEGER, condition TEXT, PRIMARY KEY (city, date) ) ''') def insert(self, city, record): try: self.conn.execute(''' INSERT OR IGNORE INTO weather VALUES (?,?,?,?,?) ''', (city, record['date'], record['temp_max'], record['temp_min'], record.get('weather_condition'))) self.conn.commit() except Exception as e: print(f"插入失败: {e}")

七、常见问题与解决方案

Q1: 网站返回403 Forbidden

原因：服务器检测到非浏览器特征
解决：

• 更新User-Agent池，使用最新Chrome/Firefox UA

• 添加更多浏览器特征头：Accept-Encoding: gzip, deflate, br，Sec-Ch-Ua等

• 使用curl_cffi库模拟TLS指纹

Q2: 表单提交后返回的仍是原页面（未查询成功）

原因：遗漏了必要参数（如__EVENTVALIDATION）或请求方式错误
解决：

• 使用浏览器开发者工具的网络标签，精确复制POST请求的所有参数

• 检查是否缺少Content-Type: application/x-www-form-urlencoded

Q3: 数据表格结构动态加载（Ajax）

解决方案：

• 方法1：使用selenium或playwright自动化浏览器

• 方法2：抓取XHR请求的JSON接口（更高效）

示例使用playwright：

python

from playwright.sync_api import sync_playwright def fetch_with_playwright(url): with sync_playwright() as p: browser = p.chromium.launch(headless=True) page = browser.new_page() page.goto(url) # 点击查询按钮 page.select_option('#year', '2023') page.select_option('#month', '12') page.click('#queryBtn') page.wait_for_selector('.weather-table') html = page.content() browser.close() return html

Q4: 网站使用JavaScript加密参数

进阶对策：

• 使用pyexecjs或node执行加密函数

• 逆向工程加密逻辑（耗时长，需权衡）

• 直接使用selenium绕过前端加密

八、法律与伦理注意事项

1. 遵守robots.txt协议：爬取前检查https://www.tianqi.com/robots.txt

2. 控制请求频率：建议每秒不超过1次，避免对服务器造成压力

3. 数据使用范围：爬取的数据仅用于个人学习、科研，不可商业转售

4. 用户隐私保护：不爬取任何个人用户信息

九、扩展：从表单提交到API爬取

许多现代天气网站提供公开API（需注册获取密钥）。若爬虫被封禁严重，可考虑迁移至合法API：

API调用示例：

python

import requests def fetch_weather_api(city, date): api_key = "YOUR_KEY" url = f"http://api.weatherapi.com/v1/history.json" params = { 'key': api_key, 'q': city, 'dt': date } resp = requests.get(url, params=params) return resp.json()

十、总结与进一步学习

本文完整实现了从表单提交到数据持久化的天气历史爬虫，涵盖了：

• 动态表单参数提取

• 会话维持与请求伪装

• HTML表格的鲁棒解析

• 异常重试与延时控制

• 生产级代码组织