dbms_sanya/python/plugins/file_browser/utils.py

import json
from multiprocessing.dummy import Pool as ThreadPool
from pathlib import Path
from random import choice

import requests
import yaml
from qgis.core import QgsSettings

TIANDITU_HOME_URL = "https://www.tianditu.gov.cn/"
PLUGIN_NAME = "tianditu-tools"

PluginDir = Path(__file__).parent

HEADER = {
    "User-Agent": "Mozilla/5.0 QGIS/32400/Windows 10 Version 2009",
    "Referer": "https://www.tianditu.gov.cn/",
}


def get_extramap_status():
    summary = load_yaml(PluginDir.joinpath("maps/summary.yml"))
    default_status = {}
    for section in summary:
        section_data = load_yaml(PluginDir.joinpath(f"maps/{section}.yml"))
        default_status[section] = list(section_data["maps"].keys())
    return default_status


class PluginConfig:
    def __init__(self):
        self.conf = QgsSettings()
        self.conf_name = "tianditu-tools"
        self.section_tianditu = f"{self.conf_name}/Tianditu"

    def init_config(self):
        # 初始化配置文件
        if not self.conf.contains("tianditu-tools/Tianditu/key"):
            print("初始化配置文件")
            # 初始化
            self.conf.setValue(f"{self.section_tianditu}/key", "")
            self.conf.setValue(f"{self.section_tianditu}/keyList", "")
            self.conf.setValue(f"{self.section_tianditu}/random", True)
            self.conf.setValue(f"{self.section_tianditu}/random_key", False)
            self.conf.setValue(f"{self.section_tianditu}/subdomain", "t0")
        if not self.conf.contains("tianditu-tools/Other/extramap_status"):
            print("初始化 extra map 文件")
            self.conf.setValue(
                f"{self.conf_name}/Other/extramap_status", str(get_extramap_status())
            )
        # 升级到保存多个key的版本
        if not self.conf.contains(f"{self.section_tianditu}/keyList"):
            self.conf.setValue(f"{self.section_tianditu}/random_key", False)
            # 保存原来的key
            if self.get_key() != "":
                self.conf.setValue(f"{self.section_tianditu}/keyList", self.get_key())
            else:
                self.conf.setValue(f"{self.section_tianditu}/keyList", "")

    def get_key_list(self):
        data_str = self.get_value("/Tianditu/keyList")
        if data_str == "":
            return []
        return data_str.split(",")

    def save_key_list(self, data_list):
        self.conf.setValue(f"{self.section_tianditu}/keyList", ",".join(data_list))

    def get_extra_maps_status(self):
        data = self.get_value("Other/extramap_status")
        return json.loads(data.replace("'", '"'))

    def set_extra_maps_status(self, data):
        self.conf.setValue(f"{self.conf_name}/Other/extramap_status", str(data))

    def get_value(self, name):
        return self.conf.value(f"{self.conf_name}/{name}")

    def get_bool_value(self, name):
        return self.conf.value(f"{self.conf_name}/{name}", type=bool)

    def set_value(self, name, value):
        self.conf.setValue(f"{self.conf_name}/{name}", value)

    def get_key(self):
        return self.get_value("Tianditu/key")

    def get_random_key(self):
        key_list = self.get_key_list()
        return choice(key_list)

    def set_key(self, key):
        key_to_set = ""
        if key is not None:
            key_to_set = key
        self.conf.setValue(f"{self.section_tianditu}/key", key_to_set)


def got(url, headers=None, timeout=6):
    try:
        response = requests.get(url, headers=headers, timeout=timeout)
    except (requests.exceptions.ConnectionError, requests.exceptions.Timeout):
        response = None
    return response


def tianditu_map_url(maptype: str, token: str, subdomain: str) -> str:
    """
    返回天地图url

    Args:
        maptype (str): 类型
        token (str): 天地图key
        subdomain (str): 使用的子域名

    Returns:
        str: 返回天地图XYZ瓦片地址
    """
    url = f"https://{subdomain}.tianditu.gov.cn/"
    url += (
        f"{maptype}_w/wmts?SERVICE=WMTS&REQUEST=GetTile&VERSION=1.0.0&LAYER={maptype}"
    )
    url += "&STYLE=default&TILEMATRIXSET=w&FORMAT=tiles&TileCol={x}&TileRow={y}&TileMatrix={z}"
    url += f"&tk={token}"
    return url


