2026 技术观察：智能财务共享进入自动审核阶段，发票识别、费用合规和异常报销成为新重点

import json
from datetime import datetime
from collections import defaultdict


EXPENSE_CLAIMS = [
    {
        "claim_id": "C001",
        "employee_id": "E001",
        "department": "销售部",
        "expense_type": "travel",
        "amount": 860,
        "invoice_no": "INV20260701001",
        "invoice_type": "交通票据",
        "submit_date": "2026-07-01"
    },
    {
        "claim_id": "C002",
        "employee_id": "E002",
        "department": "市场部",
        "expense_type": "entertainment",
        "amount": 3200,
        "invoice_no": "INV20260701002",
        "invoice_type": "餐饮发票",
        "submit_date": "2026-07-02"
    },
    {
        "claim_id": "C003",
        "employee_id": "E001",
        "department": "销售部",
        "expense_type": "office",
        "amount": 150,
        "invoice_no": "INV20260701001",
        "invoice_type": "办公用品发票",
        "submit_date": "2026-07-03"
    }
]


EXPENSE_LIMITS = {
    "travel": 1200,
    "entertainment": 2000,
    "office": 500,
    "training": 3000
}

def detect_duplicate_invoice(claims):
    invoice_map = defaultdict(list)

    for claim in claims:
        invoice_map[claim["invoice_no"]].append(
            claim["claim_id"]
        )

    duplicates = {}

    for invoice_no, claim_ids in invoice_map.items():
        if len(claim_ids) > 1:
            duplicates[invoice_no] = claim_ids

    return duplicates

def check_expense_limit(claim):
    limit = EXPENSE_LIMITS.get(
        claim["expense_type"],
        0
    )

    if limit == 0:
        return {
            "status": "unknown",
            "risk_score": 2,
            "message": "费用类型未配置标准。"
        }

    if claim["amount"] > limit:
        return {
            "status": "over_limit",
            "risk_score": 4,
            "message": "报销金额超过费用标准。"
        }

    if claim["amount"] > limit * 0.85:
        return {
            "status": "near_limit",
            "risk_score": 2,
            "message": "报销金额接近费用上限。"
        }

    return {
        "status": "normal",
        "risk_score": 0,
        "message": "报销金额符合标准。"
    }

EXPECTED_INVOICE_TYPE = {
    "travel": ["交通票据", "住宿发票"],
    "entertainment": ["餐饮发票"],
    "office": ["办公用品发票"],
    "training": ["培训发票"]
}


def check_invoice_type_match(claim):
    expected_types = EXPECTED_INVOICE_TYPE.get(
        claim["expense_type"],
        []
    )

    if claim["invoice_type"] not in expected_types:
        return {
            "matched": False,
            "risk_score": 3,
            "message": "发票类型与费用类型不匹配。"
        }

    return {
        "matched": True,
        "risk_score": 0,
        "message": "发票类型匹配。"
    }

def analyze_employee_claim_frequency(claims):
    employee_count = defaultdict(int)

    for claim in claims:
        employee_count[claim["employee_id"]] += 1

    results = {}

    for employee_id, count in employee_count.items():
        if count >= 3:
            results[employee_id] = {
                "risk_level": "medium",
                "message": "员工短期报销频率较高，建议关注。"
            }
        else:
            results[employee_id] = {
                "risk_level": "normal",
                "message": "员工报销频率正常。"
            }

    return results

def evaluate_claim(claim, duplicates, frequency_result):
    risk_score = 0
    issues = []

    if claim["invoice_no"] in duplicates:
        risk_score += 5
        issues.append("该发票号码存在重复报销风险。")

    limit_result = check_expense_limit(claim)
    type_result = check_invoice_type_match(claim)
    freq_result = frequency_result.get(
        claim["employee_id"],
        {}
    )

    risk_score += limit_result["risk_score"]
    risk_score += type_result["risk_score"]

    if freq_result.get("risk_level") == "medium":
        risk_score += 1
        issues.append(freq_result["message"])

    if limit_result["status"] != "normal":
        issues.append(limit_result["message"])

    if not type_result["matched"]:
        issues.append(type_result["message"])

    if risk_score >= 7:
        decision = "manual_review"
    elif risk_score >= 3:
        decision = "need_check"
    else:
        decision = "auto_pass"

    return {
        "claim_id": claim["claim_id"],
        "employee_id": claim["employee_id"],
        "department": claim["department"],
        "amount": claim["amount"],
        "risk_score": risk_score,
        "decision": decision,
        "issues": 30503.t.kuaisou.com 
        "evaluate_time": datetime.now().isoformat()
    }

def run_finance_shared_review():
    duplicates = detect_duplicate_invoice(
        EXPENSE_CLAIMS
    )

    frequency_result = analyze_employee_claim_frequency(
        EXPENSE_CLAIMS
    )

    review_results = []

    for claim in EXPENSE_CLAIMS:
        review_results.append(
            evaluate_claim(
                claim,
                duplicates,
                frequency_result
            )
        )

    decision_count = defaultdict(int)

    for result in review_results:
        decision_count[result["decision"]] += 1

    report = {
        "report_name": "智能财务共享费用审核报告",
        "duplicate_invoices": duplicates,
        "decision_count": dict(decision_count),
        "review_results": 30523.t.kuaisou.com 
        "generate_time": datetime.now().isoformat()
    }

    return report


if __name__ == "__main__":
    report = run_finance_shared_review()

    print(json.dumps(
        report,
        ensure_ascii=False,
        indent=2
    ))