132 lines
4.9 KiB
Python
132 lines
4.9 KiB
Python
from typing import List
|
|
import requests as rq
|
|
from . import Schemas
|
|
from Utils.DataUtils import JsDict
|
|
|
|
|
|
def make_query_url(params_dict):
|
|
str_list = []
|
|
for key, v in params_dict.items():
|
|
if v is not None and v != "":
|
|
str_list.append(f"{key}={v}")
|
|
return "&".join(str_list)
|
|
|
|
|
|
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_url = "http://open.api.tianyancha.com/services/open/ic/holder/2.0"
|
|
# 实控人
|
|
actual_control_url = "http://open.api.tianyancha.com/services/open/ic/actualControl/3.0"
|
|
# 法律诉讼
|
|
law_suit_url = "http://open.api.tianyancha.com/services/open/jr/lawSuit/3.0"
|
|
# 舆情信息
|
|
news_url = "http://open.api.tianyancha.com/services/open/ps/news/2.0"
|
|
|
|
|
|
class TianyanchaApi:
|
|
"""
|
|
天眼查接口
|
|
"""
|
|
|
|
def __init__(self, token: str, config: TianyanchaConfig = TianyanchaConfig):
|
|
self.config = config
|
|
self.token = token
|
|
self.rq = rq
|
|
|
|
def get_data(self, api_url, keyword):
|
|
url = f"{api_url}?keyword={keyword}"
|
|
return self.get_data_core(url)
|
|
|
|
def get_data_core(self, full_url):
|
|
headers = {'Authorization': self.token}
|
|
res = self.rq.get(full_url, headers=headers)
|
|
data = res.json()
|
|
if data['error_code'] == 0:
|
|
return JsDict(data['result'])
|
|
|
|
def get_list(self, api_url, keyword):
|
|
"""
|
|
获取列表
|
|
"""
|
|
query_url = f"{api_url}?keyword={keyword}"
|
|
return self.get_list_core(query_url)
|
|
|
|
def get_list_core(self, query_url):
|
|
headers = {'Authorization': self.token}
|
|
page_num = 1
|
|
page_size = 20
|
|
item_list = []
|
|
# 查询第一页的获取total
|
|
url = f"{query_url}&pageNum={page_num}&pageSize={page_size}"
|
|
res = self.rq.get(url, headers=headers)
|
|
data = res.json()
|
|
if data['error_code'] == 0:
|
|
total = data['result']['total']
|
|
temp_list = [JsDict(item) for item in data['result']['items']]
|
|
item_list.extend(temp_list)
|
|
# todo:是否获取所有列表
|
|
# if total > len(item_list):
|
|
# for i in range(1, (total - len(item_list)) // page_size + 1):
|
|
# url = f"{api_url}?keyword={keyword}&pageNum={page_num}&pageSize={page_size}"
|
|
# res = self.rq.get(url, headers=headers)
|
|
# data = res.json()
|
|
# if data['error_code'] == 0:
|
|
# if data['result']['total']:
|
|
# temp_list = [JsDict(item) for item in data['result']['items']]
|
|
# item_list.extend(temp_list)
|
|
#
|
|
return item_list
|
|
|
|
def get_company_base_info(self, keyword) -> Schemas.CompanyBaseInfoRes:
|
|
return self.get_data(self.config.company_base_info_url, keyword)
|
|
|
|
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[Schemas.CompanyShareHolderItemInfo]:
|
|
"""
|
|
返回公司股东信息列表
|
|
"""
|
|
item_list = self.get_list(self.config.company_share_holder_url, keyword)
|
|
return item_list
|
|
|
|
def get_actual_control(self, keyword) -> List[Schemas.ActualControlItemInfo]:
|
|
"""
|
|
查询公司实控人，多个结果则为疑似
|
|
"""
|
|
url = f"{self.config.company_share_holder}?keyword={keyword}"
|
|
headers = {'Authorization': self.token}
|
|
res = self.rq.get(url, headers=headers)
|
|
data = res.json()
|
|
item_list = []
|
|
if data['error_code'] == 0:
|
|
item_list.extend([JsDict(item) for item in data['result']["actualControllerList"]])
|
|
return item_list
|
|
|
|
def get_law_suit(self, keyword) -> List[Schemas.LawSuitItemInfo]:
|
|
"""
|
|
查询公司法律诉讼信息列表
|
|
"""
|
|
item_list = self.get_list(self.config.law_suit_url, keyword)
|
|
return item_list
|
|
|
|
def get_news(self, name, startTime=None, endTime=None, tags=[]) -> List[Schemas.CompanyNewsItemInfo]:
|
|
query_url = make_query_url({"name": name, "startTime": startTime, "endTime": endTime, "tags": ",".join(tags)})
|
|
full_query_url = f"{self.config.news_url}?{query_url}"
|
|
item_list = self.get_list_core(full_query_url)
|
|
return item_list
|