from __future__ import annotations import json from datetime import UTC, date, datetime from decimal import Decimal from sqlalchemy import create_engine from sqlalchemy.orm import Session, sessionmaker from sqlalchemy.pool import StaticPool from app.core.agent_enums import AgentAssetDomain from app.db.base import Base from app.models.agent_asset import AgentAsset from app.models.financial_record import ExpenseClaim, ExpenseClaimItem from app.schemas.agent_asset import AgentAssetRiskRuleGenerateRequest from app.services.agent_asset_rule_library import AgentAssetRuleLibraryManager from app.services.agent_asset_spreadsheet import RISK_RULES_LIBRARY from app.services.risk_rule_generation import RiskRuleGenerationService from app.services.risk_rule_template_executor import RiskRuleTemplateExecutor class NullRuntimeChatService: def complete(self, *args, **kwargs) -> None: return None class CompositeRuntimeChatService: def complete(self, *args, **kwargs) -> str: return json.dumps( { "name": "招待发票说明校验", "description": "招待报销已取得发票但缺少客户说明时进入复核。", "template_key": "composite_rule_v1", "semantic_type": "entertainment_invoice_reason_check", "field_keys": ["attachment.invoice_no", "claim.reason"], "condition_summary": "D=发票号码，E=报销事由；D存在且E未说明客户名称时命中。", "rule_ir": { "facts": [ {"id": "D", "label": "发票号码", "fields": ["attachment.invoice_no"]}, {"id": "E", "label": "报销事由", "fields": ["claim.reason"]}, ], "hit_logic": "D AND NOT CONTAINS(E, 客户)", }, "conditions": [ { "id": "invoice_present", "operator": "exists_any", "fields": ["attachment.invoice_no"], }, { "id": "missing_customer_reason", "operator": "not_contains_any", "fields": ["claim.reason"], "keywords": ["客户", "拜访对象"], }, ], "hit_logic": {"all": ["invoice_present", "missing_customer_reason"]}, "formula": "HIT WHEN EXISTS(invoice_no) AND NOT CONTAINS(reason, 客户|拜访对象)", "message_template": "招待发票已上传，但事由缺少客户或拜访对象说明。", "keywords": [], "exception_keywords": [], "flow": { "start": "招待报销提交", "evidence": "读取发票号码和报销事由", "decision": "是否有发票且事由缺少客户说明", "pass": "客户说明完整，继续流转", "fail": "缺少客户说明，进入复核", }, }, ensure_ascii=False, ) class LodgingSemanticRuntimeChatService: def complete(self, *args, **kwargs) -> str: return json.dumps( { "name": "住宿城市日期一致性校验", "description": "住宿票据的城市和日期需要能对应本次差旅行程，缺少合理说明时进入复核。", "template_key": "composite_rule_v1", "semantic_type": "travel_lodging_city_date_consistency", "field_keys": [ "attachment.hotel_city", "attachment.stay_start_date", "attachment.stay_end_date", "attachment.issue_date", "claim.location", "item.item_location", "attachment.route_cities", "claim.trip_start_date", "claim.trip_end_date", "item.item_date", "claim.reason", "item.item_reason", ], "condition_summary": ( "D=住宿票据事实，A=住宿城市，B=本次行程城市范围，T=住宿日期或开票日期，" "R=出差起止日期；D存在且[(A不属于B)或(T超出R)]且无合理说明时命中。" ), "rule_ir": { "facts": [ {"id": "D", "label": "住宿票据事实", "fields": ["attachment.hotel_city", "attachment.ocr_text"]}, {"id": "A", "label": "住宿城市", "fields": ["attachment.hotel_city"]}, {"id": "B", "label": "本次行程城市范围", "fields": ["claim.location", "item.item_location", "attachment.route_cities"]}, {"id": "T", "label": "住宿日期或开票日期", "fields": ["attachment.stay_start_date", "attachment.stay_end_date", "attachment.issue_date"]}, {"id": "R", "label": "出差起止日期", "fields": ["claim.trip_start_date", "claim.trip_end_date", "item.item_date"]}, {"id": "E", "label": "合理说明", "fields": ["claim.reason", "item.item_reason"]}, ], "hit_logic": "D AND ((A NOT_IN B) OR DATE_OUTSIDE(T,R)) AND NOT CONTAINS(E, exception_keywords)", }, "conditions": [ { "id": "lodging_document_present", "operator": "exists_any", "fields": ["attachment.hotel_city", "attachment.ocr_text"], }, { "id": "lodging_city_outside_trip_scope", "operator": "not_in_scope", "left_fields": ["attachment.hotel_city"], "right_fields": ["claim.location", "item.item_location", "attachment.route_cities"], }, { "id": "lodging_date_outside_trip_range", "operator": "date_outside_range", "date_fields": ["attachment.stay_start_date", "attachment.stay_end_date", "attachment.issue_date"], "range_start_fields": ["claim.trip_start_date", "item.item_date"], "range_end_fields": ["claim.trip_end_date", "item.item_date"], }, { "id": "missing_reasonable_exception", "operator": "not_contains_any", "fields": ["claim.reason", "item.item_reason"], "keywords": ["延期", "改签", "临时任务"], }, ], "hit_logic": { "all": [ "lodging_document_present", {"any": ["lodging_city_outside_trip_scope", "lodging_date_outside_trip_range"]}, "missing_reasonable_exception", ] }, "formula": "D AND ((A NOT_IN B) OR DATE_OUTSIDE(T,R)) AND NOT EXCEPTION(E)", "message_template": "住宿票据城市或日期无法与本次差旅行程形成一致关系，且未识别到合理说明。", "keywords": [], "exception_keywords": ["延期", "改签", "临时任务"], "flow": { "start": "差旅住宿报销提交", "evidence": "读取住宿票据事实、行程范围和合理说明", "decision": "住宿城市或日期是否脱离本次行程，且是否缺少合理说明", "pass": "城市和日期均能对应行程，或已有合理说明", "fail": "城市或日期无法对应行程，进入复核", }, }, ensure_ascii=False, ) 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 _read_payload(manager: AgentAssetRuleLibraryManager, asset: AgentAsset) -> dict: return manager.read_rule_library_json( library=RISK_RULES_LIBRARY, file_name=asset.config_json["rule_document"]["file_name"], ) def test_lodging_city_date_rule_generates_explainable_composite_json(tmp_path) -> None: text = ( "差旅住宿报销时，先确认已上传住宿发票或酒店水单；再读取报销事由、申报目的地、" "住宿城市、开票日期和行程城市。若住宿票据显示的城市不在本次差旅行程范围内，" "或住宿发生时间明显早于出差开始、晚于出差结束，且没有延期、改签、临时任务等说明，" "则标记为高风险，要求补充住宿原因、行程证明或重新提交票据。" ) with build_session() as db: manager = AgentAssetRuleLibraryManager(rule_root=tmp_path / "rules") service = RiskRuleGenerationService( db, rule_library_manager=manager, runtime_chat_service=LodgingSemanticRuntimeChatService(), ) asset_id = service.generate_rule_asset( AgentAssetRiskRuleGenerateRequest( business_domain=AgentAssetDomain.EXPENSE, expense_category="hotel", rule_title="住宿城市日期一致性校验", risk_level="high", natural_language=text, requires_attachment=True, ), actor="pytest", ) asset = db.get(AgentAsset, asset_id) assert asset is not None payload = _read_payload(manager, asset) assert payload["template_key"] == "composite_rule_v1" assert payload["semantic_type"] == "travel_lodging_city_date_consistency" assert payload["params"]["semantic_type"] == "travel_lodging_city_date_consistency" assert payload["params"]["keywords"] == [] assert "风险关键词" not in payload["params"]["condition_summary"] assert "attachment.stay_start_date" in payload["params"]["field_keys"] assert "claim.trip_start_date" in payload["params"]["field_keys"] assert payload["params"]["rule_ir"]["facts"] assert payload["params"]["hit_logic"]["all"][1]["any"] == [ "lodging_city_outside_trip_scope", "lodging_date_outside_trip_range", ] def test_composite_lodging_executor_hits_mismatch_and_respects_exception() -> None: manifest = { "template_key": "composite_rule_v1", "params": { "template_key": "composite_rule_v1", "semantic_type": "travel_lodging_city_date_consistency", "condition_summary": "住宿城市或日期不在本次差旅行程范围内且无合理说明时命中。", "conditions": [ { "id": "lodging_document_present", "operator": "exists_any", "fields": ["attachment.hotel_city", "attachment.ocr_text"], }, { "id": "lodging_city_outside_trip_scope", "operator": "not_in_scope", "left_fields": ["attachment.hotel_city"], "right_fields": ["claim.location", "item.item_location", "attachment.route_cities"], }, { "id": "lodging_date_outside_trip_range", "operator": "date_outside_range", "date_fields": ["attachment.stay_start_date", "attachment.stay_end_date"], "range_start_fields": ["claim.trip_start_date", "item.item_date"], "range_end_fields": ["claim.trip_end_date", "item.item_date"], }, { "id": "missing_reasonable_exception", "operator": "not_contains_any", "fields": ["claim.reason", "item.item_reason"], "keywords": ["延期", "改签", "临时任务"], }, ], "hit_logic": { "all": [ "lodging_document_present", {"any": ["lodging_city_outside_trip_scope", "lodging_date_outside_trip_range"]}, "missing_reasonable_exception", ] }, }, "outcomes": {"fail": {"severity": "high"}}, } claim = _build_claim(reason="去上海出差住宿", location="上海") contexts = [ { "document_info": { "hotel_city": "北京", "stay_start_date": "2026-05-11", "stay_end_date": "2026-05-12", "fields": [ {"key": "hotel_city", "label": "住宿城市", "value": "北京"}, {"key": "stay_start_date", "label": "入住日期", "value": "2026-05-11"}, {"key": "stay_end_date", "label": "离店日期", "value": "2026-05-12"}, ], }, "ocr_text": "北京酒店住宿发票", } ] result = RiskRuleTemplateExecutor().evaluate(manifest, claim=claim, contexts=contexts) assert result is not None assert result["evidence"]["condition_results"]["lodging_city_outside_trip_scope"] is True claim_with_exception = _build_claim(reason="去上海出差住宿，因临时任务改签至北京", location="上海") assert RiskRuleTemplateExecutor().evaluate( manifest, claim=claim_with_exception, contexts=contexts ) is None def test_model_generated_composite_rule_is_preserved_for_other_categories(tmp_path) -> None: with build_session() as db: manager = AgentAssetRuleLibraryManager(rule_root=tmp_path / "rules") service = RiskRuleGenerationService( db, rule_library_manager=manager, runtime_chat_service=CompositeRuntimeChatService(), ) asset_id = service.generate_rule_asset( AgentAssetRiskRuleGenerateRequest( business_domain=AgentAssetDomain.EXPENSE, expense_category="meal", rule_title="招待发票客户说明校验", risk_level="medium", natural_language="招待报销时，如果已经上传发票但报销事由没有客户或拜访对象说明，则提示中风险。", ), actor="pytest", ) asset = db.get(AgentAsset, asset_id) assert asset is not None payload = _read_payload(manager, asset) assert payload["template_key"] == "composite_rule_v1" assert payload["semantic_type"] == "entertainment_invoice_reason_check" assert payload["params"]["conditions"][0]["operator"] == "exists_any" assert payload["params"]["hit_logic"] == {"all": ["invoice_present", "missing_customer_reason"]} assert payload["params"]["message_template"] == "招待发票已上传，但事由缺少客户或拜访对象说明。" claim = _build_claim(reason="招待费", location="上海") result = RiskRuleTemplateExecutor().evaluate( payload, claim=claim, contexts=[{"document_info": {"invoice_no": "INV-20260526001"}}], ) assert result is not None assert result["message"] == "招待发票已上传，但事由缺少客户或拜访对象说明。" claim.reason = "招待客户 ACME 的餐费" assert RiskRuleTemplateExecutor().evaluate( payload, claim=claim, contexts=[{"document_info": {"invoice_no": "INV-20260526001"}}], ) is None def _build_claim(*, reason: str, location: str) -> ExpenseClaim: claim = ExpenseClaim( claim_no="TEST-COMPOSITE-RISK", employee_name="测试员工", department_name="测试部门", expense_type="差旅费", reason=reason, location=location, amount=Decimal("680.00"), currency="CNY", invoice_count=1, occurred_at=datetime(2026, 5, 11, tzinfo=UTC), status="draft", ) claim.trip_start_date = date(2026, 5, 10) claim.trip_end_date = date(2026, 5, 12) claim.items = [ ExpenseClaimItem( item_date=date(2026, 5, 11), item_type="住宿费", item_reason=reason, item_location=location, item_amount=Decimal("680.00"), ) ] return claim