from DBHelper.MongoHelper import MongoHelper from Manage.Company.Objects.CompanyIndex import CompanyIndexApi, HeadInfo, UpdateTime, BasicInfo, FinancialData, \ OperatingRisk, FinancialAnalysis, CCRating, ESGRating, FinancialAnalysisDetails, FinancialAnalysisIndex class CompanyIndexApiImpl(CompanyIndexApi): """企业主页接口实现类""" db = MongoHelper("tfse_v0.21") head_info = HeadInfo() update_time = UpdateTime() basic_info = BasicInfo() financial_data = FinancialData() operating_risk = OperatingRisk() financial_analysis = FinancialAnalysis() cc_rating = CCRating() esg_rating = ESGRating() def get_head_info(self): """获取头部信息""" self.head_info.email = self.db.find_single_column( "应用端", "企业用户", {"企业ID": self.head_info.cid}, "邮箱" ) result = self.db.find_single_data( "企业数据", "企业数据_更新汇总", {"企业ID": self.head_info.cid}, ["企业名称", "一级行业"] ) self.head_info.name = result['企业名称'] self.head_info.industry = result['一级行业'] def get_update_time(self): """获取更新日期""" if self.update_time.table in ["财务数据"]: update_time = "报告期" elif self.update_time.table in ["工商信息", "经营风险", "财务分析", "ESG评价分析"]: update_time = "更新日期" elif self.update_time.table in ["综合信用分析"]: update_time = "更新时间" else: update_time = False if update_time: results = self.db.find_all_data_with_single_sort( "企业数据", self.update_time.table_map[self.update_time.table], {"企业ID": self.update_time.cid}, [update_time], {update_time: -1} ) if results: self.update_time.update_time = [result[update_time] for result in results] else: self.update_time.update_time = [] def get_basic_info(self): """获取基本工商信息""" if self.basic_info.__dict__.__contains__('update_time'): find_body = {"企业ID": self.basic_info.cid, "更新日期": self.basic_info.update_time} else: find_body = {"企业ID": self.basic_info.cid} result = self.db.find_single_data( "企业数据", "基本工商信息", find_body, ["工商信息", "股东信息", "主要成员"] ) if not result: result = {'工商信息': {}, "股东信息": {}, "主要成员": {}} self.basic_info.business_info = self.dict_to_set(instance=self.basic_info.BusinessInfo, data=result['工商信息']) self.basic_info.share_holder = self.instance_list_to_set(instance=self.basic_info.ShareHolder, data=result['股东信息']) self.basic_info.main_member = self.instance_list_to_set(instance=self.basic_info.MainMember, data=result['主要成员']) def get_financial_data(self): """获取财务数据""" result = self.db.find_single_data( "企业数据", "基本财务数据", {"企业ID": self.financial_data.cid, "报告期": self.financial_data.report_date}, ["报告期", "资产负债表", "利润表", "补充数据表", "财务指标"] ) if not result: result = {'报告期': None, "资产负债表": {}, "利润表": {}, "补充数据表": {}, "财务指标": {}} self.financial_data.report_date = result['报告期'] self.financial_data.balance_sheet = self.dict_to_set(instance=self.financial_data.BalanceSheet, data=result['资产负债表']) self.financial_data.profit_sheet = self.dict_to_set(instance=self.financial_data.ProfitSheet, data=result['利润表']) self.financial_data.appendix_data = self.dict_to_set(instance=self.financial_data.AppendixDataSheet, data=result['补充数据表']) self.financial_data.financial_index = self.dict_to_set(instance=self.financial_data.FinancialIndex, data=result['财务指标']) def get_operating_risk(self): """获取经营风险分析""" if self.operating_risk.__dict__.__contains__('update_time'): find_body = {"企业ID": self.operating_risk.cid, "更新时间": self.operating_risk.update_time} else: find_body = {"企业ID": self.operating_risk.cid} result = self.db.find_single_data( "企业数据", "经营风险分析", find_body, ["评价ID", "风险级别", "风险分数", "列入失信名单", "合规风险", "经营风险", "周边风险", "变更记录"] ) if not result: result = {'评价ID': None, "风险级别": None, "风险分数": None, "列入失信名单": None, "合规风险": {}, "经营风险": {}, "周边风险": {}, "变更记录": {}} self.operating_risk.rid = result['评价ID'] self.operating_risk.risk_level = result['风险级别'] self.operating_risk.risk_score = result['风险分数'] self.operating_risk.in_untrustworthy_list = result['列入失信名单'] self.operating_risk.compliance_risk = self.instance_list_to_set(instance=OperatingRisk().InfluenceObj, data=result['合规风险']) self.operating_risk.operating_risk = self.instance_list_to_set(instance=OperatingRisk().InfluenceObj, data=result['经营风险']) self.operating_risk.peripheral_risk = self.instance_list_to_set(instance=OperatingRisk().DetailObj, data=result['周边风险']) self.operating_risk.change_records = self.instance_list_to_set(instance=OperatingRisk().InfluenceObj, data=result['变更记录']) def get_financial_analysis(self): """获取财务分析数据""" if self.financial_analysis.__dict__.__contains__('update_time'): find_body = {"企业ID": self.financial_analysis.cid, "更新时间": self.financial_analysis.update_time} else: find_body = {"企业ID": self.financial_analysis.cid} result = self.db.find_single_data( "企业数据", "财务要素分析", find_body, ["评价ID", "财报期", "财务得分年度比较", "财务得分同行比较", "财务得分较去年变化", "财务得分较同行差异", "指标详情"] ) if not result: self.financial_analysis.rid = None self.financial_analysis.report_date = None self.financial_analysis.compare_to_years = None self.financial_analysis.compare_to_industry = None self.financial_analysis.delta_to_years = None self.financial_analysis.delta_to_industry = None self.financial_analysis.index_details = None else: self.financial_analysis.rid = result['评价ID'] self.financial_analysis.report_date = result['财报期'] for item in result['财务得分年度比较']: item["指标"] = self.dict_to_set(instance=FinancialAnalysisIndex, data=item['指标']) self.financial_analysis.compare_to_years = self.instance_list_to_set(instance=FinancialAnalysis.CompareToYears, data=result['财务得分年度比较']) result['财务得分同行比较']['公司水平'] = self.dict_to_set(instance=FinancialAnalysisIndex, data=result['财务得分同行比较']['公司水平']) result['财务得分同行比较']['平均水平'] = self.dict_to_set(instance=FinancialAnalysisIndex, data=result['财务得分同行比较']['平均水平']) self.financial_analysis.compare_to_industry = self.dict_to_set(instance=FinancialAnalysis.CompareToIndustry, data=result['财务得分同行比较']) self.financial_analysis.delta_to_years = self.dict_to_set(instance=FinancialAnalysisIndex, data=result['财务得分较去年变化']) self.financial_analysis.delta_to_industry = self.dict_to_set(instance=FinancialAnalysisIndex, data=result['财务得分较同行差异']) self.financial_analysis.index_details = self.instance_list_to_set(instance=FinancialAnalysisDetails, data=result['指标详情'])