60 lines
1.6 KiB
Python
60 lines
1.6 KiB
Python
# -*- coding = utf-8 -*-
|
|
# @Time : 2025/2/13 下午7:40
|
|
# @File : helper.py
|
|
# @Software : PyCharm
|
|
# @Author : xingc
|
|
# @Desc :
|
|
import asyncio
|
|
import functools
|
|
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)
|