tfse-model-api-v0.2/Certificate/CertificateImpl.py

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