feat: restore source parity and harden agent runtime
This commit is contained in:
@@ -0,0 +1,91 @@
|
||||
#!/usr/bin/env python
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
预览 MVP:参数单元格替换(Phase 2)
|
||||
完整 dataSource 扩展在 Phase 3 convert 引擎实现
|
||||
"""
|
||||
import copy
|
||||
import re
|
||||
from typing import Any, Dict, List
|
||||
|
||||
|
||||
_PARAM_PATTERN = re.compile(r"#\{([^}]+)\}")
|
||||
|
||||
|
||||
def _replace_params_in_value(value: Any, params: Dict[str, Any]) -> Any:
|
||||
if not isinstance(value, str):
|
||||
return value
|
||||
def repl(m):
|
||||
key = m.group(1).strip()
|
||||
if key in params:
|
||||
return str(params[key])
|
||||
return m.group(0)
|
||||
return _PARAM_PATTERN.sub(repl, value)
|
||||
|
||||
|
||||
def apply_parameter_cells(
|
||||
snapshot: Dict[str, Any],
|
||||
cells_meta: Dict[str, Any],
|
||||
params: Dict[str, Any],
|
||||
) -> Dict[str, Any]:
|
||||
"""将 parameter 类型绑定写入 snapshot(MVP)"""
|
||||
result = copy.deepcopy(snapshot)
|
||||
if not result or not cells_meta:
|
||||
return result
|
||||
|
||||
cell_list = cells_meta.get("cells") or []
|
||||
sheets = result.get("sheets") or {}
|
||||
|
||||
for cell in cell_list:
|
||||
if cell.get("type") != "parameter":
|
||||
continue
|
||||
sheet_id = cell.get("sheet")
|
||||
row = cell.get("row", 0)
|
||||
col = cell.get("col", 0)
|
||||
custom = cell.get("custom") or {}
|
||||
text = custom.get("value") or custom.get("text") or ""
|
||||
if isinstance(text, str):
|
||||
text = _replace_params_in_value(text, params)
|
||||
sheet = sheets.get(sheet_id)
|
||||
if not sheet:
|
||||
continue
|
||||
cell_data = sheet.setdefault("cellData", {})
|
||||
row_data = cell_data.setdefault(str(row), {})
|
||||
cell_obj = row_data.setdefault(str(col), {})
|
||||
cell_obj["v"] = text
|
||||
if "custom" in cell_obj:
|
||||
cell_obj["custom"] = {**cell_obj.get("custom", {}), "value": text}
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def apply_snapshot_placeholders(
|
||||
snapshot: Dict[str, Any],
|
||||
params: Dict[str, Any],
|
||||
) -> Dict[str, Any]:
|
||||
"""扫描 snapshot 所有单元格,将 v / custom 字符串中的 #{param} 替换为查询参数"""
|
||||
if not snapshot or not params:
|
||||
return snapshot or {}
|
||||
result = copy.deepcopy(snapshot)
|
||||
sheets = result.get("sheets") or {}
|
||||
for sheet in sheets.values():
|
||||
if not isinstance(sheet, dict):
|
||||
continue
|
||||
cell_data = sheet.get("cellData") or {}
|
||||
for row in cell_data.values():
|
||||
if not isinstance(row, dict):
|
||||
continue
|
||||
for cell in row.values():
|
||||
if not isinstance(cell, dict):
|
||||
continue
|
||||
v = cell.get("v")
|
||||
if isinstance(v, str) and "#{" in v:
|
||||
cell["v"] = _replace_params_in_value(v, params)
|
||||
custom = cell.get("custom")
|
||||
if isinstance(custom, dict):
|
||||
for key, val in list(custom.items()):
|
||||
if isinstance(val, str) and "#{" in val:
|
||||
custom[key] = _replace_params_in_value(val, params)
|
||||
cell["custom"] = custom
|
||||
result["sheets"] = sheets
|
||||
return result
|
||||
Reference in New Issue
Block a user