Python Requests + XPath 实战：高效爬取户外轨迹数据与反爬策略

1. 项目背景与目标

户外运动爱好者经常会在六只脚等平台分享他们的GPS轨迹数据，这些数据包含丰富的时空信息，如经纬度、海拔、速度等。对于研究人员和数据分析师来说，这些数据可以用来分析用户行为、优化景区规划或研究城市交通模式。

本项目的目标是构建一个健壮的Python爬虫，能够：

• 自动登录六只脚平台
• 按关键词搜索并爬取多页轨迹数据
• 处理常见的反爬机制
• 将数据规范存储为结构化格式
• 实现可配置的爬取参数

2. 技术选型与准备

2.1 核心工具链

# 主要依赖库 import requests # 网络请求 from lxml import etree # HTML解析 import pandas as pd # 数据处理 from fake_useragent import UserAgent # 随机User-Agent生成 import time # 请求间隔控制 import json # JSON数据处理 import os # 文件系统操作 import re # 正则表达式

2.2 环境配置建议

1. 创建虚拟环境：

python -m venv trail_env source trail_env/bin/activate # Linux/Mac trail_env\Scripts\activate # Windows

2. 安装依赖：

pip install requests lxml pandas fake-useragent

3. 核心爬取逻辑实现

3.1 网站结构分析

六只脚网站的典型数据结构：

3.2 分步骤实现

3.2.1 登录与会话保持

def get_csrf_token(): """获取CSRF令牌""" login_url = "http://www.foooooot.com/accounts/login/" headers = {"User-Agent": UserAgent().random} response = requests.get(login_url, headers=headers) # 使用正则提取CSRF token pattern = r"<input type='hidden' name='csrfmiddlewaretoken' value='(.*?)'" csrf_token = re.search(pattern, response.text).group(1) return csrf_token def login(session, csrf_token, email, password): """执行登录操作""" login_data = { "csrfmiddlewaretoken": csrf_token, "email": email, "password": password, "next": "/accounts/login_complete/" } headers = { "User-Agent": UserAgent().random, "Referer": "http://www.foooooot.com/accounts/login/" } response = session.post( "http://www.foooooot.com/accounts/login/", data=login_data, headers=headers ) return response.status_code == 200

3.2.2 分页爬取轨迹ID

def get_trip_ids(session, keyword, start_page=1, max_pages=30): """获取指定关键词的轨迹ID列表""" trip_ids = [] current_page = start_page while True: url = f"http://www.foooooot.com/search/trip/all/1/all/time/descent/?page={current_page}&keyword={keyword}" response = session.get(url) if response.status_code != 200: break tree = etree.HTML(response.text) links = tree.xpath('//p[@class="trip-title"]/a/@href') if not links: break # 提取轨迹ID：/trip/123456/ → 123456 page_ids = [link.split("/")[2] for link in links] trip_ids.extend(page_ids) # 控制请求频率 time.sleep(3 + random.random() * 2) current_page += 1 if current_page > max_pages: break return trip_ids

3.2.3 获取轨迹详情数据

def get_trip_data(session, trip_id): """获取单条轨迹的详细数据""" json_url = f"http://www.foooooot.com/trip/{trip_id}/trackjson/" try: response = session.get(json_url, timeout=10) if response.status_code == 200: return response.json() except Exception as e: print(f"Error fetching trip {trip_id}: {str(e)}") return None

4. 反爬策略应对方案

4.1 常见反爬机制及对策

4.2 健壮性增强实现

def safe_request(session, url, max_retries=3, timeout=10): """带重试机制的请求函数""" for attempt in range(max_retries): try: response = session.get( url, timeout=timeout, headers={"User-Agent": UserAgent().random} ) if response.status_code == 200: return response elif response.status_code == 403: raise Exception("IP可能被封禁") except Exception as e: if attempt == max_retries - 1: raise wait_time = (2 ** attempt) + random.random() time.sleep(wait_time) return None

5. 数据存储与处理

5.1 数据结构设计

典型的轨迹数据包含以下字段：

