python-fastapi-mm-zhcs-yj-api/common/enc/chemical_company_data.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

from . import mpfun
from models import *
from sqlalchemy.orm import Session
from database import get_local_db

# 危化企业信息表

# 加密和HMAC签名
def sign_row(db: Session, row: ChemicalCompany) -> None:
    if row.sign != '':
        return
    
    province = mpfun.base64_data(row.province) # 省
    city = mpfun.base64_data(row.city) # 地市
    area = mpfun.base64_data(row.area) # 区县
    company_name = mpfun.base64_data(row.company_name) # 企业名称
    company_code = mpfun.base64_data(row.company_code) # 企业编码
    company_type = mpfun.base64_data(row.company_type) # 企业类型
    credit_code = mpfun.base64_data(row.credit_code) # 统一社会信用代码
    industry_classification = mpfun.base64_data(row.industry_classification) # 行业分类
    
    emergency_contact_number = mpfun.enc_data(row.emergency_contact_number) # 安全值班电话
    emergency_response_hotline = mpfun.enc_data(row.emergency_response_hotline) # 应急咨询服务电话
    qyagjgfzr_phone = mpfun.enc_data(row.qyagjgfzr_phone) # 企业安管机构负责人手机号
    qyfgaqfzr_phone = mpfun.enc_data(row.qyfgaqfzr_phone) # 企业分管安全负责人手机号

    xkzbh = mpfun.base64_data(row.xkzbh) # 许可证编号
    issuance_date = mpfun.base64_data(row.issuance_date) # 发证日期
    company_registration_number = mpfun.base64_data(row.company_registration_number) # 证书有效期
    factory_address = mpfun.base64_data(row.factory_address) # 厂区地址
    formatted_address = mpfun.base64_data(row.formatted_address) # 地址
    safe_level = mpfun.base64_data(row.safe_level) # 安全标准化等级

    sign_data = ",".join([province, city, area, company_name, company_code, company_type, credit_code, industry_classification, emergency_contact_number, emergency_response_hotline, 
                          qyagjgfzr_phone, qyfgaqfzr_phone, xkzbh, issuance_date, company_registration_number, factory_address, formatted_address, safe_level])
    sign_hmac = mpfun.sign_data(sign_data)
    # print('sign_tbl_user sign_data:', sign_data)
    # print('sign_tbl_user sign_hmac:', sign_hmac)

    row.emergency_contact_number = emergency_contact_number
    row.emergency_response_hotline = emergency_response_hotline
    row.qyagjgfzr_phone = qyagjgfzr_phone
    row.qyfgaqfzr_phone = qyfgaqfzr_phone
    
    row.sign = sign_hmac
    
    db.commit()

# 比较字段合并字符串是否和MAC值匹配上,调用密码服务器[验证HMAC]接口
def sign_valid_row(row: ChemicalCompany) -> bool:
    if row.sign == '':
        return True

    # 关键字段合并字符串
    sign_data = get_sign_str(row)
    # print('sys_user sign_data:', sign_data)
    
    # 原HMACSM3数值
    sign_hmac = row.sign
    # print('sign_hmac:', sign_hmac)

    return mpfun.hmac_verify(sign_data, sign_hmac)

# 生成待签名的字符串
def get_sign_str(row: ChemicalCompany) -> str:
    province = mpfun.base64_data(row.province) # 省
    city = mpfun.base64_data(row.city) # 地市
    area = mpfun.base64_data(row.area) # 区县
    company_name = mpfun.base64_data(row.company_name) # 企业名称
    company_code = mpfun.base64_data(row.company_code) # 企业编码
    company_type = mpfun.base64_data(row.company_type) # 企业类型
    credit_code = mpfun.base64_data(row.credit_code) # 统一社会信用代码
    industry_classification = mpfun.base64_data(row.industry_classification) # 行业分类
    
    emergency_contact_number = (row.emergency_contact_number) # 安全值班电话
    emergency_response_hotline = (row.emergency_response_hotline) # 应急咨询服务电话
    emergency_contacqyagjgfzr_phonet_number = (row.qyagjgfzr_phone) # 企业安管机构负责人手机号
    qyfgaqfzr_phone = (row.qyfgaqfzr_phone) # 企业分管安全负责人手机号

    xkzbh = mpfun.base64_data(row.xkzbh) # 许可证编号
    issuance_date = mpfun.base64_data(row.issuance_date) # 发证日期
    company_registration_number = mpfun.base64_data(row.company_registration_number) # 证书有效期
    factory_address = mpfun.base64_data(row.factory_address) # 厂区地址
    formatted_address = mpfun.base64_data(row.formatted_address) # 地址
    safe_level = mpfun.base64_data(row.safe_level) # 安全标准化等级

    # 关键字段合并字符串
    sign_data = ",".join([province, city, area, company_name, company_code, company_type, credit_code, industry_classification, emergency_contact_number, emergency_response_hotline, 
                          emergency_contacqyagjgfzr_phonet_number, qyfgaqfzr_phone, xkzbh, issuance_date, company_registration_number, factory_address, formatted_address, safe_level])
    return sign_data

# 生成HAMC签名值
def get_sign_hmac(row: ChemicalCompany) -> str:
    sign_data = get_sign_str(row)
    return mpfun.sign_data(sign_data) 

# 对所有数据进行签名
def sign_table():
    print('sign_chemical_company table =====>>>')
    with get_local_db() as db:
        rows = db.query(ChemicalCompany).filter(ChemicalCompany.sign == '').all()
        for row in rows:
            sign_row(db, row)