77 lines
2.7 KiB
Python
77 lines
2.7 KiB
Python
"""紧急封杀后台账号：硬删任职、摘超管、禁用客服、改密、清角色。
|
|
|
|
用法:
|
|
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} 在白名单，拒绝。确要封加 --also-whitelist'
|
|
)
|
|
return
|
|
|
|
from users.business_models import User
|
|
from jituan.models import AdminAssignment
|
|
from gvsdsdk.models import UserRole as SdkUserRole
|
|
import bcrypt
|
|
|
|
users = list(User.objects.filter(Phone=phone))
|
|
if not users:
|
|
self.stderr.write(f'未找到 {phone}')
|
|
return
|
|
|
|
new_pwd = f'LOCKED_{phone}_REVOKED'
|
|
for u in users:
|
|
self.stdout.write(f'HIT uid={u.UserUID} super={u.IsSuperuser}')
|
|
u.IsSuperuser = False
|
|
u.IsStaff = False
|
|
u.SetPassword(new_pwd)
|
|
u.save()
|
|
|
|
n, _ = AdminAssignment.objects.filter(yonghuid=u.UserUID).delete()
|
|
self.stdout.write(f' 硬删任职={n}')
|
|
|
|
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}')
|
|
|
|
# 顺手清全库 NULL 俱乐部垃圾重复
|
|
from jituan.services.admin_assignments import dedupe_active_assignments
|
|
d = dedupe_active_assignments()
|
|
self.stdout.write(f'全库去重硬删重复={d}')
|
|
|
|
self.stdout.write(self.style.SUCCESS(
|
|
f'已封杀 {phone}。密码={new_pwd}。请 restart gunicorn。'
|
|
))
|