243 lines
7.8 KiB
Python
243 lines
7.8 KiB
Python
import os
|
||
|
||
from Certificate.PdfCertificate import CertificateGenerator
|
||
from DBHelper.MongoHelper import MongoHelper
|
||
from datetime import datetime
|
||
from Certificate.scripts.path_tool import get_json_path, get_gen_certificate_path
|
||
import json
|
||
|
||
|
||
class GenerateCertificate:
|
||
"""证书生成类"""
|
||
db = MongoHelper("tfse_v0.21")
|
||
|
||
cid = None
|
||
rid = None
|
||
company = None
|
||
type = None
|
||
|
||
@staticmethod
|
||
def generate_code(param1, param2, param3):
|
||
"""
|
||
生成校验码（4位）
|
||
第一位由项目类型+项目编号生成
|
||
后三位由时间戳生成
|
||
Parameters:
|
||
param1: string 项目类型
|
||
param2: string 项目编号
|
||
param3: string 时间戳
|
||
Returns:
|
||
result: type 校验码
|
||
"""
|
||
|
||
def generate_01():
|
||
"""
|
||
生成第一位校验码
|
||
由项目类型+项目编号生成
|
||
Parameters:
|
||
-
|
||
Returns:
|
||
result: string 校验码
|
||
"""
|
||
param1_list = list(param1)
|
||
res = 0
|
||
for item in param1_list:
|
||
res += letter[item]
|
||
if param2 == '01':
|
||
res = res * 2
|
||
else:
|
||
res = res * 1
|
||
|
||
res_list = list(str(res))
|
||
return res_list[0]
|
||
|
||
def generate_02():
|
||
"""
|
||
生成后三位校验码
|
||
由时间戳生成
|
||
Parameters:
|
||
-
|
||
Returns:
|
||
result: string 校验码
|
||
"""
|
||
|
||
def generate_random(data):
|
||
"""
|
||
通过数组中的数字生成随机数
|
||
Parameters:
|
||
data: 数字列表
|
||
Returns:
|
||
result: string 校验码
|
||
"""
|
||
res = list(str(sum([int(x) for x in data])))
|
||
return str(res[0])
|
||
|
||
# Paramter
|
||
param3_list = list(param3)
|
||
param3_01 = generate_random(param3_list[0:3])
|
||
param3_02 = generate_random(param3_list[3:6])
|
||
param3_03 = generate_random(param3_list[6:9])
|
||
|
||
return param3_01 + param3_02 + param3_03
|
||
|
||
with open(get_json_path(common='letter.json'), 'r', encoding='utf-8') as f:
|
||
letter = json.load(f)
|
||
result = generate_01() + generate_02()
|
||
|
||
return result
|
||
|
||
@staticmethod
|
||
def effective_date():
|
||
"""
|
||
生成有效日期
|
||
Parameters:
|
||
-
|
||
Returns:
|
||
result: 有效日期
|
||
"""
|
||
year = datetime.now().year
|
||
|
||
date = '{}年1月1日至{}年1月1日'.format(year, year + 1)
|
||
|
||
return date
|
||
|
||
def generate_number(self):
|
||
"""
|
||
生成证书编号
|
||
Parameters:
|
||
-
|
||
Returns:
|
||
result: 证书编号
|
||
"""
|
||
project_type = 'CCRE'
|
||
project_number = '01'
|
||
project_time = str(datetime.now().strftime('%Y%m%d%H%M'))
|
||
project_code = self.generate_code(project_type, project_number, project_time)
|
||
|
||
number = project_type + project_number + project_time + project_code
|
||
|
||
return number
|
||
|
||
def generate_esg_number(self):
|
||
"""
|
||
生成esg证书编号
|
||
Parameters:
|
||
-
|
||
Returns:
|
||
result: 证书编号
|
||
"""
|
||
project_type = 'ESG'
|
||
project_number = '02'
|
||
project_time = str(datetime.now().strftime('%Y%m%d%H%M'))
|
||
project_code = self.generate_code(project_type, project_number, project_time)
|
||
|
||
number = project_type + project_number + project_time + project_code
|
||
|
||
return number
|
||
|
||
def generate_certificate(self):
|
||
"""证书生成"""
|
||
|
||
def get_certificate_data():
|
||
"""
|
||
获取证书所需数据
|
||
Parameters:
|
||
param: dict 查询条件（包括企业名称/cid/rid）
|
||
Returns:
|
||
result: dict 证书所需数据
|
||
"""
|
||
|
||
def general():
|
||
"""
|
||
获取综合信用评价证书所需数据
|
||
Parameters:
|
||
-
|
||
Returns:
|
||
result: dict 证书所需数据
|
||
"""
|
||
# 企业基本信息
|
||
data_01 = self.db.find_single_data("企业数据", "基本工商信息", {'企业ID': self.cid}, ['工商信息'])
|
||
# 评价结果
|
||
data_02 = self.db.find_single_data("综信评价数据", "评价结果", {'评价ID': self.rid}, ['信用等级'])
|
||
info = dict()
|
||
info['企业名称'] = self.company
|
||
info['统一社会信用代码'] = data_01['工商信息']['纳税人识别号']
|
||
info['企业级别'] = data_02['信用等级']
|
||
info['证书编号'] = self.generate_number()
|
||
info['有效日期'] = self.effective_date()
|
||
return info
|
||
|
||
def esg():
|
||
"""
|
||
获取esg评价证书所需数据
|
||
Parameters:
|
||
-
|
||
Returns:
|
||
result: dict 证书所需数据
|
||
"""
|
||
# 企业基本信息
|
||
data_01 = self.db.find_single_data("企业数据", "基本工商信息", {'企业ID': self.cid}, ['工商信息'])
|
||
# 评价结果
|
||
data_02 = self.db.find_single_data("ESG评价数据", "评价结果", {'评价ID': self.rid}, ['ESG得分', '评价等级'])
|
||
info = dict()
|
||
info['企业名称'] = self.company
|
||
info['统一社会信用代码'] = data_01['工商信息']['纳税人识别号']
|
||
info['ESG得分'] = data_02['ESG得分']
|
||
info['评价等级'] = data_02['评价等级']
|
||
info['证书编号'] = self.generate_esg_number()
|
||
info['有效日期'] = self.effective_date()
|
||
return info
|
||
|
||
if self.type == 'esg':
|
||
res = esg()
|
||
else:
|
||
res = general()
|
||
|
||
return res
|
||
|
||
def certificate():
|
||
"""
|
||
生成证书pdf
|
||
Parameters:
|
||
param: dict 证书所需数据
|
||
Returns:
|
||
res: 处理成功返回报告FileID；处理失败返回False
|
||
"""
|
||
# 生成PDF报告文件，以时间戳命名，临时存储
|
||
certificate_file_name = '{}_{}.pdf'.format(self.company, datetime.now().strftime('%Y%m%d'))
|
||
pdf_report = CertificateGenerator(name=certificate_file_name, text_data=text_data)
|
||
if len(text_data) > 5:
|
||
pdf_report.gen_esg()
|
||
else:
|
||
pdf_report.gen()
|
||
|
||
return certificate_file_name
|
||
|
||
def save_certificate():
|
||
"""
|
||
保存证书pdf
|
||
Parameters:
|
||
'-'
|
||
Returns:
|
||
res: 处理成功返回报告FileID；处理失败返回False
|
||
"""
|
||
# 保存PDF报告文件，删除临时存储的报告
|
||
if len(text_data) > 5:
|
||
file_id = self.db.upsert_certificate_file('文件', 'ESG证书', certifice_pdf)
|
||
else:
|
||
file_id = self.db.upsert_certificate_file('文件', '综信证书', certifice_pdf)
|
||
certificate_path = get_gen_certificate_path(name=certifice_pdf)
|
||
os.remove(certificate_path)
|
||
|
||
# 保存PDF报告成功，返回FileID;保存PDF报告文件失败，返回False
|
||
return file_id if file_id else False
|
||
|
||
text_data = get_certificate_data()
|
||
certifice_pdf = certificate()
|
||
result = save_certificate()
|
||
|
||
if result is not False:
|
||
return {"info": "生成证书成功", "result": {"FileID": result}}, 200
|
||
else:
|
||
return {"info": "生成证书失败"}, 210
|