You need to enable JavaScript to run this app.
最新活动
大模型
产品
解决方案
定价
生态与合作
支持与服务
开发者
了解我们

如何使用Python的requests模块批量爬取BC Assessment网站地图中的所有物业地址

如何使用Python的requests模块批量爬取BC Assessment网站地图中的所有物业地址

看起来你已经摸透了这个网站的ArcGIS API套路啦!当前你的代码已经能针对单个地图点位查询对应物业地址,要实现批量爬取,核心是从单点查询升级为范围遍历/全量图层请求,我来帮你梳理具体的优化思路和代码:

一、先理清楚当前代码的逻辑

你现在的代码做了这几件关键的事:

  • 访问目标页面获取ArcGIS服务的tokenmapserverUrl(这俩是调用API的核心凭证和端点)
  • 用单个经纬度点位(x,y)发起空间查询,返回对应点位的物业信息
  • 解析返回的JSON数据,提取出地址字段

要批量爬取,我们需要把「单点查询」改成「范围覆盖查询」,再处理API的分页限制,就能拿到指定区域内的所有物业地址了。

二、批量爬取的具体步骤和优化代码

1. 确定爬取的地理范围

首先你得明确要爬取哪块区域,比如维多利亚市的Web Mercator投影范围(可以从地图上拖拽边界,然后在开发者工具里抓包获取对应的xmin/ymin/xmax/ymax)。

2. 用范围查询替代单点查询

ArcGIS API支持用矩形范围(esriGeometryEnvelope)来查询该区域内的所有要素,比逐个点位查询效率高得多。同时要处理API的分页限制(默认最多返回1000条数据),通过resultOffsetresultRecordCount参数来分页获取全量数据。

3. 优化后的完整代码

import re
import json
import time
import urllib3
import requests
from pprint import pprint

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

def fetch_all_properties():
    # 目标页面链接
    link = 'https://www.bcassessment.ca/Property/Info/SjAwMDAwQzRZMQ=='
    headers = {
        'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
        'accept-language': 'en-US,en;q=0.9',
        'host': 'www.bcassessment.ca',
        'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36'
    }
    headers_ano = {
        'accept': '*/*',
        'accept-language': 'en-US,en;q=0.9',
        'host': 'arcgis.bcassessment.ca',
        'origin': 'https://www.bcassessment.ca',
        'referer': 'https://www.bcassessment.ca/',
        'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36'
    }

    with requests.Session() as session:
        # 第一步:获取token和API查询端点
        resp = session.get(link, headers=headers, verify=False)
        # 提取token和mapserver地址
        token = re.findall(r"gistoken[^']+'(.*?)';", resp.text)[0]
        query_url = re.findall(r"mapserverUrl[^']+'(.*?)';", resp.text)[0] + "/0/query"
        
        # 第二步:定义要爬取的地理范围(示例为维多利亚市某区域,你可以替换成自己的范围)
        # 注意:这个范围是Web Mercator投影(wkid:102100)的矩形边界
        min_x, min_y = -13745000, 6195000
        max_x, max_y = -13738000, 6202000
        geometry = json.dumps({
            'xmin': min_x,
            'ymin': min_y,
            'xmax': max_x,
            'ymax': max_y,
            'spatialReference': {'wkid': 102100, 'latestWkid': 3857}
        })

        # 初始化分页参数
        result_offset = 0
        result_record_count = 1000  # API默认最大单次返回量
        all_properties = []

        while True:
            params = {
                'f': 'json',
                'where': '1=1',  # 条件为所有要素
                'geometry': geometry,
                'returnGeometry': 'true',
                'geometryType': 'esriGeometryEnvelope',  # 用范围查询替代单点
                'inSR': '102100',
                'outFields': 'AFP_OID,UNIT_NUMBER,TOTAL_ASSESSED,ROLL,AREA_EVBC,JUR,TOTAL_LAND,TOTAL_BUILDING,ADDRESS,DESCRIPTION,STREET_NUMBER,STREET_NAME,OID_EVBC,SHORT_ADDRESS,IS_STRATA,FARM_FLAG,UTILITY_FLAG,MAJ_INDUSTRY_FLAG,MANAGED_FOREST_FLAG',
                'orderByFields': 'STREET_NAME,STREET_NUMBER,UNIT_NUMBER,SHORT_ADDRESS',
                'outSR': '102100',
                'token': token,
                'resultOffset': result_offset,
                'resultRecordCount': result_record_count
            }

            # 发起请求,加个延迟避免被封
            time.sleep(1)
            res = session.get(query_url, params=params, headers=headers_ano, verify=False)
            print(f"请求状态码: {res.status_code}, 当前偏移量: {result_offset}")

            if res.status_code != 200:
                print(f"请求失败,状态码: {res.status_code}")
                break

            data = res.json()
            features = data.get('features', [])
            if not features:
                # 没有更多数据了,退出循环
                break

            # 提取地址并保存
            for item in features:
                addr = item['attributes']['ADDRESS']
                all_properties.append(addr)
                print(addr)

            # 更新偏移量,准备下一页
            result_offset += result_record_count

        print(f"\n共爬取到 {len(all_properties)} 条物业地址")
        return all_properties

if __name__ == "__main__":
    fetch_all_properties()

三、关键注意事项

  • 地理范围调整:你需要替换代码里的min_x, min_y, max_x, max_y为你实际要爬取的区域边界,可以通过在网站地图上拖拽选择区域,然后在开发者工具的网络请求里抓取对应的边界参数。
  • 反爬策略:加time.sleep(1)避免请求过于频繁,不要短时间内发起大量请求,防止被网站封禁IP。
  • API限制:部分ArcGIS服务会有请求频率或数据量限制,如果遇到返回空数据或错误,检查参数是否正确,或者缩小爬取范围。
  • 合规性:请遵守BC Assessment网站的使用条款,不要将爬取的数据用于商业用途或违反规定的场景。

内容来源于stack exchange

火山引擎 最新活动