fix: 删除命令强制仅 lsx,预检清单,禁止误删其他俱乐部

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
XingQue
2026-07-23 22:04:13 +08:00
parent 34344fa5b2
commit 0d9139d5cc

View File

@@ -1,8 +1,17 @@
"""按微信商户号彻底删除俱乐部及其绑定配置(不删业务订单/真实用户)
"""仅清理「龙先生电竞」(club_id=lsx) 的配置;绝对不碰其他俱乐部
安全规则(硬约束):
1. 只允许操作 club_id=lsx或显式传入且必须等于 lsx
2. 若商户号命中了非 lsx 俱乐部:只打印警告,绝不删除
3. 默认 dry-run / --precheck只列出将删内容与条数不写库
4. --apply 必须再带 --i-understand-lsx-only 才真正删除
用法:
python manage.py delete_club_by_mch_id 17480975501
python manage.py delete_club_by_mch_id 17480975501 --apply
# 预检(推荐先跑这个,把输出发给你确认)
python manage.py delete_club_by_mch_id 17480975501 --precheck
# 确认无误后再删(双确认)
python manage.py delete_club_by_mch_id 17480975501 --apply --i-understand-lsx-only
"""
from __future__ import annotations
@@ -41,7 +50,6 @@ try:
except Exception:
AdminAuditLog = None
# 展示/运营配置config app
try:
from config.models import (
ClubMiniappIcon,
@@ -55,232 +63,365 @@ try:
Szjilu,
Tupianpeizhi,
)
except Exception: # pragma: no cover
except Exception:
Lunbo = Gonggao = Tupianpeizhi = PopupPage = Szjilu = None
ClubMiniappIcon = ClubPindao = ClubPindaoImage = None
MiniappScriptScene = MiniappScriptAutoReply = None
def _delete_qs(model, club_id, stdout, label=None):
# 本命令唯一允许删除的俱乐部
ALLOWED_CLUB_ID = 'lsx'
ALLOWED_CLUB_NAME_HINT = '龙先生'
# ORM 配置模型(均按 club_id 过滤)
ORM_MODELS = [
('支付通道 ClubPaymentChannel', ClubPaymentChannel),
('会员价 ClubHuiyuanPrice', ClubHuiyuanPrice),
('会员套餐 ClubHuiyuanBundleInclude', ClubHuiyuanBundleInclude),
('利率 ClubLilubiao', ClubLilubiao),
('提现配置 ClubWithdrawConfig', ClubWithdrawConfig),
('提现配额 ClubTixianQuota', ClubTixianQuota),
('商品类型上架 ClubShangpinLeixingConfig', ClubShangpinLeixingConfig),
('考核称号 ClubKaoheChenghaoConfig', ClubKaoheChenghaoConfig),
('发单分红利率 ClubFadanFenhongLilv', ClubFadanFenhongLilv),
('接单考试配置 ClubDashouExamConfig', ClubDashouExamConfig),
('接单考试题图 ClubDashouExamQuestionImage', ClubDashouExamQuestionImage),
('接单考试题 ClubDashouExamQuestion', ClubDashouExamQuestion),
('接单考试通过 ClubDashouExamPass', ClubDashouExamPass),
('身份标签绑定 IdentityTagBind', IdentityTagBind),
('身份标签 IdentityTag', IdentityTag),
('支付通道流水 PayChannelLedger', PayChannelLedger),
('OpenID绑定 UserWxOpenid', UserWxOpenid),
('客服任职 AdminAssignment', AdminAssignment),
('审计日志 AdminAuditLog', AdminAuditLog),
('轮播 Lunbo', Lunbo),
('公告 Gonggao', Gonggao),
('图片配置 Tupianpeizhi', Tupianpeizhi),
('弹窗页 PopupPage', PopupPage),
('收支记录配置 Szjilu', Szjilu),
('小程序图标 ClubMiniappIcon', ClubMiniappIcon),
('频道图 ClubPindaoImage', ClubPindaoImage),
('频道 ClubPindao', ClubPindao),
('话术场景 MiniappScriptScene', MiniappScriptScene),
('话术自动回复 MiniappScriptAutoReply', MiniappScriptAutoReply),
]
# SQL 兜底表(仅 DELETE WHERE club_id=%s绝不会无 club_id 条件)
SQL_TABLES = [
'club_payment_channel',
'club_huiyuan_price',
'club_huiyuan_bundle_include',
'club_lilubiao',
'club_withdraw_config',
'club_tixian_quota',
'club_shangpin_leixing_config',
'club_kaohe_chenghao_config',
'club_fadan_fenhong_lilv',
'club_dashou_exam_config',
'club_dashou_exam_question',
'club_dashou_exam_question_image',
'club_dashou_exam_pass',
'identity_tag_bind',
'identity_tag',
'pay_channel_ledger',
'user_wx_openid',
'admin_assignment',
'admin_audit_log',
'lunbo',
'gonggao',
'tupianpeizhi',
'popup_page',
'szjilu',
'club_miniapp_icon',
'club_pindao',
'club_pindao_image',
'miniapp_script_scene',
'miniapp_script_auto_reply',
'daily_income_stat',
'daily_payout_stat',
'platform_income_log',
]
# 明确不删(业务数据)
NEVER_DELETE_NOTE = [
'User 用户主表 / 各身份扩展表(打手商家等)',
'订单 Order / 退款 / 处罚等业务流水',
'商品 Shangpin若 club_id=lsx 有商品也不会在本命令删;如需另开命令)',
'其他俱乐部 xq / xzj / ufo 等任何数据',
]
def _count_model(model, club_id):
if model is None:
return 0
name = label or model.__name__
return None
try:
qs = model.query.filter(club_id=club_id)
n = qs.count()
if n:
qs.delete()
stdout.write(f' - {name}: 删除 {n}')
else:
stdout.write(f' - {name}: 0')
return n
except Exception as exc:
stdout.write(f' - {name}: 跳过 ({exc})')
return 0
return model.query.filter(club_id=club_id).count()
except Exception:
return None
def _find_clubs_by_mch(mch_id: str):
"""club.mch_id / withdraw_mch_id / payment_channel.mch_id / extra_json 命中。"""
clubs = list(
def _assert_lsx_only(club_id: str):
cid = (club_id or '').strip()
if cid != ALLOWED_CLUB_ID:
raise CommandError(
f'安全拦截:本命令只允许操作 club_id={ALLOWED_CLUB_ID}(龙先生电竞),'
f'拒绝 club_id={cid!r}'
)
def _club_snapshot(club: Club):
return {
'club_id': club.club_id,
'name': club.name,
'status': club.status,
'wx_appid': club.wx_appid or '',
'official_appid': club.official_appid or '',
'official_secret_set': bool((club.official_secret or '').strip()),
'official_token_set': bool((club.official_token or '').strip()),
'encoding_aes_key_set': bool((club.encoding_aes_key or '').strip()),
'template_id': club.template_id or '',
'mch_id': club.mch_id or '',
'withdraw_mch_id': getattr(club, 'withdraw_mch_id', '') or '',
'pay_app_id': club.pay_app_id or '',
'pay_key_path': club.pay_key_path or '',
'pay_cert_path': club.pay_cert_path or '',
'private_key_path': club.private_key_path or '',
'platform_cert_dir': club.platform_cert_dir or '',
'goeasy_appkey_set': bool((club.goeasy_appkey or '').strip()),
'h5_domain': club.h5_domain or '',
'oss_prefix': club.oss_prefix or '',
'mini_pay_channel': getattr(club, 'mini_pay_channel', '') or '',
}
def _find_other_clubs_with_mch(mch_id: str, exclude_club_id: str):
"""商户号撞到其他俱乐部时列出,绝不删除它们。"""
others = list(
Club.query.filter(
Q(mch_id=mch_id) | Q(withdraw_mch_id=mch_id)
)
).exclude(club_id=exclude_club_id)
)
found_ids = {c.club_id for c in clubs}
for ch in ClubPaymentChannel.query.filter(mch_id=mch_id):
if ch.club_id not in found_ids:
c = Club.query.filter(club_id=ch.club_id).first()
ch_club_ids = (
ClubPaymentChannel.query.filter(mch_id=mch_id)
.exclude(club_id=exclude_club_id)
.values_list('club_id', flat=True)
.distinct()
)
for cid in ch_club_ids:
if not any(c.club_id == cid for c in others):
c = Club.query.filter(club_id=cid).first()
if c:
clubs.append(c)
found_ids.add(c.club_id)
# extra_json 里可能写 merchant_id
for ch in ClubPaymentChannel.query.all().only('id', 'club_id', 'mch_id', 'extra_json'):
extra = ch.extra_json or {}
mid = str(extra.get('merchant_id') or extra.get('mch_id') or '').strip()
if mid == mch_id and ch.club_id not in found_ids:
c = Club.query.filter(club_id=ch.club_id).first()
if c:
clubs.append(c)
found_ids.add(c.club_id)
return clubs
others.append(c)
return others
def _purge_club(club: Club, stdout, style, apply: bool, remove_cert_dir: bool):
cid = club.club_id
stdout.write(style.WARNING(f'\n>>> 俱乐部 {cid} ({club.name})'))
stdout.write(f' mch_id={club.mch_id or "(空)"} withdraw_mch_id={getattr(club, "withdraw_mch_id", "") or "(空)"}')
stdout.write(f' wx_appid={club.wx_appid or "(空)"}')
stdout.write(f' pay_key_path={club.pay_key_path or "()"}')
stdout.write(f' pay_cert_path={club.pay_cert_path or "(空)"}')
stdout.write(f' private_key_path={club.private_key_path or "(空)"}')
def _precheck_report(stdout, style, club: Club, mch_id: str):
snap = _club_snapshot(club)
stdout.write(style.SUCCESS('=' * 64))
stdout.write(style.SUCCESS('【预检】仅针对龙先生电竞 club_id=lsx'))
stdout.write(style.SUCCESS('=' * 64))
stdout.write(f'请求商户号: {mch_id or "(未指定,仅按 lsx 清理)"}')
stdout.write('')
stdout.write('—— 俱乐部主表字段 ——')
for k, v in snap.items():
stdout.write(f' {k}: {v if v != "" else "(空)"}')
channel_n = ClubPaymentChannel.query.filter(club_id=cid).count()
stdout.write(f' payment_channel 条数={channel_n}')
if mch_id:
club_mch = (club.mch_id or '').strip()
club_wm = (getattr(club, 'withdraw_mch_id', '') or '').strip()
ch_hit = ClubPaymentChannel.query.filter(club_id=ALLOWED_CLUB_ID, mch_id=mch_id).count()
matched = club_mch == mch_id or club_wm == mch_id or ch_hit > 0
if not matched:
stdout.write(style.WARNING(
f'\n注意lsx 当前 mch_id={club_mch or "(空)"} / withdraw={club_wm or "(空)"} / '
f'通道命中={ch_hit},与请求商户号 {mch_id} 不完全一致。'
f'本命令仍只清理 lsx不会动其他俱乐部。'
))
else:
stdout.write(style.SUCCESS(f'\n商户号匹配确认lsx 绑定了 {mch_id}'))
if not apply:
stdout.write(' (dry-run未删除)')
return
others = _find_other_clubs_with_mch(mch_id, ALLOWED_CLUB_ID)
if others:
stdout.write(style.ERROR(
f'\n!! 其他俱乐部也配置了商户号 {mch_id}(只警告,绝对不删):'
))
for o in others:
stdout.write(
f' - KEEP {o.club_id} ({o.name}) mch={o.mch_id} withdraw={getattr(o, "withdraw_mch_id", "")}'
)
else:
stdout.write('其他俱乐部未发现同商户号(安全)。')
stdout.write('\n—— 将删除的配置表(仅 club_id=lsx——')
total = 0
for label, model in ORM_MODELS:
n = _count_model(model, ALLOWED_CLUB_ID)
if n is None:
stdout.write(f' [跳过不可用] {label}')
continue
mark = '将删' if n else ''
stdout.write(f' [{mark}] {label}: {n}')
total += n or 0
stdout.write('\n—— SQL 兜底表计数(仅 WHERE club_id=lsx——')
with connection.cursor() as cursor:
for table in SQL_TABLES:
try:
cursor.execute(
f'SELECT COUNT(*) FROM `{table}` WHERE club_id=%s',
[ALLOWED_CLUB_ID],
)
n = cursor.fetchone()[0]
stdout.write(f' [{("将删" if n else "")}] {table}: {n}')
total += n
except Exception as exc:
stdout.write(f' [无此表/跳过] {table}: {exc}')
cert_dir = os.path.join(club_wx_cert_root(), ALLOWED_CLUB_ID)
stdout.write('\n—— 证书目录 ——')
if os.path.isdir(cert_dir):
file_n = 0
for _root, _dirs, files in os.walk(cert_dir):
file_n += len(files)
stdout.write(f' [将删目录] {cert_dir} (约 {file_n} 个文件)')
else:
stdout.write(f' [无目录] {cert_dir}')
stdout.write('\n—— 明确不删 ——')
for tip in NEVER_DELETE_NOTE:
stdout.write(f' · {tip}')
stdout.write('')
stdout.write(style.SUCCESS(f'预检合计(配置行粗略合计,含重复计数可能):约 {total}'))
stdout.write(style.WARNING(
'确认无误后再执行:\n'
f' python manage.py delete_club_by_mch_id {mch_id or "17480975501"} '
'--apply --i-understand-lsx-only'
))
stdout.write(style.SUCCESS('=' * 64))
def _delete_lsx(stdout, style, remove_cert_dir: bool):
_assert_lsx_only(ALLOWED_CLUB_ID)
club = Club.query.filter(club_id=ALLOWED_CLUB_ID).first()
if not club:
raise CommandError(f'俱乐部 {ALLOWED_CLUB_ID} 不存在,无需删除')
# 名称再确认一层(非强制阻断,只警告)
if ALLOWED_CLUB_NAME_HINT not in (club.name or ''):
stdout.write(style.WARNING(
f'警告:{ALLOWED_CLUB_ID} 的 name={club.name!r} 不含「{ALLOWED_CLUB_NAME_HINT}」,'
f'但仍按 club_id=lsx 删除(主键约束)。'
))
with transaction.atomic():
# jituan 配置表
for model in (
ClubPaymentChannel,
ClubHuiyuanPrice,
ClubHuiyuanBundleInclude,
ClubLilubiao,
ClubWithdrawConfig,
ClubTixianQuota,
ClubShangpinLeixingConfig,
ClubKaoheChenghaoConfig,
ClubFadanFenhongLilv,
ClubDashouExamConfig,
ClubDashouExamQuestionImage,
ClubDashouExamQuestion,
ClubDashouExamPass,
IdentityTagBind,
IdentityTag,
PayChannelLedger,
UserWxOpenid,
AdminAssignment,
AdminAuditLog,
):
_delete_qs(model, cid, stdout)
for label, model in ORM_MODELS:
if model is None:
continue
n = model.query.filter(club_id=ALLOWED_CLUB_ID).count()
if n:
model.query.filter(club_id=ALLOWED_CLUB_ID).delete()
stdout.write(f' - 已删 {label}: {n}')
# 展示配置
for model in (
Lunbo, Gonggao, Tupianpeizhi, PopupPage, Szjilu,
ClubMiniappIcon, ClubPindaoImage, ClubPindao,
MiniappScriptScene, MiniappScriptAutoReply,
):
_delete_qs(model, cid, stdout)
# 兜底:扫一遍已知配置表(防 ORM 漏删)
with connection.cursor() as cursor:
tables = [
'club_payment_channel',
'club_huiyuan_price',
'club_huiyuan_bundle_include',
'club_lilubiao',
'club_withdraw_config',
'club_tixian_quota',
'club_shangpin_leixing_config',
'club_kaohe_chenghao_config',
'club_fadan_fenhong_lilv',
'club_dashou_exam_config',
'club_dashou_exam_question',
'club_dashou_exam_question_image',
'club_dashou_exam_pass',
'identity_tag_bind',
'identity_tag',
'pay_channel_ledger',
'user_wx_openid',
'admin_assignment',
'admin_audit_log',
'lunbo',
'gonggao',
'tupianpeizhi',
'popup_page',
'szjilu',
'club_miniapp_icon',
'club_pindao',
'club_pindao_image',
'miniapp_script_scene',
'miniapp_script_auto_reply',
'daily_income_stat',
'daily_payout_stat',
'platform_income_log',
]
for table in tables:
for table in SQL_TABLES:
try:
cursor.execute(f'SELECT COUNT(*) FROM `{table}` WHERE club_id=%s', [cid])
cursor.execute(
f'SELECT COUNT(*) FROM `{table}` WHERE club_id=%s',
[ALLOWED_CLUB_ID],
)
n = cursor.fetchone()[0]
if n:
cursor.execute(f'DELETE FROM `{table}` WHERE club_id=%s', [cid])
stdout.write(f' - SQL {table}: 删除 {n}')
cursor.execute(
f'DELETE FROM `{table}` WHERE club_id=%s',
[ALLOWED_CLUB_ID],
)
stdout.write(f' - SQL 已删 {table}: {n}')
except Exception:
pass
Club.query.filter(club_id=cid).delete()
stdout.write(style.SUCCESS(f' - Club 主表已删除: {cid}'))
# 最后删主表(再次校验主键)
deleted, _ = Club.query.filter(club_id=ALLOWED_CLUB_ID).delete()
stdout.write(style.SUCCESS(f' - 已删 Club 主表 lsx (affected={deleted})'))
if remove_cert_dir:
root = club_wx_cert_root()
cert_dir = os.path.join(root, cid)
cert_dir = os.path.join(club_wx_cert_root(), ALLOWED_CLUB_ID)
if os.path.isdir(cert_dir):
if apply:
shutil.rmtree(cert_dir, ignore_errors=True)
stdout.write(style.SUCCESS(f' - 已删除证书目录: {cert_dir}'))
else:
stdout.write(f' - 将删除证书目录: {cert_dir}')
else:
stdout.write(f' - 证书目录不存在: {cert_dir}')
shutil.rmtree(cert_dir, ignore_errors=True)
stdout.write(style.SUCCESS(f' - 已删证书目录 {cert_dir}'))
class Command(BaseCommand):
help = '按商户号删除俱乐部及其全部绑定配置(默认 dry-run'
help = '【仅 lsx/龙先生】按商户号预检/删除俱乐部配置;绝不碰其他俱乐部'
def add_arguments(self, parser):
parser.add_argument(
'mch_id',
nargs='?',
default='17480975501',
help='微信商户号(默认 17480975501',
help='用于核对的商户号(默认 17480975501;删除范围仍固定为 lsx',
)
parser.add_argument(
'--precheck',
action='store_true',
help='只预检列出将删内容(默认行为也是预检)',
)
parser.add_argument('--apply', action='store_true', help='真正删除(需配合确认开关)')
parser.add_argument(
'--i-understand-lsx-only',
action='store_true',
help='确认仅删除龙先生 lsx不涉及其他俱乐部',
)
parser.add_argument('--apply', action='store_true', help='真正删除')
parser.add_argument(
'--keep-cert-dir',
action='store_true',
help='不删磁盘 wx_certs/{club_id} 目录',
help='保留磁盘 wx_certs/lsx 目录',
)
parser.add_argument(
'--club-id',
type=str,
default=ALLOWED_CLUB_ID,
help='必须为 lsx传其他值会直接拒绝',
)
def handle(self, *args, **options):
club_id = (options.get('club_id') or ALLOWED_CLUB_ID).strip()
_assert_lsx_only(club_id)
mch_id = str(options.get('mch_id') or '').strip()
if not mch_id:
raise CommandError('请提供商户号')
apply = bool(options.get('apply'))
remove_cert = not bool(options.get('keep_cert_dir'))
confirm = bool(options.get('i_understand_lsx_only'))
# 默认/显式 precheck 都走预检;只有 apply+确认才删除
do_delete = apply and confirm
clubs = _find_clubs_by_mch(mch_id)
# 也列出仅有通道命中、主表已无的 club_id
orphan_channels = list(
ClubPaymentChannel.query.filter(mch_id=mch_id).exclude(
club_id__in=[c.club_id for c in clubs]
club = Club.query.filter(club_id=ALLOWED_CLUB_ID).first()
if not club:
raise CommandError(
f'未找到 club_id={ALLOWED_CLUB_ID}。不会删除任何其他俱乐部。'
)
)
self.stdout.write(self.style.SUCCESS('=' * 60))
self.stdout.write(f'目标商户号 mch_id = {mch_id}')
self.stdout.write(f'命中俱乐部: {len(clubs)}')
if not clubs and not orphan_channels:
self.stdout.write(self.style.WARNING('未找到绑定该商户号的俱乐部/支付通道,无需删除。'))
_precheck_report(self.stdout, self.style, club, mch_id)
if not do_delete:
if apply and not confirm:
self.stdout.write(self.style.ERROR(
'\n已拒绝删除:缺少 --i-understand-lsx-only防止误删'
))
self.stdout.write(self.style.WARNING('\n当前仅预检,数据库未改动。'))
return
for club in clubs:
_purge_club(club, self.stdout, self.style, apply, remove_cert)
if orphan_channels:
self.stdout.write(self.style.WARNING('\n仅有支付通道、无 Club 主表的残留:'))
for ch in orphan_channels:
self.stdout.write(
f' channel id={ch.id} club_id={ch.club_id} name={ch.name} mch_id={ch.mch_id}'
)
if apply:
cid = ch.club_id
with transaction.atomic():
ClubPaymentChannel.query.filter(id=ch.id).delete()
# 顺带清同 club_id 残留配置
for model in (
ClubPaymentChannel, PayChannelLedger, AdminAssignment, UserWxOpenid,
):
_delete_qs(model, cid, self.stdout)
self.stdout.write(self.style.SUCCESS(f' 已删通道 id={ch.id}'))
if not apply:
self.stdout.write(self.style.WARNING(
f'\n当前为预览。确认后执行:\n'
f' python manage.py delete_club_by_mch_id {mch_id} --apply'
# 删除前再拦一道:其他俱乐部同商户号只警告
others = _find_other_clubs_with_mch(mch_id, ALLOWED_CLUB_ID) if mch_id else []
if others:
self.stdout.write(self.style.ERROR(
'其他俱乐部存在同商户号配置,本命令仍只删 lsx下列俱乐部保持不动'
))
else:
self.stdout.write(self.style.SUCCESS('\n删除完成。'))
for o in others:
self.stdout.write(f' KEEP {o.club_id}')
self.stdout.write(self.style.WARNING('\n开始删除(仅 lsx...'))
_delete_lsx(
self.stdout,
self.style,
remove_cert_dir=not bool(options.get('keep_cert_dir')),
)
self.stdout.write(self.style.SUCCESS('\n完成:仅龙先生 lsx 配置已清理。'))