企业官网建设流程全解析

1. 项目概述与核心价值

天气预报数据采集是Python爬虫学习的经典实战案例。2345天气网作为国内主流气象服务平台，其数据更新及时、覆盖城市全面，非常适合作为爬虫教学的目标网站。这个项目不仅能帮助初学者掌握基础的网页数据抓取技术，更能学习到如何处理动态加载内容、解析复杂页面结构等实用技能。

相比简单的静态页面抓取，天气预报数据采集涉及以下几个技术难点：动态参数生成、AJAX请求分析、数据包捕获与解析。通过这个案例，你将系统性地掌握从网页分析到数据存储的完整爬虫开发流程。我在实际工作中发现，天气预报类爬虫项目的技术栈可以复用到电商价格监控、舆情监测等更多实际场景中。

2. 环境准备与工具选型

2.1 基础环境配置

推荐使用Python 3.8+版本进行开发，主要依赖库包括：

• requests（2.26+）：用于发送HTTP请求
• BeautifulSoup4（4.10+）：HTML解析库
• pandas（1.3+）：数据清洗与存储
• selenium（4.1+）：处理动态加载内容

安装命令：

pip install requests beautifulsoup4 pandas selenium

对于浏览器自动化工具，建议搭配ChromeDriver使用。需要注意ChromeDriver版本必须与本地安装的Chrome浏览器版本严格匹配，这是新手最容易出错的环节之一。

2.2 开发工具推荐

1. Chrome开发者工具（F12）：用于分析网络请求、查看页面元素
2. Postman：接口测试工具，验证API请求参数
3. Fiddler/Charles：抓包工具，辅助分析复杂请求
4. PyCharm/VSCode：代码编写与调试

提示：Windows用户建议将chromedriver.exe放在Python安装目录下的Scripts文件夹中，可以避免路径配置问题。

3. 网站分析与请求构造

3.1 页面结构解析

打开2345天气网（www.weather.com.cn），通过检查元素可以发现：

1. 城市选择采用下拉菜单动态加载
2. 天气数据通过AJAX异步请求获取
3. 每个城市有唯一的citycode标识
4. 主要数据接口格式为：/weather/[citycode].shtml

关键发现：直接访问网页源代码并不能获取完整的天气数据，必须分析XHR请求才能找到真实的数据接口。

3.2 抓包分析与接口定位

使用Chrome开发者工具的Network面板，筛选XHR请求后可以发现：

1. 城市列表接口：/data/city3jdata/china.html
2. 天气数据接口：/data/sk/[citycode].html
3. 请求头需要包含Referer字段
4. 部分接口有签名验证（sign参数）

通过多次请求分析，我们发现citycode的生成规则是：省份编号+城市编号+区县编号。例如北京朝阳区的citycode是"101010300"。

3.3 请求参数逆向

对于有反爬机制的接口，需要特别关注以下参数：

1. timestamp：当前时间戳
2. sign：MD5加密字符串
3. token：会话标识

通过JavaScript调试可以发现，sign参数是通过拼接固定字符串+时间戳后取MD5生成的。Python实现代码如下：

import hashlib import time def generate_sign(): timestamp = str(int(time.time())) raw_str = f"secret_key_{timestamp}" return hashlib.md5(raw_str.encode()).hexdigest()

4. 爬虫核心代码实现

4.1 城市列表获取

首先需要获取完整的城市编码列表：

def get_city_codes(): base_url = "http://www.weather.com.cn/data/city3jdata/china.html" headers = { "Referer": "http://www.weather.com.cn/", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)" } response = requests.get(base_url, headers=headers) provinces = response.json() city_codes = [] for prov_code in provinces: city_url = f"http://www.weather.com.cn/data/city3jdata/provshi/{prov_code}.html" cities = requests.get(city_url, headers=headers).json() for city_code in cities: area_url = f"http://www.weather.com.cn/data/city3jdata/station/{prov_code}{city_code}.html" areas = requests.get(area_url, headers=headers).json() for area_code in areas: full_code = f"{prov_code}{city_code}{area_code}" city_codes.append(full_code) return city_codes

