X-Financial/server/tests/test_expense_claim_service.py

from __future__ import annotations

import re
import uuid
from datetime import UTC, date, datetime, timedelta
from decimal import Decimal

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.pool import StaticPool

from app.api.deps import CurrentUserContext
from app.db.base import Base
from app.models.budget import BudgetAllocation, BudgetReservation, BudgetTransaction
from app.models.employee import Employee
from app.models.financial_record import ExpenseClaim, ExpenseClaimItem
from app.models.organization import OrganizationUnit
from app.schemas.ontology import OntologyParseRequest
from app.schemas.ocr import OcrRecognizeBatchRead, OcrRecognizeDocumentRead
from app.schemas.reimbursement import ExpenseClaimItemCreate, ExpenseClaimItemUpdate, ExpenseClaimUpdate
from app.services.agent_conversations import AgentConversationService
from app.services.expense_claim_attachment_storage import ExpenseClaimAttachmentStorage
from app.services.expense_claims import ExpenseClaimService
from app.services.ontology import SemanticOntologyService
from app.services.ocr import OcrService


def build_claim(*, expense_type: str, location: str) -> ExpenseClaim:
    claim = ExpenseClaim(
        id="claim-1",
        claim_no="EXP-202605-001",
        employee_id="emp-1",
        employee_name="张三",
        department_id="dept-1",
        department_name="市场部",
        project_code=None,
        expense_type=expense_type,
        reason="费用报销",
        location=location,
        amount=Decimal("88.00"),
        currency="CNY",
        invoice_count=1,
        occurred_at=datetime(2026, 5, 13, tzinfo=UTC),
        submitted_at=None,
        status="draft",
        approval_stage="待提交",
        risk_flags_json=[],
    )
    claim.items = [
        ExpenseClaimItem(
            id="item-1",
            claim_id="claim-1",
            item_date=date(2026, 5, 13),
            item_type=expense_type,
            item_reason="费用报销",
            item_location=location,
            item_amount=Decimal("88.00"),
            invoice_id="invoice-1",
        )
    ]
    return claim


