944 lines
30 KiB
Python
944 lines
30 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
"""父格 + 聚合驱动的数据源扩展（全表拓扑，支持跨行上父格 / 横向父格树）。"""
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
from dataclasses import dataclass
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
from online_dev.report_manager.engine.parent_cells import (
|
|
cell_key,
|
|
filter_rows_by_parents,
|
|
resolve_field_path,
|
|
resolve_parents,
|
|
_index_data_sources,
|
|
_get_nested_value,
|
|
_int_coord,
|
|
)
|
|
from online_dev.report_manager.engine.polymerize import BindData, build_bind_list, expand_span
|
|
|
|
CellKey = Tuple[str, int, int]
|
|
|
|
|
|
@dataclass
|
|
class ExpandedSlot:
|
|
"""单元格一次绑定扩展占用的行/列区间（左闭右开）。"""
|
|
|
|
value: Any
|
|
data_list: List[Dict[str, Any]]
|
|
start: int
|
|
end: int
|
|
|
|
|
|
def _expand_direction(custom: Dict[str, Any]) -> str:
|
|
expand = (custom.get("expand") or custom.get("expandDirection") or "").lower()
|
|
if expand in ("down", "list", "vertical"):
|
|
return "down"
|
|
if expand in ("right", "horizontal", "across"):
|
|
return "right"
|
|
fill = (custom.get("fillDirection") or "").lower()
|
|
if fill in ("portrait", "vertical", "down"):
|
|
return "down"
|
|
if fill in ("landscape", "horizontal", "right", "across"):
|
|
return "right"
|
|
return "none"
|
|
|
|
|
|
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 _field_name(cell: Dict[str, Any]) -> str:
|
|
custom = cell.get("custom") or {}
|
|
return str(custom.get("field") or custom.get("bindField") or "")
|
|
|
|
|
|
def _poly_list(cell: Dict[str, Any]) -> bool:
|
|
return str((cell.get("custom") or {}).get("polymerizationType") or "1") == "1"
|
|
|
|
|
|
def _poly_summary(cell: Dict[str, Any]) -> bool:
|
|
return str((cell.get("custom") or {}).get("polymerizationType") or "1") == "3"
|
|
|
|
|
|
def _poly_group(cell: Dict[str, Any]) -> bool:
|
|
return str((cell.get("custom") or {}).get("polymerizationType") or "1") == "2"
|
|
|
|
|
|
def _should_merge_group(cell: Dict[str, Any]) -> bool:
|
|
"""分组列是否合并单元格（custom.mergeCell，默认开启）。"""
|
|
if not _poly_group(cell):
|
|
return False
|
|
merge = (cell.get("custom") or {}).get("mergeCell")
|
|
if merge is None:
|
|
return True
|
|
if isinstance(merge, str):
|
|
return merge.strip().lower() not in ("0", "false", "no", "")
|
|
return bool(merge)
|
|
|
|
|
|
def _add_vertical_merge_region(
|
|
sheets: Dict[str, Any],
|
|
sheet_id: str,
|
|
start_row: int,
|
|
end_row: int,
|
|
col: int,
|
|
) -> None:
|
|
"""分组列：同组多行合并为一个单元格（预览/导出）。"""
|
|
if end_row <= start_row:
|
|
return
|
|
sheet = sheets.get(sheet_id)
|
|
if not sheet:
|
|
return
|
|
region = {
|
|
"startRow": start_row,
|
|
"endRow": end_row,
|
|
"startColumn": col,
|
|
"endColumn": col,
|
|
}
|
|
merges = sheet.setdefault("mergeData", [])
|
|
if region not in merges:
|
|
merges.append(region)
|
|
|
|
|
|
def _compute_down_band_lengths(
|
|
down_by_sheet: Dict[str, List[Dict[str, Any]]],
|
|
datasets: Dict[str, List[Any]],
|
|
) -> Dict[Tuple[str, int], int]:
|
|
lengths: Dict[Tuple[str, int], int] = {}
|
|
for sheet_id, down_cells in down_by_sheet.items():
|
|
bands: Dict[int, List[Dict[str, Any]]] = {}
|
|
for cell in down_cells:
|
|
bands.setdefault(_int_coord(cell.get("row")), []).append(cell)
|
|
for start_row, band in bands.items():
|
|
max_len = max(
|
|
(len(datasets.get(_dataset_name(item)) or []) for item in band),
|
|
default=0,
|
|
)
|
|
lengths[(sheet_id, start_row)] = max_len
|
|
return lengths
|
|
|
|
|
|
def _shift_sheet_rows_down(
|
|
sheets: Dict[str, Any],
|
|
sheet_id: str,
|
|
from_row: int,
|
|
delta: int,
|
|
) -> None:
|
|
"""将 from_row 及以下的 cellData 整体下移 delta 行（列表扩展前保留汇总/static 行）。"""
|
|
if delta <= 0:
|
|
return
|
|
sheet = sheets.get(sheet_id)
|
|
if not sheet:
|
|
return
|
|
cell_data = sheet.get("cellData") or {}
|
|
shifted: Dict[str, Any] = {}
|
|
for row_key, row_obj in cell_data.items():
|
|
row = _int_coord(row_key, 0)
|
|
target_key = str(row + delta) if row >= from_row else row_key
|
|
if target_key in shifted and row >= from_row:
|
|
existing = shifted[target_key]
|
|
if isinstance(existing, dict) and isinstance(row_obj, dict):
|
|
merged = {**existing, **row_obj}
|
|
shifted[target_key] = merged
|
|
else:
|
|
shifted[target_key] = row_obj
|
|
else:
|
|
shifted[target_key] = row_obj
|
|
sheet["cellData"] = shifted
|
|
|
|
|
|
def _summary_filter_rows(
|
|
cell: Dict[str, Any],
|
|
rows: List[Dict[str, Any]],
|
|
by_pos: Dict[CellKey, Dict[str, Any]],
|
|
by_sheet: Dict[str, List[Dict[str, Any]]],
|
|
) -> List[Dict[str, Any]]:
|
|
"""汇总格：仅 custom 父格按切片统计；default/none 对整表数据集聚合（合计行）。"""
|
|
custom = cell.get("custom") or {}
|
|
left_pt = custom.get("leftParentCellType") or "default"
|
|
top_pt = custom.get("topParentCellType") or "default"
|
|
if left_pt != "custom" and top_pt != "custom":
|
|
return rows
|
|
left, top = resolve_parents(cell, by_pos, by_sheet)
|
|
left_bind = top_bind = None
|
|
if left:
|
|
lb = build_bind_list(left, rows)
|
|
left_bind = lb[0].data_list if lb else None
|
|
if top:
|
|
tb = build_bind_list(top, rows)
|
|
top_bind = tb[0].data_list if tb else None
|
|
return filter_rows_by_parents(
|
|
rows,
|
|
cell,
|
|
left_parent=left,
|
|
top_parent=top,
|
|
left_bind=left_bind,
|
|
top_bind=top_bind,
|
|
)
|
|
|
|
|
|
def _summary_output_row(
|
|
sheet_id: str,
|
|
template_row: int,
|
|
band_lengths: Dict[Tuple[str, int], int],
|
|
) -> int:
|
|
"""汇总格位于列表模板行下方时，输出到扩展后的末行（对齐 JNPF 合计行）。"""
|
|
anchor: Optional[Tuple[int, int]] = None
|
|
for (sid, start_row), max_len in band_lengths.items():
|
|
if sid != sheet_id or max_len <= 0 or template_row <= start_row:
|
|
continue
|
|
if anchor is None or start_row > anchor[0]:
|
|
anchor = (start_row, max_len)
|
|
if anchor is None:
|
|
return template_row
|
|
start_row, max_len = anchor
|
|
return template_row + (max_len - 1)
|
|
|
|
|
|
def _template_style_ref(
|
|
sheets: Dict[str, Any],
|
|
sheet_id: str,
|
|
row: int,
|
|
col: int,
|
|
) -> Optional[Any]:
|
|
sheet = sheets.get(sheet_id)
|
|
if not sheet:
|
|
return None
|
|
cell_data = sheet.get("cellData") or {}
|
|
cell = (cell_data.get(str(row)) or {}).get(str(col))
|
|
if not isinstance(cell, dict):
|
|
return None
|
|
return cell.get("s")
|
|
|
|
|
|
def _set_cell_value(
|
|
sheets: Dict[str, Any],
|
|
sheet_id: str,
|
|
row: int,
|
|
col: int,
|
|
value: Any,
|
|
*,
|
|
style_template_row: Optional[int] = None,
|
|
style_template_col: Optional[int] = None,
|
|
) -> None:
|
|
sheet = sheets.get(sheet_id)
|
|
if not sheet:
|
|
return
|
|
cell_data = sheet.setdefault("cellData", {})
|
|
row_data = cell_data.setdefault(str(row), {})
|
|
cell_obj = row_data.setdefault(str(col), {})
|
|
style_ref = None
|
|
if style_template_row is not None:
|
|
style_ref = _template_style_ref(sheets, sheet_id, style_template_row, col)
|
|
elif style_template_col is not None:
|
|
style_ref = _template_style_ref(sheets, sheet_id, row, style_template_col)
|
|
if style_ref is not None:
|
|
cell_obj["s"] = style_ref
|
|
if value is None:
|
|
cell_obj["v"] = ""
|
|
elif isinstance(value, (dict, list)):
|
|
cell_obj["v"] = str(value)
|
|
else:
|
|
cell_obj["v"] = value
|
|
|
|
|
|
def _apply_fill_empty(
|
|
sheets: Dict[str, Any],
|
|
sheet_id: str,
|
|
custom: Dict[str, Any],
|
|
*,
|
|
direction: str,
|
|
anchor_row: int,
|
|
anchor_col: int,
|
|
data_len: int,
|
|
) -> None:
|
|
if not custom.get("fillEmptyRows"):
|
|
return
|
|
try:
|
|
fill_n = int(custom.get("fillEmptyNum") or 1)
|
|
except (TypeError, ValueError):
|
|
fill_n = 1
|
|
if fill_n < 1:
|
|
return
|
|
for offset in range(fill_n):
|
|
if direction == "down":
|
|
_set_cell_value(
|
|
sheets, sheet_id, anchor_row + data_len + offset, anchor_col, ""
|
|
)
|
|
else:
|
|
_set_cell_value(
|
|
sheets, sheet_id, anchor_row, anchor_col + data_len + offset, ""
|
|
)
|
|
|
|
|
|
def _slot_at_index(slots: List[ExpandedSlot], index: int) -> Optional[ExpandedSlot]:
|
|
if 0 <= index < len(slots):
|
|
return slots[index]
|
|
return None
|
|
|
|
|
|
def _slot_covering(slots: List[ExpandedSlot], pos: int) -> Optional[ExpandedSlot]:
|
|
for s in slots:
|
|
if s.start <= pos < s.end:
|
|
return s
|
|
return None
|
|
|
|
|
|
def _children_of(
|
|
parent: Dict[str, Any],
|
|
candidates: List[Dict[str, Any]],
|
|
by_pos: Dict[CellKey, Dict[str, Any]],
|
|
by_sheet: Dict[str, List[Dict[str, Any]]],
|
|
) -> List[Dict[str, Any]]:
|
|
pk = cell_key(parent)
|
|
out: List[Dict[str, Any]] = []
|
|
for c in candidates:
|
|
left, top = resolve_parents(c, by_pos, by_sheet)
|
|
if (left and cell_key(left) == pk) or (top and cell_key(top) == pk):
|
|
out.append(c)
|
|
return sorted(out, key=lambda x: (_int_coord(x.get("row")), _int_coord(x.get("col"))))
|
|
|
|
|
|
def _is_down_root(
|
|
cell: Dict[str, Any],
|
|
down_keys: set[CellKey],
|
|
by_pos: Dict[CellKey, Dict[str, Any]],
|
|
by_sheet: Dict[str, List[Dict[str, Any]]],
|
|
) -> bool:
|
|
left, top = resolve_parents(cell, by_pos, by_sheet)
|
|
if left and cell_key(left) in down_keys:
|
|
return False
|
|
if top and cell_key(top) in down_keys:
|
|
return False
|
|
return True
|
|
|
|
|
|
def _rows_for_bind(
|
|
rows: List[Dict[str, Any]],
|
|
cell: Dict[str, Any],
|
|
*,
|
|
left_parent: Optional[Dict[str, Any]],
|
|
top_parent: Optional[Dict[str, Any]],
|
|
left_slot: Optional[ExpandedSlot],
|
|
top_slot: Optional[ExpandedSlot],
|
|
) -> List[Dict[str, Any]]:
|
|
left_bind = left_slot.data_list if left_slot else None
|
|
top_bind = top_slot.data_list if top_slot else None
|
|
return filter_rows_by_parents(
|
|
rows,
|
|
cell,
|
|
left_parent=left_parent,
|
|
top_parent=top_parent,
|
|
left_bind=left_bind,
|
|
top_bind=top_bind,
|
|
)
|
|
|
|
|
|
def apply_pending_group_merges(
|
|
snapshot: Dict[str, Any],
|
|
pending: List[Tuple[str, int, int, int]],
|
|
) -> Dict[str, Any]:
|
|
sheets = snapshot.get("sheets") or {}
|
|
for sheet_id, col, start_row, end_row in pending:
|
|
_add_vertical_merge_region(sheets, sheet_id, start_row, end_row, col)
|
|
return snapshot
|
|
|
|
|
|
def _visit_down(
|
|
cell: Dict[str, Any],
|
|
rows: List[Dict[str, Any]],
|
|
row_cursor: int,
|
|
*,
|
|
sheets: Dict[str, Any],
|
|
sheet_id: str,
|
|
down_cells: List[Dict[str, Any]],
|
|
datasets: Dict[str, List[Any]],
|
|
by_pos: Dict[CellKey, Dict[str, Any]],
|
|
by_sheet: Dict[str, List[Dict[str, Any]]],
|
|
row_registry: Dict[CellKey, List[ExpandedSlot]],
|
|
bind_index: int = 0,
|
|
left_slot: Optional[ExpandedSlot] = None,
|
|
top_slot: Optional[ExpandedSlot] = None,
|
|
parent_scoped: bool = False,
|
|
pending_group_merges: Optional[List[Tuple[str, int, int, int]]] = None,
|
|
) -> int:
|
|
custom = cell.get("custom") or {}
|
|
field = resolve_field_path(_field_name(cell), _dataset_name(cell))
|
|
col = _int_coord(cell.get("col"))
|
|
anchor_row = _int_coord(cell.get("row"))
|
|
left_p, top_p = resolve_parents(cell, by_pos, by_sheet)
|
|
|
|
if not parent_scoped:
|
|
if left_slot is None and left_p:
|
|
left_slot = _slot_at_index(
|
|
row_registry.get(cell_key(left_p), []), bind_index
|
|
)
|
|
if top_slot is None and top_p:
|
|
top_slot = _slot_at_index(
|
|
row_registry.get(cell_key(top_p), []), bind_index
|
|
)
|
|
filtered = _rows_for_bind(
|
|
rows,
|
|
cell,
|
|
left_parent=left_p,
|
|
top_parent=top_p,
|
|
left_slot=left_slot,
|
|
top_slot=top_slot,
|
|
)
|
|
else:
|
|
filtered = [r if isinstance(r, dict) else {} for r in rows]
|
|
|
|
binds = build_bind_list(cell, filtered)
|
|
children = _children_of(cell, down_cells, by_pos, by_sheet)
|
|
child_cols = {_int_coord(ch.get("col")) for ch in children}
|
|
slots: List[ExpandedSlot] = []
|
|
end_row = row_cursor
|
|
|
|
for bi, bind in enumerate(binds):
|
|
block_start = end_row
|
|
if children:
|
|
child_end = block_start
|
|
for ch in children:
|
|
child_end = _visit_down(
|
|
ch,
|
|
bind.data_list,
|
|
block_start,
|
|
sheets=sheets,
|
|
sheet_id=sheet_id,
|
|
down_cells=down_cells,
|
|
datasets=datasets,
|
|
by_pos=by_pos,
|
|
by_sheet=by_sheet,
|
|
row_registry=row_registry,
|
|
bind_index=bi,
|
|
parent_scoped=True,
|
|
pending_group_merges=pending_group_merges,
|
|
)
|
|
block_len = max(1, child_end - block_start)
|
|
else:
|
|
block_len = expand_span(bind, cell)
|
|
|
|
if col not in child_cols:
|
|
style_row = anchor_row
|
|
if _poly_list(cell):
|
|
for i in range(block_len):
|
|
r = block_start + i
|
|
val = (
|
|
_get_nested_value(bind.data_list[i], field)
|
|
if i < len(bind.data_list)
|
|
else ""
|
|
)
|
|
_set_cell_value(
|
|
sheets,
|
|
sheet_id,
|
|
r,
|
|
col,
|
|
val,
|
|
style_template_row=style_row,
|
|
)
|
|
elif _poly_group(cell):
|
|
for i in range(block_len):
|
|
r = block_start + i
|
|
_set_cell_value(
|
|
sheets,
|
|
sheet_id,
|
|
r,
|
|
col,
|
|
bind.value,
|
|
style_template_row=style_row,
|
|
)
|
|
if (
|
|
pending_group_merges is not None
|
|
and block_len > 1
|
|
and _should_merge_group(cell)
|
|
):
|
|
pending_group_merges.append(
|
|
(sheet_id, col, block_start, block_start + block_len - 1)
|
|
)
|
|
else:
|
|
for i in range(block_len):
|
|
r = block_start + i
|
|
_set_cell_value(
|
|
sheets,
|
|
sheet_id,
|
|
r,
|
|
col,
|
|
bind.value,
|
|
style_template_row=style_row,
|
|
)
|
|
|
|
slots.append(
|
|
ExpandedSlot(bind.value, bind.data_list, block_start, block_start + block_len)
|
|
)
|
|
end_row = block_start + block_len
|
|
|
|
key = cell_key(cell)
|
|
row_registry[key] = row_registry.get(key, []) + slots
|
|
|
|
if custom.get("fillEmptyRows"):
|
|
_apply_fill_empty(
|
|
sheets,
|
|
sheet_id,
|
|
custom,
|
|
direction="down",
|
|
anchor_row=anchor_row,
|
|
anchor_col=col,
|
|
data_len=end_row - row_cursor,
|
|
)
|
|
return end_row
|
|
|
|
|
|
def _expand_sheet_down(
|
|
sheets: Dict[str, Any],
|
|
sheet_id: str,
|
|
down_cells: List[Dict[str, Any]],
|
|
datasets: Dict[str, List[Any]],
|
|
by_pos: Dict[CellKey, Dict[str, Any]],
|
|
by_sheet: Dict[str, List[Dict[str, Any]]],
|
|
pending_group_merges: Optional[List[Tuple[str, int, int, int]]] = None,
|
|
) -> None:
|
|
down_keys = {cell_key(c) for c in down_cells}
|
|
row_registry: Dict[CellKey, List[ExpandedSlot]] = {}
|
|
roots = [c for c in down_cells if _is_down_root(c, down_keys, by_pos, by_sheet)]
|
|
if not roots:
|
|
roots = down_cells
|
|
|
|
bands: Dict[int, List[Dict[str, Any]]] = {}
|
|
for c in roots:
|
|
bands.setdefault(_int_coord(c.get("row")), []).append(c)
|
|
|
|
for start_row in sorted(bands.keys()):
|
|
cursor = start_row
|
|
for root in sorted(bands[start_row], key=lambda x: _int_coord(x.get("col"))):
|
|
ds = _dataset_name(root)
|
|
raw = datasets.get(ds) or []
|
|
rows = [r if isinstance(r, dict) else {} for r in raw]
|
|
cursor = max(
|
|
cursor,
|
|
_visit_down(
|
|
root,
|
|
rows,
|
|
cursor,
|
|
sheets=sheets,
|
|
sheet_id=sheet_id,
|
|
down_cells=down_cells,
|
|
datasets=datasets,
|
|
by_pos=by_pos,
|
|
by_sheet=by_sheet,
|
|
row_registry=row_registry,
|
|
pending_group_merges=pending_group_merges,
|
|
),
|
|
)
|
|
|
|
|
|
def _legacy_down_band(
|
|
sheets: Dict[str, Any],
|
|
sheet_id: str,
|
|
start_row: int,
|
|
band: List[Dict[str, Any]],
|
|
datasets: Dict[str, List[Any]],
|
|
pending_group_merges: Optional[List[Tuple[str, int, int, int]]] = None,
|
|
) -> None:
|
|
max_len = 0
|
|
for item in band:
|
|
max_len = max(max_len, len(datasets.get(_dataset_name(item)) or []))
|
|
for item in sorted(band, key=lambda x: _int_coord(x.get("col"))):
|
|
ds = _dataset_name(item)
|
|
field = resolve_field_path(_field_name(item), ds)
|
|
col = _int_coord(item.get("col"))
|
|
data_rows = [r if isinstance(r, dict) else {} for r in datasets.get(ds) or []]
|
|
custom = item.get("custom") or {}
|
|
if _poly_group(item):
|
|
binds = build_bind_list(item, data_rows)
|
|
row_cursor = start_row
|
|
for bind in binds:
|
|
span = max(1, len(bind.data_list))
|
|
for i in range(span):
|
|
_set_cell_value(
|
|
sheets,
|
|
sheet_id,
|
|
row_cursor + i,
|
|
col,
|
|
bind.value,
|
|
style_template_row=start_row,
|
|
)
|
|
if (
|
|
span > 1
|
|
and pending_group_merges is not None
|
|
and _should_merge_group(item)
|
|
):
|
|
pending_group_merges.append(
|
|
(sheet_id, col, row_cursor, row_cursor + span - 1)
|
|
)
|
|
row_cursor += span
|
|
if custom.get("fillEmptyRows"):
|
|
_apply_fill_empty(
|
|
sheets,
|
|
sheet_id,
|
|
custom,
|
|
direction="down",
|
|
anchor_row=start_row,
|
|
anchor_col=col,
|
|
data_len=row_cursor - start_row,
|
|
)
|
|
continue
|
|
for i in range(max_len):
|
|
if i < len(data_rows):
|
|
val = _get_nested_value(data_rows[i], field)
|
|
else:
|
|
val = ""
|
|
_set_cell_value(
|
|
sheets,
|
|
sheet_id,
|
|
start_row + i,
|
|
col,
|
|
val,
|
|
style_template_row=start_row,
|
|
)
|
|
if custom.get("fillEmptyRows"):
|
|
_apply_fill_empty(
|
|
sheets,
|
|
sheet_id,
|
|
custom,
|
|
direction="down",
|
|
anchor_row=start_row,
|
|
anchor_col=col,
|
|
data_len=max_len,
|
|
)
|
|
|
|
|
|
def _has_down_parent_link(
|
|
band: List[Dict[str, Any]],
|
|
down_keys: set[CellKey],
|
|
by_pos: Dict[CellKey, Dict[str, Any]],
|
|
by_sheet: Dict[str, List[Dict[str, Any]]],
|
|
) -> bool:
|
|
for c in band:
|
|
left, top = resolve_parents(c, by_pos, by_sheet)
|
|
if left and cell_key(left) in down_keys:
|
|
return True
|
|
if top and cell_key(top) in down_keys:
|
|
return True
|
|
return False
|
|
|
|
|
|
def _visit_right(
|
|
cell: Dict[str, Any],
|
|
rows: List[Dict[str, Any]],
|
|
col_cursor: int,
|
|
*,
|
|
sheets: Dict[str, Any],
|
|
sheet_id: str,
|
|
row: int,
|
|
right_cells: List[Dict[str, Any]],
|
|
by_pos: Dict[CellKey, Dict[str, Any]],
|
|
by_sheet: Dict[str, List[Dict[str, Any]]],
|
|
col_registry: Dict[CellKey, List[ExpandedSlot]],
|
|
bind_index: int = 0,
|
|
left_slot: Optional[ExpandedSlot] = None,
|
|
top_slot: Optional[ExpandedSlot] = None,
|
|
parent_scoped: bool = False,
|
|
) -> int:
|
|
custom = cell.get("custom") or {}
|
|
field = resolve_field_path(_field_name(cell), _dataset_name(cell))
|
|
col = _int_coord(cell.get("col"))
|
|
anchor_col = col
|
|
left_p, top_p = resolve_parents(cell, by_pos, by_sheet)
|
|
|
|
if not parent_scoped:
|
|
if left_slot is None and left_p:
|
|
left_slot = _slot_at_index(
|
|
col_registry.get(cell_key(left_p), []), bind_index
|
|
)
|
|
if top_slot is None and top_p:
|
|
top_slot = _slot_at_index(
|
|
col_registry.get(cell_key(top_p), []), bind_index
|
|
)
|
|
filtered = _rows_for_bind(
|
|
rows,
|
|
cell,
|
|
left_parent=left_p,
|
|
top_parent=top_p,
|
|
left_slot=left_slot,
|
|
top_slot=top_slot,
|
|
)
|
|
else:
|
|
filtered = [r if isinstance(r, dict) else {} for r in rows]
|
|
|
|
binds = build_bind_list(cell, filtered)
|
|
children = _children_of(cell, right_cells, by_pos, by_sheet)
|
|
child_cols = {_int_coord(ch.get("col")) for ch in children}
|
|
slots: List[ExpandedSlot] = []
|
|
end_col = col_cursor
|
|
|
|
for bi, bind in enumerate(binds):
|
|
block_start = end_col
|
|
if children:
|
|
child_end = block_start
|
|
for ch in children:
|
|
child_end = _visit_right(
|
|
ch,
|
|
bind.data_list,
|
|
block_start,
|
|
sheets=sheets,
|
|
sheet_id=sheet_id,
|
|
row=row,
|
|
right_cells=right_cells,
|
|
by_pos=by_pos,
|
|
by_sheet=by_sheet,
|
|
col_registry=col_registry,
|
|
bind_index=bi,
|
|
parent_scoped=True,
|
|
)
|
|
block_len = max(1, child_end - block_start)
|
|
else:
|
|
block_len = expand_span(bind, cell)
|
|
|
|
if col not in child_cols:
|
|
for i in range(block_len):
|
|
c = block_start + i
|
|
if _poly_list(cell):
|
|
val = (
|
|
_get_nested_value(bind.data_list[i], field)
|
|
if i < len(bind.data_list)
|
|
else ""
|
|
)
|
|
else:
|
|
val = bind.value
|
|
_set_cell_value(
|
|
sheets,
|
|
sheet_id,
|
|
row,
|
|
c,
|
|
val,
|
|
style_template_col=anchor_col,
|
|
)
|
|
|
|
slots.append(
|
|
ExpandedSlot(bind.value, bind.data_list, block_start, block_start + block_len)
|
|
)
|
|
end_col = block_start + block_len
|
|
|
|
key = cell_key(cell)
|
|
col_registry[key] = col_registry.get(key, []) + slots
|
|
|
|
if custom.get("fillEmptyRows"):
|
|
_apply_fill_empty(
|
|
sheets,
|
|
sheet_id,
|
|
custom,
|
|
direction="right",
|
|
anchor_row=row,
|
|
anchor_col=col,
|
|
data_len=end_col - col_cursor,
|
|
)
|
|
return end_col
|
|
|
|
|
|
def _is_right_root(
|
|
cell: Dict[str, Any],
|
|
right_keys: set[CellKey],
|
|
by_pos: Dict[CellKey, Dict[str, Any]],
|
|
by_sheet: Dict[str, List[Dict[str, Any]]],
|
|
) -> bool:
|
|
left, top = resolve_parents(cell, by_pos, by_sheet)
|
|
if left and cell_key(left) in right_keys:
|
|
return False
|
|
if top and cell_key(top) in right_keys:
|
|
return False
|
|
return True
|
|
|
|
|
|
def _expand_sheet_right(
|
|
sheets: Dict[str, Any],
|
|
sheet_id: str,
|
|
row: int,
|
|
right_cells: List[Dict[str, Any]],
|
|
datasets: Dict[str, List[Any]],
|
|
by_pos: Dict[CellKey, Dict[str, Any]],
|
|
by_sheet: Dict[str, List[Dict[str, Any]]],
|
|
) -> None:
|
|
right_keys = {cell_key(c) for c in right_cells}
|
|
col_registry: Dict[CellKey, List[ExpandedSlot]] = {}
|
|
roots = [c for c in right_cells if _is_right_root(c, right_keys, by_pos, by_sheet)]
|
|
if not roots:
|
|
roots = right_cells
|
|
|
|
cursor = min(_int_coord(c.get("col")) for c in roots)
|
|
for root in sorted(roots, key=lambda x: _int_coord(x.get("col"))):
|
|
ds = _dataset_name(root)
|
|
raw = datasets.get(ds) or []
|
|
rows_data = [r if isinstance(r, dict) else {} for r in raw]
|
|
cursor = max(
|
|
cursor,
|
|
_visit_right(
|
|
root,
|
|
rows_data,
|
|
cursor,
|
|
sheets=sheets,
|
|
sheet_id=sheet_id,
|
|
row=row,
|
|
right_cells=right_cells,
|
|
by_pos=by_pos,
|
|
by_sheet=by_sheet,
|
|
col_registry=col_registry,
|
|
),
|
|
)
|
|
|
|
|
|
def _legacy_right_band(
|
|
sheets: Dict[str, Any],
|
|
sheet_id: str,
|
|
row: int,
|
|
band: List[Dict[str, Any]],
|
|
datasets: Dict[str, List[Any]],
|
|
) -> None:
|
|
max_len = max((len(datasets.get(_dataset_name(c)) or []) for c in band), default=0)
|
|
for item in sorted(band, key=lambda x: _int_coord(x.get("col"))):
|
|
ds = _dataset_name(item)
|
|
field = resolve_field_path(_field_name(item), ds)
|
|
start_col = _int_coord(item.get("col"))
|
|
data_rows = datasets.get(ds) or []
|
|
for i in range(max_len):
|
|
if i < len(data_rows) and isinstance(data_rows[i], dict):
|
|
val = _get_nested_value(data_rows[i], field)
|
|
else:
|
|
val = ""
|
|
_set_cell_value(
|
|
sheets,
|
|
sheet_id,
|
|
row,
|
|
start_col + i,
|
|
val,
|
|
style_template_col=start_col,
|
|
)
|
|
|
|
|
|
def apply_data_source_cells(
|
|
snapshot: Dict[str, Any],
|
|
cells_meta: Dict[str, Any],
|
|
datasets: Dict[str, List[Any]],
|
|
) -> Tuple[Dict[str, Any], List[Tuple[str, int, int, int]]]:
|
|
result = copy.deepcopy(snapshot)
|
|
sheets = result.get("sheets") or {}
|
|
pending_group_merges: List[Tuple[str, int, int, int]] = []
|
|
cell_list = cells_meta.get("cells") or []
|
|
by_pos, by_sheet = _index_data_sources(cell_list)
|
|
|
|
down_by_sheet: Dict[str, List[Dict[str, Any]]] = {}
|
|
right_by_sheet_row: Dict[Tuple[str, int], List[Dict[str, Any]]] = {}
|
|
singles: List[dict] = []
|
|
|
|
for cell in cell_list:
|
|
if cell.get("type") != "dataSource":
|
|
continue
|
|
custom = cell.get("custom") or {}
|
|
if not _dataset_name(cell) or not _field_name(cell):
|
|
continue
|
|
direction = _expand_direction(custom)
|
|
sheet_id = str(cell.get("sheet", "sheet1"))
|
|
|
|
if direction == "down":
|
|
down_by_sheet.setdefault(sheet_id, []).append(cell)
|
|
elif direction == "right":
|
|
row = _int_coord(cell.get("row"), 0)
|
|
right_by_sheet_row.setdefault((sheet_id, row), []).append(cell)
|
|
else:
|
|
singles.append(cell)
|
|
|
|
band_lengths = _compute_down_band_lengths(down_by_sheet, datasets)
|
|
for (sheet_id, start_row), max_len in sorted(
|
|
band_lengths.items(), key=lambda x: x[0][1], reverse=True
|
|
):
|
|
delta = max(0, max_len - 1)
|
|
if delta > 0:
|
|
_shift_sheet_rows_down(sheets, sheet_id, start_row + 1, delta)
|
|
|
|
for sheet_id, down_cells in down_by_sheet.items():
|
|
down_keys = {cell_key(c) for c in down_cells}
|
|
bands: Dict[int, List[Dict[str, Any]]] = {}
|
|
for c in down_cells:
|
|
bands.setdefault(_int_coord(c.get("row")), []).append(c)
|
|
uses_tree = False
|
|
for band in bands.values():
|
|
if _has_down_parent_link(band, down_keys, by_pos, by_sheet):
|
|
uses_tree = True
|
|
break
|
|
if uses_tree:
|
|
_expand_sheet_down(
|
|
sheets,
|
|
sheet_id,
|
|
down_cells,
|
|
datasets,
|
|
by_pos,
|
|
by_sheet,
|
|
pending_group_merges,
|
|
)
|
|
else:
|
|
for start_row, band in bands.items():
|
|
_legacy_down_band(
|
|
sheets,
|
|
sheet_id,
|
|
start_row,
|
|
band,
|
|
datasets,
|
|
pending_group_merges,
|
|
)
|
|
|
|
for (sheet_id, row), band in right_by_sheet_row.items():
|
|
right_keys = {cell_key(c) for c in band}
|
|
if _has_down_parent_link(band, right_keys, by_pos, by_sheet):
|
|
_expand_sheet_right(
|
|
sheets, sheet_id, row, band, datasets, by_pos, by_sheet
|
|
)
|
|
else:
|
|
_legacy_right_band(sheets, sheet_id, row, band, datasets)
|
|
|
|
for cell in singles:
|
|
sheet_id = str(cell.get("sheet", "sheet1"))
|
|
template_row = _int_coord(cell.get("row"), 0)
|
|
col = _int_coord(cell.get("col"), 0)
|
|
row = (
|
|
_summary_output_row(sheet_id, template_row, band_lengths)
|
|
if _poly_summary(cell)
|
|
else template_row
|
|
)
|
|
raw = datasets.get(_dataset_name(cell)) or []
|
|
rows = [r if isinstance(r, dict) else {} for r in raw]
|
|
if _poly_summary(cell):
|
|
filtered = _summary_filter_rows(cell, rows, by_pos, by_sheet)
|
|
else:
|
|
left, top = resolve_parents(cell, by_pos, by_sheet)
|
|
left_bind = top_bind = None
|
|
if left:
|
|
lb = build_bind_list(left, rows)
|
|
left_bind = lb[0].data_list if lb else None
|
|
if top:
|
|
tb = build_bind_list(top, rows)
|
|
top_bind = tb[0].data_list if tb else None
|
|
filtered = filter_rows_by_parents(
|
|
rows,
|
|
cell,
|
|
left_parent=left,
|
|
top_parent=top,
|
|
left_bind=left_bind,
|
|
top_bind=top_bind,
|
|
)
|
|
binds = build_bind_list(cell, filtered)
|
|
val = binds[0].value if binds else ""
|
|
_set_cell_value(sheets, sheet_id, row, col, val)
|
|
|
|
result["sheets"] = sheets
|
|
return result, pending_group_merges
|