fix: 紧急封杀后台账号命令 nuke_kefu_admin

Co-authored-by: Cursor <cursoragent@cursor.com>

jituan/management/commands/nuke_kefu_admin.py

@@ -0,0 +1,79 @@
+"""紧急封杀后台账号：摘超管、停全部任职、禁用客服、改密、清角色。
+
+用法（服务器）:
+  python manage.py nuke_kefu_admin 13837714110
+"""
+from django.core.management.base import BaseCommand
+from django.utils import timezone
+
+
+class Command(BaseCommand):
+    help = '紧急封杀指定手机号的后台权限（非交互）'
+
+    def add_arguments(self, parser):
+        parser.add_argument('phone', type=str, help='要封杀的手机号')
+        parser.add_argument(
+            '--also-whitelist',
+            action='store_true',
+            help='连代码白名单手机号也允许封（默认拒绝封白名单）',
+        )
+
+    def handle(self, *args, **options):
+        phone = (options['phone'] or '').strip()
+        if not phone:
+            self.stderr.write('缺少手机号')
+            return
+
+        from jituan.constants import SUPER_ADMIN_PHONES
+        if phone in SUPER_ADMIN_PHONES and not options['also_whitelist']:
+            self.stderr.write(
+                f'{phone} 在 SUPER_ADMIN_PHONES 白名单，拒绝封杀。'
+                f'确要封再加 --also-whitelist'
+            )
+            return
+
+        from users.business_models import User
+        from jituan.models import AdminAssignment
+
+        users = list(User.objects.filter(Phone=phone))
+        if not users:
+            self.stderr.write(f'未找到手机号 {phone}')
+            return
+
+        import bcrypt
+        from gvsdsdk.models import UserRole as SdkUserRole
+
+        new_pwd = f'LOCKED_{phone}_REVOKED'
+        for u in users:
+            self.stdout.write(
+                f'处理 UserUID={u.UserUID} IsSuperuser={u.IsSuperuser} IsStaff={u.IsStaff}'
+            )
+            u.IsSuperuser = False
+            u.IsStaff = False
+            u.SetPassword(new_pwd)
+            u.save()
+
+            n_asg = AdminAssignment.objects.filter(yonghuid=u.UserUID).update(
+                status=0,
+                UpdateTime=timezone.now(),
+            )
+            self.stdout.write(f'  任职停用行数={n_asg}')
+
+            n_role = SdkUserRole.objects.filter(UserUUID=u.UserUUID).delete()
+            self.stdout.write(f'  功能角色删除={n_role}')
+
+            try:
+                kefu = u.KefuProfile
+                kefu.zhuangtai = 0
+                kefu.erjimima = bcrypt.hashpw(
+                    b'LOCKED_REVOKED', bcrypt.gensalt(rounds=12)
+                ).decode('utf-8')
+                kefu.save(update_fields=['zhuangtai', 'erjimima', 'UpdateTime'])
+                self.stdout.write('  客服状态已禁用 + 二级密码已废')
+            except Exception as e:
+                self.stdout.write(f'  无客服扩展或跳过: {e}')
+
+        self.stdout.write(self.style.SUCCESS(
+            f'已封杀 {phone}（共 {len(users)} 个账号）。对方旧 token 需重新登录才会彻底失效；'
+            f'主密码已改为 LOCKED_{phone}_REVOKED，勿外传，事后请再改。'
+        ))