"""可解释规则只处理显式字段与上下文，绝不从缺失数据推测金额或法定税率。""" 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}