
简介这份资源面向具备一定Python基础的开发者与数据采集学习者围绕高德地图与58租房两个典型场景提供从地图数据获取到房源信息抓取的完整源码实现可用于地理信息处理、爬虫入门与Web可视化等练习。压缩包共11个文件约4.41MB包含2个Python脚本分别承担地图与租房数据抓取逻辑1个HTML页面与1个CSV数据文件用于结果展示与存储另有3个XML配置、2个DOC说明文档及TXT、IML等辅助文件覆盖代码、配置与使用说明各环节。资源已有1311人学习下载说明其在实际练习中具备一定参考价值。读者可借此了解高德地图接口调用方式、58租房页面数据解析思路、抓取结果落盘与前端展示的衔接流程并对照配置说明文档完成环境搭建与运行调试适合作为爬虫与地图应用方向的练手项目。1. 用 Python 调高德地图从地址解析到路径规划一套能跑通的工程方案做本地生活、物流调度、门店管理这类系统时绕不开一个需求把一堆文字地址变成经纬度再算算两点之间开车要多久。手动查几百条数据就废了。这时候 Python 配合高德地图开放平台的 Web 服务 API就是最省事的路径。它不需要你懂前端地图渲染只要会发 HTTP 请求、会解析 JSON就能把地理编码、逆地理编码、路径规划、POI 搜索这些能力嵌进自己的脚本或后端服务里。我见过不少团队一开始想自己搭地图服务折腾两周后乖乖回来调 API——原因很简单数据更新和维护成本根本扛不住。这篇笔记就按我实际做过的项目节奏把 key 申请、请求封装、批量处理、配额控制、坐标系坑点一次讲清楚新手能照着跑熟手能直接拿去改。2. 动手之前把高德 Web 服务 API 的账算明白2.1 为什么选 Web 服务 API 而不是 JS SDK高德开放平台提供好几套东西JavaScript SDK 主要给浏览器里嵌地图用Android/iOS SDK 给原生 App 用而 Web 服务 API 是一组 HTTP 接口专门给后端或者脚本调用。如果你的场景是「后台批量处理地址」「定时算配送距离」「给数据库补经纬度」那 Web 服务 API 是唯一合理的选择。它不依赖浏览器环境不挑操作系统Python 的 requests 库就能直接发请求。我一般会先确认三件事第一你的请求是服务端发起的不是给用户浏览器直接调的第二你不需要在地图上画标记、画轨迹只需要数据第三你能接受一定的 QPS 限制。这三条满足就走 Web 服务 API。反过来如果你要做的是「用户打开网页看到地图上有个点」那还是老老实实上 JS SDK别用后端代理去绕绕出来的体验很别扭。2.2 三类 Key 的区别与申请路径高德的控制台里创建应用时会让你选 Key 的类型。常见的有「Web 服务」「Web 端JS API」「Android 平台」「iOS 平台」。做 Python 后端调用必须选Web 服务类型。选错了会直接报INVALID_USER_SCODE或者USERKEY_PLAT_NOMATCH这个坑我踩过——当时图省事用了一个 JS API 的 Key 去调地理编码接口返回的错误码查了半天文档才对上号。申请流程大致是注册账号 → 实名认证 → 控制台创建应用 → 添加 Key → 绑定服务类型。个人开发者有免费配额企业认证后配额更高。具体数字高德会调整以控制台实时显示为准我不在这里写死。重点提醒一句Key 不要硬编码在代码里提交到仓库用环境变量或者配置文件加载这是基本的安全习惯。2.3 配额、QPS 与并发策略的取舍免费配额通常按「日调用量」和「并发量」两个维度限制。日调用量好理解就是一天最多调多少次并发量指的是同一秒内能同时处理多少个请求。很多人只盯着日调用量结果脚本一跑就报CUQPS_HAS_EXCEEDED_THE_LIMIT这就是并发超了。我的做法是单线程串行 固定间隔最稳。如果数据量大用线程池但把并发数压到 2~3并且在每次请求之间加time.sleep。不要一上来就开 20 个线程猛冲封禁风险不说错误处理也会变得很难看。下面是一个最小可用的请求封装把 Key 读取、超时、重试都包进去import os import time import requests # 从环境变量读取 Key避免硬编码 AMAP_KEY os.environ.get(AMAP_WEB_KEY) BASE_URL https://restapi.amap.com/v3 def amap_get(path, params, max_retry3): 统一的高德 Web 服务请求封装 :param path: 接口路径如 /geocode/geo :param params: 业务参数 dict不含 key :param max_retry: 失败重试次数 params dict(params) params[key] AMAP_KEY for attempt in range(max_retry): try: resp requests.get(BASE_URL path, paramsparams, timeout5) data resp.json() # 高德返回 status1 表示成功0 表示失败 if data.get(status) 1: return data # 配额或并发问题退避后重试 if data.get(infocode) in (10003, 10004, 10021): time.sleep(1.5 * (attempt 1)) continue # 其他业务错误直接抛出不浪费时间重试 raise RuntimeError(fAPI error: {data.get(infocode)} {data.get(info)}) except requests.RequestException as e: if attempt max_retry - 1: raise time.sleep(1.0) raise RuntimeError(max retry exceeded)这段代码里几个参数值得说清楚。timeout5是必须的不设超时的话遇到网络抖动脚本会挂死。infocode是高德的错误码10003是日配额超限10004是并发超限10021是 QPS 超限这三个都适合退避重试。其他错误比如地址解析不出来重试也没用直接抛出来让上层处理。max_retry3是我常用的值再多就说明不是偶发问题了。3. 地理编码与逆地理编码地址和经纬度互转的完整实现3.1 地理编码接口的参数与返回结构地理编码就是把「某市某区某路某号」这种文字地址转成经纬度。接口路径是/geocode/geo核心参数有这几个参数是否必填说明address是结构化地址越完整越准city否指定城市能显著提高重名地址的命中率batch否设为 true 可批量但单次上限受限制output否默认 JSON不用改返回的geocodes是一个数组每个元素里有location字段格式是「经度,纬度」。注意顺序是经度在前。我见过有人把经纬度搞反了结果点飘到几千公里外排查半天。def geocode(address, cityNone): 单条地址转经纬度返回 (lng, lat) 或 None params {address: address} if city: params[city] city data amap_get(/geocode/geo, params) geocodes data.get(geocodes, []) if not geocodes: return None location geocodes[0].get(location) if not location: return None lng, lat location.split(,) return float(lng), float(lat)逻辑很直白拿到geocodes数组取第一个结果的location按逗号切开转 float。这里有个细节——geocodes可能为空数组说明地址没解析出来这时候返回 None 让调用方决定怎么处理不要硬塞一个默认坐标那会污染数据。3.2 批量地址处理控制节奏比写代码更重要批量处理是真正见功力的地方。假设你有一万条地址要转坐标直接 for 循环调接口大概率跑到一半就被限流。我的策略是分三步先去重再分批最后带检查点。去重是因为很多地址是重复的比如同一个商圈的不同门店地址字符串可能完全一样。用一个 dict 缓存已经查过的地址能省下大量配额。分批是每处理 N 条就落一次盘万一脚本崩了不用从头再来。检查点就是记录处理到哪了。import json import time def batch_geocode(address_list, cityNone, cache_filegeo_cache.json, interval0.15): 批量地理编码带缓存和断点续跑 :param address_list: 地址字符串列表 :param city: 统一城市限定 :param cache_file: 缓存文件路径 :param interval: 每次请求间隔秒数 # 加载已有缓存 try: with open(cache_file, r, encodingutf-8) as f: cache json.load(f) except FileNotFoundError: cache {} results {} for idx, addr in enumerate(address_list): if addr in cache: results[addr] cache[addr] continue try: coord geocode(addr, citycity) except Exception as e: print(f[{idx}] {addr} 失败: {e}) coord None cache[addr] coord results[addr] coord # 每 50 条落一次盘 if idx % 50 0: with open(cache_file, w, encodingutf-8) as f: json.dump(cache, f, ensure_asciiFalse) time.sleep(interval) # 最终落盘 with open(cache_file, w, encodingutf-8) as f: json.dump(cache, f, ensure_asciiFalse) return resultsinterval0.15是我在免费配额下常用的值大概每秒 6~7 次请求不会触发 QPS 限制。如果你是企业配额可以适当调小但别低于 0.05。缓存文件用 JSON 存好处是可读、可手动修坏处是数据量特别大时加载慢。超过十万条建议换 SQLite这个后面讲。3.3 逆地理编码从坐标反查地址与 POI逆地理编码是反过来的给经纬度返回附近的地址描述、POI 列表、行政区划。接口是/geocode/regeo核心参数是location格式「经度,纬度」。还有一个extensions参数设为all会返回更详细的 POI 和道路信息设为base只返回基本地址。def regeo(lng, lat, extensionsbase): 逆地理编码返回格式化地址字符串 params { location: f{lng},{lat}, extensions: extensions } data amap_get(/geocode/regeo, params) regeocode data.get(regeocode, {}) return regeocode.get(formatted_address)这个接口我主要用来做数据校验——比如用户填的地址解析出来的坐标反查回去看看是不是同一个地方偏差太大就说明地址有问题。extensionsall返回的pois数组可以用来做「附近有什么」的功能但注意它返回的 POI 数量有限别指望拿它当完整的 POI 数据库用。4. 路径规划与距离测量驾车、步行、骑行怎么选4.1 四种出行方式的接口差异高德的路径规划接口按出行方式分了四个驾车/direction/driving、步行/direction/walking、骑行/direction/bicycling、公交/direction/transit/integrated。参数结构大同小异都是传起点和终点的经纬度返回距离、耗时、路线步骤。区别在于驾车接口支持strategy参数可以选最快路线、最短路线、避开拥堵等公交接口必须传city参数因为不同城市的公交数据是分开的步行和骑行接口相对简单没有太多策略可选。def driving_route(origin, destination, strategy0): 驾车路径规划 :param origin: (lng, lat) 起点 :param destination: (lng, lat) 终点 :param strategy: 0速度优先 1费用优先 2距离优先 :return: (距离米, 耗时秒) params { origin: f{origin[0]},{origin[1]}, destination: f{destination[0]},{destination[1]}, strategy: strategy, extensions: base } data amap_get(/direction/driving, params) paths data.get(route, {}).get(paths, []) if not paths: return None, None path paths[0] return int(path[distance]), int(path[duration])strategy这个参数值得展开说。默认 0 是速度优先适合大多数配送场景1 是费用优先会尽量避开收费路段2 是距离优先但实际耗时可能更长。我做过一个同城配送的项目一开始用默认策略后来发现有些订单算出来的时间比实际短很多换成考虑实时路况的策略后才对上。不过实时路况需要额外的服务权限免费配额下不一定能用这个要看你自己的账号情况。4.2 批量算距离矩阵接口比循环调用省配额如果你要算的是「一个起点到多个终点」或者「多个起点到多个终点」的距离千万别循环调路径规划接口。高德有个专门的矩阵接口/distance一次请求能返回多个起点终点组合的距离和耗时配额消耗比逐个调用低得多。def distance_matrix(origins, destinations, type_1): 距离测量矩阵 :param origins: 起点列表最多 100 个 :param destinations: 终点列表最多 100 个 :param type_: 1驾车 0直线 3步行 :return: 嵌套列表results[i][j] 是 origins[i] 到 destinations[j] 的距离 params { origins: |.join(f{lng},{lat} for lng, lat in origins), destination: |.join(f{lng},{lat} for lng, lat in destinations), type: type_ } data amap_get(/distance, params) results data.get(results, []) matrix [] for item in results: row [int(e[distance]) for e in item.get(elements, [])] matrix.append(row) return matrix注意origins和destination这两个参数名不对称——起点是复数origins终点是单数destination这是高德接口的历史遗留写错了会报参数错误。单次请求的坐标数量上限是 100超了要分批。返回的elements里除了distance还有duration按需取用。4.3 坐标系问题GCJ-02 与 WGS-84 的转换这是最容易翻车的地方。高德用的是 GCJ-02 坐标系也就是俗称的火星坐标。如果你手上的经纬度是从 GPS 设备或者某些国际标准地图来的那是 WGS-84直接丢给高德接口会偏移几百米。反过来高德返回的坐标拿去跟 WGS-84 的数据叠加也会对不上。常见做法是写一个转换函数在数据入口和出口做一次转换。网上有成熟的转换算法核心是一个非线性偏移公式。我一般会在项目里放一个coord_convert.py提供wgs84_to_gcj02和gcj02_to_wgs84两个函数所有进出高德接口的坐标都过一遍。import math PI 3.1415926535897932384626 A 6378245.0 EE 0.00669342162296594323 def _transform_lat(lng, lat): ret -100.0 2.0 * lng 3.0 * lat 0.2 * lat * lat \ 0.1 * lng * lat 0.2 * math.sqrt(abs(lng)) ret (20.0 * math.sin(6.0 * lng * PI) 20.0 * math.sin(2.0 * lng * PI)) * 2.0 / 3.0 ret (20.0 * math.sin(lat * PI) 40.0 * math.sin(lat / 3.0 * PI)) * 2.0 / 3.0 ret (160.0 * math.sin(lat / 12.0 * PI) 320 * math.sin(lat * PI / 30.0)) * 2.0 / 3.0 return ret def _transform_lng(lng, lat): ret 300.0 lng 2.0 * lat 0.1 * lng * lng \ 0.1 * lng * lat 0.1 * math.sqrt(abs(lng)) ret (20.0 * math.sin(6.0 * lng * PI) 20.0 * math.sin(2.0 * lng * PI)) * 2.0 / 3.0 ret (20.0 * math.sin(lng * PI) 40.0 * math.sin(lng / 3.0 * PI)) * 2.0 / 3.0 ret (150.0 * math.sin(lng / 12.0 * PI) 300.0 * math.sin(lng / 30.0 * PI)) * 2.0 / 3.0 return ret def wgs84_to_gcj02(lng, lat): WGS-84 转 GCJ-02 dlat _transform_lat(lng - 105.0, lat - 35.0) dlng _transform_lng(lng - 105.0, lat - 35.0) radlat lat / 180.0 * PI magic math.sin(radlat) magic 1 - EE * magic * magic sqrtmagic math.sqrt(magic) dlat (dlat * 180.0) / ((A * (1 - EE)) / (magic * sqrtmagic) * PI) dlng (dlng * 180.0) / (A / sqrtmagic * math.cos(radlat) * PI) return lng dlng, lat dlat这个转换不是可逆的精确变换gcj02_to_wgs84一般用迭代逼近或者直接反向减掉偏移量精度够用就行。关键是要统一整个项目里只允许一种坐标系在流转在边界处转换不要一会儿转一会儿不转那样数据会乱成一锅粥。5. 避坑与排查那些让我加班到凌晨的报错5.1 返回 INVALID_USER_KEY 或 USERKEY_PLAT_NOMATCH现象是接口直接返回错误码连业务数据都没有。原因通常是 Key 类型不对或者 Key 被删了、被禁用了。USERKEY_PLAT_NOMATCH专门指 Key 的平台类型和调用的接口不匹配比如用 JS API 的 Key 调 Web 服务接口。解决办法是去控制台确认 Key 的服务类型重新建一个 Web 服务的 Key。另外注意Key 绑定域名或 IP 白名单后从别的机器调用也会失败本地调试时要么不绑要么把当前出口 IP 加进去。5.2 地址解析结果为空或偏差巨大现象是geocodes返回空数组或者返回的坐标明显不对。原因有几个地址太模糊比如只写「某某路」没有门牌号城市重名比如「城关区」好几个城市都有地址里有特殊字符或全角半角混用。解决办法是尽量补全地址加上城市限定参数city并且对地址做预处理——去掉多余空格、统一标点、把「号」后面的内容保留完整。如果还是不行可以先用 POI 搜索接口找到相关地点再用那个地点的坐标这条路比硬解析地址靠谱。5.3 并发超限导致批量任务中断现象是脚本跑着跑着开始大量报CUQPS_HAS_EXCEEDED_THE_LIMIT然后整个任务卡住。原因是请求发得太快超过了账号的并发配额。解决办法是降并发、加间隔并且把重试逻辑写对。我一般会把并发控制在 2 以内间隔 0.2 秒以上。如果数据量实在大考虑分时段跑比如晚上跑一批白天跑一批避开自己的业务高峰期。另外矩阵接口和路径规划接口的配额是分开算的别以为地理编码没用完就能随便调路径规划。5.4 坐标系不统一导致距离计算错误现象是算出来的距离明显偏大或偏小或者路径规划给出的路线绕了一大圈。原因就是起点和终点用了不同的坐标系一个 GCJ-02 一个 WGS-84高德按 GCJ-02 去算自然对不上。解决办法是在数据入库时就标记坐标系来源所有进入高德接口的坐标统一转成 GCJ-02。我习惯在数据库里加一个coord_sys字段值为gcj02或wgs84查询时按需转换这样追溯问题很方便。5.5 缓存文件损坏导致断点续跑失败现象是脚本重启后加载缓存报 JSON 解析错误整个任务从头开始。原因是上次写入缓存时脚本被强制中断文件写了一半。解决办法是写缓存时先写临时文件写完再原子替换或者用 SQLite 代替 JSONSQLite 有事务保护不会出现半截文件。如果已经损坏了手动打开文件看看能不能截取到最后一个完整的 JSON 对象能救则救不能救就删了重跑。这个坑我踩过一次丢了几千条已解析的数据后来全部改用 SQLite 存缓存。6. 进阶技巧把配额用在刀刃上6.1 用 SQLite 做本地地理缓存JSON 缓存适合小数据量超过几万条之后加载和保存都会变慢。换成 SQLite 之后查询和写入都是常数级开销而且支持并发读。建一张表字段就三个address主键、lng、lat再加一个updated_at记录时间。每次解析前先查表命中就直接返回没命中再调接口调完写回。这样重复地址永远不会浪费配额。import sqlite3 def init_cache_db(db_pathgeo_cache.db): conn sqlite3.connect(db_path) conn.execute( CREATE TABLE IF NOT EXISTS geo_cache ( address TEXT PRIMARY KEY, lng REAL, lat REAL, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) conn.commit() return conn def get_from_cache(conn, address): cur conn.execute(SELECT lng, lat FROM geo_cache WHERE address ?, (address,)) row cur.fetchone() return (row[0], row[1]) if row else None def save_to_cache(conn, address, lng, lat): conn.execute( INSERT OR REPLACE INTO geo_cache (address, lng, lat) VALUES (?, ?, ?), (address, lng, lat) ) conn.commit()用INSERT OR REPLACE是为了处理地址重复但坐标更新的情况。updated_at字段可以用来做过期清理比如超过 30 天的缓存重新解析一次保证数据新鲜度。6.2 矩阵接口的坐标数量上限与分批策略矩阵接口单次最多 100 个起点乘 100 个终点但实际用的时候别真塞满因为返回的数据量很大解析也耗时。我一般控制在 20×20 以内超过就分批。分批的逻辑是按起点切块每块 20 个起点终点全量传进去这样一次请求能覆盖 20 个起点到所有终点的距离。如果终点也超过 100那就双向分批写个嵌套循环。def batch_matrix(origins, destinations, chunk_size20): 分批调用矩阵接口返回完整矩阵 full_matrix [] for i in range(0, len(origins), chunk_size): chunk origins[i:i chunk_size] sub_matrix distance_matrix(chunk, destinations) full_matrix.extend(sub_matrix) time.sleep(0.3) # 批次之间留间隔 return full_matrixchunk_size20是我实测比较稳的值再大容易触发超时。time.sleep(0.3)是给接口喘息的时间别省这点等待。6.3 用路径规划接口做配送顺序优化有了距离矩阵之后可以做一个简单的配送顺序优化。比如一个骑手要送 10 个点从站点出发怎么排顺序总距离最短。这是典型的旅行商问题精确解算不动但用贪心算法就能得到一个不错的近似解每次从当前位置选最近的未访问点走过去再选下一个。def greedy_route(start_idx, matrix): 贪心配送顺序 :param start_idx: 起点在矩阵中的索引 :param matrix: 距离矩阵matrix[i][j] 是 i 到 j 的距离 :return: 访问顺序列表 n len(matrix) visited {start_idx} route [start_idx] current start_idx while len(visited) n: # 找最近的未访问点 next_idx min( (j for j in range(n) if j not in visited), keylambda j: matrix[current][j] ) route.append(next_idx) visited.add(next_idx) current next_idx return route这个贪心解不保证最优但比随机顺序强太多。实际项目中我会把这个顺序作为初始解再用 2-opt 做局部优化交换两条边看总距离是否下降下降就保留。迭代几十轮之后结果通常能接近最优解的 90% 以上。注意矩阵接口返回的距离是驾车距离如果骑手是骑车要用骑行接口重新算别混用。6.4 监控配额消耗与告警最后说一个容易被忽略的点配额监控。免费配额用完了不会自动提醒等到接口报错才发现就晚了。我的做法是在请求封装里加一个计数器每次成功调用就累加写到本地文件或者 Redis 里。每天定时检查超过阈值就发告警。阈值设多少看你账号的日配额一般设到 80% 就提醒留出缓冲。import datetime def check_quota(usage_filequota_usage.json, daily_limit5000): 检查当日配额使用情况 today datetime.date.today().isoformat() try: with open(usage_file, r) as f: usage json.load(f) except FileNotFoundError: usage {} count usage.get(today, 0) if count daily_limit * 0.8: print(f警告今日已调用 {count} 次接近上限 {daily_limit}) return count这个计数器要跟请求封装绑在一起每次amap_get成功返回就加一。别用多线程同时写文件会冲突用个锁或者干脆写到 Redis 的原子计数器里。做地理相关的开发最深的体会就是接口本身不难难的是数据质量和配额管理。我见过太多项目在坐标系上翻车也见过因为没做缓存把免费配额一天跑光的。把上面这些封装和检查点做好后面换任何地图服务商迁移成本都很低。希望帮到你。本文还有配套的精品资源点击获取