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, "数据类型": "组合工商信息"})