diff --git a/a_long_dianjing/management/commands/audit_freeze_feature.py b/a_long_dianjing/management/commands/audit_freeze_feature.py new file mode 100644 index 0000000..8f45fb6 --- /dev/null +++ b/a_long_dianjing/management/commands/audit_freeze_feature.py @@ -0,0 +1,483 @@ +""" +资金冻结功能 — 只读验收脚本(不修改任何业务数据) + +用法(生产环境): + python manage.py audit_freeze_feature + python manage.py audit_freeze_feature --verbose + python manage.py audit_freeze_feature --sample-users 20 + +说明: + - 仅 SELECT / 纯函数计算 / 单元测试,绝不写入或更新业务表 + - 可反复执行,不影响线上资金 +""" +from __future__ import annotations + +import decimal +import sys +import traceback +from dataclasses import dataclass, field +from typing import List, Optional + +from django.core.management.base import BaseCommand +from django.db import connection +from django.db.models import Q, Sum + +from yonghu.freeze_service import ( + DONGJIE_MODE_FIXED, + DONGJIE_MODE_RATIO, + FREEZE_LEIXING_SET, + FreezeConfig, + compute_freeze_split, + get_leixing_meta, + is_freeze_leixing, +) +from yonghu.models import ( + TixianDongjieGonggongPeizhi, + TixianDongjieJilu, + Tixianjilu, + TixianShenheJilu, + UserDashou, + UserGuanshi, + UserShangjia, +) + + +@dataclass +class CheckResult: + name: str + passed: bool + detail: str = '' + severity: str = 'error' # error | warn | info + + +@dataclass +class AuditReport: + results: List[CheckResult] = field(default_factory=list) + + def add(self, name: str, passed: bool, detail: str = '', severity: str = 'error'): + self.results.append(CheckResult(name, passed, detail, severity)) + + @property + def ok(self) -> bool: + return all(r.passed for r in self.results if r.severity == 'error') + + def print_report(self, verbose: bool = False): + errors = [r for r in self.results if not r.passed and r.severity == 'error'] + warns = [r for r in self.results if not r.passed and r.severity == 'warn'] + + print('\n========== 资金冻结只读验收报告 ==========\n') + for r in self.results: + if r.passed: + mark = '✓' + elif r.severity == 'warn': + mark = '⚠' + else: + mark = '✗' + line = f'{mark} [{r.severity.upper()}] {r.name}' + if r.detail and (verbose or not r.passed): + line += f'\n {r.detail}' + print(line) + + print('\n---------- 汇总 ----------') + print(f'错误: {len(errors)} 警告: {len(warns)} 通过项: {sum(1 for r in self.results if r.passed)}') + if errors: + print('\n❌ 存在需关注的错误项,请逐项排查。') + elif warns: + print('\n⚠️ 无致命错误,但有警告项建议人工复核。') + else: + print('\n✅ 只读验收全部通过(算法+结构+存量数据一致性)。') + + +def _q(v) -> decimal.Decimal: + return decimal.Decimal(str(v or 0)).quantize(decimal.Decimal('0.01')) + + +def _table_has_column(table: str, column: str) -> bool: + with connection.cursor() as cursor: + cursor.execute( + """ + SELECT COUNT(*) FROM information_schema.COLUMNS + WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s AND COLUMN_NAME = %s + """, + [table, column], + ) + return cursor.fetchone()[0] > 0 + + +def _table_exists(table: str) -> bool: + with connection.cursor() as cursor: + cursor.execute( + """ + SELECT COUNT(*) FROM information_schema.TABLES + WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s + """, + [table], + ) + return cursor.fetchone()[0] > 0 + + +def check_schema(report: AuditReport): + """数据库字段/表是否齐全(只读查 information_schema)""" + required_tables = { + 'withdraw_config': ['dongjie_quanju_enabled'], + 'tixian_dongjie_gonggong_peizhi': ['leixing', 'role_enabled', 'dongjie_mode', 'dongjie_bili', 'dongjie_guding_jine'], + 'tixian_dongjie_jilu': ['yonghuid', 'leixing', 'jine', 'caozuo'], + 'user_dashou': [ + 'dongjie_chi', 'dandu_dongjie_enabled', 'dongjie_mode', 'dongjie_bili', 'dongjie_guding_jine', + 'dongjie_chi_yajin', 'dandu_dongjie_yajin_enabled', 'dongjie_yajin_mode', 'dongjie_yajin_bili', 'dongjie_yajin_guding_jine', + ], + 'user_guanshi': ['dongjie_chi', 'dandu_dongjie_enabled', 'dongjie_mode', 'dongjie_bili', 'dongjie_guding_jine'], + 'user_shangjia': ['dongjie_chi', 'dandu_dongjie_enabled', 'dongjie_mode', 'dongjie_bili', 'dongjie_guding_jine'], + 'tixianjilu': ['benchi_dongjie_jine', 'shenqing_jine'], + 'tixian_shenhe_jilu': ['benchi_dongjie_jine', 'shenqing_jine'], + } + missing = [] + for table, cols in required_tables.items(): + if not _table_exists(table): + missing.append(f'表缺失: {table}') + continue + for col in cols: + if not _table_has_column(table, col): + missing.append(f'{table}.{col}') + report.add( + '数据库表字段齐全', + len(missing) == 0, + '缺失: ' + ', '.join(missing) if missing else f'已检查 {len(required_tables)} 张表', + ) + + +def check_public_config_seed(report: AuditReport): + """公共冻结配置四角色种子数据""" + found = set(TixianDongjieGonggongPeizhi.objects.values_list('leixing', flat=True)) + expected = set(FREEZE_LEIXING_SET) + missing = expected - found + report.add( + '公共冻结配置种子(1/2/5/6)', + len(missing) == 0, + f'缺少 leixing: {sorted(missing)}' if missing else f'已有 {len(found)} 条', + ) + + +def check_algorithm_unit_tests(report: AuditReport): + """纯函数算法验收(设计文档第五节表格 + 最终确认规则,不写库)""" + failures = [] + + cases = [ + # (name, balance, pool, config, apply, expect_withdraw, expect_freeze, expect_freeze_only) + ('无配置', '100', '0', None, '50', '50', '0', False), + ('比例10%_表例_申请50', '100', '0', FreezeConfig(1, decimal.Decimal('0.10'), decimal.Decimal('0'), 'p'), '50', '50', '10', False), + ('比例10%_表例_申请90', '100', '0', FreezeConfig(1, decimal.Decimal('0.10'), decimal.Decimal('0'), 'p'), '90', '90', '10', False), + ('比例10%_超额拒绝', '100', '0', FreezeConfig(1, decimal.Decimal('0.10'), decimal.Decimal('0'), 'p'), '91', None, None, None), + ('固定_池30_申请50_全入池', '50', '30', FreezeConfig(2, decimal.Decimal('0'), decimal.Decimal('100'), 'p'), '50', '0', '50', True), + ('固定_池30_申请80_填池70提10', '80', '30', FreezeConfig(2, decimal.Decimal('0'), decimal.Decimal('100'), 'p'), '80', '10', '70', False), + ('固定_池已满', '50', '100', FreezeConfig(2, decimal.Decimal('0'), decimal.Decimal('100'), 'p'), '50', '50', '0', False), + ] + + for name, bal, pool, cfg, apply, ew, ef, efo in cases: + try: + r = compute_freeze_split(_q(bal), _q(pool), cfg, _q(apply)) + if ew is None: + failures.append(f'{name}: 应拒绝但未拒绝') + continue + if r.withdraw_jine != _q(ew) or r.freeze_jine != _q(ef) or r.freeze_only != efo: + failures.append( + f'{name}: 期望 withdraw={ew} freeze={ef} only={efo}, ' + f'实际 withdraw={r.withdraw_jine} freeze={r.freeze_jine} only={r.freeze_only}' + ) + except ValueError: + if ew is None: + continue + failures.append(f'{name}: 不应拒绝但抛错') + except Exception as e: + failures.append(f'{name}: {e}') + + # 手续费应基于 withdraw_jine 而非 apply_jine(比例30%余额100申请50 → withdraw=50) + r = compute_freeze_split( + _q('100'), _q('0'), + FreezeConfig(DONGJIE_MODE_RATIO, decimal.Decimal('0.30'), decimal.Decimal('0'), 'p'), + _q('50'), + ) + if r.withdraw_jine != _q('50'): + failures.append('手续费基数校验: withdraw 应为 50') + + report.add( + '冻结拆分算法单元用例', + len(failures) == 0, + '\n '.join(failures) if failures else f'通过 {len(cases)} 组用例', + ) + + +def check_negative_pools(report: AuditReport): + """冻结池不得为负""" + issues = [] + for row in UserDashou.objects.filter(dongjie_chi__lt=0).values_list('user__yonghuid', 'dongjie_chi')[:10]: + issues.append(f'打手佣金池 {row[0]}={row[1]}') + for row in UserDashou.objects.filter(dongjie_chi_yajin__lt=0).values_list('user__yonghuid', 'dongjie_chi_yajin')[:10]: + issues.append(f'打手押金池 {row[0]}={row[1]}') + for row in UserGuanshi.objects.filter(dongjie_chi__lt=0).values_list('user__yonghuid', 'dongjie_chi')[:10]: + issues.append(f'管事池 {row[0]}={row[1]}') + for row in UserShangjia.objects.filter(dongjie_chi__lt=0).values_list('user__yonghuid', 'dongjie_chi')[:10]: + issues.append(f'商家池 {row[0]}={row[1]}') + report.add('冻结池无负数', len(issues) == 0, '; '.join(issues) if issues else '全部 >= 0') + + +def check_pool_vs_logs(report: AuditReport, sample_limit: int = 50): + """ + 冻结池余额 vs 流水账:对每个有流水的用户, + expected_pool = sum(申请冻结) - sum(管理员解冻) + """ + mismatches = [] + log_groups = ( + TixianDongjieJilu.objects + .values('yonghuid', 'leixing') + .annotate( + apply_sum=Sum('jine', filter=Q(caozuo=TixianDongjieJilu.CAOZUO_APPLY)), + unfreeze_sum=Sum('jine', filter=Q(caozuo=TixianDongjieJilu.CAOZUO_UNFREEZE)), + ) + ) + count = 0 + for g in log_groups: + if count >= sample_limit: + break + yid = g['yonghuid'] + lx = g['leixing'] + expected = _q(g['apply_sum'] or 0) - _q(g['unfreeze_sum'] or 0) + actual = _get_pool_balance(yid, lx) + if actual is None: + continue + if abs(actual - expected) > decimal.Decimal('0.01'): + mismatches.append(f'{yid} leixing={lx} 池={actual} 流水推算={expected} 差={actual - expected}') + count += 1 + + # 也检查有池余额但无流水的异常(可能是历史手工改库) + orphan_pools = _find_pools_without_logs(limit=10) + detail_parts = [] + if mismatches: + detail_parts.append('流水不一致: ' + '; '.join(mismatches[:5])) + if orphan_pools: + detail_parts.append('有池无流水(前10): ' + '; '.join(orphan_pools)) + report.add( + '冻结池与流水账一致', + len(mismatches) == 0, + '\n '.join(detail_parts) if detail_parts else f'抽查 {count} 组用户流水,池余额与流水一致', + severity='warn' if orphan_pools and not mismatches else 'error', + ) + + +def _get_pool_balance(yonghuid: str, leixing: int) -> Optional[decimal.Decimal]: + try: + meta = get_leixing_meta(leixing) + except ValueError: + return None + Model = meta['model'] + pool_field = meta['pool'] + try: + if leixing in (1, 5): + obj = Model.objects.get(user__yonghuid=yonghuid) + else: + obj = Model.objects.get(user__yonghuid=yonghuid) + return _q(getattr(obj, pool_field)) + except Model.DoesNotExist: + return None + + +def _find_pools_without_logs(limit: int = 10) -> List[str]: + issues = [] + for dashou in UserDashou.objects.filter(Q(dongjie_chi__gt=0) | Q(dongjie_chi_yajin__gt=0))[:limit * 2]: + yid = dashou.user.yonghuid + if dashou.dongjie_chi > 0 and not TixianDongjieJilu.objects.filter(yonghuid=yid, leixing=1).exists(): + issues.append(f'{yid} 佣金池={dashou.dongjie_chi}') + if dashou.dongjie_chi_yajin > 0 and not TixianDongjieJilu.objects.filter(yonghuid=yid, leixing=5).exists(): + issues.append(f'{yid} 押金池={dashou.dongjie_chi_yajin}') + if len(issues) >= limit: + break + return issues + + +def check_tixian_record_fields(report: AuditReport, limit: int = 200): + """有冻结的提现记录:字段关系是否合理(只读)""" + issues = [] + qs = list( + Tixianjilu.objects + .filter(benchi_dongjie_jine__gt=0) + .order_by('-id')[:limit] + ) + for t in qs: + sj = _q(t.shenqing_jine) + fee = _q(t.shouxufei) + arrive = _q(t.jine) + freeze = _q(t.benchi_dongjie_jine) + # 到账 + 手续费 应等于申请扣款额 + if abs((arrive + fee) - sj) > decimal.Decimal('0.02'): + issues.append(f'tixian#{t.id} shenqing={sj} fee={fee} arrive={arrive} 不平') + # 有冻结时申请扣款应 > 0(freeze_only 不应有 tixian 记录) + if sj <= 0 and freeze > 0: + issues.append(f'tixian#{t.id} 有冻结但 shenqing_jine=0(异常:freeze_only 不应建单)') + if freeze < 0: + issues.append(f'tixian#{t.id} benchi_dongjie_jine 为负') + + freeze_only_leak = Tixianjilu.objects.filter(shenqing_jine=0, benchi_dongjie_jine__gt=0).count() + if freeze_only_leak: + issues.append(f'疑似 freeze_only 误建单: {freeze_only_leak} 条') + + report.add( + '提现记录字段一致性', + len(issues) == 0, + '\n '.join(issues[:8]) if issues else f'抽查 {len(qs)} 条含冻结记录,字段关系正常', + ) + + +def check_audit_sync(report: AuditReport, limit: int = 100): + """自动打款:审核单与提现记录 benchi_dongjie / shenqing_jine 同步""" + issues = [] + audits = list( + TixianShenheJilu.objects + .filter(benchi_dongjie_jine__gt=0) + .order_by('-id')[:limit] + ) + for a in audits: + if not a.tixianjilu_id: + continue + t = Tixianjilu.objects.filter(pk=a.tixianjilu_id).first() + if not t: + issues.append(f'audit#{a.id} 关联 tixian 缺失') + continue + if _q(a.shenqing_jine) != _q(t.shenqing_jine): + issues.append(f'audit#{a.id}/tixian#{t.id} shenqing_jine 不一致') + if _q(a.benchi_dongjie_jine) != _q(t.benchi_dongjie_jine): + issues.append(f'audit#{a.id}/tixian#{t.id} benchi_dongjie_jine 不一致') + + report.add( + '自动打款审核单与提现记录同步', + len(issues) == 0, + '\n '.join(issues[:8]) if issues else f'抽查 {len(audits)} 条审核单', + ) + + +def check_code_wiring(report: AuditReport): + """接口/模块是否接入(只读 import)""" + issues = [] + try: + from django.urls import get_resolver + urls = {p.name for p in get_resolver().url_patterns if hasattr(p, 'name') and p.name} + # 递归找 houtai + resolver = get_resolver() + houtai_found = False + for pattern in resolver.url_patterns: + if hasattr(pattern, 'url_patterns') and 'houtai' in str(pattern.pattern): + for sub in pattern.url_patterns: + if 'hqdjdongjie' in str(sub.pattern): + houtai_found = True + if not houtai_found: + issues.append('未找到 /houtai/hqdjdongjie 路由') + except Exception as e: + issues.append(f'URL检查异常: {e}') + + modules = [ + 'yonghu.freeze_service', + 'houtai.dongjie_views', + ] + for mod in modules: + try: + __import__(mod) + except Exception as e: + issues.append(f'无法导入 {mod}: {e}') + + # 手动/自动是否都接 freeze + import inspect + from yonghu import views as yv + from yonghu import tixian_shenhe_services as ts + if 'apply_balance_freeze' not in inspect.getsource(ts.create_audit_application): + issues.append('自动打款 create_audit_application 未引用 apply_balance_freeze') + if '_submit_manual_withdraw_with_freeze' not in inspect.getsource(yv.TixianShenqingView): + issues.append('手动打款未找到 _submit_manual_withdraw_with_freeze') + + report.add( + '代码接入(自动+手动+后台API)', + len(issues) == 0, + '; '.join(issues) if issues else 'freeze_service / hqdjdongjie / 双通道已接入', + ) + + +def check_final_business_rules_doc(report: AuditReport): + """ + 对照「最终确认」业务规则(与用户线上一致,非 md 初稿第六节驳回回滚冻结) + """ + rules = [ + '支持 leixing 1/2/5/6', + '手动+自动打款共用冻结', + '比例:先切冻结再对剩余扣手续费提现', + '固定填不满池:只入池不建审核单', + '驳回:仅退 shenqing_jine,冻结池不回滚', + '管理员解冻:池→余额', + ] + impl_notes = [] + impl_notes.append(f'FREEZE_LEIXING_SET={sorted(FREEZE_LEIXING_SET)}') + impl_notes.append('驳回逻辑: refund_balance(shenqing_jine) 无 benchi 回滚 — 见 tixian_shenhe_services.refund_balance') + impl_notes.append('流水 caozuo 仅 1申请冻结/3管理员解冻,无驳回回滚类型(符合最终规则)') + + # 验证驳回代码确实不回滚冻结 + import inspect + from yonghu.tixian_shenhe_services import refund_balance, reject_audit_and_refund + src_reject = inspect.getsource(reject_audit_and_refund) + src_refund = inspect.getsource(refund_balance) + no_rollback = 'benchi_dongjie' not in src_reject and 'dongjie_chi' not in src_refund + report.add( + '最终业务规则: 驳回不退冻结池', + no_rollback, + '\n '.join(rules + ['---'] + impl_notes), + severity='error' if not no_rollback else 'info', + ) + + +def check_global_switch_default(report: AuditReport): + """总开关默认关闭,不影响未开启时的存量用户""" + from peizhi.models import WithdrawConfig + wc = WithdrawConfig.objects.filter(id=1).first() + enabled = bool(wc.dongjie_quanju_enabled) if wc else False + active_roles = TixianDongjieGonggongPeizhi.objects.filter(role_enabled=True).count() + report.add( + '当前冻结开关状态', + True, + f'全员开关={enabled}, 已启用角色数={active_roles}(只读展示,非错误)', + severity='info', + ) + + +class Command(BaseCommand): + help = '资金冻结功能只读验收(不修改任何数据)' + + def add_arguments(self, parser): + parser.add_argument('--verbose', action='store_true', help='输出详细信息') + parser.add_argument('--sample-users', type=int, default=50, help='流水池一致性抽查用户数上限') + + def handle(self, *args, **options): + verbose = options['verbose'] + sample = options['sample_users'] + report = AuditReport() + + self.stdout.write(self.style.NOTICE('开始只读验收(不会写入任何业务数据)...\n')) + + checks = [ + lambda: check_schema(report), + lambda: check_public_config_seed(report), + lambda: check_algorithm_unit_tests(report), + lambda: check_code_wiring(report), + lambda: check_final_business_rules_doc(report), + lambda: check_global_switch_default(report), + lambda: check_negative_pools(report), + lambda: check_pool_vs_logs(report, sample), + lambda: check_tixian_record_fields(report), + lambda: check_audit_sync(report), + ] + + for fn in checks: + try: + fn() + except Exception as e: + report.add(fn.__name__, False, f'{e}\n{traceback.format_exc()[-300:]}') + + report.print_report(verbose) + + if not report.ok: + sys.exit(1)