一、技术背景与合规性说明

POI（Point of Interest）数据是地理信息系统（GIS）的核心要素，包含商户名称、地址、坐标、分类等关键信息。在商业分析、物流规划、城市研究等领域，高质量的POI数据是决策支持的基础。但需明确：直接爬取网页内容或未经授权的API调用可能违反服务条款，合法获取途径应通过官方地图API或数据服务接口。

百度地图开放平台提供标准化的Web服务API，开发者需申请开发者密钥（AK）并遵守《百度地图开放平台服务条款》。本文以合法API调用为核心，重点讨论技术实现细节。

二、API调用基础与认证机制

1. 密钥申请与权限配置

访问百度地图开放平台控制台，完成以下步骤：

注册开发者账号并创建应用
获取Access Key（AK）
配置服务权限（如Web服务API、Place API等）
绑定域名白名单（防止密钥泄露）

2. 基础请求结构

以POI检索API为例，典型请求URL格式如下：

https://api.map.baidu.com/place/v2/search?
  query={关键词}&
  location={经度,纬度}&
  radius=2000&
  output=json&
  ak={您的AK}&
  page_size=20&
  page_num=1

关键参数说明：

query：检索关键词（如”餐厅”）
location：中心点坐标（如”116.404,39.915”）
radius：检索半径（单位：米）
page_size：每页结果数（最大50）
page_num：页码

3. 响应数据解析

返回JSON包含以下核心字段：

{
  "status": 0,
  "message": "ok",
  "results": [
    {
      "name": "示例餐厅",
      "location": {
        "lat": 39.915,
        "lng": 116.404
      },
      "address": "北京市朝阳区...",
      "uid": "唯一标识符",
      "detail_info": {...}
    }
  ]
}

需特别处理：

状态码status=0表示成功
uid字段可用于后续详情查询
分页需循环请求直至results为空

三、反爬机制与应对策略

1. 常见限制

QPS限制：默认每秒2次请求（可申请提升）
日调用量限制：基础版每日5000次（企业版可扩容）
IP限制：同一IP异常请求可能被封禁

2. 优化方案

分布式代理池

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
class BaiduMapClient:
    def __init__(self, ak_list):
        self.session = requests.Session()
        retries = Retry(total=3, backoff_factor=1)
        self.session.mount('https://', HTTPAdapter(max_retries=retries))
        self.ak_pool = ak_list  # 多AK轮询
        self.current_ak_index = 0
    def get_poi(self, params):
        params['ak'] = self.ak_pool[self.current_ak_index % len(self.ak_pool)]
        self.current_ak_index += 1
        response = self.session.get(
            "https://api.map.baidu.com/place/v2/search",
            params=params
        )
        return response.json()

请求间隔控制

import time
import random
def safe_request(client, params, min_delay=1, max_delay=3):
    time.sleep(random.uniform(min_delay, max_delay))
    return client.get_poi(params)

四、数据存储与处理优化

1. 数据库设计建议

推荐使用PostgreSQL+PostGIS扩展：

CREATE TABLE poi_data (
    id SERIAL PRIMARY KEY,
    uid VARCHAR(64) UNIQUE NOT NULL,
    name VARCHAR(128) NOT NULL,
    address TEXT,
    location GEOGRAPHY(Point, 4326),
    category VARCHAR(32),
    province VARCHAR(32),
    city VARCHAR(32),
    district VARCHAR(32),
    update_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

2. 批量插入优化

import psycopg2
from psycopg2.extras import execute_batch
def batch_insert(data_list):
    conn = psycopg2.connect("dbname=poi user=postgres")
    cur = conn.cursor()
    try:
        execute_batch(cur,
            """INSERT INTO poi_data 
               (uid, name, address, lon, lat, category) 
               VALUES (%s, %s, %s, %s, %s, %s)""",
            [(d['uid'], d['name'], d['address'], 
              d['location']['lng'], d['location']['lat'], 
              d['detail_info']['type']) for d in data_list],
            page_size=100
        )
        conn.commit()
    finally:
        cur.close()
        conn.close()

五、性能优化实践

1. 空间索引加速

CREATE INDEX idx_poi_location ON poi_data USING GIST(location);
CREATE INDEX idx_poi_uid ON poi_data (uid);

2. 增量更新策略

def get_updated_poi(last_check_time):
    params = {
        'query': '',
        'bounds': '116.3,39.8,116.5,40.0',  # 矩形区域
        'timestamp': last_check_time,  # 仅返回更新数据
        'output': 'json'
    }
    # 实现逻辑...

3. 错误处理与重试机制

def robust_request(client, params, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = safe_request(client, params)
            if response['status'] == 0:
                return response
            elif response['status'] == 101:  # AK无效
                raise ValueError("Invalid AK")
            elif response['status'] == 110:  # 访问超限
                time.sleep(60 * (attempt + 1))
                continue
        except requests.exceptions.RequestException as e:
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)
    return None

六、进阶应用场景

1. 区域POI密度分析

import pandas as pd
from shapely.geometry import Point, Polygon
def calculate_density(poi_df, boundary_polygon):
    points = [Point(x, y) for x, y in zip(poi_df['lon'], poi_df['lat'])]
    within = [p.within(boundary_polygon) for p in points]
    return sum(within) / boundary_polygon.area  # 单位面积POI数

2. 竞品分布可视化

使用Pyecharts实现：

from pyecharts import options as opts
from pyecharts.charts import Geo
def visualize_competitors(poi_data):
    geo = Geo()
    geo.add_schema(maptype="北京")
    data = [(d['name'], [d['location']['lng'], d['location']['lat']]) 
            for d in poi_data]
    geo.add("竞品分布", data, type_="scatter")
    return geo.render("competitors.html")

七、最佳实践总结

合规优先：严格遵守API使用条款，避免法律风险
容错设计：实现完善的重试机制和日志记录
资源控制：合理设置QPS和并发数，避免被封禁
数据质量：建立数据校验流程，处理异常值
成本优化：根据需求选择合适的服务等级（免费版/企业版）

通过系统化的API调用、反爬应对、数据处理和性能优化，开发者可以构建稳定、高效的百度地图POI数据采集系统，为各类地理信息应用提供可靠的数据支持。

百度地图POI数据高效爬取策略与实现指南