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 from Utils.ErrorUtil import LogConditionCheckFailed 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() 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['所有者权益合计'] financial_report.balance_sheet_list.append(balance_sheet.dict_to_save()) def income_sheet_script(): """利润表""" income_sheet = IncomeSheet() 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['净利润'] financial_report.income_sheet_list.append(income_sheet.dict_to_save()) def appendix_sheet_script(): """补充数据表""" appendix_sheet = AppendixDataSheet() for sheet in financial['补充数据表']: appendix_sheet.report_date = sheet['报告期'] appendix_sheet.rd_expenses = sheet['研发费用'] appendix_sheet.interest_disbursement = sheet['计入财务费的利息支出'] appendix_sheet.interest_expense = sheet['资本化利息支出'] financial_report.appendix_data_sheet_list.append(appendix_sheet.dict_to_save()) financial_report = FinancialReport() financial_report.balance_sheet_list = list() financial_report.income_sheet_list = list() financial_report.appendix_data_sheet_list = list() balance_sheet_script() income_sheet_script() appendix_sheet_script() 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 '保存成功' def exec_cc_rating(self): """提交填报数据，开始评价""" # 根据cid查询评价记录得到rid find = {"企业ID": self.cid, "评价项目": "综合信用评价", "进行状态": "进行"} rid = self.db.find_single_data("企业数据", "评价记录", find, ['评价ID']) if not rid: return "没有正在进行的评价", 200 self.db.update_single_data( '企业数据', '评价记录', rid, {"提交填报时间": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}) 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) self.update_process_status(rid['评价ID'], {"评价流程.模型打分": 1}) else: self.update_process_status(rid['评价ID'], {"评价流程.模型打分": -1}) raise LogConditionCheckFailed("模型" "打分失败", 200) 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) self.update_process_status(rid['评价ID'], {"评价流程.报告生成": 1}) else: self.update_process_status(rid['评价ID'], {"评价流程.报告生成": -1}) raise LogConditionCheckFailed("生成报告失败", 200) 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) self.update_process_status(rid['评价ID'], {"评价流程.证书生成": 1}) else: self.update_process_status(rid['评价ID'], {"评价流程.证书生成": -1}) raise LogConditionCheckFailed("生成证书失败", 200) 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: self.update_process_status(rid['评价ID'], {"评价流程.综合信用评价主页": 1}) self.update_process_status(rid['评价ID'], {"评价流程.财务要素评价主页": 1}) self.update_process_status(rid['评价ID'], {"评价流程.公司风险分析主页": 1}) else: self.update_process_status(rid['评价ID'], {"评价流程.综合信用评价主页": -1}) raise LogConditionCheckFailed("生成主页数据失败", 200) def update_rating_records(): """更新完成时间和进行状态""" finish_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) finish_date = time.strftime("%Y-%m-%d", time.localtime()) update_data = {"进行状态": "完成", "评价完成时间": finish_time, "评价完成日期": finish_date} self.db.update_single_data('企业数据', '评价记录', rid, update_data) def __main__(): general_model() gen_report() gen_certified() company_index() update_rating_records() __main__()