# yonghu/ranking_tasks.py """ 排行榜数据转移任务 - 生产环境专用 在清空前将数据转移到排行榜表 """ from celery import shared_task from django.db import transaction from django.utils import timezone from datetime import datetime, timedelta import logging from decimal import Decimal from users.models import ( UserMain, UserDashou, UserShangjia, UserGuanshi, UserBoss, RankingRecord ) logger = logging.getLogger(__name__) def get_user_nicheng(user, shenfen): """ 获取用户昵称 规则：打手用打手nicheng，其他用老板nickname """ try: if shenfen == 2: # 打手 dashou_profile = getattr(user, 'dashou_profile', None) if dashou_profile and dashou_profile.nicheng: return dashou_profile.nicheng else: # 管事或商家 boss_profile = getattr(user, 'boss_profile', None) if boss_profile and boss_profile.nickname: return boss_profile.nickname # 默认返回用户ID return f"用户{user.yonghuid}" except Exception: return f"用户{user.yonghuid}" @shared_task def zhuanyi_ribang(): """ 转移日榜数据 - 每天23:55执行 在清零前将今日数据转移到排行榜表 """ try: # 统计日期（今天） today = timezone.now().date() logger.info(f"开始转移日榜数据，日期: {today}") records_to_create = [] # 1. 转移打手日榜（今日收益 > 0） dashou_list = UserDashou.objects.filter( jinrishouyi__gt=0 # 只转移收益大于0的 ).select_related('user').order_by('-jinrishouyi') for i, dashou in enumerate(dashou_list, start=1): if i > 100: # 最多100条 break user = dashou.user record = RankingRecord( zhouqi=1, # 日榜 tongji_date=today, shenfen=2, # 打手 yonghuid=user.yonghuid, paiming=i, zhizhi=dashou.jinrishouyi, nicheng=get_user_nicheng(user, 2), avatar=user.avatar or '' ) records_to_create.append(record) # 2. 转移商家日榜（今日流水 > 0） shangjia_list = UserShangjia.objects.filter( jinriliushui__gt=0 # 只转移流水大于0的 ).select_related('user').order_by('-jinriliushui') for i, shangjia in enumerate(shangjia_list, start=1): if i > 100: # 最多100条 break user = shangjia.user record = RankingRecord( zhouqi=1, tongji_date=today, shenfen=4, # 商家 yonghuid=user.yonghuid, paiming=i, zhizhi=shangjia.jinriliushui, nicheng=get_user_nicheng(user, 4), avatar=user.avatar or '' ) records_to_create.append(record) # 3. 转移管事日榜（今日充值人数 > 0） guanshi_list = UserGuanshi.objects.filter( jinrichongzhi__gt=0 # 只转移充值人数大于0的 ).select_related('user').order_by('-jinrichongzhi') for i, guanshi in enumerate(guanshi_list, start=1): if i > 100: # 最多100条 break user = guanshi.user record = RankingRecord( zhouqi=1, tongji_date=today, shenfen=3, # 管事 yonghuid=user.yonghuid, paiming=i, zhizhi=Decimal(guanshi.jinrichongzhi), nicheng=get_user_nicheng(user, 3), avatar=user.avatar or '' ) records_to_create.append(record) # 批量保存 if records_to_create: with transaction.atomic(): RankingRecord.objects.bulk_create(records_to_create) logger.info(f"日榜数据转移完成，保存{len(records_to_create)}条记录") result = { 'success': True, 'date': today.strftime('%Y-%m-%d'), 'total_count': len(records_to_create), } return result except Exception as e: logger.error(f"日榜数据转移失败: {str(e)}", exc_info=True) raise @shared_task def zhuanyi_yuebang(): """ 转移月榜数据 - 每月最后一天23:50执行 在清零前将本月数据转移到排行榜表 """ try: # 🔥 获取当前时间 now = timezone.now() today = now.date() # 🔥 判断今天是否是当月的最后一天 # 方法：计算明天的日期，如果明天的月份和今天不同，那么今天就是最后一天 tomorrow = today + timedelta(days=1) if tomorrow.month == today.month: # 今天不是最后一天，直接返回 logger.info(f"今天不是当月最后一天（{today}），跳过月榜转移") return { 'success': True, 'message': f'今天不是当月最后一天，跳过执行', 'is_last_day': False, 'today': today.strftime('%Y-%m-%d'), } logger.info(f"今天是当月最后一天（{today}），开始转移月榜数据") # 🔥 统计日期为当月第一天（如2024-01-01） tongji_date = datetime(today.year, today.month, 1).date() records_to_create = [] # 1. 转移打手月榜（今月收益 > 0） dashou_list = UserDashou.objects.filter( jinyueshouyi__gt=0 # 只转移收益大于0的 ).select_related('user').order_by('-jinyueshouyi') for i, dashou in enumerate(dashou_list, start=1): if i > 100: # 最多100条 break user = dashou.user record = RankingRecord( zhouqi=2, # 月榜 tongji_date=tongji_date, shenfen=2, # 打手 yonghuid=user.yonghuid, paiming=i, zhizhi=dashou.jinyueshouyi, nicheng=get_user_nicheng(user, 2), avatar=user.avatar or '' ) records_to_create.append(record) # 2. 转移商家月榜（今月流水 > 0） shangjia_list = UserShangjia.objects.filter( jinyueliushui__gt=0 # 只转移流水大于0的 ).select_related('user').order_by('-jinyueliushui') for i, shangjia in enumerate(shangjia_list, start=1): if i > 100: # 最多100条 break user = shangjia.user record = RankingRecord( zhouqi=2, tongji_date=tongji_date, shenfen=4, # 商家 yonghuid=user.yonghuid, paiming=i, zhizhi=shangjia.jinyueliushui, nicheng=get_user_nicheng(user, 4), avatar=user.avatar or '' ) records_to_create.append(record) # 3. 转移管事月榜（今月充值人数 > 0） guanshi_list = UserGuanshi.objects.filter( jinyuechongzhi__gt=0 # 只转移充值人数大于0的 ).select_related('user').order_by('-jinyuechongzhi') for i, guanshi in enumerate(guanshi_list, start=1): if i > 100: # 最多100条 break user = guanshi.user record = RankingRecord( zhouqi=2, tongji_date=tongji_date, shenfen=3, # 管事 yonghuid=user.yonghuid, paiming=i, zhizhi=Decimal(guanshi.jinyuechongzhi), nicheng=get_user_nicheng(user, 3), avatar=user.avatar or '' ) records_to_create.append(record) # 批量保存 if records_to_create: with transaction.atomic(): RankingRecord.objects.bulk_create(records_to_create) logger.info(f"月榜数据转移完成，保存{len(records_to_create)}条记录") result = { 'success': True, 'month': tongji_date.strftime('%Y-%m'), 'total_count': len(records_to_create), 'is_last_day': True, 'today': today.strftime('%Y-%m-%d'), } return result except Exception as e: logger.error(f"月榜数据转移失败: {str(e)}", exc_info=True) raise @shared_task def qingli_jiulishuju(): """ 清理旧历史数据 - 每月1号凌晨1点执行 清理3个月前的排行榜数据，保留最近3个月的数据 """ try: # 计算3个月前的日期 three_months_ago = timezone.now().date() - timedelta(days=90) # 删除旧数据 deleted_count, _ = RankingRecord.objects.filter( tongji_date__lt=three_months_ago ).delete() logger.info(f"清理旧历史数据完成，删除{deleted_count}条记录") return { 'success': True, 'deleted_count': deleted_count, 'before_date': three_months_ago.strftime('%Y-%m-%d') } except Exception as e: logger.error(f"清理旧历史数据失败: {str(e)}") raise