193 lines
7.4 KiB
Python
193 lines
7.4 KiB
Python
from datetime import datetime
|
|
from typing import List
|
|
import requests as rq
|
|
from . import Schemas
|
|
from Utils.DataUtils import JsDict
|
|
from Utils.MongoUtils import MongoConnect
|
|
|
|
|
|
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)
|
|
|
|
|
|
error_code_dic = {
|
|
0: "请求成功",
|
|
300000: "无数据",
|
|
300001: "请求失败",
|
|
300002: "账号失效",
|
|
300003: "账号过期",
|
|
300004: "访问频率过快",
|
|
300005: "无权限访问此api",
|
|
300006: "余额不足",
|
|
300007: "剩余次数不足",
|
|
300008: "缺少必要参数",
|
|
300009: "账号信息有误",
|
|
300010: "URL不存在",
|
|
300011: "此IP无权限访问此api",
|
|
300012: "报告生成中",
|
|
}
|
|
|
|
|
|
class TianyanchaApiDataType:
|
|
工商信息 = "工商信息"
|
|
股东信息列表 = "股东信息列表"
|
|
公司实控人列表 = "公司实控人列表"
|
|
法律诉讼信息列表 = "法律诉讼信息列表"
|
|
舆情信息列表 = "舆情信息列表"
|
|
组合工商信息 = "组合工商信息"
|
|
|
|
|
|
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_shareholder_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"
|
|
# 工商信息组合接口
|
|
full_base_info_url = "http://open.api.tianyancha.com/services/open/cb/ic/2.0"
|
|
|
|
|
|
class TianyanchaApi:
|
|
"""
|
|
天眼查接口
|
|
"""
|
|
|
|
def __init__(self, token: str, config: TianyanchaConfig = TianyanchaConfig,
|
|
mongo_connect: MongoConnect = None,
|
|
mongo_db_name="天眼查"
|
|
):
|
|
self.config = config
|
|
self.token = token
|
|
self.rq = rq
|
|
self.mongo_connect = mongo_connect
|
|
self.mongo_db_name = mongo_db_name
|
|
|
|
def get_data(self, api_url, keyword, query_info={}):
|
|
url = f"{api_url}?keyword={keyword}"
|
|
return self.get_data_core(url, query_info)
|
|
|
|
def get_data_core(self, full_url, query_info={}):
|
|
if self.mongo_connect:
|
|
item = self.mongo_connect.db[self.mongo_db_name].find_one({"url": full_url})
|
|
if item:
|
|
print("请求缓存成功")
|
|
return JsDict(item['data'])
|
|
headers = {'Authorization': self.token}
|
|
res = self.rq.get(full_url, headers=headers)
|
|
data = res.json()
|
|
if data['error_code'] == 0:
|
|
print("请求成功")
|
|
if self.mongo_connect:
|
|
self.mongo_connect.db[self.mongo_db_name].insert_one({
|
|
"url": full_url,
|
|
"data": data['result'],
|
|
"query_info": query_info,
|
|
"create_time": datetime.now(),
|
|
"update_time": datetime.now()
|
|
})
|
|
return JsDict(data['result'])
|
|
else:
|
|
print(full_url)
|
|
print(error_code_dic[data['error_code']])
|
|
# raise Exception(error_code_dic[data['error_code']])
|
|
|
|
def get_list(self, api_url, keyword, query_info={}):
|
|
"""
|
|
获取列表
|
|
"""
|
|
query_url = f"{api_url}?keyword={keyword}"
|
|
return self.get_list_core(query_url, query_info)
|
|
|
|
def get_list_core(self, query_url, query_info={}):
|
|
page_num = 1
|
|
page_size = 20
|
|
item_list = []
|
|
# 查询第一页的获取total
|
|
url = f"{query_url}&pageNum={page_num}&pageSize={page_size}"
|
|
result = self.get_data_core(url, query_info)
|
|
print(result)
|
|
total = min(result['total'], 50)
|
|
temp_list = [JsDict(item) for item in 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"{query_url}&pageNum={page_num}&pageSize={page_size}"
|
|
# result = self.get_data_core(url)
|
|
# temp_list = [JsDict(item) for item in 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,
|
|
query_info={"公司名称": keyword, "数据类型": TianyanchaApiDataType.工商信息})
|
|
|
|
def company_name_search(self, keyword) -> List[str]:
|
|
"""
|
|
返回公司名称列表
|
|
"""
|
|
url = f"{self.config.search_url}?word={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"] == '公司名称匹配']
|
|
else:
|
|
print(data)
|
|
return company_names
|
|
|
|
def get_company_shareholder(self, keyword) -> List[Schemas.CompanyShareHolderItemInfo]:
|
|
"""
|
|
返回公司股东信息列表
|
|
"""
|
|
item_list = self.get_list(self.config.company_shareholder_url, keyword,
|
|
query_info={"公司名称": keyword, "数据类型": TianyanchaApiDataType.股东信息列表})
|
|
return item_list
|
|
|
|
def get_actual_control(self, keyword) -> List[Schemas.ActualControlItemInfo]:
|
|
"""
|
|
查询公司实控人，多个结果则为疑似
|
|
"""
|
|
data = self.get_data(self.config.actual_control_url, keyword,
|
|
{"公司名称": keyword, "数据类型": TianyanchaApiDataType.公司实控人列表})
|
|
if data:
|
|
return data['actualControllerList']
|
|
|
|
def get_law_suit(self, keyword) -> List[Schemas.LawSuitItemInfo]:
|
|
"""
|
|
查询公司法律诉讼信息列表
|
|
"""
|
|
item_list = self.get_list(self.config.law_suit_url, keyword,
|
|
{"公司名称": keyword, "数据类型": TianyanchaApiDataType.法律诉讼信息列表})
|
|
return item_list
|
|
|
|
def get_news(self, name, start_time=None, end_time=None, tags=[]) -> List[Schemas.CompanyNewsItemInfo]:
|
|
"""
|
|
获取公司舆情信息
|
|
"""
|
|
query_url = make_query_url({"name": name, "startTime": start_time, "endTime": end_time, "tags": ",".join(tags)})
|
|
full_query_url = f"{self.config.news_url}?{query_url}"
|
|
item_list = self.get_list_core(full_query_url,
|
|
{"公司名称": name, "数据类型": TianyanchaApiDataType.舆情信息列表,
|
|
"start_time": start_time,
|
|
"end_time": end_time, "tags": tags})
|
|
return item_list
|
|
|
|
def get_full_base_info(self, keyword) -> Schemas.FullBaseInfo:
|
|
"""
|
|
组合获取工商信息
|
|
"""
|
|
return self.get_data(self.config.full_base_info_url, keyword, {"公司名称": keyword, "数据类型": "组合工商信息"})
|