code/ai_finance_agent/backend/app/domain/rules.py

"""可解释规则只处理显式字段与上下文，绝不从缺失数据推测金额或法定税率。"""
from datetime import date, timedelta
from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
from typing import Any
from app.shared.common import cents

CORE_RULES = {"invoice_date_valid", "invoice_amount_valid", "invoice_header_valid", "invoice_no_duplicate"}


def result(rule: dict, passed: bool, message: str, severity: str | None = None,
           details: dict | None = None, execution_status: str = "success") -> dict:
    return {"rule_code": rule["code"], "rule_version": rule.get("version", 1), "passed": passed,
            "message": message, "severity": severity or rule.get("severity", "warning"),
            "details": details or {}, "execution_status": execution_status}


def check_rule(rule: dict, invoice: dict, context: dict) -> dict:
    """每个分支保留实际值和阈值，历史记录能解释为什么触发，而非只留布尔值。"""
    code, params = rule["code"], rule.get("params", {})
    if not rule.get("enabled", True):
        return result(rule, True, "管理员停用了可选检查", execution_status="not_applicable")
    scope = rule.get("scope", {})
    if scope.get("types") and context.get("type") not in scope["types"] or scope.get("departments") and context.get("department_id") not in scope["departments"]:
        return result(rule, True, "不在该规则适用范围", execution_status="not_applicable")
    if code == "invoice_date_valid":
        value = invoice.get("invoice_date")
        if not value:
            return result(rule, False, "发票日期缺失，无法确认有效期", "error")
        try:
            actual = date.fromisoformat(str(value))
        except (ValueError, TypeError):
            return result(rule, False, "日期格式必须为 YYYY-MM-DD", "error")
        today = date.fromisoformat(context["today"]) if context.get("today") else date.today()
        if actual > today:
            return result(rule, False, "发票日期晚于审核业务日期", "error", {"actual": value, "today": str(today)})
        age = (today - actual).days
        if age > params.get("max_age_days", 365):
            return result(rule, False, "发票超过配置的有效期，需复核", "warning", {"age_days": age, "limit": params.get("max_age_days", 365)})
        return result(rule, True, "发票日期未触发异常", details={"age_days": age})
    if code == "invoice_amount_valid":
        try:
            amount = cents(invoice.get("amount"), False)
            tax = cents(invoice.get("tax_amount"))
        except Exception:
            return result(rule, False, "金额/税额缺失或精度非法", "error")
        if amount > cents(params.get("large_amount", "100000.00")):
            return result(rule, False, "单张发票超过大额提醒阈值", "warning", {"actual": invoice["amount"], "limit": params.get("large_amount")})
        return result(rule, True, "金额为正且税额精度有效", details={"net_cents": amount, "tax_cents": tax})
    if code == "invoice_header_valid":
        if not str(invoice.get("buyer_name") or "").strip() or not str(invoice.get("seller_name") or "").strip():
            return result(rule, False, "买方或卖方抬头缺失", "error")
        company = params.get("company_name", "").strip()
        if company and invoice["buyer_name"].strip() != company:
            return result(rule, False, "买方名称与配置企业全称不同", "error", {"actual": invoice["buyer_name"], "expected": company})
        return result(rule, True, "抬头完整" + ("且符合企业全称" if company else "；未配置企业全称，仅核对完整性"))
    if code == "invoice_no_duplicate":
        file_hash = invoice.get("file_hash")
        if not file_hash:
            return result(rule, False, "缺少文件指纹，无法完成重复校验", "error")
        duplicate = file_hash in context.get("existing_hashes", []) or context.get("same_hash_count", {}).get(file_hash, 0) > 1
        return result(rule, not duplicate, "发现相同文件内容的有效报销" if duplicate else "未发现同内容重复", "critical", {"file_hash": file_hash, "method": "sha256_content_only"})
    if code == "tax_calculation_valid":
        if invoice.get("tax_rate") is None or invoice.get("amount") is None or invoice.get("tax_amount") is None:
            return result(rule, False, "未填税率/金额/税额，税额待人工核验", "warning", execution_status="skipped")
        try:
            rate = Decimal(str(invoice["tax_rate"]))
            if not rate.is_finite() or not 0 <= rate <= 1:
                raise ValueError()
            expected = (Decimal(str(invoice["amount"])) * rate).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
            actual = Decimal(str(invoice["tax_amount"]))
            passed = abs(expected - actual) <= Decimal(str(params.get("tolerance", "0.01")))
            return result(rule, passed, "税额符合填报税率" if passed else "税额与填报税率计算不符", details={"expected": str(expected), "actual": str(actual), "tax_rate": str(rate), "tolerance": params.get("tolerance", "0.01")})
        except (InvalidOperation, ValueError, TypeError):
            return result(rule, False, "税率或金额非法，不能进行计算")
    if code == "tax_number_valid":
        value = str(invoice.get("tax_number") or "")
        valid = len(value) in params.get("lengths", [15, 18, 20]) and value.isalnum() and value.isascii()
        return result(rule, valid, "税号格式符合示例长度规则，未进行税务验真" if valid else "税号为空或长度/字符不符合配置", details={"length": len(value), "allowed_lengths": params.get("lengths", [15, 18, 20])})
    raise ValueError("政策规则必须经检索服务执行")


