python3 异步-协程版：常用高德方法总结

import requests
import time
import csv
from math import ceil
import asyncio
from aiohttp import ClientSession, TCPConnector, client_exceptions
from random import choice


class KEY:
    def __init__(self):
        self.keys = list(csv.reader(open(u'高德秘钥.csv', 'r')))
        self.keys_length = len(self.keys)
        self.i = 0

    def get_key(self):
        return self.keys[self.i][1]

    def next_key(self):
        self.i += 1
        if self.i == self.keys_length:
            print("秘钥耗尽")
            self.i = 0
        return self.get_key()

    def random_key(self):
        return choice(self.keys)[1]

    def check_out(self):
        lon = 104.12159
        lat = 30.65447
        while self.i < self.keys_length:
            print(self.i)
            base_url = 'https://restapi.amap.com/v3/assistant/coordinate/convert'
            _params_ = {'locations': '{},{}'.format(lon, lat),
                        'coordsys': 'gps', 'output': 'json', 'key': self.keys[self.i][1]}

            infocode = int(requests.get(base_url, _params_).json()['infocode'])
            if infocode == 10001:
                print(self.keys[self.i])

            self.i += 1

    def get_all_key(self):
        self.keys = list(csv.reader(open(u'高德秘钥.csv', 'r')))

    def drop(self, key):
        if key in self.keys:
            self.keys.remove(key)

    def get_len(self):
        return len(self.keys)

class Method:
    def __init__(self):
        self.keys = KEY()
        self.k = self.keys.random_key()
        self.event_loop = asyncio.get_event_loop()

    # 访问网址
    async def get_response(self, base_url_, params_):
        """
        :param base_url_:  网址
        :param params_: 访问网址的参数字典
        :return: 访问网址的响应内容
        """
        async with ClientSession(connector=TCPConnector(verify_ssl=False)) as session:
            while True:
                try:
                    async with session.get(base_url_, params=params_) as r:
                        response_ = await r.json()
                        infocode = int(response_['infocode'])
                        if infocode == 10000:
                            return response_
                        else:
                            print(infocode)
                        print("切换key,重新查询中")
                        self.k = self.keys.random_key()
                        params_['key'] = self.k
                except client_exceptions.ClientConnectorError:
                    print('wait a moment')
                    for i in range(60, 0, -1):
                        print("倒计时{}秒".format(i), flush=True, end="" * 10)
                        time.sleep(1)
                except Exception as e:
                    print(e)
                    for i in range(60, 0, -1):
                        print("倒计时{}秒".format(i), flush=True, end="" * 10)
                        time.sleep(1)

    # gps坐标转化高德坐标
    def exchange(self, coordinate):
        """
        :param coordinate: 坐标 经度，纬度|经度，纬度……
        :return: 高德坐标
        """
        base_url = 'https://restapi.amap.com/v3/assistant/coordinate/convert'
        task = []
        for c in coordinate:
            _params_ = {'locations': c, 'coordsys': 'gps', 'output': 'json', 'key': self.k}
            task.append(self.get_response(base_url, _params_))

        response = self.event_loop.run_until_complete(asyncio.gather(*task))
        result = []
        for r in response:
            result.append(r['locations'])
        return result

    # 获取行政区信息
    def get_region(self, coordinate_array,  batch=False):
        """
        :param coordinate_array: 坐标数组 【"经度，纬度"，……】
        :param batch: 批量查询
        :return: 省，市，区（县）,详细地址
        """

        # 坐标转换
        coordinate = []
        for c in coordinate_array:
            coordinate += [c]
        coordinate = self.exchange(coordinate)

        task = []
        for c in coordinate:
            base_url = 'https://restapi.amap.com/v3/geocode/regeo'
            _params_ = {'location': c, 'extensions': 'all',
                        'output': 'json', 'key': self.k, 'batch': str(batch).lower()}
            task.append(self.get_response(base_url, _params_))
        response_array = self.event_loop.run_until_complete(asyncio.gather(*task))

        for response in response_array:
            if not batch:
                province = response['regeocode']['addressComponent']['province']
                city = response['regeocode']['addressComponent']['city']
                district = response['regeocode']['addressComponent']['district']
                address = response['regeocode']['formatted_address']
                pois = [[r['type'].split(';')[0], float(r['distance'])] for r in response['regeocode']['pois'][:3]]

                yield province, city, district, address, pois
            else:
                regeocodes = response['regeocodes']
                result = []
                for r in regeocodes:
                    r['pois'] = sorted(r['pois'], key=lambda x: x['distance'])
                    result.append(
                        (
                            r['addressComponent']['province'],
                            r['addressComponent']['city'],
                            r['addressComponent']['district'],
                            r['formatted_address'],
                            [[i['type'].split(';')[0], float(i['distance'])] for i in r['pois'][:3]]
                        )
                    )
                yield result

    # 获取POI信息（多边形搜索--矩形搜索）
    def get_poi(self, coordinate_array, lon_gap, lat_gap, poi_type, output_poiname=False):
        """
        :param coordinate: 【【经度，纬度】，【经度，纬度】】 矩形的中心经纬度
        :param lon_gap: 矩形经度间隔
        :param lat_gap: 矩形纬度间隔
        :param poi_type: 查询POI类型
        :param output_poiname: 输出POI name
        :return: 相应poi数量
        """
        # 坐标转换
        coordinate = []
        for c in coordinate_array:
            coordinate += [c[0]-lon_gap/2, c[1]+lat_gap/2, c[0]+lon_gap/2, c[1]-lat_gap/2]
        coordinate = self.exchange(coordinate)

        base_url = 'https://restapi.amap.com/v3/place/polygon'

        task = []
        for i in range(0, len(coordinate), 4):
            params = {'polygon': '{},{}|{},{}'.format(coordinate[i], coordinate[i+1], coordinate[i+2], coordinate[i+3]),
                      'types': poi_type, 'output': 'json', 'key': self.keys.random_key(), 'page': 1, 'offset': 20}

            task.append(self.get_response(base_url, params))
        response = self.event_loop.run_until_complete(asyncio.gather(*task))

        for r in response:
            n = int(r['count'])
            task = []
            if output_poiname:
                print('Crawling POI {} name'.format(poi_type))
                name = ['{}({})'.format(e['name'], e['typecode']) for e in r['pois']]
                while params['page'] < ceil(n/20):
                    params['page'] += 1
                    task.append(self.get_response(base_url, params))
                result = self.event_loop.run_until_complete(asyncio.gather(*task))
                for res in result:
                    name += ['{}({})'.format(e['name'], e['typecode']) for e in res['pois']]

                yield n, name
            else:
                yield n

    def get_distance(self, coordinate_array):
        """
        :param coordinate_array: [o_lon, o_lat, d_lon, d_lat]
        :return:
        """
        # 坐标转换
        coordinate = []
        for c in coordinate_array:
            coordinate += c
        coordinate = self.exchange(coordinate)

        base_url = 'https://restapi.amap.com/v3/direction/driving'
        task = []

        for i in range(0, len(coordinate), 2):
            parameters = {
                'key': self.keys.random_key(), 'strategy': 2, 'nosteps': 1, 'extensions': 'base',
                'origin': coordinate[i],
                'destination': coordinate[i+1]
            }
            task.append(self.get_response(base_url, parameters))
        response = self.event_loop.run_until_complete(asyncio.gather(*task))

        for r in response:
            yield float(r['route']['paths'][0]['distance'])