feat(achievement): 添加判定核心、通知模块并接入判题链路

achievement/checker.py

@@ -0,0 +1,96 @@
+"""成就判定核心，被异步任务和管理命令共用。
+
+判定刻意做成"一次查询取全部候选 + 内存比对"，与 problemset 里逐条查询的
+旧写法相反：一次判题只多 3~4 条 SQL。
+"""
+
+import logging
+
+from django.db import transaction
+from django.db.models import F
+
+from achievement.metrics import META_METRICS, METRIC_REGISTRY, build_ctx
+from achievement.models import Achievement, Operator, UserAchievement, UserStat
+
+logger = logging.getLogger(__name__)
+
+
+def get_or_create_stat(user):
+    stat, _ = UserStat.objects.get_or_create(user=user)
+    return stat
+
+
+def evaluate(user, metrics, only_metrics=None):
+    """返回该用户应解锁但尚未解锁的成就列表。"""
+    unlocked_ids = set(UserAchievement.objects.filter(user=user).values_list("achievement_id", flat=True))
+    qs = Achievement.objects.filter(visible=True).exclude(id__in=unlocked_ids)
+    if only_metrics is not None:
+        qs = qs.filter(metric__in=only_metrics)
+
+    hits = []
+    for achievement in qs:
+        value = metrics.get(achievement.metric)
+        # 指标从未产生有效值时 key 不存在，直接跳过：
+        # 否则 min_ac_code_chars 这类极小值指标会对新用户恒成立
+        if value is None:
+            continue
+        if achievement.operator == Operator.GTE and value >= achievement.threshold:
+            hits.append(achievement)
+        elif achievement.operator == Operator.LTE and value <= achievement.threshold:
+            hits.append(achievement)
+    return hits
+
+
+def unlock(user, achievements, backfilled=False, notified=False):
+    """写入解锁记录并累加 unlock_count，返回实际新建的记录。
+
+    刻意逐条 get_or_create 而不是 bulk_create：unique_user_achievement 约束负责
+    并发竞态，was_created 是"这一条确实是我新建的"的唯一可信判据。用
+    bulk_create(ignore_conflicts=True) 则无法区分新建与已存在，并发判题时会把
+    unlock_count 重复累加（获得率永久偏高），并对同一个奖杯重复推送通知。
+
+    循环次数是"本次新解锁的成就数"，常态为 0，因此常态零查询。
+    """
+    created = []
+    for achievement in achievements:
+        record, was_created = UserAchievement.objects.get_or_create(
+            user=user,
+            achievement=achievement,
+            defaults={"backfilled": backfilled, "notified": notified},
+        )
+        if was_created:
+            Achievement.objects.filter(id=achievement.id).update(unlock_count=F("unlock_count") + 1)
+            created.append(record)
+    return created
+
+
+def run_for_submission(user, submission):
+    """判题后的完整流程：更新指标 → 第一轮判定 → 元指标第二轮判定。"""
+    ctx = build_ctx(user.id, submission)
+    if ctx["skip"]:
+        # 比赛提交不计入成就
+        return []
+
+    with transaction.atomic():
+        stat = UserStat.objects.select_for_update().get_or_create(user=user)[0]
+        for key, m in METRIC_REGISTRY.items():
+            if key in META_METRICS:
+                continue
+            m.on_submission(stat.metrics, submission, ctx)
+        stat.save(update_fields=["metrics", "update_time"])
+
+    first = unlock(user, evaluate(user, stat.metrics))
+    if not first:
+        return []
+
+    # 第二轮：只重算元指标、只判定依赖元指标的成就，不再有第三轮
+    for key in META_METRICS:
+        value = METRIC_REGISTRY[key].recompute(user)
+        if value is None:
+            stat.metrics.pop(key, None)
+        else:
+            stat.metrics[key] = value
+    stat.save(update_fields=["metrics", "update_time"])
+    second = unlock(user, evaluate(user, stat.metrics, only_metrics=META_METRICS))
+
+    return first + second

achievement/notify.py

@@ -0,0 +1,59 @@
+"""解锁通知。
+
+通知走推拉结合，UserAchievement.notified 是唯一的真相来源：
+- 拉（主）：前端在布局层拉 /api/achievements/pending，覆盖全部场景
+- 推（增强）：WebSocket 只负责把"当场那一下"的延迟压到几百毫秒
+
+必须推拉结合的原因：前端 WebSocket 不是常驻连接，useSubmissionWebSocket
+只在问题页且有提交监听时建连，纯推会丢消息。
+"""
+
+import logging
+
+from utils.websocket import push_to_user
+
+logger = logging.getLogger(__name__)
+
+
+def notify_achievements(user_id, records):
+    """records: list[UserAchievement]，已带 select_related('achievement')。"""
+    if not records:
+        return
+    payload = [
+        {
+            "id": r.achievement_id,
+            "name": r.achievement.name,
+            "description": r.achievement.description,
+            "icon": r.achievement.icon,
+            "rarity": r.achievement.rarity,
+            "kind": "achievement",
+        }
+        for r in records
+    ]
+    _push(user_id, payload)
+
+
+def notify_badges(user_id, badges):
+    """badges: list[ProblemSetBadge]。题单奖章复用同一个弹窗组件。"""
+    if not badges:
+        return
+    payload = [
+        {
+            "id": b.id,
+            "name": b.name,
+            "description": b.description,
+            "icon": b.icon,
+            "rarity": "bronze",
+            "kind": "badge",
+        }
+        for b in badges
+    ]
+    _push(user_id, payload)
+
+
+def _push(user_id, payload):
+    # 推送失败不影响已入库的解锁记录，前端下次拉 pending 时仍会补弹
+    try:
+        push_to_user(user_id, "achievement_unlocked", {"achievements": payload})
+    except Exception as e:
+        logger.error(f"Failed to push achievement notification: user_id={user_id}, error={e}")

achievement/tasks.py

@@ -0,0 +1,26 @@
+import logging
+
+import dramatiq
+
+from account.models import User
+from achievement import checker
+from achievement.notify import notify_achievements
+from submission.models import Submission
+from utils.shortcuts import DRAMATIQ_WORKER_ARGS
+
+logger = logging.getLogger(__name__)
+
+
+@dramatiq.actor(**DRAMATIQ_WORKER_ARGS())
+def check_achievements(user_id, submission_id):
+    """判题完成后的成就判定。
+
+    所有异常在此吞掉：成就算错绝不能影响判题结果，这也是选异步的意义。
+    """
+    try:
+        user = User.objects.get(id=user_id)
+        submission = Submission.objects.get(id=submission_id)
+        records = checker.run_for_submission(user, submission)
+        notify_achievements(user_id, records)
+    except Exception as e:
+        logger.exception(f"check_achievements failed: user_id={user_id}, submission_id={submission_id}, error={e}")

judge/tasks.py

@@ -1,6 +1,7 @@
 import dramatiq
 
 from account.models import User
+from achievement.tasks import check_achievements
 from judge.dispatcher import JudgeDispatcher
 from judge.sql_dispatcher import SQLJudgeDispatcher
 from submission.models import Submission
@@ -17,3 +18,6 @@ def judge_task(submission_id, problem_id):
         SQLJudgeDispatcher(submission_id, problem_id).judge()
     else:
         JudgeDispatcher(submission_id, problem_id).judge()
+
+    # 判题结束后异步判定成就；投递失败不影响判题结果
+    check_achievements.send(submission.user_id, submission_id)