def check_url_status(url: str) -> object:
    """
    检查url状态
    Args:
        url (str): url

    Returns:
        object: {"code": 0}
        code:
            -1:网络异常
            0: 正常
            1: 非法key
            12: 权限类型错误
            1000: 未知错误
    """
    res = got(url, headers=HEADER)
    msg = {"code": 0}
    if res is not None:
        if res.status_code == 403:
            msg["code"] = res.json()["code"]  # 1:非法key 12:权限类型错误
            msg["msg"] = res.json()["msg"]
            msg["resolve"] = res.json()["resolve"]
        elif res.status_code == 200:
            msg["code"] = 0
        else:
            msg["code"] = 1000  # 未知错误
            msg["msg"] = "未知错误 "
            msg["resolve"] = f"错误代码:{res.status_code}"
    else:
        msg = {"code": -1, "msg": "网络错误", "resolve": "请检查网络连接"}
    return msg


def check_subdomain(url: str) -> int:
    """对子域名进行测速

    Args:
        url (str): 瓦片url

    Returns:
        int: 子域名对应的延迟数(毫秒), -1 表示连接失败
    """
    response = got(url, headers=HEADER, timeout=8)
    if response:
        millisecond = response.elapsed.total_seconds() * 1000
    else:
        millisecond = -1
    return int(millisecond)


def check_subdomains(url_list: list) -> list:
    """对子域名列表进行测速

    Args:
        url_list (list): 由不同子域名组成的瓦片url列表

    Returns:
        list: 每个子域名对应的延迟数(毫秒)组成的列表
    """
    pool = ThreadPool(4)
    ping_list = pool.map(check_subdomain, url_list)
    pool.close()
    pool.join()
    return ["❌" if x == -1 else f"{x} ms" for x in ping_list]


def load_yaml(file_path: Path):
    """
    读取YAML文件
    """
    with open(file_path, "r", encoding="utf-8") as f:
        return yaml.safe_load(f)


class TiandituAPI:
    """实现天地图搜索API"""

    def __init__(self, token: str):
        self.token = token
        self.header = HEADER

    def get(self, url: str, payload: dict) -> object:
        """实现get请求

        Args:
            url (str): url
            payload (dict): 传递参数

        Returns:
            object: {"code": 1为正常, -1为异常, "data": 请求数据}
        """
        timeout = 8
        try:
            res = requests.get(
                url, headers=self.header, params=payload, timeout=timeout
            )
            if res.ok:
                return {"code": 1, "data": res.json()}
            return {"code": -1, "message": f"请求失败 Status Code:{res.status_code}"}
        except TimeoutError as error:
            return {"code": -1, "message": str(error)}

    def api_search_v2(self, keyword: str, specify: str = None) -> object:
        """天地图地名搜索V2接口

        API说明: http://lbs.tianditu.gov.cn/server/search2.html

        Args:
            keyword (str): 搜索关键词
            specify (str, optional): 指定行政区的国标码 默认不传入

        Returns:
            object: 返回
        """
        #
        url = "http://api.tianditu.gov.cn/v2/search"
        data = {
            "keyWord": keyword,  # 搜索的关键字
            "mapBound": "-180,-90,180,90",  # 查询的地图范围(minx,miny,maxx,maxy) | -180,-90至180,90
            "level": 18,  # 目前查询的级别 | 1-18级
            "queryType": 1,  # 搜索类型 | 1:普通搜索（含地铁公交） 7：地名搜索
            "start": 0,  # 返回结果起始位（用于分页和缓存）默认0 | 0-300，表示返回结果的起始位置。
            "count": 10,  # 返回的结果数量（用于分页和缓存）| 1-300，返回结果的条数。
            "show": 1,  # 返回poi结果信息类别 | 取值为1，则返回基本poi信息;取值为2，则返回详细poi信息
        }
        if specify:
            data["specify"] = specify
        payload = {"postStr": str(data), "type": "query", "tk": self.token}
        return self.get(url, payload)

    def api_geocoder(self, keyword: str) -> object:
        """天地图地理编码接口

        API说明: http://lbs.tianditu.gov.cn/server/geocodinginterface.html

        Args:
            keyword (str): _description_

        Returns:
            object: _description_
        """
        url = "http://api.tianditu.gov.cn/geocoder"
        data = {
            "keyWord": keyword,  # 搜索的关键字
        }
        payload = {"ds": str(data), "tk": self.token}
        return self.get(url, payload)

    def api_regeocoder(self, lon: float, lat: float) -> object:
        """天地图逆地理编码接口

        API说明: http://lbs.tianditu.gov.cn/server/geocoding.html

        Args:
            lon (float): 纬度值
            lat (float): 经度值

        Returns:
            object: 逆地理编码数据
        """
        url = "http://api.tianditu.gov.cn/geocoder"
        data = {"lon": lon, "lat": lat, "ver": 1}
        payload = {"postStr": str(data), "type": "geocode", "tk": self.token}
        return self.get(url, payload)