新增资金冻结只读验收脚本 audit_freeze_feature，不修改任何业务数据。

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

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)