{ "timestamp": 1625097600, # UNIX时间戳 "latitude": 39.9075, # 纬度 "longitude": 116.3972, # 经度 "altitude": 50.2, # 海拔(米) "speed": 5.3, # 速度(km/h) "distance": 1250.4 # 累计距离(米) }

5.2 存储方案比较

5.3 示例存储实现

def save_to_csv(data, keyword, trip_id): """保存单条轨迹数据到CSV""" os.makedirs(f"{keyword}_tracks", exist_ok=True) filename = f"{keyword}_tracks/{trip_id}.csv" df = pd.DataFrame(data) df.to_csv(filename, index=False, encoding='utf-8-sig') print(f"Saved {len(df)} points to {filename}") def save_to_sqlite(data, db_path="tracks.db"): """保存到SQLite数据库""" import sqlite3 conn = sqlite3.connect(db_path) df = pd.DataFrame(data) df.to_sql("tracks", conn, if_exists="append", index=False) conn.close()

6. 完整爬虫架构

6.1 主程序流程

def main(): # 初始化配置 keyword = input("请输入搜索关键词: ") start_page = int(input("请输入起始页码: ")) max_pages = int(input("请输入最大爬取页数: ")) # 创建会话 session = requests.Session() # 登录流程 csrf_token = get_csrf_token() if not login(session, csrf_token, "your_email", "your_password"): print("登录失败") return # 获取轨迹ID列表 trip_ids = get_trip_ids(session, keyword, start_page, max_pages) print(f"共发现{len(trip_ids)}条轨迹") # 爬取每条轨迹数据 for trip_id in trip_ids: try: data = get_trip_data(session, trip_id) if data: save_to_csv(data, keyword, trip_id) time.sleep(2 + random.random()) # 随机延迟 except Exception as e: print(f"处理轨迹{trip_id}时出错: {str(e)}") continue print("爬取任务完成") if __name__ == "__main__": main()

6.2 配置参数建议

# config.py DEFAULT_CONFIG = { "request": { "timeout": 15, "retries": 3, "delay": (2, 5) # 随机延迟范围(秒) }, "auth": { "email": "your@email.com", "password": "your_password" }, "storage": { "format": "csv", # csv/json/sqlite "output_dir": "./data" } }

7. 高级技巧与优化

7.1 性能优化策略

1. 异步请求：使用aiohttp替代requests

import aiohttp import asyncio async def fetch(session, url): async with session.get(url) as response: return await response.json() async def main(): async with aiohttp.ClientSession() as session: tasks = [fetch(session, url) for url in urls] return await asyncio.gather(*tasks)

2. 连接池复用：配置requests.Session的连接池

session = requests.Session() adapter = requests.adapters.HTTPAdapter( pool_connections=10, pool_maxsize=50, max_retries=3 ) session.mount('http://', adapter)

7.2 数据质量保障

1. 数据验证：

def validate_track_point(point): required_fields = ['latitude', 'longitude', 'timestamp'] return all(field in point for field in required_fields)

2. 异常数据处理：

def clean_data(data): # 移除无效数据点 return [p for p in data if -90 <= p['latitude'] <= 90 and -180 <= p['longitude'] <= 180]

8. 可视化与分析示例

8.1 使用Pandas进行基础分析

import pandas as pd import matplotlib.pyplot as plt # 加载数据 df = pd.read_csv("岳麓山_tracks/123456.csv") # 计算基本统计量 stats = df.describe() print(stats) # 绘制海拔变化图 plt.figure(figsize=(10, 4)) df['altitude'].plot(title="海拔变化") plt.xlabel("轨迹点序号") plt.ylabel("海拔(米)") plt.show()

8.2 使用Folium绘制轨迹地图

import folium # 创建地图 m = folium.Map( location=[df['latitude'].mean(), df['longitude'].mean()], zoom_start=14 ) # 添加轨迹线 folium.PolyLine( list(zip(df['latitude'], df['longitude'])), color='blue', weight=2.5, opacity=1 ).add_to(m) # 保存为HTML m.save('trail_map.html')

9. 项目扩展方向

1. 分布式爬取：使用Scrapy-Redis实现分布式爬虫
2. 实时监控：添加Prometheus监控指标
3. 数据管道：集成Apache Airflow实现定期爬取
4. API服务：使用FastAPI提供数据查询接口

提示：在实际项目中，建议添加完善的日志记录和异常处理机制，可以使用Python的logging模块实现分级日志记录。