113 lines
3.0 KiB
Python
113 lines
3.0 KiB
Python
# -*- 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')
|