fix(affinity-P1): CR-002 + WR-001 加 Affinity 模型 DB CHECK 约束 + 单例硬约束

AffinityRule / AffinityLevel / AffinitySetting 三表共 13 条 CheckConstraint
覆盖 min ≤ max / 各类 cap > 0 / cooldown ≥ 0 / companion_time 配套字段必填 /
decay 区间合法 / initial ≤ max 等不变量。

AffinitySetting 加 pk=1 单例硬约束（CR-002 + WR-001 联动）+ save() 强制 pk=1，
形成事实单例防御并发首次插入重复行。

模型 clean() 提供 Python 级兜底（给 DRF / admin 友好错误信息）；
AffinityLevel.save() 自动 full_clean 触发跨行区间不重叠校验（为 WR-009 铺路）。

详见 docs/REVIEW-affinity-P1.md CR-002 / WR-001。

qy_lty/docs/修改记录.md

@@ -23,6 +23,26 @@
 
 <!-- 新的修改记录添加在此处下方，最新的在最前面 -->
 
+### [2026-05-13] 好感度系统 P1 审查修复 B — Affinity 模型 DB CHECK 约束 + WR-001 单例硬约束（CR-002 + WR-001）
+
+配套审查报告：[docs/REVIEW-affinity-P1.md](REVIEW-affinity-P1.md)（CR-002 + WR-001）
+配套修复报告：[docs/REVIEW-affinity-P1-FIX-REPORT.md](REVIEW-affinity-P1-FIX-REPORT.md)
+
+- **文件路径**:
+  - `userapp/models.py`（修改 — `AffinityRule.Meta.constraints` 加 5 条 CheckConstraint + `clean()`；`AffinityLevel.Meta.constraints` 加 2 条 CheckConstraint + `clean()` + `save()` 自动 full_clean；`AffinitySetting.Meta.constraints` 加 6 条 CheckConstraint（含 pk=1 单例硬约束）+ `clean()` + `save()` 强制 pk=1；imports 段补 CheckConstraint / F / ValidationError）
+  - `userapp/migrations/0007_add_affinity_check_constraints.py`（**新建** — 由 makemigrations 自动生成，13 条 AddConstraint）
+- **修改类型**: 新增 + 修复Bug
+- **修改内容**:
+  - **CR-002（Critical）**：
+    - AffinityRule：`min_change ≤ max_change` / `cooldown_seconds ≥ 0` / `single_cap > 0` / `daily_cap > 0` / 陪伴时长规则必须设置 `min_continuous_minutes > 0 ∧ max_count_per_day > 0`
+    - AffinityLevel：`min_affinity ≤ max_affinity` / `reward_currency ≥ 0`
+    - AffinitySetting：`decay_min_decay ≤ decay_max_decay ≤ decay_cap` / `initial_affinity ≤ max_affinity` / `decay_min_floor ≤ max_affinity` / `daily_cap > 0`
+  - **WR-001（Warning）**：AffinitySetting 加 `pk=1` 单例硬约束 + save() 强制 pk=1，配合形成事实单例（CHECK 约束跨行不可，但能阻止任何非 1 主键的写入）
+  - 所有模型 `clean()` 提供 Python 级兜底，给 DRF / admin 友好错误信息（DB 级 CheckConstraint 是最终防线）
+  - AffinityLevel.save() 自动调 full_clean 触发跨等级区间不重叠校验（WR-009 多层兜底，详见 Commit D）；提供 `skip_clean=True` 后门给迁移 / fixture 场景
+- **修改原因**: P1 审查指出（详见 REVIEW-affinity-P1.md CR-002）模型字段对管理后台 / shell / 直 SQL 写入毫无保护，P2 服务层 `random.randint(rule.min_change, rule.max_change)` 等运算会被脏数据击穿（ValueError 抛出、冷却永久解锁、上限永远命中等）；同时审查中 WR-001 指出 AffinitySetting 单例保证在并发下脆弱
+- **跨项目联动**: 无 — 仅服务端 DB 约束 + 模型校验层；管理后台前端在写入前应捕获 ValidationError 显示给运营，但接口契约本身未动
+
 ### [2026-05-13] 好感度系统 P1 审查修复 A — UserDevice 软删语义修正（CR-001 + IN-005）
 
 配套审查报告：[docs/REVIEW-affinity-P1.md](REVIEW-affinity-P1.md)（CR-001 + IN-005）

