257 lines
12 KiB
Python
257 lines
12 KiB
Python
from DBHelper.MongoHelper import MongoHelper
|
|
from Manage.Company.Objects.CompanyIndex import *
|
|
|
|
|
|
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()
|
|
rating_records = RatingRecords()
|
|
|
|
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['报告期']
|
|
result['资产负债表']['报告期'] = result['报告期']
|
|
result['利润表']['报告期'] = result['报告期']
|
|
result['补充数据表']['报告期'] = result['报告期']
|
|
result['财务指标']['报告期'] = 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=FinancialIndex, data=item['指标'])
|
|
self.financial_analysis.compare_to_years = self.instance_list_to_set(instance=FinancialAnalysis.CompareToYears, data=result['财务得分年度比较'])
|
|
|
|
result['财务得分同行比较']['公司水平'] = self.dict_to_set(instance=FinancialIndex, data=result['财务得分同行比较']['公司水平'])
|
|
result['财务得分同行比较']['平均水平'] = self.dict_to_set(instance=FinancialIndex, 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=FinancialIndex, data=result['财务得分较去年变化'])
|
|
self.financial_analysis.delta_to_industry = self.dict_to_set(instance=FinancialIndex, data=result['财务得分较同行差异'])
|
|
self.financial_analysis.index_details = self.instance_list_to_set(instance=FinancialAnalysisDetails, data=result['指标详情'])
|
|
|
|
def get_cc_rating(self):
|
|
"""综合信用分析"""
|
|
if self.cc_rating.__dict__.__contains__('update_time'):
|
|
find_body = {"企业ID": self.cc_rating.cid, "更新时间": self.cc_rating.update_time}
|
|
else:
|
|
find_body = {"企业ID": self.cc_rating.cid}
|
|
|
|
result = self.db.find_single_data(
|
|
"企业数据",
|
|
"综合信用分析",
|
|
find_body,
|
|
["评价ID", "企业名称", "综合信用表现", "信用分析", "指标表格", "历史级别"]
|
|
)
|
|
|
|
if not result:
|
|
self.cc_rating.rid = None
|
|
self.cc_rating.name = None
|
|
self.cc_rating.cc_rating_radar = None
|
|
self.cc_rating.credit_analysis = None
|
|
self.cc_rating.index_table = None
|
|
self.cc_rating.history_level = None
|
|
else:
|
|
self.cc_rating.rid = result['评价ID']
|
|
self.cc_rating.name = result['企业名称']
|
|
|
|
result['综合信用表现']['指标雷达']['指标得分'] = self.dict_to_set(instance=CCRatingIndexRadarFactors, data=result['综合信用表现']['指标雷达']['指标得分'])
|
|
result['综合信用表现']['指标雷达']['最大分数'] = self.dict_to_set(instance=CCRatingIndexRadarFactors, data=result['综合信用表现']['指标雷达']['最大分数'])
|
|
self.cc_rating.cc_rating_radar = self.dict_to_set(instance=CCRating.IndexRadar, data=result['综合信用表现']['指标雷达'])
|
|
|
|
self.cc_rating.credit_analysis = self.dict_to_set(instance=CCRating.CreditAnalysis, data=result['信用分析'])
|
|
|
|
result['指标表格']["财务指标"] = self.dict_to_set(instance=FinancialIndex, data=result['指标表格']["财务指标"])
|
|
result['指标表格']["风险指标"] = self.dict_to_set(instance=CCRatingRiskIndex, data=result['指标表格']["风险指标"])
|
|
result['指标表格']["经营指标"] = self.dict_to_set(instance=CCRatingOperatingIndex, data=result['指标表格']["经营指标"])
|
|
self.cc_rating.index_table = self.dict_to_set(instance=CCRating.IndexTable, data=result['指标表格'])
|
|
|
|
self.cc_rating.history_level = self.instance_list_to_set(instance=CCRating.RankLevelsInHistory, data=result['历史级别'])
|
|
|
|
def get_esg_rating(self):
|
|
"""ESG评价分析"""
|
|
if self.esg_rating.__dict__.__contains__('update_time'):
|
|
find_body = {"企业ID": self.esg_rating.cid, "更新日期": self.esg_rating.update_time}
|
|
else:
|
|
find_body = {"企业ID": self.esg_rating.cid}
|
|
|
|
result = self.db.find_single_data(
|
|
"企业数据",
|
|
"ESG评价分析",
|
|
find_body,
|
|
["评价ID", "ESG评级", "维度得分情况"]
|
|
)
|
|
|
|
if not result:
|
|
self.esg_rating.rid = None
|
|
self.esg_rating.esg_level = None
|
|
self.esg_rating.factors_score = None
|
|
else:
|
|
self.esg_rating.rid = result['评价ID']
|
|
self.esg_rating.esg_level = self.dict_to_set(instance=ESGRating.ESGLevel, data=result['ESG评级'])
|
|
self.esg_rating.factors_score = self.instance_list_to_set(instance=ESGRating.FactorsScore, data=result['维度得分情况'])
|
|
|
|
def get_rating_records(self):
|
|
""""""
|
|
result = self.db.find_all_data_with_single_sort(
|
|
"企业数据",
|
|
"评价记录",
|
|
{"企业ID": self.rating_records.cid},
|
|
["评价ID", "评价项目", "进行状态", "评价结果", "报告fid", "证书fid", "开始填报时间", "提交填报时间", "评价完成时间"],
|
|
{"开始填报时间": -1}
|
|
)
|
|
|
|
if not result:
|
|
self.rating_records.rating_records = []
|
|
else:
|
|
|
|
for item in result:
|
|
item['报告'] = '/admin/file?fid={}&file=report&rating={}'.format(item.pop('报告fid'), item['评价项目'])
|
|
item['证书'] = '/admin/file?fid={}&file=certificate&rating={}'.format(item.pop('证书fid'), item['评价项目'])
|
|
|
|
self.rating_records.rating_records = self.instance_list_to_set(instance=RatingRecord, data=result)
|