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

一、爬虫项目背景与目标在数据驱动的时代天气数据作为基础的环境信息在农业预测、旅游规划、能源管理、历史事件回溯分析等领域具有重要价值。然而主流天气网站通常仅提供有限的历史数据免费查询且往往需要用户手动选择日期并点击查询按钮才能获取。对于需要批量获取长时间序列数据的场景人工操作显然不可行。因此开发一个能够自动提交表单、模拟查询行为、精确提取历史天气数据的爬虫成为数据采集工程师的必备技能。本文将以中国知名天气数据平台“天气网”https://www.tianqi.com 为目标系统讲解如何利用Python爬虫技术实现从日期选择、表单提交、响应解析到数据持久化的完整流程。本项目的核心难点在于处理动态表单参数、维持会话状态、应对反爬虫机制、解析非结构化HTML文本。我们将采用最新稳定的技术栈包括requests、BeautifulSoup4、pandas以及fake_useragent提供可直接投入生产的代码。核心功能点自动构造目标城市的天气查询URL模拟浏览器提交年份和月份参数解析返回的HTML表格提取日期、最高温度、最低温度、天气状况支持多日期范围批量爬取输出结构化数据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 Queue6.2 代理IP池集成6.3 数据清洗进阶6.4 增量爬取与去重七、常见问题与解决方案Q1: 网站返回403 ForbiddenQ2: 表单提交后返回的仍是原页面未查询成功Q3: 数据表格结构动态加载AjaxQ4: 网站使用JavaScript加密参数八、法律与伦理注意事项九、扩展从表单提交到API爬取十、总结与进一步学习二、技术选型与环境搭建2.1 为什么选择这些技术库库名版本要求核心作用替代方案requests2.31.0处理HTTP请求维持Session提交表单数据httpx, aiohttpBeautifulSoup44.12.0解析HTML文档使用CSS选择器提取数据lxml, parselfake_useragent1.4.0随机生成User-Agent头规避反爬手动维护UA列表pandas2.0.0数据清洗与导出时间序列处理csv模块 openpyxlretrying1.3.3请求失败自动重试tenacity, 手动循环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验证安装pythonimport sys print(sys.version) import requests print(requests.__version__) # 应输出 2.31.x 及以上2.3 目标网站分析目标URL模式以北京为例https://www.tianqi.com/lishi/beijing/{year}{month}.htmlbeijing为城市拼音{year}{month}为6位数字如202401代表2024年1月关键发现该网站历史数据按月份分页存储直接访问构造的URL即可获取整月数据无需额外的POST请求但部分类似网站如 weather.com需要提交表单参数。为展示“表单提交”技术点我们将模拟访问带有月份选择器的实际交互流程——即先发送GET获取页面解析隐藏的表单token再POST请求提交查询。2.4 反爬虫机制识别使用开发者工具F12分析请求特征User-Agent校验服务器检测请求头中的UA非浏览器标识返回403Referer检查部分页面要求来源页为本站Cookie依赖首次访问需设置会话Cookie请求频率限制同IP短时间内请求过多会触发封禁动态Token表单中包含一次性token参数针对上述机制我们的应对策略使用fake_useragent伪造浏览器UA正确设置Referer和Origin头使用requests.Session()自动管理Cookie添加随机延迟1~3秒从页面中提取token动态填充表单三、核心爬虫架构设计3.1 整体流程图text开始 → 用户输入城市拼音 → 生成目标URL → 发起GET请求获取主页面 → 提取表单隐藏参数token等→ 构造POST数据年份、月份→ 发送POST请求获取结果页面 → 解析HTML表格 → 清洗数据 → 循环下个月 → 最终合并导出CSV3.2 模块划分NetworkManager负责HTTP请求处理重试、头信息、代理FormExtractor从初始页面解析表单字段DataParser使用BeautifulSoup解析天气表格DataExporter将爬取结果保存为CSV/ExcelScheduler控制请求频率和日期范围循环四、详细代码实现4.1 网络请求模块带重试和伪装pythonimport requests from fake_useragent import UserAgent from retrying import retry import time import random class WeatherHTTPClient: 封装HTTP请求支持会话保持和自动重试 def __init__(self, retry_max_attempts3, timeout10): self.session requests.Session() self.timeout timeout self.retry_attempts retry_max_attempts self.ua UserAgent() # 设置默认请求头 self.session.headers.update({ Accept: text/html,application/xhtmlxml,application/xml;q0.9,image/webp,*/*;q0.8, Accept-Language: zh-CN,zh;q0.8,zh-TW;q0.7,zh-HK;q0.5,en-US;q0.3,en;q0.2, Accept-Encoding: gzip, deflate, br, Connection: keep-alive, Upgrade-Insecure-Requests: 1, }) def _get_random_headers(self, refererNone): 生成随机User-Agent并可选设置Referer headers { User-Agent: self.ua.random, } if referer: headers[Referer] referer return headers retry(stop_max_attempt_number3, wait_fixed2000) def get(self, url, paramsNone, refererNone): 带重试的GET请求 headers self._get_random_headers(referer) response self.session.get( url, paramsparams, headersheaders, timeoutself.timeout ) response.raise_for_status() # 部分网站使用GBK编码需自动检测 if charsetgbk in response.text.lower() or charsetgb2312 in response.text.lower(): response.encoding gbk else: response.encoding response.apparent_encoding return response retry(stop_max_attempt_number3, wait_fixed2000) def post(self, url, dataNone, refererNone): 带重试的POST请求 headers self._get_random_headers(referer) response self.session.post( url, datadata, headersheaders, timeoutself.timeout ) response.raise_for_status() response.encoding response.apparent_encoding return response def close(self): self.session.close()4.2 表单参数提取器很多天气网站使用隐藏input传递CSRF token或会话标识。我们需要从HTML中提取这些参数。pythonfrom bs4 import BeautifulSoup class FormParameterExtractor: 从页面中提取表单提交所需的参数 staticmethod def extract_hidden_inputs(html_content, form_idNone): 提取表单中的所有隐藏输入字段 :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_namesNone): 提取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 None4.3 天气数据解析器解析历史天气表格是关键步骤。以天气网为例每个月的页面中包含一个表格列通常包括日期、最高温度、最低温度、天气、风向等。pythonclass 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(stripTrue) for th in thead.find_all(th)] else: # 尝试从第一行获取 first_row table.find(tr) headers [td.get_text(stripTrue) 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(stripTrue) # 构造完整日期: 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(stripTrue) row_data[temp_max] WeatherDataParser._extract_temperature(max_temp) if temp_min in col_mapping: min_temp cells[col_mapping[temp_min]].get_text(stripTrue) row_data[temp_min] WeatherDataParser._extract_temperature(min_temp) if condition in col_mapping: row_data[weather_condition] cells[col_mapping[condition]].get_text(stripTrue) if wind in col_mapping: row_data[wind] cells[col_mapping[wind]].get_text(stripTrue) # 只有包含有效日期才添加 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 None4.4 主爬虫逻辑表单提交模拟以下代码演示完整流程访问包含日期选择器的页面 → 提取表单参数 → 提交POST请求 → 解析响应。pythonimport 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 fhttps://www.tianqi.com/lishi/{city_pinyin}/ self.query_url https://www.tianqi.com/lishi/{}/query # 假设的提交端点 logging.basicConfig(levellogging.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_idaspnetForm) 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, datapost_data, refererself.base_url) return response.text def crawl_month(self, year, month): 爬取单月数据组合方式直接构造URL模式 # 方法A: 若网站支持直接按月URL访问则使用此方式更稳定 direct_url fhttps://www.tianqi.com/lishi/{self.city}/{year}{month:02d}.html self.logger.info(f直接访问: {direct_url}) try: resp self.http_client.get(direct_url, refererself.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(day1) end_first end_date.replace(day1) 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(yearcurrent.year1, month1) else: current current.replace(monthcurrent.month1) return all_data def close(self): self.http_client.close()4.5 数据导出模块pythonimport pandas as pd from pathlib import Path class WeatherDataExporter: staticmethod def to_csv(data, filenameweather_history.csv, encodingutf-8-sig): 导出为CSV文件 if not data: print(无数据可导出) return df pd.DataFrame(data) # 按日期排序 df[date] pd.to_datetime(df[date]) df.sort_values(date, inplaceTrue) df[date] df[date].dt.strftime(%Y-%m-%d) df.to_csv(filename, indexFalse, encodingencoding) print(f成功导出 {len(df)} 条记录到 {filename}) return df staticmethod def to_excel(data, filenameweather_history.xlsx): 导出为Excel文件 if not data: return df pd.DataFrame(data) df[date] pd.to_datetime(df[date]) df.sort_values(date, inplaceTrue) df.to_excel(filename, indexFalse, engineopenpyxl) print(f成功导出到 {filename})4.6 异常处理与日志增强pythonimport 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, paramsNone, refererNone): return super().get(url, params, referer) log_request def post(self, url, dataNone, refererNone): return super().post(url, data, referer)五、完整运行示例与参数调优pythondef main(): # 配置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(weather_spider.log), logging.StreamHandler() ] ) # 初始化爬虫以北京为例 spider WeatherHistorySpider(city_pinyinbeijing) # 爬取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(connectionredis_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封禁集成免费或付费代理pythonimport 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℃pythondef 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存储避免重复爬取pythonimport sqlite3 class WeatherDB: def __init__(self, db_pathweather.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, brSec-Ch-Ua等使用curl_cffi库模拟TLS指纹Q2: 表单提交后返回的仍是原页面未查询成功原因遗漏了必要参数如__EVENTVALIDATION或请求方式错误解决使用浏览器开发者工具的网络标签精确复制POST请求的所有参数检查是否缺少Content-Type: application/x-www-form-urlencodedQ3: 数据表格结构动态加载Ajax解决方案方法1使用selenium或playwright自动化浏览器方法2抓取XHR请求的JSON接口更高效示例使用playwrightpythonfrom playwright.sync_api import sync_playwright def fetch_with_playwright(url): with sync_playwright() as p: browser p.chromium.launch(headlessTrue) 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 htmlQ4: 网站使用JavaScript加密参数进阶对策使用pyexecjs或node执行加密函数逆向工程加密逻辑耗时长需权衡直接使用selenium绕过前端加密八、法律与伦理注意事项遵守robots.txt协议爬取前检查https://www.tianqi.com/robots.txt控制请求频率建议每秒不超过1次避免对服务器造成压力数据使用范围爬取的数据仅用于个人学习、科研不可商业转售用户隐私保护不爬取任何个人用户信息九、扩展从表单提交到API爬取许多现代天气网站提供公开API需注册获取密钥。若爬虫被封禁严重可考虑迁移至合法APIAPI服务商免费额度历史数据深度OpenWeatherMap1000次/天5年WeatherAPI10000次/月历史仅付费和风天气1000次/天30天历史API调用示例pythonimport requests def fetch_weather_api(city, date): api_key YOUR_KEY url fhttp://api.weatherapi.com/v1/history.json params { key: api_key, q: city, dt: date } resp requests.get(url, paramsparams) return resp.json()十、总结与进一步学习本文完整实现了从表单提交到数据持久化的天气历史爬虫涵盖了动态表单参数提取会话维持与请求伪装HTML表格的鲁棒解析异常重试与延时控制生产级代码组织