qy_lty/userapp/migrations/0007_add_affinity_check_constraints.py

@@ -0,0 +1,65 @@
+# Generated by Django 5.2.12 on 2026-05-13 02:11
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('userapp', '0006_migrate_favorability_to_userdevice'),
+    ]
+
+    operations = [
+        migrations.AddConstraint(
+            model_name='affinitylevel',
+            constraint=models.CheckConstraint(condition=models.Q(('min_affinity__lte', models.F('max_affinity'))), name='affinitylevel_min_le_max'),
+        ),
+        migrations.AddConstraint(
+            model_name='affinitylevel',
+            constraint=models.CheckConstraint(condition=models.Q(('reward_currency__gte', 0)), name='affinitylevel_currency_nonneg'),
+        ),
+        migrations.AddConstraint(
+            model_name='affinityrule',
+            constraint=models.CheckConstraint(condition=models.Q(('min_change__lte', models.F('max_change'))), name='affinityrule_min_le_max'),
+        ),
+        migrations.AddConstraint(
+            model_name='affinityrule',
+            constraint=models.CheckConstraint(condition=models.Q(('cooldown_seconds__gte', 0)), name='affinityrule_cooldown_nonneg'),
+        ),
+        migrations.AddConstraint(
+            model_name='affinityrule',
+            constraint=models.CheckConstraint(condition=models.Q(('single_cap__gt', 0)), name='affinityrule_single_cap_positive'),
+        ),
+        migrations.AddConstraint(
+            model_name='affinityrule',
+            constraint=models.CheckConstraint(condition=models.Q(('daily_cap__gt', 0)), name='affinityrule_daily_cap_positive'),
+        ),
+        migrations.AddConstraint(
+            model_name='affinityrule',
+            constraint=models.CheckConstraint(condition=models.Q(models.Q(('trigger_type', 'companion_time'), _negated=True), models.Q(('min_continuous_minutes__gt', 0), ('max_count_per_day__gt', 0)), _connector='OR'), name='affinityrule_companion_fields_present'),
+        ),
+        migrations.AddConstraint(
+            model_name='affinitysetting',
+            constraint=models.CheckConstraint(condition=models.Q(('decay_min_decay__lte', models.F('decay_max_decay'))), name='affinitysetting_decay_min_le_max'),
+        ),
+        migrations.AddConstraint(
+            model_name='affinitysetting',
+            constraint=models.CheckConstraint(condition=models.Q(('decay_max_decay__lte', models.F('decay_cap'))), name='affinitysetting_decay_within_cap'),
+        ),
+        migrations.AddConstraint(
+            model_name='affinitysetting',
+            constraint=models.CheckConstraint(condition=models.Q(('initial_affinity__lte', models.F('max_affinity'))), name='affinitysetting_initial_le_max'),
+        ),
+        migrations.AddConstraint(
+            model_name='affinitysetting',
+            constraint=models.CheckConstraint(condition=models.Q(('decay_min_floor__lte', models.F('max_affinity'))), name='affinitysetting_floor_le_max'),
+        ),
+        migrations.AddConstraint(
+            model_name='affinitysetting',
+            constraint=models.CheckConstraint(condition=models.Q(('daily_cap__gt', 0)), name='affinitysetting_daily_cap_positive'),
+        ),
+        migrations.AddConstraint(
+            model_name='affinitysetting',
+            constraint=models.CheckConstraint(condition=models.Q(('pk', 1)), name='affinitysetting_singleton'),
+        ),
+    ]

qy_lty/userapp/models.py

@@ -1,7 +1,8 @@
+from django.core.exceptions import ValidationError
 from django.db import models
 from django.conf import settings
 from django.contrib.auth.models import AbstractUser, Group, Permission
-from django.db.models import UniqueConstraint, Q
+from django.db.models import CheckConstraint, F, Q, UniqueConstraint
 from rest_framework.authtoken.models import Token as DefaultToken
 
 
