397 lines
15 KiB
Python
397 lines
15 KiB
Python
"""
|
|
天眼查API
|
|
"""
|
|
import json
|
|
import requests
|
|
|
|
from common import judege_api_code, general_request
|
|
from db import conserve_data, find_db_data, conserve_data_by_hold
|
|
from setting import DB_SFFX, DB_JYFX, DB_TYFX, DB_GSBJ, DB_JYZK, DB_ZSCQ, DB_GSFZ
|
|
|
|
token = "32737167-cb63-4ce9-9397-d66169488f51"
|
|
headers = {'Authorization': token}
|
|
|
|
|
|
# 司法风险
|
|
def judicial_risk(name):
|
|
url = 'http://open.api.tianyancha.com/services/open/cb/judicial/2.0?keyword={}'
|
|
url = url.format(name)
|
|
# 请求url
|
|
res = json.loads(requests.get(url, headers=headers).text)
|
|
# 判断 error_code状态码
|
|
code = res['error_code']
|
|
status = judege_api_code(code)
|
|
# 返回请求成功则保存result数据
|
|
if status == '请求成功':
|
|
result = res['result']
|
|
insert = dict()
|
|
insert['开庭公告'] = result['ktAnnouncementList']
|
|
insert['被执行人'] = result['zhixingList']
|
|
insert['法院公告'] = result['courtAnnouncementList']
|
|
insert['立案信息'] = result['courtRegisterList']
|
|
insert['失信人'] = result['dishonestList']
|
|
insert['诉讼'] = result['lawSuitList']
|
|
insert['送达公告'] = result['sendAnnouncementList']
|
|
for key, value in insert.items():
|
|
conserve_data(DB_SFFX, name, key, value)
|
|
return '保存成功'
|
|
elif status == '无数据':
|
|
insert = dict()
|
|
insert['开庭公告'] = dict()
|
|
insert['被执行人'] = dict()
|
|
insert['法院公告'] = dict()
|
|
insert['立案信息'] = dict()
|
|
insert['失信人'] = dict()
|
|
insert['诉讼'] = dict()
|
|
insert['送达公告'] = dict()
|
|
for key, value in insert.items():
|
|
conserve_data(DB_SFFX, name, key, value)
|
|
return '保存成功'
|
|
else:
|
|
return status
|
|
|
|
|
|
# 经营风险-严重违法
|
|
def serious_violation(name):
|
|
url = 'http://open.api.tianyancha.com/services/open/mr/illegalinfo/2.0?pageSize=20&keyword={}&pageNum=1'
|
|
url = url.format(name)
|
|
res = general_request(url, name, DB_JYFX, '严重违法')
|
|
return res
|
|
|
|
|
|
# 经营风险-行政处罚
|
|
def administrative_sanction(name):
|
|
url = 'http://open.api.tianyancha.com/services/open/mr/punishmentInfo/3.0?pageSize=20&keyword={}&pageNum=1'
|
|
url = url.format(name)
|
|
res = general_request(url, name, DB_JYFX, '行政处罚')
|
|
return res
|
|
|
|
|
|
# 经营风险-环保处罚
|
|
def environmental_punishment(name):
|
|
url = 'http://open.api.tianyancha.com/services/open/mr/environmentalPenalty/2.0?name={}&pageSize=20&pageNum=1'
|
|
url = url.format(name)
|
|
res = general_request(url, name, DB_JYFX, '环保处罚')
|
|
return res
|
|
|
|
|
|
# 经营风险-经营异常
|
|
def abnormal_operation(name):
|
|
url = 'http://open.api.tianyancha.com/services/open/mr/abnormal/2.0?pageSize=20&keyword={}&pageNum=1'
|
|
url = url.format(name)
|
|
res = general_request(url, name, DB_JYFX, '经营异常')
|
|
return res
|
|
|
|
|
|
# 经营风险-欠税公告
|
|
def tax_arrears_announcement(name):
|
|
url = 'http://open.api.tianyancha.com/services/open/mr/ownTax/2.0?pageSize=20&keyword={}&pageNum=1'
|
|
url = url.format(name)
|
|
res = general_request(url, name, DB_JYFX, '欠税公告')
|
|
return res
|
|
|
|
|
|
# 经营风险-税收违法
|
|
def tax_violation(name):
|
|
url = 'http://open.api.tianyancha.com/services/open/mr/taxContravention/2.0?pageSize=20&keyword={}&pageNum=1'
|
|
url = url.format(name)
|
|
res = general_request(url, name, DB_JYFX, '税收违法')
|
|
return res
|
|
|
|
|
|
# 经营风险-股权出质
|
|
def equity_pledge(name):
|
|
url = 'http://open.api.tianyancha.com/services/open/mr/equityInfo/2.0?pageSize=20&keyword={}&pageNum=1'
|
|
url = url.format(name)
|
|
res = general_request(url, name, DB_JYFX, '股权出质')
|
|
return res
|
|
|
|
|
|
# 企业天眼风险-周边风险
|
|
def peripheral_risk(name):
|
|
url = 'http://open.api.tianyancha.com/services/open/risk/riskInfo/2.0?keyword={}'
|
|
url = url.format(name)
|
|
|
|
# 请求url
|
|
res = json.loads(requests.get(url, headers=headers).text)
|
|
# 判断 error_code状态码
|
|
code = res['error_code']
|
|
status = judege_api_code(code)
|
|
# 返回请求成功则保存result数据
|
|
if status == '请求成功':
|
|
result = res['result']
|
|
insert = dict()
|
|
insert['自身风险'] = result['riskList'][0]
|
|
insert['周边风险'] = result['riskList'][1]
|
|
insert['历史风险'] = result['riskList'][2]
|
|
insert['预警提醒'] = result['riskList'][3]
|
|
for key, value in insert.items():
|
|
conserve_data(DB_TYFX, name, key, value)
|
|
return '保存成功'
|
|
elif status == '无数据':
|
|
insert = dict()
|
|
insert['自身风险'] = dict()
|
|
insert['周边风险'] = dict()
|
|
insert['历史风险'] = dict()
|
|
insert['预警提醒'] = dict()
|
|
for key, value in insert.items():
|
|
conserve_data(DB_TYFX, name, key, value)
|
|
return '保存成功'
|
|
else:
|
|
return status
|
|
|
|
|
|
# 公司背景-企业基本信息
|
|
def company_baseinfo(name):
|
|
url = 'http://open.api.tianyancha.com/services/open/ic/baseinfo/normal?keyword={}'
|
|
url = url.format(name)
|
|
# 请求url
|
|
res = json.loads(requests.get(url, headers=headers).text)
|
|
# 判断 error_code状态码
|
|
code = res['error_code']
|
|
status = judege_api_code(code)
|
|
# 返回请求成功则保存result数据
|
|
if status == '请求成功':
|
|
insert = res['result']
|
|
conserve_data(DB_GSBJ, name, '基本信息', insert)
|
|
return '保存成功'
|
|
elif status == '无数据':
|
|
insert = dict()
|
|
conserve_data(DB_GSBJ, name, '基本信息', insert)
|
|
return '保存成功'
|
|
else:
|
|
return status
|
|
|
|
|
|
# 公司背景-企业股东
|
|
def corporate_shareholders(name):
|
|
def method_01():
|
|
url = 'http://open.api.tianyancha.com/services/open/ic/holder/2.0?pageSize=20&keyword={}&pageNum=1'
|
|
url = url.format(name)
|
|
res = general_request(url, name, DB_GSBJ, '企业股东')
|
|
return res
|
|
|
|
def method_02():
|
|
url = 'http://open.api.tianyancha.com/services/open/stock/shareholder/2.0?&keyword={}&type=1'
|
|
url = url.format(name)
|
|
res = general_request(url, name, DB_GSBJ, '十大股东')
|
|
return res
|
|
|
|
records = find_db_data(db='公司背景', col='基本信息', query={"企业名称": name})[0]
|
|
if '上市' in records['基本信息']['companyOrgType']:
|
|
info = method_02()
|
|
return info
|
|
else:
|
|
info = method_01()
|
|
return info
|
|
|
|
|
|
# 公司背景-对外投资
|
|
def outbound_investment(name):
|
|
url = 'http://open.api.tianyancha.com/services/open/ic/inverst/2.0?pageSize=20&keyword={}&pageNum=1'
|
|
url = url.format(name)
|
|
res = general_request(url, name, DB_GSBJ, '对外投资')
|
|
return res
|
|
|
|
|
|
# 公司背景-主要人员
|
|
def principal_officer(name):
|
|
url = 'http://open.api.tianyancha.com/services/open/ic/staff/2.0?keyword={}&pageNum=1&pageSize=20'
|
|
url = url.format(name)
|
|
res = general_request(url, name, DB_GSBJ, '主要人员')
|
|
return res
|
|
|
|
|
|
# 公司背景-变更记录
|
|
def change_record(name):
|
|
url = 'http://open.api.tianyancha.com/services/open/ic/changeinfo/2.0?keyword={}&pageNum=1&pageSize=20'
|
|
url = url.format(name)
|
|
res = general_request(url, name, DB_GSBJ, '变更记录')
|
|
return res
|
|
|
|
|
|
# 经营状况-进出口信用
|
|
def import_export_credit(name):
|
|
url = 'http://open.api.tianyancha.com/services/open/m/importAndExport/2.0?keyword={}'
|
|
url = url.format(name)
|
|
# 请求url
|
|
res = json.loads(requests.get(url, headers=headers).text)
|
|
# 判断 error_code状态码
|
|
code = res['error_code']
|
|
status = judege_api_code(code)
|
|
# 返回请求成功则保存result数据
|
|
if status == '请求成功':
|
|
insert = res['result']
|
|
conserve_data(DB_JYZK, name, '进出口信用', insert)
|
|
return '保存成功'
|
|
elif status == '无数据':
|
|
insert = dict()
|
|
conserve_data(DB_JYZK, name, '进出口信用', insert)
|
|
return '保存成功'
|
|
else:
|
|
return status
|
|
|
|
|
|
# 经营状况-资质证书
|
|
def qualification_certificate(name):
|
|
url = 'http://open.api.tianyancha.com/services/open/m/certificate/2.0?name={}&pageSize=20&pageNum=1'
|
|
url = url.format(name)
|
|
res = general_request(url, name, DB_JYZK, '资质证书')
|
|
return res
|
|
|
|
|
|
# 知识产权
|
|
def intellectual_property(name):
|
|
url = 'http://open.api.tianyancha.com/services/open/cb/ipr/2.0?keyword={}'
|
|
url = url.format(name)
|
|
# 请求url
|
|
res = json.loads(requests.get(url, headers=headers).text)
|
|
# 判断 error_code状态码
|
|
code = res['error_code']
|
|
status = judege_api_code(code)
|
|
# 返回请求成功则保存result数据
|
|
if status == '请求成功':
|
|
result = res['result']
|
|
insert = dict()
|
|
insert['商标'] = result['tmList']
|
|
insert['软件著作权'] = result['copyRegList']
|
|
insert['作品著作权'] = result['copyRegWorksList']
|
|
insert['网站备案'] = result['icpList']
|
|
insert['专利'] = result['patentList']
|
|
for key, value in insert.items():
|
|
conserve_data(DB_ZSCQ, name, key, value)
|
|
return '保存成功'
|
|
elif status == '无数据':
|
|
insert = dict()
|
|
insert['商标'] = dict()
|
|
insert['软件著作权'] = dict()
|
|
insert['作品著作权'] = dict()
|
|
insert['网站备案'] = dict()
|
|
insert['专利'] = dict()
|
|
for key, value in insert.items():
|
|
conserve_data(DB_ZSCQ, name, key, value)
|
|
return '保存成功'
|
|
else:
|
|
return status
|
|
|
|
|
|
# 公司发展-融资历史
|
|
def financing_history(name):
|
|
url = 'http://open.api.tianyancha.com/services/open/cd/findHistoryRongzi/2.0?pageSize=20&keyword={}&pageNum=1'
|
|
url = url.format(name)
|
|
res = general_request(url, name, DB_GSFZ, '融资历史')
|
|
return res
|
|
|
|
|
|
# 人员控股企业
|
|
def holding_enterprise(h_id, h_name):
|
|
url = 'http://open.api.tianyancha.com/services/open/human/companyholding/2.0?hid={}&pageSize=20&pageNum=1&'
|
|
url = url.format(h_id)
|
|
# 根据传入url进行请求
|
|
res = json.loads(requests.get(url, headers=headers).text)
|
|
# 判断请求返回结果中的error_code
|
|
code = res['error_code']
|
|
status = judege_api_code(code)
|
|
# status为请求成功则保存result数据
|
|
if status == '请求成功':
|
|
insert = dict()
|
|
insert['result'] = res['result']['items']
|
|
insert['total'] = res['result']['total']
|
|
conserve_data_by_hold(DB_GSBJ, h_id, h_name, '人员控股企业', insert)
|
|
return '保存成功'
|
|
elif status == '无数据':
|
|
insert = dict()
|
|
conserve_data_by_hold(DB_GSBJ, h_id, h_name, '人员控股企业', insert)
|
|
return '保存成功'
|
|
else:
|
|
return status
|
|
|
|
|
|
# 保存股东数据
|
|
def shareholder_information(name):
|
|
basic_info = find_db_data(db='公司背景', col='基本信息', query={'企业名称': name})[0]
|
|
if '上市' in basic_info['基本信息']['companyOrgType']:
|
|
# 判断企业大股东类型
|
|
result = find_db_data(db='公司背景', col='十大股东', query={'企业名称': name})
|
|
if len(result) > 0:
|
|
if result[0]['十大股东']['holderList'][0]['cType'] == 1:
|
|
# 1-公司
|
|
company_name = result[0]['十大股东']['holderList'][0]['name']
|
|
# 股东为公司，需要再次调用基本信息/企业股东天眼查接口
|
|
res = company_baseinfo(company_name)
|
|
if res == '保存成功':
|
|
res = corporate_shareholders(company_name)
|
|
else:
|
|
# 2-自然人
|
|
h_id = result[0]['十大股东']['holderList'][0]['id']
|
|
h_name = result[0]['十大股东']['holderList'][0]['name']
|
|
res = holding_enterprise(h_id, h_name)
|
|
return res
|
|
else:
|
|
# 判断企业大股东类型
|
|
result = find_db_data(db='公司背景', col='企业股东', query={'企业名称': name})
|
|
if len(result) > 0:
|
|
if result[0]['企业股东']['result'][0]['type'] == 1:
|
|
# 1-公司
|
|
company_name = result[0]['企业股东']['result'][0]['name']
|
|
# 股东为公司，需要再次调用基本信息/企业股东天眼查接口
|
|
res = company_baseinfo(company_name)
|
|
if res == '保存成功':
|
|
res = corporate_shareholders(company_name)
|
|
else:
|
|
# 2-自然人
|
|
h_id = result[0]['企业股东']['result'][0]['id']
|
|
h_name = result[0]['企业股东']['result'][0]['name']
|
|
res = holding_enterprise(h_id, h_name)
|
|
return res
|
|
|
|
|
|
# 保存全量数据
|
|
def save_full_data(name):
|
|
"""
|
|
调用天眼查api
|
|
保存天眼查全量数据
|
|
"""
|
|
res = judicial_risk(name)
|
|
if res == '保存成功':
|
|
res = serious_violation(name)
|
|
if res == '保存成功':
|
|
res = administrative_sanction(name)
|
|
if res == '保存成功':
|
|
res = environmental_punishment(name)
|
|
if res == '保存成功':
|
|
res = abnormal_operation(name)
|
|
if res == '保存成功':
|
|
res = tax_arrears_announcement(name)
|
|
if res == '保存成功':
|
|
res = tax_violation(name)
|
|
if res == '保存成功':
|
|
res = equity_pledge(name)
|
|
if res == '保存成功':
|
|
res = peripheral_risk(name)
|
|
if res == '保存成功':
|
|
res = company_baseinfo(name)
|
|
if res == '保存成功':
|
|
res = corporate_shareholders(name)
|
|
if res == '保存成功':
|
|
res = outbound_investment(name)
|
|
if res == '保存成功':
|
|
res = import_export_credit(name)
|
|
if res == '保存成功':
|
|
res = qualification_certificate(name)
|
|
if res == '保存成功':
|
|
res = intellectual_property(name)
|
|
if res == '保存成功':
|
|
res = financing_history(name)
|
|
if res == '保存成功':
|
|
res = principal_officer(name)
|
|
if res == '保存成功':
|
|
res = change_record(name)
|
|
if res == '保存成功':
|
|
res = shareholder_information(name)
|
|
return res
|
|
|
|
|
|
if __name__ == '__main__':
|
|
r = shareholder_information('大族激光科技产业集团股份有限公司')
|
|
print(r) |