1525 lines
57 KiB
Python
1525 lines
57 KiB
Python
#!/usr/bin/env python3
|
|
# Copyright (c) 2026 Lark Technologies Pte. Ltd.
|
|
# SPDX-License-Identifier: MIT
|
|
"""Check Lark Sheet chart quality, placement, and numeric source-data issues.
|
|
|
|
The single required argument is a spreadsheet URL or spreadsheet token. By
|
|
default every worksheet is checked; pass --worksheet-id to restrict the check
|
|
to one worksheet reference_id.
|
|
|
|
Numeric source checks sample at most 50 data points per series and request at
|
|
most 2000 source cells per chart across the style and typed-value reads,
|
|
including headers and gaps between series.
|
|
Sampled zero/constant values do not establish a whole-series issue.
|
|
|
|
Exit codes:
|
|
0: check completed and no issue was found
|
|
1: the check could not be completed (CLI/read/response error)
|
|
2: check completed and at least one chart-quality issue was found
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
from typing import Any
|
|
|
|
from lark_sheet_read_cli import (
|
|
LarkCliError,
|
|
emit_error,
|
|
envelope_data,
|
|
resolve_target_sheets,
|
|
run_sheets,
|
|
sheet_identifier,
|
|
sheet_title,
|
|
)
|
|
from lark_chart_size_rules import MAX_ASPECT_RATIO, MAX_CHART_WIDTH, minimum_chart_size
|
|
|
|
ACTION = "chart_quality_check"
|
|
DEFAULT_COLUMN_WIDTH = 105.0
|
|
DEFAULT_ROW_HEIGHT = 27.0
|
|
MAX_CELL_READ_SIZE = 2_000
|
|
MAX_SOURCE_SAMPLE_POINTS = 50
|
|
|
|
|
|
CellBounds = tuple[int, int, int, int]
|
|
CellCache = dict[tuple[str, str, str, bool], dict[str, Any]]
|
|
SeriesProfile = dict[str, Any]
|
|
|
|
|
|
def _parse_a1_bounds(cell_range: str) -> CellBounds:
|
|
value = str(cell_range).rsplit("!", 1)[-1].replace("$", "")
|
|
match = re.fullmatch(r"([A-Za-z]+)(\d+)(?::([A-Za-z]+)(\d+))?", value)
|
|
if not match:
|
|
raise ValueError(f"Invalid A1 range: {cell_range!r}")
|
|
start_column = column_to_index(match.group(1))
|
|
start_row = int(match.group(2))
|
|
end_column = column_to_index(match.group(3) or match.group(1))
|
|
end_row = int(match.group(4) or match.group(2))
|
|
if end_row < start_row or end_column < start_column:
|
|
raise ValueError(f"Invalid A1 range: {cell_range!r}")
|
|
return start_row, end_row, start_column, end_column
|
|
|
|
|
|
def _format_a1_bounds(bounds: CellBounds) -> str:
|
|
start_row, end_row, start_column, end_column = bounds
|
|
return (
|
|
f"{index_to_column(start_column)}{start_row}:"
|
|
f"{index_to_column(end_column)}{end_row}"
|
|
)
|
|
|
|
|
|
def _bounds_area(bounds: CellBounds) -> int:
|
|
start_row, end_row, start_column, end_column = bounds
|
|
return (end_row - start_row + 1) * (end_column - start_column + 1)
|
|
|
|
|
|
def _merge_bounds(first: CellBounds, second: CellBounds) -> CellBounds:
|
|
return (
|
|
min(first[0], second[0]),
|
|
max(first[1], second[1]),
|
|
min(first[2], second[2]),
|
|
max(first[3], second[3]),
|
|
)
|
|
|
|
|
|
def _cluster_cell_reads(
|
|
items: list[tuple[dict[str, Any], CellBounds]],
|
|
) -> list[dict[str, Any]]:
|
|
clusters: list[dict[str, Any]] = []
|
|
for rectangle, bounds in sorted(items, key=lambda item: (item[1][0], item[1][2])):
|
|
if clusters:
|
|
merged = _merge_bounds(clusters[-1]["bounds"], bounds)
|
|
if _bounds_area(merged) <= MAX_CELL_READ_SIZE:
|
|
clusters[-1]["bounds"] = merged
|
|
clusters[-1]["members"].append((rectangle, bounds))
|
|
continue
|
|
clusters.append({"bounds": bounds, "members": [(rectangle, bounds)]})
|
|
return clusters
|
|
|
|
|
|
def column_to_index(column: str) -> int:
|
|
value = 0
|
|
text = str(column).strip().upper()
|
|
if not text or not text.isalpha():
|
|
raise ValueError(f"Invalid column: {column!r}")
|
|
for char in text:
|
|
value = value * 26 + ord(char) - ord("A") + 1
|
|
return value - 1
|
|
|
|
|
|
def index_to_column(index: int) -> str:
|
|
if index < 0:
|
|
raise ValueError(f"Invalid column index: {index}")
|
|
chars: list[str] = []
|
|
value = index + 1
|
|
while value:
|
|
value, remainder = divmod(value - 1, 26)
|
|
chars.append(chr(ord("A") + remainder))
|
|
return "".join(reversed(chars))
|
|
|
|
|
|
def _span_bounds(span: str, *, columns: bool) -> tuple[int, int]:
|
|
start, separator, end = str(span).partition(":")
|
|
end = end if separator else start
|
|
if columns:
|
|
return column_to_index(start), column_to_index(end)
|
|
return int(start) - 1, int(end) - 1
|
|
|
|
|
|
def _size_edges(
|
|
groups: Any,
|
|
*,
|
|
count: int,
|
|
span_key: str,
|
|
size_key: str,
|
|
columns: bool,
|
|
default_size: float,
|
|
) -> tuple[list[float], bool]:
|
|
sizes: list[float | None] = [None] * count
|
|
if isinstance(groups, list):
|
|
for group in groups:
|
|
if not isinstance(group, dict) or group.get(span_key) is None:
|
|
continue
|
|
start, end = _span_bounds(str(group[span_key]), columns=columns)
|
|
size = float(group.get(size_key, default_size))
|
|
for index in range(max(0, start), min(count - 1, end) + 1):
|
|
sizes[index] = max(0.0, size)
|
|
|
|
used_default = any(size is None for size in sizes)
|
|
resolved = [default_size if size is None else size for size in sizes]
|
|
edges = [0.0]
|
|
for size in resolved:
|
|
edges.append(edges[-1] + size)
|
|
return edges, used_default
|
|
|
|
|
|
def build_layout(
|
|
structure: dict[str, Any], row_count: int, column_count: int
|
|
) -> tuple[list[float], list[float], list[str]]:
|
|
row_groups = structure.get("row_heights")
|
|
column_groups = structure.get("col_widths", structure.get("column_widths"))
|
|
row_edges, row_defaulted = _size_edges(
|
|
row_groups,
|
|
count=row_count,
|
|
span_key="rows",
|
|
size_key="height",
|
|
columns=False,
|
|
default_size=DEFAULT_ROW_HEIGHT,
|
|
)
|
|
column_edges, column_defaulted = _size_edges(
|
|
column_groups,
|
|
count=column_count,
|
|
span_key="cols",
|
|
size_key="width",
|
|
columns=True,
|
|
default_size=DEFAULT_COLUMN_WIDTH,
|
|
)
|
|
warnings: list[str] = []
|
|
if row_defaulted:
|
|
warnings.append("部分行缺少高度信息，按 27 px 估算")
|
|
if column_defaulted:
|
|
warnings.append("部分列缺少宽度信息，按 105 px 估算")
|
|
return row_edges, column_edges, warnings
|
|
|
|
|
|
def _first_dict(value: Any) -> dict[str, Any] | None:
|
|
if isinstance(value, dict):
|
|
return value
|
|
if isinstance(value, list):
|
|
return next((item for item in value if isinstance(item, dict)), None)
|
|
return None
|
|
|
|
|
|
def extract_sheet_structure(data: dict[str, Any]) -> dict[str, Any]:
|
|
sheet = _first_dict(data.get("sheets")) or _first_dict(data.get("sheet"))
|
|
return sheet or data
|
|
|
|
|
|
def extract_charts(data: dict[str, Any], sheet_id: str, title: str) -> list[dict[str, Any]]:
|
|
sheets = data.get("sheets")
|
|
if isinstance(sheets, list):
|
|
for sheet in sheets:
|
|
if not isinstance(sheet, dict):
|
|
continue
|
|
if sheet_identifier(sheet) == sheet_id or sheet_title(sheet) == title:
|
|
charts = sheet.get("charts")
|
|
return [chart for chart in charts if isinstance(chart, dict)] if isinstance(charts, list) else []
|
|
charts = data.get("charts")
|
|
return [chart for chart in charts if isinstance(chart, dict)] if isinstance(charts, list) else []
|
|
|
|
|
|
def chart_rectangle(
|
|
chart: dict[str, Any], row_edges: list[float], column_edges: list[float]
|
|
) -> dict[str, Any]:
|
|
details = chart.get("details") if isinstance(chart.get("details"), dict) else chart
|
|
position = details.get("position") if isinstance(details.get("position"), dict) else {}
|
|
offset = details.get("offset") if isinstance(details.get("offset"), dict) else {}
|
|
size = details.get("size") if isinstance(details.get("size"), dict) else {}
|
|
|
|
row = int(position["row"])
|
|
column = column_to_index(str(position["col"]))
|
|
if row < 0 or column < 0 or row >= len(row_edges) - 1 or column >= len(column_edges) - 1:
|
|
raise ValueError(f"anchor outside sheet: {position!r}")
|
|
|
|
width = float(size["width"])
|
|
height = float(size["height"])
|
|
if width <= 0 or height <= 0:
|
|
raise ValueError(f"invalid chart size: {size!r}")
|
|
|
|
left = column_edges[column] + float(offset.get("col_offset", 0) or 0)
|
|
top = row_edges[row] + float(offset.get("row_offset", 0) or 0)
|
|
return {
|
|
"chart_id": str(chart.get("chart_id") or chart.get("id") or ""),
|
|
"anchor_cell": f"{index_to_column(column)}{row + 1}",
|
|
"left": left,
|
|
"top": top,
|
|
"right": left + width,
|
|
"bottom": top + height,
|
|
"width": width,
|
|
"height": height,
|
|
}
|
|
|
|
|
|
def intersection(first: dict[str, Any], second: dict[str, Any]) -> dict[str, float] | None:
|
|
left = max(float(first["left"]), float(second["left"]))
|
|
top = max(float(first["top"]), float(second["top"]))
|
|
right = min(float(first["right"]), float(second["right"]))
|
|
bottom = min(float(first["bottom"]), float(second["bottom"]))
|
|
if right <= left or bottom <= top:
|
|
return None
|
|
return {
|
|
"width": round(right - left, 2),
|
|
"height": round(bottom - top, 2),
|
|
"area": round((right - left) * (bottom - top), 2),
|
|
}
|
|
|
|
|
|
def chart_context(rectangle: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"chart_id": rectangle["chart_id"],
|
|
"anchor_cell": rectangle["anchor_cell"],
|
|
"rectangle_px": {
|
|
"left": round(rectangle["left"], 2),
|
|
"top": round(rectangle["top"], 2),
|
|
"right": round(rectangle["right"], 2),
|
|
"bottom": round(rectangle["bottom"], 2),
|
|
"width": round(rectangle["width"], 2),
|
|
"height": round(rectangle["height"], 2),
|
|
},
|
|
}
|
|
|
|
|
|
def _covered_indexes(edges: list[float], start: float, end: float) -> list[int]:
|
|
return [
|
|
index
|
|
for index in range(len(edges) - 1)
|
|
if edges[index + 1] > start and edges[index] < end
|
|
]
|
|
|
|
|
|
def rectangle_cell_range(
|
|
rectangle: dict[str, Any], row_edges: list[float], column_edges: list[float]
|
|
) -> str | None:
|
|
rows = _covered_indexes(row_edges, max(0.0, rectangle["top"]), rectangle["bottom"])
|
|
columns = _covered_indexes(column_edges, max(0.0, rectangle["left"]), rectangle["right"])
|
|
if not rows or not columns:
|
|
return None
|
|
return f"{index_to_column(columns[0])}{rows[0] + 1}:{index_to_column(columns[-1])}{rows[-1] + 1}"
|
|
|
|
|
|
def _has_content(cell: Any) -> bool:
|
|
if not isinstance(cell, dict):
|
|
return False
|
|
for key in ("value", "formula", "note"):
|
|
value = cell.get(key)
|
|
if value not in (None, ""):
|
|
return True
|
|
return bool(cell.get("rich_text") or cell.get("multiple_values"))
|
|
|
|
|
|
def _iter_cells(data: dict[str, Any]):
|
|
ranges = data.get("ranges")
|
|
if not isinstance(ranges, list):
|
|
return
|
|
for result_range in ranges:
|
|
if not isinstance(result_range, dict):
|
|
continue
|
|
cells = result_range.get("cells")
|
|
rows = result_range.get("row_indices")
|
|
columns = result_range.get("col_indices")
|
|
if not isinstance(cells, list):
|
|
continue
|
|
for row_offset, row in enumerate(cells):
|
|
if not isinstance(row, list):
|
|
continue
|
|
row_number = int(rows[row_offset]) if isinstance(rows, list) and row_offset < len(rows) else row_offset + 1
|
|
for column_offset, cell in enumerate(row):
|
|
column = columns[column_offset] if isinstance(columns, list) and column_offset < len(columns) else index_to_column(column_offset)
|
|
yield row_number, column_to_index(str(column)), cell
|
|
|
|
|
|
def non_empty_cells(
|
|
data: dict[str, Any], sample_limit: int, bounds: CellBounds | None = None
|
|
) -> tuple[int, list[str], bool]:
|
|
count = 0
|
|
samples: list[str] = []
|
|
truncated = bool(data.get("has_more"))
|
|
ranges = data.get("ranges")
|
|
if not isinstance(ranges, list):
|
|
return 0, [], truncated
|
|
for result_range in ranges:
|
|
if not isinstance(result_range, dict):
|
|
continue
|
|
truncated = truncated or bool(result_range.get("truncated"))
|
|
for row_number, column_index, cell in _iter_cells(data):
|
|
if bounds and not (
|
|
bounds[0] <= row_number <= bounds[1]
|
|
and bounds[2] <= column_index <= bounds[3]
|
|
):
|
|
continue
|
|
if not _has_content(cell):
|
|
continue
|
|
count += 1
|
|
if len(samples) < sample_limit:
|
|
samples.append(f"{index_to_column(column_index)}{row_number}")
|
|
return count, samples, truncated
|
|
|
|
|
|
def _read_cells(
|
|
cache: CellCache,
|
|
locator: dict[str, str],
|
|
*,
|
|
sheet_id: str | None,
|
|
sheet_name: str | None,
|
|
cell_range: str,
|
|
include: str,
|
|
skip_hidden: bool = False,
|
|
timeout: int,
|
|
) -> dict[str, Any]:
|
|
selector = f"id:{sheet_id}" if sheet_id else f"name:{sheet_name}"
|
|
key = (selector, cell_range, include, skip_hidden)
|
|
if key not in cache:
|
|
flags: dict[str, Any] = {"range": cell_range, "include": include}
|
|
if skip_hidden:
|
|
flags["skip-hidden"] = True
|
|
cache[key] = envelope_data(
|
|
run_sheets(
|
|
"+cells-get",
|
|
**locator,
|
|
**({"sheet_id": sheet_id} if sheet_id else {"sheet_name": sheet_name}),
|
|
flags=flags,
|
|
timeout=timeout,
|
|
)
|
|
)
|
|
return cache[key]
|
|
|
|
|
|
def _read_typed_table(
|
|
cache: CellCache,
|
|
locator: dict[str, str],
|
|
*,
|
|
sheet_id: str | None,
|
|
sheet_name: str | None,
|
|
cell_range: str,
|
|
timeout: int,
|
|
) -> tuple[list[list[Any]], list[str], bool]:
|
|
selector = f"id:{sheet_id}" if sheet_id else f"name:{sheet_name}"
|
|
key = (selector, cell_range, "typed_table", False)
|
|
if key not in cache:
|
|
cache[key] = envelope_data(
|
|
run_sheets(
|
|
"+table-get",
|
|
**locator,
|
|
**({"sheet_id": sheet_id} if sheet_id else {"sheet_name": sheet_name}),
|
|
flags={"range": cell_range, "no-header": True},
|
|
timeout=timeout,
|
|
)
|
|
)
|
|
data = cache[key]
|
|
sheets = data.get("sheets")
|
|
if not isinstance(sheets, list) or len(sheets) != 1 or not isinstance(sheets[0], dict):
|
|
return [], [], True
|
|
table = sheets[0]
|
|
rows = table.get("data")
|
|
columns = table.get("columns")
|
|
dtypes = table.get("dtypes")
|
|
if not isinstance(rows, list) or not isinstance(columns, list) or not isinstance(dtypes, dict):
|
|
return [], [], True
|
|
def value_kind(dtype: Any) -> str:
|
|
value = str(dtype or "").lower()
|
|
if value in {"number", "int64", "float64"}:
|
|
return "number"
|
|
if value in {"bool", "boolean"}:
|
|
return "bool"
|
|
if value.startswith("datetime"):
|
|
return "date"
|
|
return "string"
|
|
|
|
types = [value_kind(dtypes.get(str(column))) for column in columns]
|
|
truncated = bool(data.get("truncated")) or bool(table.get("truncated"))
|
|
return rows, types, truncated
|
|
|
|
|
|
def _typed_cell(
|
|
rows: list[list[Any]],
|
|
column_types: list[str],
|
|
row_offset: int,
|
|
column_offset: int,
|
|
visible_offsets: set[tuple[int, int]],
|
|
) -> tuple[Any, str]:
|
|
if (
|
|
row_offset < 0
|
|
or row_offset >= len(rows)
|
|
or not isinstance(rows[row_offset], list)
|
|
or column_offset < 0
|
|
or column_offset >= len(rows[row_offset])
|
|
):
|
|
return None, ""
|
|
value = rows[row_offset][column_offset]
|
|
if isinstance(value, bool):
|
|
return value, "bool"
|
|
if isinstance(value, (int, float)):
|
|
return value, "number"
|
|
kind = column_types[column_offset] if column_offset < len(column_types) else ""
|
|
if kind != "string" or not isinstance(value, str) or not _looks_numeric(value):
|
|
return value, kind
|
|
|
|
# table-get infers one dtype per physical column. A hidden cell or another
|
|
# row-series can widen that dtype and stringify otherwise numeric cells.
|
|
# Treat such cells as unknown instead of reporting a false storage issue.
|
|
numeric_string_count = 0
|
|
for other_row_offset, row in enumerate(rows):
|
|
if not isinstance(row, list) or column_offset >= len(row):
|
|
continue
|
|
other = row[column_offset]
|
|
if other not in (None, "") and (other_row_offset, column_offset) not in visible_offsets:
|
|
return value, "unknown"
|
|
if other in (None, "") or (
|
|
isinstance(other, (int, float)) and not isinstance(other, bool)
|
|
):
|
|
continue
|
|
if isinstance(other, str) and _looks_numeric(other):
|
|
numeric_string_count += 1
|
|
continue
|
|
return value, "unknown"
|
|
return value, "string" if numeric_string_count == 1 else "ambiguous_string"
|
|
|
|
|
|
def _chart_snapshot(chart: dict[str, Any]) -> dict[str, Any]:
|
|
details = chart.get("details") if isinstance(chart.get("details"), dict) else chart
|
|
snapshot = details.get("snapshot")
|
|
return snapshot if isinstance(snapshot, dict) else {}
|
|
|
|
|
|
def _chart_type(snapshot: dict[str, Any]) -> str:
|
|
plot_area = snapshot.get("plotArea")
|
|
plot = plot_area.get("plot") if isinstance(plot_area, dict) else None
|
|
return str(plot.get("type") or "").lower() if isinstance(plot, dict) else ""
|
|
|
|
|
|
def _plot(snapshot: dict[str, Any]) -> dict[str, Any]:
|
|
plot_area = snapshot.get("plotArea")
|
|
plot = plot_area.get("plot") if isinstance(plot_area, dict) else None
|
|
return plot if isinstance(plot, dict) else {}
|
|
|
|
|
|
def _static_series_profiles(snapshot: dict[str, Any]) -> list[SeriesProfile]:
|
|
data = snapshot.get("data")
|
|
dim2 = data.get("dim2") if isinstance(data, dict) else None
|
|
fields = dim2.get("fields") if isinstance(dim2, dict) else None
|
|
if not isinstance(fields, list):
|
|
return []
|
|
profiles: list[SeriesProfile] = []
|
|
for offset, field in enumerate(fields, start=1):
|
|
if not isinstance(field, dict):
|
|
continue
|
|
values = []
|
|
for value in field.get("parsedValues") or []:
|
|
numeric = (
|
|
float(value)
|
|
if isinstance(value, (int, float)) and not isinstance(value, bool)
|
|
else _numeric_text_value(value) if isinstance(value, str) else None
|
|
)
|
|
if numeric is not None:
|
|
values.append(numeric)
|
|
profiles.append(
|
|
{
|
|
"dimension_index": offset,
|
|
"series_name": str(field.get("name") or f"Series {offset}"),
|
|
"point_count": len(field.get("parsedValues") or []),
|
|
"numeric_value_count": len(values),
|
|
"unique_numeric_values": list(dict.fromkeys(values))[:2],
|
|
"source_sheet": "",
|
|
"source_range": "",
|
|
"series_range": "",
|
|
}
|
|
)
|
|
return profiles
|
|
|
|
|
|
def _labeled_series_indexes(
|
|
snapshot: dict[str, Any], profiles: list[SeriesProfile]
|
|
) -> set[int]:
|
|
plot = _plot(snapshot)
|
|
available = {
|
|
int(profile["dimension_index"])
|
|
for profile in profiles
|
|
if profile.get("dimension_index") is not None
|
|
}
|
|
labeled = set(available) if isinstance(plot.get("labels"), dict) else set()
|
|
series = plot.get("series")
|
|
if isinstance(series, list):
|
|
labeled.update(
|
|
int(item["index"])
|
|
for item in series
|
|
if isinstance(item, dict)
|
|
and item.get("index") is not None
|
|
and isinstance(item.get("labels"), dict)
|
|
)
|
|
return labeled & available
|
|
|
|
|
|
def _constant_labeled_series(
|
|
chart: dict[str, Any], profiles: list[SeriesProfile]
|
|
) -> list[dict[str, Any]]:
|
|
chart_id = str(chart.get("chart_id") or chart.get("id") or "")
|
|
snapshot = _chart_snapshot(chart)
|
|
labeled = _labeled_series_indexes(snapshot, profiles)
|
|
return [
|
|
{
|
|
"chart_id": chart_id,
|
|
"dimension_index": profile["dimension_index"],
|
|
"series_name": profile["series_name"],
|
|
"source_sheet": profile["source_sheet"],
|
|
"source_range": profile["source_range"],
|
|
"series_range": profile["series_range"],
|
|
"reason": "constant_labeled_series",
|
|
"data_point_count": profile["point_count"],
|
|
"constant_value": profile["unique_numeric_values"][0],
|
|
"suggested_fix": "remove_series_labels_or_use_one_sparse_marker",
|
|
}
|
|
for profile in profiles
|
|
if int(profile.get("dimension_index", -1)) in labeled
|
|
and profile.get("constant_check_unverifiable") is not True
|
|
and int(profile.get("numeric_value_count", 0)) >= 2
|
|
and len(profile.get("unique_numeric_values") or []) == 1
|
|
]
|
|
|
|
|
|
def _unbound_secondary_axis(chart: dict[str, Any]) -> dict[str, Any] | None:
|
|
snapshot = _chart_snapshot(chart)
|
|
plot_area = snapshot.get("plotArea")
|
|
if not isinstance(plot_area, dict):
|
|
return None
|
|
plot = plot_area.get("plot")
|
|
if not isinstance(plot, dict) or str(plot.get("type") or "").lower() != "combo":
|
|
return None
|
|
|
|
axes = plot_area.get("axes")
|
|
has_right_axis = isinstance(axes, list) and any(
|
|
isinstance(axis, dict)
|
|
and str(axis.get("type") or "").lower() == "y"
|
|
and str(axis.get("position") or axis.get("axisPosition") or "").lower() == "right"
|
|
for axis in axes
|
|
)
|
|
series = plot.get("series")
|
|
configured_series = (
|
|
[item for item in series if isinstance(item, dict)]
|
|
if isinstance(series, list)
|
|
else []
|
|
)
|
|
if not has_right_axis or not configured_series:
|
|
return None
|
|
if str(plot.get("yAxisPosition") or "").lower() == "right" or any(
|
|
str(item.get("yAxisPosition") or "").lower() == "right"
|
|
for item in configured_series
|
|
):
|
|
return None
|
|
|
|
return {
|
|
"chart_id": str(chart.get("chart_id") or chart.get("id") or ""),
|
|
"reason": "secondary_axis_has_no_bound_series",
|
|
"series_indexes": [
|
|
int(item["index"])
|
|
for item in configured_series
|
|
if isinstance(item.get("index"), (int, float))
|
|
],
|
|
"suggested_fix": "bind_the_intended_combo_series_to_the_right_axis",
|
|
}
|
|
|
|
|
|
def _undersized_chart(chart: dict[str, Any]) -> dict[str, Any] | None:
|
|
details = chart.get("details") if isinstance(chart.get("details"), dict) else chart
|
|
snapshot = _chart_snapshot(chart)
|
|
chart_type = _chart_type(snapshot)
|
|
size = details.get("size")
|
|
if not isinstance(size, dict):
|
|
return None
|
|
width = size.get("width")
|
|
height = size.get("height")
|
|
if (
|
|
not isinstance(width, (int, float))
|
|
or isinstance(width, bool)
|
|
or width <= 0
|
|
or not isinstance(height, (int, float))
|
|
or isinstance(height, bool)
|
|
or height <= 0
|
|
):
|
|
return None
|
|
actual = {
|
|
"width": float(width),
|
|
"height": float(height),
|
|
}
|
|
minimum = minimum_chart_size(chart_type)
|
|
if actual["width"] >= minimum["width"] and actual["height"] >= minimum["height"]:
|
|
return None
|
|
return {
|
|
"chart_id": str(chart.get("chart_id") or chart.get("id") or ""),
|
|
"reason": "chart_below_minimum_size",
|
|
"chart_type": chart_type,
|
|
"actual_size": actual,
|
|
"minimum_size": minimum,
|
|
"suggested_fix": "run_lark_chart_size_advisor_before_resizing",
|
|
}
|
|
|
|
|
|
def _overwide_chart(chart: dict[str, Any]) -> dict[str, Any] | None:
|
|
details = chart.get("details") if isinstance(chart.get("details"), dict) else chart
|
|
snapshot = _chart_snapshot(chart)
|
|
chart_type = _chart_type(snapshot)
|
|
size = details.get("size") if isinstance(details.get("size"), dict) else {}
|
|
width = size.get("width")
|
|
height = size.get("height")
|
|
if (
|
|
not isinstance(width, (int, float))
|
|
or isinstance(width, bool)
|
|
or width <= 0
|
|
or not isinstance(height, (int, float))
|
|
or isinstance(height, bool)
|
|
or height <= 0
|
|
):
|
|
return None
|
|
width = float(width)
|
|
height = float(height)
|
|
minimum = minimum_chart_size(chart_type)
|
|
aspect_ratio = width / height if height > 0 else 0
|
|
width_exceeded = width > MAX_CHART_WIDTH
|
|
aspect_ratio_exceeded = (
|
|
width >= minimum["width"]
|
|
and height >= minimum["height"]
|
|
and aspect_ratio > MAX_ASPECT_RATIO
|
|
)
|
|
if not width_exceeded and not aspect_ratio_exceeded:
|
|
return None
|
|
return {
|
|
"chart_id": str(chart.get("chart_id") or chart.get("id") or ""),
|
|
"reason": "chart_too_wide",
|
|
"chart_type": chart_type,
|
|
"actual_size": {"width": width, "height": height},
|
|
"actual_aspect_ratio": round(aspect_ratio, 2),
|
|
"maximum_width": MAX_CHART_WIDTH,
|
|
"maximum_aspect_ratio": MAX_ASPECT_RATIO,
|
|
"suggested_fix": "use_size_advisor_create_flags_or_change_chart_structure",
|
|
}
|
|
|
|
|
|
def _numeric_dimensions(snapshot: dict[str, Any]) -> list[tuple[int, str]]:
|
|
data = snapshot.get("data")
|
|
if not isinstance(data, dict):
|
|
return []
|
|
dim2 = data.get("dim2")
|
|
series = dim2.get("series") if isinstance(dim2, dict) else None
|
|
chart_type = _chart_type(snapshot)
|
|
dimensions: list[tuple[int, str]] = []
|
|
if isinstance(series, list):
|
|
for offset, serie in enumerate(series):
|
|
if not isinstance(serie, dict) or serie.get("index") is None:
|
|
continue
|
|
if str(serie.get("aggregateType") or "").lower() == "counta":
|
|
continue
|
|
role = str(serie.get("role") or "").lower()
|
|
if chart_type == "bubble":
|
|
role = role or ("x", "y", "group", "size")[min(offset, 3)]
|
|
if role not in {"x", "y", "size"}:
|
|
continue
|
|
dimensions.append((int(serie["index"]), role or "value"))
|
|
|
|
plot_area = snapshot.get("plotArea")
|
|
axes = plot_area.get("axes") if isinstance(plot_area, dict) else None
|
|
continuous_x = chart_type == "scatter"
|
|
if isinstance(axes, list):
|
|
for axis in axes:
|
|
if not isinstance(axis, dict) or str(axis.get("type") or "").lower() != "x":
|
|
continue
|
|
position = axis.get("position")
|
|
if (
|
|
(position is None or str(position).lower() in {"bottom", "x"})
|
|
and str(axis.get("valueType") or "").lower() == "linear"
|
|
):
|
|
continuous_x = True
|
|
break
|
|
dim1 = data.get("dim1")
|
|
serie = dim1.get("serie") if isinstance(dim1, dict) else None
|
|
if continuous_x and chart_type != "bubble" and isinstance(serie, dict) and serie.get("index") is not None:
|
|
dimensions.append((int(serie["index"]), "x"))
|
|
|
|
return list(dict.fromkeys(dimensions))
|
|
|
|
|
|
def _series_aggregate_type(data: dict[str, Any], dimension_index: int) -> str:
|
|
dim2 = data.get("dim2")
|
|
value_series = dim2.get("series") if isinstance(dim2, dict) else None
|
|
source_series = next(
|
|
(
|
|
item
|
|
for item in (value_series if isinstance(value_series, list) else [])
|
|
if isinstance(item, dict)
|
|
and item.get("index") is not None
|
|
and int(item["index"]) == dimension_index
|
|
),
|
|
{},
|
|
)
|
|
return str(source_series.get("aggregateType") or "sum").lower()
|
|
|
|
|
|
def _aggregation_can_change_constant(data: dict[str, Any], dimension_index: int) -> bool:
|
|
dim1 = data.get("dim1")
|
|
category_series = dim1.get("serie") if isinstance(dim1, dict) else None
|
|
if not isinstance(category_series, dict) or category_series.get("aggregate") is False:
|
|
return False
|
|
return _series_aggregate_type(data, dimension_index) in {"sum", "count", "counta"}
|
|
|
|
|
|
def _parse_chart_ref(value: str, default_sheet: str) -> tuple[str, str, CellBounds]:
|
|
raw = str(value).strip()
|
|
sheet_name = default_sheet
|
|
cell_range = raw
|
|
if "!" in raw:
|
|
sheet_name, cell_range = raw.rsplit("!", 1)
|
|
sheet_name = sheet_name.strip()
|
|
if len(sheet_name) >= 2 and sheet_name[0] == sheet_name[-1] == "'":
|
|
sheet_name = sheet_name[1:-1].replace("''", "'")
|
|
return sheet_name, cell_range.replace("$", ""), _parse_a1_bounds(cell_range)
|
|
|
|
|
|
def _looks_numeric(value: str) -> bool:
|
|
return _numeric_text_value(value) is not None
|
|
|
|
|
|
def _numeric_text_value(value: str) -> float | None:
|
|
text = value.strip()
|
|
if not text:
|
|
return None
|
|
text = re.sub(r"^([+-]?)[\$\u00a5\uffe5\u20ac\u00a3]", r"\1", text)
|
|
if text.endswith("%"):
|
|
text = text[:-1]
|
|
# Group separators must be consistent and separate groups of three digits.
|
|
if not re.fullmatch(
|
|
r"[+-]?(?:(?:\d+|\d{1,3}([, ])\d{3}(?:\1\d{3})*)(?:\.\d*)?|\.\d+)"
|
|
r"(?:[eE][+-]?\d+)?",
|
|
text,
|
|
):
|
|
return None
|
|
return float(text.replace(" ", "").replace(",", ""))
|
|
|
|
|
|
def _zero_state(value: Any) -> str:
|
|
if value in (None, ""):
|
|
return "empty"
|
|
if isinstance(value, (int, float)) and not isinstance(value, bool):
|
|
return "zero" if value == 0 else "nonzero"
|
|
if isinstance(value, str):
|
|
numeric_value = _numeric_text_value(value)
|
|
if numeric_value is not None:
|
|
return "zero" if numeric_value == 0 else "nonzero"
|
|
return "nonzero"
|
|
|
|
|
|
def _update_series_state(state: dict[str, Any], value: Any) -> None:
|
|
state[_zero_state(value)] = True
|
|
numeric = (
|
|
float(value)
|
|
if isinstance(value, (int, float)) and not isinstance(value, bool)
|
|
else _numeric_text_value(value) if isinstance(value, str) else None
|
|
)
|
|
if numeric is None:
|
|
return
|
|
state["numeric_value_count"] += 1
|
|
if len(state["unique_numeric_values"]) < 2:
|
|
state["unique_numeric_values"].add(numeric)
|
|
|
|
|
|
def _cells_truncated(data: dict[str, Any]) -> bool:
|
|
ranges = data.get("ranges")
|
|
return bool(data.get("has_more")) or any(
|
|
isinstance(item, dict) and item.get("truncated")
|
|
for item in (ranges if isinstance(ranges, list) else [])
|
|
)
|
|
|
|
|
|
def _numeric_source_issues(
|
|
chart: dict[str, Any],
|
|
*,
|
|
owner_sheet_id: str,
|
|
owner_sheet_name: str,
|
|
cache: CellCache,
|
|
locator: dict[str, str],
|
|
timeout: int,
|
|
sample_limit: int,
|
|
) -> tuple[
|
|
list[dict[str, Any]],
|
|
list[dict[str, Any]],
|
|
list[dict[str, str]],
|
|
list[SeriesProfile],
|
|
]:
|
|
chart_id = str(chart.get("chart_id") or chart.get("id") or "")
|
|
snapshot = _chart_snapshot(chart)
|
|
data = snapshot.get("data")
|
|
if not isinstance(data, dict):
|
|
return [], [], [{"chart_id": chart_id, "reason": "chart snapshot.data is missing"}], []
|
|
if data.get("isStaticData") is True:
|
|
return [], [], [], _static_series_profiles(snapshot)
|
|
dimensions = _numeric_dimensions(snapshot)
|
|
if not dimensions:
|
|
return [], [], [], []
|
|
refs = data.get("refs")
|
|
if not isinstance(refs, list) or not refs:
|
|
return [], [], [{"chart_id": chart_id, "reason": "chart data.refs is missing"}], []
|
|
|
|
parsed_refs: list[tuple[str, str, CellBounds]] = []
|
|
unverifiable: list[dict[str, str]] = []
|
|
for ref in refs:
|
|
raw_ref = ref.get("value") if isinstance(ref, dict) else ref
|
|
try:
|
|
parsed_refs.append(_parse_chart_ref(str(raw_ref), owner_sheet_name))
|
|
except (TypeError, ValueError) as exc:
|
|
unverifiable.append({"chart_id": chart_id, "reason": str(exc)})
|
|
return [], [], unverifiable, []
|
|
|
|
direction = str(data.get("direction") or "column").lower()
|
|
skip_hidden = data.get("includeHiddenOrFilter") is not True
|
|
mapped: list[tuple[int, str, int, str, str, CellBounds]] = []
|
|
for dimension_index, role in dimensions:
|
|
offset = 0
|
|
for source_sheet, source_range, bounds in parsed_refs:
|
|
dimension_count = (
|
|
bounds[3] - bounds[2] + 1 if direction == "column" else bounds[1] - bounds[0] + 1
|
|
)
|
|
if offset < dimension_index <= offset + dimension_count:
|
|
mapped.append(
|
|
(
|
|
dimension_index,
|
|
role,
|
|
dimension_index - offset,
|
|
source_sheet,
|
|
source_range,
|
|
bounds,
|
|
)
|
|
)
|
|
break
|
|
offset += dimension_count
|
|
else:
|
|
unverifiable.append(
|
|
{
|
|
"chart_id": chart_id,
|
|
"reason": f"numeric dimension index {dimension_index} is outside data.refs",
|
|
}
|
|
)
|
|
|
|
detached = str(data.get("headerMode") or "").lower() == "detached"
|
|
mapped_by_ref: dict[
|
|
tuple[str, str, CellBounds], list[tuple[int, str, int]]
|
|
] = {}
|
|
for dimension_index, role, local_index, source_sheet, source_range, bounds in mapped:
|
|
mapped_by_ref.setdefault((source_sheet, source_range, bounds), []).append(
|
|
(dimension_index, role, local_index)
|
|
)
|
|
|
|
issue_groups: dict[tuple[int, str, str, str, str, str], list[str]] = {}
|
|
issue_counts: dict[tuple[int, str, str, str, str, str], int] = {}
|
|
degenerate_series: list[dict[str, Any]] = []
|
|
series_profiles: list[SeriesProfile] = []
|
|
remaining_source_cells = MAX_CELL_READ_SIZE
|
|
remaining_dimension_spans = sum(
|
|
max(local_index for _, _, local_index in ref_dimensions)
|
|
- min(local_index for _, _, local_index in ref_dimensions)
|
|
+ 1
|
|
for ref_dimensions in mapped_by_ref.values()
|
|
)
|
|
for (source_sheet, source_range, bounds), ref_dimensions in mapped_by_ref.items():
|
|
dimension_start = bounds[2] if direction == "column" else bounds[0]
|
|
selected = {
|
|
dimension_start + local_index - 1: (dimension_index, role)
|
|
for dimension_index, role, local_index in ref_dimensions
|
|
}
|
|
dimension_span = max(selected) - min(selected) + 1
|
|
header_points = 0 if detached else 1
|
|
# Each sampled rectangle is read twice: cells-get supplies coordinates,
|
|
# styles, and hidden/filter semantics; table-get supplies typed values.
|
|
cells_per_dimension = remaining_source_cells // remaining_dimension_spans
|
|
remaining_dimension_spans -= dimension_span
|
|
point_axis_size = (
|
|
bounds[1] - bounds[0] + 1
|
|
if direction == "column"
|
|
else bounds[3] - bounds[2] + 1
|
|
)
|
|
if point_axis_size <= header_points:
|
|
unverifiable.append(
|
|
{
|
|
"chart_id": chart_id,
|
|
"reason": f"source has no data points: {source_sheet}!{source_range}",
|
|
}
|
|
)
|
|
continue
|
|
sample_points = min(MAX_SOURCE_SAMPLE_POINTS, max(0, (cells_per_dimension - header_points) // 2))
|
|
if sample_points == 0:
|
|
unverifiable.append({
|
|
"chart_id": chart_id,
|
|
"reason": f"source sampling 2000-cell budget cannot cover {source_sheet}!{source_range}",
|
|
})
|
|
continue
|
|
point_count = sample_points + header_points
|
|
if direction == "column":
|
|
checked_bounds = (
|
|
bounds[0],
|
|
min(bounds[1], bounds[0] + point_count - 1),
|
|
min(selected),
|
|
max(selected),
|
|
)
|
|
else:
|
|
checked_bounds = (
|
|
min(selected),
|
|
max(selected),
|
|
bounds[2],
|
|
min(bounds[3], bounds[2] + point_count - 1),
|
|
)
|
|
typed_bounds = (
|
|
(checked_bounds[0] + header_points, checked_bounds[1], checked_bounds[2], checked_bounds[3])
|
|
if direction == "column"
|
|
else (checked_bounds[0], checked_bounds[1], checked_bounds[2] + header_points, checked_bounds[3])
|
|
)
|
|
remaining_source_cells -= _bounds_area(checked_bounds) + _bounds_area(typed_bounds)
|
|
checked_range = _format_a1_bounds(checked_bounds)
|
|
same_sheet = source_sheet == owner_sheet_name
|
|
cells_data = _read_cells(
|
|
cache,
|
|
locator,
|
|
sheet_id=owner_sheet_id if same_sheet else None,
|
|
sheet_name=None if same_sheet else source_sheet,
|
|
cell_range=checked_range,
|
|
include="value,style",
|
|
skip_hidden=skip_hidden,
|
|
timeout=timeout,
|
|
)
|
|
cells_truncated = _cells_truncated(cells_data)
|
|
typed_rows, typed_columns, table_truncated = _read_typed_table(
|
|
cache,
|
|
locator,
|
|
sheet_id=owner_sheet_id if same_sheet else None,
|
|
sheet_name=None if same_sheet else source_sheet,
|
|
cell_range=_format_a1_bounds(typed_bounds),
|
|
timeout=timeout,
|
|
)
|
|
if cells_truncated:
|
|
unverifiable.append(
|
|
{"chart_id": chart_id, "reason": f"cells-get truncated for {source_sheet}!{checked_range}"}
|
|
)
|
|
if table_truncated:
|
|
unverifiable.append(
|
|
{
|
|
"chart_id": chart_id,
|
|
"reason": (
|
|
"table-get truncated or returned invalid typed data for "
|
|
f"{source_sheet}!{_format_a1_bounds(typed_bounds)}"
|
|
),
|
|
}
|
|
)
|
|
truncated = cells_truncated or table_truncated
|
|
states = {
|
|
coordinate: {
|
|
"zero": False,
|
|
"empty": False,
|
|
"nonzero": False,
|
|
"sample_point_count": 0,
|
|
"numeric_value_count": 0,
|
|
"unique_numeric_values": set(),
|
|
}
|
|
for coordinate in selected
|
|
}
|
|
type_unverifiable_dimensions: set[int] = set()
|
|
visible_offsets = {
|
|
(row_number - typed_bounds[0], column_index - typed_bounds[2])
|
|
for row_number, column_index, _ in _iter_cells(cells_data)
|
|
if typed_bounds[0] <= row_number <= typed_bounds[1]
|
|
and typed_bounds[2] <= column_index <= typed_bounds[3]
|
|
}
|
|
for row_number, column_index, cell in _iter_cells(cells_data):
|
|
coordinate = column_index if direction == "column" else row_number
|
|
dimension = selected.get(coordinate)
|
|
if dimension is None:
|
|
continue
|
|
if not detached and (
|
|
(direction == "column" and row_number == bounds[0])
|
|
or (direction != "column" and column_index == bounds[2])
|
|
):
|
|
continue
|
|
states[coordinate]["sample_point_count"] += 1
|
|
row_offset = row_number - typed_bounds[0]
|
|
column_offset = column_index - typed_bounds[2]
|
|
value, raw_type = _typed_cell(
|
|
typed_rows,
|
|
typed_columns,
|
|
row_offset,
|
|
column_offset,
|
|
visible_offsets,
|
|
)
|
|
_update_series_state(states[coordinate], value)
|
|
number_format = (
|
|
cell.get("cell_styles", {}).get("number_format")
|
|
if isinstance(cell, dict) and isinstance(cell.get("cell_styles"), dict)
|
|
else None
|
|
)
|
|
reason = ""
|
|
if _looks_numeric(str(value or "")):
|
|
if str(number_format or "").strip() == "@":
|
|
reason = "numeric_value_uses_text_format"
|
|
elif raw_type == "string":
|
|
reason = "numeric_value_stored_as_text"
|
|
elif raw_type in {"ambiguous_string", "unknown"}:
|
|
type_unverifiable_dimensions.add(dimension[0])
|
|
if reason:
|
|
dimension_index, role = dimension
|
|
key = (
|
|
dimension_index,
|
|
role,
|
|
source_sheet,
|
|
source_range,
|
|
checked_range,
|
|
reason,
|
|
)
|
|
issue_counts[key] = issue_counts.get(key, 0) + 1
|
|
samples = issue_groups.setdefault(key, [])
|
|
if len(samples) < sample_limit:
|
|
samples.append(f"{index_to_column(column_index)}{row_number}")
|
|
|
|
if truncated:
|
|
continue
|
|
for dimension_index in sorted(type_unverifiable_dimensions):
|
|
unverifiable.append(
|
|
{
|
|
"chart_id": chart_id,
|
|
"reason": (
|
|
"numeric storage type cannot be attributed to individual cells after "
|
|
f"typed column coercion for {source_sheet}!{checked_range}, "
|
|
f"dimension {dimension_index}"
|
|
),
|
|
}
|
|
)
|
|
zero_candidates = {
|
|
coordinate
|
|
for coordinate, state in states.items()
|
|
if not state["nonzero"]
|
|
and _series_aggregate_type(data, selected[coordinate][0]) != "count"
|
|
}
|
|
data_start = bounds[0] + (0 if detached else 1)
|
|
data_column = bounds[2] + (0 if detached else 1)
|
|
dim2 = data.get("dim2")
|
|
value_series = dim2.get("series") if isinstance(dim2, dict) else None
|
|
for coordinate, state in states.items():
|
|
dimension_index, role = selected[coordinate]
|
|
if direction == "column":
|
|
point_total = max(0, bounds[1] - data_start + 1)
|
|
series_range = (
|
|
f"{index_to_column(coordinate)}{data_start}:"
|
|
f"{index_to_column(coordinate)}{bounds[1]}"
|
|
if point_total
|
|
else ""
|
|
)
|
|
else:
|
|
point_total = max(0, bounds[3] - data_column + 1)
|
|
series_range = (
|
|
f"{index_to_column(data_column)}{coordinate}:"
|
|
f"{index_to_column(bounds[3])}{coordinate}"
|
|
if point_total
|
|
else ""
|
|
)
|
|
source_series = next(
|
|
(
|
|
item
|
|
for item in (value_series if isinstance(value_series, list) else [])
|
|
if isinstance(item, dict)
|
|
and item.get("index") is not None
|
|
and int(item["index"]) == dimension_index
|
|
),
|
|
{},
|
|
)
|
|
profile = {
|
|
"dimension_index": dimension_index,
|
|
"series_name": str(
|
|
source_series.get("name")
|
|
or source_series.get("nameRef")
|
|
or f"Series {dimension_index}"
|
|
),
|
|
"point_count": point_total,
|
|
"sample_point_count": state["sample_point_count"],
|
|
"checked_range": checked_range,
|
|
"sampled": (
|
|
checked_bounds[1] < bounds[1]
|
|
if direction == "column"
|
|
else checked_bounds[3] < bounds[3]
|
|
),
|
|
"numeric_value_count": state["numeric_value_count"],
|
|
"unique_numeric_values": list(state["unique_numeric_values"]),
|
|
"source_sheet": source_sheet,
|
|
"source_range": source_range,
|
|
"series_range": series_range,
|
|
}
|
|
aggregate_type = _series_aggregate_type(data, dimension_index)
|
|
if aggregate_type == "count":
|
|
profile["constant_check_unverifiable"] = True
|
|
constant_labeled = (
|
|
aggregate_type != "count"
|
|
and state["numeric_value_count"] >= 2
|
|
and len(state["unique_numeric_values"]) == 1
|
|
and dimension_index
|
|
in _labeled_series_indexes(snapshot, [{"dimension_index": dimension_index}])
|
|
)
|
|
if profile["sampled"]:
|
|
profile["constant_check_unverifiable"] = True
|
|
if coordinate in zero_candidates or constant_labeled:
|
|
unverifiable.append({
|
|
"chart_id": chart_id,
|
|
"reason": (
|
|
"zero/constant-series check is unverifiable outside sampled source "
|
|
f"{source_sheet}!{checked_range} for dimension {dimension_index}"
|
|
),
|
|
})
|
|
elif constant_labeled and _aggregation_can_change_constant(data, dimension_index):
|
|
profile["constant_check_unverifiable"] = True
|
|
unverifiable.append(
|
|
{
|
|
"chart_id": chart_id,
|
|
"reason": (
|
|
"constant-series check is unverifiable after category "
|
|
f"aggregation for dimension {dimension_index}"
|
|
),
|
|
}
|
|
)
|
|
series_profiles.append(profile)
|
|
if coordinate not in zero_candidates or profile["sampled"]:
|
|
continue
|
|
degenerate_series.append(
|
|
{
|
|
"chart_id": chart_id,
|
|
"dimension_index": dimension_index,
|
|
"role": role,
|
|
"source_sheet": source_sheet,
|
|
"source_range": source_range,
|
|
"series_range": series_range,
|
|
"reason": (
|
|
"numeric_series_all_zero_or_empty"
|
|
if states[coordinate]["zero"]
|
|
else "numeric_series_all_empty"
|
|
),
|
|
"data_point_count": point_total,
|
|
}
|
|
)
|
|
|
|
issues = [
|
|
{
|
|
"chart_id": chart_id,
|
|
"dimension_index": key[0],
|
|
"role": key[1],
|
|
"source_sheet": key[2],
|
|
"source_range": key[3],
|
|
"checked_range": key[4],
|
|
"reason": key[5],
|
|
"suggested_fix": (
|
|
"set_numeric_number_format"
|
|
if key[5] == "numeric_value_uses_text_format"
|
|
else "rewrite_as_number_and_set_numeric_number_format"
|
|
),
|
|
"affected_sample_cell_count": issue_counts[key],
|
|
"sample_cells": samples,
|
|
}
|
|
for key, samples in issue_groups.items()
|
|
]
|
|
return issues, degenerate_series, unverifiable, series_profiles
|
|
|
|
|
|
def _locator(target: str) -> dict[str, str]:
|
|
return {"url": target} if target.startswith(("http://", "https://")) else {"spreadsheet_token": target}
|
|
|
|
|
|
def _sheet_counts(sheet: dict[str, Any]) -> tuple[int, int]:
|
|
row_count = int(sheet.get("row_count") or sheet.get("rowCount") or 0)
|
|
column_count = int(sheet.get("column_count") or sheet.get("columnCount") or 0)
|
|
if row_count <= 0 or column_count <= 0:
|
|
raise LarkCliError(f"Missing row_count/column_count for sheet {sheet_title(sheet)!r}")
|
|
return row_count, column_count
|
|
|
|
|
|
def check_sheet(
|
|
locator: dict[str, str],
|
|
sheet: dict[str, Any],
|
|
*,
|
|
timeout: int,
|
|
sample_limit: int,
|
|
cell_cache: CellCache | None = None,
|
|
) -> dict[str, Any]:
|
|
cell_cache = cell_cache if cell_cache is not None else {}
|
|
sheet_id = sheet_identifier(sheet)
|
|
title = sheet_title(sheet)
|
|
if not sheet_id:
|
|
raise LarkCliError(f"Missing sheet_id for sheet {title!r}")
|
|
|
|
chart_data = envelope_data(
|
|
run_sheets("+chart-list", **locator, sheet_id=sheet_id, timeout=timeout)
|
|
)
|
|
charts = extract_charts(chart_data, sheet_id, title)
|
|
unverifiable: list[dict[str, str]] = []
|
|
expected_chart_count = sheet.get("chart_count")
|
|
if expected_chart_count is not None and int(expected_chart_count) != len(charts):
|
|
unverifiable.append(
|
|
{
|
|
"chart_id": "",
|
|
"reason": (
|
|
f"chart-list returned {len(charts)} charts, "
|
|
f"but workbook-info reported {int(expected_chart_count)}"
|
|
),
|
|
}
|
|
)
|
|
if not charts:
|
|
return {
|
|
"sheet_id": sheet_id,
|
|
"sheet_name": title,
|
|
"chart_count": 0,
|
|
"sheet_size_px": None,
|
|
"chart_overlaps": [],
|
|
"cell_content_overlaps": [],
|
|
"numeric_source_format_issues": [],
|
|
"numeric_source_samples": [],
|
|
"degenerate_numeric_series": [],
|
|
"constant_labeled_series": [],
|
|
"unbound_secondary_axes": [],
|
|
"undersized_charts": [],
|
|
"overwide_charts": [],
|
|
"out_of_visible_range": [],
|
|
"unverifiable_charts": unverifiable,
|
|
"issue_count": 0,
|
|
"unverifiable_count": len(unverifiable),
|
|
"warnings": [],
|
|
}
|
|
|
|
row_count, column_count = _sheet_counts(sheet)
|
|
structure_data = envelope_data(
|
|
run_sheets(
|
|
"+sheet-info",
|
|
**locator,
|
|
sheet_id=sheet_id,
|
|
flags={"include": "row_heights,col_widths"},
|
|
timeout=timeout,
|
|
)
|
|
)
|
|
row_edges, column_edges, warnings = build_layout(
|
|
extract_sheet_structure(structure_data), row_count, column_count
|
|
)
|
|
|
|
rectangles: list[dict[str, Any]] = []
|
|
for chart in charts:
|
|
chart_id = str(chart.get("chart_id") or chart.get("id") or "")
|
|
if not chart_id:
|
|
unverifiable.append({"chart_id": "", "reason": "chart is missing chart_id"})
|
|
continue
|
|
try:
|
|
rectangles.append(chart_rectangle(chart, row_edges, column_edges))
|
|
except (KeyError, TypeError, ValueError) as exc:
|
|
unverifiable.append({"chart_id": chart_id, "reason": str(exc)})
|
|
|
|
overlaps: list[dict[str, Any]] = []
|
|
for index, first in enumerate(rectangles):
|
|
for second in rectangles[index + 1 :]:
|
|
overlap = intersection(first, second)
|
|
if overlap:
|
|
overlaps.append(
|
|
{
|
|
"chart_ids": [first["chart_id"], second["chart_id"]],
|
|
"charts": [chart_context(first), chart_context(second)],
|
|
"intersection": overlap,
|
|
}
|
|
)
|
|
|
|
sheet_width = column_edges[-1]
|
|
sheet_height = row_edges[-1]
|
|
out_of_bounds: list[dict[str, Any]] = []
|
|
content_overlaps: list[dict[str, Any]] = []
|
|
covered_items: list[tuple[dict[str, Any], CellBounds]] = []
|
|
for rectangle in rectangles:
|
|
overflow = {
|
|
"left": round(max(0.0, -rectangle["left"]), 2),
|
|
"top": round(max(0.0, -rectangle["top"]), 2),
|
|
"right": round(max(0.0, rectangle["right"] - sheet_width), 2),
|
|
"bottom": round(max(0.0, rectangle["bottom"] - sheet_height), 2),
|
|
}
|
|
if any(overflow.values()):
|
|
out_of_bounds.append({**chart_context(rectangle), "overflow_px": overflow})
|
|
|
|
covered_range = rectangle_cell_range(rectangle, row_edges, column_edges)
|
|
if not covered_range:
|
|
continue
|
|
covered_items.append((rectangle, _parse_a1_bounds(covered_range)))
|
|
|
|
for cluster in _cluster_cell_reads(covered_items):
|
|
read_range = _format_a1_bounds(cluster["bounds"])
|
|
cells_data = _read_cells(
|
|
cell_cache,
|
|
locator,
|
|
sheet_id=sheet_id,
|
|
sheet_name=None,
|
|
cell_range=read_range,
|
|
include="value,formula,comment",
|
|
timeout=timeout,
|
|
)
|
|
for rectangle, bounds in cluster["members"]:
|
|
covered_range = _format_a1_bounds(bounds)
|
|
count, samples, truncated = non_empty_cells(cells_data, sample_limit, bounds)
|
|
if truncated:
|
|
unverifiable.append(
|
|
{
|
|
"chart_id": rectangle["chart_id"],
|
|
"reason": f"cells-get truncated for {read_range}",
|
|
}
|
|
)
|
|
if count:
|
|
content_overlaps.append(
|
|
{
|
|
**chart_context(rectangle),
|
|
"covered_range": covered_range,
|
|
"non_empty_cell_count": count,
|
|
"sample_cells": samples,
|
|
}
|
|
)
|
|
|
|
numeric_source_issues: list[dict[str, Any]] = []
|
|
numeric_source_samples: list[dict[str, Any]] = []
|
|
degenerate_numeric_series: list[dict[str, Any]] = []
|
|
constant_series_issues: list[dict[str, Any]] = []
|
|
unbound_secondary_axes: list[dict[str, Any]] = []
|
|
undersized_charts: list[dict[str, Any]] = []
|
|
overwide_charts: list[dict[str, Any]] = []
|
|
for chart in charts:
|
|
issues, degenerate, source_unverifiable, profiles = _numeric_source_issues(
|
|
chart,
|
|
owner_sheet_id=sheet_id,
|
|
owner_sheet_name=title,
|
|
cache=cell_cache,
|
|
locator=locator,
|
|
timeout=timeout,
|
|
sample_limit=sample_limit,
|
|
)
|
|
numeric_source_issues.extend(issues)
|
|
numeric_source_samples.extend(
|
|
{"chart_id": str(chart.get("chart_id") or chart.get("id") or ""), **profile}
|
|
for profile in profiles
|
|
if "checked_range" in profile
|
|
)
|
|
degenerate_numeric_series.extend(degenerate)
|
|
unverifiable.extend(source_unverifiable)
|
|
constant_series_issues.extend(_constant_labeled_series(chart, profiles))
|
|
unbound_secondary_axis = _unbound_secondary_axis(chart)
|
|
if unbound_secondary_axis:
|
|
unbound_secondary_axes.append(unbound_secondary_axis)
|
|
undersized = _undersized_chart(chart)
|
|
if undersized:
|
|
undersized_charts.append(undersized)
|
|
overwide = _overwide_chart(chart)
|
|
if overwide:
|
|
overwide_charts.append(overwide)
|
|
|
|
issue_count = (
|
|
len(overlaps)
|
|
+ len(out_of_bounds)
|
|
+ len(content_overlaps)
|
|
+ len(numeric_source_issues)
|
|
+ len(degenerate_numeric_series)
|
|
+ len(constant_series_issues)
|
|
+ len(unbound_secondary_axes)
|
|
+ len(undersized_charts)
|
|
+ len(overwide_charts)
|
|
)
|
|
return {
|
|
"sheet_id": sheet_id,
|
|
"sheet_name": title,
|
|
"chart_count": len(charts),
|
|
"sheet_size_px": {"width": round(sheet_width, 2), "height": round(sheet_height, 2)},
|
|
"chart_overlaps": overlaps,
|
|
"cell_content_overlaps": content_overlaps,
|
|
"numeric_source_format_issues": numeric_source_issues,
|
|
"numeric_source_samples": numeric_source_samples,
|
|
"degenerate_numeric_series": degenerate_numeric_series,
|
|
"constant_labeled_series": constant_series_issues,
|
|
"unbound_secondary_axes": unbound_secondary_axes,
|
|
"undersized_charts": undersized_charts,
|
|
"overwide_charts": overwide_charts,
|
|
"out_of_visible_range": out_of_bounds,
|
|
"unverifiable_charts": unverifiable,
|
|
"issue_count": issue_count,
|
|
"unverifiable_count": len(unverifiable),
|
|
"warnings": warnings,
|
|
}
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
description=(
|
|
"Check chart overlap, covered cell content, worksheet boundary overflow, "
|
|
"minimum size, excessive width, constant labeled series, numeric source-cell "
|
|
"formats, all-zero/empty numeric series, and unbound combo-chart secondary axes."
|
|
)
|
|
)
|
|
parser.add_argument("sheet_id", help="Spreadsheet URL or spreadsheet token")
|
|
parser.add_argument("--worksheet-id", help="Only check this worksheet reference_id")
|
|
parser.add_argument("--timeout", type=int, default=60)
|
|
parser.add_argument("--sample-limit", type=int, default=10)
|
|
return parser.parse_args()
|
|
|
|
|
|
def success_envelope(results: list[dict[str, Any]]) -> dict[str, Any]:
|
|
issue_count = sum(result["issue_count"] for result in results)
|
|
unverifiable_count = sum(result["unverifiable_count"] for result in results)
|
|
warnings = [
|
|
f"{result['sheet_name'] or result['sheet_id']}: {warning}"
|
|
for result in results
|
|
for warning in result["warnings"]
|
|
]
|
|
return {
|
|
"ok": True,
|
|
"engine": "lark",
|
|
"action": ACTION,
|
|
"data": {
|
|
"passed": issue_count == 0 and unverifiable_count == 0,
|
|
"scope_note": (
|
|
"out_of_visible_range checks worksheet drawable bounds, not a device-specific browser viewport; "
|
|
"numeric source checks sample at most the first 50 data points of each chart value dimension "
|
|
"for formats, all-zero/empty, and constant-value checks; zero/constant results are conclusive "
|
|
"only when that sample covers the complete series, otherwise they are marked unverifiable"
|
|
),
|
|
"summary": {
|
|
"worksheet_count": len(results),
|
|
"chart_count": sum(result["chart_count"] for result in results),
|
|
"issue_count": issue_count,
|
|
"unverifiable_count": unverifiable_count,
|
|
},
|
|
"sheets": results,
|
|
},
|
|
"warnings": warnings,
|
|
}
|
|
|
|
|
|
def report_exit_code(report: dict[str, Any]) -> int:
|
|
if report["data"]["passed"]:
|
|
return 0
|
|
if report["data"]["summary"]["issue_count"] > 0:
|
|
return 2
|
|
return 1
|
|
|
|
|
|
def main() -> None:
|
|
args = parse_args()
|
|
locator = _locator(args.sheet_id)
|
|
cell_cache: CellCache = {}
|
|
try:
|
|
workbook_data = envelope_data(
|
|
run_sheets("+workbook-info", **locator, timeout=args.timeout)
|
|
)
|
|
sheets = resolve_target_sheets(workbook_data, sheet_id=args.worksheet_id)
|
|
if not args.worksheet_id:
|
|
sheets = [sheet for sheet in sheets if not bool(sheet.get("is_hidden"))]
|
|
if not sheets:
|
|
raise LarkCliError("No visible worksheet matched")
|
|
results = [
|
|
check_sheet(
|
|
locator,
|
|
sheet,
|
|
timeout=args.timeout,
|
|
sample_limit=args.sample_limit,
|
|
cell_cache=cell_cache,
|
|
)
|
|
for sheet in sheets
|
|
]
|
|
except (LarkCliError, KeyError, TypeError, ValueError) as exc:
|
|
emit_error(ACTION, str(exc))
|
|
raise SystemExit(1) from exc
|
|
|
|
report = success_envelope(results)
|
|
print(json.dumps(report, ensure_ascii=False, indent=2))
|
|
exit_code = report_exit_code(report)
|
|
if exit_code:
|
|
raise SystemExit(exit_code)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|