from __future__ import annotations from pathlib import Path from sqlalchemy import select from app.core.agent_enums import ( AgentAssetStatus, ) from app.core.logging import get_logger from app.models.agent_asset import AgentAsset from app.services.agent_asset_spreadsheet import ( COMPANY_COMMUNICATION_EXPENSE_RULE_CODE, COMPANY_COMMUNICATION_EXPENSE_RULE_FILENAME, COMPANY_PREAPPROVAL_RULE_CODE, COMPANY_PREAPPROVAL_RULE_FILENAME, COMPANY_TRAVEL_EXPENSE_RULE_CODE, COMPANY_TRAVEL_EXPENSE_RULE_FILENAME, FINANCE_RULES_LIBRARY, AgentAssetSpreadsheetManager, ) from app.services.agent_foundation_constants import ( COMPANY_COMMUNICATION_RULE_SCENARIO_JSON, COMPANY_PREAPPROVAL_RULE_SCENARIO_JSON, COMPANY_TRAVEL_RULE_SCENARIO_JSON, ) from app.services.finance_rule_catalog import ( DEPRECATED_FINANCE_RULE_CODES, DEPRECATED_FINANCE_RULE_REPLACEMENTS, ) logger = get_logger("app.services.agent_foundation") class AgentFoundationSpreadsheetMixin: def sync_finance_rule_assets_from_catalog(self) -> int: synced_count = self._ensure_core_finance_rule_asset_metadata() self._hide_deprecated_finance_rule_assets() self.db.flush() return synced_count def _ensure_core_finance_rule_asset_metadata(self) -> int: synced_count = 0 synced_count += int( self._ensure_core_finance_rule_asset( code=COMPANY_TRAVEL_EXPENSE_RULE_CODE, scenario_category=COMPANY_TRAVEL_RULE_SCENARIO_JSON[0], finance_rule_sheet="差旅住宿费标准", expense_types=["travel", "hotel", "transport"], ) ) synced_count += int( self._ensure_core_finance_rule_asset( code=COMPANY_COMMUNICATION_EXPENSE_RULE_CODE, scenario_category=COMPANY_COMMUNICATION_RULE_SCENARIO_JSON[0], finance_rule_sheet="通信费报销标准", expense_types=["communication"], ) ) synced_count += int( self._ensure_core_finance_rule_asset( code=COMPANY_PREAPPROVAL_RULE_CODE, scenario_category=COMPANY_PREAPPROVAL_RULE_SCENARIO_JSON[0], finance_rule_sheet="费用申请审批规则", expense_types=["meal", "entertainment", "office", "all"], ) ) return synced_count def _ensure_core_finance_rule_asset( self, *, code: str, scenario_category: str, finance_rule_sheet: str, expense_types: list[str], ) -> bool: asset = self.db.scalar(select(AgentAsset).where(AgentAsset.code == code)) if asset is None: return False asset.scenario_json = [scenario_category] asset.config_json = { **(asset.config_json or {}), "enabled": True, "tag": "财务规则", "detail_mode": "spreadsheet", "rule_library": FINANCE_RULES_LIBRARY, "scenario_category": scenario_category, "ai_review_category": scenario_category, "finance_rule_code": code, "finance_rule_sheet": finance_rule_sheet, "expense_types": expense_types, "business_stage": ["expense_application", "reimbursement"], "budget_required": True, } return True def _hide_deprecated_finance_rule_assets(self) -> None: for code in DEPRECATED_FINANCE_RULE_CODES: asset = self.db.scalar(select(AgentAsset).where(AgentAsset.code == code)) if asset is None: continue asset.status = AgentAssetStatus.DISABLED.value asset.scenario_json = ["已废弃"] replacement = DEPRECATED_FINANCE_RULE_REPLACEMENTS.get(code) if replacement == COMPANY_TRAVEL_EXPENSE_RULE_CODE: deprecated_reason = ( "交通/住宿细分并入公司差旅费报销规则，不再作为独立财务规则展示。" ) elif replacement == COMPANY_PREAPPROVAL_RULE_CODE: deprecated_reason = ( "申请审批阈值已并入公司费用申请审批规则，不再作为独立财务规则展示。" ) else: deprecated_reason = ( "该费用类型没有独立职务金额分档，额度控制转入预算中心，" "不再作为独立财务规则表展示。" ) asset.config_json = { **(asset.config_json or {}), "enabled": False, "tag": "废弃规则", "deprecated": True, "deprecated_reason": deprecated_reason, } if replacement: asset.config_json["replaced_by"] = replacement def _ensure_company_travel_rule_spreadsheet_seed( self, asset: AgentAsset, *, version: str, actor_name: str, ): manager = AgentAssetSpreadsheetManager() manager.ensure_rule_library_dirs() live_document = manager.store_rule_library_spreadsheet( library=FINANCE_RULES_LIBRARY, file_name=COMPANY_TRAVEL_EXPENSE_RULE_FILENAME, content=self._read_or_build_company_travel_rule_file(manager), actor_name=actor_name, source="rule-library", ) existing_document = ( asset.config_json.get("rule_document") if isinstance(asset.config_json, dict) else None ) storage_key = ( str(existing_document.get("storage_key") or "").strip() if isinstance(existing_document, dict) else "" ) if storage_key: try: existing_path = manager.resolve_storage_path(storage_key) except FileNotFoundError: existing_path = None if existing_path is not None and existing_path.exists(): asset.config_json = { **(asset.config_json or {}), "detail_mode": "spreadsheet", "tag": "财务规则", "rule_library": FINANCE_RULES_LIBRARY, "rule_document": { **AgentAssetSpreadsheetManager.build_rule_document_config( live_document, asset_version=version, ), "storage_key": live_document.storage_key, }, } return live_document asset.config_json = { **(asset.config_json or {}), "detail_mode": "spreadsheet", "tag": "财务规则", "rule_library": FINANCE_RULES_LIBRARY, "rule_document": { **AgentAssetSpreadsheetManager.build_rule_document_config( live_document, asset_version=version, ), "storage_key": live_document.storage_key, }, } return live_document def _ensure_company_communication_rule_spreadsheet_seed( self, asset: AgentAsset, *, version: str, actor_name: str, ): return self._ensure_finance_rule_spreadsheet_seed( asset, version=version, actor_name=actor_name, file_name=COMPANY_COMMUNICATION_EXPENSE_RULE_FILENAME, fallback_sheet_name="通信费报销规则", ) def _ensure_company_preapproval_rule_spreadsheet_seed( self, asset: AgentAsset, *, version: str, actor_name: str, ): return self._ensure_finance_rule_spreadsheet_seed( asset, version=version, actor_name=actor_name, file_name=COMPANY_PREAPPROVAL_RULE_FILENAME, fallback_sheet_name="费用申请审批规则", workbook_sheets=[ ( "费用申请审批规则", [ [ "费用类型代码", "费用类型", "触发条件", "阈值金额", "前置要求", "审批要求", "风险动作", "备注", ], [ "meal/entertainment", "业务招待费", "单次费用金额大于 500 元", 500, "必须先提交费用申请单，并说明客户、参与人和招待事由", "申请单需按审批链完成审批后方可报销", "报销阶段未关联已通过申请单时标记高风险", "适配 meal 与 entertainment 两个本体费用类型", ], [ "office", "办公用品费", "单次或批量采购金额大于 2000 元", 2000, "必须先提交办公采购或费用申请单", "申请单需经直属领导审批；如触发预算管控则继续预算复核", "报销阶段未关联已通过申请单时标记高风险", "覆盖办公用品、办公耗材、低值易耗品等场景", ], [ "all", "通用大额费用", "任意费用金额大于 2000 元", 2000, "必须进入费用申请和审批流程", "至少完成直属领导审批；按预算和财务规则继续流转", "报销阶段未关联已通过申请单时标记高风险", "差旅、通信等已有专项规则时可同时适用专项规则", ], ], ), ( "字段说明", [ ["字段", "说明"], ["费用类型代码", "使用系统本体费用类型，不新增非本体字段"], ["阈值金额", "单位为人民币元，执行时按 claim.amount 进行数值比较"], ["前置要求", "说明是否需要事前申请以及申请单需要包含的信息"], ["审批要求", "说明申请单进入审批链后的最低审批要求"], ["风险动作", "说明报销阶段未满足规则时的系统处理"], ], ), ], ) @staticmethod def _read_or_build_company_travel_rule_file( manager: AgentAssetSpreadsheetManager, ) -> bytes: live_key = ( Path("rules") / FINANCE_RULES_LIBRARY / COMPANY_TRAVEL_EXPENSE_RULE_FILENAME ).as_posix() live_path = manager.resolve_storage_path(live_key) if live_path.exists(): return live_path.read_bytes() return AgentAssetSpreadsheetManager.build_blank_rule_workbook("差旅费报销规则") def _ensure_finance_rule_spreadsheet_seed( self, asset: AgentAsset, *, version: str, actor_name: str, file_name: str, fallback_sheet_name: str, workbook_sheets: list[tuple[str, list[list[object]]]] | None = None, ): manager = AgentAssetSpreadsheetManager() manager.ensure_rule_library_dirs() live_document = manager.store_rule_library_spreadsheet( library=FINANCE_RULES_LIBRARY, file_name=file_name, content=self._read_or_build_finance_rule_file( manager, file_name=file_name, fallback_sheet_name=fallback_sheet_name, workbook_sheets=workbook_sheets, ), actor_name=actor_name, source="rule-library", ) existing_document = ( asset.config_json.get("rule_document") if isinstance(asset.config_json, dict) else None ) storage_key = ( str(existing_document.get("storage_key") or "").strip() if isinstance(existing_document, dict) else "" ) if storage_key: try: existing_path = manager.resolve_storage_path(storage_key) except FileNotFoundError: existing_path = None if existing_path is not None and existing_path.exists(): asset.config_json = { **(asset.config_json or {}), "detail_mode": "spreadsheet", "tag": "财务规则", "rule_library": FINANCE_RULES_LIBRARY, "rule_document": { **AgentAssetSpreadsheetManager.build_rule_document_config( live_document, asset_version=version, ), "storage_key": live_document.storage_key, }, } return live_document asset.config_json = { **(asset.config_json or {}), "detail_mode": "spreadsheet", "tag": "财务规则", "rule_library": FINANCE_RULES_LIBRARY, "rule_document": { **AgentAssetSpreadsheetManager.build_rule_document_config( live_document, asset_version=version, ), "storage_key": live_document.storage_key, }, } return live_document @staticmethod def _read_or_build_finance_rule_file( manager: AgentAssetSpreadsheetManager, *, file_name: str, fallback_sheet_name: str, workbook_sheets: list[tuple[str, list[list[object]]]] | None = None, ) -> bytes: live_key = ( Path("rules") / FINANCE_RULES_LIBRARY / file_name ).as_posix() live_path = manager.resolve_storage_path(live_key) if live_path.exists(): return live_path.read_bytes() if workbook_sheets is not None: return AgentAssetSpreadsheetManager.build_rule_workbook(workbook_sheets) return AgentAssetSpreadsheetManager.build_blank_rule_workbook(fallback_sheet_name)