from __future__ import annotations from pathlib import Path from sqlalchemy import select from app.core.agent_enums import ( AgentAssetContentType, AgentAssetDomain, AgentAssetType, AgentReviewStatus, 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_TRAVEL_ALLOWANCE_RULE_CODE, COMPANY_TRAVEL_ALLOWANCE_RULE_FILENAME, COMPANY_PREAPPROVAL_RULE_CODE, COMPANY_PREAPPROVAL_RULE_FILENAME, COMPANY_TRAVEL_EXPENSE_RULE_CODE, COMPANY_TRAVEL_EXPENSE_RULE_FILENAME, COMPANY_TRAVEL_GRADE_MAPPING_RULE_CODE, COMPANY_TRAVEL_GRADE_MAPPING_RULE_FILENAME, COMPANY_TRAVEL_SEASON_MAPPING_RULE_CODE, COMPANY_TRAVEL_SEASON_MAPPING_RULE_FILENAME, COMPANY_TRAVEL_TRANSPORT_ESTIMATE_RULE_CODE, COMPANY_TRAVEL_TRANSPORT_ESTIMATE_RULE_FILENAME, COMPANY_TRAVEL_TRANSPORT_RULE_CODE, COMPANY_TRAVEL_TRANSPORT_RULE_FILENAME, FINANCE_RULES_LIBRARY, AgentAssetSpreadsheetManager, ) from app.services.agent_foundation_constants import ( COMPANY_COMMUNICATION_RULE_SCENARIO_JSON, COMPANY_COMMUNICATION_RULE_VERSION, COMPANY_PREAPPROVAL_RULE_SCENARIO_JSON, COMPANY_PREAPPROVAL_RULE_VERSION, COMPANY_TRAVEL_RULE_SCENARIO_JSON, COMPANY_TRAVEL_RULE_VERSION, ) from app.services.finance_rule_catalog import ( DEPRECATED_FINANCE_RULE_CODES, DEPRECATED_FINANCE_RULE_REPLACEMENTS, ) from app.services.agent_foundation_preapproval_spreadsheet import ( build_preapproval_rule_workbook_sheets, ) 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, name="差旅住宿报销标准", description="按地区和职级维护差旅住宿费报销上限。", scenario_category=COMPANY_TRAVEL_RULE_SCENARIO_JSON[0], finance_rule_sheet="差旅住宿费标准", expense_types=["hotel"], version=COMPANY_TRAVEL_RULE_VERSION, reviewer="顾承宇", file_name=COMPANY_TRAVEL_EXPENSE_RULE_FILENAME, workbook_content=AgentAssetSpreadsheetManager.build_travel_lodging_rule_template(), rule_template_label="差旅住宿 Excel 模板", travel_policy_component="lodging", ) ) synced_count += int( self._ensure_core_finance_rule_asset( code=COMPANY_TRAVEL_ALLOWANCE_RULE_CODE, name="出差补助报销标准", description="按地区维护伙食补助、基本出差补贴和补助合计。", scenario_category=COMPANY_TRAVEL_RULE_SCENARIO_JSON[0], finance_rule_sheet="出差补助标准", expense_types=["travel"], version=COMPANY_TRAVEL_RULE_VERSION, reviewer="顾承宇", file_name=COMPANY_TRAVEL_ALLOWANCE_RULE_FILENAME, workbook_content=AgentAssetSpreadsheetManager.build_travel_allowance_rule_template(), rule_template_label="出差补助 Excel 模板", travel_policy_component="allowance", ) ) synced_count += int( self._ensure_core_finance_rule_asset( code=COMPANY_TRAVEL_TRANSPORT_RULE_CODE, name="交通工具等级标准", description="按员工职级维护飞机、火车等长途交通工具等级。", scenario_category=COMPANY_TRAVEL_RULE_SCENARIO_JSON[0], finance_rule_sheet="交通工具等级标准", expense_types=["travel", "transport"], version=COMPANY_TRAVEL_RULE_VERSION, reviewer="顾承宇", file_name=COMPANY_TRAVEL_TRANSPORT_RULE_FILENAME, workbook_content=AgentAssetSpreadsheetManager.build_travel_transport_rule_template(), rule_template_label="交通工具等级 Excel 模板", travel_policy_component="transport", ) ) synced_count += int( self._ensure_core_finance_rule_asset( code=COMPANY_TRAVEL_TRANSPORT_ESTIMATE_RULE_CODE, name="交通费用预估表", description="按出发城市、目的地和交通方式维护申请阶段预算占用的交通费用预估金额。", scenario_category=COMPANY_TRAVEL_RULE_SCENARIO_JSON[0], finance_rule_sheet="交通费用预估表", expense_types=["travel", "transport"], version=COMPANY_TRAVEL_RULE_VERSION, reviewer="顾承宇", file_name=COMPANY_TRAVEL_TRANSPORT_ESTIMATE_RULE_FILENAME, workbook_content=AgentAssetSpreadsheetManager.build_travel_transport_estimate_rule_template(), rule_template_label="交通费用预估 Excel 模板", travel_policy_component="transport_estimate", ) ) synced_count += int( self._ensure_core_finance_rule_asset( code=COMPANY_TRAVEL_GRADE_MAPPING_RULE_CODE, name="差旅职级映射表", description="明确 P0-P8 九级职级与住宿、交通规则列之间的对应关系，其中 P8 为董事会。", scenario_category=COMPANY_TRAVEL_RULE_SCENARIO_JSON[0], finance_rule_sheet="差旅职级映射表", expense_types=["hotel", "travel", "transport"], version=COMPANY_TRAVEL_RULE_VERSION, reviewer="顾承宇", file_name=COMPANY_TRAVEL_GRADE_MAPPING_RULE_FILENAME, workbook_content=AgentAssetSpreadsheetManager.build_travel_grade_mapping_template(), rule_template_label="差旅职级映射 Excel 模板", travel_policy_component="grade_mapping", ) ) synced_count += int( self._ensure_core_finance_rule_asset( code=COMPANY_TRAVEL_SEASON_MAPPING_RULE_CODE, name="地区淡旺季映射表", description="明确住宿标准中旺季地区、旺季月份和旺季超标限额的对应关系。", scenario_category=COMPANY_TRAVEL_RULE_SCENARIO_JSON[0], finance_rule_sheet="地区淡旺季映射表", expense_types=["hotel", "travel"], version=COMPANY_TRAVEL_RULE_VERSION, reviewer="顾承宇", file_name=COMPANY_TRAVEL_SEASON_MAPPING_RULE_FILENAME, workbook_content=AgentAssetSpreadsheetManager.build_travel_season_mapping_template(), rule_template_label="地区淡旺季映射 Excel 模板", travel_policy_component="season_mapping", ) ) synced_count += int( self._ensure_core_finance_rule_asset( code=COMPANY_COMMUNICATION_EXPENSE_RULE_CODE, name="公司通信费报销规则", description="通过 Excel 明细表维护员工通信费报销标准、专项补充口径和审批要求。", scenario_category=COMPANY_COMMUNICATION_RULE_SCENARIO_JSON[0], finance_rule_sheet="通信费报销标准", expense_types=["communication"], version=COMPANY_COMMUNICATION_RULE_VERSION, reviewer="顾承宇", file_name=COMPANY_COMMUNICATION_EXPENSE_RULE_FILENAME, workbook_content=AgentAssetSpreadsheetManager.build_company_communication_rule_template(), rule_template_label="通信费报销 Excel 模板", finance_rule_code="expense.communication.policy", refresh_workbook_content=True, ) ) synced_count += int( self._ensure_core_finance_rule_asset( code=COMPANY_PREAPPROVAL_RULE_CODE, name="公司费用申请审批规则", description="通过 Excel 明细表维护业务招待、办公用品和通用大额费用的事前申请与审批阈值。", scenario_category=COMPANY_PREAPPROVAL_RULE_SCENARIO_JSON[0], finance_rule_sheet="费用申请审批规则", expense_types=["meal", "entertainment", "office", "all"], version=COMPANY_PREAPPROVAL_RULE_VERSION, reviewer="顾承宣", file_name=COMPANY_PREAPPROVAL_RULE_FILENAME, workbook_content=None, rule_template_label="费用申请审批 Excel 模板", finance_rule_code="expense.preapproval.policy", tag="申请规则", ) ) return synced_count def _ensure_core_finance_rule_asset( self, *, code: str, name: str, description: str, scenario_category: str, finance_rule_sheet: str, expense_types: list[str], version: str, reviewer: str, file_name: str, workbook_content: bytes | None, rule_template_label: str, finance_rule_code: str | None = None, travel_policy_component: str = "", tag: str = "基础规则", refresh_workbook_content: bool = False, ) -> bool: asset = self.db.scalar(select(AgentAsset).where(AgentAsset.code == code)) created_asset = asset is None if asset is None: asset = self._create_seed_asset( asset_type=AgentAssetType.RULE.value, code=code, name=name, description=description, domain=AgentAssetDomain.EXPENSE.value, scenario_json=[scenario_category], owner="财务制度管理组", reviewer=reviewer, status=AgentAssetStatus.ACTIVE.value, current_version=version, config_json={ "severity": "medium", "enabled": True, "tag": tag, "rule_tag": tag, "tags": [tag], "rule_tags": [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, "rule_template_label": rule_template_label, }, ) else: asset.name = name asset.description = description asset.owner = asset.owner or "财务制度管理组" asset.reviewer = asset.reviewer or reviewer if not str(asset.current_version or "").strip(): asset.current_version = version if not str(asset.working_version or "").strip(): asset.working_version = asset.current_version if not str(asset.published_version or "").strip(): asset.published_version = asset.current_version if not str(asset.status or "").strip() or asset.status == AgentAssetStatus.DISABLED.value: asset.status = AgentAssetStatus.ACTIVE.value asset.scenario_json = [scenario_category] config_json = { **(asset.config_json or {}), "enabled": True, "tag": tag, "rule_tag": tag, "tags": [tag], "rule_tags": [tag], "detail_mode": "spreadsheet", "rule_library": FINANCE_RULES_LIBRARY, "scenario_category": scenario_category, "ai_review_category": scenario_category, "finance_rule_code": finance_rule_code or code, "finance_rule_sheet": finance_rule_sheet, "expense_types": expense_types, "business_stage": ["expense_application", "reimbursement"], "budget_required": True, "rule_template_label": rule_template_label, } if travel_policy_component: config_json["travel_policy_component"] = travel_policy_component asset.config_json = config_json rule_document = (asset.config_json or {}).get("rule_document") has_rule_document = isinstance(rule_document, dict) and bool( str(rule_document.get("storage_key") or "").strip() ) if workbook_content is not None and ( created_asset or not has_rule_document or refresh_workbook_content ): self._ensure_finance_rule_asset_document( asset, version=version, reviewer=reviewer, file_name=file_name, content=workbook_content, force_live_document=refresh_workbook_content, ) return True def _ensure_finance_rule_asset_document( self, asset: AgentAsset, *, version: str, reviewer: str, file_name: str, content: bytes, force_live_document: bool = False, ) -> None: manager = AgentAssetSpreadsheetManager() manager.ensure_rule_library_dirs() rule_document = (asset.config_json or {}).get("rule_document") storage_key = ( str(rule_document.get("storage_key") or "").strip() if isinstance(rule_document, dict) else "" ) should_seed_file = force_live_document or not storage_key if storage_key: try: current_path = manager.resolve_storage_path(storage_key) except FileNotFoundError: current_path = None should_seed_file = should_seed_file or current_path is None or not current_path.exists() if should_seed_file: metadata = manager.store_rule_library_spreadsheet( library=FINANCE_RULES_LIBRARY, file_name=file_name, content=content, actor_name="系统初始化", source="rule-library", ) asset.config_json = { **(asset.config_json or {}), "rule_document": { **AgentAssetSpreadsheetManager.build_rule_document_config( metadata, asset_version=version, ), "storage_key": metadata.storage_key, }, } else: metadata = manager.store_rule_library_spreadsheet_snapshot( library=FINANCE_RULES_LIBRARY, asset_id=asset.id, version=version, file_name=file_name, content=content, actor_name="系统初始化", source="rule-library-version", ) self._ensure_asset_version( asset, version=version, content=AgentAssetSpreadsheetManager.build_version_markdown( rule_name=asset.name, version=version, metadata=metadata, ), content_type=AgentAssetContentType.MARKDOWN.value, change_note=f"初始化{asset.name} Excel 规则表。", created_by="系统初始化", ) self._ensure_asset_review( asset, version=version, reviewer=reviewer, review_status=AgentReviewStatus.APPROVED.value, review_note="首版 Excel 规则表已确认，可作为基础规则使用。", reviewed_at=None, ) 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_tag": "基础规则", "tags": ["基础规则"], "rule_tags": ["基础规则"], "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_tag": "基础规则", "tags": ["基础规则"], "rule_tags": ["基础规则"], "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="费用申请审批规则", tag="申请规则", workbook_sheets=build_preapproval_rule_workbook_sheets(), ) @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_travel_lodging_rule_template() 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, tag: str = "基础规则", ): 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": tag, "rule_tag": tag, "tags": [tag], "rule_tags": [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": tag, "rule_tag": tag, "tags": [tag], "rule_tags": [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)