1. 项目概述
最近在做一个商业分析项目时,需要收集全国连锁门店的基础信息。手动收集显然不现实,于是研究了一套基于Python的逆向API解决方案。这个方案不仅能获取店名、地址、电话等基础信息,还能精准抓取经纬度坐标,并支持CSV格式导出,对商业选址、竞品分析等场景特别实用。
市面上常见的爬虫教程大多停留在基础页面抓取,而实际商业应用中,我们往往需要处理反爬机制、解析非结构化数据、应对动态加载等复杂情况。本文将分享一套经过实战检验的完整方案,包含从逆向分析到数据导出的全流程,特别适合需要批量获取门店信息的市场分析人员或数据工程师。
2. 核心思路与技术选型
2.1 逆向工程 vs 公开API
传统爬虫直接请求目标网站前端页面,但现代网站普遍采用动态渲染和严格反爬措施。经过测试,我发现目标网站的移动端API接口反而防护较弱,且返回结构化JSON数据。因此决定采用逆向工程思路,通过浏览器开发者工具捕获真实API请求。
2.2 技术栈组成
- 请求库:选用
aiohttp而非requests,支持异步并发请求,实测速度提升5-8倍 - 解析工具:
json模块处理API响应,re正则表达式辅助提取关键字段 - 地理编码:高德地图API将地址转换为经纬度(需申请开发者key)
- 数据存储:
pandas处理DataFrame并导出CSV - 反爬对抗:
fake-useragent随机生成请求头,proxy代理池轮询IP
提示:商业级爬虫必须考虑请求频率控制,建议设置2-3秒的随机延迟,避免触发风控
3. 完整实现步骤
3.1 逆向分析API接口
以某连锁药店为例,在手机浏览器访问门店列表页,抓包发现核心接口:
https://api.xxx.com/store/list?cityId=010&page=1&size=20通过修改cityId参数可切换城市,page参数实现分页。响应格式如下:
{ "code": 200, "data": { "stores": [ { "name": "XX药店(中关村店)", "address": "海淀区中关村大街1号", "tel": "010-62558888", "businessHours": "08:00-22:00" } ] } }3.2 Python实现代码解析
import aiohttp import asyncio import pandas as pd from geopy.geocoders import Gaode class StoreSpider: def __init__(self): self.geocoder = Gaode(api_key='your_amap_key') self.headers = { 'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 13_2_3 like Mac OS X)' } async def fetch_store(self, session, city_id, page): url = f"https://api.xxx.com/store/list?cityId={city_id}&page={page}&size=20" async with session.get(url, headers=self.headers) as resp: data = await resp.json() stores = data['data']['stores'] results = [] for store in stores: location = self.geocoder.geocode(store['address']) results.append({ 'name': store['name'], 'address': store['address'], 'tel': store['tel'], 'lat': location.latitude, 'lng': location.longitude }) return results3.3 关键功能实现
3.3.1 城市ID映射处理
发现不同城市使用特定编码(如北京为010),需先构建城市字典:
city_mapping = { '北京': '010', '上海': '021', '广州': '020' # 其他城市可通过分析页面JS获取 }3.3.2 异步并发控制
通过信号量控制最大并发数,避免被封IP:
semaphore = asyncio.Semaphore(5) async def bounded_fetch(session, city_id, page): async with semaphore: await asyncio.sleep(random.uniform(1, 3)) return await self.fetch_store(session, city_id, page)3.3.3 数据清洗与导出
使用pandas处理数据并导出CSV:
def save_to_csv(data, filename): df = pd.DataFrame(data) # 处理重复数据 df.drop_duplicates(subset=['name','address'], inplace=True) # 标准化电话号码 df['tel'] = df['tel'].str.replace(r'(\d{3,4})-?(\d{7,8})', r'\1-\2') df.to_csv(filename, index=False, encoding='utf_8_sig')4. 实战问题与解决方案
4.1 高频访问被封IP
现象:连续请求20次后返回403错误解决方案:
- 使用代理IP池轮换(推荐付费API服务)
- 每个请求添加随机延迟(1-5秒)
- 模拟真实用户行为:随机切换User-Agent
4.2 地址解析失败
场景:部分新开商场地址无法解析坐标应对策略:
try: location = geocoder.geocode(address) except: # 尝试去除详细楼层信息后重试 location = geocoder.geocode(re.sub(r'[A-Z]层\d+号','',address))4.3 数据完整性校验
设计自动校验规则:
def validate_store(store): required_fields = ['name', 'address', 'tel', 'lat', 'lng'] return all(store.get(field) for field in required_fields)5. 性能优化技巧
- 缓存机制:对已查询的地理坐标建立本地缓存文件,减少API调用
- 断点续爬:记录已完成的页码,程序重启后自动继续
- 分布式扩展:使用Redis作为任务队列,支持多机协同爬取
- 增量更新:通过记录最后更新时间,只抓取新增或变更的门店
# 示例:使用SQLite实现简单缓存 import sqlite3 conn = sqlite3.connect('geo_cache.db') conn.execute('''CREATE TABLE IF NOT EXISTS locations (address TEXT PRIMARY KEY, lat REAL, lng REAL)''')6. 法律合规注意事项
- 严格遵守目标网站的robots.txt协议
- 数据采集后仅用于个人分析,禁止商业转售
- 控制请求频率,单城市不超过10页/分钟
- 公开成果时需脱敏处理关键信息
重要:商业使用前务必咨询法律意见,部分企业门店数据受法律保护
7. 完整项目结构建议
/store_crawler │── config.py # API密钥等配置 │── crawler.py # 主爬虫逻辑 │── utils.py # 工具函数 │── proxies.txt # 代理IP列表 │── cities.json # 城市编码映射 └── /data # 存储结果CSV实际部署时,推荐使用Docker容器化运行:
FROM python:3.9 WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . CMD ["python", "crawler.py"]8. 扩展应用场景
- 竞品分析:对比不同品牌在同一商圈的门店密度
- 选址评估:通过热力图分析优质门店的分布特征
- 市场预测:结合人口数据预测新店业绩
- 路线优化:为配送服务规划最佳路径
# 示例:使用folium生成热力图 import folium from folium.plugins import HeatMap m = folium.Map(location=[39.9, 116.3], zoom_start=12) heat_data = [[row['lat'], row['lng']] for _,row in df.iterrows()] HeatMap(heat_data).add_to(m) m.save('heatmap.html')这套方案经过多个真实项目验证,单机日均可采集10万+门店数据。建议根据具体目标网站调整逆向策略,核心思路具有通用性。数据采集只是第一步,真正的价值在于后续的商业分析应用。