| 1 | from __future__ import annotations |
| 2 | |
| 3 | import json |
| 4 | from pathlib import Path |
| 5 | from typing import Any |
| 6 | |
| 7 | from helpers.tool import Response, Tool |
| 8 | from plugins._office.helpers import artifact_editor, document_store, libreoffice |
| 9 | |
| 10 | OFFICE_EXTENSIONS = document_store.OPEN_DOCUMENT_EXTENSIONS | document_store.OOXML_EXTENSIONS |
| 11 | |
| 12 | |
| 13 | class OfficeArtifact(Tool): |
| 14 | async def execute( |
| 15 | self, |
| 16 | action: str = "", |
| 17 | kind: str = "document", |
| 18 | title: str = "Untitled", |
| 19 | format: str = "", |
| 20 | content: str = "", |
| 21 | path: str = "", |
| 22 | file_id: str = "", |
| 23 | version_id: int | str | None = None, |
| 24 | operation: str = "", |
| 25 | find: str = "", |
| 26 | replace: str = "", |
| 27 | sheet: str = "", |
| 28 | cells: Any = None, |
| 29 | rows: Any = None, |
| 30 | chart: Any = None, |
| 31 | slides: Any = None, |
| 32 | max_chars: int | str = 12000, |
| 33 | open_in_canvas: bool = False, |
| 34 | open_in_desktop: bool = False, |
| 35 | **kwargs: Any, |
| 36 | ) -> Response: |
| 37 | action = str(action or "status").strip().lower().replace("-", "_") |
| 38 | open_in_canvas = _truthy( |
| 39 | open_in_canvas |
| 40 | or kwargs.get("open_canvas") |
| 41 | or kwargs.get("open_document") |
| 42 | ) |
| 43 | open_in_desktop = _truthy( |
| 44 | open_in_desktop |
| 45 | or kwargs.get("open_desktop") |
| 46 | or kwargs.get("desktop") |
| 47 | ) |
| 48 | try: |
| 49 | if action == "create": |
| 50 | fmt = _default_office_format(kind, format) |
| 51 | doc = document_store.create_document( |
| 52 | kind=kind, |
| 53 | title=title, |
| 54 | fmt=fmt, |
| 55 | content=content, |
| 56 | path=path, |
| 57 | context_id=self._context_id(), |
| 58 | ) |
| 59 | if doc["extension"] in {"odt", "ods", "odp"}: |
| 60 | validation = libreoffice.validate_odf(doc["path"]) |
| 61 | if not validation.get("ok"): |
| 62 | return Response( |
| 63 | message=f"{self.name} create failed: {validation.get('error')}", |
| 64 | break_loop=False, |
| 65 | ) |
| 66 | if doc["extension"] == "docx": |
| 67 | validation = libreoffice.validate_docx(doc["path"]) |
| 68 | if not validation.get("ok"): |
| 69 | return Response( |
| 70 | message=f"{self.name} create failed: {validation.get('error')}", |
| 71 | break_loop=False, |
| 72 | ) |
| 73 | return self._document_response( |
| 74 | "Created office artifact.", |
| 75 | doc, |
| 76 | action=action, |
| 77 | open_in_canvas=open_in_canvas, |
| 78 | open_in_desktop=open_in_desktop, |
| 79 | ) |
| 80 | if action == "open": |
| 81 | doc = self._document_from_input(file_id=file_id, path=path) |
| 82 | return self._document_response( |
| 83 | "Opened office artifact.", |
| 84 | doc, |
| 85 | action=action, |
| 86 | open_in_canvas=open_in_canvas, |
| 87 | open_in_desktop=open_in_desktop, |
| 88 | ) |
| 89 | if action in {"read", "extract"}: |
| 90 | doc = self._document_from_input(file_id=file_id, path=path) |
| 91 | payload = { |
| 92 | "ok": True, |
| 93 | "action": "read", |
| 94 | "document": self._public_doc(doc), |
| 95 | "content": artifact_editor.read_artifact(doc, max_chars=int(max_chars or 12000)), |
| 96 | } |
| 97 | return self._json_response( |
| 98 | payload, |
| 99 | doc=doc, |
| 100 | action="read", |
| 101 | open_in_canvas=open_in_canvas, |
| 102 | open_in_desktop=open_in_desktop, |
| 103 | ) |
| 104 | if action in {"edit", "update", "patch"}: |
| 105 | doc = self._document_from_input(file_id=file_id, path=path) |
| 106 | updated_doc, payload = artifact_editor.edit_artifact( |
| 107 | doc, |
| 108 | operation=operation, |
| 109 | content=content, |
| 110 | find=find, |
| 111 | replace=replace, |
| 112 | sheet=sheet, |
| 113 | cells=cells, |
| 114 | rows=rows, |
| 115 | chart=chart, |
| 116 | slides=slides, |
| 117 | **kwargs, |
| 118 | ) |
| 119 | payload["document"] = self._public_doc(updated_doc) |
| 120 | return self._json_response( |
| 121 | payload, |
| 122 | doc=updated_doc, |
| 123 | action="edit", |
| 124 | open_in_canvas=open_in_canvas, |
| 125 | open_in_desktop=open_in_desktop, |
| 126 | ) |
| 127 | if action == "inspect": |
| 128 | doc = self._document_from_input(file_id=file_id, path=path) |
| 129 | return self._json_response( |
| 130 | {"ok": True, "action": action, "document": self._public_doc(doc)}, |
| 131 | doc=doc, |
| 132 | action=action, |
| 133 | open_in_canvas=open_in_canvas, |
| 134 | open_in_desktop=open_in_desktop, |
| 135 | ) |
| 136 | if action == "version_history": |
| 137 | doc = self._document_from_input(file_id=file_id, path=path) |
| 138 | versions = document_store.version_history(doc["file_id"]) |
| 139 | return self._json_response( |
| 140 | {"ok": True, "action": action, "versions": versions}, |
| 141 | doc=doc, |
| 142 | action=action, |
| 143 | open_in_canvas=open_in_canvas, |
| 144 | open_in_desktop=open_in_desktop, |
| 145 | ) |
| 146 | if action == "restore_version": |
| 147 | if version_id is None or str(version_id).strip() == "": |
| 148 | return Response(message="version_id is required for restore_version.", break_loop=False) |
| 149 | doc = self._document_from_input(file_id=file_id, path=path) |
| 150 | restored = document_store.restore_version(doc["file_id"], int(version_id)) |
| 151 | _ensure_office_doc(restored) |
| 152 | return self._document_response( |
| 153 | "Restored office artifact version.", |
| 154 | restored, |
| 155 | action=action, |
| 156 | open_in_canvas=open_in_canvas, |
| 157 | open_in_desktop=open_in_desktop, |
| 158 | ) |
| 159 | if action == "export": |
| 160 | doc = self._document_from_input(file_id=file_id, path=path) |
| 161 | target_format = str(kwargs.get("target_format") or kwargs.get("export_format") or "").lower().lstrip(".") |
| 162 | if target_format and target_format != doc["extension"]: |
| 163 | _ensure_office_format(target_format, "target_format") |
| 164 | result = libreoffice.convert_document(doc["path"], target_format) |
| 165 | if result.get("ok"): |
| 166 | payload = { |
| 167 | "ok": True, |
| 168 | "action": action, |
| 169 | "path": document_store.display_path(result["path"]), |
| 170 | "document": self._public_doc(doc), |
| 171 | } |
| 172 | return self._json_response( |
| 173 | payload, |
| 174 | doc=doc, |
| 175 | action=action, |
| 176 | open_in_canvas=open_in_canvas, |
| 177 | open_in_desktop=open_in_desktop, |
| 178 | ) |
| 179 | return Response( |
| 180 | message=f"{self.name} export failed: {result.get('error')}", |
| 181 | break_loop=False, |
| 182 | additional=self._additional( |
| 183 | doc, |
| 184 | action=action, |
| 185 | open_in_canvas=open_in_canvas, |
| 186 | open_in_desktop=open_in_desktop, |
| 187 | ), |
| 188 | ) |
| 189 | return self._document_response( |
| 190 | "Office artifact export path is ready.", |
| 191 | doc, |
| 192 | action=action, |
| 193 | open_in_canvas=open_in_canvas, |
| 194 | open_in_desktop=open_in_desktop, |
| 195 | ) |
| 196 | if action == "status": |
| 197 | return self._json_response({"ok": True, "action": action, "status": libreoffice.collect_status()}, action=action) |
| 198 | return Response(message=f"Unknown {self.name} action: {action}", break_loop=False) |
| 199 | except Exception as exc: |
| 200 | return Response(message=f"{self.name} {action} failed: {exc}", break_loop=False) |
| 201 | |
| 202 | def get_log_object(self): |
| 203 | return self.agent.context.log.log( |
| 204 | type="tool", |
| 205 | heading=f"icon://description {self.agent.agent_name}: Using office artifact", |
| 206 | content="", |
| 207 | kvps={**self.args, "_tool_name": self.name}, |
| 208 | _tool_name=self.name, |
| 209 | ) |
| 210 | |
| 211 | def _document_from_input(self, file_id: str = "", path: str = "") -> dict[str, Any]: |
| 212 | if file_id: |
| 213 | return _ensure_office_doc(document_store.get_document(file_id)) |
| 214 | if path: |
| 215 | return _ensure_office_doc(document_store.register_document(path, context_id=self._context_id())) |
| 216 | raise ValueError("file_id or path is required") |
| 217 | |
| 218 | def _context_id(self) -> str: |
| 219 | return self.agent.context.id if self.agent and self.agent.context else "" |
| 220 | |
| 221 | def _document_response( |
| 222 | self, |
| 223 | message: str, |
| 224 | doc: dict[str, Any], |
| 225 | action: str = "", |
| 226 | *, |
| 227 | open_in_canvas: bool = False, |
| 228 | open_in_desktop: bool = False, |
| 229 | ) -> Response: |
| 230 | payload = {"ok": True, "action": action, "message": message, "document": self._public_doc(doc)} |
| 231 | return Response( |
| 232 | message=json.dumps(payload, indent=2, ensure_ascii=False), |
| 233 | break_loop=False, |
| 234 | additional=self._additional( |
| 235 | doc, |
| 236 | action=action, |
| 237 | open_in_canvas=open_in_canvas, |
| 238 | open_in_desktop=open_in_desktop, |
| 239 | ), |
| 240 | ) |
| 241 | |
| 242 | def _json_response( |
| 243 | self, |
| 244 | payload: dict[str, Any], |
| 245 | doc: dict[str, Any] | None = None, |
| 246 | action: str = "", |
| 247 | *, |
| 248 | open_in_canvas: bool = False, |
| 249 | open_in_desktop: bool = False, |
| 250 | ) -> Response: |
| 251 | return Response( |
| 252 | message=json.dumps(payload, indent=2, ensure_ascii=False, default=str), |
| 253 | break_loop=False, |
| 254 | additional=self._additional( |
| 255 | doc, |
| 256 | action=action, |
| 257 | open_in_canvas=open_in_canvas, |
| 258 | open_in_desktop=open_in_desktop, |
| 259 | ) if doc else { |
| 260 | "_tool_name": self.name, |
| 261 | "canvas_surface": "desktop", |
| 262 | "action": action, |
| 263 | "open_in_canvas": bool(open_in_canvas), |
| 264 | "open_in_desktop": bool(open_in_desktop), |
| 265 | }, |
| 266 | ) |
| 267 | |
| 268 | def _additional( |
| 269 | self, |
| 270 | doc: dict[str, Any] | None, |
| 271 | action: str = "", |
| 272 | *, |
| 273 | open_in_canvas: bool = False, |
| 274 | open_in_desktop: bool = False, |
| 275 | ) -> dict[str, Any]: |
| 276 | if not doc: |
| 277 | return { |
| 278 | "_tool_name": self.name, |
| 279 | "canvas_surface": "desktop", |
| 280 | "action": action, |
| 281 | "open_in_canvas": bool(open_in_canvas), |
| 282 | "open_in_desktop": bool(open_in_desktop), |
| 283 | } |
| 284 | return { |
| 285 | "_tool_name": self.name, |
| 286 | "canvas_surface": "desktop", |
| 287 | "action": action, |
| 288 | "open_in_canvas": bool(open_in_canvas), |
| 289 | "open_in_desktop": bool(open_in_desktop), |
| 290 | "file_id": doc["file_id"], |
| 291 | "title": doc["basename"], |
| 292 | "format": doc["extension"], |
| 293 | "path": document_store.display_path(doc["path"]), |
| 294 | "version": document_store.item_version(doc), |
| 295 | } |
| 296 | |
| 297 | def _public_doc(self, doc: dict[str, Any]) -> dict[str, Any]: |
| 298 | return { |
| 299 | "file_id": doc["file_id"], |
| 300 | "path": document_store.display_path(doc["path"]), |
| 301 | "basename": doc["basename"], |
| 302 | "extension": doc["extension"], |
| 303 | "size": doc["size"], |
| 304 | "version": document_store.item_version(doc), |
| 305 | "last_modified": doc["last_modified"], |
| 306 | "exists": Path(doc["path"]).exists(), |
| 307 | } |
| 308 | |
| 309 | |
| 310 | def _truthy(value: Any) -> bool: |
| 311 | if isinstance(value, bool): |
| 312 | return value |
| 313 | if value is None: |
| 314 | return False |
| 315 | if isinstance(value, (int, float)): |
| 316 | return value != 0 |
| 317 | return str(value).strip().lower() in {"1", "true", "yes", "y", "on"} |
| 318 | |
| 319 | |
| 320 | def _default_office_format(kind: str, fmt: str = "") -> str: |
| 321 | normalized = str(fmt or "").strip().lower().lstrip(".") |
| 322 | if not normalized: |
| 323 | normalized_kind = str(kind or "").strip().lower() |
| 324 | if normalized_kind in {"spreadsheet", "sheet", "calc"}: |
| 325 | normalized = "ods" |
| 326 | elif normalized_kind in {"presentation", "slides", "deck", "impress"}: |
| 327 | normalized = "odp" |
| 328 | else: |
| 329 | normalized = "odt" |
| 330 | return _ensure_office_format(normalized, "format") |
| 331 | |
| 332 | |
| 333 | def _ensure_office_format(fmt: str, label: str = "format") -> str: |
| 334 | normalized = str(fmt or "").strip().lower().lstrip(".") |
| 335 | if normalized not in OFFICE_EXTENSIONS: |
| 336 | raise ValueError( |
| 337 | f"{label} must be an Office format ({', '.join(sorted(OFFICE_EXTENSIONS))}); " |
| 338 | "use text_editor for Markdown and plain text files." |
| 339 | ) |
| 340 | return normalized |
| 341 | |
| 342 | |
| 343 | def _ensure_office_doc(doc: dict[str, Any]) -> dict[str, Any]: |
| 344 | _ensure_office_format(str(doc.get("extension") or ""), "document extension") |
| 345 | return doc |