114 lines
3.7 KiB
Python
114 lines
3.7 KiB
Python
"""
|
||
阿龙电竞 - Celery工具函数
|
||
提供通用的Celery辅助函数，确保数据安全和操作稳定
|
||
【修复版】已解决装饰器导致Celery任务无法识别的问题
|
||
"""
|
||
|
||
import logging
|
||
from django.db import transaction
|
||
from datetime import datetime
|
||
from decimal import Decimal, InvalidOperation
|
||
import functools # 【核心修复】必须导入此模块
|
||
|
||
# 配置日志
|
||
logger = logging.getLogger(__name__)
|
||
|
||
def safe_decimal_operation(value, default=Decimal('0.00')):
|
||
"""
|
||
安全处理Decimal操作，防止类型转换错误
|
||
:param value: 输入值
|
||
:param default: 默认值
|
||
:return: Decimal对象
|
||
"""
|
||
try:
|
||
if value is None:
|
||
return default
|
||
|
||
if isinstance(value, Decimal):
|
||
return value
|
||
|
||
if isinstance(value, (int, float)):
|
||
return Decimal(str(value))
|
||
|
||
if isinstance(value, str):
|
||
# 移除可能的空白字符和特殊字符
|
||
value = value.strip()
|
||
if not value:
|
||
return default
|
||
return Decimal(value)
|
||
|
||
# 其他类型尝试转换
|
||
return Decimal(str(value))
|
||
except (InvalidOperation, TypeError, ValueError) as e:
|
||
logger.warning(f"Decimal转换失败: {e}, 输入值: {value}, 类型: {type(value)}, 使用默认值: {default}")
|
||
return default
|
||
except Exception as e:
|
||
logger.error(f"Decimal转换发生未知错误: {e}, 输入值: {value}, 使用默认值: {default}")
|
||
return default
|
||
|
||
def log_task_execution(task_name, success=True, details=""):
|
||
"""
|
||
记录任务执行日志
|
||
:param task_name: 任务名称
|
||
:param success: 是否成功
|
||
:param details: 详细信息
|
||
"""
|
||
try:
|
||
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
||
status = "成功" if success else "失败"
|
||
|
||
log_message = f"[{timestamp}] {task_name} - {status}"
|
||
if details:
|
||
log_message += f" | {details}"
|
||
|
||
if success:
|
||
logger.info(log_message)
|
||
else:
|
||
logger.error(log_message)
|
||
|
||
return log_message
|
||
except Exception as e:
|
||
# 即使日志记录失败也不影响主流程
|
||
print(f"日志记录失败: {e}")
|
||
return f"日志记录失败: {e}"
|
||
|
||
def rollback_on_failure(operation_name):
|
||
"""
|
||
事务回滚装饰器（修复版）
|
||
:param operation_name: 操作名称
|
||
"""
|
||
def decorator(func):
|
||
@functools.wraps(func) # 【核心修复】保留原函数元数据，使Celery能正确识别任务
|
||
def wrapper(*args, **kwargs):
|
||
try:
|
||
with transaction.atomic():
|
||
return func(*args, **kwargs)
|
||
except Exception as e:
|
||
logger.error(f"{operation_name}事务执行失败: {str(e)}")
|
||
raise
|
||
return wrapper
|
||
return decorator
|
||
|
||
def retry_on_failure(max_retries=3, delay=60):
|
||
"""
|
||
重试装饰器（修复版，可用于Celery任务）
|
||
:param max_retries: 最大重试次数
|
||
:param delay: 重试延迟（秒）
|
||
"""
|
||
def decorator(func):
|
||
@functools.wraps(func) # 【核心修复】保留原函数元数据
|
||
def wrapper(*args, **kwargs):
|
||
retries = 0
|
||
while retries < max_retries:
|
||
try:
|
||
return func(*args, **kwargs)
|
||
except Exception as e:
|
||
retries += 1
|
||
if retries == max_retries:
|
||
logger.error(f"任务{func.__name__}重试{max_retries}次后失败: {str(e)}")
|
||
raise
|
||
logger.warning(f"任务{func.__name__}第{retries}次失败，{delay}秒后重试: {str(e)}")
|
||
import time
|
||
time.sleep(delay)
|
||
return wrapper
|
||
return decorator |