222 lines
6.9 KiB
Python
222 lines
6.9 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
"""解析 JNPF 风格左/上父格(none / default / custom)。"""
|
||
from __future__ import annotations
|
||
|
||
from typing import Any, Dict, List, Optional, Tuple
|
||
|
||
CellKey = Tuple[str, int, int]
|
||
|
||
|
||
def col_letter_to_index(letters: str) -> int:
|
||
s = (letters or "A").upper()
|
||
n = 0
|
||
for ch in s:
|
||
if not ("A" <= ch <= "Z"):
|
||
continue
|
||
n = n * 26 + (ord(ch) - 64)
|
||
return max(0, n - 1)
|
||
|
||
|
||
def col_index_to_letter(col: int) -> str:
|
||
n = col
|
||
s = ""
|
||
while n >= 0:
|
||
s = chr(65 + (n % 26)) + s
|
||
n = n // 26 - 1
|
||
return s or "A"
|
||
|
||
|
||
def _int_coord(v: Any, default: int = 0) -> int:
|
||
try:
|
||
return int(v)
|
||
except (TypeError, ValueError):
|
||
return default
|
||
|
||
|
||
def cell_key(cell: Dict[str, Any]) -> CellKey:
|
||
return (
|
||
str(cell.get("sheet") or "sheet1"),
|
||
_int_coord(cell.get("row"), 0),
|
||
_int_coord(cell.get("col"), 0),
|
||
)
|
||
|
||
|
||
def _is_data_source(cell: Dict[str, Any]) -> bool:
|
||
return cell.get("type") == "dataSource"
|
||
|
||
|
||
def _index_data_sources(cells: List[Dict[str, Any]]) -> Tuple[Dict[CellKey, Dict[str, Any]], Dict[str, List[Dict[str, Any]]]]:
|
||
by_pos: Dict[CellKey, Dict[str, Any]] = {}
|
||
by_sheet: Dict[str, List[Dict[str, Any]]] = {}
|
||
for c in cells:
|
||
if not _is_data_source(c):
|
||
continue
|
||
k = cell_key(c)
|
||
by_pos[k] = c
|
||
by_sheet.setdefault(k[0], []).append(c)
|
||
return by_pos, by_sheet
|
||
|
||
|
||
def resolve_parent(
|
||
cell: Dict[str, Any],
|
||
*,
|
||
is_left: bool,
|
||
by_pos: Dict[CellKey, Dict[str, Any]],
|
||
by_sheet: Dict[str, List[Dict[str, Any]]],
|
||
) -> Optional[Dict[str, Any]]:
|
||
"""返回父格 dataSource 元数据;none 为 None,default 为最近左/上数据源格。"""
|
||
custom = cell.get("custom") or {}
|
||
ptype = (
|
||
custom.get("leftParentCellType") if is_left else custom.get("topParentCellType")
|
||
) or "default"
|
||
sheet, row, col = cell_key(cell)
|
||
|
||
if ptype == "none":
|
||
return None
|
||
|
||
if ptype == "custom":
|
||
if is_left:
|
||
letters = custom.get("leftParentCellCustomRowName") or "A"
|
||
row_num = custom.get("leftParentCellCustomColName")
|
||
else:
|
||
letters = custom.get("topParentCellCustomRowName") or "A"
|
||
row_num = custom.get("topParentCellCustomColName")
|
||
if row_num is None:
|
||
return None
|
||
try:
|
||
parent_row = int(row_num) - 1
|
||
except (TypeError, ValueError):
|
||
return None
|
||
parent_col = col_letter_to_index(str(letters))
|
||
return by_pos.get((sheet, parent_row, parent_col))
|
||
|
||
# default:同行向左 / 同列向上找最近 dataSource
|
||
candidates = by_sheet.get(sheet) or []
|
||
best: Optional[Dict[str, Any]] = None
|
||
if is_left:
|
||
for c in candidates:
|
||
cr, cc = _int_coord(c.get("row")), _int_coord(c.get("col"))
|
||
if cr == row and cc < col:
|
||
if best is None or _int_coord(best.get("col")) < cc:
|
||
best = c
|
||
else:
|
||
for c in candidates:
|
||
cr, cc = _int_coord(c.get("row")), _int_coord(c.get("col"))
|
||
if cc == col and cr < row:
|
||
if best is None or _int_coord(best.get("row")) < cr:
|
||
best = c
|
||
return best
|
||
|
||
|
||
def resolve_parents(
|
||
cell: Dict[str, Any],
|
||
by_pos: Dict[CellKey, Dict[str, Any]],
|
||
by_sheet: Dict[str, List[Dict[str, Any]]],
|
||
) -> Tuple[Optional[Dict[str, Any]], Optional[Dict[str, Any]]]:
|
||
left = resolve_parent(cell, is_left=True, by_pos=by_pos, by_sheet=by_sheet)
|
||
top = resolve_parent(cell, is_left=False, by_pos=by_pos, by_sheet=by_sheet)
|
||
custom = cell.get("custom") or {}
|
||
ds_name = _dataset_name(cell)
|
||
# 汇总格且无扩展时,JNPF 在 none+none 时清除父格
|
||
poly = str(custom.get("polymerizationType") or "1")
|
||
if poly == "3" and custom.get("leftParentCellType") == "none" and custom.get("topParentCellType") == "none":
|
||
return None, None
|
||
if left and _dataset_name(left) != ds_name:
|
||
left = None
|
||
if top and _dataset_name(top) != ds_name:
|
||
top = None
|
||
return left, top
|
||
|
||
|
||
def _dataset_name(cell: Dict[str, Any]) -> str:
|
||
custom = cell.get("custom") or {}
|
||
name = str(
|
||
custom.get("dataSetName") or custom.get("dataSet") or custom.get("alias") or ""
|
||
)
|
||
if name:
|
||
return name
|
||
field = str(custom.get("field") or custom.get("bindField") or "")
|
||
if "." in field:
|
||
return field.split(".", 1)[0]
|
||
return ""
|
||
|
||
|
||
def filter_rows_by_parents(
|
||
rows: List[Any],
|
||
cell: Dict[str, Any],
|
||
*,
|
||
left_parent: Optional[Dict[str, Any]],
|
||
top_parent: Optional[Dict[str, Any]],
|
||
left_bind: Optional[List[Dict[str, Any]]],
|
||
top_bind: Optional[List[Dict[str, Any]]],
|
||
) -> List[Dict[str, Any]]:
|
||
"""对齐 JNPF DataUtils.fetchData:按父格 bindData 切片过滤。"""
|
||
data = [r if isinstance(r, dict) else {} for r in rows]
|
||
if not left_bind and not top_bind:
|
||
return data
|
||
|
||
left_rows = left_bind
|
||
top_rows = top_bind
|
||
if left_rows is None and top_rows is not None:
|
||
return top_rows
|
||
if top_rows is None and left_rows is not None:
|
||
return left_rows
|
||
if left_rows is None or top_rows is None:
|
||
return data
|
||
|
||
left_field = _bind_field(left_parent) if left_parent else ""
|
||
top_field = _bind_field(top_parent) if top_parent else ""
|
||
left_val = _first_field_value(left_rows, left_field) if left_rows else None
|
||
top_val = _first_field_value(top_rows, top_field) if top_rows else None
|
||
|
||
from_top: List[Dict[str, Any]] = []
|
||
for row in top_rows:
|
||
if left_field and _get_nested_value(row, left_field) == left_val:
|
||
from_top.append(row)
|
||
from_left: List[Dict[str, Any]] = []
|
||
for row in left_rows:
|
||
if top_field and _get_nested_value(row, top_field) == top_val:
|
||
from_left.append(row)
|
||
return from_top if len(from_top) <= len(from_left) else from_left
|
||
|
||
|
||
def _bind_field(parent: Optional[Dict[str, Any]]) -> str:
|
||
if not parent:
|
||
return ""
|
||
custom = parent.get("custom") or {}
|
||
field = custom.get("field") or custom.get("bindField") or ""
|
||
if "." in field:
|
||
return field.split(".", 1)[1]
|
||
return field
|
||
|
||
|
||
def resolve_field_path(field: str, dataset_alias: str = "") -> str:
|
||
"""JNPF 字段常为 alias.prop,行数据一般为扁平 prop 或嵌套 prop。"""
|
||
field = str(field or "")
|
||
alias = str(dataset_alias or "")
|
||
if alias and field.startswith(f"{alias}."):
|
||
return field[len(alias) + 1 :]
|
||
return field
|
||
|
||
|
||
def _get_nested_value(row: Dict[str, Any], field: str) -> Any:
|
||
if not field:
|
||
return None
|
||
if field in row:
|
||
return row[field]
|
||
parts = field.split(".")
|
||
cur: Any = row
|
||
for p in parts:
|
||
if isinstance(cur, dict) and p in cur:
|
||
cur = cur[p]
|
||
else:
|
||
return None
|
||
return cur
|
||
|
||
|
||
def _first_field_value(rows: List[Dict[str, Any]], field: str) -> Any:
|
||
if not rows or not field:
|
||
return None
|
||
return _get_nested_value(rows[0], field)
|