@@ -166,10 +167,56 @@ class AffinityRule(models.Model):
         verbose_name = '好感度规则'
         verbose_name_plural = '好感度规则'
         ordering = ['-created_at']
+        constraints = [
+            CheckConstraint(
+                check=Q(min_change__lte=F('max_change')),
+                name='affinityrule_min_le_max',
+            ),
+            CheckConstraint(
+                check=Q(cooldown_seconds__gte=0),
+                name='affinityrule_cooldown_nonneg',
+            ),
+            CheckConstraint(
+                check=Q(single_cap__gt=0),
+                name='affinityrule_single_cap_positive',
+            ),
+            CheckConstraint(
+                check=Q(daily_cap__gt=0),
+                name='affinityrule_daily_cap_positive',
+            ),
+            # trigger_type='companion_time' 时必须配套 min_continuous_minutes / max_count_per_day
+            CheckConstraint(
+                check=(~Q(trigger_type='companion_time')) |
+                      (Q(min_continuous_minutes__gt=0) & Q(max_count_per_day__gt=0)),
+                name='affinityrule_companion_fields_present',
+            ),
+        ]
 
     def __str__(self):
         return f"{self.name} ({self.rule_key or '-'})"
 
+    def clean(self):
+        """Python 级兜底校验 — 在 admin / serializer 调用 full_clean() 时触发，
+        给前端友好错误信息；DB 级 CheckConstraint 是最终防线（详见 Meta.constraints）。
+        """
+        super().clean()
+        errors = {}
+        if self.min_change > self.max_change:
+            errors['max_change'] = '最大变化值不能小于最小变化值'
+        if self.cooldown_seconds < 0:
+            errors['cooldown_seconds'] = '冷却时间不能为负数'
+        if self.single_cap <= 0:
+            errors['single_cap'] = '单次上限必须大于 0'
+        if self.daily_cap <= 0:
+            errors['daily_cap'] = '每日上限必须大于 0'
+        if self.trigger_type == 'companion_time':
+            if not self.min_continuous_minutes or self.min_continuous_minutes <= 0:
+                errors['min_continuous_minutes'] = '陪伴时长规则必须设置 min_continuous_minutes > 0'
+            if not self.max_count_per_day or self.max_count_per_day <= 0:
+                errors['max_count_per_day'] = '陪伴时长规则必须设置 max_count_per_day > 0'
+        if errors:
+            raise ValidationError(errors)
+
 
 class AffinityLevel(models.Model):
     """好感度等级：定义不同好感度区间对应的等级和奖励
@@ -239,10 +286,57 @@ class AffinityLevel(models.Model):
         verbose_name = '好感度等级'
         verbose_name_plural = '好感度等级'
         ordering = ['level']
+        constraints = [
+            CheckConstraint(
+                check=Q(min_affinity__lte=F('max_affinity')),
+                name='affinitylevel_min_le_max',
+            ),
+            CheckConstraint(
+                check=Q(reward_currency__gte=0),
+                name='affinitylevel_currency_nonneg',
+            ),
+        ]
 
     def __str__(self):
         return f"Lv{self.level} {self.name}"
 
+    def clean(self):
+        """Python 级兜底校验 — 检查区间合法 + 与同表其它等级不重叠。
+
+        注：跨行不重叠约束 PostgreSQL CheckConstraint 表达不出（需 ExclusionConstraint +
+        btree_gist 扩展），所以放在应用层；DB 仅约束单行 min ≤ max 一致性。
+        详见审查报告 WR-009。
+        """
+        super().clean()
+        errors = {}
+        if self.min_affinity > self.max_affinity:
+            errors['max_affinity'] = '上限不能小于下限'
+        if self.reward_currency < 0:
+            errors['reward_currency'] = '虚拟货币奖励不能为负'
+        # 与其它等级的区间重叠检查（同表多行，仅 Python 层校验）
+        if not errors:
+            overlaps = type(self).objects.exclude(pk=self.pk).filter(
+                is_deleted=False,
+                min_affinity__lte=self.max_affinity,
+                max_affinity__gte=self.min_affinity,
+            )
+            if overlaps.exists():
+                names = ', '.join(f'Lv{l.level}' for l in overlaps[:5])
+                errors['min_affinity'] = f'与已存在等级 {names} 区间重叠'
+        if errors:
+            raise ValidationError(errors)
+
+    def save(self, *args, **kwargs):
+        """save 前触发 full_clean，确保跨行约束（区间不重叠）在任何写入路径都生效。
+
+        注：迁移文件内使用历史模型（apps.get_model）不会触发此 save 重载，
+        因此数据迁移可以批量写入而不被校验；正常业务代码 / admin / DRF 路径会校验。
+        """
+        # 跳过自动 clean 的开关（迁移 / fixture / seed 场景可显式传 skip_clean=True）
+        if not kwargs.pop('skip_clean', False):
+            self.full_clean()
+        super().save(*args, **kwargs)
+
 
 class AffinitySetting(models.Model):
     """好感度系统全局参数（单例表）— P1-04
