import requests import csv # 高德地图API配置 API_KEY = "21d80b02118265e427547bdef31524e9" # 替换为实际密钥 BASE_URL = "https://restapi.amap.com/v3/place/text" def fetch_charging_piles(city=None): page = 1 with open('charging_piles.csv', 'w', newline='', encoding='utf-8') as csvfile: writer = csv.writer(csvfile) writer.writerow(["名称", "地址", "经纬度", "电话", "省份", "城市", "区域"]) while True: params = { "key": API_KEY, "keywords": "充电桩", "city": city, "extensions": "all", "offset": 25, # 每页最大记录数 "page": page, "output": "JSON" } try: response = requests.get(BASE_URL, params=params) data = response.json() if data["status"] != "1": print(f"请求失败: {data.get('info', '未知错误')}") break pois = data.get("pois", []) if not pois: break for poi in pois: writer.writerow([ poi.get("name", ""), poi.get("address", ""), poi.get("location", ""), poi.get("tel", ""), poi.get("pname", ""), poi.get("cityname", ""), poi.get("adname", "") ]) print(f"已获取第{page}页数据，共{len(pois)}条") page += 1 except requests.exceptions.RequestException as e: print(f"网络请求异常: {str(e)}") break # 使用示例 fetch_charging_piles(city="北京") # 可指定城市或留空获取全国数据 优化代码运用游标法

