"""CRUD de documents de Metabase (feature reciente, Metabase 0.57+). Un document es un texto editable en Metabase (tipo Notion) serializado como ProseMirror JSON (content_type: application/json+vnd.prose-mirror). Permite embeber cards, smart links y flex containers. Nodos ProseMirror soportados (observados en Metabase v0.59): - doc, paragraph, heading (attrs.level 1-6), text - bulletList, orderedList, listItem - blockquote, codeBlock (attrs.language), horizontalRule, hardBreak - cardEmbed (attrs.id — card_id), smartLink (attrs.entityId), flexContainer (attrs.columnWidths), resizeNode, mention - image, iframe, table/tableRow/tableCell, callout, taskList/taskItem, details Marks: - bold, italic, strike, code, link (attrs.href), underline, highlight, subscript, textStyle """ from .client import MetabaseClient def metabase_list_documents(client: MetabaseClient) -> list[dict]: """Lista documents de Metabase. Endpoint: GET /api/document. Retorna `{"items": [...]}` segun la API. Args: client: Cliente autenticado. Returns: Lista de dicts con: id, name, collection_id, archived, created_at, updated_at, creator_id, content_type, entity_id. Example: >>> docs = metabase_list_documents(client) >>> for d in docs: ... print(d["id"], d["name"]) """ resp = client.request("GET", "/api/document") if isinstance(resp, dict) and "items" in resp: return resp["items"] return resp or [] def metabase_get_document(client: MetabaseClient, document_id: int) -> dict: """Obtiene un document completo con su contenido ProseMirror. Endpoint: GET /api/document/:id. Args: client: Cliente autenticado. document_id: ID del document. Returns: Dict con: id, name, document (ProseMirror tree), collection_id, archived, creator, created_at, updated_at, entity_id, content_type. Example: >>> doc = metabase_get_document(client, 1) >>> print(doc["name"]) >>> tree = doc["document"] # {"type": "doc", "content": [...]} """ return client.request("GET", f"/api/document/{document_id}") def metabase_create_document( client: MetabaseClient, name: str, document: dict, collection_id: int = 0, ) -> dict: """Crea un document nuevo. Endpoint: POST /api/document. Args: client: Cliente autenticado. name: Titulo del document (1-254 chars, no blank). document: Arbol ProseMirror JSON. Formato minimo: {"type": "doc", "content": [{"type": "paragraph", "content": [...]}]} O cadena vacia "" si se quiere arrancar en blanco. collection_id: ID de coleccion destino. 0 = root. Returns: Document creado con su id asignado. Example: >>> doc = metabase_create_document(client, "Notas", { ... "type": "doc", ... "content": [{ ... "type": "paragraph", ... "content": [{"type": "text", "text": "Hola mundo"}] ... }] ... }) """ body: dict = {"name": name, "document": document} if collection_id > 0: body["collection_id"] = collection_id return client.request("POST", "/api/document", json=body) def metabase_update_document( client: MetabaseClient, document_id: int, **fields, ) -> dict: """Actualiza un document. Solo se envian los campos pasados. Endpoint: PUT /api/document/:id. Campos tipicos: name, document, collection_id, archived. Args: client: Cliente autenticado. document_id: ID del document a actualizar. **fields: Campos a modificar. Returns: Document actualizado. Example: >>> metabase_update_document(client, 1, name="Nuevo titulo") >>> metabase_update_document(client, 1, document={"type":"doc","content":[...]}) """ return client.request("PUT", f"/api/document/{document_id}", json=fields) def metabase_archive_document(client: MetabaseClient, document_id: int) -> dict: """Archiva un document (equivalente a PUT con archived=True). Metabase exige archive previo para poder eliminar. Args: client: Cliente autenticado. document_id: ID del document. Returns: Document con archived=True. """ return client.request("PUT", f"/api/document/{document_id}", json={"archived": True}) def metabase_list_document_comments( client: MetabaseClient, document_id: int, *, include_resolved: bool = True, include_deleted: bool = False, ) -> list[dict]: """Lista los comentarios de un document. Endpoint: GET /api/comment?target_type=document&target_id=:id. Un comentario puede estar anclado a un bloque especifico via `child_target_id` (UUID que matchea `attrs._id` de un parrafo) o al document entero si `child_target_id` es null. Las respuestas tipo thread cuelgan via `parent_comment_id`. Args: client: Cliente autenticado. document_id: ID del document. include_resolved: si False, filtra comentarios con is_resolved=True. include_deleted: si False, filtra comentarios con deleted_at no null. Returns: Lista de dicts con: id, content (arbol ProseMirror), creator, creator_id, target_type, target_id, child_target_id, parent_comment_id, is_resolved, deleted_at, reactions, created_at, updated_at. Example: >>> comments = metabase_list_document_comments(client, 29) >>> for c in comments: ... text = _comment_plaintext(c["content"]) ... print(f'{c["creator"]["common_name"]}: {text}') """ resp = client.request( "GET", "/api/comment", params={"target_type": "document", "target_id": document_id}, ) items = resp.get("comments", []) if isinstance(resp, dict) else [] if not include_resolved: items = [c for c in items if not c.get("is_resolved")] if not include_deleted: items = [c for c in items if c.get("deleted_at") is None] return items def metabase_create_document_comment( client: MetabaseClient, document_id: int, content: dict, *, child_target_id: str | None = None, parent_comment_id: int | None = None, ) -> dict: """Crea un comentario en un document. Endpoint: POST /api/comment. Args: client: Cliente autenticado. document_id: ID del document donde comentar. content: Arbol ProseMirror del comentario. Ejemplo minimo: {"type":"doc","content":[{"type":"paragraph","content":[ {"type":"text","text":"mi comentario"}]}]} child_target_id: UUID del bloque del document al que se ancla el comentario (matchea `attrs._id` de un parrafo). None = a nivel doc. parent_comment_id: id del comentario al que se responde. None = top-level. Returns: Comentario creado. """ body: dict = { "target_type": "document", "target_id": document_id, "content": content, } if child_target_id is not None: body["child_target_id"] = child_target_id if parent_comment_id is not None: body["parent_comment_id"] = parent_comment_id return client.request("POST", "/api/comment", json=body) def metabase_resolve_document_comment(client: MetabaseClient, comment_id: int) -> dict: """Marca un comentario como resuelto (is_resolved=True). Endpoint: PUT /api/comment/:id. """ return client.request("PUT", f"/api/comment/{comment_id}", json={"is_resolved": True}) def metabase_delete_document(client: MetabaseClient, document_id: int) -> None: """Elimina un document. REQUIERE archivarlo primero. Endpoint: DELETE /api/document/:id. Si el document no esta archivado, Metabase responde: "Document must be archived before it can be deleted." Args: client: Cliente autenticado. document_id: ID del document a eliminar. """ client.request("DELETE", f"/api/document/{document_id}") def metabase_move_document( client: MetabaseClient, document_id: int, collection_id: int | None, ) -> dict: """Mueve un document a otra coleccion. Thin wrapper sobre metabase_update_document: envia solo collection_id. Endpoint: PUT /api/document/:id con {collection_id: ...}. Args: client: Cliente autenticado. document_id: ID del document a mover. collection_id: ID de coleccion destino. None mueve a la raiz ("Our analytics"). Returns: Document actualizado con el nuevo collection_id. Example: >>> doc = metabase_move_document(client, 42, collection_id=7) >>> print(doc["collection_id"]) # 7 >>> # Mover a raiz: >>> doc = metabase_move_document(client, 42, collection_id=None) """ return client.request( "PUT", f"/api/document/{document_id}", json={"collection_id": collection_id}, ) def metabase_copy_document( client: MetabaseClient, document_id: int, name: str | None = None, collection_id: int | None = None, ) -> dict: """Copia un document (Metabase no tiene endpoint nativo). Obtiene el document original con metabase_get_document, luego crea uno nuevo con metabase_create_document clonando el contenido ProseMirror. Si name=None usa "{original_name} (copia)". Si collection_id=None copia a la misma coleccion del original. Realiza 2 requests HTTP: GET /api/document/:id + POST /api/document. Args: client: Cliente autenticado. document_id: ID del document a copiar. name: Nombre del nuevo document. None = "{original} (copia)". collection_id: Coleccion destino. None = misma coleccion del original. Returns: Document nuevo recien creado con su id asignado. Example: >>> copy = metabase_copy_document(client, 42) >>> print(copy["name"]) # "Mi documento (copia)" >>> print(copy["id"]) # nuevo ID >>> # Clonar a otra coleccion con nombre personalizado: >>> copy = metabase_copy_document(client, 42, name="Backup Q1", collection_id=5) """ original = metabase_get_document(client, document_id) new_name = name if name is not None else f"{original['name']} (copia)" dest_collection = collection_id if collection_id is not None else original.get("collection_id", 0) doc_content = original.get("document", "") body: dict = {"name": new_name, "document": doc_content} if dest_collection: body["collection_id"] = dest_collection return client.request("POST", "/api/document", json=body)