Files
X-Financial/server/src/app/services/agent_asset_spreadsheet.py
caoxiaozhu d460ee0fe7 fix(agent): 修复规则中心表格版本和修改记录
补齐规则资产 JSON 读写接口和前端调用,修复 AuditView 导入缺失。

Excel 在线编辑改为比对所有页签并生成最近修改记录,版本快照统一保存到 rules/finance-rules/.versions。

隔离规则表测试存储,避免测试或旧入口写入真实规则目录与 storage/agent_assets。
2026-05-19 15:41:53 +00:00

611 lines
23 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
from __future__ import annotations
import hashlib
import json
import mimetypes
import re
from dataclasses import asdict, dataclass
from datetime import UTC, datetime
from io import BytesIO
from pathlib import Path
from xml.sax.saxutils import escape
from zipfile import ZIP_DEFLATED, ZipFile
from openpyxl import load_workbook
from app.core.config import SERVER_DIR, get_settings
RULE_SPREADSHEET_BLOCK_PATTERN = re.compile(
r"```rule-spreadsheet\s*(\{.*?\})\s*```",
re.DOTALL,
)
COMPANY_TRAVEL_EXPENSE_RULE_CODE = "rule.expense.company_travel_expense_reimbursement"
COMPANY_TRAVEL_EXPENSE_RULE_FILENAME = "公司差旅费报销规则.xlsx"
COMPANY_COMMUNICATION_EXPENSE_RULE_CODE = "rule.expense.company_communication_expense_reimbursement"
COMPANY_COMMUNICATION_EXPENSE_RULE_FILENAME = "公司通信费报销规则.xlsx"
FINANCE_RULES_LIBRARY = "finance-rules"
RISK_RULES_LIBRARY = "risk-rules"
RULE_LIBRARY_NAMES = {FINANCE_RULES_LIBRARY, RISK_RULES_LIBRARY}
SPREADSHEET_MIME_TYPE = (
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
)
@dataclass(slots=True)
class RuleSpreadsheetMeta:
file_name: str
storage_key: str
mime_type: str
size_bytes: int
checksum: str
updated_at: str
updated_by: str
source: str = "upload"
class AgentAssetSpreadsheetManager:
def __init__(
self,
storage_root: Path | None = None,
rule_root: Path | None = None,
) -> None:
settings = get_settings()
self.storage_root = Path(storage_root or settings.resolved_storage_root_dir).resolve()
self.asset_root = (self.storage_root / "agent_assets").resolve()
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 store_spreadsheet(
self,
*,
asset_id: str,
version: str,
file_name: str,
content: bytes,
actor_name: str,
source: str = "upload",
) -> RuleSpreadsheetMeta:
return self.store_rule_library_spreadsheet_snapshot(
library=FINANCE_RULES_LIBRARY,
asset_id=asset_id,
version=version,
file_name=file_name,
content=content,
actor_name=actor_name,
source=source,
)
def store_rule_library_spreadsheet(
self,
*,
library: str,
file_name: str,
content: bytes,
actor_name: str,
source: str = "rule-library",
) -> RuleSpreadsheetMeta:
normalized_library = str(library or "").strip()
if normalized_library not in RULE_LIBRARY_NAMES:
raise ValueError("规则库目录不合法。")
normalized_name = Path(str(file_name or "").strip()).name.strip()
if not normalized_name:
raise ValueError("规则表文件名不能为空。")
if not content:
raise ValueError("规则表文件内容不能为空。")
self.ensure_rule_library_dirs()
relative_path = Path("rules") / normalized_library / normalized_name
target_path = (SERVER_DIR / relative_path).resolve()
try:
target_path.relative_to(self.rule_root)
except ValueError:
raise ValueError("规则库文件路径不合法。") from None
target_path.parent.mkdir(parents=True, exist_ok=True)
target_path.write_bytes(content)
mime_type = mimetypes.guess_type(normalized_name)[0] or SPREADSHEET_MIME_TYPE
return RuleSpreadsheetMeta(
file_name=normalized_name,
storage_key=relative_path.as_posix(),
mime_type=mime_type,
size_bytes=len(content),
checksum=hashlib.sha256(content).hexdigest(),
updated_at=datetime.now(UTC).isoformat(),
updated_by=str(actor_name or "system").strip() or "system",
source=source,
)
def store_rule_library_spreadsheet_snapshot(
self,
*,
library: str,
asset_id: str,
version: str,
file_name: str,
content: bytes,
actor_name: str,
source: str = "rule-library-version",
) -> RuleSpreadsheetMeta:
normalized_library = str(library or "").strip()
if normalized_library not in RULE_LIBRARY_NAMES:
raise ValueError("规则库目录不合法。")
raw_asset_id = str(asset_id or "").strip()
raw_version = str(version or "").strip()
normalized_asset_id = Path(raw_asset_id).name.strip()
normalized_version = Path(raw_version).name.strip()
normalized_name = Path(str(file_name or "").strip()).name.strip()
if (
not normalized_asset_id
or normalized_asset_id in {".", ".."}
or normalized_asset_id != raw_asset_id
):
raise ValueError("规则资产 ID 不合法。")
if (
not normalized_version
or normalized_version in {".", ".."}
or normalized_version != raw_version
):
raise ValueError("规则表版本号不合法。")
if not normalized_name:
raise ValueError("规则表文件名不能为空。")
if not content:
raise ValueError("规则表文件内容不能为空。")
self.ensure_rule_library_dirs()
relative_path = (
Path("rules")
/ normalized_library
/ ".versions"
/ normalized_asset_id
/ normalized_version
/ normalized_name
)
target_path = (SERVER_DIR / relative_path).resolve()
try:
target_path.relative_to(self.rule_root)
except ValueError:
raise ValueError("规则库版本文件路径不合法。") from None
target_path.parent.mkdir(parents=True, exist_ok=True)
target_path.write_bytes(content)
mime_type = mimetypes.guess_type(normalized_name)[0] or SPREADSHEET_MIME_TYPE
return RuleSpreadsheetMeta(
file_name=normalized_name,
storage_key=relative_path.as_posix(),
mime_type=mime_type,
size_bytes=len(content),
checksum=hashlib.sha256(content).hexdigest(),
updated_at=datetime.now(UTC).isoformat(),
updated_by=str(actor_name or "system").strip() or "system",
source=source,
)
def resolve_storage_path(self, storage_key: str) -> Path:
normalized = Path(str(storage_key or "").strip())
if not normalized.parts:
raise FileNotFoundError("规则表文件不存在。")
if normalized.parts[0] == "rules":
resolved = (SERVER_DIR / normalized).resolve()
allowed_root = self.rule_root
else:
resolved = (self.storage_root / normalized).resolve()
allowed_root = self.storage_root
try:
resolved.relative_to(allowed_root)
except ValueError:
raise FileNotFoundError("规则表文件不存在。") from None
return resolved
@staticmethod
def parse_version_markdown(markdown: str) -> RuleSpreadsheetMeta | None:
match = RULE_SPREADSHEET_BLOCK_PATTERN.search(str(markdown or ""))
if match is None:
return None
try:
payload = json.loads(match.group(1))
except json.JSONDecodeError:
return None
if not isinstance(payload, dict):
return None
return RuleSpreadsheetMeta(
file_name=str(payload.get("file_name") or "").strip(),
storage_key=str(payload.get("storage_key") or "").strip(),
mime_type=str(payload.get("mime_type") or SPREADSHEET_MIME_TYPE).strip()
or SPREADSHEET_MIME_TYPE,
size_bytes=int(payload.get("size_bytes") or 0),
checksum=str(payload.get("checksum") or "").strip(),
updated_at=str(payload.get("updated_at") or "").strip(),
updated_by=str(payload.get("updated_by") or "system").strip() or "system",
source=str(payload.get("source") or "upload").strip() or "upload",
)
@staticmethod
def build_version_markdown(
*,
rule_name: str,
version: str,
metadata: RuleSpreadsheetMeta,
) -> str:
sections = [
f"# {rule_name}",
"",
"## 规则载体",
"",
"- 详情类型Excel 表格",
f"- 当前规则版本:`{version}`",
f"- 表格文件:`{metadata.file_name}`",
f"- 最近更新人:{metadata.updated_by}",
f"- 最近更新时间:{metadata.updated_at}",
"",
"## 使用说明",
"",
"- 管理员可直接在规则中心内联编辑 Excel 表格,并通过 ONLYOFFICE 回写新版本。",
"- 上传新的 Excel 文件后,会自动生成新的规则版本快照。",
"- 切换到历史版本时仅提供预览,不允许直接覆盖历史快照。",
"",
"```rule-spreadsheet",
json.dumps(asdict(metadata), ensure_ascii=False, indent=2),
"```",
]
return "\n".join(sections)
@staticmethod
def build_rule_document_config(
metadata: RuleSpreadsheetMeta,
*,
asset_version: str,
) -> dict[str, object]:
return {
"kind": "spreadsheet",
"file_name": metadata.file_name,
"mime_type": metadata.mime_type,
"size_bytes": metadata.size_bytes,
"checksum": metadata.checksum,
"updated_at": metadata.updated_at,
"updated_by": metadata.updated_by,
"source": metadata.source,
"asset_version": asset_version,
}
@staticmethod
def build_company_travel_rule_template() -> bytes:
standard_rows = [
["费用分类", "适用场景", "票据要求", "报销标准", "审批要求", "备注"],
[
"长途交通",
"飞机、高铁、火车等跨城出行",
"行程单、车票、发票",
"据实报销",
"超预算需直属领导审批",
"优先选择公共交通",
],
[
"住宿费",
"出差住宿",
"酒店发票、入住清单",
"一线城市 650/晚;二线城市 500/晚;其他城市 380/晚",
"超标需总监审批",
"协议酒店优先",
],
[
"市内交通",
"出租车、网约车、地铁、公交",
"发票或电子行程单",
"150/天",
"超限需补充说明",
"夜间或无公共交通场景可豁免",
],
[
"餐补",
"出差期间日常补助",
"无需票据",
"120/天",
"系统自动核定",
"当天往返默认不享受",
],
[
"招待餐费",
"客户接待或项目宴请",
"餐饮发票、参与人清单",
"300/人",
"需业务负责人审批",
"需关联客户或项目",
],
]
instruction_rows = [
["字段", "填写说明"],
["费用分类", "建议保持固定选项,避免审批口径漂移。"],
["适用场景", "写清楚业务场景,例如客户拜访、项目驻场、参会等。"],
["票据要求", "必须明确哪些单据为必传,哪些场景允许补充说明替代。"],
["报销标准", "建议拆成统一金额、按城市等级、按职级分档三类口径。"],
["审批要求", "超标、例外、补录等情形应写清升级审批链。"],
["备注", "记录豁免条件、灰度口径或制度来源。"],
["版本建议", "每次修改表格后在规则中心同步生成一个新的规则版本。"],
]
return _build_xlsx_bytes(
[
("差旅报销标准", standard_rows),
("填表说明", instruction_rows),
]
)
@staticmethod
def build_blank_rule_workbook(sheet_name: str = "规则配置") -> bytes:
return _build_xlsx_bytes([(sheet_name, [[""]])])
@staticmethod
def rebuild_from_uploaded_content(content: bytes) -> bytes:
if not content:
raise ValueError("待导入的表格内容不能为空。")
try:
workbook = load_workbook(
filename=BytesIO(content),
read_only=True,
data_only=False,
)
except Exception as exc: # noqa: BLE001
raise ValueError("无法解析上传的 Excel 表格。") from exc
sheets: list[tuple[str, list[list[object]]]] = []
for worksheet in workbook.worksheets:
rows = [
list(row)
for row in worksheet.iter_rows(values_only=True)
]
sheets.append((worksheet.title, _trim_empty_table(rows)))
if not sheets:
raise ValueError("上传的 Excel 表格中没有可导入的工作表。")
return _build_xlsx_bytes(sheets)
def _build_xlsx_bytes(sheets: list[tuple[str, list[list[object]]]]) -> bytes:
created_at = datetime.now(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z")
workbook_buffer = BytesIO()
with ZipFile(workbook_buffer, "w", ZIP_DEFLATED) as archive:
archive.writestr("[Content_Types].xml", _build_content_types_xml(sheets))
archive.writestr("_rels/.rels", _build_root_rels_xml())
archive.writestr("docProps/app.xml", _build_app_xml(sheets))
archive.writestr("docProps/core.xml", _build_core_xml(created_at))
archive.writestr("xl/workbook.xml", _build_workbook_xml(sheets))
archive.writestr("xl/_rels/workbook.xml.rels", _build_workbook_rels_xml(sheets))
archive.writestr("xl/styles.xml", _build_styles_xml())
for index, (_, rows) in enumerate(sheets, start=1):
archive.writestr(
f"xl/worksheets/sheet{index}.xml",
_build_sheet_xml(rows),
)
return workbook_buffer.getvalue()
def _build_content_types_xml(sheets: list[tuple[str, list[list[object]]]]) -> str:
overrides = [
(
'<Override PartName="/xl/workbook.xml" '
'ContentType="application/vnd.openxmlformats-officedocument.'
'spreadsheetml.sheet.main+xml"/>'
),
(
'<Override PartName="/xl/styles.xml" '
'ContentType="application/vnd.openxmlformats-officedocument.'
'spreadsheetml.styles+xml"/>'
),
(
'<Override PartName="/docProps/core.xml" '
'ContentType="application/vnd.openxmlformats-package.core-properties+xml"/>'
),
(
'<Override PartName="/docProps/app.xml" '
'ContentType="application/vnd.openxmlformats-officedocument.'
'extended-properties+xml"/>'
),
]
overrides.extend(
[
(
f'<Override PartName="/xl/worksheets/sheet{index}.xml" '
'ContentType="application/vnd.openxmlformats-officedocument.'
'spreadsheetml.worksheet+xml"/>'
)
for index, _ in enumerate(sheets, start=1)
]
)
return (
'<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
'<Types xmlns="http://schemas.openxmlformats.org/package/2006/content-types">'
'<Default Extension="rels" '
'ContentType="application/vnd.openxmlformats-package.relationships+xml"/>'
'<Default Extension="xml" ContentType="application/xml"/>'
f'{"".join(overrides)}'
"</Types>"
)
def _build_root_rels_xml() -> str:
return (
'<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
'<Relationships xmlns="http://schemas.openxmlformats.org/package/2006/relationships">'
'<Relationship Id="rId1" '
'Type="http://schemas.openxmlformats.org/officeDocument/2006/'
'relationships/officeDocument" Target="xl/workbook.xml"/>'
'<Relationship Id="rId2" '
'Type="http://schemas.openxmlformats.org/package/2006/relationships/'
'metadata/core-properties" Target="docProps/core.xml"/>'
'<Relationship Id="rId3" '
'Type="http://schemas.openxmlformats.org/officeDocument/2006/'
'relationships/extended-properties" Target="docProps/app.xml"/>'
"</Relationships>"
)
def _build_app_xml(sheets: list[tuple[str, list[list[object]]]]) -> str:
titles = "".join(
[f'<vt:lpstr>{escape(name)}</vt:lpstr>' for name, _ in sheets]
)
sheet_count = len(sheets)
return (
'<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
'<Properties xmlns="http://schemas.openxmlformats.org/officeDocument/2006/'
'extended-properties" '
'xmlns:vt="http://schemas.openxmlformats.org/officeDocument/2006/docPropsVTypes">'
'<Application>Microsoft Excel</Application>'
'<HeadingPairs><vt:vector size="2" baseType="variant">'
"<vt:variant><vt:lpstr>Worksheets</vt:lpstr></vt:variant>"
f"<vt:variant><vt:i4>{sheet_count}</vt:i4></vt:variant>"
"</vt:vector></HeadingPairs>"
f'<TitlesOfParts><vt:vector size="{sheet_count}" baseType="lpstr">'
f"{titles}</vt:vector></TitlesOfParts>"
"</Properties>"
)
def _build_core_xml(created_at: str) -> str:
return (
'<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
'<cp:coreProperties xmlns:cp="http://schemas.openxmlformats.org/package/'
'2006/metadata/core-properties" '
'xmlns:dc="http://purl.org/dc/elements/1.1/" '
'xmlns:dcterms="http://purl.org/dc/terms/" '
'xmlns:dcmitype="http://purl.org/dc/dcmitype/" '
'xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">'
"<dc:creator>X-Financial</dc:creator>"
"<cp:lastModifiedBy>X-Financial</cp:lastModifiedBy>"
f'<dcterms:created xsi:type="dcterms:W3CDTF">{created_at}</dcterms:created>'
f'<dcterms:modified xsi:type="dcterms:W3CDTF">{created_at}</dcterms:modified>'
"</cp:coreProperties>"
)
def _build_workbook_xml(sheets: list[tuple[str, list[list[object]]]]) -> str:
sheet_items = "".join(
[
f'<sheet name="{escape(name)}" sheetId="{index}" r:id="rId{index}"/>'
for index, (name, _) in enumerate(sheets, start=1)
]
)
return (
'<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
'<workbook xmlns="http://schemas.openxmlformats.org/spreadsheetml/2006/main" '
'xmlns:r="http://schemas.openxmlformats.org/officeDocument/2006/relationships">'
"<bookViews><workbookView/></bookViews>"
f"<sheets>{sheet_items}</sheets>"
"</workbook>"
)
def _build_workbook_rels_xml(sheets: list[tuple[str, list[list[object]]]]) -> str:
relationships = "".join(
[
(
f'<Relationship Id="rId{index}" '
'Type="http://schemas.openxmlformats.org/officeDocument/2006/'
f'relationships/worksheet" Target="worksheets/sheet{index}.xml"/>'
)
for index, _ in enumerate(sheets, start=1)
]
)
relationships += (
f'<Relationship Id="rId{len(sheets) + 1}" '
'Type="http://schemas.openxmlformats.org/officeDocument/2006/relationships/styles" '
'Target="styles.xml"/>'
)
return (
'<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
'<Relationships xmlns="http://schemas.openxmlformats.org/package/2006/relationships">'
f"{relationships}"
"</Relationships>"
)
def _build_styles_xml() -> str:
return (
'<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
'<styleSheet xmlns="http://schemas.openxmlformats.org/spreadsheetml/2006/main">'
'<fonts count="1"><font><sz val="11"/><name val="Calibri"/></font></fonts>'
'<fills count="2"><fill><patternFill patternType="none"/></fill>'
'<fill><patternFill patternType="gray125"/></fill></fills>'
'<borders count="1"><border><left/><right/><top/><bottom/><diagonal/></border></borders>'
'<cellStyleXfs count="1">'
'<xf numFmtId="0" fontId="0" fillId="0" borderId="0"/>'
"</cellStyleXfs>"
'<cellXfs count="1">'
'<xf numFmtId="0" fontId="0" fillId="0" borderId="0" xfId="0"/>'
"</cellXfs>"
'<cellStyles count="1"><cellStyle name="Normal" xfId="0" builtinId="0"/></cellStyles>'
'</styleSheet>'
)
def _build_sheet_xml(rows: list[list[object]]) -> str:
normalized_rows = rows or [[""]]
max_column_count = max((len(row) for row in normalized_rows), default=1)
worksheet_rows: list[str] = []
for row_index, row in enumerate(normalized_rows, start=1):
cells: list[str] = []
for column_index, cell in enumerate(row, start=1):
ref = f"{_column_letter(column_index)}{row_index}"
text = "" if cell is None else str(cell)
preserve = ' xml:space="preserve"' if text.strip() != text or "\n" in text else ""
cells.append(
f'<c r="{ref}" t="inlineStr"><is><t{preserve}>{escape(text)}</t></is></c>'
)
worksheet_rows.append(f'<row r="{row_index}">{"".join(cells)}</row>')
dimension = f"A1:{_column_letter(max_column_count)}{len(normalized_rows)}"
return (
'<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
'<worksheet xmlns="http://schemas.openxmlformats.org/spreadsheetml/2006/main">'
f'<dimension ref="{dimension}"/>'
"<sheetViews><sheetView workbookViewId=\"0\"/></sheetViews>"
"<sheetFormatPr defaultRowHeight=\"18\"/>"
f"<sheetData>{''.join(worksheet_rows)}</sheetData>"
"</worksheet>"
)
def _column_letter(index: int) -> str:
value = max(1, int(index))
result = ""
while value > 0:
value, remainder = divmod(value - 1, 26)
result = f"{chr(65 + remainder)}{result}"
return result
def _trim_empty_table(rows: list[list[object]]) -> list[list[object]]:
normalized_rows = [list(row) for row in rows]
while normalized_rows and all(cell in (None, "") for cell in normalized_rows[-1]):
normalized_rows.pop()
if not normalized_rows:
return [[""]]
max_column = 0
for row in normalized_rows:
for index, cell in enumerate(row, start=1):
if cell not in (None, ""):
max_column = max(max_column, index)
if max_column <= 0:
return [[""]]
return [row[:max_column] for row in normalized_rows]