spb/uqo-chat

1"""inspect_excel / edit_excel — read and modify an uploaded or generated workbook."""23from __future__ import annotations45import re6from typing import Any78from pydantic import BaseModel, Field910from app.llm.schemas import Artifact, ToolResult11from app.services import files as file_service12from app.tools import excel_ops13from app.tools.registry import ToolContext, registry141516class InspectArgs(BaseModel):17    file_id: str18    sheet: str | None = None19    max_rows: int = Field(60, ge=5, le=300)202122class EditArgs(BaseModel):23    file_id: str24    operations: list[dict[str, Any]] = Field(default_factory=list)25    filename: str | None = None262728async def _load(ctx: ToolContext, file_id: str) -> tuple[Any, bytes] | ToolResult:29    rec = await file_service.get_file(file_id)30    if not rec or rec.user_id != ctx.user_id:31        return ToolResult(content="Fichier introuvable. Utilise un file_id listé dans le contexte "32                          "(fichiers déposés ou classeurs déjà générés).", error=True)33    if file_service.ext_of(rec.filename) not in {"xlsx", "xlsm"}:34        return ToolResult(content=f"« {rec.filename} » n'est pas un classeur .xlsx.", error=True)35    try:36        return rec, file_service.read_bytes(rec)37    except FileNotFoundError:38        return ToolResult(content="Le fichier a expiré ; demande à l'étudiant de le redéposer.", error=True)394041async def inspect(args: dict[str, Any], ctx: ToolContext) -> ToolResult:42    loaded = await _load(ctx, args["file_id"])43    if isinstance(loaded, ToolResult):44        return loaded45    rec, data = loaded46    await ctx.report("running", f"Lecture de {rec.filename}…")47    info = excel_ops.inspect(data, args.get("sheet"), max_rows=args.get("max_rows", 60))48    parts = [f"Classeur « {rec.filename} » (file_id {rec.id}). Feuilles : "49             + ", ".join(s["name"] for s in info["sheets"])]50    for s in info["sheets"]:51        parts.append(f"\n## Feuille « {s['name']} » ({s['dims']}, {s['max_row']} lignes × {s['max_col']} col., "52                     f"{s['charts']} graphique(s))\n" + "\n".join(s["cells"][:200]))53    if info["defined_names"]:54        parts.append("Noms définis : " + ", ".join(info["defined_names"]))55    parts.append("\nLes formules sont affichées telles quelles (non calculées). Pour modifier : edit_excel "56                 "avec des opérations (set_cell, add_column avec formula '=E{r}*0.02', add_row, format, "57                 "add_sheet, add_chart…).")58    return ToolResult(content="\n".join(parts)[:40000],59                      payload={"filename": rec.filename, "file_id": rec.id, "preview": excel_ops.preview(data)},60                      meta={"summary": f"Lecture : {rec.filename}"})616263def _version_name(name: str) -> str:64    stem, ext = (name.rsplit(".", 1) + ["xlsx"])[:2]65    m = re.search(r"_v(\d+)$", stem)66    if m:67        return f"{stem[: m.start()]}_v{int(m.group(1)) + 1}.{ext}"68    return f"{stem}_v2.{ext}"697071async def edit(args: dict[str, Any], ctx: ToolContext) -> ToolResult:72    loaded = await _load(ctx, args["file_id"])73    if isinstance(loaded, ToolResult):74        return loaded75    rec, data = loaded76    ops = args.get("operations") or []77    if not ops:78        return ToolResult(content="Aucune opération fournie. Utilise inspect_excel pour lire, puis edit_excel "79                          "avec `operations`.", error=True)80    await ctx.report("running", f"Modification de {rec.filename} ({len(ops)} opération(s))…")81    try:82        new_data, log = excel_ops.apply_operations(data, ops)83    except excel_ops.ExcelOpError as exc:84        return ToolResult(content=f"Modification refusée : {exc}", error=True)85    filename = args.get("filename") or _version_name(rec.filename)86    if not filename.lower().endswith(".xlsx"):87        filename += ".xlsx"88    new_rec = await file_service.store_artifact(ctx.user_id, ctx.conversation_id, filename, new_data, ftype="xlsx")89    pv = excel_ops.preview(new_data)90    art = Artifact(type="xlsx", file_id=new_rec.id, filename=new_rec.filename, url=f"/api/v1/files/{new_rec.id}",91                   preview=pv)92    return ToolResult(93        content=f"Classeur modifié et enregistré sous « {new_rec.filename} » (file_id {new_rec.id}). Opérations :\n"94                + "\n".join(log) + "\nLe nouveau fichier est affiché à l'étudiant avec un bouton Télécharger ; "95                "l'original est conservé.",96        artifacts=[art],97        payload={"filename": new_rec.filename, "file_id": new_rec.id, "source": rec.filename,98                 "source_file_id": rec.id, "changes": log, "preview": pv},99        meta={"summary": f"Excel modifié : {new_rec.filename} ({len(log)} op.)"},100    )101102103registry.register("inspect_excel", inspect, InspectArgs)104registry.register("edit_excel", edit, EditArgs, heavy=True)105