321 lines
11 KiB
Python
321 lines
11 KiB
Python
"""
|
|
接口数据获取类
|
|
"""
|
|
|
|
import json
|
|
import math
|
|
from copy import deepcopy
|
|
from datetime import datetime
|
|
from typing import Dict, Any, Union, List
|
|
|
|
import requests
|
|
import requests as rq
|
|
from utils.data_utils import JsDict
|
|
from utils.mongo_utils import MongoConnect
|
|
|
|
|
|
def make_query_url(url, params_dict: dict):
|
|
str_list = []
|
|
for key, v in params_dict.items():
|
|
if v is not None and v != "":
|
|
str_list.append(f"{key}={v}")
|
|
add_content = "&".join(str_list)
|
|
return f"{url}?{add_content}"
|
|
|
|
|
|
error_code_dic = {
|
|
0: "请求成功",
|
|
300000: "无数据",
|
|
300001: "请求失败",
|
|
300002: "账号失效",
|
|
300003: "账号过期",
|
|
300004: "访问频率过快",
|
|
300005: "无权限访问此api",
|
|
300006: "余额不足",
|
|
300007: "剩余次数不足",
|
|
300008: "缺少必要参数",
|
|
300009: "账号信息有误",
|
|
300010: "URL不存在",
|
|
300011: "此IP无权限访问此api",
|
|
300012: "报告生成中",
|
|
}
|
|
|
|
api_config = {
|
|
"股权变更": {
|
|
"type": "list",
|
|
"url": "http://open.api.tianyancha.com/services/open/ic/holderChange/2.0"
|
|
},
|
|
# 注意参数是name
|
|
"新闻舆情": {
|
|
"type": "list",
|
|
"url": "http://open.api.tianyancha.com/services/open/ps/news/2.0"
|
|
},
|
|
"司法风险": {
|
|
"type": "dict",
|
|
"url": "http://open.api.tianyancha.com/services/open/cb/judicial/2.0"
|
|
},
|
|
"司法解析": {
|
|
"type": "list",
|
|
"url": "http://open.api.tianyancha.com/services/open/jr/judicialCase/2.0"
|
|
},
|
|
"资质证书": {
|
|
"type": "list",
|
|
"url": "http://open.api.tianyancha.com/services/open/m/certificate/2.0"
|
|
},
|
|
"历史经营异常": {
|
|
"type": "list",
|
|
"url": "http://open.api.tianyancha.com/services/open/hi/abnormal/2.0"
|
|
},
|
|
"双随机抽查": {
|
|
"type": "list",
|
|
"url": "http://open.api.tianyancha.com/services/open/m/doubleRandomCheck/2.0"
|
|
},
|
|
"双随机抽查详情": {
|
|
"type": "list",
|
|
"url": "http://open.api.tianyancha.com/services/open/m/doubleRandomCheckDetail/2.0"
|
|
},
|
|
"重要股东质押": {
|
|
"type": "list",
|
|
"url": "http://open.api.tianyancha.com/services/open/mr/stockPledge/shareholder/2.0"
|
|
},
|
|
"重要股东质押详情": {
|
|
"type": "dict",
|
|
"url": "http://open.api.tianyancha.com/services/open/mr/stockPledge/shareholder/detail/2.0"
|
|
},
|
|
"破产重整": {
|
|
"type": "list",
|
|
"url": "http://open.api.tianyancha.com/services/open/jr/bankruptcy/2.0"
|
|
},
|
|
"司法拍卖": {
|
|
"type": "list",
|
|
"url": "http://open.api.tianyancha.com/services/open/mr/judicialSale/3.0"
|
|
},
|
|
"企业基本信息": {
|
|
"type": "dict",
|
|
"url": "https://open.api.tianyancha.com/services/open/ic/baseinfo/normal"
|
|
},
|
|
"企业专利信息": {
|
|
"type": "list",
|
|
"url": "http://open.api.tianyancha.com/services/open/ipr/patents/3.0"
|
|
},
|
|
"知识产权出质": {
|
|
"type": "list",
|
|
"url": "http://open.api.tianyancha.com/services/v4/open/getPledgeReg"
|
|
}, "知识产权": {
|
|
"type": "dict",
|
|
"url": "http://open.api.tianyancha.com/services/open/cb/ipr/3.0"
|
|
},
|
|
"税务评级": {
|
|
"type": "list",
|
|
"url": "http://open.api.tianyancha.com/services/open/m/taxCredit/2.0"
|
|
},
|
|
"进出口信用": {
|
|
"type": "dict",
|
|
"url": "http://open.api.tianyancha.com/services/open/m/importAndExport/2.0"
|
|
},
|
|
"税收违法": {
|
|
"type": "list",
|
|
"url": "http://open.api.tianyancha.com/services/open/mr/taxContravention/2.0"
|
|
},
|
|
"欠税公告": {
|
|
"type": "list",
|
|
"url": "http://open.api.tianyancha.com/services/open/mr/ownTax/2.0"
|
|
},
|
|
"行政许可": {
|
|
"type": "list",
|
|
"url": "http://open.api.tianyancha.com/services/open/m/getAdministrativeLicense/2.0"
|
|
},
|
|
"行政处罚": {
|
|
"type": "list",
|
|
"url": "http://open.api.tianyancha.com/services/open/mr/punishmentInfo/3.0"
|
|
},
|
|
"经营异常": {
|
|
"type": "list",
|
|
"url": "http://open.api.tianyancha.com/services/open/mr/abnormal/2.0"
|
|
},
|
|
"严重违法": {
|
|
"type": "list",
|
|
"url": "http://open.api.tianyancha.com/services/open/mr/illegalinfo/2.0"
|
|
},
|
|
"终本案件": {
|
|
"type": "list",
|
|
"url": "http://open.api.tianyancha.com/services/open/jr/endCase/2.0"
|
|
},
|
|
"失信被执行人": {
|
|
"type": "list",
|
|
"url": "http://open.api.tianyancha.com/services/v4/open/human/dishonest"
|
|
},
|
|
"被执行人": {
|
|
"type": "list",
|
|
"url": "http://open.api.tianyancha.com/services/open/jr/zhixinginfo/2.0"
|
|
},
|
|
"限制消费令": {
|
|
"type": "list",
|
|
"url": "http://open.api.tianyancha.com/services/open/jr/consumptionRestriction/2.0"
|
|
},
|
|
"质押比例": {
|
|
"type": "dict",
|
|
"url": "http://open.api.tianyancha.com/services/open/mr/stockPledge/ratio/2.0"
|
|
},
|
|
"企业三要素": {
|
|
"type": "dict",
|
|
"url": "http://open.api.tianyancha.com/services/open/ic/verify/2.0"
|
|
},
|
|
"工商信息": {
|
|
"type": "dict",
|
|
"url": "http://open.api.tianyancha.com/services/open/cb/ic/2.0"
|
|
},
|
|
"产品信息": {
|
|
"type": "list",
|
|
"url": "http://open.api.tianyancha.com/services/open/m/appbkInfo/2.0"
|
|
}, "股权结构图": {
|
|
"type": "dict",
|
|
"url": "http://open.api.tianyancha.com/services/v4/open/equityRatio"
|
|
}, "供应商": {
|
|
"type": "dict",
|
|
"url": "http://open.api.tianyancha.com/services/open/m/supply/2.0"
|
|
}, "企业招投标信息": {
|
|
"type": "list",
|
|
"url": "http://open.api.tianyancha.com/services/open/m/bids/2.0"
|
|
},
|
|
}
|
|
|
|
|
|
class TianyanchaApi:
|
|
"""
|
|
天眼查接口
|
|
"""
|
|
|
|
def __init__(self, token: str,
|
|
mongo_connect: MongoConnect = None,
|
|
list_max_count=100,
|
|
mongo_db_name="天眼查",
|
|
api_config: Dict[str, Dict] = api_config
|
|
):
|
|
self.config = api_config
|
|
self.token = token
|
|
self.rq = rq
|
|
self.mongo_connect = mongo_connect
|
|
self.mongo_db_name = mongo_db_name
|
|
self.list_max_count = list_max_count
|
|
|
|
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, error_code_dic[data['error_code']])
|
|
|
|
# raise Exception(error_code_dic[data['error_code']])
|
|
|
|
def get(self, name, query: dict, query_info: dict = {}, page=None):
|
|
total = 0
|
|
url_config = self.config.get(name)
|
|
if not url_config:
|
|
raise Exception(f"无 {name}")
|
|
url = url_config['url']
|
|
full_url = make_query_url(url, query)
|
|
if url_config["type"] == "list":
|
|
data, total = self.get_list_core(full_url, {**query, **query_info},
|
|
page * 20 if page else url_config.get('list_max_count'))
|
|
else:
|
|
data = self.get_data_core(full_url, {**query, **query_info})
|
|
return data, total
|
|
|
|
def get_total(self, name, query: dict, query_info: dict = {}):
|
|
total = 0
|
|
url_config = self.config.get(name)
|
|
if not url_config:
|
|
raise Exception(f"无 {name}")
|
|
url = url_config['url']
|
|
full_url = make_query_url(url, query)
|
|
if url_config["type"] == "list":
|
|
total = self.get_total_core(full_url, {**query, **query_info})
|
|
return int(total)
|
|
|
|
def get_total_core(self, query_url, query_info={}):
|
|
page_num = 1
|
|
page_size = 20
|
|
# 查询第一页的获取total
|
|
url = f"{query_url}&pageNum={page_num}&pageSize={page_size}"
|
|
result = self.get_data_core(url, query_info)
|
|
if not result:
|
|
return 0
|
|
total = result['total']
|
|
return total
|
|
|
|
def get_list_core(self, query_url, query_info={}, list_max_count=None):
|
|
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)
|
|
if not result:
|
|
return [], 0
|
|
total = int(result['total'])
|
|
print(f"共{total}条数据")
|
|
max_count = list_max_count or self.list_max_count
|
|
if max_count:
|
|
print(result['total'], max_count, "result['total']")
|
|
total = min(int(result['total']), max_count)
|
|
temp_list = [JsDict(item) for item in result['items']]
|
|
item_list.extend(temp_list)
|
|
if total > len(item_list):
|
|
for page_num in range(2, math.ceil(total / page_size) + 1):
|
|
url = f"{query_url}&pageNum={page_num}&pageSize={page_size}"
|
|
result = self.get_data_core(url, query_info) or []
|
|
temp_list = [JsDict(item) for item in result['items']]
|
|
item_list.extend(temp_list)
|
|
return item_list, total
|
|
|
|
|
|
class ApiItemConfig:
|
|
url: str
|
|
data_schema: Dict[str, Any]
|
|
paging: bool
|
|
method: str = 'GET'
|
|
|
|
|
|
class Api:
|
|
def __init__(self, source_config: Dict[str, ApiItemConfig], token: str):
|
|
self.source_config = source_config
|
|
self.token = token
|
|
|
|
def get_list_data_total(self, name, params: Dict):
|
|
pass
|
|
|
|
def get_data(self, name, params: Dict) -> Union[List[Any], Dict[str, Any]]:
|
|
source = self.source_config[name]
|
|
full_url = self.make_url(source.url, method=source.method, params=params)
|
|
return self.get_core(url=full_url)
|
|
|
|
def get_core(self, url):
|
|
data = requests.get(url, headers={'Authorization': self.token}).json()
|
|
if data['error_code'] == 0:
|
|
return JsDict(data['result'])
|
|
else:
|
|
print(error_code_dic[data['error_code']])
|
|
|
|
def make_url(self, url: str, method='GET', params: Dict = {}, page=None, page_size=None):
|
|
if method == 'GET':
|
|
params = deepcopy(params)
|
|
url = requests.Request('GET', url, params=params).prepare().url
|
|
return url
|