"""Spreadsheet tools (#153 A6) — read and mutate existing ``.xlsx`` workbooks. Complements :mod:`backend.tools.documents` (``create_xlsx`` creates a *new* file; here the assistant can read and edit one that already exists): * ``list_xlsx_sheets`` — READ, sheet names + dimensions; * ``xlsx_to_markdown`` — READ, bounded markdown table for the LLM context; * ``update_xlsx_cells`` — WRITE, batch cell edits (wraps the guarded service); * ``append_xlsx_rows`` — WRITE, append whole rows at the end of a sheet. Mutation tools go through :func:`backend.services.mutations.edit_xlsx_cells`, which already carries the #153 P0 guards: per-file lock, atomic replace, formula neutralisation (``allow_formula`` opt-in) and the lossy-write 409. """ from __future__ import annotations import logging from pathlib import Path from typing import Any from backend.services.errors import ServiceError from backend.services.paths import resolve_safe_path from backend.services.vaults import get_vault_root from backend.tools.context import ToolContext, ToolError, ToolRisk from backend.tools.registry import tool from backend.tools.schemas import ( AppendXlsxRowsInput, ListXlsxSheetsInput, UpdateXlsxCellsInput, XlsxToMarkdownInput, ) logger = logging.getLogger("obsigate.tools.spreadsheets") # xlsx_to_markdown ceiling: a workbook is a data dump, not prose. The table is # for the LLM context, so both axes are bounded (same spirit as A5's index cap). MAX_MD_ROWS = 100 MAX_MD_COLS = 20 MAX_MD_CHARS = 20_000 def _workbook_path(vault: str, path: str) -> Path: """Resolve and validate a vault-relative ``.xlsx`` path.""" path = (path or "").strip() if not path.lower().endswith(".xlsx"): raise ToolError("Extension attendue : .xlsx", code="invalid_arguments") try: root = get_vault_root(vault) except ServiceError as e: raise ToolError(e.message, code=e.code, details=e.details) from e return resolve_safe_path(root, path) def _map_service_error(e: ServiceError) -> ToolError: return ToolError(e.message, code=e.code, details=e.details) @tool( name="list_xlsx_sheets", description=( "List the sheets of an .xlsx workbook with their dimensions " "(rows x columns) and whether the display caps truncate them. " "Use before editing to pick the right sheet name." ), input_model=ListXlsxSheetsInput, risk=ToolRisk.READ, requires_vault=True, ) def list_xlsx_sheets(ctx: ToolContext, params: ListXlsxSheetsInput) -> dict[str, Any]: """Return sheet names and extents of the workbook.""" from backend.xlsx_reader import MAX_COLS, MAX_ROWS, _sheet_extent file_path = _workbook_path(params.vault, params.path) try: from openpyxl import load_workbook wb = load_workbook(str(file_path), read_only=True, data_only=True) except ServiceError as e: raise _map_service_error(e) from e except Exception as e: raise ToolError(f"Classeur illisible: {e}", code="invalid") from e try: sheets = [] for ws in wb.worksheets: total_rows, total_cols = _sheet_extent(ws) sheets.append( { "name": ws.title, "total_rows": total_rows, "total_cols": total_cols, "truncated": total_rows > MAX_ROWS or total_cols > MAX_COLS, } ) return {"vault": params.vault, "path": params.path, "sheets": sheets} finally: wb.close() @tool( name="xlsx_to_markdown", description=( "Read a sheet of an .xlsx workbook as a bounded markdown table " "(up to 100 rows x 20 columns). Use to inspect spreadsheet data " "before answering or editing." ), input_model=XlsxToMarkdownInput, risk=ToolRisk.READ, requires_vault=True, ) def xlsx_to_markdown(ctx: ToolContext, params: XlsxToMarkdownInput) -> dict[str, Any]: """Render one sheet as a markdown table for the LLM context.""" from openpyxl import load_workbook from backend.xlsx_reader import _fmt file_path = _workbook_path(params.vault, params.path) try: wb = load_workbook(str(file_path), read_only=True, data_only=True) except ServiceError as e: raise _map_service_error(e) from e except Exception as e: raise ToolError(f"Classeur illisible: {e}", code="invalid") from e try: if params.sheet: if params.sheet not in wb.sheetnames: raise ToolError( f"Feuille introuvable: {params.sheet}", code="not_found" ) ws = wb[params.sheet] else: ws = wb.active title = ws.title rows: list[list[str]] = [] truncated = False for row in ws.iter_rows( min_row=1, max_row=MAX_MD_ROWS, max_col=MAX_MD_COLS, values_only=True ): cells = [_fmt(v) for v in row] if not any(c.strip() for c in cells): continue rows.append(cells) # Real tail beyond the caps? Probe one row further. probe = list( ws.iter_rows( min_row=MAX_MD_ROWS + 1, max_row=MAX_MD_ROWS + 1, max_col=MAX_MD_COLS, values_only=True, ) ) if any(any(str(v or "").strip() for v in r) for r in probe): truncated = True finally: wb.close() lines: list[str] = [] if rows: header = rows[0] lines.append("| " + " | ".join(header) + " |") lines.append("|" + "|".join("---" for _ in header) + "|") for row in rows[1:]: lines.append("| " + " | ".join(row) + " |") table = "\n".join(lines)[:MAX_MD_CHARS] return { "vault": params.vault, "path": params.path, "sheet": title, "rows": len(rows), "cols": max((len(r) for r in rows), default=0), "truncated": truncated, "markdown": table, } @tool( name="update_xlsx_cells", description=( "Edit cells of an existing .xlsx workbook. ``cells`` maps A1 " "references to new values (max 500). A value starting with '=' or " "'@' is stored as TEXT unless allow_formula is set (DDE guard). " "Editing a workbook carrying features openpyxl cannot rewrite " "requires force=true (cached formula results, slicers…)." ), input_model=UpdateXlsxCellsInput, risk=ToolRisk.WRITE, requires_vault=True, ) def update_xlsx_cells(ctx: ToolContext, params: UpdateXlsxCellsInput) -> dict[str, Any]: """Wrap the guarded cell-edit service.""" from backend.services.mutations import edit_xlsx_cells if not params.cells: raise ToolError("Aucune cellule fournie", code="invalid_arguments") try: result = edit_xlsx_cells( params.vault, params.path, params.sheet, dict(params.cells), allow_formula=params.allow_formula, force=params.force, ) except ServiceError as e: raise _map_service_error(e) from e return { "status": "ok", "vault": result["vault"], "path": result["path"], "sheet": params.sheet, "cells": len(params.cells), } @tool( name="append_xlsx_rows", description=( "Append rows at the end of a sheet of an existing .xlsx workbook. " "Values are typed like in the viewer (numbers, TRUE/FALSE, FR dates " "JJ/MM/AAAA). The workbook is rewritten atomically with a backup." ), input_model=AppendXlsxRowsInput, risk=ToolRisk.WRITE, requires_vault=True, ) def append_xlsx_rows(ctx: ToolContext, params: AppendXlsxRowsInput) -> dict[str, Any]: """Append whole rows below the last used row of the sheet.""" from openpyxl import load_workbook from openpyxl.utils import get_column_letter from backend.services.mutations import _coerce_xlsx_value, edit_xlsx_cells if not params.rows: raise ToolError("Aucune ligne fournie", code="invalid_arguments") if len(params.rows) > 500: raise ToolError("Trop de lignes (max 500)", code="invalid_arguments") file_path = _workbook_path(params.vault, params.path) try: wb = load_workbook(str(file_path), read_only=True, data_only=True) try: if params.sheet not in wb.sheetnames: raise ToolError( f"Feuille introuvable: {params.sheet}", code="not_found" ) ws = wb[params.sheet] first_free = (ws.max_row or 0) + 1 finally: wb.close() except ServiceError as e: raise _map_service_error(e) from e except ToolError: raise except Exception as e: raise ToolError(f"Classeur illisible: {e}", code="invalid") from e cells: dict[str, Any] = {} for i, row in enumerate(params.rows): for j, value in enumerate(row): if value is None or (isinstance(value, str) and not value.strip()): continue ref = f"{get_column_letter(j + 1)}{first_free + i}" cells[ref] = _coerce_xlsx_value(value) if not cells: raise ToolError("Aucune valeur fournie", code="invalid_arguments") try: result = edit_xlsx_cells( params.vault, params.path, params.sheet, cells, allow_formula=params.allow_formula, force=params.force, ) except ServiceError as e: raise _map_service_error(e) from e return { "status": "ok", "vault": result["vault"], "path": result["path"], "sheet": params.sheet, "rows": len(params.rows), "first_row": first_free, }