fix(agent): 修复规则中心表格版本和修改记录
补齐规则资产 JSON 读写接口和前端调用,修复 AuditView 导入缺失。 Excel 在线编辑改为比对所有页签并生成最近修改记录,版本快照统一保存到 rules/finance-rules/.versions。 隔离规则表测试存储,避免测试或旧入口写入真实规则目录与 storage/agent_assets。
This commit is contained in:
84
server/src/app/services/agent_asset_rule_library.py
Normal file
84
server/src/app/services/agent_asset_rule_library.py
Normal file
@@ -0,0 +1,84 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from app.core.config import SERVER_DIR
|
||||
from app.services.agent_asset_spreadsheet import RULE_LIBRARY_NAMES
|
||||
|
||||
JSON_RULE_MIME_TYPE = "application/json"
|
||||
|
||||
|
||||
class AgentAssetRuleLibraryManager:
|
||||
def __init__(self, rule_root: Path | None = None) -> None:
|
||||
self.rule_root = Path(rule_root or (SERVER_DIR / "rules")).resolve()
|
||||
|
||||
def ensure_rule_library_dirs(self) -> None:
|
||||
for library in sorted(RULE_LIBRARY_NAMES):
|
||||
(self.rule_root / library).mkdir(parents=True, exist_ok=True)
|
||||
|
||||
def resolve_rule_library_path(self, *, library: str, file_name: str) -> Path:
|
||||
normalized_library = str(library or "").strip()
|
||||
if normalized_library not in RULE_LIBRARY_NAMES:
|
||||
raise ValueError("Invalid rule library.")
|
||||
|
||||
normalized_name = Path(str(file_name or "").strip()).name.strip()
|
||||
if not normalized_name or not normalized_name.endswith(".json"):
|
||||
raise ValueError("Rule JSON file name must end with .json.")
|
||||
|
||||
library_dir = (self.rule_root / normalized_library).resolve()
|
||||
target_path = (library_dir / normalized_name).resolve()
|
||||
try:
|
||||
target_path.relative_to(library_dir)
|
||||
except ValueError:
|
||||
raise ValueError("Invalid rule JSON path.") from None
|
||||
return target_path
|
||||
|
||||
def read_rule_library_json(self, *, library: str, file_name: str) -> dict[str, Any]:
|
||||
target_path = self.resolve_rule_library_path(library=library, file_name=file_name)
|
||||
if not target_path.exists():
|
||||
raise FileNotFoundError("Rule JSON file not found.")
|
||||
|
||||
try:
|
||||
payload = json.loads(target_path.read_text(encoding="utf-8"))
|
||||
except json.JSONDecodeError as exc:
|
||||
raise ValueError("Rule JSON file is invalid.") from exc
|
||||
|
||||
if not isinstance(payload, dict):
|
||||
raise ValueError("Rule JSON payload must be an object.")
|
||||
return payload
|
||||
|
||||
def write_rule_library_json(
|
||||
self,
|
||||
*,
|
||||
library: str,
|
||||
file_name: str,
|
||||
payload: dict[str, Any],
|
||||
) -> dict[str, Any]:
|
||||
if not isinstance(payload, dict):
|
||||
raise ValueError("Rule JSON payload must be an object.")
|
||||
|
||||
rule_code = str(payload.get("rule_code") or "").strip()
|
||||
if not rule_code:
|
||||
raise ValueError("Rule JSON must include rule_code.")
|
||||
|
||||
evaluator = str(payload.get("evaluator") or "").strip()
|
||||
if not evaluator:
|
||||
raise ValueError("Rule JSON must include evaluator.")
|
||||
|
||||
target_path = self.resolve_rule_library_path(library=library, file_name=file_name)
|
||||
target_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
target_path.write_text(
|
||||
f"{json.dumps(payload, ensure_ascii=False, indent=2)}\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
return payload
|
||||
|
||||
def list_rule_library_json_files(self, *, library: str) -> list[str]:
|
||||
library_dir = self.resolve_rule_library_path(
|
||||
library=library,
|
||||
file_name="placeholder.json",
|
||||
).parent
|
||||
library_dir.mkdir(parents=True, exist_ok=True)
|
||||
return sorted(path.name for path in library_dir.glob("*.json") if path.is_file())
|
||||
Reference in New Issue
Block a user