Files
lark-sheets/scripts/lark_profile_table.py

615 lines
22 KiB
Python

#!/usr/bin/env python3
# Copyright (c) 2026 Lark Technologies Pte. Ltd.
# SPDX-License-Identifier: MIT
"""Profile a candidate table range in a Lark sheet."""
from __future__ import annotations
import argparse
import re
from typing import Any
from lark_sheet_range import col_to_index, format_range, index_to_col, parse_range
from lark_sheet_read_cli import (
LarkCliError,
add_spreadsheet_args,
emit_error,
emit_success,
envelope_data,
run_sheets,
)
from lark_detect_subtables import CsvGrid, parse_annotated_csv
ACTION = "profile_table"
ERROR_VALUES = ("#VALUE!", "#DIV/0!", "#REF!", "#NAME?", "#NULL!", "#NUM!", "#N/A")
TOTAL_KEYWORDS = ("合计", "总计", "小计", "汇总", "累计")
TOTAL_EN_RE = re.compile(r"\b(?:grand total|subtotal|total)\b", re.IGNORECASE)
SIGNATURE_KEYWORDS = ("编制人", "审核人", "审批人", "负责人", "经理", "签名")
def _is_number(value: str) -> bool:
text = value.strip().replace(",", "")
if text.endswith("%"):
text = text[:-1]
try:
float(text)
return True
except ValueError:
return False
def _is_date_like(value: str) -> bool:
text = value.strip()
if re.match(r"^\d{4}[-/]\d{1,2}[-/]\d{1,2}(?:\s+\d{1,2}:\d{2}(:\d{2})?)?$", text):
return True
if re.match(r"^\d{8}$", text):
year = int(text[:4])
month = int(text[4:6])
day = int(text[6:8])
return 1900 <= year <= 2100 and 1 <= month <= 12 and 1 <= day <= 31
if re.match(r"^\d{6}$", text):
year = int(text[:4])
month = int(text[4:6])
return 1900 <= year <= 2100 and 1 <= month <= 12
return False
def _classify_value(value: str) -> str:
text = value.strip()
if not text:
return "empty"
if any(err in text for err in ERROR_VALUES):
return "error"
if re.match(r"^0\d+$", text) or re.match(r"^\d{11,}$", text):
return "string_id"
if _is_date_like(text):
return "date"
if _is_number(text):
return "number"
if text.lower() in {"true", "false", "yes", "no"}:
return "bool"
return "string"
def _row_values(grid: CsvGrid, row_index: int) -> list[str]:
return grid.values[row_index]
def _header_score(row: list[str]) -> tuple[int, int]:
non_empty = [value for value in row if value.strip()]
if not non_empty:
return 0, 0
strings = sum(1 for value in non_empty if _classify_value(value) in {"string", "string_id"})
data_like = sum(
1 for value in non_empty if _classify_value(value) in {"number", "date", "bool"}
)
return strings * 2 + len(non_empty) - data_like * 3, len(non_empty)
def _header_candidate_score(grid: CsvGrid, index: int) -> int:
score, _ = _header_score(grid.values[index])
following = grid.values[index + 1 : index + 4]
following_non_empty = [value for row in following for value in row if value.strip()]
following_data_like = sum(
1 for value in following_non_empty if _classify_value(value) in {"number", "date", "bool"}
)
# A schema row is usually followed by data-shaped values; use this as a
# small tie-breaker while preferring earlier candidates with equal evidence.
score += min(3, following_data_like)
return score - index // 2
def detect_header_row(grid: CsvGrid, scan_rows: int = 20) -> int | None:
best_row = None
best_score = None
for idx, row in enumerate(grid.values[:scan_rows]):
_, non_empty_count = _header_score(row)
if not non_empty_count:
continue
score = _header_candidate_score(grid, idx)
if best_score is None or score > best_score:
best_score = score
best_row = grid.row_numbers[idx]
return best_row
def _possible_multi_row_header(grid: CsvGrid, header_row: int | None) -> bool:
if header_row is None or header_row not in grid.row_numbers:
return False
header_index = grid.row_numbers.index(header_row)
if header_index + 1 >= len(grid.values):
return False
header_score, header_non_empty = _header_score(grid.values[header_index])
next_score, next_non_empty = _header_score(grid.values[header_index + 1])
if header_non_empty < 2 or next_non_empty < 2:
return False
return next_score >= max(4, (header_score * 3 + 3) // 4)
def _header_row_not_first(grid: CsvGrid, header_row: int | None) -> bool:
if header_row is None:
return False
first_non_empty = next(
(
row_num
for row_num, row in zip(grid.row_numbers, grid.values)
if any(value.strip() for value in row)
),
None,
)
return first_non_empty is not None and first_non_empty != header_row
def detect_special_rows(grid: CsvGrid, header_row: int | None) -> list[dict[str, Any]]:
specials = []
for idx, row in enumerate(grid.values):
row_num = grid.row_numbers[idx]
if header_row is not None and row_num <= header_row:
continue
joined = " ".join(value.strip() for value in row if value.strip())
if not joined:
specials.append({"row": row_num, "reason": "empty_row", "sample": ""})
continue
if any(keyword in joined for keyword in TOTAL_KEYWORDS) or TOTAL_EN_RE.search(joined):
specials.append({"row": row_num, "reason": "total_row", "sample": joined[:120]})
elif any(keyword in joined for keyword in SIGNATURE_KEYWORDS):
specials.append({"row": row_num, "reason": "signature_row", "sample": joined[:120]})
return specials
def _last_data_row(grid: CsvGrid, header_row: int | None, specials: list[dict[str, Any]]) -> int | None:
special_rows = {item["row"] for item in specials if item["reason"] != "empty_row"}
for idx in range(len(grid.row_numbers) - 1, -1, -1):
row_num = grid.row_numbers[idx]
if header_row is not None and row_num <= header_row:
continue
if row_num in special_rows:
continue
if any(value.strip() for value in grid.values[idx]):
return row_num
return None
def _column_values(grid: CsvGrid, col_index: int, start_row: int, end_row: int | None) -> list[str]:
values = []
for idx, row_num in enumerate(grid.row_numbers):
if row_num < start_row:
continue
if end_row is not None and row_num > end_row:
continue
row = grid.values[idx]
values.append(row[col_index] if col_index < len(row) else "")
return values
def _type_guess(type_counts: dict[str, int]) -> str:
non_empty_counts = {key: value for key, value in type_counts.items() if key != "empty" and value}
if not non_empty_counts:
return "empty"
if len(non_empty_counts) == 1:
return next(iter(non_empty_counts))
if "error" in non_empty_counts:
return "mixed_with_errors"
if "string_id" in non_empty_counts and set(non_empty_counts) <= {"string_id", "string"}:
return "string_id"
return "mixed"
def profile_columns(
grid: CsvGrid,
*,
header_row: int | None,
data_start_row: int,
data_end_row: int | None,
) -> list[dict[str, Any]]:
header_values = []
if header_row in grid.row_numbers:
header_values = _row_values(grid, grid.row_numbers.index(header_row))
profiles = []
for idx, col in enumerate(grid.col_letters):
name = ""
if idx < len(header_values):
name = header_values[idx].strip()
if not name:
name = f"unnamed_{col}"
values = _column_values(grid, idx, data_start_row, data_end_row)
type_counts: dict[str, int] = {}
examples = []
error_count = 0
for value in values:
value_type = _classify_value(value)
type_counts[value_type] = type_counts.get(value_type, 0) + 1
if value_type == "error":
error_count += 1
if value.strip() and len(examples) < 5:
examples.append(value)
non_empty = len(values) - type_counts.get("empty", 0)
warnings = []
if type_counts.get("string_id"):
warnings.append("long_numeric_like_id")
if error_count:
warnings.append("formula_or_value_errors")
if non_empty and type_counts.get("empty", 0) / len(values) > 0.5:
warnings.append("many_empty_cells")
if _type_guess(type_counts) == "mixed":
warnings.append("mixed_value_types")
profiles.append(
{
"name": name,
"col": col,
"non_empty": non_empty,
"empty": type_counts.get("empty", 0),
"type_guess": _type_guess(type_counts),
"type_distribution": type_counts,
"examples": examples,
"warnings": warnings,
}
)
return profiles
def _field_map(columns: list[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]:
result: dict[str, list[dict[str, Any]]] = {}
for column in columns:
name = column["name"]
result.setdefault(name, []).append(
{
"col": column["col"],
"type_guess": column["type_guess"],
"warnings": column["warnings"],
}
)
return result
def _risk_warnings(
*,
columns: list[dict[str, Any]],
special_rows: list[dict[str, Any]],
header_row: int | None,
data_range: str | None,
possible_multi_row_header: bool,
header_row_not_first: bool,
data_range_has_gaps: bool,
data_range_has_col_gaps: bool,
hidden_rows: list[int],
hidden_columns: list[str],
) -> list[str]:
warnings = {warning for column in columns for warning in column["warnings"]}
names = [column["name"] for column in columns]
if len(names) != len(set(names)):
warnings.add("duplicate_headers")
if any(name.startswith("unnamed_") for name in names):
warnings.add("unnamed_columns")
if header_row is None:
warnings.add("header_not_detected")
if data_range is None:
warnings.add("data_range_not_detected")
if any(item["reason"] != "empty_row" for item in special_rows):
warnings.add("special_rows_present")
if any(item["reason"] == "empty_row" for item in special_rows):
warnings.add("empty_rows_present")
if possible_multi_row_header:
warnings.add("possible_multi_row_header")
if header_row_not_first:
warnings.add("header_row_not_first")
if data_range_has_gaps:
warnings.add("data_range_has_gaps")
if data_range_has_col_gaps:
warnings.add("data_range_has_col_gaps")
if hidden_rows:
warnings.add("hidden_rows_in_range")
if hidden_columns:
warnings.add("hidden_columns_in_range")
return sorted(warnings)
def _col_segments(col_letters: list[str]) -> list[list[str]]:
"""Group column letters into contiguous runs of real column indices.
With --skip-hidden the returned columns can be non-consecutive (A, C when B
is hidden). Reporting a single data_range across that gap would let a caller
write the hidden-column-free data back as if it were contiguous, shifting
every value right of the gap.
"""
if not col_letters:
return []
indices = [col_to_index(col) for col in col_letters]
segments = []
start = previous = indices[0]
for current in indices[1:]:
if current == previous + 1:
previous = current
continue
segments.append([index_to_col(start), index_to_col(previous)])
start = previous = current
segments.append([index_to_col(start), index_to_col(previous)])
return segments
def _row_segments(row_numbers: list[int]) -> list[list[int]]:
if not row_numbers:
return []
segments = []
start = end = row_numbers[0]
for row in row_numbers[1:]:
if row == end + 1:
end = row
continue
segments.append([start, end])
start = end = row
segments.append([start, end])
return segments
def _write_hints(
grid: CsvGrid,
*,
header_row: int | None,
data_start: int,
bounds,
hidden_columns: list[str] | None = None,
skip_hidden: bool = False,
) -> dict[str, Any]:
last_non_empty_col = None
for idx, col in enumerate(grid.col_letters):
if any(idx < len(row) and row[idx].strip() for row in grid.values):
last_non_empty_col = col_to_index(col)
safe_col_num = (last_non_empty_col + 1) if last_non_empty_col else bounds.start_col
# Step over hidden columns. Under --skip-hidden they are absent from the
# grid entirely, so "one past the last visible column" can land ON a hidden
# column that holds data — and when the hidden columns sit at the right edge
# there is no gap in the returned letters either, so data_range_has_col_gaps
# stays silent too. Appending there would overwrite data nobody can see.
hidden_indices = {col_to_index(col) for col in (hidden_columns or [])}
skipped_hidden: list[str] = []
while safe_col_num in hidden_indices:
skipped_hidden.append(index_to_col(safe_col_num))
safe_col_num += 1
safe_col = index_to_col(safe_col_num)
hints = {
"last_non_empty_col": index_to_col(last_non_empty_col) if last_non_empty_col else None,
}
if not (skip_hidden and hidden_columns is None):
hints.update(
{
"safe_append_col": safe_col,
"safe_append_header_cell": f"{safe_col}{header_row}" if header_row else None,
"safe_append_data_start_cell": f"{safe_col}{data_start}",
}
)
if skipped_hidden:
hints["skipped_hidden_cols"] = skipped_hidden
return hints
def profile_grid(
grid: CsvGrid,
source_range: str,
*,
skip_hidden: bool = False,
hidden_rows: list[int] | None = None,
hidden_columns: list[str] | None = None,
all_hidden_columns: list[str] | None = None,
header_scan_rows: int = 20,
) -> dict[str, Any]:
max_row = max(grid.row_numbers, default=1)
max_col = max((col_to_index(col) for col in grid.col_letters), default=1)
bounds = parse_range(source_range, max_row=max_row, max_col=max_col)
header_row = detect_header_row(grid, header_scan_rows)
data_start = (header_row + 1) if header_row else bounds.start_row
specials = detect_special_rows(grid, header_row)
data_end = _last_data_row(grid, header_row, specials)
data_range = None
if data_end is not None and data_end >= data_start:
data_range = format_range(data_start, bounds.start_col, data_end, bounds.end_col)
columns = profile_columns(
grid,
header_row=header_row,
data_start_row=data_start,
data_end_row=data_end,
)
data_row_numbers = []
if data_end is not None and data_end >= data_start:
data_row_numbers = [
row for row in grid.row_numbers if data_start <= row <= data_end
]
data_row_segments = _row_segments(data_row_numbers)
data_range_has_gaps = bool(data_row_numbers) and (
data_row_numbers[0] != data_start or len(data_row_segments) > 1
)
data_rows = len(data_row_numbers)
data_col_segments = _col_segments(grid.col_letters)
data_range_has_col_gaps = len(data_col_segments) > 1
hidden_rows = hidden_rows or []
hidden_columns = hidden_columns or []
possible_multi_row_header = _possible_multi_row_header(grid, header_row)
header_row_not_first = _header_row_not_first(grid, header_row)
return {
"summary": {
"range": source_range,
"header_row": header_row,
"data_range": data_range,
"data_rows": data_rows,
"data_row_segments": data_row_segments,
"data_col_segments": data_col_segments,
"column_count": len(columns),
"special_rows_count": len(specials),
},
"range": source_range,
"header_row": header_row,
"data_range": data_range,
"data_row_segments": data_row_segments,
"data_col_segments": data_col_segments,
"columns": columns,
"field_map": _field_map(columns),
"risk_warnings": _risk_warnings(
columns=columns,
special_rows=specials,
header_row=header_row,
data_range=data_range,
possible_multi_row_header=possible_multi_row_header,
header_row_not_first=header_row_not_first,
data_range_has_gaps=data_range_has_gaps,
data_range_has_col_gaps=data_range_has_col_gaps,
hidden_rows=hidden_rows,
hidden_columns=hidden_columns,
),
"visibility": {
"skip_hidden": skip_hidden,
"hidden_rows_in_range": hidden_rows,
"hidden_columns_in_range": hidden_columns,
},
"write_hints": _write_hints(
grid,
header_row=header_row,
data_start=data_start,
bounds=bounds,
hidden_columns=all_hidden_columns,
skip_hidden=skip_hidden,
),
"special_rows": specials,
}
def _hidden_rows_and_columns(grid: CsvGrid, layout: dict[str, Any]) -> tuple[list[int], list[str]]:
def indexes(key: str) -> set[int]:
values = layout.get(key)
if not isinstance(values, list):
return set()
result = set()
for value in values:
try:
result.add(int(value) + 1) # +sheet-info uses zero-based indices.
except (TypeError, ValueError):
continue
return result
hidden_row_indexes = indexes("hidden_rows")
hidden_columns = layout.get("hidden_cols") or layout.get("hidden_columns") or []
hidden_column_letters = set()
for value in hidden_columns if isinstance(hidden_columns, list) else []:
if isinstance(value, str) and value.isalpha():
hidden_column_letters.add(value.upper())
else:
try:
hidden_column_letters.add(index_to_col(int(value) + 1))
except (TypeError, ValueError):
continue
rows = sorted(row for row in grid.row_numbers if row in hidden_row_indexes)
columns = [col for col in grid.col_letters if col.upper() in hidden_column_letters]
return rows, columns
def _all_hidden_columns(layout: dict[str, Any]) -> list[str]:
"""Every hidden column on the sheet, not just those inside the grid.
_hidden_rows_and_columns intersects with the returned grid, which is the
right scope for the "there are hidden rows/columns in what you read"
warnings. The append hint needs the opposite: the columns that are NOT in
the grid precisely because they are hidden.
"""
letters: list[str] = []
raw = layout.get("hidden_cols") or layout.get("hidden_columns") or []
for value in raw if isinstance(raw, list) else []:
if isinstance(value, str) and value.isalpha():
letters.append(value.upper())
continue
try:
letters.append(index_to_col(int(value) + 1))
except (TypeError, ValueError):
continue
return letters
def profile_table(args) -> tuple[dict[str, Any], list[str]]:
warnings = []
if args.header_scan_rows < 1:
raise ValueError("--header-scan-rows must be at least 1")
csv_data = envelope_data(
run_sheets(
"+csv-get",
url=args.url,
spreadsheet_token=args.spreadsheet_token,
sheet_id=args.sheet_id,
sheet_name=args.sheet_name,
flags={
"range": args.range,
"max_chars": args.max_chars,
"skip_hidden": True if args.skip_hidden else None,
},
timeout=args.timeout,
)
)
source_range = str(csv_data.get("actual_range") or args.range)
if csv_data.get("has_more"):
raise LarkCliError(
f"+csv-get truncated the requested range at {source_range}; narrow --range before profiling"
)
grid = parse_annotated_csv(
csv_data.get("annotated_csv", ""),
csv_data.get("col_indices"),
csv_data.get("row_indices"),
source_range,
)
if grid.row_numbers_inferred:
warnings.append("CSV row numbers were inferred from the requested range")
hidden_rows: list[int] = []
hidden_columns: list[str] = []
all_hidden_columns: list[str] | None = None
# Fetched in BOTH modes. Under --skip-hidden the hidden rows/columns are
# absent from the grid, so hidden_rows/hidden_columns (which are scoped to
# what the grid contains) come back empty and no warning fires — but the
# write hints still need to know where the hidden columns are, or
# safe_append_col can point at one that holds data. all_hidden_columns is
# the unscoped list used for exactly that.
try:
layout = envelope_data(
run_sheets(
"+sheet-info",
url=args.url,
spreadsheet_token=args.spreadsheet_token,
sheet_id=args.sheet_id,
sheet_name=args.sheet_name,
flags={"include": "hidden_rows,hidden_cols"},
timeout=args.timeout,
)
)
hidden_rows, hidden_columns = _hidden_rows_and_columns(grid, layout)
all_hidden_columns = _all_hidden_columns(layout)
except LarkCliError as exc:
warnings.append(f"hidden row/column detection unavailable: {exc}")
return profile_grid(
grid,
source_range,
skip_hidden=args.skip_hidden,
hidden_rows=hidden_rows,
hidden_columns=hidden_columns,
all_hidden_columns=all_hidden_columns,
header_scan_rows=args.header_scan_rows,
), warnings
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
add_spreadsheet_args(parser, require_sheet=True, allow_sheet=True)
parser.add_argument("--range", required=True)
parser.add_argument("--max-chars", type=int, default=25000)
parser.add_argument("--header-scan-rows", type=int, default=20)
parser.add_argument("--skip-hidden", action="store_true")
parser.add_argument("--timeout", type=int, default=60)
args = parser.parse_args()
try:
data, warnings = profile_table(args)
except (LarkCliError, ValueError, TypeError) as exc:
emit_error(ACTION, str(exc))
emit_success(ACTION, data, warnings)
if __name__ == "__main__":
main()