"""P2-01 好感度系统服务层 — 唯一写入入口 所有好感度变化（设备事件、手机事件、衰减任务、管理员调整）**必须**经由 `AffinityService.apply()` 处理，确保单一冷却 / 日上限 / 钳位 / 日志 / 等级 更新 / 奖励派发 / WS 推送的语义闭环。 设计文档：§4.3 触发计算流程 """ from __future__ import annotations import logging import random from dataclasses import dataclass, field from typing import Any, Dict, Optional, Tuple from django.db import transaction from django.db.models import F from device_interaction.models import UserDevice from userapp.models import ( AffinityLog, AffinityRule, AffinitySetting, UserAffinityDailyCounter, ) from . import counters as redis_counters from . import levels as level_utils from . import rewards as reward_utils from . import ws logger = logging.getLogger(__name__) # ---------- 返回类型 ---------- class ApplyOutcome: """ApplyResult.outcome 的常量集合""" APPLIED = 'applied' # 成功写入 NOOP_NO_RULE = 'noop_no_rule' # 规则不存在或已禁用 / 软删 NOOP_COOLDOWN = 'noop_cooldown' # 冷却中 NOOP_RULE_DAILY_CAP = 'noop_rule_daily_cap' # 本规则今日已达上限 NOOP_GLOBAL_DAILY_CAP = 'noop_global_daily_cap' # 全局今日已达上限 NOOP_EVENT_DUP = 'noop_event_duplicate' # 同 event_id 60s 内已处理 NOOP_VALUE_BOUNDARY = 'noop_value_boundary' # 当前已在边界（0 还想扣 / 上限还想加） ERROR = 'error' # 异常 @dataclass class ApplyResult: outcome: str change_value: int = 0 before_value: int = 0 after_value: int = 0 rule_key: str = '' old_level: int = 0 new_level: int = 0 log_id: Optional[int] = None rewards_granted: list = field(default_factory=list) rewards_failed: list = field(default_factory=list) error: Optional[str] = None @property def is_applied(self) -> bool: return self.outcome == ApplyOutcome.APPLIED # ---------- 主入口 ---------- class AffinityService: """好感度系统服务层""" @classmethod def apply( cls, *, user_id: int, device_id: int, rule_key: str, source: str, event_id: str = '', metadata: Optional[Dict[str, Any]] = None, operator_admin_id: Optional[int] = None, reason: str = '', ) -> ApplyResult: """好感度变化的唯一写入入口。 参数： user_id: ParadiseUser.id（用于 AffinityLog.user 关联及 WS group 路由） device_id: UserDevice.id（注意是绑定 ID，不是 Device.id） rule_key: AffinityRule.rule_key（如 'chat' / 'gift' / 'decay' / 'admin'） source: AffinityLog.SOURCE_CHOICES 之一 event_id: 客户端事件 UUID，用于 60s 内幂等去重 metadata: 扩展上下文 JSON operator_admin_id / reason: 管理员调整时填写 返回 ApplyResult，调用方根据 outcome 决定后续动作。 """ # 0) event_id 幂等防重 if event_id and redis_counters.event_already_processed(event_id): return ApplyResult( outcome=ApplyOutcome.NOOP_EVENT_DUP, rule_key=rule_key, ) # 1) 取规则 rule = ( AffinityRule.objects .filter(rule_key=rule_key, is_enabled=True, is_deleted=False) .first() ) if rule is None: return ApplyResult(outcome=ApplyOutcome.NOOP_NO_RULE, rule_key=rule_key) # 2) 冷却检查 if redis_counters.is_in_cooldown(device_id, rule_key): return ApplyResult(outcome=ApplyOutcome.NOOP_COOLDOWN, rule_key=rule_key) # 3) 取设备（要求绑定有效） try: user_device = UserDevice.active.get(id=device_id, user_id=user_id) except UserDevice.DoesNotExist: logger.warning( '[affinity.service] device %s 不属于 user %s 或已解绑，跳过', device_id, user_id, ) return ApplyResult( outcome=ApplyOutcome.ERROR, rule_key=rule_key, error='UserDevice not found or unbound', ) # 4) 计算本次变化值（[min, max] 闭区间随机；P1 CHECK 保证 min <= max） raw_change = random.randint(rule.min_change, rule.max_change) # single_cap 钳位（保护性 — P1 CHECK 已保证 single_cap > 0） if raw_change > rule.single_cap: raw_change = rule.single_cap elif raw_change < -rule.single_cap: raw_change = -rule.single_cap if raw_change == 0: return ApplyResult( outcome=ApplyOutcome.NOOP_VALUE_BOUNDARY, rule_key=rule_key, change_value=0, ) # 5) 本规则日上限检查（用绝对值累加） abs_change = abs(raw_change) rule_today = redis_counters.get_rule_daily(device_id, rule_key) if rule_today + abs_change > rule.daily_cap: # 允许部分通过：剩余空间 > 0 时按剩余空间钳位 remain = rule.daily_cap - rule_today if remain <= 0: return ApplyResult( outcome=ApplyOutcome.NOOP_RULE_DAILY_CAP, rule_key=rule_key, ) # 按正负方向取剩余空间 raw_change = remain if raw_change > 0 else -remain abs_change = remain # 6) 全局日上限检查（仅正向汇总，衰减不占用） setting = AffinitySetting.get_solo() if raw_change > 0: global_today = redis_counters.get_global_daily(device_id) if global_today + abs_change > setting.global_daily_cap: remain = setting.global_daily_cap - global_today if remain <= 0: return ApplyResult( outcome=ApplyOutcome.NOOP_GLOBAL_DAILY_CAP, rule_key=rule_key, ) raw_change = remain abs_change = remain # 7) 原子更新 UserDevice.favorability + 钳位 try: with transaction.atomic(): # 锁住该 UserDevice 行，避免并发写竞争 ud_locked = UserDevice.objects.select_for_update().get(id=user_device.id) before_value = ud_locked.favorability after_value = before_value + raw_change # 钳位 [0, max_affinity]（管理员调整也走此分支，因此天然遵守 CR-001 决策 6） after_value = max(0, min(setting.max_affinity, after_value)) actual_change = after_value - before_value if actual_change == 0: # 边界场景：当前值已经在边界，本次实际无变化 return ApplyResult( outcome=ApplyOutcome.NOOP_VALUE_BOUNDARY, rule_key=rule_key, before_value=before_value, after_value=after_value, ) ud_locked.favorability = after_value # 同步刷新 last_active_at（衰减判断依赖） from django.utils import timezone as djtz ud_locked.last_active_at = djtz.now() ud_locked.save(update_fields=['favorability', 'last_active_at']) # 8) 写 AffinityLog（含 cross-app FK 与冗余 rule_key） log = AffinityLog.objects.create( user_id=user_id, device=ud_locked, rule=rule, rule_key=rule_key, change_value=actual_change, before_value=before_value, after_value=after_value, source=source, event_id=event_id or None, # WR-004：空 -> NULL operator_admin_id=operator_admin_id, reason=reason, metadata=metadata or {}, ) # 9) 更新数据库兜底计数器（Redis 是热路径，DB 是审计兜底） today_date = redis_counters.local_today_date(setting.timezone) counter, _ = UserAffinityDailyCounter.objects.select_for_update().get_or_create( device=ud_locked, rule=rule, date=today_date, defaults={'accumulated_change': 0, 'trigger_count': 0}, ) counter.accumulated_change = F('accumulated_change') + actual_change counter.trigger_count = F('trigger_count') + 1 counter.save(update_fields=['accumulated_change', 'trigger_count']) # 10) 更新等级缓存（事务内做，避免半成品状态） old_level, new_level, _matched = level_utils.update_device_level( ud_locked, save=True, ) except Exception as exc: logger.exception( '[affinity.service] apply 失败 user=%s device=%s rule=%s: %s', user_id, device_id, rule_key, exc, ) return ApplyResult( outcome=ApplyOutcome.ERROR, rule_key=rule_key, error=str(exc), ) # 事务已提交，下面是非原子的副作用：Redis 计数 / 推送 / 奖励派发 # 11) Redis 计数器累加（仅在数据真正写入后再扣冷却 / 日上限配额） redis_counters.set_cooldown(device_id, rule_key, rule.cooldown_seconds) redis_counters.incr_rule_daily(device_id, rule_key, abs(actual_change)) if actual_change > 0: redis_counters.incr_global_daily(device_id, abs(actual_change)) if event_id: redis_counters.mark_event_processed(event_id) # 12) 跨级奖励发放（A3 方案 B — 每级独立事务） grant_result = reward_utils.RewardGrantResult() if new_level > old_level: grant_result = reward_utils.grant_levels(ud_locked, old_level, new_level) # 13) WS 推送 if setting.enable_notify: ws.push_affinity_update( user_id=user_id, device_id=device_id, change=actual_change, before=before_value, after=after_value, rule_key=rule_key, source=source, ) if new_level > old_level: ws.push_level_up( user_id=user_id, device_id=device_id, old_level=old_level, new_level=new_level, rewards=grant_result.granted, ) elif new_level < old_level: ws.push_level_down( user_id=user_id, device_id=device_id, old_level=old_level, new_level=new_level, ) return ApplyResult( outcome=ApplyOutcome.APPLIED, change_value=actual_change, before_value=before_value, after_value=after_value, rule_key=rule_key, old_level=old_level, new_level=new_level, log_id=log.id, rewards_granted=grant_result.granted, rewards_failed=grant_result.failed, ) # ---------- 管理员调整专用入口（绕过 rule，直接给数值）---------- @classmethod def admin_adjust( cls, *, user_id: int, device_id: int, delta: int, operator_admin_id: int, reason: str = '', batch: bool = False, ) -> ApplyResult: """管理员手动调整。 - 不查 AffinityRule（rule=NULL，rule_key='admin'） - 不查冷却 / 日上限 - 依然钳位 [0, max_affinity]（决策 6：管理员不能突破上限） - source=admin_adjust_single 或 admin_adjust_batch - 仍写 AffinityLog、更新等级缓存、发奖励、推 WS """ from userapp.models import AffinitySetting source = 'admin_adjust_batch' if batch else 'admin_adjust_single' try: user_device = UserDevice.active.get(id=device_id, user_id=user_id) except UserDevice.DoesNotExist: return ApplyResult( outcome=ApplyOutcome.ERROR, rule_key='admin', error='UserDevice not found or unbound', ) if delta == 0: return ApplyResult( outcome=ApplyOutcome.NOOP_VALUE_BOUNDARY, rule_key='admin', ) setting = AffinitySetting.get_solo() try: with transaction.atomic(): ud_locked = UserDevice.objects.select_for_update().get(id=user_device.id) before_value = ud_locked.favorability after_value = max(0, min(setting.max_affinity, before_value + delta)) actual_change = after_value - before_value if actual_change == 0: return ApplyResult( outcome=ApplyOutcome.NOOP_VALUE_BOUNDARY, rule_key='admin', before_value=before_value, after_value=after_value, ) ud_locked.favorability = after_value from django.utils import timezone as djtz ud_locked.last_active_at = djtz.now() ud_locked.save(update_fields=['favorability', 'last_active_at']) log = AffinityLog.objects.create( user_id=user_id, device=ud_locked, rule=None, rule_key='admin', change_value=actual_change, before_value=before_value, after_value=after_value, source=source, operator_admin_id=operator_admin_id, reason=reason, metadata={'requested_delta': delta}, ) old_level, new_level, _matched = level_utils.update_device_level( ud_locked, save=True, ) except Exception as exc: logger.exception( '[affinity.service.admin_adjust] 失败 user=%s device=%s delta=%s', user_id, device_id, delta, ) return ApplyResult( outcome=ApplyOutcome.ERROR, rule_key='admin', error=str(exc), ) # 副作用 grant_result = reward_utils.RewardGrantResult() if new_level > old_level: grant_result = reward_utils.grant_levels(ud_locked, old_level, new_level) if setting.enable_notify: ws.push_affinity_update( user_id=user_id, device_id=device_id, change=actual_change, before=before_value, after=after_value, rule_key='admin', source=source, ) if new_level > old_level: ws.push_level_up( user_id=user_id, device_id=device_id, old_level=old_level, new_level=new_level, rewards=grant_result.granted, ) elif new_level < old_level: ws.push_level_down( user_id=user_id, device_id=device_id, old_level=old_level, new_level=new_level, ) return ApplyResult( outcome=ApplyOutcome.APPLIED, change_value=actual_change, before_value=before_value, after_value=after_value, rule_key='admin', old_level=old_level, new_level=new_level, log_id=log.id, rewards_granted=grant_result.granted, rewards_failed=grant_result.failed, )