def build_session() -> Session:
    engine = create_engine(
        "sqlite+pysqlite:///:memory:",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    Base.metadata.create_all(bind=engine)
    session_factory = sessionmaker(bind=engine, autoflush=False, autocommit=False)
    return session_factory()


def _count_claims(db: Session) -> int:
    return int(db.query(ExpenseClaim).count())


def _seed_budget_allocation(
    db: Session,
    *,
    department_id: str | None,
    department_name: str,
    subject_code: str = "travel",
    amount: Decimal = Decimal("50000.00"),
    period_key: str = "2026Q2",
) -> BudgetAllocation:
    allocation = BudgetAllocation(
        budget_no=f"BUD-TEST-{uuid.uuid4().hex[:8]}",
        fiscal_year=2026,
        period_type="quarter",
        period_key=period_key,
        department_id=department_id,
        department_name=department_name,
        cost_center=None,
        project_code=None,
        subject_code=subject_code,
        subject_name=subject_code,
        original_amount=amount,
        adjusted_amount=Decimal("0.00"),
        status="active",
        warning_threshold=Decimal("80.00"),
        control_action="block",
    )
    db.add(allocation)
    db.commit()
    return allocation


def test_validate_claim_for_submission_allows_office_claim_without_location() -> None:
    service = ExpenseClaimService.__new__(ExpenseClaimService)
    claim = build_claim(expense_type="office", location="待补充")

    issues = service._validate_claim_for_submission(claim)

    assert "业务地点未完善" not in issues
    assert not any("缺少地点" in item for item in issues)


def test_validate_claim_for_submission_allows_transport_claim_without_location() -> None:
    service = ExpenseClaimService.__new__(ExpenseClaimService)
    claim = build_claim(expense_type="transport", location="待补充")

    issues = service._validate_claim_for_submission(claim)

    assert "业务地点未完善" not in issues
    assert not any("缺少地点" in item for item in issues)


def test_validate_claim_for_submission_still_requires_location_for_travel_claim() -> None:
    service = ExpenseClaimService.__new__(ExpenseClaimService)
    claim = build_claim(expense_type="travel", location="待补充")

    issues = service._validate_claim_for_submission(claim)

    assert "业务地点未完善" in issues
    assert any("缺少地点" in item for item in issues)


def test_save_or_submit_preview_does_not_create_claim_without_explicit_action() -> None:
    user_id = "preview-only@example.com"
    message = "业务发生时间:2026-03-04，打车去客户现场，交通费32元，请帮我看看怎么报"

    with build_session() as db:
        employee = Employee(
            employee_no="E5100",
            name="预览员工",
            email=user_id,
        )
        db.add(employee)
        db.commit()
        ontology = SemanticOntologyService(db).parse(
            OntologyParseRequest(
                query=message,
                user_id=user_id,
            )
        )
        before_count = _count_claims(db)

        result = ExpenseClaimService(db).save_or_submit_from_ontology(
            run_id=ontology.run_id,
            user_id=user_id,
            message=message,
            ontology=ontology,
            context_json={
                "name": "预览员工",
                "user_input_text": message,
            },
        )

        assert result["preview_only"] is True
        assert result["status"] == "preview"
        assert "报销测算参考：" in result["message"]
        assert "| 项目 | 当前信息 | 复核口径 |" in result["message"]
        assert "交通票据金额 + 住宿标准" not in result["message"]
        assert _count_claims(db) == before_count


def test_save_or_submit_persists_claim_only_after_save_draft_action() -> None:
    user_id = "save-draft-explicit@example.com"
    message = "业务发生时间:2026-03-04，打车去客户现场，交通费32元，请帮我看看怎么报"

    with build_session() as db:
        employee = Employee(
            employee_no="E5101",
            name="保存员工",
            email=user_id,
        )
        db.add(employee)
        db.commit()
        ontology = SemanticOntologyService(db).parse(
            OntologyParseRequest(
                query=message,
                user_id=user_id,
            )
        )
        before_count = _count_claims(db)

        result = ExpenseClaimService(db).save_or_submit_from_ontology(
            run_id=ontology.run_id,
            user_id=user_id,
            message=message,
            ontology=ontology,
            context_json={
                "name": "保存员工",
                "user_input_text": message,
                "review_action": "save_draft",
            },
        )

        assert result["draft_only"] is True
        assert result["claim_id"]
        assert result["status"] == "draft"
        assert _count_claims(db) == before_count + 1


def test_save_draft_persists_user_changed_expense_category() -> None:
    user_id = "save-draft-category@example.com"
    message = "业务发生时间:2026-03-04，打车去客户现场，交通费32元，请帮我看看怎么报"

    with build_session() as db:
        employee = Employee(
            employee_no="E5102",
            name="分类员工",
            email=user_id,
        )
        db.add(employee)
        db.commit()
        ontology = SemanticOntologyService(db).parse(
            OntologyParseRequest(
                query=message,
                user_id=user_id,
            )
        )

        result = ExpenseClaimService(db).save_or_submit_from_ontology(
            run_id=ontology.run_id,
            user_id=user_id,
            message=message,
            ontology=ontology,
            context_json={
                "name": "分类员工",
                "user_input_text": message,
                "review_action": "save_draft",
                "review_form_values": {
                    "expense_type": "办公用品费",
                    "amount": "32元",
                    "occurred_date": "2026-03-04",
                    "reason": "右侧核对后改为办公用品费",
                },
            },
        )

        claim = db.get(ExpenseClaim, result["claim_id"])
        assert claim is not None
        assert claim.expense_type == "office"
        assert claim.items[0].item_type == "office"


def test_unsaved_conversation_expires_after_retention_but_saved_conversation_stays() -> None:
    with build_session() as db:
        service = AgentConversationService(db)
        unsaved = service.get_or_create_conversation(
            conversation_id="conv-unsaved-expire",
            user_id="expire@example.com",
            source="user_message",
            context_json={"session_type": "expense"},
        )
        saved = service.get_or_create_conversation(
            conversation_id="conv-saved-keep",
            user_id="expire@example.com",
            source="user_message",
            context_json={
                "session_type": "expense",
                "draft_claim_id": "claim-saved",
            },
        )
        old_time = datetime.now(UTC) - timedelta(days=4)
        unsaved.updated_at = old_time
        saved.updated_at = old_time
        db.add_all([unsaved, saved])
        db.commit()

        deleted_count = service.prune_expired_conversations(retention_days=3)

        assert deleted_count == 1
        assert service.get_conversation("conv-unsaved-expire") is None
        assert service.get_conversation("conv-saved-keep") is not None


def test_resolve_expense_type_maps_office_supplies_review_value_to_office() -> None:
    expense_type = ExpenseClaimService._resolve_expense_type(
        [],
        context_json={
            "review_form_values": {
                "expense_type": "办公用品"
            }
        },
    )

    assert expense_type == "office"


def test_resolve_expense_type_maps_riding_fare_review_value_to_transport() -> None:
    expense_type = ExpenseClaimService._resolve_expense_type(
        [],
        context_json={
            "review_form_values": {
                "expense_type": "乘车费用"
            }
        },
    )

    assert expense_type == "transport"


def test_upsert_draft_from_ontology_defers_multi_document_association_choice() -> None:
    user_id = "zhangsan@example.com"

    with build_session() as db:
        employee = Employee(
            employee_no="E5001",
            name="张三",
            email=user_id,
        )
        db.add(employee)
        db.flush()
        existing_claim = ExpenseClaim(
            claim_no="EXP-202605-010",
            employee_id=employee.id,
            employee_name="张三",
            department_name="市场部",
            project_code=None,
            expense_type="transport",
            reason="原有交通报销",
            location="深圳",
            amount=Decimal("20.00"),
            currency="CNY",
            invoice_count=1,
            occurred_at=datetime(2026, 5, 13, tzinfo=UTC),
            status="draft",
            approval_stage="待提交",
            risk_flags_json=[],
        )
        existing_claim.items = [
            ExpenseClaimItem(
                claim_id=existing_claim.id,
                item_date=date(2026, 5, 13),
                item_type="transport",
                item_reason="原有交通报销",
                item_location="深圳",
                item_amount=Decimal("20.00"),
                invoice_id="old-trip.png",
            )
        ]
        db.add(existing_claim)
        db.commit()

        ontology = SemanticOntologyService(db).parse(
            OntologyParseRequest(
                query="我上传了两张交通票据，帮我生成报销草稿",
                user_id=user_id,
            )
        )
        service = ExpenseClaimService(db)
        result = service.upsert_draft_from_ontology(
            run_id=ontology.run_id,
            user_id=user_id,
            message="我上传了两张交通票据，帮我生成报销草稿",
            ontology=ontology,
            context_json={
                "name": "张三",
                "attachment_names": ["didi-trip.png", "parking-ticket.jpg"],
                "attachment_count": 2,
                "draft_claim_id": existing_claim.id,
                "ocr_documents": [
                    {
                        "filename": "didi-trip.png",
                        "summary": "滴滴出行 支付金额 32 元",
                        "text": "滴滴出行 支付金额 32 元",
                    },
                    {
                        "filename": "parking-ticket.jpg",
                        "summary": "停车费 合计 18 元",
                        "text": "停车费 合计 18 元",
                    },
                ],
            },
        )

        db.refresh(existing_claim)
        assert result["pending_association_decision"] is True
        assert result["association_candidate_claim_id"] == existing_claim.id
        assert existing_claim.invoice_count == 1
        assert len(existing_claim.items) == 1
        assert existing_claim.items[0].invoice_id == "old-trip.png"


def test_linked_document_supplement_keeps_existing_claim_expense_type() -> None:
    user_id = "type-lock@example.com"

    with build_session() as db:
        employee = Employee(
            employee_no="E5010",
            name="类型锁定员工",
            email=user_id,
        )
        db.add(employee)
        db.flush()
        existing_claim = ExpenseClaim(
            claim_no="EXP-202605-020",
            employee_id=employee.id,
            employee_name="类型锁定员工",
            department_name="市场部",
            project_code=None,
            expense_type="transport",
            reason="原有交通报销",
            location="深圳",
            amount=Decimal("32.00"),
            currency="CNY",
            invoice_count=1,
            occurred_at=datetime(2026, 5, 13, tzinfo=UTC),
            status="draft",
            approval_stage="待提交",
            risk_flags_json=[],
        )
        existing_claim.items = [
            ExpenseClaimItem(
                claim_id=existing_claim.id,
                item_date=date(2026, 5, 13),
                item_type="transport",
                item_reason="原有交通报销",
                item_location="深圳",
                item_amount=Decimal("32.00"),
                invoice_id="old-trip.png",
            )
        ]
        db.add(existing_claim)
        db.commit()

        context_json = {
            "name": "类型锁定员工",
            "review_action": "link_to_existing_draft",
            "draft_claim_id": existing_claim.id,
            "attachment_names": ["hotel-invoice.pdf"],
            "attachment_count": 1,
            "ocr_documents": [
                {
                    "filename": "hotel-invoice.pdf",
                    "document_type": "hotel_invoice",
                    "scene_code": "hotel",
                    "scene_label": "住宿票据",
                    "summary": "酒店住宿 发票金额 300 元",
                    "text": "酒店住宿 发票金额 ￥300.00",
                    "document_fields": [
                        {"key": "amount", "label": "金额", "value": "300"},
                        {"key": "merchant", "label": "酒店名称", "value": "上海酒店"},
                    ],
                }
            ],
        }
        ontology = SemanticOntologyService(db).parse(
            OntologyParseRequest(
                query="把酒店发票补充到现有草稿",
                user_id=user_id,
                context_json=context_json,
            )
        )
        ExpenseClaimService(db).upsert_draft_from_ontology(
            run_id=ontology.run_id,
            user_id=user_id,
            message="把酒店发票补充到现有草稿",
            ontology=ontology,
            context_json=context_json,
        )

        db.refresh(existing_claim)
        assert existing_claim.expense_type == "transport"
        assert any(item.item_type == "hotel_ticket" for item in existing_claim.items)


def test_upsert_draft_from_ontology_keeps_reason_missing_for_attachment_only_upload() -> None:
    user_id = "wangwu@example.com"

    with build_session() as db:
        employee = Employee(
            employee_no="E5003",
            name="王五",
            email=user_id,
        )
        db.add(employee)
        db.commit()

        ontology = SemanticOntologyService(db).parse(
            OntologyParseRequest(
                query="我上传了 1 份票据，请结合附件名称给出报销建议并尽量生成草稿。",
                user_id=user_id,
            )
        )
        service = ExpenseClaimService(db)
        result = service.upsert_draft_from_ontology(
            run_id=ontology.run_id,
            user_id=user_id,
            message="我上传了 1 份票据，请结合附件名称给出报销建议并尽量生成草稿。\n附件名称：didi-trip.png",
            ontology=ontology,
            context_json={
                "name": "王五",
                "user_input_text": "",
                "attachment_names": ["didi-trip.png"],
                "attachment_count": 1,
                "ocr_documents": [
                    {
                        "filename": "didi-trip.png",
                        "summary": "滴滴出行 支付金额 32 元",
                        "text": "滴滴出行 支付金额 32 元",
                        "document_type": "taxi_receipt",
                        "scene_code": "transport",
                    }
                ],
            },
        )

        claim = db.get(ExpenseClaim, result["claim_id"])
        assert claim is not None
        assert claim.reason == "待补充"


def test_upsert_draft_from_ontology_strips_recognized_business_time_from_reason() -> None:
    user_id = "transport-time@example.com"
    message = "业务发生时间:2026-03-04，送客户去林萃小区办事，请报销乘车费用"

    with build_session() as db:
        employee = Employee(
            employee_no="E5004",
            name="赵六",
            email=user_id,
        )
        db.add(employee)
        db.commit()

        ontology = SemanticOntologyService(db).parse(
            OntologyParseRequest(
                query=message,
                user_id=user_id,
            )
        )
        result = ExpenseClaimService(db).upsert_draft_from_ontology(
            run_id=ontology.run_id,
            user_id=user_id,
            message=message,
            ontology=ontology,
            context_json={
                "name": "赵六",
                "user_input_text": message,
            },
        )

        claim = db.get(ExpenseClaim, result["claim_id"])
        assert claim is not None
        assert claim.occurred_at.date() == date(2026, 3, 4)
        assert claim.reason == "送客户去林萃小区办事，请报销乘车费用"
        assert len(claim.items) == 1
        assert claim.items[0].item_date == date(2026, 3, 4)
        assert claim.items[0].item_reason == "送客户去林萃小区办事，请报销乘车费用"
        assert "客户单位" not in result["message"]
        assert "票据附件" not in result["message"]
        assert "费用明细" not in result["message"]


def test_upsert_draft_from_ontology_supports_link_or_create_for_multi_documents() -> None:
    user_id = "lisi@example.com"

    with build_session() as db:
        employee = Employee(
            employee_no="E5002",
            name="李四",
            email=user_id,
        )
        db.add(employee)
        db.flush()
        existing_claim = ExpenseClaim(
            claim_no="EXP-202605-011",
            employee_id=employee.id,
            employee_name="李四",
            department_name="销售部",
            project_code=None,
            expense_type="transport",
            reason="原有交通报销",
            location="上海",
            amount=Decimal("20.00"),
            currency="CNY",
            invoice_count=1,
            occurred_at=datetime(2026, 5, 13, tzinfo=UTC),
            status="draft",
            approval_stage="待提交",
            risk_flags_json=[],
        )
        existing_claim.items = [
            ExpenseClaimItem(
                claim_id=existing_claim.id,
                item_date=date(2026, 5, 13),
                item_type="transport",
                item_reason="原有交通报销",
                item_location="上海",
                item_amount=Decimal("20.00"),
                invoice_id="existing.png",
            )
        ]
        db.add(existing_claim)
        db.commit()

        ontology = SemanticOntologyService(db).parse(
            OntologyParseRequest(
                query="我上传了两张交通票据，帮我生成报销草稿",
                user_id=user_id,
            )
        )
        service = ExpenseClaimService(db)
        context_json = {
            "name": "李四",
            "attachment_names": ["didi-trip.png", "parking-ticket.jpg"],
            "attachment_count": 2,
            "draft_claim_id": existing_claim.id,
            "ocr_documents": [
                {
                    "filename": "didi-trip.png",
                    "summary": "滴滴出行",
                    "text": "滴滴出行 支付金额 32.50 元",
                    "document_type": "taxi_receipt",
                    "scene_code": "transport",
                    "document_fields": [{"key": "amount", "label": "支付金额", "value": "32.50"}],
                },
                {
                    "filename": "parking-ticket.jpg",
                    "summary": "停车票",
                    "text": "停车费 合计 18 元",
                    "document_type": "parking_toll_receipt",
                    "scene_code": "transport",
                    "document_fields": [{"key": "total_amount", "label": "合计金额", "value": "18"}],
                },
            ],
        }

        link_result = service.upsert_draft_from_ontology(
            run_id=ontology.run_id,
            user_id=user_id,
            message="把这两张票据关联到已有草稿",
            ontology=ontology,
            context_json={
                **context_json,
                "review_action": "link_to_existing_draft",
            },
        )

        db.refresh(existing_claim)
        assert link_result["claim_id"] == existing_claim.id
        assert existing_claim.invoice_count == 3
        assert len(existing_claim.items) == 3
        assert float(existing_claim.amount) == 70.5

        create_result = service.upsert_draft_from_ontology(
            run_id=f"{ontology.run_id}-new",
            user_id=user_id,
            message="单独新建一张报销单",
            ontology=ontology,
            context_json={
                **context_json,
                "review_action": "create_new_claim_from_documents",
            },
        )

        assert create_result["claim_id"] != existing_claim.id
        new_claim = db.get(ExpenseClaim, create_result["claim_id"])
        assert new_claim is not None
        assert new_claim.invoice_count == 2
        assert len(new_claim.items) == 2
        assert float(new_claim.amount) == 50.5


def test_link_existing_draft_blocks_duplicate_uploaded_invoice() -> None:
    user_id = "duplicate@example.com"

    with build_session() as db:
        employee = Employee(
            employee_no="E5010",
            name="重复票据员工",
            email=user_id,
        )
        db.add(employee)
        db.flush()
        existing_claim = ExpenseClaim(
            claim_no="EXP-202605-021",
            employee_id=employee.id,
            employee_name="重复票据员工",
            department_name="销售部",
            project_code=None,
            expense_type="transport",
            reason="原有交通报销",
            location="上海",
            amount=Decimal("32.50"),
            currency="CNY",
            invoice_count=1,
            occurred_at=datetime(2026, 5, 13, tzinfo=UTC),
            status="draft",
            approval_stage="待提交",
            risk_flags_json=[],
        )
        existing_claim.items = [
            ExpenseClaimItem(
                claim_id=existing_claim.id,
                item_date=date(2026, 5, 13),
                item_type="transport",
                item_reason="原有交通报销",
                item_location="上海",
                item_amount=Decimal("32.50"),
                invoice_id="didi-trip.png",
            )
        ]
        db.add(existing_claim)
        db.commit()

        context_json = {
            "name": "重复票据员工",
            "review_action": "link_to_existing_draft",
            "draft_claim_id": existing_claim.id,
            "attachment_names": ["didi-trip.png"],
            "attachment_count": 1,
            "ocr_documents": [
                {
                    "filename": "didi-trip.png",
                    "summary": "滴滴出行 支付金额 32.50 元",
                    "text": "滴滴出行 支付金额 32.50 元",
                    "document_type": "taxi_receipt",
                    "scene_code": "transport",
                    "document_fields": [{"key": "amount", "label": "支付金额", "value": "32.50"}],
                }
            ],
        }
        ontology = SemanticOntologyService(db).parse(
            OntologyParseRequest(
                query="把这张票据关联到已有草稿",
                user_id=user_id,
                context_json=context_json,
            )
        )

        result = ExpenseClaimService(db).upsert_draft_from_ontology(
            run_id=ontology.run_id,
            user_id=user_id,
            message="把这张票据关联到已有草稿",
            ontology=ontology,
            context_json=context_json,
        )

        db.refresh(existing_claim)
        assert result["duplicate_attachment_blocked"] is True
        assert result["submission_blocked"] is True
        assert "重复" in result["message"]
        assert "重新上传不同的票据" in result["message"]
        assert len(existing_claim.items) == 1
        assert existing_claim.invoice_count == 1
        assert float(existing_claim.amount) == 32.5


def test_upsert_travel_draft_uses_ticket_item_types_and_auto_allowance() -> None:
    user_id = "travel-allowance@example.com"

    with build_session() as db:
        employee = Employee(
            employee_no="E5010",
            name="差旅员工",
            email=user_id,
            grade="P4",
        )
        db.add(employee)
        db.commit()

        ontology = SemanticOntologyService(db).parse(
            OntologyParseRequest(
                query="我去北京出差 3 天，上传了火车票，帮我生成差旅费报销草稿",
                user_id=user_id,
            )
        )
        result = ExpenseClaimService(db).upsert_draft_from_ontology(
            run_id=ontology.run_id,
            user_id=user_id,
            message="我去北京出差 3 天，上传了火车票，帮我生成差旅费报销草稿",
            ontology=ontology,
            context_json={
                "name": "差旅员工",
                "grade": "P4",
                "attachment_names": ["train-ticket.png"],
                "attachment_count": 1,
                "review_form_values": {
                    "expense_type": "差旅费",
                    "location": "北京",
                    "time_range": "2026-05-13 至 2026-05-15",
                },
                "business_time_context": {
                    "mode": "range",
                    "start_date": "2026-05-13",
                    "end_date": "2026-05-15",
                    "display_value": "2026-05-13 至 2026-05-15",
                },
                "ocr_documents": [
                    {
                        "filename": "train-ticket.png",
                        "document_type": "train_ticket",
                        "document_type_label": "火车/高铁票",
                        "scene_code": "travel",
                        "scene_label": "差旅费",
                        "summary": "中国铁路电子客票 广州南-北京南 二等座 票价 354 元",
                        "text": "中国铁路电子客票 广州南-北京南 二等座 票价:￥354.00",
                        "document_fields": [
                            {"key": "amount", "label": "票价", "value": "￥354.00"},
                            {"key": "route", "label": "行程", "value": "广州南-北京南"},
                        ],
                    }
                ],
            },
        )

        claim = db.get(ExpenseClaim, result["claim_id"])
        assert claim is not None
        assert claim.expense_type == "travel"
        assert claim.invoice_count == 1
        assert len(claim.items) == 2
        train_item = next(item for item in claim.items if item.item_type == "train_ticket")
        allowance_item = next(item for item in claim.items if item.item_type == "travel_allowance")
        assert train_item.item_amount == Decimal("354.00")
        assert train_item.item_reason == "广州南-北京南"
        assert allowance_item.item_amount == Decimal("300.00")
        assert allowance_item.invoice_id is None
        assert allowance_item.is_system_generated is True
        assert claim.amount == Decimal("654.00")

        with pytest.raises(ValueError, match="系统自动计算"):
            ExpenseClaimService(db).update_claim_item(
                claim_id=claim.id,
                item_id=allowance_item.id,
                payload=ExpenseClaimItemUpdate(item_amount=Decimal("1.00")),
                current_user=CurrentUserContext(
                    username=user_id,
                    name="差旅员工",
                    role_codes=[],
                    is_admin=False,
                ),
            )


def test_upsert_travel_draft_uses_explicit_text_days_for_allowance() -> None:
    user_id = "travel-explicit-days@example.com"
    message = "业务发生时间:2026-05-20 至 2026-05-23，去上海支撑上海电力服务器部署，出差3天，申请差旅费报销"

    with build_session() as db:
        employee = Employee(
            employee_no="E5012",
            name="文本差旅员工",
            email=user_id,
            grade="P4",
        )
        db.add(employee)
        db.commit()

        ontology = SemanticOntologyService(db).parse(
            OntologyParseRequest(
                query=message,
                user_id=user_id,
                context_json={"name": "文本差旅员工", "grade": "P4"},
            )
        )
        result = ExpenseClaimService(db).upsert_draft_from_ontology(
            run_id=ontology.run_id,
            user_id=user_id,
            message=message,
            ontology=ontology,
            context_json={
                "name": "文本差旅员工",
                "grade": "P4",
                "user_input_text": message,
                "review_form_values": {
                    "expense_type": "差旅费",
                    "business_location": "上海",
                    "reason": "去上海支撑上海电力服务器部署，出差3天",
                    "time_range": "2026-05-20 至 2026-05-23",
                    "business_time": "2026-05-20 至 2026-05-23",
                },
                "business_time_context": {
                    "mode": "range",
                    "start_date": "2026-05-20",
                    "end_date": "2026-05-23",
                    "display_value": "2026-05-20 至 2026-05-23",
                },
            },
        )

        claim = db.get(ExpenseClaim, result["claim_id"])
        assert claim is not None
        assert claim.expense_type == "travel"
        allowance_item = next(item for item in claim.items if item.item_type == "travel_allowance")
        assert allowance_item.item_amount == Decimal("300.00")
        assert "3天" in allowance_item.item_reason
        assert allowance_item.item_date == date(2026, 5, 22)
        assert claim.amount == Decimal("300.00")


def test_sync_travel_claim_adds_allowance_from_manual_ticket_dates() -> None:
    with build_session() as db:
        employee = Employee(
            employee_no="E5011",
            name="手工差旅员工",
            email="manual-travel-allowance@example.com",
            grade="P4",
        )
        db.add(employee)
        db.flush()

        claim = build_claim(expense_type="travel", location="北京")
        claim.employee_id = employee.id
        claim.employee_name = employee.name
        claim.items[0].item_date = date(2026, 5, 13)
        claim.items[0].item_type = "train_ticket"
        claim.items[0].item_reason = "广州南-北京南"
        claim.items[0].item_location = "北京"
        claim.items[0].item_amount = Decimal("354.00")
        claim.items.append(
            ExpenseClaimItem(
                claim_id=claim.id,
                item_date=date(2026, 5, 15),
                item_type="train_ticket",
                item_reason="北京南-广州南",
                item_location="北京",
                item_amount=Decimal("354.00"),
                invoice_id="return-train.png",
            )
        )
        db.add(claim)
        db.commit()

        service = ExpenseClaimService(db)
        service._sync_claim_from_items(claim)
        db.commit()
        db.refresh(claim)

        allowance_item = next(item for item in claim.items if item.item_type == "travel_allowance")
        assert allowance_item.item_amount == Decimal("300.00")
        assert "3天" in allowance_item.item_reason
        assert allowance_item.invoice_id is None
        assert claim.amount == Decimal("1008.00")


def test_update_claim_item_allows_placeholder_date_reason_and_amount() -> None:
    current_user = CurrentUserContext(
        username="emp-1",
        name="张三",
        role_codes=[],
        is_admin=False,
    )

    with build_session() as db:
        claim = build_claim(expense_type="office", location="深圳")
        db.add(claim)
        db.commit()

        updated = ExpenseClaimService(db).update_claim_item(
            claim_id=claim.id,
            item_id=claim.items[0].id,
            payload=ExpenseClaimItemUpdate(
                item_reason="",
                item_location="",
                item_amount=Decimal("0.00"),
            ),
            current_user=current_user,
        )

        assert updated is not None
        db.refresh(claim)
        assert claim.items[0].item_date == date(2026, 5, 13)
        assert claim.items[0].item_reason == ""
        assert claim.items[0].item_location == ""
        assert claim.items[0].item_amount == Decimal("0.00")


def test_upsert_draft_from_ontology_updates_returned_claim_and_preserves_return_events() -> None:
    user_id = "returned-owner@example.com"
    return_flag = {
        "source": "manual_return",
        "return_event_id": "return-event-1",
        "message": "第一次退回：附件缺失。",
        "reason": "附件缺失。",
        "return_count": 1,
        "return_stage": "直属领导审批",
        "return_stage_key": "direct_manager",
        "risk_points": ["附件缺失或不清晰"],
    }

    with build_session() as db:
        employee = Employee(
            employee_no="E5004",
            name="赵六",
            email=user_id,
        )
        db.add(employee)
        db.flush()
        existing_claim = ExpenseClaim(
            claim_no="EXP-202605-012",
            employee_id=employee.id,
            employee_name="赵六",
            department_name="市场部",
            project_code=None,
            expense_type="transport",
            reason="原有交通报销",
            location="上海",
            amount=Decimal("20.00"),
            currency="CNY",
            invoice_count=1,
            occurred_at=datetime(2026, 5, 13, tzinfo=UTC),
            status="returned",
            approval_stage="待提交",
            risk_flags_json=[return_flag],
        )
        existing_claim.items = [
            ExpenseClaimItem(
                claim_id=existing_claim.id,
                item_date=date(2026, 5, 13),
                item_type="transport",
                item_reason="原有交通报销",
                item_location="上海",
                item_amount=Decimal("20.00"),
                invoice_id="old-trip.png",
            )
        ]
        db.add(existing_claim)
        db.commit()

        ontology = SemanticOntologyService(db).parse(
            OntologyParseRequest(
                query="我补充了交通票据，更新这张退回单据",
                user_id=user_id,
            )
        )
        ontology.risk_flags = ["系统识别：票据金额待人工核对。"]

        result = ExpenseClaimService(db).upsert_draft_from_ontology(
            run_id=ontology.run_id,
            user_id=user_id,
            message="我补充了交通票据，更新这张退回单据",
            ontology=ontology,
            context_json={
                "name": "赵六",
                "draft_claim_id": existing_claim.id,
                "attachment_names": ["new-trip.png"],
                "attachment_count": 1,
                "ocr_documents": [
                    {
                        "filename": "new-trip.png",
                        "summary": "滴滴出行 支付金额 32 元",
                        "text": "滴滴出行 支付金额 32 元",
                        "document_type": "taxi_receipt",
                        "scene_code": "transport",
                    }
                ],
            },
        )

        db.refresh(existing_claim)
        assert result["claim_id"] == existing_claim.id
        assert existing_claim.status == "draft"
        assert "系统识别：票据金额待人工核对。" in existing_claim.risk_flags_json
        manual_returns = [
            flag
            for flag in list(existing_claim.risk_flags_json or [])
            if isinstance(flag, dict) and flag.get("source") == "manual_return"
        ]
        assert manual_returns == [return_flag]


def test_generate_claim_no_uses_re_prefix_timestamp_and_random_suffix() -> None:
    with build_session() as db:
        db.add_all(
            [
                ExpenseClaim(
                    claim_no="EXP-202605-001",
                    employee_name="张三",
                    department_name="市场部",
                    project_code=None,
                    expense_type="transport",
                    reason="交通报销",
                    location="深圳",
                    amount=Decimal("10.00"),
                    currency="CNY",
                    invoice_count=1,
                    occurred_at=datetime(2026, 5, 10, tzinfo=UTC),
                    status="draft",
                    approval_stage="待提交",
                    risk_flags_json=[],
                ),
                ExpenseClaim(
                    claim_no="EXP-202605-003",
                    employee_name="李四",
                    department_name="销售部",
                    project_code=None,
                    expense_type="transport",
                    reason="交通报销",
                    location="上海",
                    amount=Decimal("20.00"),
                    currency="CNY",
                    invoice_count=1,
                    occurred_at=datetime(2026, 5, 11, tzinfo=UTC),
                    status="submitted",
                    approval_stage="审批中",
                    risk_flags_json=[],
                ),
            ]
        )
        db.commit()

        service = ExpenseClaimService(db)

        assert re.fullmatch(
            r"RE-\d{14}-[A-HJ-NP-Z2-9]{8}",
            service._generate_claim_no(datetime(2026, 5, 14, tzinfo=UTC)),
        )


def test_upsert_draft_from_ontology_retries_claim_no_conflict() -> None:
    user_id = "zhaoliu-claimno@example.com"

    with build_session() as db:
        employee = Employee(
            employee_no="E5006",
            name="赵六",
            email=user_id,
        )
        db.add(employee)
        db.flush()
        db.add(
            ExpenseClaim(
                claim_no="RE-20260525101010-ABCDEFGH",
                employee_name="历史单据",
                department_name="财务部",
                project_code=None,
                expense_type="other",
                reason="历史草稿",
                location="北京",
                amount=Decimal("0.00"),
                currency="CNY",
                invoice_count=0,
                occurred_at=datetime(2026, 5, 12, tzinfo=UTC),
                status="submitted",
                approval_stage="审批中",
                risk_flags_json=[],
            )
        )
        db.commit()

        ontology = SemanticOntologyService(db).parse(
            OntologyParseRequest(
                query="帮我生成报销草稿，我昨天交通费 13.4 元",
                user_id=user_id,
            )
        )
        service = ExpenseClaimService(db)
        generated_claim_nos = iter(
            ["RE-20260525101010-ABCDEFGH", "RE-20260525101010-HGFEDCBA"]
        )
        service._generate_claim_no = lambda occurred_at: next(generated_claim_nos)

        result = service.upsert_draft_from_ontology(
            run_id=ontology.run_id,
            user_id=user_id,
            message="帮我生成报销草稿，我昨天交通费 13.4 元",
            ontology=ontology,
            context_json={
                "name": "赵六",
                "user_input_text": "帮我生成报销草稿，我昨天交通费 13.4 元",
            },
        )

        created_claim = db.get(ExpenseClaim, result["claim_id"])
        assert created_claim is not None
        assert created_claim.claim_no == "RE-20260525101010-HGFEDCBA"
        assert result["claim_no"] == "RE-20260525101010-HGFEDCBA"


def test_create_claim_item_adds_blank_draft_row_without_forcing_attachment() -> None:
    current_user = CurrentUserContext(
        username="emp-1",
        name="张三",
        role_codes=[],
        is_admin=False,
    )

    with build_session() as db:
        claim = build_claim(expense_type="office", location="深圳南山")
        db.add(claim)
        db.commit()

        service = ExpenseClaimService(db)
        updated = service.create_claim_item(
            claim_id=claim.id,
            payload=ExpenseClaimItemCreate(),
            current_user=current_user,
        )

        assert updated is not None
        assert len(updated.items) == 2
        assert updated.amount == Decimal("88.00")
        assert updated.invoice_count == 1

        new_item = next(item for item in updated.items if item.id != "item-1")
        assert new_item.item_type == "office"
        assert new_item.item_reason == ""
        assert new_item.item_location == ""
        assert new_item.item_amount == Decimal("0.00")
        assert new_item.invoice_id is None


def test_update_claim_reason_only_allows_draft_pending_submission() -> None:
    current_user = CurrentUserContext(
        username="emp-1",
        name="张三",
        role_codes=[],
        is_admin=False,
    )

    with build_session() as db:
        claim = build_claim(expense_type="travel", location="北京")
        db.add(claim)
        db.commit()

        service = ExpenseClaimService(db)
        updated = service.update_claim(
            claim_id=claim.id,
            payload=ExpenseClaimUpdate(reason="去北京客户现场出差，处理项目验收事项"),
            current_user=current_user,
        )

        assert updated is not None
        assert updated.reason == "去北京客户现场出差，处理项目验收事项"

        claim.status = "submitted"
        claim.submitted_at = datetime(2026, 5, 14, tzinfo=UTC)
        claim.approval_stage = "直属领导审批"
        db.commit()

        with pytest.raises(ValueError, match="草稿待提交"):
            service.update_claim(
                claim_id=claim.id,
                payload=ExpenseClaimUpdate(reason="提交后不能改"),
                current_user=current_user,
            )


def test_update_claim_item_reanalyzes_existing_attachment(monkeypatch, tmp_path) -> None:
    current_user = CurrentUserContext(
        username="emp-1",
        name="张三",
        role_codes=[],
        is_admin=False,
    )

    def fake_recognize(
        self,
        files: list[tuple[str, bytes, str | None]],
    ) -> OcrRecognizeBatchRead:
        return OcrRecognizeBatchRead(
            total_file_count=1,
            success_count=1,
            documents=[
                OcrRecognizeDocumentRead(
                    filename="office-note.png",
                    media_type="image/png",
                    text="办公用品发票 金额88元 2026-05-13",
                    summary="识别到办公用品发票，金额 88 元。",
                    avg_score=0.98,
                    line_count=1,
                    page_count=1,
                    warnings=[],
                )
            ],
        )

    monkeypatch.setattr(OcrService, "recognize_files", fake_recognize)
    monkeypatch.setattr(ExpenseClaimAttachmentStorage, "root", lambda self: tmp_path)

    with build_session() as db:
        claim = build_claim(expense_type="office", location="深圳南山")
        claim.invoice_count = 0
        claim.items[0].invoice_id = None
        claim.items[0].item_reason = "办公用品采购"
        db.add(claim)
        db.commit()

        service = ExpenseClaimService(db)
        service.upload_claim_item_attachment(
            claim_id=claim.id,
            item_id=claim.items[0].id,
            filename="office-note.png",
            content=b"fake-image-bytes",
            media_type="image/png",
            current_user=current_user,
        )

        uploaded_meta = service.get_claim_item_attachment_meta(
            claim_id=claim.id,
            item_id=claim.items[0].id,
            current_user=current_user,
        )
        assert uploaded_meta is not None
        assert uploaded_meta["preview_kind"] == "image"
        assert uploaded_meta["preview_url"].endswith(
            f"/reimbursements/claims/{claim.id}/items/{claim.items[0].id}/attachment/preview"
        )
        assert uploaded_meta["analysis"]["severity"] == "pass"
        assert uploaded_meta["document_info"]["document_type"] == "office_invoice"
        assert uploaded_meta["requirement_check"]["matches"] is True

        updated = service.update_claim_item(
            claim_id=claim.id,
            item_id=claim.items[0].id,
            payload=ExpenseClaimItemUpdate(
                item_type="transport",
                item_reason="打车报销",
            ),
            current_user=current_user,
        )

        assert updated is not None
        assert any(flag.get("source") == "attachment_analysis" for flag in updated.risk_flags_json)

        refreshed_meta = service.get_claim_item_attachment_meta(
            claim_id=claim.id,
            item_id=claim.items[0].id,
            current_user=current_user,
        )
        assert refreshed_meta is not None
        assert refreshed_meta["analysis"]["severity"] == "high"
        assert refreshed_meta["requirement_check"]["matches"] is False
        assert any("附件类型要求" in point for point in refreshed_meta["analysis"]["points"])


def test_upload_train_ticket_attachment_backfills_item_amount(monkeypatch, tmp_path) -> None:
    current_user = CurrentUserContext(
        username="emp-1",
        name="张三",
        role_codes=[],
        is_admin=False,
    )

    def fake_recognize(
        self,
        files: list[tuple[str, bytes, str | None]],
    ) -> OcrRecognizeBatchRead:
        return OcrRecognizeBatchRead(
            total_file_count=1,
            success_count=1,
            documents=[
                OcrRecognizeDocumentRead(
                    filename="train-ticket.png",
                    media_type="image/png",
                    text="中国铁路电子客票 广州南-北京南 二等座 2026-02-20 08:30开 票价:￥354.00",
                    summary="铁路电子客票，2026-02-20 08:30 广州南至北京南，票价 354 元。",
                    avg_score=0.98,
                    line_count=1,
                    page_count=1,
                    document_type="train_ticket",
                    document_type_label="火车/高铁票",
                    scene_code="travel",
                    scene_label="差旅费",
                    document_fields=[
                        {"key": "invoice_date", "label": "开票日期", "value": "2026-02-18"},
                        {"key": "trip_date", "label": "行程日期", "value": "2026-02-20 08:30"},
                        {"key": "fare", "label": "票价", "value": "￥354.00"},
                    ],
                )
            ],
        )

    monkeypatch.setattr(OcrService, "recognize_files", fake_recognize)
    monkeypatch.setattr(ExpenseClaimAttachmentStorage, "root", lambda self: tmp_path)

    with build_session() as db:
        claim = build_claim(expense_type="travel", location="北京")
        claim.amount = Decimal("0.00")
        claim.invoice_count = 0
        claim.items[0].item_amount = Decimal("0.00")
        claim.items[0].invoice_id = None
        db.add(claim)
        db.commit()

        service = ExpenseClaimService(db)
        updated = service.upload_claim_item_attachment(
            claim_id=claim.id,
            item_id=claim.items[0].id,
            filename="train-ticket.png",
            content=b"fake-image-bytes",
            media_type="image/png",
            current_user=current_user,
        )

        assert updated is not None
        assert updated["item_amount"] == Decimal("354.00")
        assert updated["item_date"] == "2026-02-20"
        assert updated["item_type"] == "train_ticket"
        assert updated["item_reason"] == "广州南-北京南"
        assert updated["claim_amount"] == Decimal("354.00")
        db.refresh(claim)
        assert claim.items[0].item_amount == Decimal("354.00")
        assert claim.items[0].item_type == "train_ticket"
        assert claim.items[0].item_date == date(2026, 2, 20)
        assert claim.items[0].item_reason == "广州南-北京南"
        assert claim.amount == Decimal("354.00")
        uploaded_meta = service.get_claim_item_attachment_meta(
            claim_id=claim.id,
            item_id=claim.items[0].id,
            current_user=current_user,
        )
        assert uploaded_meta is not None
        assert uploaded_meta["document_info"]["document_type"] == "train_ticket"
        assert any(
            field["label"] == "列车出发时间" and field["value"] == "2026-02-20 08:30"
            for field in uploaded_meta["document_info"]["fields"]
        )
        assert any(
            field["label"] == "开票日期" and field["value"] == "2026-02-18"
            for field in uploaded_meta["document_info"]["fields"]
        )
        assert any(
            field["label"] == "票价" and field["value"] == "￥354.00"
            for field in uploaded_meta["document_info"]["fields"]
        )
        assert not any("用途字段" in point for point in uploaded_meta["analysis"]["points"])


def test_upload_hotel_attachment_audits_date_like_amount(monkeypatch, tmp_path) -> None:
    current_user = CurrentUserContext(
        username="emp-1",
        name="张三",
        role_codes=[],
        is_admin=False,
    )

    def fake_recognize(
        self,
        files: list[tuple[str, bytes, str | None]],
    ) -> OcrRecognizeBatchRead:
        return OcrRecognizeBatchRead(
            total_file_count=1,
            success_count=1,
            documents=[
                OcrRecognizeDocumentRead(
                    filename="hotel-invoice.png",
                    media_type="image/png",
                    text="北京中心酒店 总费用是828元 入住日期 2026-02-20 离店日期 2026-02-21",
                    summary="酒店住宿票据，住宿总费用 828 元。",
                    avg_score=0.96,
                    line_count=1,
                    page_count=1,
                    document_type="hotel_invoice",
                    document_type_label="酒店住宿票据",
                    scene_code="hotel",
                    scene_label="住宿票据",
                    document_fields=[
                        {"key": "amount", "label": "金额", "value": "2026元"},
                        {"key": "hotel_name", "label": "酒店", "value": "北京中心酒店"},
                        {"key": "check_in", "label": "入住日期", "value": "2026-02-20"},
                        {"key": "check_out", "label": "离店日期", "value": "2026-02-21"},
                    ],
                )
            ],
        )

    monkeypatch.setattr(OcrService, "recognize_files", fake_recognize)
    monkeypatch.setattr(ExpenseClaimAttachmentStorage, "root", lambda self: tmp_path)

    with build_session() as db:
        claim = build_claim(expense_type="hotel", location="北京")
        claim.amount = Decimal("0.00")
        claim.invoice_count = 0
        claim.items[0].item_type = "hotel"
        claim.items[0].item_reason = ""
        claim.items[0].item_amount = Decimal("0.00")
        claim.items[0].invoice_id = None
        db.add(claim)
        db.commit()

        service = ExpenseClaimService(db)
        updated = service.upload_claim_item_attachment(
            claim_id=claim.id,
            item_id=claim.items[0].id,
            filename="hotel-invoice.png",
            content=b"fake-image-bytes",
            media_type="image/png",
            current_user=current_user,
        )

        assert updated is not None
        assert updated["item_type"] == "hotel_ticket"
        assert updated["item_amount"] == Decimal("828.00")
        assert updated["claim_amount"] == Decimal("828.00")
        db.refresh(claim)
        assert claim.items[0].item_amount == Decimal("828.00")
        assert claim.amount == Decimal("828.00")
        uploaded_meta = service.get_claim_item_attachment_meta(
            claim_id=claim.id,
            item_id=claim.items[0].id,
            current_user=current_user,
        )
        assert uploaded_meta is not None
        assert uploaded_meta["analysis"]["severity"] == "medium"
        assert any("费用核算" in point and "828.00 元" in point for point in uploaded_meta["analysis"]["points"])
        assert not any("2026.00 元与报销金额" in point for point in uploaded_meta["analysis"]["points"])


def test_upload_hotel_attachment_flags_amount_over_travel_policy(monkeypatch, tmp_path) -> None:
    current_user = CurrentUserContext(
        username="emp-hotel-risk@example.com",
        name="张三",
        role_codes=[],
        is_admin=False,
    )

    def fake_recognize(
        self,
        files: list[tuple[str, bytes, str | None]],
    ) -> OcrRecognizeBatchRead:
        return OcrRecognizeBatchRead(
            total_file_count=1,
            success_count=1,
            documents=[
                OcrRecognizeDocumentRead(
                    filename="hotel-risk.png",
                    media_type="image/png",
                    text="北京全季酒店 住宿 1晚 金额800元 2026-05-13",
                    summary="北京全季酒店住宿发票，住宿 1 晚，金额 800 元。",
                    avg_score=0.98,
                    line_count=1,
                    page_count=1,
                    document_type="hotel_invoice",
                    document_type_label="酒店住宿票据",
                    scene_code="hotel",
                    scene_label="住宿票据",
                    document_fields=[
                        {"key": "merchant_name", "label": "商户", "value": "北京全季酒店"},
                        {"key": "amount", "label": "金额", "value": "800元"},
                        {"key": "date", "label": "日期", "value": "2026-05-13"},
                    ],
                )
            ],
        )

    monkeypatch.setattr(OcrService, "recognize_files", fake_recognize)
    monkeypatch.setattr(ExpenseClaimAttachmentStorage, "root", lambda self: tmp_path)

    with build_session() as db:
        employee = Employee(
            employee_no="E7401",
            name="张三",
            email="emp-hotel-risk@example.com",
            grade="P4",
        )
        db.add(employee)
        db.flush()

        claim = build_claim(expense_type="travel", location="北京")
        claim.employee = employee
        claim.employee_id = employee.id
        claim.reason = "北京客户现场出差"
        claim.amount = Decimal("0.00")
        claim.invoice_count = 0
        claim.items[0].item_type = "hotel"
        claim.items[0].item_reason = "北京住宿"
        claim.items[0].item_location = "北京"
        claim.items[0].item_amount = Decimal("0.00")
        claim.items[0].invoice_id = None
        db.add(claim)
        db.commit()

        service = ExpenseClaimService(db)
        updated = service.upload_claim_item_attachment(
            claim_id=claim.id,
            item_id=claim.items[0].id,
            filename="hotel-risk.png",
            content=b"fake-image-bytes",
            media_type="image/png",
            current_user=current_user,
        )

        assert updated is not None
        uploaded_meta = service.get_claim_item_attachment_meta(
            claim_id=claim.id,
            item_id=claim.items[0].id,
            current_user=current_user,
        )
        assert uploaded_meta is not None
        analysis = uploaded_meta["analysis"]
        assert analysis["severity"] == "high"
        assert analysis["headline"] == "AI提示：住宿金额超出报销标准"
        assert "保留在单据中" in analysis["summary"]
        assert "特殊情况" in analysis["summary"]
        assert any("住宿标准" in point and "800.00 元" in point for point in analysis["points"])
        assert any("住宿费按员工职级" in basis for basis in analysis["rule_basis"])
        db.refresh(claim)
        hotel_item = next(item for item in claim.items if str(item.invoice_id or "").strip())
        assert hotel_item.item_amount == Decimal("800.00")
        assert claim.invoice_count == 1
        assert any(
            isinstance(flag, dict)
            and str(flag.get("source") or "").strip() == "attachment_analysis"
            and flag.get("item_id") == hotel_item.id
            and str(flag.get("severity") or "").strip() == "high"
            for flag in list(claim.risk_flags_json or [])
        )


def test_attachment_analysis_does_not_compare_business_purpose_with_ticket_scene() -> None:
    with build_session() as db:
        claim = build_claim(expense_type="travel", location="上海")
        claim.items[0].item_type = "train_ticket"
        claim.items[0].item_reason = "2026-02-20 至 2026-02-23，支撑上海电力项目部署"
        claim.items[0].item_amount = Decimal("354.00")
        db.add(claim)
        db.commit()

        document = OcrRecognizeDocumentRead(
            filename="train-ticket.png",
            media_type="image/png",
            text="中国铁路电子客票 上海虹桥-武汉 二等座 2026-02-20 票价:￥354.00",
            summary="铁路电子客票，上海虹桥至武汉，票价 354 元。",
            avg_score=0.98,
            line_count=1,
            page_count=1,
            document_type="train_ticket",
            document_type_label="火车/高铁票",
            scene_code="travel",
            scene_label="差旅票据",
            document_fields=[
                {"key": "amount", "label": "票价", "value": "￥354.00"},
                {"key": "date", "label": "日期", "value": "2026-02-20"},
                {"key": "route", "label": "行程", "value": "上海虹桥-武汉"},
            ],
        )

        analysis = ExpenseClaimService(db)._build_attachment_analysis(
            document=document,
            item=claim.items[0],
        )

        assert analysis["severity"] == "medium"
        assert not any("用途字段" in point for point in analysis["points"])
        assert any("行程说明" in point and "起始地-目的地" in point for point in analysis["points"])


def test_attachment_risk_flag_message_uses_specific_points(monkeypatch, tmp_path) -> None:
    with build_session() as db:
        claim = build_claim(expense_type="travel", location="上海")
        claim.items[0].invoice_id = "invoice.png"
        db.add(claim)
        db.commit()

        generic_summary = "当前附件可见部分内容，但金额、用途、日期或附件类型仍有缺失或不一致。"
        file_path = tmp_path / "invoice.png"
        file_path.write_bytes(b"fake")
        service = ExpenseClaimService(db)
        monkeypatch.setattr(
            ExpenseClaimAttachmentStorage,
            "resolve_path",
            lambda self, storage_key: file_path,
        )
        monkeypatch.setattr(
            service._attachment_storage,
            "read_meta",
            lambda path: {
                "analysis": {
                    "severity": "medium",
                    "label": "中风险",
                    "summary": generic_summary,
                    "points": [
                        "日期字段：未识别到开票日期或业务发生日期。",
                        "金额字段：附件识别金额 300.00 元与报销金额 88.00 元不一致。",
                    ],
                }
            },
        )

        flags = service._build_claim_attachment_risk_flags([claim.items[0]])

        assert len(flags) == 1
        assert "日期字段：未识别到开票日期或业务发生日期。" in flags[0]["message"]
        assert "当前附件可见部分内容" not in flags[0]["message"]
        assert flags[0]["summary"] == generic_summary
        assert flags[0]["points"] == [
            "日期字段：未识别到开票日期或业务发生日期。",
            "金额字段：附件识别金额 300.00 元与报销金额 88.00 元不一致。",
        ]


def test_upload_ride_receipt_backfills_item_reason_from_addresses(monkeypatch, tmp_path) -> None:
    current_user = CurrentUserContext(
        username="emp-1",
        name="张三",
        role_codes=[],
        is_admin=False,
    )

    def fake_recognize(
        self,
        files: list[tuple[str, bytes, str | None]],
    ) -> OcrRecognizeBatchRead:
        return OcrRecognizeBatchRead(
            total_file_count=1,
            success_count=1,
            documents=[
                OcrRecognizeDocumentRead(
                    filename="ride-receipt.png",
                    media_type="image/png",
                    text="滴滴出行订单 起点：深圳北站 终点：腾讯滨海大厦 实付金额：42.00元",
                    summary="滴滴出行乘车票据，深圳北站到腾讯滨海大厦，金额 42 元。",
                    avg_score=0.98,
                    line_count=1,
                    page_count=1,
                    document_type="taxi_receipt",
                    document_type_label="出租车/网约车票据",
                    scene_code="transport",
                    scene_label="交通票据",
                    document_fields=[
                        {"key": "start_location", "label": "起点", "value": "深圳北站"},
                        {"key": "end_location", "label": "终点", "value": "腾讯滨海大厦"},
                        {"key": "amount", "label": "实付金额", "value": "42.00元"},
                    ],
                )
            ],
        )

    monkeypatch.setattr(OcrService, "recognize_files", fake_recognize)
    monkeypatch.setattr(ExpenseClaimAttachmentStorage, "root", lambda self: tmp_path)

    with build_session() as db:
        claim = build_claim(expense_type="transport", location="深圳")
        claim.amount = Decimal("0.00")
        claim.invoice_count = 0
        claim.items[0].item_type = "transport"
        claim.items[0].item_reason = "打车报销"
        claim.items[0].item_amount = Decimal("0.00")
        claim.items[0].invoice_id = None
        db.add(claim)
        db.commit()

        service = ExpenseClaimService(db)
        updated = service.upload_claim_item_attachment(
            claim_id=claim.id,
            item_id=claim.items[0].id,
            filename="ride-receipt.png",
            content=b"fake-image-bytes",
            media_type="image/png",
            current_user=current_user,
        )

        assert updated is not None
        db.refresh(claim)
        assert claim.items[0].item_type == "ride_ticket"
        assert claim.items[0].item_reason == "深圳北站-腾讯滨海大厦"
        assert claim.items[0].item_amount == Decimal("42.00")
        assert claim.amount == Decimal("42.00")


def test_delete_claim_item_removes_row_and_attachment_files(monkeypatch, tmp_path) -> None:
    current_user = CurrentUserContext(
        username="emp-1",
        name="张三",
        role_codes=[],
        is_admin=False,
    )

    def fake_recognize(
        self,
        files: list[tuple[str, bytes, str | None]],
    ) -> OcrRecognizeBatchRead:
        return OcrRecognizeBatchRead(
            total_file_count=1,
            success_count=1,
            documents=[
                OcrRecognizeDocumentRead(
                    filename="office-note.png",
                    media_type="image/png",
                    text="办公用品发票 金额88元 2026-05-13",
                    summary="识别到办公用品发票，金额 88 元。",
                    avg_score=0.98,
                    line_count=1,
                    page_count=1,
                    warnings=[],
                )
            ],
        )

    monkeypatch.setattr(OcrService, "recognize_files", fake_recognize)
    monkeypatch.setattr(ExpenseClaimAttachmentStorage, "root", lambda self: tmp_path)

    with build_session() as db:
        claim = build_claim(expense_type="office", location="深圳南山")
        claim.invoice_count = 0
        claim.items[0].invoice_id = None
        claim.items[0].item_reason = "办公用品采购"
        db.add(claim)
        db.commit()

        service = ExpenseClaimService(db)
        upload_payload = service.upload_claim_item_attachment(
            claim_id=claim.id,
            item_id=claim.items[0].id,
            filename="office-note.png",
            content=b"fake-image-bytes",
            media_type="image/png",
            current_user=current_user,
        )

        assert upload_payload is not None
        attachment_root = tmp_path / claim.id / claim.items[0].id
        assert attachment_root.exists()

        delete_payload = service.delete_claim_item(
            claim_id=claim.id,
            item_id=claim.items[0].id,
            current_user=current_user,
        )

        assert delete_payload is not None
        assert delete_payload["claim_id"] == claim.id
        refreshed_claim = service.get_claim(claim.id, current_user)
        assert refreshed_claim is not None
        assert refreshed_claim.items == []
        assert refreshed_claim.amount == Decimal("0.00")
        assert refreshed_claim.invoice_count == 0
        assert not attachment_root.exists()


def test_delete_claim_removes_all_claim_attachment_files(monkeypatch, tmp_path) -> None:
    current_user = CurrentUserContext(
        username="emp-1",
        name="张三",
        role_codes=[],
        is_admin=False,
    )
    monkeypatch.setattr(ExpenseClaimAttachmentStorage, "root", lambda self: tmp_path)

    with build_session() as db:
        claim = build_claim(expense_type="office", location="深圳南山")
        attachment_dir = tmp_path / claim.id / claim.items[0].id
        attachment_dir.mkdir(parents=True)
        attachment_path = attachment_dir / "office-note.png"
        attachment_path.write_bytes(b"fake-image-bytes")
        (attachment_dir / "office-note.png.meta.json").write_text("{}", encoding="utf-8")
        orphan_path = tmp_path / claim.id / "orphan-preview.png"
        orphan_path.write_bytes(b"orphan-preview")
        claim.items[0].invoice_id = f"{claim.id}/{claim.items[0].id}/office-note.png"
        db.add(claim)
        db.commit()
        conversation = AgentConversationService(db).get_or_create_conversation(
            conversation_id=None,
            user_id=current_user.username,
            source="user_message",
            context_json={
                "session_type": "expense",
                "draft_claim_id": claim.id,
            },
        )
        claim_id = claim.id
        claim_root = tmp_path / claim.id

        deleted = ExpenseClaimService(db).delete_claim(claim_id, current_user)

        assert deleted is not None
        assert db.get(ExpenseClaim, claim_id) is None
        assert not claim_root.exists()
        assert AgentConversationService(db).get_conversation(conversation.conversation_id) is None


def test_attachment_preview_resolves_legacy_filename_in_claim_item_directory(monkeypatch, tmp_path) -> None:
    current_user = CurrentUserContext(
        username="emp-1",
        name="张三",
        role_codes=[],
        is_admin=False,
    )

    monkeypatch.setattr(ExpenseClaimAttachmentStorage, "root", lambda self: tmp_path)

    with build_session() as db:
        claim = build_claim(expense_type="transport", location="上海")
        claim.items[0].invoice_id = "legacy-ticket.pdf"
        db.add(claim)
        db.commit()

        attachment_dir = tmp_path / claim.id / claim.items[0].id
        attachment_dir.mkdir(parents=True)
        file_path = attachment_dir / "legacy-ticket.pdf"
        file_path.write_bytes(b"legacy-pdf-bytes")
        (attachment_dir / "legacy-ticket.pdf.meta.json").write_text(
            '{"file_name":"legacy-ticket.pdf","media_type":"application/pdf","previewable":true}',
            encoding="utf-8",
        )

        payload = ExpenseClaimService(db).get_claim_item_attachment_preview_content(
            claim_id=claim.id,
            item_id=claim.items[0].id,
            current_user=current_user,
        )

        assert payload is not None
        resolved_path, media_type, filename = payload
        assert resolved_path == file_path
        assert media_type == "application/pdf"
        assert filename == "legacy-ticket.pdf"


def test_submit_claim_runs_ai_review_and_routes_to_direct_manager() -> None:
    current_user = CurrentUserContext(
        username="emp-submit@example.com",
        name="张三",
        role_codes=[],
        is_admin=False,
    )

    with build_session() as db:
        manager = Employee(
            employee_no="E7000",
            name="李经理",
            email="manager@example.com",
        )
        employee = Employee(
            employee_no="E7001",
            name="张三",
            email="emp-submit@example.com",
            manager=manager,
        )
        claim = build_claim(expense_type="transport", location="上海")
        claim.employee = employee
        claim.employee_id = employee.id
        claim.items[0].invoice_id = "taxi-ticket.png"
        db.add_all([manager, employee, claim])
        db.commit()

        submitted = ExpenseClaimService(db).submit_claim(claim.id, current_user)

        assert submitted is not None
        assert submitted.status == "submitted"
        assert submitted.approval_stage == "直属领导审批"
        assert submitted.submitted_at is not None


def test_submit_claim_allows_returned_claim_to_be_resubmitted() -> None:
    current_user = CurrentUserContext(
        username="emp-submit@example.com",
        name="张三",
        role_codes=[],
        is_admin=False,
    )

    with build_session() as db:
        manager = Employee(
            employee_no="E7100",
            name="李经理",
            email="manager-returned@example.com",
        )
        employee = Employee(
            employee_no="E7101",
            name="张三",
            email="emp-submit@example.com",
            manager=manager,
        )
        claim = build_claim(expense_type="transport", location="上海")
        claim.employee = employee
        claim.employee_id = employee.id
        claim.status = "returned"
        claim.approval_stage = "待补充"
        claim.items[0].invoice_id = "taxi-ticket.png"
        db.add_all([manager, employee, claim])
        db.commit()

        submitted = ExpenseClaimService(db).submit_claim(claim.id, current_user)

        assert submitted is not None
        assert submitted.status == "submitted"
        assert submitted.approval_stage == "直属领导审批"
        assert submitted.submitted_at is not None


def test_submit_claim_backfills_department_from_current_employee() -> None:
    current_user = CurrentUserContext(
        username="emp-dept@example.com",
        name="张三",
        role_codes=[],
        is_admin=False,
    )

    with build_session() as db:
        department = OrganizationUnit(
            unit_code="D7200",
            name="销售部",
        )
        manager = Employee(
            employee_no="E7200",
            name="李经理",
            email="manager-dept@example.com",
        )
        employee = Employee(
            employee_no="E7201",
            name="张三",
            email="emp-dept@example.com",
            organization_unit=department,
            manager=manager,
        )
        claim = build_claim(expense_type="transport", location="待补充")
        claim.employee = None
        claim.employee_id = None
        claim.employee_name = "张三"
        claim.department_id = None
        claim.department_name = "待补充"
        claim.items[0].item_location = "待补充"
        db.add_all([department, manager, employee, claim])
        db.commit()

        submitted = ExpenseClaimService(db).submit_claim(claim.id, current_user)

        assert submitted is not None
        assert submitted.status == "submitted"
        assert submitted.department_id == department.id
        assert submitted.department_name == "销售部"
        assert submitted.approval_stage == "直属领导审批"


def test_submit_claim_routes_high_risk_attachment_to_approval_with_review_flag(
    monkeypatch,
    tmp_path,
) -> None:
    current_user = CurrentUserContext(
        username="emp-risk@example.com",
        name="张三",
        role_codes=[],
        is_admin=False,
    )

    def fake_recognize(
        self,
        files: list[tuple[str, bytes, str | None]],
    ) -> OcrRecognizeBatchRead:
        return OcrRecognizeBatchRead(
            total_file_count=1,
            success_count=1,
            documents=[
                OcrRecognizeDocumentRead(
                    filename="taxi-note.png",
                    media_type="image/png",
                    text="滴滴出行电子发票 金额120元 2026-05-13",
                    summary="识别到交通出行发票，金额 120 元。",
                    avg_score=0.97,
                    line_count=1,
                    page_count=1,
                    warnings=[],
                )
            ],
        )

    monkeypatch.setattr(OcrService, "recognize_files", fake_recognize)
    monkeypatch.setattr(ExpenseClaimAttachmentStorage, "root", lambda self: tmp_path)

    with build_session() as db:
        manager = Employee(
            employee_no="E7100",
            name="李经理",
            email="manager2@example.com",
        )
        employee = Employee(
            employee_no="E7101",
            name="张三",
            email="emp-risk@example.com",
            manager=manager,
        )
        claim = build_claim(expense_type="office", location="深圳南山")
        claim.employee = employee
        claim.employee_id = employee.id
        claim.invoice_count = 0
        claim.items[0].invoice_id = None
        claim.items[0].item_reason = "办公用品采购"
        db.add_all([manager, employee, claim])
        db.commit()

        service = ExpenseClaimService(db)
        service.upload_claim_item_attachment(
            claim_id=claim.id,
            item_id=claim.items[0].id,
            filename="taxi-note.png",
            content=b"fake-image-bytes",
            media_type="image/png",
            current_user=current_user,
        )

        submitted = service.submit_claim(claim.id, current_user)

        assert submitted is not None
        assert submitted.status == "submitted"
        assert submitted.approval_stage == "直属领导审批"
        assert submitted.submitted_at is not None
        assert any(
            isinstance(flag, dict) and str(flag.get("source") or "").strip() == "submission_review"
            for flag in list(submitted.risk_flags_json or [])
        )


def test_submit_claim_routes_travel_route_mismatch_to_approval_with_review_flag(
    monkeypatch,
    tmp_path,
) -> None:
    current_user = CurrentUserContext(
        username="emp-travel@example.com",
        name="张三",
        role_codes=[],
        is_admin=False,
    )

    def fake_recognize(
        self,
        files: list[tuple[str, bytes, str | None]],
    ) -> OcrRecognizeBatchRead:
        documents: list[OcrRecognizeDocumentRead] = []
        for filename, _, media_type in files:
            if filename == "outbound.png":
                documents.append(
                    OcrRecognizeDocumentRead(
                        filename=filename,
                        media_type=media_type or "image/png",
                        text="电子行程单 2026-05-13 经济舱 武汉-上海 金额 480元 航班号 MU5101",
                        summary="武汉到上海机票",
                        avg_score=0.98,
                        line_count=1,
                        page_count=1,
                        document_type="flight_itinerary",
                        document_type_label="机票/航班行程单",
                        scene_code="travel",
                        scene_label="差旅票据",
                        document_fields=[
                            {"key": "route", "label": "行程", "value": "武汉-上海"},
                            {"key": "amount", "label": "金额", "value": "480元"},
                            {"key": "date", "label": "日期", "value": "2026-05-13"},
                        ],
                        warnings=[],
                    )
                )
            elif filename == "onward.png":
                documents.append(
                    OcrRecognizeDocumentRead(
                        filename=filename,
                        media_type=media_type or "image/png",
                        text="电子行程单 2026-05-14 经济舱 上海-成都 金额 360元 航班号 MU5402",
                        summary="上海到成都机票",
                        avg_score=0.98,
                        line_count=1,
                        page_count=1,
                        document_type="flight_itinerary",
                        document_type_label="机票/航班行程单",
                        scene_code="travel",
                        scene_label="差旅票据",
                        document_fields=[
                            {"key": "route", "label": "行程", "value": "上海-成都"},
                            {"key": "amount", "label": "金额", "value": "360元"},
                            {"key": "date", "label": "日期", "value": "2026-05-14"},
                        ],
                        warnings=[],
                    )
                )
        return OcrRecognizeBatchRead(
            total_file_count=len(files),
            success_count=len(documents),
            documents=documents,
        )

    monkeypatch.setattr(OcrService, "recognize_files", fake_recognize)
    monkeypatch.setattr(ExpenseClaimAttachmentStorage, "root", lambda self: tmp_path)

    with build_session() as db:
        manager = Employee(
            employee_no="E7200",
            name="李经理",
            email="manager-travel@example.com",
        )
        employee = Employee(
            employee_no="E7201",
            name="张三",
            email="emp-travel@example.com",
            grade="P4",
            manager=manager,
        )
        db.add_all([manager, employee])
        db.flush()

        claim = build_claim(expense_type="travel", location="上海")
        claim.reason = "上海客户现场出差"
        claim.employee = employee
        claim.employee_id = employee.id
        claim.items = [
            ExpenseClaimItem(
                id="travel-item-1",
                claim_id=claim.id,
                item_date=date(2026, 5, 13),
                item_type="travel",
                item_reason="赴上海客户现场",
                item_location="上海",
                item_amount=Decimal("480.00"),
                invoice_id=None,
            ),
            ExpenseClaimItem(
                id="travel-item-2",
                claim_id=claim.id,
                item_date=date(2026, 5, 14),
                item_type="travel",
                item_reason="赴上海客户现场",
                item_location="上海",
                item_amount=Decimal("360.00"),
                invoice_id=None,
            ),
        ]
        claim.amount = Decimal("840.00")
        claim.invoice_count = 0
        db.add(claim)
        db.commit()

        service = ExpenseClaimService(db)
        service.upload_claim_item_attachment(
            claim_id=claim.id,
            item_id="travel-item-1",
            filename="outbound.png",
            content=b"outbound-image",
            media_type="image/png",
            current_user=current_user,
        )
        service.upload_claim_item_attachment(
            claim_id=claim.id,
            item_id="travel-item-2",
            filename="onward.png",
            content=b"onward-image",
            media_type="image/png",
            current_user=current_user,
        )

        submitted = service.submit_claim(claim.id, current_user)

        assert submitted is not None
        assert submitted.status == "submitted"
        assert submitted.approval_stage == "直属领导审批"
        assert any(
            isinstance(flag, dict)
            and str(flag.get("source") or "").strip() == "submission_review"
            and (
                "多城市" in str(flag.get("message") or "")
                or "终点" in str(flag.get("message") or "")
            )
            for flag in list(submitted.risk_flags_json or [])
        )


def test_submit_claim_routes_hotel_amount_over_travel_policy_to_approval_with_review_flag(
    monkeypatch,
    tmp_path,
) -> None:
    current_user = CurrentUserContext(
        username="emp-hotel@example.com",
        name="张三",
        role_codes=[],
        is_admin=False,
    )

    def fake_recognize(
        self,
        files: list[tuple[str, bytes, str | None]],
    ) -> OcrRecognizeBatchRead:
        documents: list[OcrRecognizeDocumentRead] = []
        for filename, _, media_type in files:
            if filename == "beijing-trip.png":
                documents.append(
                    OcrRecognizeDocumentRead(
                        filename=filename,
                        media_type=media_type or "image/png",
                        text="电子行程单 2026-05-13 经济舱 武汉-北京 金额 520元 航班号 MU6101",
                        summary="武汉到北京机票",
                        avg_score=0.97,
                        line_count=1,
                        page_count=1,
                        document_type="flight_itinerary",
                        document_type_label="机票/航班行程单",
                        scene_code="travel",
                        scene_label="差旅票据",
                        document_fields=[
                            {"key": "route", "label": "行程", "value": "武汉-北京"},
                            {"key": "amount", "label": "金额", "value": "520元"},
                            {"key": "date", "label": "日期", "value": "2026-05-13"},
                        ],
                        warnings=[],
                    )
                )
            elif filename == "beijing-hotel.png":
                documents.append(
                    OcrRecognizeDocumentRead(
                        filename=filename,
                        media_type=media_type or "image/png",
                        text="北京全季酒店 1晚 金额 880元 2026-05-13",
                        summary="北京全季酒店住宿发票",
                        avg_score=0.98,
                        line_count=1,
                        page_count=1,
                        document_type="hotel_invoice",
                        document_type_label="酒店住宿票据",
                        scene_code="hotel",
                        scene_label="住宿票据",
                        document_fields=[
                            {"key": "merchant_name", "label": "商户", "value": "北京全季酒店"},
                            {"key": "amount", "label": "金额", "value": "880元"},
                            {"key": "date", "label": "日期", "value": "2026-05-13"},
                        ],
                        warnings=[],
                    )
                )
        return OcrRecognizeBatchRead(
            total_file_count=len(files),
            success_count=len(documents),
            documents=documents,
        )

    monkeypatch.setattr(OcrService, "recognize_files", fake_recognize)
    monkeypatch.setattr(ExpenseClaimAttachmentStorage, "root", lambda self: tmp_path)

    with build_session() as db:
        manager = Employee(
            employee_no="E7300",
            name="李经理",
            email="manager-hotel@example.com",
        )
        employee = Employee(
            employee_no="E7301",
            name="张三",
            email="emp-hotel@example.com",
            grade="P4",
            manager=manager,
        )
        db.add_all([manager, employee])
        db.flush()

        claim = build_claim(expense_type="travel", location="北京")
        claim.reason = "北京客户现场出差"
        claim.employee = employee
        claim.employee_id = employee.id
        claim.items = [
            ExpenseClaimItem(
                id="hotel-trip-item",
                claim_id=claim.id,
                item_date=date(2026, 5, 13),
                item_type="travel",
                item_reason="赴北京客户现场",
                item_location="北京",
                item_amount=Decimal("520.00"),
                invoice_id=None,
            ),
            ExpenseClaimItem(
                id="hotel-item",
                claim_id=claim.id,
                item_date=date(2026, 5, 13),
                item_type="hotel",
                item_reason="北京住宿",
                item_location="北京",
                item_amount=Decimal("880.00"),
                invoice_id=None,
            ),
        ]
        claim.amount = Decimal("1400.00")
        claim.invoice_count = 0
        db.add(claim)
        db.commit()

        service = ExpenseClaimService(db)
        service.upload_claim_item_attachment(
            claim_id=claim.id,
            item_id="hotel-trip-item",
            filename="beijing-trip.png",
            content=b"travel-image",
            media_type="image/png",
            current_user=current_user,
        )
        service.upload_claim_item_attachment(
            claim_id=claim.id,
            item_id="hotel-item",
            filename="beijing-hotel.png",
            content=b"hotel-image",
            media_type="image/png",
            current_user=current_user,
        )

        submitted = service.submit_claim(claim.id, current_user)

        assert submitted is not None
        assert submitted.status == "submitted"
        assert submitted.approval_stage == "直属领导审批"
        assert any(
            isinstance(flag, dict)
            and str(flag.get("source") or "").strip() == "submission_review"
            and "住宿标准" in str(flag.get("message") or "")
            for flag in list(submitted.risk_flags_json or [])
        )


def test_list_claims_scopes_to_current_user_id_even_when_names_duplicate() -> None:
    current_user = CurrentUserContext(
        username="zhangsan1@example.com",
        name="张三",
        role_codes=["manager"],
        is_admin=False,
    )

    with build_session() as db:
        employee_a = Employee(
            employee_no="E2001",
            name="张三",
            email="zhangsan1@example.com",
        )
        employee_b = Employee(
            employee_no="E2002",
            name="张三",
            email="zhangsan2@example.com",
        )
        db.add_all([employee_a, employee_b])
        db.flush()
        db.add_all(
            [
                ExpenseClaim(
                    claim_no="EXP-DUP-001",
                    employee_id=employee_a.id,
                    employee_name="张三",
                    department_name="市场部",
                    project_code="PRJ-A",
                    expense_type="travel",
                    reason="本人报销",
                    location="上海",
                    amount=Decimal("120.00"),
                    currency="CNY",
                    invoice_count=1,
                    occurred_at=datetime(2026, 5, 12, 9, 0, tzinfo=UTC),
                    submitted_at=datetime(2026, 5, 12, 10, 0, tzinfo=UTC),
                    status="submitted",
                    approval_stage="finance_review",
                    risk_flags_json=[],
                ),
                ExpenseClaim(
                    claim_no="EXP-DUP-002",
                    employee_id=employee_b.id,
                    employee_name="张三",
                    department_name="销售部",
                    project_code="PRJ-B",
                    expense_type="meal",
                    reason="他人报销",
                    location="杭州",
                    amount=Decimal("300.00"),
                    currency="CNY",
                    invoice_count=1,
                    occurred_at=datetime(2026, 5, 12, 12, 0, tzinfo=UTC),
                    submitted_at=datetime(2026, 5, 12, 13, 0, tzinfo=UTC),
                    status="approved",
                    approval_stage="completed",
                    risk_flags_json=[],
                ),
            ]
        )
        db.commit()

        claims = ExpenseClaimService(db).list_claims(current_user)

        assert len(claims) == 1
        assert claims[0].claim_no == "EXP-DUP-001"


def test_list_claims_allows_finance_to_view_all_records() -> None:
    current_user = CurrentUserContext(
        username="finance@example.com",
        name="财务",
        role_codes=["finance"],
        is_admin=False,
    )

    with build_session() as db:
        db.add_all(
            [
                ExpenseClaim(
                    claim_no="EXP-FIN-101",
                    employee_name="甲",
                    department_name="A部",
                    project_code="PRJ-A",
                    expense_type="travel",
                    reason="A 报销",
                    location="上海",
                    amount=Decimal("120.00"),
                    currency="CNY",
                    invoice_count=1,
                    occurred_at=datetime(2026, 5, 11, 9, 0, tzinfo=UTC),
                    submitted_at=datetime(2026, 5, 11, 10, 0, tzinfo=UTC),
                    status="submitted",
                    approval_stage="finance_review",
                    risk_flags_json=[],
                ),
                ExpenseClaim(
                    claim_no="EXP-FIN-102",
                    employee_name="乙",
                    department_name="B部",
                    project_code="PRJ-B",
                    expense_type="meal",
                    reason="B 报销",
                    location="杭州",
                    amount=Decimal("300.00"),
                    currency="CNY",
                    invoice_count=1,
                    occurred_at=datetime(2026, 5, 11, 12, 0, tzinfo=UTC),
                    submitted_at=datetime(2026, 5, 11, 13, 0, tzinfo=UTC),
                    status="approved",
                    approval_stage="completed",
                    risk_flags_json=[],
                ),
            ]
        )
        db.commit()

        claims = ExpenseClaimService(db).list_claims(current_user)

        assert len(claims) == 1
        assert claims[0].claim_no == "EXP-FIN-101"


def test_list_claims_allows_executive_to_view_all_records() -> None:
    current_user = CurrentUserContext(
        username="executive@example.com",
        name="高管",
        role_codes=["executive"],
        is_admin=False,
    )

    with build_session() as db:
        db.add_all(
            [
                ExpenseClaim(
                    claim_no="EXP-EXE-101",
                    employee_name="甲",
                    department_name="A部",
                    project_code="PRJ-A",
                    expense_type="travel",
                    reason="A 报销",
                    location="上海",
                    amount=Decimal("120.00"),
                    currency="CNY",
                    invoice_count=1,
                    occurred_at=datetime(2026, 5, 11, 9, 0, tzinfo=UTC),
                    submitted_at=datetime(2026, 5, 11, 10, 0, tzinfo=UTC),
                    status="submitted",
                    approval_stage="直属领导审批",
                    risk_flags_json=[],
                ),
                ExpenseClaim(
                    claim_no="EXP-EXE-102",
                    employee_name="乙",
                    department_name="B部",
                    project_code="PRJ-B",
                    expense_type="meal",
                    reason="B 报销",
                    location="杭州",
                    amount=Decimal("300.00"),
                    currency="CNY",
                    invoice_count=1,
                    occurred_at=datetime(2026, 5, 11, 12, 0, tzinfo=UTC),
                    submitted_at=datetime(2026, 5, 11, 13, 0, tzinfo=UTC),
                    status="approved",
                    approval_stage="completed",
                    risk_flags_json=[],
                ),
            ]
        )
        db.commit()

        claims = ExpenseClaimService(db).list_claims(current_user)

        assert len(claims) == 1
        assert claims[0].claim_no == "EXP-EXE-101"


def test_list_claims_keeps_own_archived_claim_for_finance_applicant() -> None:
    current_user = CurrentUserContext(
        username="finance@example.com",
        name="财务",
        role_codes=["finance"],
        is_admin=False,
    )

    with build_session() as db:
        db.add(
            ExpenseClaim(
                claim_no="EXP-FIN-OWN-ARCH",
                employee_name="财务",
                department_name="财务部",
                project_code="PRJ-FIN",
                expense_type="meal",
                reason="本人报销",
                location="上海",
                amount=Decimal("88.00"),
                currency="CNY",
                invoice_count=1,
                occurred_at=datetime(2026, 5, 12, 9, 0, tzinfo=UTC),
                submitted_at=datetime(2026, 5, 12, 10, 0, tzinfo=UTC),
                status="approved",
                approval_stage="归档入账",
                risk_flags_json=[],
            )
        )
        db.commit()

        claims = ExpenseClaimService(db).list_claims(current_user)

        assert len(claims) == 1
        assert claims[0].claim_no == "EXP-FIN-OWN-ARCH"


def test_list_archived_claims_returns_company_archived_records_for_finance() -> None:
    current_user = CurrentUserContext(
        username="finance@example.com",
        name="财务",
        role_codes=["finance"],
        is_admin=False,
    )

    with build_session() as db:
        db.add_all(
            [
                ExpenseClaim(
                    claim_no="EXP-ARCH-101",
                    employee_name="甲",
                    department_name="A部",
                    project_code="PRJ-A",
                    expense_type="travel",
                    reason="A 报销",
                    location="上海",
                    amount=Decimal("120.00"),
                    currency="CNY",
                    invoice_count=1,
                    occurred_at=datetime(2026, 5, 11, 9, 0, tzinfo=UTC),
                    submitted_at=datetime(2026, 5, 11, 10, 0, tzinfo=UTC),
                    status="approved",
                    approval_stage="归档入账",
                    risk_flags_json=[],
                ),
                ExpenseClaim(
                    claim_no="EXP-ARCH-102",
                    employee_name="乙",
                    department_name="B部",
                    project_code="PRJ-B",
                    expense_type="meal",
                    reason="B 报销",
                    location="杭州",
                    amount=Decimal("300.00"),
                    currency="CNY",
                    invoice_count=1,
                    occurred_at=datetime(2026, 5, 11, 12, 0, tzinfo=UTC),
                    submitted_at=datetime(2026, 5, 11, 13, 0, tzinfo=UTC),
                    status="submitted",
                    approval_stage="财务审批",
                    risk_flags_json=[],
                ),
                ExpenseClaim(
                    claim_no="AP-20260525120000-ABCDEFGH",
                    employee_name="丙",
                    department_name="C部",
                    project_code="PRJ-C",
                    expense_type="travel_application",
                    reason="C 申请",
                    location="成都",
                    amount=Decimal("800.00"),
                    currency="CNY",
                    invoice_count=0,
                    occurred_at=datetime(2026, 5, 11, 14, 0, tzinfo=UTC),
                    submitted_at=datetime(2026, 5, 11, 15, 0, tzinfo=UTC),
                    status="approved",
                    approval_stage="审批完成",
                    risk_flags_json=[],
                ),
                ExpenseClaim(
                    claim_no="AP-20260525123000-HGFEDCBA",
                    employee_name="丁",
                    department_name="D部",
                    project_code="PRJ-D",
                    expense_type="travel_application",
                    reason="D 申请",
                    location="北京",
                    amount=Decimal("500.00"),
                    currency="CNY",
                    invoice_count=0,
                    occurred_at=datetime(2026, 5, 11, 16, 0, tzinfo=UTC),
                    submitted_at=datetime(2026, 5, 11, 17, 0, tzinfo=UTC),
                    status="submitted",
                    approval_stage="直属领导审批",
                    risk_flags_json=[],
                ),
            ]
        )
        db.commit()

        claims = ExpenseClaimService(db).list_archived_claims(current_user)

        assert {claim.claim_no for claim in claims} == {
            "EXP-ARCH-101",
            "AP-20260525120000-ABCDEFGH",
        }


def test_list_archived_claims_returns_only_own_records_for_regular_employee() -> None:
    current_user = CurrentUserContext(
        username="zhangsan@example.com",
        name="张三",
        role_codes=["employee"],
        is_admin=False,
    )

    with build_session() as db:
        db.add_all(
            [
                ExpenseClaim(
                    claim_no="EXP-ARCH-EMP",
                    employee_name="张三",
                    department_name="研发部",
                    project_code="PRJ-EMP",
                    expense_type="travel",
                    reason="本人报销",
                    location="北京",
                    amount=Decimal("200.00"),
                    currency="CNY",
                    invoice_count=1,
                    occurred_at=datetime(2026, 5, 10, 9, 0, tzinfo=UTC),
                    submitted_at=datetime(2026, 5, 10, 10, 0, tzinfo=UTC),
                    status="approved",
                    approval_stage="归档入账",
                    risk_flags_json=[],
                ),
                ExpenseClaim(
                    claim_no="AP-20260525130000-ABCDEFGH",
                    employee_name="李四",
                    department_name="研发部",
                    project_code="PRJ-EMP",
                    expense_type="travel_application",
                    reason="他人申请",
                    location="上海",
                    amount=Decimal("500.00"),
                    currency="CNY",
                    invoice_count=0,
                    occurred_at=datetime(2026, 5, 10, 9, 0, tzinfo=UTC),
                    submitted_at=datetime(2026, 5, 10, 10, 0, tzinfo=UTC),
                    status="approved",
                    approval_stage="审批完成",
                    risk_flags_json=[],
                ),
            ]
        )
        db.commit()

        claims = ExpenseClaimService(db).list_archived_claims(current_user)

        assert [claim.claim_no for claim in claims] == ["EXP-ARCH-EMP"]


def test_finance_can_return_but_cannot_delete_submitted_claim() -> None:
    current_user = CurrentUserContext(
        username="finance@example.com",
        name="财务",
        role_codes=["finance"],
        is_admin=False,
    )

    with build_session() as db:
        claim = ExpenseClaim(
            claim_no="EXP-RET-101",
            employee_name="张三",
            department_name="市场部",
            project_code="PRJ-A",
            expense_type="travel",
            reason="差旅报销",
            location="上海",
            amount=Decimal("120.00"),
            currency="CNY",
            invoice_count=1,
            occurred_at=datetime(2026, 5, 11, 9, 0, tzinfo=UTC),
            submitted_at=datetime(2026, 5, 11, 10, 0, tzinfo=UTC),
            status="submitted",
            approval_stage="直属领导审批",
            risk_flags_json=[],
        )
        db.add(claim)
        db.commit()
        claim_id = claim.id

        service = ExpenseClaimService(db)
        returned = service.return_claim(claim_id, current_user, reason="资料不完整")

        assert returned is not None
        assert returned.status == "returned"
        assert returned.approval_stage == "待提交"
        assert any(
            isinstance(flag, dict)
            and flag.get("source") == "manual_return"
            and flag.get("message") == "资料不完整"
            for flag in returned.risk_flags_json
        )

        with pytest.raises(ValueError, match="只有高级财务人员可以删除"):
            service.delete_claim(claim_id, current_user)

        assert db.get(ExpenseClaim, claim_id) is not None


def test_executive_can_delete_submitted_claim() -> None:
    current_user = CurrentUserContext(
        username="executive-delete@example.com",
        name="高管",
        role_codes=["executive"],
        is_admin=False,
    )

    with build_session() as db:
        claim = ExpenseClaim(
            claim_no="EXP-DEL-EXEC-101",
            employee_name="张三",
            department_name="市场部",
            project_code="PRJ-A",
            expense_type="travel",
            reason="差旅报销",
            location="上海",
            amount=Decimal("120.00"),
            currency="CNY",
            invoice_count=1,
            occurred_at=datetime(2026, 5, 11, 9, 0, tzinfo=UTC),
            submitted_at=datetime(2026, 5, 11, 10, 0, tzinfo=UTC),
            status="submitted",
            approval_stage="财务审批",
            risk_flags_json=[],
        )
        db.add(claim)
        db.commit()
        claim_id = claim.id

        deleted = ExpenseClaimService(db).delete_claim(claim_id, current_user)

        assert deleted is not None
        assert deleted.claim_no == "EXP-DEL-EXEC-101"
        assert db.get(ExpenseClaim, claim_id) is None


def test_executive_cannot_delete_archived_claim() -> None:
    current_user = CurrentUserContext(
        username="executive-archive-delete@example.com",
        name="高管",
        role_codes=["executive"],
        is_admin=False,
    )

    with build_session() as db:
        claim = ExpenseClaim(
            claim_no="EXP-DEL-ARCHIVE-101",
            employee_name="张三",
            department_name="市场部",
            project_code="PRJ-A",
            expense_type="travel",
            reason="差旅报销",
            location="上海",
            amount=Decimal("120.00"),
            currency="CNY",
            invoice_count=1,
            occurred_at=datetime(2026, 5, 11, 9, 0, tzinfo=UTC),
            submitted_at=datetime(2026, 5, 11, 10, 0, tzinfo=UTC),
            status="approved",
            approval_stage="归档入账",
            risk_flags_json=[],
        )
        db.add(claim)
        db.commit()
        claim_id = claim.id

        with pytest.raises(ValueError, match="已归档单据不能删除"):
            ExpenseClaimService(db).delete_claim(claim_id, current_user)

        assert db.get(ExpenseClaim, claim_id) is not None


def test_admin_can_delete_archived_claim() -> None:
    current_user = CurrentUserContext(
        username="superadmin",
        name="系统管理员",
        role_codes=["manager"],
        is_admin=True,
    )

    with build_session() as db:
        claim = ExpenseClaim(
            claim_no="EXP-DEL-ARCHIVE-102",
            employee_name="张三",
            department_name="市场部",
            project_code="PRJ-A",
            expense_type="travel",
            reason="差旅报销",
            location="上海",
            amount=Decimal("120.00"),
            currency="CNY",
            invoice_count=1,
            occurred_at=datetime(2026, 5, 11, 9, 0, tzinfo=UTC),
            submitted_at=datetime(2026, 5, 11, 10, 0, tzinfo=UTC),
            status="approved",
            approval_stage="归档入账",
            risk_flags_json=[],
        )
        db.add(claim)
        db.commit()
        claim_id = claim.id

        deleted = ExpenseClaimService(db).delete_claim(claim_id, current_user)

        assert deleted is not None
        assert deleted.claim_no == "EXP-DEL-ARCHIVE-102"
        assert db.get(ExpenseClaim, claim_id) is None


def test_direct_manager_can_return_subordinate_claim_to_pending_submission() -> None:
    current_user = CurrentUserContext(
        username="manager-return@example.com",
        name="李经理",
        role_codes=["manager"],
        is_admin=False,
    )

    with build_session() as db:
        manager = Employee(
            employee_no="E8100",
            name="李经理",
            email="manager-return@example.com",
        )
        employee = Employee(
            employee_no="E8101",
            name="张三",
            email="zhangsan-return@example.com",
            manager=manager,
        )
        db.add_all([manager, employee])
        db.flush()
        claim = ExpenseClaim(
            claim_no="EXP-RET-201",
            employee_id=employee.id,
            employee_name="张三",
            department_name="市场部",
            project_code="PRJ-A",
            expense_type="transport",
            reason="交通报销",
            location="上海",
            amount=Decimal("66.00"),
            currency="CNY",
            invoice_count=1,
            occurred_at=datetime(2026, 5, 12, 9, 0, tzinfo=UTC),
            submitted_at=datetime(2026, 5, 12, 10, 0, tzinfo=UTC),
            status="submitted",
            approval_stage="直属领导审批",
            risk_flags_json=[],
        )
        db.add(claim)
        db.commit()
        claim_id = claim.id

        returned = ExpenseClaimService(db).return_claim(claim_id, current_user, reason="请补充行程说明")

        assert returned is not None
        assert returned.status == "returned"
        assert returned.approval_stage == "待提交"
        assert returned.submitted_at is None
        assert any(
            isinstance(flag, dict)
            and flag.get("source") == "manual_return"
            and flag.get("message") == "请补充行程说明"
            for flag in returned.risk_flags_json
        )


def test_direct_manager_can_approve_subordinate_claim_to_finance_review() -> None:
    current_user = CurrentUserContext(
        username="manager-approve@example.com",
        name="李经理",
        role_codes=["manager"],
        is_admin=False,
    )

    with build_session() as db:
        manager = Employee(
            employee_no="E8110",
            name="李经理",
            email="manager-approve@example.com",
        )
        employee = Employee(
            employee_no="E8111",
            name="张三",
            email="zhangsan-approve@example.com",
            manager=manager,
        )
        db.add_all([manager, employee])
        db.flush()
        claim = ExpenseClaim(
            claim_no="EXP-APP-201",
            employee_id=employee.id,
            employee_name="张三",
            department_name="市场部",
            project_code="PRJ-A",
            expense_type="transport",
            reason="交通报销",
            location="上海",
            amount=Decimal("66.00"),
            currency="CNY",
            invoice_count=1,
            occurred_at=datetime(2026, 5, 12, 9, 0, tzinfo=UTC),
            submitted_at=datetime(2026, 5, 12, 10, 0, tzinfo=UTC),
            status="submitted",
            approval_stage="直属领导审批",
            risk_flags_json=[],
        )
        db.add(claim)
        db.commit()
        claim_id = claim.id

        approved = ExpenseClaimService(db).approve_claim(
            claim_id,
            current_user,
            opinion="情况属实，同意报销。",
        )

        assert approved is not None
        assert approved.status == "submitted"
        assert approved.approval_stage == "财务审批"
        assert approved.submitted_at is not None
        assert any(
            isinstance(flag, dict)
            and flag.get("source") == "manual_approval"
            and flag.get("event_type") == "expense_claim_approval"
            and flag.get("opinion") == "情况属实，同意报销。"
            and flag.get("previous_approval_stage") == "直属领导审批"
            and flag.get("next_approval_stage") == "财务审批"
            for flag in approved.risk_flags_json
        )


def test_application_submit_skips_ai_review_and_receipt_requirements(monkeypatch: pytest.MonkeyPatch) -> None:
    current_user = CurrentUserContext(
        username="application-owner@example.com",
        name="张三",
        role_codes=["employee"],
        is_admin=True,
    )

    with build_session() as db:
        claim = ExpenseClaim(
            claim_no="APP-20260525-SUBMIT",
            employee_name="张三",
            department_name="交付部",
            project_code="PRJ-A",
            expense_type="travel_application",
            reason="支撑国网服务器上线部署",
            location="上海",
            amount=Decimal("12000.00"),
            currency="CNY",
            invoice_count=0,
            occurred_at=datetime(2026, 5, 25, 9, 0, tzinfo=UTC),
            submitted_at=None,
            status="draft",
            approval_stage="待提交",
            risk_flags_json=[
                {
                    "source": "submission_review",
                    "severity": "medium",
                    "message": "旧 AI 预审提示不应保留到申请单提交结果。",
                }
            ],
        )
        db.add(claim)
        db.commit()
        claim_id = claim.id
        service = ExpenseClaimService(db)

        def fail_ai_review(_claim: ExpenseClaim) -> dict[str, object]:
            raise AssertionError("费用申请提交不应进入 AI 预审")

        monkeypatch.setattr(service, "_run_ai_submission_review", fail_ai_review)

        submitted = service.submit_claim(claim_id, current_user)

        assert submitted is not None
        assert submitted.status == "submitted"
        assert submitted.approval_stage == "直属领导审批"
        assert submitted.invoice_count == 0
        assert submitted.items == []
        assert not any(
            isinstance(flag, dict) and flag.get("source") == "submission_review"
            for flag in submitted.risk_flags_json
        )
        assert any(
            isinstance(flag, dict)
            and flag.get("source") == "application_submission"
            and flag.get("event_type") == "expense_application_submission"
            and flag.get("next_approval_stage") == "直属领导审批"
            for flag in submitted.risk_flags_json
        )


def test_application_submit_reserves_budget_once() -> None:
    current_user = CurrentUserContext(
        username="application-budget-owner@example.com",
        name="张三",
        role_codes=["employee"],
        is_admin=True,
    )

    with build_session() as db:
        _seed_budget_allocation(
            db,
            department_id="dept-budget",
            department_name="交付部",
            amount=Decimal("50000.00"),
        )
        claim = ExpenseClaim(
            claim_no="APP-20260525-BUDGET",
            employee_id="emp-budget",
            employee_name="张三",
            department_id="dept-budget",
            department_name="交付部",
            project_code=None,
            expense_type="travel_application",
            reason="客户现场交付",
            location="上海",
            amount=Decimal("12000.00"),
            currency="CNY",
            invoice_count=0,
            occurred_at=datetime(2026, 5, 25, 9, 0, tzinfo=UTC),
            submitted_at=None,
            status="draft",
            approval_stage="待提交",
            risk_flags_json=[],
        )
        db.add(claim)
        db.commit()

        submitted = ExpenseClaimService(db).submit_claim(claim.id, current_user)

        assert submitted is not None
        reservations = db.query(BudgetReservation).all()
        assert len(reservations) == 1
        assert reservations[0].source_type == "application"
        assert reservations[0].source_id == claim.id
        assert reservations[0].amount == Decimal("12000.00")
        transactions = db.query(BudgetTransaction).all()
        assert any(item.transaction_type == "reserve" for item in transactions)
        assert any(
            isinstance(flag, dict)
            and flag.get("source") == "budget_control"
            and flag.get("event_type") == "budget_reserved"
            for flag in submitted.risk_flags_json
        )


def test_application_submit_blocks_when_budget_insufficient_without_state_change() -> None:
    current_user = CurrentUserContext(
        username="application-budget-block@example.com",
        name="张三",
        role_codes=["employee"],
        is_admin=True,
    )

    with build_session() as db:
        _seed_budget_allocation(
            db,
            department_id="dept-budget-block",
            department_name="交付部",
            amount=Decimal("1000.00"),
        )
        claim = ExpenseClaim(
            claim_no="APP-20260525-BLOCK",
            employee_id="emp-budget-block",
            employee_name="张三",
            department_id="dept-budget-block",
            department_name="交付部",
            project_code=None,
            expense_type="travel_application",
            reason="客户现场交付",
            location="上海",
            amount=Decimal("12000.00"),
            currency="CNY",
            invoice_count=0,
            occurred_at=datetime(2026, 5, 25, 9, 0, tzinfo=UTC),
            submitted_at=None,
            status="draft",
            approval_stage="待提交",
            risk_flags_json=[],
        )
        db.add(claim)
        db.commit()

        with pytest.raises(ValueError):
            ExpenseClaimService(db).submit_claim(claim.id, current_user)

        db.refresh(claim)
        assert claim.status == "draft"
        assert db.query(BudgetReservation).count() == 0
        assert db.query(BudgetTransaction).count() == 0


def test_application_submit_skips_budget_for_non_demo_subject() -> None:
    current_user = CurrentUserContext(
        username="application-budget-skip@example.com",
        name="张三",
        role_codes=["employee"],
        is_admin=True,
    )

    with build_session() as db:
        _seed_budget_allocation(
            db,
            department_id="dept-budget-skip",
            department_name="交付部",
            amount=Decimal("1000.00"),
        )
        claim = ExpenseClaim(
            claim_no="APP-20260525-SKIP",
            employee_id="emp-budget-skip",
            employee_name="张三",
            department_id="dept-budget-skip",
            department_name="交付部",
            project_code=None,
            expense_type="software_application",
            reason="采购演示软件服务",
            location="深圳",
            amount=Decimal("12000.00"),
            currency="CNY",
            invoice_count=0,
            occurred_at=datetime(2026, 5, 25, 9, 0, tzinfo=UTC),
            submitted_at=None,
            status="draft",
            approval_stage="待提交",
            risk_flags_json=[],
        )
        db.add(claim)
        db.commit()

        submitted = ExpenseClaimService(db).submit_claim(claim.id, current_user)

        assert submitted is not None
        assert submitted.status == "submitted"
        assert submitted.approval_stage == "直属领导审批"
        assert db.query(BudgetReservation).count() == 0
        assert db.query(BudgetTransaction).count() == 0
        assert not any(
            isinstance(flag, dict) and str(flag.get("source") or "").strip() == "budget_control"
            for flag in submitted.risk_flags_json
        )


def test_direct_manager_can_approve_application_claim_to_reimbursement_draft() -> None:
    current_user = CurrentUserContext(
        username="manager-application-approve@example.com",
        name="李经理",
        role_codes=["manager"],
        is_admin=False,
    )

    with build_session() as db:
        manager = Employee(
            employee_no="E8112",
            name="李经理",
            email="manager-application-approve@example.com",
        )
        employee = Employee(
            employee_no="E8113",
            name="张三",
            email="zhangsan-application-approve@example.com",
            manager=manager,
        )
        db.add_all([manager, employee])
        db.flush()
        claim = ExpenseClaim(
            claim_no="APP-20260525-APPROVE",
            employee_id=employee.id,
            employee_name="张三",
            department_name="交付部",
            project_code="PRJ-A",
            expense_type="travel_application",
            reason="支撑国网服务器上线部署",
            location="上海",
            amount=Decimal("12000.00"),
            currency="CNY",
            invoice_count=0,
            occurred_at=datetime(2026, 5, 25, 9, 0, tzinfo=UTC),
            submitted_at=datetime(2026, 5, 25, 10, 0, tzinfo=UTC),
            status="submitted",
            approval_stage="直属领导审批",
            risk_flags_json=[],
        )
        db.add(claim)
        db.commit()
        claim_id = claim.id

        approved = ExpenseClaimService(db).approve_claim(
            claim_id,
            current_user,
            opinion="业务必要，同意申请。",
        )

        assert approved is not None
        assert approved.status == "approved"
        assert approved.approval_stage == "审批完成"
        archived_claims = ExpenseClaimService(db).list_archived_claims(
            CurrentUserContext(
                username="finance-archive@example.com",
                name="财务归档员",
                role_codes=["finance"],
                is_admin=False,
            )
        )
        assert any(claim.claim_no == "APP-20260525-APPROVE" for claim in archived_claims)
        generated_draft = db.query(ExpenseClaim).filter(ExpenseClaim.claim_no.like("RE-%")).one()
        assert generated_draft.status == "draft"
        assert generated_draft.approval_stage == "待提交"
        assert generated_draft.expense_type == "travel"
        assert generated_draft.employee_id == employee.id
        assert generated_draft.employee_name == "张三"
        assert generated_draft.department_name == "交付部"
        assert generated_draft.reason == "支撑国网服务器上线部署"
        assert generated_draft.location == "上海"
        assert generated_draft.amount == Decimal("12000.00")
        assert generated_draft.invoice_count == 0
        assert generated_draft.items == []
        assert any(
            isinstance(flag, dict)
            and flag.get("source") == "application_handoff"
            and flag.get("event_type") == "expense_application_to_reimbursement_draft"
            and flag.get("application_claim_no") == "APP-20260525-APPROVE"
            and flag.get("leader_opinion") == "业务必要，同意申请。"
            for flag in generated_draft.risk_flags_json
        )
        assert any(
            isinstance(flag, dict)
            and flag.get("source") == "manual_approval"
            and flag.get("event_type") == "expense_application_approval"
            and flag.get("opinion") == "业务必要，同意申请。"
            and flag.get("previous_approval_stage") == "直属领导审批"
            and flag.get("next_status") == "approved"
            and flag.get("next_approval_stage") == "审批完成"
            and flag.get("generated_draft_claim_id") == generated_draft.id
            and flag.get("generated_draft_claim_no") == generated_draft.claim_no
            for flag in approved.risk_flags_json
        )


def test_application_approval_transfers_budget_reservation_to_reimbursement_draft() -> None:
    owner = CurrentUserContext(
        username="application-budget-owner-approve@example.com",
        name="张三",
        role_codes=["employee"],
        is_admin=False,
    )
    manager_user = CurrentUserContext(
        username="manager-application-budget@example.com",
        name="李经理",
        role_codes=["manager"],
        is_admin=False,
    )

    with build_session() as db:
        manager = Employee(
            employee_no="M-BUDGET-APP",
            name="李经理",
            email="manager-application-budget@example.com",
        )
        employee = Employee(
            employee_no="E-BUDGET-APP",
            name="张三",
            email="application-budget-owner-approve@example.com",
            manager=manager,
        )
        db.add_all([manager, employee])
        db.flush()
        _seed_budget_allocation(
            db,
            department_id="dept-budget-transfer",
            department_name="交付部",
            amount=Decimal("50000.00"),
        )
        claim = ExpenseClaim(
            claim_no="APP-20260525-TRANSFER",
            employee_id=employee.id,
            employee_name="张三",
            department_id="dept-budget-transfer",
            department_name="交付部",
            project_code=None,
            expense_type="travel_application",
            reason="客户现场交付",
            location="上海",
            amount=Decimal("12000.00"),
            currency="CNY",
            invoice_count=0,
            occurred_at=datetime(2026, 5, 25, 9, 0, tzinfo=UTC),
            submitted_at=None,
            status="draft",
            approval_stage="待提交",
            risk_flags_json=[],
        )
        db.add(claim)
        db.commit()
        service = ExpenseClaimService(db)

        service.submit_claim(claim.id, owner)
        approved = service.approve_claim(claim.id, manager_user, opinion="同意申请")
        generated_draft = db.query(ExpenseClaim).filter(ExpenseClaim.claim_no.like("RE-%")).one()
        reservation = db.query(BudgetReservation).one()

        assert approved is not None
        assert reservation.source_type == "claim"
        assert reservation.source_id == generated_draft.id
        assert reservation.source_no == generated_draft.claim_no
        assert any(item.transaction_type == "transfer" for item in db.query(BudgetTransaction).all())
        assert any(
            isinstance(flag, dict)
            and flag.get("event_type") == "budget_reservation_transferred"
            for flag in generated_draft.risk_flags_json
        )


def test_direct_manager_approval_requires_leader_opinion() -> None:
    current_user = CurrentUserContext(
        username="manager-application-required-opinion@example.com",
        name="李经理",
        role_codes=["manager"],
        is_admin=False,
    )

    with build_session() as db:
        manager = Employee(
            employee_no="E8122",
            name="李经理",
            email="manager-application-required-opinion@example.com",
        )
        employee = Employee(
            employee_no="E8123",
            name="张三",
            email="zhangsan-application-required-opinion@example.com",
            manager=manager,
        )
        db.add_all([manager, employee])
        db.flush()
        claim = ExpenseClaim(
            claim_no="APP-20260525-REQUIRE-OPINION",
            employee_id=employee.id,
            employee_name="张三",
            department_name="交付部",
            project_code="PRJ-A",
            expense_type="travel_application",
            reason="支撑国网服务器上线部署",
            location="上海",
            amount=Decimal("12000.00"),
            currency="CNY",
            invoice_count=0,
            occurred_at=datetime(2026, 5, 25, 9, 0, tzinfo=UTC),
            submitted_at=datetime(2026, 5, 25, 10, 0, tzinfo=UTC),
            status="submitted",
            approval_stage="直属领导审批",
            risk_flags_json=[],
        )
        db.add(claim)
        db.commit()
        claim_id = claim.id

        with pytest.raises(ValueError, match="领导审核意见不能为空"):
            ExpenseClaimService(db).approve_claim(
                claim_id,
                current_user,
                opinion="   ",
            )

        db.refresh(claim)
        assert claim.status == "submitted"
        assert claim.approval_stage == "直属领导审批"
        assert db.query(ExpenseClaim).filter(ExpenseClaim.claim_no.like("RE-%")).count() == 0


def test_finance_approve_reimbursement_consumes_budget_reservation() -> None:
    current_user = CurrentUserContext(
        username="finance-budget-approve@example.com",
        name="财务",
        role_codes=["finance"],
        is_admin=False,
    )

    with build_session() as db:
        allocation = _seed_budget_allocation(
            db,
            department_id="dept-finance-budget",
            department_name="交付部",
            amount=Decimal("50000.00"),
        )
        claim = ExpenseClaim(
            claim_no="RE-20260525-BUDGET",
            employee_id="emp-finance-budget",
            employee_name="张三",
            department_id="dept-finance-budget",
            department_name="交付部",
            project_code=None,
            expense_type="travel",
            reason="客户现场交付",
            location="上海",
            amount=Decimal("12000.00"),
            currency="CNY",
            invoice_count=1,
            occurred_at=datetime(2026, 5, 25, 9, 0, tzinfo=UTC),
            submitted_at=datetime(2026, 5, 25, 10, 0, tzinfo=UTC),
            status="submitted",
            approval_stage="财务审批",
            risk_flags_json=[],
        )
        db.add(claim)
        db.flush()
        reservation = BudgetReservation(
            reservation_no=f"BRS-TEST-{uuid.uuid4().hex[:8]}",
            allocation_id=allocation.id,
            source_type="claim",
            source_id=claim.id,
            source_no=claim.claim_no,
            source_status="active",
            amount=Decimal("12000.00"),
            context_json={},
        )
        db.add(reservation)
        db.commit()

        approved = ExpenseClaimService(db).approve_claim(claim.id, current_user, opinion="同意入账")

        assert approved is not None
        db.refresh(reservation)
        assert reservation.source_status == "consumed"
        assert reservation.consumed_amount == Decimal("12000.00")
        assert db.query(BudgetTransaction).filter(BudgetTransaction.transaction_type == "consume").count() == 1
        assert any(
            isinstance(flag, dict)
            and flag.get("source") == "budget_control"
            and flag.get("event_type") == "budget_consumed"
            for flag in approved.risk_flags_json
        )


def test_finance_can_approve_claim_to_archive_stage() -> None:
    current_user = CurrentUserContext(
        username="finance-approve@example.com",
        name="财务复核",
        role_codes=["finance"],
        is_admin=False,
    )

    with build_session() as db:
        claim = ExpenseClaim(
            claim_no="EXP-FIN-APP-201",
            employee_name="张三",
            department_name="市场部",
            project_code="PRJ-A",
            expense_type="transport",
            reason="交通报销",
            location="上海",
            amount=Decimal("66.00"),
            currency="CNY",
            invoice_count=1,
            occurred_at=datetime(2026, 5, 12, 9, 0, tzinfo=UTC),
            submitted_at=datetime(2026, 5, 12, 10, 0, tzinfo=UTC),
            status="submitted",
            approval_stage="财务审批",
            risk_flags_json=[],
        )
        db.add(claim)
        db.commit()
        claim_id = claim.id

        approved = ExpenseClaimService(db).approve_claim(
            claim_id,
            current_user,
            opinion="票据与明细一致，同意入账。",
        )

        assert approved is not None
        assert approved.status == "approved"
        assert approved.approval_stage == "归档入账"
        assert any(
            isinstance(flag, dict)
            and flag.get("source") == "finance_approval"
            and flag.get("event_type") == "expense_claim_finance_approval"
            and flag.get("opinion") == "票据与明细一致，同意入账。"
            and flag.get("previous_approval_stage") == "财务审批"
            and flag.get("next_approval_stage") == "归档入账"
            for flag in approved.risk_flags_json
        )


def test_return_claim_rejects_already_returned_claim_without_adding_event() -> None:
    current_user = CurrentUserContext(
        username="finance-returned@example.com",
        name="财务",
        role_codes=["finance"],
        is_admin=False,
    )
    return_flag = {
        "source": "manual_return",
        "return_event_id": "return-event-existing",
        "message": "请补充附件。",
        "return_count": 1,
        "return_stage": "直属领导审批",
        "return_stage_key": "direct_manager",
    }

    with build_session() as db:
        claim = ExpenseClaim(
            claim_no="EXP-RET-202",
            employee_name="张三",
            department_name="市场部",
            project_code="PRJ-A",
            expense_type="transport",
            reason="交通报销",
            location="上海",
            amount=Decimal("66.00"),
            currency="CNY",
            invoice_count=1,
            occurred_at=datetime(2026, 5, 12, 9, 0, tzinfo=UTC),
            submitted_at=None,
            status="returned",
            approval_stage="待提交",
            risk_flags_json=[return_flag],
        )
        db.add(claim)
        db.commit()
        claim_id = claim.id

        with pytest.raises(ValueError, match="无需重复退回"):
            ExpenseClaimService(db).return_claim(claim_id, current_user, reason="重复退回")

        db.refresh(claim)
        manual_returns = [
            flag
            for flag in list(claim.risk_flags_json or [])
            if isinstance(flag, dict) and flag.get("source") == "manual_return"
        ]
        assert manual_returns == [return_flag]


def test_return_claim_records_each_return_event_with_stage_reason_and_counts() -> None:
    current_user = CurrentUserContext(
        username="finance-return@example.com",
        name="财务复核",
        role_codes=["finance"],
        is_admin=False,
    )

    with build_session() as db:
        claim = ExpenseClaim(
            claim_no="EXP-RET-301",
            employee_name="张三",
            department_name="市场部",
            project_code="PRJ-A",
            expense_type="transport",
            reason="交通报销",
            location="上海",
            amount=Decimal("66.00"),
            currency="CNY",
            invoice_count=1,
            occurred_at=datetime(2026, 5, 12, 9, 0, tzinfo=UTC),
            submitted_at=datetime(2026, 5, 12, 10, 0, tzinfo=UTC),
            status="submitted",
            approval_stage="直属领导审批",
            risk_flags_json=[],
        )
        db.add(claim)
        db.commit()
        claim_id = claim.id

        service = ExpenseClaimService(db)
        first_returned = service.return_claim(
            claim_id,
            current_user,
            reason="发票金额与明细金额不一致，请重新核对。",
            reason_codes=["invoice_mismatch", "business_explanation"],
        )
        assert first_returned is not None

        first_returned.status = "submitted"
        first_returned.approval_stage = "财务审批"
        first_returned.submitted_at = datetime(2026, 5, 12, 11, 0, tzinfo=UTC)
        db.commit()

        second_returned = service.return_claim(
            claim_id,
            current_user,
            reason="超标说明仍不完整，请补充制度例外依据。",
            reason_codes=["over_policy"],
        )

        assert second_returned is not None
        return_events = [
            flag
            for flag in list(second_returned.risk_flags_json or [])
            if isinstance(flag, dict) and flag.get("source") == "manual_return"
        ]
        assert len(return_events) == 2
        assert return_events[0]["return_count"] == 1
        assert return_events[0]["stage_return_count"] == 1
        assert return_events[0]["return_stage"] == "直属领导审批"
        assert return_events[0]["reason_codes"] == ["invoice_mismatch", "business_explanation"]
        assert return_events[0]["risk_points"] == ["票据类型/金额与明细不一致", "业务事由/地点/人员信息不完整"]
        assert return_events[0]["reason"] == "发票金额与明细金额不一致，请重新核对。"
        assert return_events[0]["operator_role_codes"] == ["finance"]
        assert return_events[1]["return_count"] == 2
        assert return_events[1]["stage_return_count"] == 1
        assert return_events[1]["return_stage"] == "财务审批"
        assert return_events[1]["risk_points"] == ["超出制度标准或缺少超标说明"]


def test_submit_returned_claim_preserves_manual_return_events() -> None:
    current_user = CurrentUserContext(
        username="emp-submit-returned@example.com",
        name="张三",
        role_codes=[],
        is_admin=False,
    )
    return_flag = {
        "source": "manual_return",
        "return_event_id": "return-event-submit",
        "message": "第一次退回：业务说明不完整。",
        "reason": "业务说明不完整。",
        "return_count": 1,
        "return_stage": "直属领导审批",
        "return_stage_key": "direct_manager",
        "risk_points": ["业务事由/地点/人员信息不完整"],
    }

    with build_session() as db:
        manager = Employee(
            employee_no="E8200",
            name="李经理",
            email="manager-submit-returned@example.com",
        )
        employee = Employee(
            employee_no="E8201",
            name="张三",
            email="emp-submit-returned@example.com",
            manager=manager,
        )
        claim = build_claim(expense_type="office", location="上海")
        claim.employee = employee
        claim.employee_id = employee.id
        claim.employee_name = "张三"
        claim.department_name = "市场部"
        claim.status = "returned"
        claim.approval_stage = "待提交"
        claim.risk_flags_json = [return_flag]
        db.add_all([manager, employee, claim])
        db.commit()
        conversation = AgentConversationService(db).get_or_create_conversation(
            conversation_id=None,
            user_id=current_user.username,
            source="user_message",
            context_json={
                "session_type": "expense",
                "draft_claim_id": claim.id,
            },
        )
        conversation_id = conversation.conversation_id

        submitted = ExpenseClaimService(db).submit_claim(claim.id, current_user)

        assert submitted is not None
        assert submitted.status == "submitted"
        assert submitted.approval_stage == "直属领导审批"
        assert any(
            isinstance(flag, dict)
            and flag.get("source") == "manual_return"
            and flag.get("return_event_id") == "return-event-submit"
            for flag in list(submitted.risk_flags_json or [])
        )
        assert AgentConversationService(db).get_conversation(conversation_id) is None


def test_manager_personal_claims_exclude_subordinate_pending_approval_claims() -> None:
    current_user = CurrentUserContext(
        username="manager-personal@example.com",
        name="李经理",
        role_codes=["manager"],
        is_admin=False,
    )

    with build_session() as db:
        manager = Employee(
            employee_no="E8300",
            name="李经理",
            email="manager-personal@example.com",
        )
        employee = Employee(
            employee_no="E8301",
            name="张三",
            email="zhangsan-personal@example.com",
            manager=manager,
        )
        db.add_all([manager, employee])
        db.flush()
        db.add_all(
            [
                ExpenseClaim(
                    claim_no="EXP-MGR-OWN",
                    employee_id=manager.id,
                    employee_name="李经理",
                    department_name="市场部",
                    project_code="PRJ-MGR",
                    expense_type="office",
                    reason="本人报销",
                    location="上海",
                    amount=Decimal("88.00"),
                    currency="CNY",
                    invoice_count=1,
                    occurred_at=datetime(2026, 5, 12, 9, 0, tzinfo=UTC),
                    submitted_at=None,
                    status="draft",
                    approval_stage="待提交",
                    risk_flags_json=[],
                ),
                ExpenseClaim(
                    claim_no="EXP-MGR-SUB",
                    employee_id=employee.id,
                    employee_name="张三",
                    department_name="市场部",
                    project_code="PRJ-MGR",
                    expense_type="transport",
                    reason="下属待审批报销",
                    location="上海",
                    amount=Decimal("66.00"),
                    currency="CNY",
                    invoice_count=1,
                    occurred_at=datetime(2026, 5, 12, 10, 0, tzinfo=UTC),
                    submitted_at=datetime(2026, 5, 12, 11, 0, tzinfo=UTC),
                    status="submitted",
                    approval_stage="直属领导审批",
                    risk_flags_json=[],
                ),
            ]
        )
        db.commit()

        service = ExpenseClaimService(db)
        personal_claims = service.list_claims(current_user)
        approval_claims = service.list_approval_claims(current_user)

        assert [claim.claim_no for claim in personal_claims] == ["EXP-MGR-OWN"]
        assert [claim.claim_no for claim in approval_claims] == ["EXP-MGR-SUB"]


def test_list_approval_claims_allows_direct_manager_to_view_pending_claims_for_approval() -> None:
    current_user = CurrentUserContext(
        username="manager@example.com",
        name="李经理",
        role_codes=["manager"],
        is_admin=False,
    )

    with build_session() as db:
        manager = Employee(
            employee_no="E8000",
            name="李经理",
            email="manager@example.com",
        )
        employee = Employee(
            employee_no="E8001",
            name="张三",
            email="zhangsan@example.com",
            manager=manager,
        )
        outsider_manager = Employee(
            employee_no="E8002",
            name="王经理",
            email="other-manager@example.com",
        )
        outsider = Employee(
            employee_no="E8003",
            name="李四",
            email="lisi@example.com",
            manager=outsider_manager,
        )
        db.add_all([manager, employee, outsider_manager, outsider])
        db.flush()
        db.add_all(
            [
                ExpenseClaim(
                    claim_no="EXP-MGR-201",
                    employee_id=employee.id,
                    employee_name="张三",
                    department_name="市场部",
                    project_code="PRJ-MGR",
                    expense_type="transport",
                    reason="滴滴报销",
                    location="上海",
                    amount=Decimal("66.00"),
                    currency="CNY",
                    invoice_count=1,
                    occurred_at=datetime(2026, 5, 12, 9, 0, tzinfo=UTC),
                    submitted_at=datetime(2026, 5, 12, 10, 0, tzinfo=UTC),
                    status="submitted",
                    approval_stage="直属领导审批",
                    risk_flags_json=[],
                ),
                ExpenseClaim(
                    claim_no="EXP-MGR-202",
                    employee_id=outsider.id,
                    employee_name="李四",
                    department_name="销售部",
                    project_code="PRJ-OTHER",
                    expense_type="meal",
                    reason="客户用餐",
                    location="杭州",
                    amount=Decimal("188.00"),
                    currency="CNY",
                    invoice_count=1,
                    occurred_at=datetime(2026, 5, 12, 12, 0, tzinfo=UTC),
                    submitted_at=datetime(2026, 5, 12, 13, 0, tzinfo=UTC),
                    status="submitted",
                    approval_stage="直属领导审批",
                    risk_flags_json=[],
                ),
            ]
        )
        db.commit()

        claims = ExpenseClaimService(db).list_approval_claims(current_user)

        assert len(claims) == 1
        assert claims[0].claim_no == "EXP-MGR-201"


def test_list_approval_claims_limits_finance_to_finance_stage_claims() -> None:
    current_user = CurrentUserContext(
        username="finance-approval-list@example.com",
        name="财务",
        role_codes=["finance"],
        is_admin=False,
    )

    with build_session() as db:
        db.add_all(
            [
                ExpenseClaim(
                    claim_no="EXP-FIN-LIST-201",
                    employee_name="张三",
                    department_name="市场部",
                    project_code="PRJ-FIN",
                    expense_type="transport",
                    reason="直属领导待审",
                    location="上海",
                    amount=Decimal("66.00"),
                    currency="CNY",
                    invoice_count=1,
                    occurred_at=datetime(2026, 5, 12, 9, 0, tzinfo=UTC),
                    submitted_at=datetime(2026, 5, 12, 10, 0, tzinfo=UTC),
                    status="submitted",
                    approval_stage="直属领导审批",
                    risk_flags_json=[],
                ),
                ExpenseClaim(
                    claim_no="EXP-FIN-LIST-202",
                    employee_name="李四",
                    department_name="销售部",
                    project_code="PRJ-FIN",
                    expense_type="meal",
                    reason="财务待审",
                    location="杭州",
                    amount=Decimal("188.00"),
                    currency="CNY",
                    invoice_count=1,
                    occurred_at=datetime(2026, 5, 12, 12, 0, tzinfo=UTC),
                    submitted_at=datetime(2026, 5, 12, 13, 0, tzinfo=UTC),
                    status="submitted",
                    approval_stage="财务审批",
                    risk_flags_json=[],
                ),
            ]
        )
        db.commit()

        claims = ExpenseClaimService(db).list_approval_claims(current_user)

        assert [claim.claim_no for claim in claims] == ["EXP-FIN-LIST-202"]