urban-investment-research/ThirdPartyApi/Tianyancha/Apis.py

from typing import List

from .Schemas import CompanyBaseInfoRes, CompanyShareHolderItemInfo

from Utils.DataUtils import JsDict


class TianyanchaConfig:
    # 公司工商信息接口
    company_base_info_url = "http://open.api.tianyancha.com/services/open/ic/baseinfo/normal"
    # 公司搜索接口
    search_url = "http://open.api.tianyancha.com/services/open/search/2.0"
    # 公司股东信息获取接口
    company_share_holder = "http://open.api.tianyancha.com/services/open/ic/holder/2.0"


class TianyanchaApi:
    def __init__(self, token: str, config: TianyanchaConfig = TianyanchaConfig):
        self.config = config
        self.token = token

    def get_company_base_info(self, keyword) -> CompanyBaseInfoRes:
        url = f"{self.config.company_base_info_url}?keyword={keyword}"
        headers = {'Authorization': self.token}
        res = self.rq.get(url, headers=headers)
        data = res.json()
        if data['error_code'] == 0:
            return JsDict(data['result'])

    def company_name_search(self, keyword) -> List[str]:
        """
        返回公司名称列表
        """
        url = f"{self.config.search_url}?keyword={keyword}"
        headers = {'Authorization': self.token}
        res = self.rq.get(url, headers=headers)
        data = res.json()
        company_names = []
        if data['error_code'] == 0:
            company_names = [item["name"] for item in data['result']['items'] if item["matchType"] == '公司名称匹配']
        return company_names

    def get_company_share_holder(self, keyword) -> List[CompanyShareHolderItemInfo]:
        """
        返回公司股东信息列表
        """
        url = f"{self.config.company_share_holder}?keyword={keyword}"
        headers = {'Authorization': self.token}
        res = self.rq.get(url, headers=headers)
        data = res.json()
        holder_list = []
        if data['error_code'] == 0:
            holder_list = [JsDict(item) for item in data['result']]
        return holder_list