"""inspect_excel / edit_excel — read and modify an uploaded or generated workbook.""" from __future__ import annotations import re from typing import Any from pydantic import BaseModel, Field from app.llm.schemas import Artifact, ToolResult from app.services import files as file_service from app.tools import excel_ops from app.tools.registry import ToolContext, registry class InspectArgs(BaseModel): file_id: str sheet: str | None = None max_rows: int = Field(60, ge=5, le=300) class EditArgs(BaseModel): file_id: str operations: list[dict[str, Any]] = Field(default_factory=list) filename: str | None = None async def _load(ctx: ToolContext, file_id: str) -> tuple[Any, bytes] | ToolResult: rec = await file_service.get_file(file_id) if not rec or rec.user_id != ctx.user_id: return ToolResult(content="Fichier introuvable. Utilise un file_id listé dans le contexte " "(fichiers déposés ou classeurs déjà générés).", error=True) if file_service.ext_of(rec.filename) not in {"xlsx", "xlsm"}: return ToolResult(content=f"« {rec.filename} » n'est pas un classeur .xlsx.", error=True) try: return rec, file_service.read_bytes(rec) except FileNotFoundError: return ToolResult(content="Le fichier a expiré ; demande à l'étudiant de le redéposer.", error=True) async def inspect(args: dict[str, Any], ctx: ToolContext) -> ToolResult: loaded = await _load(ctx, args["file_id"]) if isinstance(loaded, ToolResult): return loaded rec, data = loaded await ctx.report("running", f"Lecture de {rec.filename}…") info = excel_ops.inspect(data, args.get("sheet"), max_rows=args.get("max_rows", 60)) parts = [f"Classeur « {rec.filename} » (file_id {rec.id}). Feuilles : " + ", ".join(s["name"] for s in info["sheets"])] for s in info["sheets"]: parts.append(f"\n## Feuille « {s['name']} » ({s['dims']}, {s['max_row']} lignes × {s['max_col']} col., " f"{s['charts']} graphique(s))\n" + "\n".join(s["cells"][:200])) if info["defined_names"]: parts.append("Noms définis : " + ", ".join(info["defined_names"])) parts.append("\nLes formules sont affichées telles quelles (non calculées). Pour modifier : edit_excel " "avec des opérations (set_cell, add_column avec formula '=E{r}*0.02', add_row, format, " "add_sheet, add_chart…).") return ToolResult(content="\n".join(parts)[:40000], payload={"filename": rec.filename, "file_id": rec.id, "preview": excel_ops.preview(data)}, meta={"summary": f"Lecture : {rec.filename}"}) def _version_name(name: str) -> str: stem, ext = (name.rsplit(".", 1) + ["xlsx"])[:2] m = re.search(r"_v(\d+)$", stem) if m: return f"{stem[: m.start()]}_v{int(m.group(1)) + 1}.{ext}" return f"{stem}_v2.{ext}" async def edit(args: dict[str, Any], ctx: ToolContext) -> ToolResult: loaded = await _load(ctx, args["file_id"]) if isinstance(loaded, ToolResult): return loaded rec, data = loaded ops = args.get("operations") or [] if not ops: return ToolResult(content="Aucune opération fournie. Utilise inspect_excel pour lire, puis edit_excel " "avec `operations`.", error=True) await ctx.report("running", f"Modification de {rec.filename} ({len(ops)} opération(s))…") try: new_data, log = excel_ops.apply_operations(data, ops) except excel_ops.ExcelOpError as exc: return ToolResult(content=f"Modification refusée : {exc}", error=True) filename = args.get("filename") or _version_name(rec.filename) if not filename.lower().endswith(".xlsx"): filename += ".xlsx" new_rec = await file_service.store_artifact(ctx.user_id, ctx.conversation_id, filename, new_data, ftype="xlsx") pv = excel_ops.preview(new_data) art = Artifact(type="xlsx", file_id=new_rec.id, filename=new_rec.filename, url=f"/api/v1/files/{new_rec.id}", preview=pv) return ToolResult( content=f"Classeur modifié et enregistré sous « {new_rec.filename} » (file_id {new_rec.id}). Opérations :\n" + "\n".join(log) + "\nLe nouveau fichier est affiché à l'étudiant avec un bouton Télécharger ; " "l'original est conservé.", artifacts=[art], payload={"filename": new_rec.filename, "file_id": new_rec.id, "source": rec.filename, "source_file_id": rec.id, "changes": log, "preview": pv}, meta={"summary": f"Excel modifié : {new_rec.filename} ({len(log)} op.)"}, ) registry.register("inspect_excel", inspect, InspectArgs) registry.register("edit_excel", edit, EditArgs, heavy=True)