along_django/yonghu/management/commands/audit_freeze_feature.py

"""
资金冻结功能 — 只读验收脚本（不修改任何业务数据）

用法（生产环境）：
    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_RATIO,
    FREEZE_LEIXING_SET,
    FreezeConfig,
    compute_freeze_split,
    get_leixing_meta,
)
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 = [
        ('无配置', '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}')

    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 流水账"""
    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:
        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} 不平')
        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):
    """自动打款：审核单与提现记录同步"""
    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:
        resolver = __import__('django.urls', fromlist=['get_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}')

    for mod in ('yonghu.freeze_service', 'houtai.dongjie_views'):
        try:
            __import__(mod)
        except Exception as e:
            issues.append(f'无法导入 {mod}: {e}')

    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):
    """对照最终确认业务规则"""
    rules = [
        '支持 leixing 1/2/5/6',
        '手动+自动打款共用冻结',
        '比例：先切冻结再对剩余扣手续费提现',
        '固定填不满池：只入池不建审核单',
        '驳回：仅退 shenqing_jine，冻结池不回滚',
        '管理员解冻：池→余额',
    ]
    impl_notes = [
        f'FREEZE_LEIXING_SET={sorted(FREEZE_LEIXING_SET)}',
        '驳回逻辑: refund_balance(shenqing_jine) 无 benchi 回滚',
        '流水 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)