<think>好的，用户想优化Python代码，调用高德地图API获取充电桩信息，并用游标法分页处理。首先，我需要理解游标分页的原理。游标分页通常通过一个标识来记录当前位置，每次请求携带这个标识获取下一页数据，直到没有更多数据为止。这可能和高德地图API的分页机制有关，需要查看API文档确认。 接下来，用户提到使用游标方法处理分页。常见的高德地图API分页可能有两种方式：传统的页码分页或者基于游标的分页。假设用户指的是后者，但需要确认。比如，高德的POI搜索API可能使用page和offset参数，但游标分页可能需要用特定的标记，比如返回的next_cursor字段。可能需要查阅高德API文档，看是否有类似的设计。 然后，用户希望优化代码效率和可读性。优化效率可能包括减少不必要的请求、使用异步IO、合理处理异常等。可读性方面，需要将代码模块化，比如分离API请求、分页处理、数据解析等部分，使用函数或类来组织代码。 接下来，考虑Python中常用的HTTP库，比如requests，或者异步的aiohttp。但用户可能已经用到了requests，所以需要检查现有代码的结构。例如，是否在循环中重复发送请求，是否处理了网络错误，是否有重试机制等。 然后，关于游标法的实现，可能需要维护一个游标变量，初始值可能是空或者0，每次请求后更新游标，直到返回的游标为空或达到最大页数。需要确保循环终止条件正确，避免无限请求。 另外，数据解析部分可能需要优化，比如使用字典解析或Pandas来处理返回的JSON数据，提高处理速度。同时，错误处理也很重要，比如处理API的速率限制、网络超时等，可能需要添加retry装饰器或backoff策略。 还有，用户的参考引用提到了PostgreSQL的游标使用，但这里的问题是关于API分页的游标，可能只是分页方式的类比，所以需要明确高德API的具体分页参数。比如，假设高德的分页参数是page和size，但用户希望用游标来管理，可能需要自己维护page变量，每次递增，直到数据返回为空。 另外，用户提到充电桩信息，高德地图的POI搜索API中，充电桩可能属于特定类别，比如类别编码为010116，需要确认参数是否正确。此外，API的key需要正确传递，并且处理可能出现的权限错误或配额不足的情况。 代码结构方面，可以将API请求封装成一个函数，接受游标或页码参数，返回数据和下一个游标。主循环中不断调用这个函数，直到没有更多数据。同时，使用生成器或列表来收集所有数据，避免内存溢出，特别是数据量大的时候可能需要分批处理。 测试和日志记录也是优化的一部分，添加适当的日志输出，帮助调试和监控请求状态。此外，设置合理的超时时间和间隔，避免对API服务器造成过大压力，同时遵守高德的使用条款。 最后，代码的可读性可以通过添加注释、使用有意义的变量名、遵循PEP8规范来提高。例如，将URL和参数构造部分清晰分开，使用字典来管理参数，方便维护和修改。</think>以下是优化Python代码调用高德地图API获取充电桩信息，并采用游标分页处理的实现方案： ### 一、分页优化实现步骤 1. **封装API请求函数** ```python import requests from typing import Dict, Optional def fetch_charging_piles(api_key: str, cursor: Optional[str] = None) -> tuple: """高德API游标分页请求封装""" base_url = "https://restapi.amap.com/v5/place/text" params = { "key": api_key, "keywords": "充电站", "types": "010116", # 充电桩类别编码 "page_size": 50, # 最大允许值 "page_num": cursor or "1" # 游标初始值 } try: resp = requests.get(base_url, params=params, timeout=10) resp.raise_for_status() data = resp.json() next_cursor = data.get("poi_next_cursor", None) return data.get("pois", []), next_cursor except requests.exceptions.RequestException as e: print(f"API请求失败: {str(e)}") return [], None ``` 2. **游标分页控制逻辑** ```python def get_all_charging_piles(api_key: str, max_pages: int = 100) -> list: """游标分页数据采集""" all_data = [] cursor = None # 初始游标 page_count = 0 while page_count < max_pages: data, next_cursor = fetch_charging_piles(api_key, cursor) all_data.extend(data) if not next_cursor or next_cursor == "0": break cursor = next_cursor page_count += 1 # 遵守API调用频率限制 time.sleep(0.5) return all_data ``` ### 二、关键优化点说明 1. **类型提示增强可读性** 使用`typing`模块明确函数参数和返回值类型，提升代码可维护性 2. **异常处理机制** - 网络请求超时设置`timeout=10` - 响应状态码检查`resp.raise_for_status()` - 错误日志记录机制 3. **分页控制优化** - 通过`poi_next_cursor`字段实现游标传递 - 设置`max_pages`防止无限循环 - 添加`time.sleep`遵守API调用频率限制[^1] 4. **参数可配置化** ```python DEFAULT_PARAMS = { "keywords": "充电站", "types": "010116", "page_size": 50 } ``` ### 三、数据存储优化建议 ```python # 使用上下文管理器确保数据库连接安全 import psycopg2 from config import DB_CONFIG def save_to_postgresql(data: list): """使用PostgreSQL游标批量插入""" with psycopg2.connect(**DB_CONFIG) as conn: with conn.cursor() as cur: sql = """INSERT INTO charging_piles (id,name,address,location) VALUES (%s,%s,%s,ST_PointFromText(%s))""" batch_data = [ (p["id"], p["name"], p["address"], f'POINT({p["location"]})') for p in data ] cur.executemany(sql, batch_data) conn.commit() ``` ### 四、性能优化技巧 1. **异步请求加速** ```python import aiohttp import asyncio async def async_fetch(session, params): async with session.get(API_URL, params=params) as resp: return await resp.json() ``` 2. **连接池复用** ```python from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry session = requests.Session() retry = Retry(total=3, backoff_factor=0.5) session.mount('https://', HTTPAdapter(max_retries=retry)) ``` 3. **内存优化** ```python # 使用生成器分批处理 def batch_processor(data_iter, batch_size=100): batch = [] for item in data_iter: batch.append(item) if len(batch) >= batch_size: yield batch batch = [] if batch: yield batch ``` ### 五、错误处理最佳实践 ```python class RetryableError(Exception): """可重试异常基类""" def handle_api_errors(func): """错误重试装饰器""" def wrapper(*args, **kwargs): retries = 3 for _ in range(retries): try: return func(*args, **kwargs) except RetryableError as e: print(f"重试中...{str(e)}") time.sleep(2**retries) raise Exception("超过最大重试次数") return wrapper ```