450 lines
20 KiB
Python
450 lines
20 KiB
Python
from DBHelper.MongoHelper import MongoHelper
|
||
from cc_rating.CCRatingObj import RatingInfo, CreditRatingInput, FinancialReport, BusinessQuestionnaire
|
||
from cc_rating.CCRatingUtils import *
|
||
|
||
|
||
class RatingImpl(CreditRatingInput, RatingInfo, BusinessQuestionnaire, FinancialReport):
|
||
"""企业综合信用评价"""
|
||
|
||
db = MongoHelper('tfse_v0.21')
|
||
|
||
def records(self):
|
||
"""评价记录"""
|
||
cc_report = '/file/pdf?file_id={}'
|
||
esg_report = '/file/pdf?file_id={}'
|
||
cc_certification = "/file/pdf?file_id={}"
|
||
esg_certification = "/file/pdf?file_id={}"
|
||
|
||
rating_recprds = self.db.find_all_data(
|
||
"企业数据",
|
||
"评价记录",
|
||
{"企业ID": self.cid},
|
||
["评价ID", "评价项目", "评价方式", "进行状态", "评价结果", "报告fid", "证书fid"]
|
||
)
|
||
|
||
for record in rating_recprds:
|
||
if rating_recprds is not None:
|
||
if record['评价项目'] == '综合信用评价':
|
||
record['报告'] = cc_report.format(record['报告fid']) if record['报告fid'] is not None else None
|
||
record['证书'] = cc_certification.format(record['证书fid']) if record['证书fid'] is not None else None
|
||
record.pop('报告fid')
|
||
record.pop('证书fid')
|
||
else:
|
||
record['报告'] = esg_report.format(record['报告fid']) if record['报告fid'] is not None else None
|
||
record['证书'] = esg_certification.format(record['证书fid']) if record['证书fid'] is not None else None
|
||
record.pop('报告fid')
|
||
record.pop('证书fid')
|
||
return rating_recprds
|
||
|
||
def start_cc_rating(self):
|
||
"""开始综合评价"""
|
||
|
||
def check_input_data():
|
||
"""
|
||
input_data: 已存在的填报数据
|
||
根据企业ID+当前年度,查询是否存在填报数据
|
||
若存在填报数据，则返回填报数据
|
||
若不存在填报数据，则返回False
|
||
"""
|
||
input_data = self.db.find_single_data(
|
||
"综信评价数据",
|
||
"填报数据",
|
||
{"企业ID": self.cid, "评价年度": get_current_year()},
|
||
["评价ID", "企业名称", "评价年度", "行业选择", "经营问卷", "财务填报"]
|
||
)
|
||
return input_data if input_data else False
|
||
|
||
def case_input_exist(param):
|
||
"""
|
||
param: dict 填报数据
|
||
根据评价ID查询评价进行状态
|
||
若评价已完成，返回‘本年度已评价’的提示，且返回空对象
|
||
若评价进行中，返回‘继续填报’的提示，且返回填报数据
|
||
"""
|
||
# rating_status: 评价服务状态
|
||
rating_status = self.db.find_single_data(
|
||
"企业数据",
|
||
"评价记录",
|
||
{"评价ID": param["评价ID"]},
|
||
["进行状态"]
|
||
)
|
||
# 把不需要返回的字段删掉
|
||
param.pop('评价ID')
|
||
# 方法返回结果
|
||
return ("本年度已评价", {}) if rating_status == "完成" else ("继续填报", param)
|
||
|
||
def case_input_not_exist():
|
||
"""不存在填报数据"""
|
||
# new_rid: 新生成填报ID
|
||
new_rid = gen_new_rid(8)
|
||
# report_dates: 生成报告期
|
||
report_dates = make_3_report_dates()
|
||
# company_name: 企业名称
|
||
company_name = self.db.find_single_data(
|
||
"应用端",
|
||
"企业用户",
|
||
{"企业ID": self.cid},
|
||
["企业名称"]
|
||
)
|
||
|
||
def prepare_input_template():
|
||
"""
|
||
根据模板生成新的填报数据
|
||
input_template: 填报数据模板
|
||
"""
|
||
input_template = self.credit_rating_dict_to_save()
|
||
input_template['评价ID'] = new_rid
|
||
input_template['企业ID'] = self.cid
|
||
input_template['企业名称'] = company_name['企业名称']
|
||
input_template['评价年度'] = get_current_year()
|
||
input_template['经营问卷'] = self.business_questionnaire_dict_to_save()
|
||
input_template['财务填报'] = self.financial_dict_to_save()
|
||
balance_sheet = list()
|
||
income_sheet = list()
|
||
appendix_sheet = list()
|
||
for n in range(3):
|
||
balance_sheet.append(self.BalanceSheet().dict_to_save())
|
||
income_sheet.append(self.IncomeSheet().dict_to_save())
|
||
appendix_sheet.append(self.AppendixDataSheet().dict_to_save())
|
||
input_template['财务填报']['资产负债表'] = balance_sheet
|
||
input_template['财务填报']['利润表'] = income_sheet
|
||
input_template['财务填报']['补充数据表'] = appendix_sheet
|
||
for sheets in ['资产负债表', '利润表', '补充数据表']:
|
||
for i in range(3):
|
||
input_template['财务填报'][sheets][i]['报告期'] = report_dates[i]
|
||
# 插入新的填报记录
|
||
self.db.insert_single_data(
|
||
'综信评价数据',
|
||
'填报数据',
|
||
input_template
|
||
)
|
||
|
||
def prepare_rating_record_template():
|
||
"""根据模板生成新的评价记录"""
|
||
self.rid = new_rid
|
||
self.company = company_name['企业名称']
|
||
self.project = '综合信用评价'
|
||
self.methods = '企业申报'
|
||
self.status = '进行'
|
||
self.begin_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
|
||
rating_record_template = self.rating_info_dict_to_save()
|
||
self.db.insert_single_data(
|
||
'企业数据',
|
||
'评价记录',
|
||
rating_record_template
|
||
)
|
||
|
||
def prepare_process_template():
|
||
"""根据数据模板生成新的评价流程数据"""
|
||
process_template = get_data_template('process')
|
||
process_template['评价ID'] = new_rid
|
||
process_template['企业ID'] = self.cid
|
||
self.db.insert_single_data(
|
||
'综信评价数据',
|
||
'评价流程',
|
||
process_template
|
||
)
|
||
|
||
# 评价流程
|
||
prepare_process_template()
|
||
# 评价记录
|
||
prepare_rating_record_template()
|
||
# 填报数据
|
||
prepare_input_template()
|
||
# 风险数据
|
||
prepare_risk_data(self.cid, new_rid, company_name)
|
||
return "开始填报", check_input_data()
|
||
|
||
def execute_func():
|
||
"""判断是否有填报记录"""
|
||
input_data = check_input_data()
|
||
if input_data is not False:
|
||
return case_input_exist(input_data)
|
||
else:
|
||
return case_input_not_exist()
|
||
|
||
return execute_func()
|
||
|
||
def save_cc_rating_input(self, input_form):
|
||
"""保存综合填报信息"""
|
||
self.cid = self.cid
|
||
self.project = "综合信用评价"
|
||
self.status = "进行"
|
||
# 根据cid查询评价记录得到rid
|
||
find = self.rating_info_dict_to_save(columns=["企业ID", "评价项目", "进行状态"])
|
||
rid = self.db.find_single_data("企业数据", "评价记录", find, ['评价ID'])
|
||
|
||
def business_questionnaire_script():
|
||
"""财务问卷"""
|
||
business = input_form['经营问卷']
|
||
index = get_data_template('index')
|
||
for k, v in index.items():
|
||
self.__setattr__(k, business[v])
|
||
return self.business_questionnaire_dict_to_save()
|
||
|
||
def financial_report_script():
|
||
"""财务填报"""
|
||
financial = input_form['财务填报']
|
||
|
||
def balance_sheet_script():
|
||
"""资产负债表"""
|
||
balance_sheet = self.BalanceSheet()
|
||
new_balance_sheet = list()
|
||
for sheet in financial['资产负债表']:
|
||
balance_sheet.report_date = sheet['报告期']
|
||
balance_sheet.accounts_receivable = sheet['应收账款']
|
||
balance_sheet.stock = sheet['存货']
|
||
balance_sheet.total_current_assets = sheet['流动资产合计']
|
||
balance_sheet.total_assets = sheet['资产总计']
|
||
balance_sheet.short_loan = sheet['短期借款']
|
||
balance_sheet.one_year_liabilities = sheet['一年内到期非流动负债']
|
||
balance_sheet.total_current_liabilities = sheet['流动负债合计']
|
||
balance_sheet.long_term_loan = sheet['长期借款']
|
||
balance_sheet.total_liabilities = sheet['负债合计']
|
||
balance_sheet.total_owners_equity = sheet['所有者权益合计']
|
||
new_balance_sheet.append(balance_sheet.dict_to_save())
|
||
return new_balance_sheet
|
||
|
||
def income_sheet_script():
|
||
"""利润表"""
|
||
income_sheet = self.IncomeSheet()
|
||
new_income_sheet = list()
|
||
for sheet in financial['利润表']:
|
||
income_sheet.report_date = sheet['报告期']
|
||
income_sheet.operating_income = sheet['营业收入']
|
||
income_sheet.operating_cost = sheet['营业成本']
|
||
income_sheet.total_profit = sheet['利润总额']
|
||
income_sheet.net_profit = sheet['净利润']
|
||
new_income_sheet.append(income_sheet.dict_to_save())
|
||
return new_income_sheet
|
||
|
||
def appendix_sheet_script():
|
||
"""补充数据表"""
|
||
appendix_sheet = self.AppendixDataSheet()
|
||
new_appendix_sheet = list()
|
||
for sheet in financial['补充数据表']:
|
||
appendix_sheet.report_date = sheet['报告期']
|
||
appendix_sheet.rd_expenses = sheet['研发费用']
|
||
appendix_sheet.interest_disbursement = sheet['计入财务费的利息支出']
|
||
appendix_sheet.interest_expense = sheet['资本化利息支出']
|
||
new_appendix_sheet.append(appendix_sheet.dict_to_save())
|
||
return new_appendix_sheet
|
||
|
||
self.balance_sheet_list = balance_sheet_script()
|
||
self.income_sheet_list = income_sheet_script()
|
||
self.appendix_data_sheet_list = appendix_sheet_script()
|
||
return self.financial_dict_to_save()
|
||
|
||
def rating_input():
|
||
"""校验填报数据"""
|
||
self.rid = rid['评价ID']
|
||
self.cid = input_form['企业ID']
|
||
self.company = input_form['企业名称']
|
||
self.year = input_form['评价年度']
|
||
self.industry = input_form['行业选择']
|
||
self.business_questionnaire = business_questionnaire_script()
|
||
self.financial_report = financial_report_script()
|
||
input_data = self.credit_rating_dict_to_save()
|
||
self.db.update_single_data(
|
||
'综信评价数据',
|
||
'填报数据',
|
||
rid,
|
||
input_data
|
||
)
|
||
|
||
def rating_process():
|
||
"""更新评价流程"""
|
||
process = self.db.find_single_data(
|
||
'综信评价数据',
|
||
'评价流程',
|
||
rid,
|
||
['评价流程']
|
||
)
|
||
if process:
|
||
if process['评价流程']['问卷填报'] == 0:
|
||
update_process_status(rid['评价ID'], {"评价流程.问卷填报": 1})
|
||
if process['评价流程']['财务数据填报'] == 0:
|
||
update_process_status(rid['评价ID'], {"评价流程.财务数据填报": 1})
|
||
|
||
rating_input()
|
||
rating_process()
|
||
return '保存成功'
|
||
|
||
def exec_cc_rating(self):
|
||
"""提交填报数据，开始评价"""
|
||
|
||
def check_rating_record():
|
||
"""校验评价记录"""
|
||
self.project = '综合信用评价'
|
||
self.status = '进行'
|
||
find = self.rating_info_dict_to_save(columns=['企业ID', '评价项目', '进行状态'])
|
||
record = self.db.find_single_data(
|
||
'企业数据',
|
||
'评价记录',
|
||
find,
|
||
['评价ID']
|
||
)
|
||
return record
|
||
|
||
rid = check_rating_record()
|
||
self.submit_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
|
||
update_time = self.rating_info_dict_to_save(columns=['提交填报时间'])
|
||
self.db.update_single_data('企业数据', '评价记录', rid, update_time)
|
||
if not rid:
|
||
return "没有正在进行的评价", 200
|
||
|
||
def get_rating_data():
|
||
"""获取打分需要的数据"""
|
||
input_data = self.db.find_single_data(
|
||
'综信评价数据',
|
||
'填报数据',
|
||
rid,
|
||
["评价ID", "企业ID", "企业名称", "评价年度", "行业选择", "经营问卷", "财务填报"]
|
||
)
|
||
risk_data = self.db.find_single_data(
|
||
'综信评价数据',
|
||
'风险数据',
|
||
rid,
|
||
["经营风险", "合规风险", "关联风险"]
|
||
)
|
||
return {**input_data, **{"风险数据": risk_data}}
|
||
|
||
rating_input = get_rating_data()
|
||
|
||
def general_model():
|
||
"""打分模型"""
|
||
url = "https://api.fecribd.com/tfse_rating/rating/general_model"
|
||
token = "X0gSlC!YE8jmr2jJr&ilcjS83j!tsoh5"
|
||
headers = {'token': token, "Content-Type": "application/json;charset=UTF-8"}
|
||
data = json.dumps({"input_data": rating_input})
|
||
res = requests.post(url, headers=headers, data=data)
|
||
if res.status_code == 200:
|
||
update_data = {
|
||
"评价结果": json.loads(res.text)['result'],
|
||
}
|
||
self.db.update_single_data("企业数据", "评价记录", rid, update_data)
|
||
update_process_status(rid['评价ID'], {"评价流程.模型打分": 1})
|
||
return True
|
||
else:
|
||
update_process_status(rid['评价ID'], {"评价流程.模型打分": -1})
|
||
return False
|
||
|
||
def gen_report():
|
||
"""生成报告"""
|
||
url = "https://api.fecribd.com/tfse_rating/report/gen_general_rating_report"
|
||
token = "X0gSlC!YE8jmr2jJr&ilcjS83j!tsoh5"
|
||
headers = {'token': token, "Content-Type": "application/json;charset=UTF-8"}
|
||
data = {
|
||
"company": rating_input['企业名称'],
|
||
"cid": rating_input['企业ID'],
|
||
"rid": rating_input['评价ID']
|
||
}
|
||
res = requests.post(url, headers=headers, data=json.dumps(data))
|
||
if res.status_code == 200:
|
||
update_data = {
|
||
"报告fid": json.loads(res.text)['result']['FileID']
|
||
}
|
||
self.db.update_single_data("企业数据", "评价记录", rid, update_data)
|
||
update_process_status(rid['评价ID'], {"评价流程.报告生成": 1})
|
||
return True
|
||
else:
|
||
update_process_status(rid['评价ID'], {"评价流程.报告生成": -1})
|
||
return False
|
||
|
||
def gen_certified():
|
||
"""生成证书"""
|
||
url = "https://api.fecribd.com/tfse_rating/certificate/generate_certificate"
|
||
token = "X0gSlC!YE8jmr2jJr&ilcjS83j!tsoh5"
|
||
headers = {'token': token, "Content-Type": "application/json;charset=UTF-8"}
|
||
data = {
|
||
"company": rating_input['企业名称'],
|
||
"cid": rating_input['企业ID'],
|
||
"rid": rating_input['评价ID']
|
||
}
|
||
res = requests.post(url, headers=headers, data=json.dumps(data))
|
||
if res.status_code == 200:
|
||
update_data = {
|
||
"证书fid": json.loads(res.text)['result']['FileID']
|
||
}
|
||
self.db.update_single_data("企业数据", "评价记录", rid, update_data)
|
||
update_process_status(rid['评价ID'], {"评价流程.证书生成": 1})
|
||
return True
|
||
else:
|
||
update_process_status(rid['评价ID'], {"评价流程.证书生成": -1})
|
||
return False
|
||
|
||
def company_index():
|
||
url = "https://api.fecribd.com/etl_tfse/company/company_index_etl"
|
||
token = "dmfd7FshT!5Wng9^gcCcQV7T6FBnVgl4"
|
||
headers = {'token': token, "Content-Type": "application/json;charset=UTF-8"}
|
||
data = {
|
||
"rid": rating_input['评价ID']
|
||
}
|
||
res = requests.post(url, headers=headers, data=json.dumps(data))
|
||
if res.status_code == 200:
|
||
update_process_status(rid['评价ID'], {"评价流程.综合信用评价主页": 1})
|
||
update_process_status(rid['评价ID'], {"评价流程.财务要素评价主页": 1})
|
||
update_process_status(rid['评价ID'], {"评价流程.公司风险分析主页": 1})
|
||
return True
|
||
else:
|
||
update_process_status(rid['评价ID'], {"评价流程.综合信用评价主页": -1})
|
||
return False
|
||
|
||
if not general_model():
|
||
return "打分失败", 210
|
||
|
||
if not gen_report():
|
||
return "生成报告失败", 210
|
||
|
||
if not gen_certified():
|
||
return "生成证书失败", 210
|
||
|
||
if not company_index():
|
||
return "准备主页数据失败", 210
|
||
|
||
# 更新完成时间和进行状态
|
||
self.finish_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
|
||
self.finish_date = time.strftime("%Y-%m-%d", time.localtime())
|
||
update_time = self.rating_info_dict_to_save(columns=['评价完成时间', '评价完成日期'])
|
||
self.db.update_single_data('企业数据', '评价记录', rid, update_time)
|
||
self.db.update_single_data("企业数据", "评价记录", rid, {"进行状态": "完成"})
|
||
|
||
return "执行成功", 200
|
||
|
||
def questionnaire_obj(self):
|
||
"""问卷选择"""
|
||
def company_type():
|
||
"""获取企业类型"""
|
||
industrys = self.db.find_single_data(
|
||
'综信评价数据',
|
||
'填报数据',
|
||
{'企业ID': self.cid, '评价年度': get_current_year()},
|
||
['行业选择']
|
||
)
|
||
if not len(industrys['行业选择']) > 1:
|
||
return None
|
||
else:
|
||
return industrys['行业选择'][0]
|
||
|
||
industry = company_type()
|
||
try:
|
||
questionnaire_json = '/static/questionnaire/{}.json'.format(industry)
|
||
with open(os.path.dirname(__file__) + questionnaire_json, "r", encoding='utf-8') as f:
|
||
data = json.load(f)
|
||
return data
|
||
except FileNotFoundError:
|
||
data = [
|
||
{"label": "环境", "children": []},
|
||
{"label": "社会责任", "children": []},
|
||
{"label": "公司治理", "children": []}
|
||
]
|
||
return data
|
||
|
||
@staticmethod
|
||
def choose_industry():
|
||
"""行业选择接口"""
|
||
industry_json = get_data_template('industry')
|
||
return industry_json
|
||
|
||
|