import json import os import random import re import time import base64 import requests from werkzeug.security import generate_password_hash from Crypto.PublicKey import RSA from Crypto.Cipher import PKCS1_v1_5 as PKCS1_cipher from Board.board_db import FIND_DATA, INSERT_DATA from TestCenter.tc_db import update_user_info_by_cid, find_user_info_by_cid, delete_data, find_data, update_data, \ delete_pdf, find_admin_data, insert_admin_data def search_evaluate_record_scripts(types): """ 根据评价类型 Parameters: types str 评价类型 Returns: result list 评价数据库名称 """ if types == '综合信用评价': result = ['评价结果', '报告数据', '综合信用报告', '评价证书', '综合评价填报'] else: result = ['ESG评价结果', 'ESG报告数据', 'esg报告', 'ESG评价证书', 'ESG评价填报'] return result def delete_rid_cascade_scripts(rid, record, data): """ 级联删除与评价ID有关的评价数据 1.根据rid和评价类型决定删除评价类型（综合信用评价，ESG评价） 2.评价记录选择（删除记录：删除跟此rid相关的所有数据；更改状态：可不删除综合评价填报） Parameters: rid: 评价ID record: 评价记录 data: 评价数据 Returns: res: True 执行结果 """ def delete_all_records(): """ 根据rid删除评价数据库相匹配数据 Parameters: - Returns: info str 删除结果 """ # 删除填报数据 if '综合信用填报' in data: delete_data('评价', '综合评价填报', {'评价ID': rid}) elif 'ESG评价填报' in data: delete_data('评价', 'ESG评价填报', {'评价ID': rid}) else: return '如选删除记录，请勾选综合评价填报' # 删除与rid相关的所有数据 delete_data('评价', data[0], {'评价ID': rid}) delete_data('评价', data[1], {'评价ID': rid}) report_id = rids[0]['报告fid'] delete_pdf('评价', data[2], report_id) certificate_id = rids[0]['证书fid'] delete_pdf('评价', data[3], certificate_id) delete_data('评价', '评价记录', {'评价ID': rid}) return '删除成功' def delete_part_records(): """ 根据rid删除评价数据库相匹配数据 Parameters: - Returns: info str 删除结果 """ # 删除相关数据 if '综合信用填报' in data: delete_data('评价', '综合评价填报', {'评价ID': rid}) elif 'ESG评价填报' in data: delete_data('评价', 'ESG评价填报', {'评价ID': rid}) # 删除rid相关的所有数据 delete_data('评价', data[0], {'评价ID': rid}) delete_data('评价', data[1], {'评价ID': rid}) report_id = rids[0]['报告fid'] delete_pdf('评价', data[2], report_id) certificate_id = rids[0]['证书fid'] delete_pdf('评价', data[3], certificate_id) # 更新此条数据 insert = {'进行状态': '进行', '评价结果': None, '评价时间': None, '报告fid': None, '证书fid': None} res = update_data('评价', '评价记录', {'评价ID': rid}, insert) if res: return '状态更改成功' else: return '评价记录数据重置失败' # main rids = find_data('评价', '评价记录', {'评价ID': rid}) if rids: # 判断状态是否已经完成 if rids[0]['进行状态'] == '完成': # 判断评价记录为哪种类型 if record == '删除记录': info = delete_all_records() else: info = delete_part_records() return info else: return '不能删除正在进行的评价记录' else: return '未查询到此rid相关记录' def delete_cid_cascade_scripts(cid, data): """ 级联删除uid有关的所有数据 Parameters: cid: 企业ID data: 数据库名称集合 Returns: res: True 执行结果 """ for name in data: delete_data('企业', name, {'企业ID': cid}) return '企业数据删除成功' def generate_rid_information_script(rid, types): """ 根据rid生成企业主页信息 Parameters: rid str 评价ID types str 评价类型 Returns: result str 生成结果 """ def rating_script(): """ 根据rid执行综合信用评价流程 Parameters: rid str 评价ID Returns: result type 返回数据 """ record = find_data('评价', '评价记录', {'评价ID': rid, '进行状态': '进行'}) if not record: return "没有正在进行的评价" inputs = find_data('评价', '综合评价填报', {'评价ID': rid}) if not inputs: return '未查询到此rid相关填报数据' risks = find_data('评价', '风险数据', {'评价ID': rid}) if not risks: return '未查询到此rid相关风险数据' risk = {"风险数据": risks[0]} risk["风险数据"].pop('企业ID') risk["风险数据"].pop('评价ID') rating_inputs = {**inputs[0], **risk} # 打分模型 def general_model(): url = "http://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_inputs}) res = requests.post(url, headers=headers, data=data) if res.status_code == 200: insert_data = { "评价结果": json.loads(res.text)['result'], "评价时间": time.strftime("%Y-%m-%d", time.localtime()) } update_data('评价', '评价记录', {"评价ID": rid}, insert_data) return True else: return False # 生成报告 def gen_report(): url = "http://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_inputs['企业名称'], "cid": rating_inputs['企业ID'], "rid": rating_inputs['评价ID'] } res = requests.post(url, headers=headers, data=json.dumps(data)) if res.status_code == 200: insert_data = { "报告fid": json.loads(res.text)['result']['FileID'] } update_data('评价', '评价记录', {"评价ID": rid}, insert_data) return True else: return False # 生成证书 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_inputs['企业名称'], "cid": rating_inputs['企业ID'], "rid": rating_inputs['评价ID'] } res = requests.post(url, headers=headers, data=json.dumps(data)) if res.status_code == 200: insert_data = { "证书fid": json.loads(res.text)['result']['FileID'] } update_data('评价', '评价记录', {"评价ID": rid}, insert_data) return True else: return False # 生成企业主页数据 def company_index(): url = "http://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_inputs['评价ID'] } res = requests.post(url, headers=headers, data=json.dumps(data)) if res.status_code == 200: return True else: return False if not general_model(): return False, "打分失败" if not gen_report(): return False, "生成报告失败" if not gen_certified(): return False, "生成证书失败" if not company_index(): return False, "准备主页数据失败" update_data('评价', '评价记录', {'评价ID': rid}, {'进行状态': '完成'}) return '执行完成' def esg_rating_script(): """ 根据rid执行esg评价流程 Parameters: rid str 评价ID Returns: result type 返回数据 """ record = find_data('评价', '评价记录', {'评价ID': rid, '进行状态': '进行'}) if not record: return "没有正在进行的评价" inputs = find_data('评价', 'ESG评价填报', {'评价ID': rid}) if not inputs: return '未查询到此rid相关填报数据' risks = find_data('评价', 'ESG风险数据', {'评价ID': rid}) if not risks: return '未查询到此rid相关风险数据' risk = {"风险数据": risks[0]} risk["风险数据"].pop('企业ID') risk["风险数据"].pop('评价ID') risk["风险数据"].pop('更新日期') rating_inputs = {**inputs[0], **risk} # 打分模型 def general_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_inputs}) res = requests.post(url, headers=headers, data=data) if res.status_code == 200: insert_data = { "评价结果": json.loads(res.text)['result'], "评价时间": time.strftime("%Y-%m-%d", time.localtime()) } update_data('评价', '评价记录', {"评价ID": rid}, insert_data) return True else: return False # 生成报告 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_inputs['企业名称'], "cid": rating_inputs['企业ID'], "rid": rating_inputs['评价ID'] } res = requests.post(url, headers=headers, data=json.dumps(data)) if res.status_code == 200: insert_data = { "报告fid": json.loads(res.text)['result']['FileID'] } update_data('评价', '评价记录', {"评价ID": rid}, insert_data) return True else: return False # 生成证书 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_inputs['企业名称'], "cid": rating_inputs['企业ID'], "rid": rating_inputs['评价ID'], "type": 'esg' } res = requests.post(url, headers=headers, data=json.dumps(data)) if res.status_code == 200: insert_data = { "证书fid": json.loads(res.text)['result']['FileID'] } update_data('评价', '评价记录', {"评价ID": rid}, insert_data) return True else: return False # 生成企业主页数据 def company_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_inputs['评价ID'] } res = requests.post(url, headers=headers, data=json.dumps(data)) if res.status_code == 200: return True else: return False if not general_model(): return "打分失败" if not gen_report(): return "生成报告失败" if not gen_certified(): return "生成证书失败" if not company_index(): return "准备主页数据失败" update_data('评价', '评价记录', {'评价ID': rid}, {'进行状态': '完成'}) return '执行完成' if types == '综合信用评价': info = rating_script() else: info = esg_rating_script() return info def company_verify_script(company, code, legal_person, cid): """ 输入企业名称、统一社会信用代码、法人姓名进行企业认证 Parameters: company: 企业名称 code: 统一社会信用代码 legal_person: 法人姓名 cid: 企业ID Returns: result: 认证结果 """ def check_verified(): # 是否认证 r = find_user_info_by_cid(cid) if len(r) < 0: return '未查到该用户信息' else: auth = r[0]['已认证'] if auth == '是': return '该用户已认证' # 认证信息有缺失 if None in [company, code, legal_person]: return '认证信息不完整' def tyc_api(): """ # 调用天眼查企业三要素验证接口 Parameters: - Returns: result: 查询结果 """ token = "5407573b-e41e-4ab4-9f46-daa0032ad0a4" base = "http://open.api.tianyancha.com/services/open/ic/verify/2.0?code={}&name={}&legalPersonName={}" url = base.format(code, company, legal_person) headers = {'Authorization': token} res = json.loads(requests.get(url, headers=headers).text) return res def main_process(): check_result = check_verified() if check_result: return check_result try: result = tyc_api() if result['result']['result'] == 1: # 初始化企业数据 init_url = "http://api.fecribd.com/etl_tfse/company/init_company_data" init_token = "dmfd7FshT!5Wng9^gcCcQV7T6FBnVgl4" init_headers = {'token': init_token} data = json.dumps({"cid": cid, "company_name": company}) requests.post(init_url, headers=init_headers, data=data) # 更新用户认证状态 update_user_info_by_cid(cid, {"企业名称": company, "已认证": "是"}) return '认证成功' else: return '认证失败' except Exception: return '认证失败' return main_process() def change_verified_script(cid): """ 将企业认证状态改为否 Parameters: cid: 企业ID Returns: res: True 执行结果 """ # 更新用户认证状态 insert_dict = dict() insert_dict['企业名称'] = None insert_dict['统一社会信用代码'] = None insert_dict['法人姓名'] = None insert_dict['法人身份证'] = None insert_dict['营业执照fid'] = None insert_dict['身份证fid'] = None info = update_user_info_by_cid(cid, {"企业名称": None, "已认证": "否", "认证信息": insert_dict}) return info def new_test_company_impl(email, company, code, legal_person): """ 新增一家测试企业 """ def check_mail_fmt(): """ 邮箱地址格式校验 Parameters: - Returns: result: 邮箱校验结果，正确返回True，不正确返回False """ regex = "^.+\\@(\\[?)[a-zA-Z0-9\\-\\.]+\\.([a-zA-Z]{2,3}|[0-9]{1,3})(\\]?)$" case = (len(email) > 7) and (re.match(regex, email) is not None) result = True if case else False return result def check_registered(): """ 检查邮箱是否被注册 Parameters: email: 邮箱 Returns: 邮箱未被注册 False 邮箱已被注册 True """ return False if FIND_DATA("TFSE", "用户", "验证记录", {"邮箱": email}) == [] else True def gen_new_cid(): """ 生成新的企业ID，如果该ID存在，则重新生成 Parameters: - Returns: new_id: 生成的企业ID """ 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 = FIND_DATA("TFSE", "用户", "用户信息", {"企业ID": new_id}) is [] while case: new_id = make_id(8) # 新ID可使用，返回新ID return new_id cid = gen_new_cid() def decrypt_data(**kwargs): """ 用私钥解密 Parameters: encrypt_msg str 加密信息 Returns: 执行正确 解密后结果 执行错误 False """ try: encrypt_msg = kwargs['encrypt_msg'] # 加密信息 with open(os.path.abspath(os.path.dirname(__file__) + '/static/rsa_private_key.pem')) as f: data = f.read() private_key = RSA.importKey(data) cipher = PKCS1_cipher.new(private_key) back_text = cipher.decrypt(base64.b64decode(encrypt_msg), 0) return back_text.decode('utf-8') except Exception: return False def create_user_by_template(): file_rel_path = "/static/user.json" with open(os.path.abspath(os.path.dirname(__file__) + file_rel_path), "r", encoding='utf-8') as f: user = json.load(f) user['企业ID'] = cid user['邮箱'] = email user['头像fid'] = "61a15b33e55c00003e0048e4" user['密码'] = generate_password_hash("Fecr1988.") user['注册时间'] = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) user['已认证'] = '否' # 插入新用户数据 INSERT_DATA("TFSE", "用户", "用户信息", user) return True def new_company_main_process(): # 校验邮箱格式 if not check_mail_fmt(): return "邮箱格式错误" # 检验邮箱是否被注册 if check_registered(): return "邮箱已被注册" # 通过以上校验，生成新用户信息，插入新用户数据 res = create_user_by_template() if res is not True: return '新增企业失败' return company_verify_script(company, code, legal_person, cid) return new_company_main_process() def get_company_process_script(types, rid, cid): """ 查询企业测评流程 Parameters: types str 数据集名称 rid str 评价ID cid str 企业ID Returns: result dict 流程数据 """ data = find_admin_data("评价", types, {"评价ID": rid, "企业ID": cid}) return data if __name__ == '__main__': insert = { "企业ID": "QZhQ12Cw", "评价ID": "bw1tcQfP", "评价流程": { "数据填报": 1, "清洗风险数据": 1, "模型打分": 1, "报告生成": 1, "证书生成": -1, "主页ESG数据生成": 0 } } insert_admin_data("评价", "ESG评价流程", insert)