| """Deterministic spreadsheet inspection."""
|
|
|
| from __future__ import annotations
|
|
|
| import json
|
| from pathlib import Path
|
|
|
| import pandas as pd
|
| from openpyxl import load_workbook
|
|
|
|
|
| def describe_workbook(path: str | Path) -> str:
|
| """Describe sheet dimensions, headers, formulas, and inferred cell data types."""
|
| workbook = load_workbook(path, data_only=False, read_only=True)
|
| description: dict[str, object] = {}
|
| for sheet in workbook.worksheets:
|
| rows = sheet.iter_rows()
|
| first_row = next(rows, ())
|
| headers = [cell.value for cell in first_row]
|
| formulas = [cell.coordinate for cell in first_row if cell.data_type == "f"]
|
| types: dict[str, int] = {}
|
| for cell in first_row:
|
| types[cell.data_type] = types.get(cell.data_type, 0) + 1
|
| formula_count = len(formulas)
|
| for row in rows:
|
| for cell in row:
|
| types[cell.data_type] = types.get(cell.data_type, 0) + 1
|
| if cell.data_type == "f":
|
| formula_count += 1
|
| if len(formulas) < 50:
|
| formulas.append(cell.coordinate)
|
| description[sheet.title] = {
|
| "rows": sheet.max_row,
|
| "columns": sheet.max_column,
|
| "headers": headers,
|
| "formula_count": formula_count,
|
| "sample_formula_cells": formulas[:50],
|
| "cell_types": types,
|
| }
|
| return json.dumps(description, ensure_ascii=False, default=str)
|
|
|
|
|
| def read_sheet(
|
| path: str | Path, sheet_name: str, max_rows: int = 1000
|
| ) -> list[dict[str, object]]:
|
| frame = pd.read_excel(path, sheet_name=sheet_name)
|
| return frame.where(pd.notna(frame), None).head(max_rows).to_dict(orient="records")
|
|
|
|
|
| def filter_rows(
|
| path: str | Path, sheet_name: str, column: str, value: str, exclude: bool = False
|
| ) -> list[dict[str, object]]:
|
| frame = pd.read_excel(path, sheet_name=sheet_name)
|
| if column not in frame.columns:
|
| raise KeyError(f"Unknown column: {column}")
|
| mask = (
|
| frame[column].astype(str).str.contains(value, case=False, regex=False, na=False)
|
| )
|
| selected = frame[~mask if exclude else mask]
|
| return selected.where(pd.notna(selected), None).to_dict(orient="records")
|
|
|
|
|
| def sum_column(
|
| path: str | Path,
|
| sheet_name: str,
|
| column: str,
|
| filter_column: str | None = None,
|
| filter_value: str | None = None,
|
| exclude: bool = False,
|
| ) -> float:
|
| frame = pd.read_excel(path, sheet_name=sheet_name)
|
| if filter_column and filter_value is not None:
|
| mask = (
|
| frame[filter_column]
|
| .astype(str)
|
| .str.contains(filter_value, case=False, regex=False, na=False)
|
| )
|
| frame = frame[~mask if exclude else mask]
|
| return float(pd.to_numeric(frame[column], errors="coerce").sum())
|
|
|
|
|
| def inspect_spreadsheet(path: str | Path, max_rows: int = 200) -> str:
|
| """Return bounded, structured workbook contents for downstream reasoning."""
|
| file_path = Path(path)
|
| if file_path.suffix.lower() not in {".xlsx", ".xls"}:
|
| raise ValueError("Expected an Excel workbook")
|
| sheets = pd.read_excel(file_path, sheet_name=None)
|
| result: dict[str, object] = {
|
| "description": json.loads(describe_workbook(file_path))
|
| }
|
| for name, frame in sheets.items():
|
| clean = frame.where(pd.notna(frame), None)
|
| result[str(name)] = {
|
| "shape": [int(frame.shape[0]), int(frame.shape[1])],
|
| "columns": [str(column) for column in frame.columns],
|
| "rows": clean.head(max_rows).to_dict(orient="records"),
|
| "truncated": len(frame) > max_rows,
|
| }
|
| return json.dumps(result, ensure_ascii=False, default=str)
|
|
|
|
|
| def answer_spreadsheet_question(question: str, path: str | Path) -> str | None:
|
| """Answer well-defined aggregation questions from workbook data when possible."""
|
| lowered = question.lower()
|
| if not (
|
| "total sales" in lowered
|
| and ("not including drinks" in lowered or "excluding drinks" in lowered)
|
| ):
|
| return None
|
| describe_workbook(path)
|
| frames = pd.read_excel(path, sheet_name=None)
|
| frame = pd.concat(frames.values(), ignore_index=True)
|
| normalized = {str(column).strip().lower(): column for column in frame.columns}
|
| value_column = next(
|
| (
|
| normalized[name]
|
| for name in ("total sales", "sales", "revenue", "amount")
|
| if name in normalized
|
| ),
|
| None,
|
| )
|
| if value_column is None:
|
| beverage = (
|
| r"\b(?:drink|beverage|soda|cola|coffee|tea|juice|water|shake|"
|
| r"smoothie|lemonade)\b"
|
| )
|
| numeric_columns = [
|
| column
|
| for column in frame.columns
|
| if pd.api.types.is_numeric_dtype(frame[column])
|
| and not pd.Series([str(column)])
|
| .str.contains(beverage, case=False, regex=True)
|
| .iloc[0]
|
| ]
|
| if not numeric_columns:
|
| return None
|
| total = frame[numeric_columns].apply(pd.to_numeric, errors="coerce").sum().sum()
|
| return f"${total:,.2f}"
|
| text_columns = [
|
| column
|
| for column in frame.columns
|
| if not pd.api.types.is_numeric_dtype(frame[column])
|
| ]
|
| if not text_columns:
|
| return None
|
| beverage = r"\b(?:drink|beverage|soda|cola|coffee|tea|juice|water|shake|smoothie|lemonade)\b"
|
| drink_mask = pd.Series(False, index=frame.index)
|
| for column in text_columns:
|
| drink_mask |= (
|
| frame[column]
|
| .astype(str)
|
| .str.contains(beverage, case=False, regex=True, na=False)
|
| )
|
| values = pd.to_numeric(frame[value_column], errors="coerce")
|
| if values.notna().sum() == 0:
|
| return None
|
| return f"${values[~drink_mask].sum():,.2f}"
|
|
|