4.2 天气数据抓取

获取到城市编码后，就可以抓取详细的天气数据：

def get_weather_data(city_code): api_url = f"http://www.weather.com.cn/data/sk/{city_code}.html" headers = { "Referer": f"http://www.weather.com.cn/weather1d/{city_code}.shtml", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)" } try: response = requests.get(api_url, headers=headers) data = response.json()["weatherinfo"] return { "city": data["city"], "temp": data["temp"], "wind": data["WD"] + data["WS"], "humidity": data["SD"], "update_time": data["time"] } except Exception as e: print(f"Error fetching data for {city_code}: {str(e)}") return None

4.3 反反爬策略实现

针对网站可能的反爬机制，我们需要实现以下防护措施：

1. 请求头完善：

headers = { "Accept": "application/json, text/javascript, */*; q=0.01", "Accept-Encoding": "gzip, deflate", "Accept-Language": "zh-CN,zh;q=0.9", "Connection": "keep-alive", "Host": "www.weather.com.cn", "Referer": "http://www.weather.com.cn/", "X-Requested-With": "XMLHttpRequest", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/96.0.4664.110 Safari/537.36" }

2. IP代理池集成：

def get_proxy(): # 这里替换为你的代理获取接口 proxy_url = "http://your-proxy-pool.com/get" try: return requests.get(proxy_url).text except: return None def make_request(url, headers): proxy = get_proxy() if proxy: proxies = { "http": f"http://{proxy}", "https": f"http://{proxy}" } return requests.get(url, headers=headers, proxies=proxies, timeout=10) else: return requests.get(url, headers=headers, timeout=10)

3. 请求频率控制：

import random import time def random_delay(): time.sleep(random.uniform(1, 3))

5. 数据存储与处理

5.1 数据清洗

获取到的原始数据需要进行清洗：

def clean_data(raw_data): # 温度处理 temp = raw_data.get("temp", "N/A") if temp != "N/A": temp = float(temp.replace("℃", "")) # 风力等级转换 wind_map = { "微风": "1-2级", "和风": "3-4级", "劲风": "5-6级" } wind = raw_data.get("wind", "N/A") wind = wind_map.get(wind, wind) return { "city": raw_data.get("city", "N/A"), "temperature": temp, "wind_level": wind, "humidity": raw_data.get("humidity", "N/A"), "update_time": raw_data.get("update_time", "N/A") }

5.2 存储方案选择

根据数据量大小可以选择不同的存储方式：

1. 小规模数据（<1万条）：

# CSV存储 import pandas as pd def save_to_csv(data_list, filename): df = pd.DataFrame(data_list) df.to_csv(filename, index=False, encoding="utf_8_sig")

2. 中等规模数据（1万-100万条）：

# MySQL存储 import pymysql def save_to_mysql(data_list): conn = pymysql.connect( host="localhost", user="root", password="yourpassword", database="weather" ) with conn.cursor() as cursor: sql = """INSERT INTO weather_data (city, temperature, wind_level, humidity, update_time) VALUES (%s, %s, %s, %s, %s)""" cursor.executemany(sql, data_list) conn.commit() conn.close()

3. 大规模数据（>100万条）：

# MongoDB存储 from pymongo import MongoClient def save_to_mongodb(data_list): client = MongoClient("mongodb://localhost:27017/") db = client["weather_db"] collection = db["weather_data"] collection.insert_many(data_list) client.close()

6. 常见问题与解决方案

6.1 请求返回403错误

可能原因及解决方案：

1. 请求头不完整：补全Referer、User-Agent等字段
2. IP被封禁：使用代理IP或降低请求频率
3. 签名验证失败：检查sign参数生成逻辑

6.2 数据解析异常

典型错误处理：

try: data = response.json() weather_info = data["weatherinfo"] except ValueError: print("Invalid JSON response") except KeyError: print("Missing 'weatherinfo' field in response")

6.3 动态加载内容抓取失败

对于无法通过API获取的数据，可以使用Selenium：

from selenium import webdriver from selenium.webdriver.chrome.options import Options def get_dynamic_content(url): chrome_options = Options() chrome_options.add_argument("--headless") driver = webdriver.Chrome(options=chrome_options) driver.get(url) # 等待动态内容加载 time.sleep(2) # 获取页面源码 html = driver.page_source driver.quit() return html

6.4 数据更新不及时

解决方案：

1. 设置定时任务（如每天6点、12点、18点各采集一次）
2. 监控数据变化，当发现更新时立即抓取
3. 实现增量采集，只获取变化的数据

7. 项目优化与扩展

7.1 性能优化方案

1. 多线程采集：

from concurrent.futures import ThreadPoolExecutor def multi_thread_crawl(city_codes, workers=5): with ThreadPoolExecutor(max_workers=workers) as executor: results = list(executor.map(get_weather_data, city_codes)) return [r for r in results if r is not None]

2. 异步IO实现：

import aiohttp import asyncio async def async_fetch(session, url): async with session.get(url) as response: return await response.json() async def main(city_codes): async with aiohttp.ClientSession() as session: tasks = [] for code in city_codes: url = f"http://www.weather.com.cn/data/sk/{code}.html" tasks.append(async_fetch(session, url)) return await asyncio.gather(*tasks)

7.2 功能扩展方向

1. 历史数据归档：定期存储天气数据，建立历史数据库
2. 天气预警监控：实时监测极端天气预警信息
3. 可视化展示：使用Pyecharts等库生成天气数据图表
4. 天气预报推送：通过邮件/微信定时发送天气预报

7.3 企业级应用建议

对于商业用途的天气数据采集，建议：

1. 使用分布式爬虫框架（Scrapy、Scrapy-Redis）
2. 搭建稳定的代理IP池
3. 实现自动化监控与告警系统
4. 设计合理的数据存储架构
5. 遵守《网络安全法》相关规定

8. 完整代码示例

以下是整合了所有功能的完整实现：

import requests import json import time import random import pandas as pd from bs4 import BeautifulSoup from concurrent.futures import ThreadPoolExecutor class WeatherSpider: def __init__(self): self.base_url = "http://www.weather.com.cn" self.headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)", "Referer": self.base_url } def get_city_codes(self): """获取所有城市编码""" prov_url = f"{self.base_url}/data/city3jdata/china.html" provinces = requests.get(prov_url, headers=self.headers).json() city_codes = [] for prov_code in provinces: city_url = f"{self.base_url}/data/city3jdata/provshi/{prov_code}.html" cities = requests.get(city_url, headers=self.headers).json() for city_code in cities: area_url = f"{self.base_url}/data/city3jdata/station/{prov_code}{city_code}.html" areas = requests.get(area_url, headers=self.headers).json() for area_code in areas: full_code = f"{prov_code}{city_code}{area_code}" city_codes.append(full_code) return city_codes def get_weather_data(self, city_code): """获取单个城市天气数据""" api_url = f"{self.base_url}/data/sk/{city_code}.html" try: response = requests.get(api_url, headers=self.headers) data = response.json()["weatherinfo"] return { "city": data["city"], "temp": data["temp"], "wind": data["WD"] + data["WS"], "humidity": data["SD"], "update_time": data["time"] } except Exception as e: print(f"Error fetching {city_code}: {str(e)}") return None def run(self, output_file="weather_data.csv"): """主运行方法""" print("开始获取城市列表...") city_codes = self.get_city_codes() print(f"共获取到{len(city_codes)}个城市编码") print("开始采集天气数据...") with ThreadPoolExecutor(max_workers=5) as executor: results = list(executor.map(self.get_weather_data, city_codes)) valid_data = [r for r in results if r is not None] print(f"成功获取{len(valid_data)}个城市的数据") df = pd.DataFrame(valid_data) df.to_csv(output_file, index=False, encoding="utf_8_sig") print(f"数据已保存至{output_file}") if __name__ == "__main__": spider = WeatherSpider() spider.run()

重要提示：实际运行时建议添加适当的延迟（如time.sleep(random.uniform(1,3))）以避免给目标服务器造成过大压力。商业用途请务必遵守网站的使用条款。