638 lines
29 KiB
Python
638 lines
29 KiB
Python
import json
|
||
import time
|
||
import random
|
||
import requests
|
||
|
||
from DBHelper.MongoHelper import MongoHelper
|
||
from Objects.Common.Rating.FECRRating import RatingRecord
|
||
from Objects.InProject.Rating.esg import EnvironmentQuestionnaire, ThreeYearDataByE, OtherTypeDataByE, \
|
||
ThreeYearDataByS, SocietyQuestionnaire, OtherTypeDataByS, ThreeYearDataByG, GovernanceQuestionnaire, \
|
||
OtherTypeDataByG, EsgInputData
|
||
from Objects.InProject.Rating.process import EsgNewProcess, EsgProcess
|
||
from Utils.ErrorUtil import ReturnConditionCheckFailed
|
||
|
||
|
||
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 make_3_report_years():
|
||
"""
|
||
生成最近三年报告期
|
||
Parameters:
|
||
-
|
||
Returns:
|
||
report_dates 三年报告期
|
||
"""
|
||
current_year = int(time.strftime("%Y", time.localtime())) - 1
|
||
date_01 = str(current_year) + '年'
|
||
date_02 = str(current_year - 1) + '年'
|
||
date_03 = str(current_year - 2) + '年'
|
||
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": "ESG评价流程", "renew_data": update_data})
|
||
requests.post(url, headers=headers, data=data)
|
||
|
||
|
||
class EsgRating(EsgInputData, CreditUtils):
|
||
"""企业ESG评价"""
|
||
db = MongoHelper('tfse_v0.21')
|
||
|
||
def start_esg_rating_obj(self):
|
||
"""开始ESG评价"""
|
||
|
||
def check_input_data():
|
||
"""
|
||
input_data: 已存在的填报数据
|
||
根据企业ID+当前年度,查询是否存在填报数据
|
||
若存在填报数据，则返回填报数据
|
||
若不存在填报数据，则返回False
|
||
"""
|
||
input_data = self.db.find_single_data(
|
||
"ESG评价数据",
|
||
"填报数据",
|
||
{"企业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_years()
|
||
# company_name: 企业名称
|
||
company_name = self.db.find_single_data(
|
||
"应用端",
|
||
"企业用户",
|
||
{"企业ID": self.cid},
|
||
["企业名称"]
|
||
)
|
||
|
||
def prepare_input_template():
|
||
"""
|
||
根据模板生成新的填报数据
|
||
input_template: 填报数据模板
|
||
"""
|
||
|
||
def environment():
|
||
"""环境问卷"""
|
||
def three_year_data(i):
|
||
"""近三年数据"""
|
||
e = ThreeYearDataByE()
|
||
e.year = report_dates[i]
|
||
e.raw_coal = None
|
||
e.coke = None
|
||
e.crude = None
|
||
e.fuel_oil = None
|
||
e.gasoline = None
|
||
e.kerosene = None
|
||
e.diesel_fuel = None
|
||
e.liquefied_petroleum_gas = None
|
||
e.natural_gas = None
|
||
e.gas = None
|
||
e.electricity = None
|
||
e.water = None
|
||
e.green_income = None
|
||
return e.dict_to_save()
|
||
|
||
environment_questionnaire = EnvironmentQuestionnaire()
|
||
environment_questionnaire.e_three_year_data = list()
|
||
for num in range(3):
|
||
environment_questionnaire.e_three_year_data.append(three_year_data(num))
|
||
environment_questionnaire.e_other_type_data = list()
|
||
return environment_questionnaire
|
||
|
||
def society():
|
||
"""社会问卷"""
|
||
def three_year_data(i):
|
||
s = ThreeYearDataByS()
|
||
s.year = report_dates[i]
|
||
s.employee_total = None
|
||
s.leaving_total = None
|
||
s.salary_level = None
|
||
s.work_length = None
|
||
s.labor_dispute = None
|
||
s.security_incident = None
|
||
s.training_times = None
|
||
s.supplier = None
|
||
s.sales = None
|
||
s.return_sales = None
|
||
s.donate = None
|
||
return s.dict_to_save()
|
||
|
||
society_questionnaire = SocietyQuestionnaire()
|
||
society_questionnaire.s_three_year_data = list()
|
||
for num in range(3):
|
||
society_questionnaire.s_three_year_data.append(three_year_data(num))
|
||
society_questionnaire.s_other_type_data = list()
|
||
return society_questionnaire
|
||
|
||
def governance():
|
||
"""治理问卷"""
|
||
def three_year_data(i):
|
||
g = ThreeYearDataByG()
|
||
g.year = report_dates[i]
|
||
g.audit_report = None
|
||
g.net_assets = None
|
||
g.net_profit = None
|
||
return g.dict_to_save()
|
||
|
||
governance_questionnaire = GovernanceQuestionnaire()
|
||
governance_questionnaire.g_three_year_data = list()
|
||
for num in range(3):
|
||
governance_questionnaire.g_three_year_data.append(three_year_data(num))
|
||
governance_questionnaire.g_other_type_data = list()
|
||
return governance_questionnaire
|
||
|
||
self.rid = new_rid
|
||
self.cid = self.cid
|
||
self.company = company_name['企业名称']
|
||
self.year = self.get_current_year()
|
||
self.industry = list()
|
||
self.current_year_income = dict()
|
||
self.current_year_income[report_dates[0]] = None
|
||
self.current_year_income[report_dates[1]] = None
|
||
self.current_year_income[report_dates[2]] = None
|
||
self.environment = environment()
|
||
self.society = society()
|
||
self.governance = governance()
|
||
input_template = self.dict_to_save()
|
||
|
||
# 插入新的填报记录
|
||
self.db.insert_single_data(
|
||
'ESG评价数据',
|
||
'填报数据',
|
||
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 = 'ESG评价'
|
||
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 = EsgProcess()
|
||
preparation = credit_process.EsgDataPreparation()
|
||
preparation.fill_in_data = 0
|
||
preparation.risk_data = 0
|
||
credit_process.data_preparation = preparation
|
||
score = credit_process.EsgModelScoring()
|
||
score.rating_result = 0
|
||
credit_process.model_scoring = score
|
||
document = credit_process.EsgDocumentGeneration()
|
||
document.report_data = 0
|
||
document.report = 0
|
||
document.certificate = 0
|
||
credit_process.document_generation = document
|
||
clean = credit_process.EsgDataCleaning()
|
||
clean.esg_evaluation_analysis = 0
|
||
credit_process.data_cleaning = clean
|
||
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(
|
||
'ESG评价数据',
|
||
'评价流程',
|
||
process_template
|
||
)
|
||
|
||
def prepare_risk_data():
|
||
# 调用清洗数据接口，执行风险数据清洗程序
|
||
url = "http://api.fecribd.com/etl_tfse/rating/esg_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_input_general(self, input_form):
|
||
"""保存填报数据"""
|
||
# 根据cid查询评价记录得到rid
|
||
find = {"企业ID": self.cid, "评价项目": "ESG评价", "进行状态": "进行"}
|
||
rid = self.db.find_single_data("企业数据", "评价记录", find, ['评价ID'])
|
||
|
||
if not rid:
|
||
return "没有正在进行的评价", 200
|
||
|
||
def handle_input_form():
|
||
"""处理表单数据"""
|
||
|
||
def environment():
|
||
"""环境问卷"""
|
||
three_year_data = input_form['环境问卷']['近三年公司数据']
|
||
for item in three_year_data:
|
||
for k, v in item.items():
|
||
if isinstance(v, float) or isinstance(v, int) or item == 0:
|
||
item[k] = float(v)
|
||
|
||
def society():
|
||
"""社会问卷"""
|
||
three_year_data = input_form['社会问卷']['近三年公司数据']
|
||
filter_list = ['员工总数（年底）', '当年离职人数', '劳资纠纷次数', '安全事故发生次数', '组织培训次数']
|
||
for item in three_year_data:
|
||
for k, v in item.items():
|
||
if k not in filter_list and (isinstance(v, float) or isinstance(v, int) or item == 0):
|
||
item[k] = float(v)
|
||
|
||
def governance():
|
||
"""治理问卷"""
|
||
three_year_data = input_form['治理问卷']['近三年公司数据']
|
||
for item in three_year_data:
|
||
for k, v in item.items():
|
||
if isinstance(v, float) or isinstance(v, int) or item == 0:
|
||
item[k] = float(v)
|
||
|
||
environment()
|
||
society()
|
||
governance()
|
||
|
||
def environment_script():
|
||
"""环境问卷校验"""
|
||
questionnaire = input_form['环境问卷']
|
||
|
||
def three_year_data_script():
|
||
"""近三年公司数据"""
|
||
three_year_data = ThreeYearDataByE()
|
||
for data in questionnaire['近三年公司数据']:
|
||
three_year_data.year = data['年份']
|
||
three_year_data.raw_coal = data['公司当年消耗的原煤（千克）']
|
||
three_year_data.coke = data['公司当年消耗的焦炭（千克）']
|
||
three_year_data.crude = data['公司当年消耗的原油（升）']
|
||
three_year_data.fuel_oil = data['公司当年消耗的燃料油（升）']
|
||
three_year_data.gasoline = data['公司当年消耗的汽油（升）']
|
||
three_year_data.kerosene = data['公司当年消耗的煤油（升）']
|
||
three_year_data.diesel_fuel = data['公司当年消耗的柴油（升）']
|
||
three_year_data.liquefied_petroleum_gas = data['公司当年消耗的液化石油气（千克）']
|
||
three_year_data.natural_gas = data['公司当年消耗的天然气（立方米）']
|
||
three_year_data.gas = data['公司当年消耗的煤气（立方米）']
|
||
three_year_data.electricity = data['公司当年消耗的电（千瓦时）']
|
||
three_year_data.water = data['公司当年耗水（吨）']
|
||
three_year_data.green_income = data['公司当年绿色业务收入（万元）——包括不限于清洁能源、清洁交通、绿色建筑、清洁技术等']
|
||
eq.e_three_year_data.append(three_year_data.dict_to_save())
|
||
|
||
def other_type_data_script():
|
||
"""其他类型问卷"""
|
||
if questionnaire['其他类型问卷']:
|
||
other_type_data = OtherTypeDataByE()
|
||
other_type_data.question_01 = questionnaire['其他类型问卷'][0][0]
|
||
other_type_data.question_02 = questionnaire['其他类型问卷'][1][0]
|
||
other_type_data.question_03 = questionnaire['其他类型问卷'][2][0]
|
||
other_type_data.question_04 = questionnaire['其他类型问卷'][3][0]
|
||
other_type_data.question_05 = questionnaire['其他类型问卷'][4][0]
|
||
other_type_data.question_06 = questionnaire['其他类型问卷'][5][0]
|
||
other_type_data.question_07 = questionnaire['其他类型问卷'][6][0]
|
||
other_type_data.question_08 = questionnaire['其他类型问卷'][7][0]
|
||
other_type_data.list_to_save()
|
||
return questionnaire['其他类型问卷']
|
||
|
||
eq = EnvironmentQuestionnaire()
|
||
eq.e_three_year_data = list()
|
||
three_year_data_script()
|
||
other_type_data_script()
|
||
return eq
|
||
|
||
def society_script():
|
||
"""社会问卷校验"""
|
||
questionnaire = input_form['社会问卷']
|
||
|
||
def three_year_data_script():
|
||
"""近三年公司数据"""
|
||
three_year_data = ThreeYearDataByS()
|
||
for data in questionnaire['近三年公司数据']:
|
||
three_year_data.year = data['年份']
|
||
three_year_data.employee_total = data['员工总数（年底）']
|
||
three_year_data.leaving_total = data['当年离职人数']
|
||
three_year_data.salary_level = data['人均薪酬水平（元/月）']
|
||
three_year_data.work_length = data['劳动合同要求工作时长（每周小时数）']
|
||
three_year_data.labor_dispute = data['劳资纠纷次数']
|
||
three_year_data.security_incident = data['安全事故发生次数']
|
||
three_year_data.training_times = data['组织培训次数']
|
||
three_year_data.supplier = data['公司从前3大供应商拿货占全部供应商比例（%）']
|
||
three_year_data.sales = data['公司前3大客户销售额占全部销售比例（%）']
|
||
three_year_data.return_sales = data['返修、退回、投诉产品对应销售额占全部销售比例（%）']
|
||
three_year_data.donate = data['扶贫+捐赠规模（万元）']
|
||
sq.s_three_year_data.append(three_year_data.dict_to_save())
|
||
|
||
def other_type_data_script():
|
||
"""其他类型问卷"""
|
||
if questionnaire['其他类型问卷']:
|
||
other_type_data = OtherTypeDataByS()
|
||
other_type_data.question_01 = questionnaire['其他类型问卷'][0]
|
||
other_type_data.question_02 = questionnaire['其他类型问卷'][1]
|
||
other_type_data.question_03 = questionnaire['其他类型问卷'][2]
|
||
other_type_data.question_04 = questionnaire['其他类型问卷'][3]
|
||
other_type_data.list_to_save()
|
||
return questionnaire['其他类型问卷']
|
||
|
||
sq = SocietyQuestionnaire()
|
||
sq.s_three_year_data = list()
|
||
three_year_data_script()
|
||
sq.s_other_type_data = other_type_data_script()
|
||
return sq
|
||
|
||
def governance_script():
|
||
"""治理问卷校验"""
|
||
questionnaire = input_form['治理问卷']
|
||
|
||
def three_year_data_script():
|
||
"""近三年公司数据"""
|
||
three_year_data = ThreeYearDataByG()
|
||
for data in questionnaire['近三年公司数据']:
|
||
three_year_data.year = data['年份']
|
||
three_year_data.audit_report = data['公司是否有审计报告']
|
||
three_year_data.net_assets = data['公司净资产（万元）']
|
||
three_year_data.net_profit = data['公司净利润（万元）']
|
||
gq.g_three_year_data.append(three_year_data.dict_to_save())
|
||
|
||
def other_type_data_script():
|
||
"""其他类型问卷"""
|
||
if questionnaire['其他类型问卷']:
|
||
other_type_data = OtherTypeDataByG()
|
||
other_type_data.question_01 = questionnaire['其他类型问卷'][0]
|
||
other_type_data.question_02 = questionnaire['其他类型问卷'][1]
|
||
other_type_data.question_03 = int(questionnaire['其他类型问卷'][2])
|
||
other_type_data.question_04 = int(questionnaire['其他类型问卷'][3])
|
||
other_type_data.question_05 = questionnaire['其他类型问卷'][4]
|
||
other_type_data.question_06 = questionnaire['其他类型问卷'][5]
|
||
other_type_data.question_07 = questionnaire['其他类型问卷'][6]
|
||
other_type_data.question_08 = questionnaire['其他类型问卷'][7]
|
||
other_type_data.list_to_save()
|
||
return questionnaire['其他类型问卷']
|
||
|
||
gq = GovernanceQuestionnaire()
|
||
gq.g_three_year_data = list()
|
||
three_year_data_script()
|
||
gq.g_other_type_data = other_type_data_script()
|
||
return gq
|
||
|
||
def rating_input():
|
||
"""校验填报数据并保存"""
|
||
self.rid = rid['评价ID']
|
||
self.company = input_form['企业名称']
|
||
self.year = input_form['评价年度']
|
||
self.industry = input_form['所属行业']
|
||
self.current_year_income = input_form['公司当年收入（万元）']
|
||
self.environment = environment_script()
|
||
self.society = society_script()
|
||
self.governance = governance_script()
|
||
input_data = self.dict_to_save()
|
||
input_data['环境问卷']['其他类型问卷'] = input_form['环境问卷']['其他类型问卷']
|
||
self.db.update_single_data(
|
||
'ESG评价数据',
|
||
'填报数据',
|
||
rid,
|
||
input_data
|
||
)
|
||
self.update_process_status(rid['评价ID'], {"评价流程.数据准备.填报数据": 1})
|
||
|
||
def __main__():
|
||
handle_input_form()
|
||
rating_input()
|
||
|
||
__main__()
|
||
|
||
def exec_esg_rating(self):
|
||
"""提交数据，开始esg评价"""
|
||
# 根据cid查询评价记录得到rid
|
||
find = {"企业ID": self.cid, "评价项目": "ESG评价", "进行状态": "进行"}
|
||
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(
|
||
'ESG评价数据',
|
||
'填报数据',
|
||
rid,
|
||
["评价ID", "企业ID", "企业名称", "评价年度", "所属行业", "公司当年收入（万元）", "环境问卷", "社会问卷", "治理问卷"]
|
||
)
|
||
risk_data = self.db.find_single_data(
|
||
'ESG评价数据',
|
||
'风险数据',
|
||
rid,
|
||
["严重违法", "失信人", "法律诉讼", "税收违法", "经营异常", "行政处罚", "被执行人"]
|
||
)
|
||
return {**input_data, **{"风险数据": risk_data}}
|
||
|
||
rating_input = get_rating_data()
|
||
|
||
def esg_model():
|
||
"""模型打分"""
|
||
url = "http://api.fecribd.com/tfse_rating/esg/esg_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)
|
||
else:
|
||
raise ReturnConditionCheckFailed("模型打分失败", 200)
|
||
|
||
def gen_report():
|
||
"""报告生成"""
|
||
url = "http://api.fecribd.com/tfse_rating/report/generate_esg_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'], {"评价流程.文档生成.ESG报告": 1})
|
||
else:
|
||
self.update_process_status(rid['评价ID'], {"评价流程.文档生成.ESG报告": -1})
|
||
raise ReturnConditionCheckFailed("报告生成失败", 200)
|
||
|
||
def gen_certified():
|
||
"""证书生成"""
|
||
url = "http://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'],
|
||
"type": "esg"
|
||
}
|
||
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'], {"评价流程.文档生成.ESG证书": 1})
|
||
else:
|
||
self.update_process_status(rid['评价ID'], {"评价流程.文档生成.ESG证书": -1})
|
||
raise ReturnConditionCheckFailed("证书生成失败", 200)
|
||
|
||
def esg_index():
|
||
url = "http://api.fecribd.com/etl_tfse/company/esg_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'], {"评价流程.数据清洗.ESG评价分析": 1})
|
||
else:
|
||
self.update_process_status(rid['评价ID'], {"评价流程.数据清洗.ESG评价分析": -1})
|
||
raise ReturnConditionCheckFailed("主页ESG数据生成失败", 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__():
|
||
esg_model()
|
||
gen_report()
|
||
gen_certified()
|
||
esg_index()
|
||
update_rating_records()
|
||
|
||
__main__()
|
||
return '执行成功', 200
|