@@ -296,20 +390,65 @@ class AffinitySetting(models.Model):
     class Meta:
         verbose_name = '好感度系统设置'
         verbose_name_plural = '好感度系统设置'
+        constraints = [
+            CheckConstraint(
+                check=Q(decay_min_decay__lte=F('decay_max_decay')),
+                name='affinitysetting_decay_min_le_max',
+            ),
+            CheckConstraint(
+                check=Q(decay_max_decay__lte=F('decay_cap')),
+                name='affinitysetting_decay_within_cap',
+            ),
+            CheckConstraint(
+                check=Q(initial_affinity__lte=F('max_affinity')),
+                name='affinitysetting_initial_le_max',
+            ),
+            CheckConstraint(
+                check=Q(decay_min_floor__lte=F('max_affinity')),
+                name='affinitysetting_floor_le_max',
+            ),
+            CheckConstraint(
+                check=Q(daily_cap__gt=0),
+                name='affinitysetting_daily_cap_positive',
+            ),
+            # 单例硬约束：所有写入必须 pk=1（CHECK 约束不能跨行限制行数，
+            # 但能强制任何 INSERT/UPDATE 的 pk 都是 1，配合 save() 强制 pk=1 形成事实单例）
+            # 详见审查报告 WR-001
+            CheckConstraint(
+                check=Q(pk=1),
+                name='affinitysetting_singleton',
+            ),
+        ]
 
     def __str__(self):
         return f"AffinitySetting(initial={self.initial_affinity}, max={self.max_affinity})"
 
+    def clean(self):
+        """Python 级兜底校验"""
+        super().clean()
+        errors = {}
+        if self.decay_min_decay > self.decay_max_decay:
+            errors['decay_max_decay'] = '单日衰减最大值不能小于最小值'
+        if self.decay_max_decay > self.decay_cap:
+            errors['decay_cap'] = '单日衰减上限不能小于最大值'
+        if self.initial_affinity > self.max_affinity:
+            errors['initial_affinity'] = '初始好感度不能超过最大好感度'
+        if self.decay_min_floor > self.max_affinity:
+            errors['decay_min_floor'] = '衰减下限不能超过最大好感度'
+        if self.daily_cap <= 0:
+            errors['daily_cap'] = '每日全局增长上限必须大于 0'
+        if errors:
+            raise ValidationError(errors)
+
     def save(self, *args, **kwargs):
-        # 强制单例：新增时如果已有记录则覆盖到现有 pk
+        # 强制单例：固定 pk=1（WR-001 升级版 — 配合 DB CHECK pk=1 约束防并发重复行）
-        if not self.pk and AffinitySetting.objects.exists():
+        if not self.pk:
-            existing = AffinitySetting.objects.first()
+            self.pk = 1
-            self.pk = existing.pk
         super().save(*args, **kwargs)
 
     @classmethod
     def get_solo(cls):
-        """获取单例实例，不存在则用默认值创建"""
+        """获取单例实例，不存在则用默认值创建（强制 pk=1）"""
         instance, _ = cls.objects.get_or_create(pk=1)
         return instance