"""清理龙先生重复/无 pem 证书俱乐部。 规则： - 候选：club_id=lsx / 名称含「龙先生」 / wx_appid=龙先生小程序 - 「有 pem」：pay_key_path / pay_cert_path / private_key_path 任一路径以 .pem 结尾 （你后台上传的证书路径都是 *.pem；脚本自动建的/空配置通常没有） - 「可删」：上述三个字段都没有以 .pem 结尾的路径 - 默认 dry-run；--apply 才删除 - 若候选里没有任何「有 pem」的对照，默认不删；加 --force-delete-empty-only 才删 用法： python manage.py cleanup_lsx_empty_club python manage.py cleanup_lsx_empty_club --apply python manage.py cleanup_lsx_empty_club --apply --remove-seed-users """ from django.core.management.base import BaseCommand from django.db import transaction from django.db.models import Q from jituan.models import ( Club, ClubDashouExamConfig, ClubDashouExamPass, ClubDashouExamQuestion, ClubDashouExamQuestionImage, ClubFadanFenhongLilv, ClubHuiyuanBundleInclude, ClubHuiyuanPrice, ClubKaoheChenghaoConfig, ClubLilubiao, ClubPaymentChannel, ClubShangpinLeixingConfig, ClubTixianQuota, ClubWithdrawConfig, ) from users.business_models import User from users.models import UserBoss, UserGuanshi, UserZuzhang LSX_APPID = 'wx7ff90e9d024fcdb8' SEED_UIDS = ('9910001', '9910002') RELATED_MODELS = ( ClubPaymentChannel, ClubHuiyuanPrice, ClubHuiyuanBundleInclude, ClubLilubiao, ClubWithdrawConfig, ClubTixianQuota, ClubShangpinLeixingConfig, ClubKaoheChenghaoConfig, ClubFadanFenhongLilv, ClubDashouExamConfig, ClubDashouExamQuestion, ClubDashouExamQuestionImage, ClubDashouExamPass, ) # 证书「文件」字段（应以 .pem 结尾）；platform_cert_dir 是目录，不参与判定 _PEM_FILE_FIELDS = ('pay_key_path', 'pay_cert_path', 'private_key_path') def _path_ends_with_pem(path): return str(path or '').strip().lower().endswith('.pem') def _has_pem_paths(club): """后台上传过证书：路径以 .pem 结尾。""" return any(_path_ends_with_pem(getattr(club, f, '')) for f in _PEM_FILE_FIELDS) def _pem_field_summary(club): lines = [] for f in _PEM_FILE_FIELDS: val = (getattr(club, f, '') or '').strip() if not val: tag = '空' elif _path_ends_with_pem(val): tag = 'pem✓' else: tag = '非pem✗' lines.append(f'{f}={val or "(空)"} [{tag}]') plat = (getattr(club, 'platform_cert_dir', '') or '').strip() lines.append(f'platform_cert_dir={plat or "(空)"} [目录仅展示]') return lines def _candidates(): return Club.query.filter( Q(club_id__iexact='lsx') | Q(club_id__icontains='lsx') | Q(name__icontains='龙先生') | Q(wx_appid=LSX_APPID) ).order_by('club_id') def _purge_club_related(club_id, stdout, style): for model in RELATED_MODELS: try: n = model.query.filter(club_id=club_id).count() if n: model.query.filter(club_id=club_id).delete() stdout.write(f' - 已删 {model.__name__} x{n}') except Exception as exc: stdout.write(style.WARNING(f' - 跳过 {model.__name__}: {exc}')) class Command(BaseCommand): help = '列出并删除龙先生相关「证书路径不以 .pem 结尾」的重复俱乐部（默认 dry-run）' def add_arguments(self, parser): parser.add_argument('--apply', action='store_true', help='真正写入删除') parser.add_argument( '--force-delete-empty-only', action='store_true', help='即使没有「有 .pem」的对照俱乐部，也允许删除非 pem 候选', ) parser.add_argument( '--remove-seed-users', action='store_true', help='同时删除 seed_lsx_invite_data 种子用户 9910001/9910002', ) parser.add_argument( '--club-id', type=str, default='', help='只处理指定 club_id（精确匹配）', ) def handle(self, *args, **options): apply = bool(options.get('apply')) force = bool(options.get('force_delete_empty_only')) only_id = (options.get('club_id') or '').strip() qs = _candidates() if only_id: qs = Club.query.filter(club_id=only_id) clubs = list(qs) if not clubs: self.stdout.write(self.style.WARNING('未找到龙先生/lsx 相关俱乐部')) return with_pem = [] without_pem = [] self.stdout.write(self.style.SUCCESS('=' * 60)) self.stdout.write('判定标准：pay_key_path / pay_cert_path / private_key_path 是否以 .pem 结尾') self.stdout.write('龙先生相关俱乐部扫描结果：') for c in clubs: keyed = _has_pem_paths(c) (with_pem if keyed else without_pem).append(c) self.stdout.write('-' * 40) self.stdout.write(f'club_id={c.club_id} name={c.name} status={c.status}') self.stdout.write(f' wx_appid={c.wx_appid or "(空)"}') for line in _pem_field_summary(c): self.stdout.write(f' {line}') self.stdout.write( f' >>> {"有 .pem 路径（保留）" if keyed else "无 .pem 路径（可删）"}' ) self.stdout.write(self.style.SUCCESS('=' * 60)) self.stdout.write(f'有 .pem: {len(with_pem)} 条；无 .pem: {len(without_pem)} 条') if not without_pem: self.stdout.write(self.style.SUCCESS('没有「无 .pem 路径」的俱乐部，无需删除。')) elif not with_pem and not force: self.stdout.write(self.style.ERROR( '全部候选都没有 .pem 路径。为安全起见不自动删。' '若确认要删，加 --force-delete-empty-only --apply' )) else: for c in without_pem: self.stdout.write(self.style.WARNING( f'将删除无 .pem 俱乐部: {c.club_id} ({c.name})' )) if not apply: self.stdout.write(' (dry-run，未真正删除；加 --apply 执行)') continue with transaction.atomic(): _purge_club_related(c.club_id, self.stdout, self.style) Club.query.filter(club_id=c.club_id).delete() self.stdout.write(self.style.SUCCESS(f' 已删除俱乐部 {c.club_id}')) if options.get('remove_seed_users'): self.stdout.write('-' * 40) self.stdout.write('处理种子用户 9910001 / 9910002 ...') for uid in SEED_UIDS: user = User.query.filter(UserUID=uid).first() if not user: self.stdout.write(f' {uid}: 不存在') continue self.stdout.write( f' {uid}: ClubID={user.ClubID} Phone={user.Phone} OpenID={user.OpenID}' ) if not apply: self.stdout.write(' (dry-run)') continue with transaction.atomic(): UserZuzhang.query.filter(user=user).delete() UserGuanshi.query.filter(user=user).delete() UserBoss.query.filter(user=user).delete() user.delete() self.stdout.write(self.style.SUCCESS(f' 已删除用户 {uid}')) if not apply: self.stdout.write(self.style.WARNING( '\n当前为预览。确认后执行：\n' ' python manage.py cleanup_lsx_empty_club --apply\n' '如需顺带删种子账号：\n' ' python manage.py cleanup_lsx_empty_club --apply --remove-seed-users' ))