1. 项目背景与需求分析
最近在帮朋友策划亲子活动时,发现很多儿童活动场所的分区说明都只发布在官网上,而且信息分散在不同页面。手动收集这些数据不仅耗时耗力,还容易遗漏关键信息。作为一个经常处理数据问题的Python开发者,我决定写个爬虫来自动化这个采集过程。
儿童活动空间的分区说明通常包含以下关键信息:
- 不同年龄段的适用区域划分
- 安全注意事项和特殊设施说明
- 开放时间和预约要求
- 各区域的功能介绍和适玩年龄
这些信息对家长规划出行非常重要,但往往需要点击多个页面才能获取完整信息。通过爬虫自动化采集,可以:
- 一次性获取所有相关页面数据
- 自动整理成结构化格式
- 建立本地数据库方便后续查询
- 设置定期更新机制保持信息时效性
2. 技术方案设计
2.1 目标网站分析
以某大型儿童乐园官网为例,其分区说明页有这些特点:
- 使用动态加载技术(AJAX)
- 关键信息有时会以图片形式呈现
- 分页采用JavaScript渲染
- 需要处理登录状态和Cookies
2.2 技术选型
基于这些特点,我选择了以下技术栈:
# 核心库 import requests from bs4 import BeautifulSoup import selenium from PIL import Image import pytesseract # 辅助工具 import pandas as pd import json import time选择理由:
- Requests + BeautifulSoup:处理静态页面解析
- Selenium:应对动态加载内容
- PIL + pytesseract:处理图片文字识别
- Pandas:数据清洗和存储
3. 爬虫实现细节
3.1 基础爬取流程
def get_page(url): headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' } try: response = requests.get(url, headers=headers) response.raise_for_status() return response.text except Exception as e: print(f"获取页面失败: {e}") return None3.2 处理动态内容
对于需要交互才能加载的内容:
from selenium.webdriver.chrome.options import Options def setup_driver(): options = Options() options.add_argument("--headless") driver = webdriver.Chrome(options=options) return driver def get_dynamic_content(driver, url): driver.get(url) time.sleep(2) # 等待渲染 return driver.page_source3.3 图片文字识别
当遇到文字图片时:
def extract_text_from_image(img_url): response = requests.get(img_url, stream=True) img = Image.open(response.raw) text = pytesseract.image_to_string(img, lang='chi_sim') return text.strip()4. 数据清洗与存储
4.1 信息提取
使用BeautifulSoup提取关键数据:
def parse_page(html): soup = BeautifulSoup(html, 'html.parser') data = { 'title': soup.find('h1').text if soup.find('h1') else '', 'sections': [], 'update_time': '' } # 提取各区域信息 for section in soup.select('.area-section'): section_data = { 'name': section.select_one('.section-title').text, 'age_range': section.select_one('.age-range').text, 'features': [li.text for li in section.select('.feature-list li')] } data['sections'].append(section_data) return data4.2 数据存储方案
提供三种存储方式供选择:
# JSON存储 def save_to_json(data, filename): with open(filename, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) # CSV存储 def save_to_csv(data, filename): df = pd.DataFrame(data['sections']) df.to_csv(filename, index=False, encoding='utf_8_sig') # 数据库存储 import sqlite3 def save_to_db(data, db_file='playground.db'): conn = sqlite3.connect(db_file) c = conn.cursor() # 创建表 c.execute('''CREATE TABLE IF NOT EXISTS sections (name TEXT, age_range TEXT, features TEXT)''') # 插入数据 for section in data['sections']: c.execute("INSERT INTO sections VALUES (?,?,?)", (section['name'], section['age_range'], ','.join(section['features']))) conn.commit() conn.close()5. 反爬策略应对
5.1 常见反爬措施
目标网站可能采用:
- IP频率限制
- User-Agent检测
- 验证码挑战
- 行为分析
5.2 应对方案
# 使用代理IP池 proxies = { 'http': 'http://proxy.example.com:8080', 'https': 'https://proxy.example.com:8080' } # 随机User-Agent from fake_useragent import UserAgent ua = UserAgent() headers = {'User-Agent': ua.random} # 请求间隔控制 import random time.sleep(random.uniform(1, 3))6. 项目优化建议
6.1 性能优化
- 使用多线程/异步请求:
import concurrent.futures def fetch_urls(urls): with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor: executor.map(get_page, urls)- 实现增量爬取:
def check_update(soup): last_update = soup.select_one('.last-updated').text # 与本地存储的最后更新时间比较 # 返回是否需要更新6.2 功能扩展
- 添加自动邮件通知功能:
import smtplib from email.mime.text import MIMEText def send_notification(subject, content): msg = MIMEText(content) msg['Subject'] = subject msg['From'] = 'your_email@example.com' msg['To'] = 'recipient@example.com' with smtplib.SMTP('smtp.example.com') as server: server.login('username', 'password') server.send_message(msg)- 开发可视化展示界面:
import matplotlib.pyplot as plt def visualize_age_ranges(data): age_ranges = [s['age_range'] for s in data['sections']] counts = {age: age_ranges.count(age) for age in set(age_ranges)} plt.bar(counts.keys(), counts.values()) plt.title('各年龄段活动区域分布') plt.show()7. 常见问题解决
7.1 编码问题
# 强制指定编码 response.encoding = response.apparent_encoding7.2 元素定位失败
# 更健壮的元素查找 element = soup.find('div', class_='section') or soup.find('div', id='section')7.3 验证码处理
# 半自动处理方案 def handle_captcha(driver): driver.save_screenshot('captcha.png') print("请查看captcha.png并输入验证码:") captcha = input() driver.find_element_by_id('captcha').send_keys(captcha)8. 完整项目结构
建议的项目目录结构:
/child_activity_crawler │── main.py # 主程序入口 │── config.py # 配置文件 │── requirements.txt # 依赖库 ├── /utils │ ├── crawler.py # 爬虫核心功能 │ ├── parser.py # 页面解析 │ └── storage.py # 数据存储 ├── /data │ ├── raw/ # 原始数据 │ └── processed/ # 处理后的数据 └── /logs # 运行日志9. 实际应用建议
- 定时执行方案:
# Linux crontab设置 0 3 * * * /usr/bin/python3 /path/to/main.py >> /path/to/logs/crawl.log 2>&1- 数据更新策略:
- 每周一凌晨3点全量更新
- 每天检查更新标志
- 发现变更时只爬取变动部分
- 异常处理机制:
import logging logging.basicConfig(filename='crawler.log', level=logging.INFO) try: # 爬取代码 except Exception as e: logging.error(f"爬取失败: {str(e)}") send_notification("爬虫异常报警", str(e))10. 法律与道德考量
- 遵守robots.txt协议
- 控制请求频率(建议≥3秒/次)
- 不爬取个人隐私信息
- 数据仅用于个人研究
- 注明数据来源
可以在代码中添加合规检查:
def check_robots(url): robots_url = f"{urlparse(url).scheme}://{urlparse(url).netloc}/robots.txt" response = requests.get(robots_url) if response.status_code == 200: print("请遵守以下爬取规则:") print(response.text)