tfse-app-api-v0.2/Rating/Credit/CreditImpl.py

445 lines
19 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import os
import time
import json
import random
import requests
from DBHelper.MongoHelper import MongoHelper
from Rating.Credit.CreditObj import RatingRecord, CreditInputData, FinancialReport, BusinessQuestionnaire, BalanceSheet, \
IncomeSheet, AppendixDataSheet
from Rating.Process.ProcessObj import CreditProcess
class CreditUtils:
"""综合信用评价相关方法"""
db = MongoHelper('tfse_v0.21')
@staticmethod
def get_current_year():
"""获取当前年度"""
current_year = time.strftime("%Y", time.localtime()) + ""
return current_year
@staticmethod
def gen_new_rid():
"""生成新的评价ID如果该ID存在则重新生成"""
db = MongoHelper('tfse_v0.21')
def make_id(num):
"""
随机生成字符串
"""
choices = 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789'
salt = ''
for i in range(num):
salt += random.choice(choices)
return salt
# 生成新ID
new_id = make_id(8)
# 检查新ID是否存在如果存在则继续生成新ID
case = db.find_all_data_with_count(
'企业数据',
'评价记录',
{'评价ID': new_id}
)
while case > 0:
new_id = make_id(8)
# 新ID可使用返回新ID
return new_id
@staticmethod
def make_3_report_dates():
"""
生成最近三年报告期 日期
Parameters:
-
Returns:
report_dates 三年报告期
"""
current_year = int(time.strftime("%Y", time.localtime())) - 1
date_01 = str(current_year) + '-12-31'
date_02 = str(current_year - 1) + '-12-31'
date_03 = str(current_year - 2) + '-12-31'
report_dates = [date_01, date_02, date_03]
return report_dates
@staticmethod
def update_process_status(rid, update_data):
"""
调用更新流程接口
Parameters:
update_data dict 需要更新的数据
Returns:
result str 更新结果
"""
url = "http://api.fecribd.com/etl_tfse/company/update_process_data"
token = "dmfd7FshT!5Wng9^gcCcQV7T6FBnVgl4"
headers = {'token': token, 'Content-Type': 'application/json; charset=UTF-8'}
data = json.dumps({"rid": rid, "types": "综合评价流程", "renew_data": update_data})
requests.post(url, headers=headers, data=data)
class CreditRating(CreditInputData, CreditUtils):
"""综合信用评价"""
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, "评价年度": self.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 = self.gen_new_rid()
# report_dates: 生成报告期
report_dates = self.make_3_report_dates()
# company_name: 企业名称
company_name = self.db.find_single_data(
"应用端",
"企业用户",
{"企业ID": self.cid},
["企业名称"]
)
def prepare_input_template():
"""
根据模板生成新的填报数据
input_template: 填报数据模板
"""
self.rid = new_rid
self.cid = self.cid
self.company = company_name['企业名称']
self.year = self.get_current_year()
self.industry = list()
def balance():
"""资产负债表"""
balance_sheet = BalanceSheet()
balance_sheet.report_date = None
balance_sheet.accounts_receivable = None
balance_sheet.stock = None
balance_sheet.total_current_assets = None
balance_sheet.total_assets = None
balance_sheet.short_loan = None
balance_sheet.one_year_liabilities = None
balance_sheet.total_current_liabilities = None
balance_sheet.long_term_loan = None
balance_sheet.total_liabilities = None
balance_sheet.total_owners_equity = None
return balance_sheet.dict_to_save()
def income():
"""利润表"""
income_sheet = IncomeSheet()
income_sheet.report_date = None
income_sheet.operating_income = None
income_sheet.operating_cost = None
income_sheet.total_profit = None
income_sheet.net_profit = None
return income_sheet.dict_to_save()
def appendix():
"""补充数据表"""
appendix_sheet = AppendixDataSheet()
appendix_sheet.report_date = None
appendix_sheet.rd_expenses = None
appendix_sheet.interest_disbursement = None
appendix_sheet.interest_expense = None
return appendix_sheet.dict_to_save()
def financial():
"""财务填报"""
financial_report = FinancialReport()
financial_report.balance_sheet_list = list()
financial_report.income_sheet_list = list()
financial_report.appendix_data_sheet_list = list()
for n in range(3):
b = balance()
b['报告期'] = report_dates[n]
i = income()
i['报告期'] = report_dates[n]
a = appendix()
a['报告期'] = report_dates[n]
financial_report.balance_sheet_list.append(b)
financial_report.income_sheet_list.append(i)
financial_report.appendix_data_sheet_list.append(a)
return financial_report
self.financial_report = financial()
self.business_questionnaire = list()
input_template = self.dict_to_save()
# 插入新的填报记录
self.db.insert_single_data(
'综信评价数据',
'填报数据',
input_template
)
def prepare_rating_record_template():
"""根据模板生成新的评价记录"""
rating_record = RatingRecord()
rating_record.rid = new_rid
rating_record.cid = self.cid
rating_record.company = company_name['企业名称']
rating_record.project = '综合信用评价'
rating_record.methods = '企业申报'
rating_record.status = '进行'
rating_record.result = None
rating_record.report_fid = None
rating_record.certificate_fid = None
rating_record.begin_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
rating_record.submit_time = None
rating_record.finish_time = None
rating_record.finish_date = None
rating_record_template = rating_record.dict_to_save()
self.db.insert_single_data(
'企业数据',
'评价记录',
rating_record_template
)
def prepare_process_template():
"""根据数据模板生成新的评价流程数据"""
credit_process = CreditProcess()
credit_process.fill_in_questionnaire = 0
credit_process.fill_in_financial_data = 0
credit_process.clean_risk_data = 0
credit_process.model_scoring = 0
credit_process.report_generation = 0
credit_process.certificate_generation = 0
credit_process.cc_evaluation_homepage = 0
credit_process.fe_evaluation_homepage = 0
credit_process.cr_analysis_homepage = 0
credit_process.am_details_homepage = 0
process_template = dict()
process_template['评价ID'] = new_rid
process_template['企业ID'] = self.cid
process_template['评价流程'] = credit_process.dict_to_save()
self.db.insert_single_data(
'综信评价数据',
'评价流程',
process_template
)
def prepare_risk_data():
# 调用清洗数据接口,执行风险数据清洗程序
url = "http://api.fecribd.com/etl_tfse/rating/risk_data"
token = "dmfd7FshT!5Wng9^gcCcQV7T6FBnVgl4"
headers = {'token': token, 'Content-Type': 'application/json; charset=UTF-8'}
data = json.dumps({"company_id": self.cid, "company_name": company_name, "evaluation_id": new_rid})
requests.post(url, headers=headers, data=data)
# 风险数据清洗完成后更新流程状态
self.update_process_status(new_rid, {"评价流程.清洗风险数据": 1})
def __main__():
prepare_input_template()
prepare_rating_record_template()
prepare_process_template()
prepare_risk_data()
__main__()
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):
"""保存综合填报信息"""
# 根据cid查询评价记录得到rid
find = {"企业ID": self.cid, "评价项目": "综合信用评价", "进行状态": "进行"}
rid = self.db.find_single_data("企业数据", "评价记录", find, ['评价ID'])
if not rid:
return "没有正在进行的评价", 200
def handle_input_form():
"""处理表单数据"""
def questionnarie():
"""经营问卷"""
questionnarie_data = input_form['经营问卷']
new_questionnaire = list()
for item in questionnarie_data:
if isinstance(item, float) or isinstance(item, int) or item == 0:
item = float(item)
new_questionnaire.append(item)
else:
new_questionnaire.append(item)
input_form['经营问卷'] = new_questionnaire
def financial():
"""财务填报"""
financial_data = input_form['财务填报']
for sheet in ['资产负债表', '利润表', '补充数据表']:
for item in financial_data[sheet]:
for key, value in item.items():
if isinstance(value, float) or isinstance(value, int) or value == 0:
item[key] = float(value)
questionnarie()
# financial()
def business_questionnaire_script():
"""经营问卷"""
business = input_form['经营问卷']
return business
def financial_report_script():
"""财务填报"""
financial = input_form['财务填报']
def balance_sheet_script():
"""资产负债表"""
balance_sheet = 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 = 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 = 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
financial_report = FinancialReport()
financial_report.balance_sheet_list = list()
financial_report.income_sheet_list = list()
financial_report.appendix_data_sheet_list = list()
return financial_report
def rating_input():
"""校验填报数据"""
self.rid = rid['评价ID']
self.cid = self.cid
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.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:
self.update_process_status(rid['评价ID'], {"评价流程.问卷填报": 1})
if process['评价流程']['财务数据填报'] == 0:
self.update_process_status(rid['评价ID'], {"评价流程.财务数据填报": 1})
handle_input_form()
rating_input()
rating_process()
return '保存成功'