def compliance(results: list[dict]) -> str:
    """critical 与 error 一样失败；跳过和未知检查不能被误汇总为全通过。"""
    if not results:
        return "fail"
    if any(not item["passed"] and item["severity"] in {"error", "critical"} for item in results):
        return "fail"
    if any(not item["passed"] or item.get("execution_status") in {"skipped", "failed", "partial"} for item in results):
        return "partial"
    return "pass"


def risk_score(invoices: list[dict], context: dict, config: dict) -> dict:
    """整数权重避免 0.3/0.6 边界的浮点累加误差；缺统计样本不加虚构分数。"""
    amounts = [cents(item["amount"]) + cents(item["tax_amount"]) for item in invoices if item.get("amount") is not None and item.get("tax_amount") is not None]
    total = sum(amounts)
    confidences = [item.get("confidence", 0) for item in invoices]
    sellers = [str(item.get("seller_name") or "").strip() for item in invoices if item.get("seller_name")]
    concentration = max([sellers.count(x) / len(sellers) for x in set(sellers)], default=0)
    minimum = min(confidences, default=0)
    dept_average = context.get("dept_average_cents")
    average_decimal = Decimal(str(dept_average)) if dept_average is not None else None
    ratio = float(Decimal(total) / average_decimal) if average_decimal is not None and average_decimal > 0 else None
    features = {"total_amount": str(Decimal(total) / 100), "invoice_count": len(invoices),
        "avg_amount": str(Decimal(total) / max(len(amounts), 1) / 100), "max_single_amount": str(Decimal(max(amounts, default=0)) / 100),
        "avg_confidence": sum(confidences) / max(len(confidences), 1), "min_confidence": minimum,
        "seller_concentration": concentration, "type_diversity": len(set(item.get("invoice_type") for item in invoices)),
        "user_submission_frequency": context.get("recent_submissions"), "dept_avg_amount_ratio": ratio,
        "statistic_status": "available" if dept_average else "insufficient_data", "window_days": config.get("window_days", 30)}
    score, reasons = 0, []
    if total > cents(config.get("amount_high", "50000.00")):
        score += 40
        reasons.append("含税总额超过高额阈值 +0.40")
    elif total > cents(config.get("amount_medium", "20000.00")):
        score += 20
        reasons.append("含税总额超过中额阈值 +0.20")
    if minimum < 0.6:
        score += 30
        reasons.append("提取置信度标记低于0.60 +0.30")
    elif minimum < 0.8:
        score += 15
        reasons.append("提取置信度标记低于0.80 +0.15")
    if concentration > 0.8 and len(invoices) > 3:
        score += 20
        reasons.append("多张发票卖方高度集中 +0.20")
    if context.get("recent_submissions") is not None and context["recent_submissions"] > config.get("frequency_limit", 10):
        score += 15
        reasons.append("近期实际提交次数偏高 +0.15")
    if ratio is not None and ratio > config.get("dept_ratio_limit", 3):
        score += 20
        reasons.append("超过部门历史平均倍数阈值 +0.20")
    return {"level": 3 if score >= 60 else 2 if score >= 30 else 1, "score": str(Decimal(min(score, 100)) / 100),
            "features": features, "reasons": reasons or ["未触发规则风险阈值"]}


def decide(state: dict) -> dict:
    """路由只读取规范化结果；没有依据或上游未评估时不能使用低风险默认值。"""
    invoices = state.get("extracted_invoices", [])
    missing = not invoices or any(item.get("amount") is None or not item.get("invoice_date") for item in invoices)
    level, status = state.get("risk_level"), state.get("compliance_status")
    if missing or level not in {1, 2, 3} or status not in {"pass", "partial", "fail"} or state.get("core_failure"):
        decision, reason = "to_manual", "关键依据缺失或核心环节未完成，转人工复核"
    elif level == 3 or status == "fail":
        decision, reason = "to_finance", "高风险或存在严重合规失败，转财务审核"
    elif level == 2 or status == "partial":
        decision, reason = "to_supervisor", "中风险或部分检查待核验，转主管复核"
    else:
        decision, reason = "auto_approve", "明确低风险且必需检查全部通过，自动批准"
    return {"decision": decision, "decision_reason": reason}