Files
ai-agent-admin/backend-fastapi/online_dev/report_manager/engine/column_layout.py
T

316 lines
10 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.
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
分栏布局(MVP
- 行分栏 colcolumnType=1 超过 maxCol 行分列 / columnType=2 分 N 栏
- 列分栏 rowcolumnType=1 超过 maxRow 列分行 / columnType=2 分 N 行
- fillEmptyRows:每栏数据不足时补空行/列
"""
from __future__ import annotations
import copy
import math
import re
from typing import Any, Dict, List, Optional, Set, Tuple
def _col_letter_to_index(col: str) -> int:
col = col.upper()
n = 0
for ch in col:
n = n * 26 + (ord(ch) - ord("A") + 1)
return n - 1
def parse_cell_range(addr: str) -> Optional[Tuple[int, int, int, int]]:
"""A2:D10 -> (r0, r1, c0, c1) 0-based 闭区间"""
if not addr or not isinstance(addr, str):
return None
m = re.match(r"^([A-Za-z]+)(\d+):([A-Za-z]+)(\d+)$", addr.strip())
if not m:
return None
c0 = _col_letter_to_index(m.group(1))
r0 = int(m.group(2)) - 1
c1 = _col_letter_to_index(m.group(3))
r1 = int(m.group(4)) - 1
if r0 > r1:
r0, r1 = r1, r0
if c0 > c1:
c0, c1 = c1, c0
return r0, r1, c0, c1
def parse_index_list(spec: Optional[str]) -> Set[int]:
"""1,2-3,6 -> 0-based 索引集合"""
result: Set[int] = set()
if not spec:
return result
for part in str(spec).split(","):
part = part.strip()
if not part:
continue
if "-" in part:
a, b = part.split("-", 1)
try:
start, end = int(a) - 1, int(b) - 1
for i in range(min(start, end), max(start, end) + 1):
result.add(i)
except ValueError:
pass
else:
try:
result.add(int(part) - 1)
except ValueError:
pass
return result
def _get_column_config_for_sheet(
layout_list: List[Any],
sheet_id: str,
) -> Optional[Dict[str, Any]]:
for item in layout_list or []:
if isinstance(item, dict) and str(item.get("sheet")) == str(sheet_id):
cfg = item.get("columnList")
return cfg if isinstance(cfg, dict) else None
return None
def _extract_region_cells(
cell_data: Dict[str, Any],
r0: int,
r1: int,
c0: int,
c1: int,
) -> Dict[Tuple[int, int], Dict[str, Any]]:
region: Dict[Tuple[int, int], Dict[str, Any]] = {}
for rk in range(r0, r1 + 1):
row = cell_data.get(str(rk))
if not isinstance(row, dict):
continue
for ck in range(c0, c1 + 1):
cell = row.get(str(ck))
if isinstance(cell, dict):
region[(rk, ck)] = copy.deepcopy(cell)
return region
def _write_cell(cell_data: Dict[str, Any], row: int, col: int, cell: Dict[str, Any]) -> None:
cell_data.setdefault(str(row), {})[str(col)] = cell
def _empty_cell() -> Dict[str, Any]:
return {"v": ""}
def _resolve_col_fence(
cfg: Dict[str, Any],
data_size: int,
) -> Optional[Tuple[int, int]]:
"""返回 (fence_num, fence_data_size)"""
if data_size < 1:
return None
column_type = str(cfg.get("columnType") or "2")
if column_type == "1":
per_fence = int(cfg.get("maxCol") or 0)
if per_fence < 1:
return None
return math.ceil(data_size / per_fence), per_fence
fence_num = int(cfg.get("rowCount") or 0)
if fence_num < 2:
return None
return fence_num, math.ceil(data_size / fence_num)
def _resolve_row_fence(
cfg: Dict[str, Any],
data_size: int,
) -> Optional[Tuple[int, int]]:
if data_size < 1:
return None
column_type = str(cfg.get("columnType") or "2")
if column_type == "1":
per_fence = int(cfg.get("maxRow") or 0)
if per_fence < 1:
return None
return math.ceil(data_size / per_fence), per_fence
fence_num = int(cfg.get("colCount") or 0)
if fence_num < 2:
return None
return fence_num, math.ceil(data_size / fence_num)
def _apply_col_split(sheet: Dict[str, Any], cfg: Dict[str, Any]) -> None:
if not cfg.get("columnState") or cfg.get("columnStyle") != "col":
return
bounds = parse_cell_range(cfg.get("columnData") or "")
if not bounds:
return
r0, r1, c0, c1 = bounds
width = c1 - c0 + 1
cell_data = sheet.setdefault("cellData", {})
region = _extract_region_cells(cell_data, r0, r1, c0, c1)
data_row_indices = sorted({r for (r, _) in region.keys()}) or list(range(r0, r1 + 1))
resolved = _resolve_col_fence(cfg, len(data_row_indices))
if not resolved:
return
fence_num, fence_data_size = resolved
fill_empty = bool(cfg.get("fillEmptyRows"))
copy_rows = parse_index_list(cfg.get("copyCol"))
for r in range(r0, r1 + 1):
row_obj = cell_data.get(str(r))
if not isinstance(row_obj, dict):
continue
for c in range(c0, c1 + 1):
if (r, c) in region and str(c) in row_obj:
del row_obj[str(c)]
for block in range(fence_num):
block_rows = data_row_indices[block * fence_data_size : (block + 1) * fence_data_size]
target_c_base = c0 + block * width
out_row = r0
for copy_r in sorted(copy_rows):
if copy_r < r0 or copy_r > r1:
continue
for dc in range(width):
src = region.get((copy_r, c0 + dc))
if src:
_write_cell(cell_data, out_row, target_c_base + dc, copy.deepcopy(src))
out_row += 1
header_offset = out_row - r0
data_written = 0
for local_i, src_r in enumerate(block_rows):
if src_r in copy_rows:
continue
dst_r = r0 + header_offset + local_i
data_written += 1
for dc in range(width):
src = region.get((src_r, c0 + dc))
if src:
_write_cell(cell_data, dst_r, target_c_base + dc, copy.deepcopy(src))
if fill_empty and data_written < fence_data_size:
for pad_i in range(data_written, fence_data_size):
dst_r = r0 + header_offset + pad_i
for dc in range(width):
col_idx = target_c_base + dc
row_obj = cell_data.get(str(dst_r)) or {}
if str(col_idx) not in row_obj:
_write_cell(cell_data, dst_r, col_idx, _empty_cell())
max_r, max_c = r1, c1
for rk, row in cell_data.items():
if not str(rk).isdigit() or not isinstance(row, dict):
continue
max_r = max(max_r, int(rk))
for ck in row.keys():
if str(ck).isdigit():
max_c = max(max_c, int(ck))
sheet["rowCount"] = max(int(sheet.get("rowCount") or 0), max_r + 5)
sheet["columnCount"] = max(int(sheet.get("columnCount") or 0), max_c + 5)
def _apply_row_split(sheet: Dict[str, Any], cfg: Dict[str, Any]) -> None:
"""列分栏:将区域内列拆成多块,纵向堆叠"""
if not cfg.get("columnState") or cfg.get("columnStyle") != "row":
return
bounds = parse_cell_range(cfg.get("columnData") or "")
if not bounds:
return
r0, r1, c0, c1 = bounds
height = r1 - r0 + 1
cell_data = sheet.setdefault("cellData", {})
region = _extract_region_cells(cell_data, r0, r1, c0, c1)
data_col_indices = sorted({c for (_, c) in region.keys()}) or list(range(c0, c1 + 1))
resolved = _resolve_row_fence(cfg, len(data_col_indices))
if not resolved:
return
fence_num, fence_data_size = resolved
fill_empty = bool(cfg.get("fillEmptyRows"))
copy_cols = parse_index_list(cfg.get("copyRow"))
for r in range(r0, r1 + 1):
row_obj = cell_data.get(str(r))
if not isinstance(row_obj, dict):
continue
for c in range(c0, c1 + 1):
if (r, c) in region and str(c) in row_obj:
del row_obj[str(c)]
for block in range(fence_num):
block_cols = data_col_indices[block * fence_data_size : (block + 1) * fence_data_size]
target_r_base = r0 + block * height
for copy_c in sorted(copy_cols):
if copy_c < c0 or copy_c > c1:
continue
for dr in range(height):
src_r = r0 + dr
src = region.get((src_r, copy_c))
if src:
_write_cell(cell_data, target_r_base + dr, copy_c, copy.deepcopy(src))
written_cols = [c for c in block_cols if c not in copy_cols]
for src_c in written_cols:
for dr in range(height):
src_r = r0 + dr
dst_r = target_r_base + dr
src = region.get((src_r, src_c))
if src:
_write_cell(cell_data, dst_r, src_c, copy.deepcopy(src))
if fill_empty and len(written_cols) < fence_data_size:
pad_need = fence_data_size - len(written_cols)
pad_candidates = [
c
for c in range(c0, c1 + 1)
if c not in copy_cols and c not in written_cols
]
for pad_c in pad_candidates[:pad_need]:
for dr in range(height):
dst_r = target_r_base + dr
row_obj = cell_data.get(str(dst_r)) or {}
if str(pad_c) not in row_obj:
_write_cell(cell_data, dst_r, pad_c, _empty_cell())
max_r = r0 + fence_num * height
max_c = c1
for rk, row in cell_data.items():
if not str(rk).isdigit() or not isinstance(row, dict):
continue
max_r = max(max_r, int(rk))
for ck in row.keys():
if str(ck).isdigit():
max_c = max(max_c, int(ck))
sheet["rowCount"] = max(int(sheet.get("rowCount") or 0), max_r + 5)
sheet["columnCount"] = max(int(sheet.get("columnCount") or 0), max_c + 5)
def apply_column_layout(
snapshot: Dict[str, Any],
layout_list: List[Any],
) -> Dict[str, Any]:
if not snapshot or not layout_list:
return snapshot or {}
result = copy.deepcopy(snapshot)
sheets = result.get("sheets") or {}
for sheet_id in result.get("sheetOrder") or list(sheets.keys()):
sheet = sheets.get(sheet_id)
if not isinstance(sheet, dict):
continue
cfg = _get_column_config_for_sheet(layout_list, sheet_id)
if not cfg or not cfg.get("columnState"):
continue
style = cfg.get("columnStyle")
if style == "col":
_apply_col_split(sheet, cfg)
elif style == "row":
_apply_row_split(sheet, cfg)
result["sheets"] = sheets
return result