# -*- coding = utf-8 -*- # @Time : 2025/2/13 下午7:40 # @File : helper.py # @Software : PyCharm # @Author : xingc # @Desc : import base64 import functools import random import string from Crypto.Cipher import AES from Crypto.Util.Padding import unpad from datetime import datetime from typing import Callable from config import settings def async_retry(max_retries: int = 3, exceptions: tuple = (Exception,)): """ 异步重试装饰器。 :param max_retries: 最大重试次数。 :param exceptions: 触发重试的异常类型。 """ def decorator(func: Callable): @functools.wraps(func) async def wrapper(*args, **kwargs): retries = 0 while retries < max_retries: try: return await func(*args, **kwargs) except exceptions as e: retries += 1 if retries >= max_retries: raise # 如果达到最大重试次数,抛出异常 return wrapper return decorator def generate_keyname(name: str, use_today: bool = False) -> str: """生成统一键名前缀""" if name.startswith(settings.PREFIX) is False: names = [settings.PREFIX, name] if use_today: today = datetime.now().strftime('%Y%m%d') names.insert(1, today) return ':'.join(names) else: return name def check_picture_for_empty(picture: dict, or_: bool = False) -> bool: """检查图片是否为空""" results = [ picture.get('front_picture') is None, picture.get('reverse_picture') is None ] return any(results) if or_ else all(results) def generate_order_number(): random_part = ''.join(random.choices(string.digits, k=18)) order_number = f"{datetime.now().strftime('%Y%m%d%H%M%S')}{random_part}" return order_number def format_date(date: datetime | None) -> str | None: if date is None: return date else: return date.strftime('%Y-%m-%d %H:%M:%S') def decrypt_aes_cbc(ciphertext_b64, key_str, iv_str): """ 等效于 Node.js CryptoJS 的解密实现 参数: ciphertext_b64: Base64编码的密文 key_str: 密钥字符串(UTF-8编码) iv_str: 初始化向量字符串(UTF-8编码) 返回: 解密后的明文(UTF-8字符串) """ # 将密钥和IV从UTF-8字符串转换为bytes key = key_str.encode('utf-8') iv = iv_str.encode('utf-8') # 确保密钥和IV长度正确 if len(key) not in [16, 24, 32]: raise ValueError("密钥长度必须是16(AES-128), 24(AES-192)或32(AES-256)字节") if len(iv) != 16: raise ValueError("IV长度必须是16字节") # 解码Base64密文 ciphertext = base64.b64decode(ciphertext_b64) # 创建解密器 cipher = AES.new(key, AES.MODE_CBC, iv) # 解密并去除PKCS7填充 decrypted = unpad(cipher.decrypt(ciphertext), AES.block_size) return decrypted.decode('utf-8')