diff --git a/bash/functions/shell/save_onlyoffice_file.md b/bash/functions/shell/save_onlyoffice_file.md new file mode 100644 index 00000000..37e78804 --- /dev/null +++ b/bash/functions/shell/save_onlyoffice_file.md @@ -0,0 +1,90 @@ +--- +name: save_onlyoffice_file +kind: function +lang: bash +domain: shell +purity: impure +version: 1.1.0 +description: "Fuerza el guardado (Ctrl+S) de un documento abierto en una instancia de OnlyOffice Desktop en Linux/X11 y confirma que llego a disco por cambio de mtime. Primer paso del flujo seguro guardar -> actualizar -> recargar; evita perder cambios no guardados cuando un build regenera el archivo leyendo del disco." +signature: "save_onlyoffice_file(file_path: string, [instance: string]) -> json" +error_type: error_go_core +tags: [onlyoffice, desktop, x11, gui, save, persist] +uses_functions: [] +uses_types: [] +file_path: bash/functions/shell/save_onlyoffice_file.sh +params: + - name: file_path + desc: "ruta al documento abierto en OnlyOffice cuyo guardado se quiere forzar. Debe existir. Se normaliza a ruta absoluta y se usa su basename para localizar la ventana." + - name: instance + desc: "nombre del slot/instancia para etiquetar la salida JSON (default: 'demo'). Usar el MISMO valor que en open/reload/close del mismo documento por coherencia." +output: "linea JSON a stdout: {\"instance\":\"\",\"file\":\"\",\"wid\":\"|null\",\"status\":\"saved\"|\"no_change\"|\"no_window\",\"dialog_confirmed\":0|1[,\"mtime_before\":N,\"mtime_after\":N]}. dialog_confirmed=1 si se envio Return para cerrar el dialogo modal de formato. Exit 0 salvo error de dependencia o archivo inexistente (exit 1)." +--- + +Fuerza el guardado (Ctrl+S) de un documento abierto en una instancia de ONLYOFFICE +Desktop Editors en Linux/X11 y confirma que el guardado llegó a disco observando el +cambio de `mtime` del archivo. + +Existe para cerrar una ventana de pérdida de datos: OnlyOffice mantiene los cambios +en memoria hasta que el usuario guarda. Cualquier proceso que regenere el archivo +leyendo del disco (un build que refresca hojas, un script de sincronización) +perdería el trabajo manual no guardado. Esta función vuelca ese trabajo a disco +ANTES de tocar el archivo, de modo que el paso de actualización pueda preservarlo. + +Es el primer paso del flujo seguro de refresco: + +``` +save_onlyoffice_file -> (actualizar el archivo en disco) -> reload_onlyoffice_file +``` + +## Ejemplo + +```bash +# Forzar el guardado de un xlsx abierto en la instancia "afiliados" +bash bash/functions/shell/save_onlyoffice_file.sh \ + /home/enmanuel/afiliados/programas_afiliados.xlsx afiliados +# {"instance":"afiliados","file":"/home/enmanuel/afiliados/programas_afiliados.xlsx","wid":"0x0a20002a","status":"saved","mtime_before":1718380000,"mtime_after":1718380042} + +# Via fn run (tras fn index) +./fn run save_onlyoffice_file /home/enmanuel/afiliados/programas_afiliados.xlsx afiliados + +# Encadenado con la actualización y la recarga (flujo seguro completo) +bash bash/functions/shell/save_onlyoffice_file.sh "$XLSX" afiliados +python build_xlsx.py # regenera solo las hojas gestionadas +./fn run reload_onlyoffice_file "$XLSX" afiliados +``` + +## Cuando usarla + +Llámala SIEMPRE justo antes de regenerar o modificar en disco un archivo que el +usuario pueda tener abierto en OnlyOffice, para no pisar sus cambios sin guardar. +Es el primer eslabón del flujo guardar -> actualizar -> recargar. Si no hay ventana +abierta para ese archivo, es un no-op seguro (status `no_window`). + +## Gotchas + +- **Orden crítico**: guarda ANTES de actualizar el archivo. Si actualizas primero y + guardas OnlyOffice después, OnlyOffice sobrescribe tu actualización con su copia + en memoria (vieja). El flujo correcto es save -> update -> reload. +- **status `no_change`**: el `mtime` no cambió. Normalmente significa que no había + cambios pendientes (no es un error). +- **Auto-confirmación del diálogo de formato (v1.1.0)**: si tras Ctrl+S el guardado no + se completa en ~1.2s, la función asume que OnlyOffice mostró un diálogo modal + ("mantener formato") y le envía Return, que acepta la opción por defecto (mantener el + formato actual). El campo `dialog_confirmed` indica si se envió. Si no había diálogo, + el Return va al editor y solo mueve de celda (no altera datos). Para suprimir el + diálogo de forma permanente, desmárcalo en OnlyOffice: Configuración avanzada → + desactivar el aviso de formato al guardar. +- **status `no_window`**: no hay ninguna ventana cuyo título contenga el basename del + archivo. No hay nada que guardar; el disco ya es la única fuente de verdad. +- **Detección por basename**: dos archivos con el mismo nombre en rutas distintas + colisionan al localizar la ventana (igual que open/reload). +- **X11 obligatorio**: depende de `xdotool` (y `stat` de coreutils). No funciona en + Wayland puro sin XWayland. +- **Foco**: la función activa la ventana (`windowactivate --sync`) para que Ctrl+S + llegue al editor. Roba el foco un instante; es esperable. + +## Capability growth log + +- v1.1.0 (2026-06-15) — auto-confirma el diálogo modal "mantener formato" enviando + Return a la ventana activa cuando el guardado no se completa en ~1.2s; añade el campo + `dialog_confirmed` a la salida JSON. diff --git a/bash/functions/shell/save_onlyoffice_file.sh b/bash/functions/shell/save_onlyoffice_file.sh new file mode 100644 index 00000000..f5083fc7 --- /dev/null +++ b/bash/functions/shell/save_onlyoffice_file.sh @@ -0,0 +1,107 @@ +#!/usr/bin/env bash +# save_onlyoffice_file — fuerza el guardado (Ctrl+S) de un documento abierto en una +# instancia de ONLYOFFICE Desktop Editors en Linux/X11 y confirma que el archivo se +# escribio a disco observando el cambio de mtime. +# +# Para que existe: OnlyOffice mantiene los cambios en memoria hasta que el usuario +# guarda. Cualquier proceso que regenere el .xlsx leyendo del disco (por ejemplo un +# build que refresca hojas) perderia el trabajo manual no guardado. Esta funcion +# vuelca ese trabajo a disco ANTES de tocar el archivo, de modo que el paso de +# actualizacion pueda preservarlo. Es el primer paso del flujo seguro: +# save_onlyoffice_file -> (actualizar el archivo) -> reload_onlyoffice_file +# +# La ventana se localiza por el basename del archivo (OnlyOffice titula la ventana +# " — ONLYOFFICE"), igual que open_onlyoffice_file. Si no hay ventana +# abierta para ese basename no hay nada que guardar: se devuelve status "no_window" +# con exit 0 (el disco ya es la unica fuente de verdad). +# +# Funcion impura: envia eventos de teclado a X11 (xdotool) y lee el estado del +# sistema de archivos. Imprime una linea JSON con el resultado a stdout. +# +# No usamos `set -e`: los pipelines de busqueda de ventanas (xdotool|head) pueden no +# matchear y no deben abortar el script. Mantenemos -u y pipefail con guardas. +set -uo pipefail + +save_onlyoffice_file() { + local file_path="${1:-}" + local instance="${2:-demo}" + + # --- 1. Validacion de dependencias del sistema --- + local dep + for dep in xdotool stat; do + if ! command -v "$dep" >/dev/null 2>&1; then + echo "error: dependencia ausente: '$dep' (instala xdotool, coreutils)" >&2 + return 1 + fi + done + + # --- 2. Validacion de argumentos --- + if [ -z "$file_path" ]; then + echo "error: uso: save_onlyoffice_file [instance]" >&2 + return 1 + fi + if [ ! -f "$file_path" ]; then + echo "error: el archivo no existe: '$file_path'" >&2 + return 1 + fi + local abs_path + abs_path="$(cd "$(dirname "$file_path")" && pwd)/$(basename "$file_path")" + local base + base="$(basename "$abs_path")" + + # --- 3. Localizar la ventana de OnlyOffice por basename --- + local wid="" + wid="$(xdotool search --name "$base" 2>/dev/null | head -1 || true)" + if [ -z "$wid" ]; then + printf '{"instance":"%s","file":"%s","wid":null,"status":"no_window"}\n' \ + "$instance" "$abs_path" + return 0 + fi + local hex + hex="$(printf '0x%08x' "$wid" 2>/dev/null || echo "$wid")" + + # --- 4. mtime antes de guardar --- + local mtime_before + mtime_before="$(stat -c %Y "$abs_path" 2>/dev/null || echo 0)" + + # --- 5. Enfocar la ventana y enviar Ctrl+S --- + xdotool windowactivate --sync "$wid" >/dev/null 2>&1 || true + xdotool key --clearmodifiers --window "$wid" ctrl+s >/dev/null 2>&1 || true + + # --- 6. Esperar el guardado; auto-confirmar el dialogo de formato si aparece --- + # OnlyOffice puede mostrar un dialogo modal ("mantener formato") al guardar. Si el + # mtime no cambia en ~1.2s asumimos que hay un modal esperando y le enviamos Return: + # acepta la opcion por defecto, que es mantener el formato actual del archivo. Si no + # habia dialogo, el Return va al editor y solo mueve de celda (inofensivo: no altera + # datos). El intento se repite mientras el guardado no se confirme. + local mtime_after="$mtime_before" i=0 confirmed=0 + local max=27 # ~8s a 0.3s por iteracion + until [ "$mtime_after" -gt "$mtime_before" ] || [ "$i" -ge "$max" ]; do + read -r -t 0.3 _ /dev/null || true + mtime_after="$(stat -c %Y "$abs_path" 2>/dev/null || echo "$mtime_before")" + i=$((i + 1)) + # A partir de ~1.2s sin guardar, confirmar el dialogo modal con Return. + if [ "$i" -ge 4 ] && [ "$mtime_after" -le "$mtime_before" ]; then + local dlg + dlg="$(xdotool getactivewindow 2>/dev/null || true)" + if [ -n "$dlg" ]; then + xdotool key --clearmodifiers --window "$dlg" Return >/dev/null 2>&1 || true + confirmed=1 + fi + fi + done + + local status="saved" + if [ "$mtime_after" -le "$mtime_before" ]; then + # Sin cambio de mtime: no habia nada pendiente que guardar. + status="no_change" + fi + printf '{"instance":"%s","file":"%s","wid":"%s","status":"%s","dialog_confirmed":%s,"mtime_before":%s,"mtime_after":%s}\n' \ + "$instance" "$abs_path" "$hex" "$status" "$confirmed" "$mtime_before" "$mtime_after" + return 0 +} + +# Ejecutable directo: `bash save_onlyoffice_file.sh [instance]`. +if [ "${BASH_SOURCE[0]}" = "${0}" ]; then + save_onlyoffice_file "$@" +fi diff --git a/docs/capabilities/obsidian.md b/docs/capabilities/obsidian.md index 7189e030..34e21803 100644 --- a/docs/capabilities/obsidian.md +++ b/docs/capabilities/obsidian.md @@ -1,6 +1,6 @@ # Capability: obsidian -CRUD headless de vaults y notas de Obsidian, tratadas como Markdown plano con frontmatter YAML y wikilinks `[[...]]`. NO depende de la app GUI de Obsidian ni de su URI scheme — manipula los archivos `.md` directamente en disco. Scriptable, rapido, con telemetria del registry. +CRUD headless de vaults y notas de Obsidian, tratadas como Markdown plano con frontmatter YAML y wikilinks `[[...]]`. El nucleo del grupo manipula los archivos `.md` directamente en disco (no necesita la app GUI). Un sub-conjunto aparte gestiona la **lista de vaults que la app de escritorio Obsidian conoce** (su config `~/.config/obsidian/obsidian.json` + el URI scheme `obsidian://`): `register_*`, `list_registered_*`, `unregister_*`, `open_obsidian_vault`. Scriptable, rapido, con telemetria del registry. Los vaults de Obsidian del usuario viven en `/home/enmanuel/Obsidian/` y estan enlazados como vaults del registry en el project `obsidian` (`projects/obsidian/vaults/`). Ver `projects/obsidian/project.md`. @@ -19,6 +19,10 @@ Los vaults de Obsidian del usuario viven en `/home/enmanuel/Obsidian/` y estan e | `search_obsidian_notes_py_obsidian` | `search_obsidian_notes(vault_dir, query, in_body=True, in_frontmatter=True) -> list` | Busca substring (case-insensitive) en las notas. Devuelve `[{path, matches:[{line, text}]}]`. | | `list_obsidian_vaults_py_obsidian` | `list_obsidian_vaults(base_dir: str) -> list` | Lista los vaults (subdirs con `.obsidian/`) bajo `base_dir`. `[{name, path}]`. | | `create_obsidian_vault_py_obsidian` | `create_obsidian_vault(parent_dir, name) -> str` | Crea un vault nuevo: carpeta + `.obsidian/app.json` minimo. Error si ya existe. | +| `register_obsidian_vault_py_obsidian` | `register_obsidian_vault(vault_path, open=False, config_path="") -> dict` | Da de alta un vault en la **app** Obsidian (entrada en `~/.config/obsidian/obsidian.json`). Idempotente por path, backup `.bak`, preserva el resto del JSON. NO toca el filesystem del vault. | +| `list_registered_obsidian_vaults_py_obsidian` | `list_registered_obsidian_vaults(config_path="") -> list` | Lista los vaults que la **app** Obsidian conoce (de `obsidian.json`), ordenados por path. `[{id, path, open, ts}]`. Distinto de `list_obsidian_vaults` (que escanea el filesystem). | +| `unregister_obsidian_vault_py_obsidian` | `unregister_obsidian_vault(vault_ref, config_path="") -> dict` | Quita un vault de la lista de la **app** Obsidian (por id o por path). NO borra la carpeta del vault. Backup `.bak`, preserva el resto del JSON. | +| `open_obsidian_vault_py_obsidian` | `open_obsidian_vault(vault, register_if_missing=True, launch=True, config_path="") -> dict` | Abre un vault en la **app** Obsidian via `obsidian://open?vault=` (lanza `xdg-open`). Registra el vault antes si falta. `launch=False` solo construye el URI. | | `slugify_obsidian_name_py_obsidian` | `slugify_obsidian_name(name: str) -> str` | **Pure.** Nombre/titulo -> slug kebab-case estable (translitera acentos, ñ->n). Estabiliza ids de nodo y nombres de archivo. | | `extract_obsidian_embeds_py_obsidian` | `extract_obsidian_embeds(body: str) -> list` | **Pure.** Solo los embeds `![[...]]` (attachments incrustados), ignorando wikilinks normales. Dedup preservando orden. | | `resolve_obsidian_embed_py_obsidian` | `resolve_obsidian_embed(vault_dir, embed_name) -> str` | Resuelve un embed `![[foto.jpg]]` a su path absoluto real (busca por basename unico en el vault). Cadena vacia si no existe. | @@ -74,7 +78,8 @@ Para una sola operacion con un id conocido, `fn run` tambien sirve: ## Fronteras (que NO cubre) -- **No habla con la app GUI** (no usa el URI scheme `obsidian://`, no abre notas en la interfaz, no dispara plugins). Si la app esta abierta, escribir en disco puede chocar con sus locks/cache — cerrar la app o refrescar manualmente. +- **El CRUD de notas no habla con la app GUI** (no abre notas en la interfaz ni dispara plugins). Si la app esta abierta, escribir en disco puede chocar con sus locks/cache — cerrar la app o refrescar manualmente. La unica interaccion con la app es la **gestion de su lista de vaults** (`register_*`/`unregister_*`/`list_registered_*` sobre `obsidian.json`) y `open_obsidian_vault` (lanza el URI `obsidian://`); estas no editan notas ni renderizan nada. +- **Single-instance gotcha**: Obsidian cachea su `obsidian.json` en memoria al arrancar. Registrar/desregistrar un vault con la app abierta no se reflejara hasta reiniciarla; `open_obsidian_vault` sobre un vault recien registrado puede dar "unable to find a vault" hasta el reinicio. - **No resuelve wikilinks a paths** automaticamente (devuelve los targets crudos). Resolver `[[nota]]` -> archivo real es responsabilidad del caller (busqueda por nombre en el vault). - **No renderiza Markdown** ni evalua Dataview/templating. Trata las notas como texto + frontmatter. - **El grafo agregado** del vault ya lo cubre `build_obsidian_graph_py_obsidian` (nodos tipados + aristas con `kind` + nodos fantasma `dangling`). Es la base de la vista grafo (sigma.js) de la app `osint_web`. Lo que sigue fuera del grupo es el *layout* visual del grafo (force-directed) — eso vive en el frontend. diff --git a/python/functions/infra/upsert_xlsx_sheet.md b/python/functions/infra/upsert_xlsx_sheet.md new file mode 100644 index 00000000..6bd091f4 --- /dev/null +++ b/python/functions/infra/upsert_xlsx_sheet.md @@ -0,0 +1,100 @@ +--- +name: upsert_xlsx_sheet +kind: function +lang: py +domain: infra +version: "1.0.0" +purity: impure +signature: "def upsert_xlsx_sheet(xlsx_path: str, sheet_name: str, records: list[dict], columns: list[str], key_col: str = \"\", preserve_cols: list[str] | None = None, formulas: dict | None = None, backup: bool = True, freeze: str = \"A2\", autofilter: bool = True) -> dict" +description: "Actualiza de forma NO DESTRUCTIVA una hoja concreta de un archivo .xlsx con openpyxl. Reescribe SOLO la hoja indicada (sheet_name) y conserva intactas las demas hojas del libro. Antes de limpiar la hoja gestionada lee, por una columna clave (key_col), los valores de las columnas de trabajo manual (preserve_cols) y los reescribe ganando sobre los datos nuevos. Cabecera estilizada (negrita, relleno, texto blanco, borde, centrado), freeze_panes, autofilter, auto-ancho de columnas, formulas por columna con placeholders {row} y {NombreColumna}, y backup .bak opcional. Devuelve un resumen con filas escritas, hojas conservadas y celdas manuales preservadas." +tags: [xlsx, openpyxl, spreadsheet, office, onlyoffice, infra] +uses_functions: [] +uses_types: [] +returns: [] +returns_optional: false +error_type: "error_go_core" +imports: [openpyxl] +params: + - name: xlsx_path + desc: "Ruta del archivo .xlsx. Si existe se abre con openpyxl (y se respaldan las demas hojas); si no existe se crea un libro nuevo eliminando la hoja por defecto vacia." + - name: sheet_name + desc: "Nombre de la hoja a (re)escribir. Es la UNICA hoja que la funcion toca; el resto del libro queda intacto. Si la hoja no existe, se crea." + - name: records + desc: "Lista de dicts; cada dict es una fila y sus claves son nombres de columna. Se escribe una fila por record en el orden definido por columns." + - name: columns + desc: "Orden canonico de columnas (nombres de cabecera). Define que columnas se escriben y en que orden. Una clave de un record que no este en columns se ignora." + - name: key_col + desc: "Nombre de la columna clave usada para emparejar filas existentes con records al preservar trabajo manual. Vacio (default) desactiva la preservacion. El match normaliza el valor (lowercase + espacios colapsados)." + - name: preserve_cols + desc: "Lista de columnas cuyos valores manuales existentes en el libro se conservan: si una celda ya tenia valor para esa clave, ese valor gana sobre el de records. None o lista vacia desactiva la preservacion." + - name: formulas + desc: "Dict opcional {columna: \"plantilla\"} o {columna: {\"f\": plantilla, \"fmt\": formato_numerico}}. La plantilla admite {row} (numero de fila actual) y {NombreColumna} (letra de la columna con ese nombre). Estas columnas se escriben como formula en cada fila y NO se rellenan desde records." + - name: backup + desc: "Si True y el archivo existe, copia a xlsx_path + '.bak' antes de escribir. El .bak es rotativo: cada llamada lo sobrescribe. Default True." + - name: freeze + desc: "Celda para freeze_panes (inmoviliza filas/columnas por encima/izquierda). Default 'A2' (congela la cabecera)." + - name: autofilter + desc: "Si True activa auto_filter sobre el rango A1 hasta la ultima columna y fila de datos. Default True." +output: "Dict con: sheet (nombre de la hoja escrita), rows_written (numero de filas de datos), other_sheets_preserved (lista con los nombres de las demas hojas conservadas), manual_cells_preserved (cuantas celdas de trabajo manual se conservaron) y backup_path (ruta del .bak creado, o cadena vacia si no hubo backup)." +tested: true +tests: ["test_no_destructivo_y_preserva_trabajo_manual", "test_crea_libro_nuevo_si_no_existe"] +test_file_path: "python/functions/infra/upsert_xlsx_sheet_test.py" +file_path: "python/functions/infra/upsert_xlsx_sheet.py" +--- + +## Ejemplo + +```python +import sys, os +sys.path.insert(0, os.path.join("python", "functions")) +from infra.upsert_xlsx_sheet import upsert_xlsx_sheet + +result = upsert_xlsx_sheet( + xlsx_path="/home/enmanuel/afiliados/programas_afiliados.xlsx", + sheet_name="Datos", + records=[ + {"Programa": "Awin", "Clicks": 1200, "Ingreso": 340}, + {"Programa": "CJ", "Clicks": 800, "Ingreso": 210}, + ], + columns=["Programa", "Clicks", "Ingreso", "EPC"], + key_col="Programa", + preserve_cols=["Ingreso"], # el ingreso anotado a mano gana sobre el dato nuevo + formulas={ + "EPC": {"f": '=IFERROR({Ingreso}{row}/{Clicks}{row},"")', "fmt": "0.00"}, + }, +) +print(result) +# {'sheet': 'Datos', 'rows_written': 2, 'other_sheets_preserved': ['Personal'], +# 'manual_cells_preserved': 1, 'backup_path': '/home/enmanuel/afiliados/programas_afiliados.xlsx.bak'} +``` + +La columna EPC se escribe como formula `=IFERROR(C2/B2,"")` (las letras se +resuelven a partir de la posicion de `Ingreso` y `Clicks` en `columns`), y la +hoja "Personal" del usuario no se toca. + +## Cuando usarla + +Usala cuando necesites regenerar UNA hoja de un .xlsx que el usuario tambien +edita a mano, sin destruir su trabajo: refrescar datos de research/scraping +manteniendo columnas anotadas manualmente, o publicar un dataset en una hoja +concreta de un libro que contiene otras hojas personales. Es la pieza de +escritura para flujos donde un editor (OnlyOffice/Excel) y un proceso +automatico comparten el mismo archivo. + +## Gotchas + +- **Impura — escribe en disco.** Pisa SOLO la hoja `sheet_name`; las demas hojas + del libro se conservan tal cual (esa es la garantia central de la funcion). +- **Requiere openpyxl** (ya instalado en `python/.venv`). +- **El .bak es rotativo**: cada llamada con `backup=True` sobrescribe + `xlsx_path + ".bak"`. No es un historial; es la copia de la version anterior. +- **Lee del disco.** Si el archivo esta abierto en OnlyOffice/Excel, GUARDA en + el editor ANTES de llamar a la funcion: ella lee la version en disco y, al + recargar el editor despues, perderias los cambios no guardados. +- **Las columnas se localizan por nombre en la cabecera (fila 1).** Si renombras + una columna entre ejecuciones, el match de `preserve_cols`/`key_col` con la + version anterior se rompe para esa columna (se trata como columna nueva). +- **`key_col` vacio o `preserve_cols` vacio** desactivan la preservacion: la hoja + se reescribe por completo desde `records`. +- Las columnas declaradas en `formulas` se escriben siempre como formula y NO se + rellenan desde `records` ni se preservan. diff --git a/python/functions/infra/upsert_xlsx_sheet.py b/python/functions/infra/upsert_xlsx_sheet.py new file mode 100644 index 00000000..73b1bc6d --- /dev/null +++ b/python/functions/infra/upsert_xlsx_sheet.py @@ -0,0 +1,232 @@ +"""Actualiza de forma NO DESTRUCTIVA una hoja concreta de un archivo .xlsx. + +Generalizacion de la logica probada en afiliados/build_xlsx.py. La garantia +central es que SOLO se reescribe la hoja indicada (sheet_name); cualquier otra +hoja del libro (trabajo del usuario) se conserva intacta. Dentro de la hoja +gestionada se preservan las columnas de trabajo manual (preserve_cols) haciendo +match por una columna clave (key_col): si la celda ya tiene valor en el libro +existente, el valor manual gana sobre el dato nuevo. + +Requiere openpyxl (instalado en python/.venv). +""" + +import os +import shutil + +from openpyxl import Workbook, load_workbook +from openpyxl.styles import Alignment, Border, Font, PatternFill, Side +from openpyxl.utils import get_column_letter + + +def _norm(value): + """Normaliza una clave para el match (lowercase, espacios colapsados).""" + return " ".join(str(value).lower().split()) + + +def _read_preserved(ws, key_col, preserve_cols): + """Lee de una hoja existente los valores manuales por clave. + + Localiza las columnas por su nombre en la cabecera (fila 1). Devuelve un + dict {clave_normalizada: {nombre_columna: valor}} con solo las celdas no + vacias de las columnas en preserve_cols. + """ + preserved = {} + if not preserve_cols or not key_col or ws.max_row < 2: + return preserved + header = {} + for c in range(1, ws.max_column + 1): + name = ws.cell(row=1, column=c).value + if name is not None: + header[name] = c + if key_col not in header: + return preserved + for r in range(2, ws.max_row + 1): + raw_key = ws.cell(row=r, column=header[key_col]).value + if raw_key in (None, ""): + continue + key = _norm(raw_key) + vals = {} + for col in preserve_cols: + if col in header: + v = ws.cell(row=r, column=header[col]).value + if v not in (None, ""): + vals[col] = v + if vals: + preserved[key] = vals + return preserved + + +def _style_objects(): + return { + "header_fill": PatternFill("solid", fgColor="1F4E78"), + "header_font": Font(bold=True, color="FFFFFF", size=11), + "border": Border(*[Side(style="thin", color="D9D9D9")] * 4), + "center": Alignment(horizontal="center", vertical="center"), + "wrap": Alignment(vertical="center", wrap_text=True), + } + + +def _normalize_formulas(formulas): + """Acepta {col: "plantilla"} o {col: {"f": ..., "fmt": ...}} y normaliza + siempre a {col: {"f": plantilla, "fmt": formato_o_None}}. + """ + out = {} + if not formulas: + return out + for col, spec in formulas.items(): + if isinstance(spec, dict): + out[col] = {"f": spec.get("f", ""), "fmt": spec.get("fmt")} + else: + out[col] = {"f": str(spec), "fmt": None} + return out + + +def upsert_xlsx_sheet( + xlsx_path: str, + sheet_name: str, + records: list, + columns: list, + key_col: str = "", + preserve_cols=None, + formulas=None, + backup: bool = True, + freeze: str = "A2", + autofilter: bool = True, +) -> dict: + """Reescribe una sola hoja de un .xlsx conservando el resto del libro. + + Args: + xlsx_path: Ruta del archivo .xlsx. Si no existe se crea un libro nuevo. + sheet_name: Nombre de la hoja a (re)escribir. Es la UNICA hoja tocada. + records: Lista de dicts; cada dict es una fila, sus claves son nombres + de columna. + columns: Orden canonico de columnas (nombres de cabecera). Define que + columnas se escriben y en que orden. + key_col: Nombre de la columna clave usada para el match al preservar + trabajo manual. Vacio (default) desactiva la preservacion. + preserve_cols: Lista de columnas cuyos valores manuales se conservan si + ya existian en el libro (ganan sobre el dato nuevo de records). + formulas: Dict opcional {columna: {"f": plantilla, "fmt": formato}} o + {columna: "plantilla"}. La plantilla admite {row} (numero de fila) + y {NombreColumna} (letra de esa columna). Estas columnas se escriben + como formula y NO se rellenan desde records. + backup: Si True y el archivo existe, copia a xlsx_path + ".bak" antes + de escribir. + freeze: Celda para freeze_panes (default "A2"). + autofilter: Si True activa auto_filter sobre el rango de datos. + + Returns: + Dict con sheet, rows_written, other_sheets_preserved, manual_cells_preserved + y backup_path. + """ + preserve_cols = preserve_cols or [] + formulas = _normalize_formulas(formulas) + + backup_path = "" + if os.path.exists(xlsx_path): + if backup: + backup_path = xlsx_path + ".bak" + shutil.copy2(xlsx_path, backup_path) + wb = load_workbook(xlsx_path) + else: + wb = Workbook() + # Quitar la hoja por defecto vacia; se crearan/usaran las gestionadas. + wb.remove(wb.active) + + # Indice de columna por nombre y letra correspondiente (basado en `columns`). + col_index = {name: i + 1 for i, name in enumerate(columns)} + col_letter = {name: get_column_letter(i + 1) for i, name in enumerate(columns)} + formula_cols = set(formulas.keys()) + + # Preservar trabajo manual ANTES de limpiar la hoja. + if sheet_name in wb.sheetnames: + ws = wb[sheet_name] + preserved = _read_preserved(ws, key_col, preserve_cols) + if ws.max_row: + ws.delete_rows(1, ws.max_row) + else: + ws = wb.create_sheet(title=sheet_name) + preserved = {} + + manual_cells_preserved = sum(len(v) for v in preserved.values()) + + st = _style_objects() + + # Cabecera estilizada (fila 1). + for name in columns: + cell = ws.cell(row=1, column=col_index[name], value=name) + cell.fill = st["header_fill"] + cell.font = st["header_font"] + cell.alignment = st["center"] + cell.border = st["border"] + + # Filas de datos. + for idx, record in enumerate(records): + er = idx + 2 + key = _norm(record.get(key_col, "")) if key_col else "" + pres = preserved.get(key, {}) if key else {} + for name in columns: + c = ws.cell(row=er, column=col_index[name]) + c.border = st["border"] + c.alignment = st["wrap"] + if name in formula_cols: + # Sustituir {row} y {NombreColumna} -> letra de columna. + template = formulas[name]["f"] + rendered = template.format(row=er, **col_letter) + c.value = rendered + fmt = formulas[name]["fmt"] + if fmt: + c.number_format = fmt + continue + if name in preserve_cols and name in pres: + c.value = pres[name] + else: + c.value = record.get(name, "") + + # Auto-ancho por columna: max(len(nombre)+2, 10) acotado a 48, mirando datos. + for name in columns: + max_len = len(name) + for record in records: + if name in formula_cols: + break + v = str(record.get(name, "")) + if len(v) > max_len: + max_len = len(v) + ws.column_dimensions[col_letter[name]].width = min(max(max_len + 2, 10), 48) + + ws.freeze_panes = freeze + + if autofilter and columns: + last_col = col_letter[columns[-1]] + ws.auto_filter.ref = f"A1:{last_col}{max(len(records) + 1, 1)}" + + wb.save(xlsx_path) + + other_sheets = [s for s in wb.sheetnames if s != sheet_name] + + return { + "sheet": sheet_name, + "rows_written": len(records), + "other_sheets_preserved": other_sheets, + "manual_cells_preserved": manual_cells_preserved, + "backup_path": backup_path, + } + + +if __name__ == "__main__": + import tempfile + + tmp = os.path.join(tempfile.mkdtemp(), "demo.xlsx") + result = upsert_xlsx_sheet( + xlsx_path=tmp, + sheet_name="Datos", + records=[ + {"Programa": "Alpha", "Clicks": 100, "Ingreso": 50}, + {"Programa": "Beta", "Clicks": 200, "Ingreso": 80}, + ], + columns=["Programa", "Clicks", "Ingreso", "EPC"], + key_col="Programa", + preserve_cols=["Ingreso"], + formulas={"EPC": {"f": '=IFERROR({Ingreso}{row}/{Clicks}{row},"")', "fmt": "0.00"}}, + ) + print(result) diff --git a/python/functions/infra/upsert_xlsx_sheet_test.py b/python/functions/infra/upsert_xlsx_sheet_test.py new file mode 100644 index 00000000..d75920d7 --- /dev/null +++ b/python/functions/infra/upsert_xlsx_sheet_test.py @@ -0,0 +1,118 @@ +"""Tests para upsert_xlsx_sheet.""" + +import os +import sys + +from openpyxl import Workbook, load_workbook + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..")) + +from functions.infra.upsert_xlsx_sheet import upsert_xlsx_sheet + + +def _build_initial_book(path): + """Crea un libro con dos hojas: 'Datos' (gestionada con valor manual) y + 'Personal' (trabajo del usuario que NO debe tocarse). + """ + wb = Workbook() + wb.remove(wb.active) + + datos = wb.create_sheet("Datos") + # Cabecera de la hoja gestionada. + for c, name in enumerate(["Programa", "Clicks", "Ingreso", "EPC"], start=1): + datos.cell(row=1, column=c, value=name) + # Fila existente para Alpha con un valor manual en "Ingreso". + datos.cell(row=2, column=1, value="Alpha") + datos.cell(row=2, column=2, value=10) + datos.cell(row=2, column=3, value=999) # valor manual a preservar + + personal = wb.create_sheet("Personal") + personal.cell(row=1, column=1, value="mis_notas") + personal.cell(row=2, column=1, value="no me toques") + personal.cell(row=2, column=2, value=1234) + + wb.save(path) + + +def test_no_destructivo_y_preserva_trabajo_manual(tmp_path): + path = str(tmp_path / "libro.xlsx") + _build_initial_book(path) + + result = upsert_xlsx_sheet( + xlsx_path=path, + sheet_name="Datos", + records=[ + {"Programa": "Alpha", "Clicks": 100, "Ingreso": 50}, + {"Programa": "Beta", "Clicks": 200, "Ingreso": 80}, + {"Programa": "Gamma", "Clicks": 300, "Ingreso": 120}, + ], + columns=["Programa", "Clicks", "Ingreso", "EPC"], + key_col="Programa", + preserve_cols=["Ingreso"], + formulas={"EPC": {"f": '=IFERROR({Ingreso}{row}/{Clicks}{row},"")', "fmt": "0.00"}}, + ) + + # El dict de retorno reporta lo esperado. + assert result["sheet"] == "Datos" + assert result["rows_written"] == 3 + assert "Personal" in result["other_sheets_preserved"] + assert result["manual_cells_preserved"] == 1 + assert result["backup_path"] == path + ".bak" + + wb = load_workbook(path) + + # (a) La hoja "Personal" sigue existiendo con sus valores intactos. + assert "Personal" in wb.sheetnames + personal = wb["Personal"] + assert personal.cell(row=1, column=1).value == "mis_notas" + assert personal.cell(row=2, column=1).value == "no me toques" + assert personal.cell(row=2, column=2).value == 1234 + + datos = wb["Datos"] + header = {datos.cell(row=1, column=c).value: c for c in range(1, datos.max_column + 1)} + + # Localizar la fila de Alpha por la columna clave. + alpha_row = None + for r in range(2, datos.max_row + 1): + if datos.cell(row=r, column=header["Programa"]).value == "Alpha": + alpha_row = r + break + assert alpha_row is not None + + # (b) El valor manual en "Ingreso" para Alpha (999) NO se pisó con 50. + assert datos.cell(row=alpha_row, column=header["Ingreso"]).value == 999 + + # (c) Se añadieron las filas nuevas (Beta y Gamma no existían antes). + programas = { + datos.cell(row=r, column=header["Programa"]).value + for r in range(2, datos.max_row + 1) + } + assert {"Alpha", "Beta", "Gamma"} <= programas + + # (d) La columna de fórmula contiene una fórmula. + epc = datos.cell(row=alpha_row, column=header["EPC"]).value + assert isinstance(epc, str) and epc.startswith("=") + assert "IFERROR" in epc + + +def test_crea_libro_nuevo_si_no_existe(tmp_path): + path = str(tmp_path / "nuevo.xlsx") + assert not os.path.exists(path) + + result = upsert_xlsx_sheet( + xlsx_path=path, + sheet_name="Hoja1", + records=[{"A": "x", "B": "y"}], + columns=["A", "B"], + ) + + assert os.path.exists(path) + # Sin archivo previo no hay backup. + assert result["backup_path"] == "" + assert result["rows_written"] == 1 + + wb = load_workbook(path) + assert wb.sheetnames == ["Hoja1"] + ws = wb["Hoja1"] + assert ws.cell(row=1, column=1).value == "A" + assert ws.cell(row=2, column=1).value == "x" diff --git a/python/functions/infra/write_xlsx_sheets.py b/python/functions/infra/write_xlsx_sheets.py new file mode 100644 index 00000000..1ba8d4c7 --- /dev/null +++ b/python/functions/infra/write_xlsx_sheets.py @@ -0,0 +1,172 @@ +"""Escribe un archivo Excel (.xlsx) multi-hoja desde datos en memoria con openpyxl. + +Funcion impura: crea (o sobrescribe) un libro Excel completo a partir de un dict +{nombre_hoja: datos}. Pensada para volcar datasets de un proceso (scraping, +queries, reports) a un .xlsx limpio de una sola pasada, sin preservar estado +previo del archivo. Para actualizar UNA hoja de un libro existente sin tocar las +demas, usar `upsert_xlsx_sheet` en su lugar. +""" + +import os + + +def write_xlsx_sheets( + out_path: str, + sheets: dict, + header_bold: bool = True, + autofit: bool = True, + freeze_header: bool = True, +) -> str: + """Escribe un .xlsx multi-hoja desde datos en memoria. + + Args: + out_path: Ruta del archivo .xlsx a escribir. Se crean los directorios + padre si faltan. Si el archivo ya existe se sobrescribe por completo. + sheets: Dict {nombre_hoja: datos}, una hoja por key en orden de + insercion. Cada valor admite dos formas: + - list[list]: la primera fila son los headers, el resto son filas + de datos. + - dict {"headers": [...], "rows": [[...], ...]}: forma explicita + separando cabeceras de filas. + Un dict sin la clave "rows" se trata como filas vacias; un dict sin + "headers" produce una hoja sin fila de cabecera. + header_bold: Si True (default) la primera fila (cabecera) se escribe en + negrita. + autofit: Si True (default) ajusta el ancho de cada columna a la longitud + maxima del contenido de esa columna (cap a 60 caracteres). + freeze_header: Si True (default) congela la fila de cabecera + (freeze_panes="A2") en cada hoja que tenga cabecera. + + Returns: + La ruta absoluta del archivo .xlsx escrito. + + Raises: + ValueError: si `sheets` esta vacio o si `out_path` esta vacio. + ImportError: si openpyxl no esta instalado en el entorno. + """ + if not out_path: + raise ValueError("out_path no puede estar vacio") + if not sheets: + raise ValueError("sheets no puede estar vacio: se necesita al menos una hoja") + + try: + from openpyxl import Workbook + from openpyxl.styles import Font + from openpyxl.utils import get_column_letter + except ImportError as exc: # pragma: no cover - dependencia del entorno + raise ImportError( + "openpyxl es requerido para write_xlsx_sheets. " + "Instalar con: cd python && uv add openpyxl" + ) from exc + + abs_path = os.path.abspath(out_path) + parent = os.path.dirname(abs_path) + if parent: + os.makedirs(parent, exist_ok=True) + + wb = Workbook() + # El Workbook nace con una hoja por defecto; la eliminamos para controlar el + # orden y los nombres exactamente desde `sheets`. + default_ws = wb.active + wb.remove(default_ws) + + bold = Font(bold=True) + max_width = 60 + + for raw_name, data in sheets.items(): + # openpyxl limita el nombre de hoja a 31 chars y prohibe ciertos caracteres. + name = str(raw_name)[:31] if raw_name else "Sheet" + ws = wb.create_sheet(title=name) + + headers, rows = _normalize_sheet(data) + + all_rows = [] + if headers is not None: + all_rows.append(headers) + all_rows.extend(rows) + + # Ancho aproximado por columna basado en el contenido de TODAS las filas. + col_widths = {} + + for r_idx, row in enumerate(all_rows, start=1): + for c_idx, value in enumerate(row, start=1): + cell = ws.cell(row=r_idx, column=c_idx, value=_coerce(value)) + if header_bold and headers is not None and r_idx == 1: + cell.font = bold + if autofit: + length = len(_display(value)) + if length > col_widths.get(c_idx, 0): + col_widths[c_idx] = length + + if autofit: + for c_idx, width in col_widths.items(): + # +2 de holgura para que el contenido no quede pegado al borde. + ws.column_dimensions[get_column_letter(c_idx)].width = min( + width + 2, max_width + ) + + if freeze_header and headers is not None: + ws.freeze_panes = "A2" + + wb.save(abs_path) + return abs_path + + +def _normalize_sheet(data): + """Devuelve (headers, rows) a partir de cualquiera de las dos formas aceptadas. + + headers es una lista o None (si no hay cabecera). rows es una lista de listas. + """ + if isinstance(data, dict): + headers = data.get("headers") + rows = data.get("rows", []) + rows = [list(r) for r in rows] + return (list(headers) if headers is not None else None, rows) + + # Forma list[list]: primera fila = headers, resto = datos. + seq = list(data) + if not seq: + return (None, []) + headers = list(seq[0]) + rows = [list(r) for r in seq[1:]] + return (headers, rows) + + +def _coerce(value): + """Convierte un valor a algo que openpyxl pueda escribir directamente. + + None, int, float, str y bool son nativos. Cualquier otro tipo se serializa + a str para evitar que openpyxl lance. + """ + if value is None or isinstance(value, (int, float, str, bool)): + return value + return str(value) + + +def _display(value) -> str: + """Representacion en texto de un valor, para medir el ancho de columna.""" + if value is None: + return "" + return str(value) + + +if __name__ == "__main__": # pragma: no cover - smoke manual + import tempfile + + out = os.path.join(tempfile.gettempdir(), "write_xlsx_sheets_demo.xlsx") + path = write_xlsx_sheets( + out, + { + "Ventas": [ + ["Producto", "Unidades", "Precio", "Activo"], + ["Teclado", 12, 29.99, True], + ["Raton", 30, 14.5, False], + ["Monitor", None, 199.0, True], + ], + "Resumen": { + "headers": ["Metrica", "Valor"], + "rows": [["Total productos", 3], ["Ingreso estimado", 6359.99]], + }, + }, + ) + print(f"Escrito: {path}") diff --git a/python/functions/obsidian/__init__.py b/python/functions/obsidian/__init__.py index 0b0ab60a..dfa09e65 100644 --- a/python/functions/obsidian/__init__.py +++ b/python/functions/obsidian/__init__.py @@ -14,6 +14,12 @@ from .search_obsidian_notes import search_obsidian_notes from .list_obsidian_vaults import list_obsidian_vaults from .create_obsidian_vault import create_obsidian_vault +# CRUD de vaults REGISTRADOS en la app de escritorio Obsidian (obsidian.json) +from .register_obsidian_vault import register_obsidian_vault +from .list_registered_obsidian_vaults import list_registered_obsidian_vaults +from .unregister_obsidian_vault import unregister_obsidian_vault +from .open_obsidian_vault import open_obsidian_vault + # Utilidades de migracion / extraccion de subgrafos (grupo obsidian) from .slugify_obsidian_name import slugify_obsidian_name from .extract_obsidian_embeds import extract_obsidian_embeds @@ -34,6 +40,10 @@ __all__ = [ "search_obsidian_notes", "list_obsidian_vaults", "create_obsidian_vault", + "register_obsidian_vault", + "list_registered_obsidian_vaults", + "unregister_obsidian_vault", + "open_obsidian_vault", "slugify_obsidian_name", "extract_obsidian_embeds", "resolve_obsidian_embed", diff --git a/python/functions/obsidian/list_registered_obsidian_vaults.md b/python/functions/obsidian/list_registered_obsidian_vaults.md new file mode 100644 index 00000000..ae989ff0 --- /dev/null +++ b/python/functions/obsidian/list_registered_obsidian_vaults.md @@ -0,0 +1,52 @@ +--- +name: list_registered_obsidian_vaults +kind: function +lang: py +domain: obsidian +version: "1.0.0" +purity: impure +signature: "def list_registered_obsidian_vaults(config_path: str = '') -> list" +description: "Lista los vaults que la app de escritorio Obsidian conoce, leyendo la clave 'vaults' de ~/.config/obsidian/obsidian.json. Opera sobre la config de la app, NO inspecciona el filesystem como list_obsidian_vaults. Devuelve una entrada por vault registrado con id, path, open y ts, ordenada por path. Lista vacia si el archivo no existe." +tags: [obsidian, vault, list, config, desktop-app, obsidian-json] +uses_functions: [] +uses_types: [] +returns: [] +returns_optional: false +error_type: "error_go_core" +imports: ["json", "os"] +params: + - name: config_path + desc: "ruta al obsidian.json de la app; vacio usa ~/.config/obsidian/obsidian.json" +output: "lista de dicts {'id','path','open','ts'} una por vault registrado, ordenada por path; lista vacia si el archivo no existe o no tiene vaults" +tested: true +tests: + - "lista vaults registrados ordenados por path" + - "config inexistente devuelve lista vacia" +test_file_path: "python/functions/obsidian/list_registered_obsidian_vaults_test.py" +file_path: "python/functions/obsidian/list_registered_obsidian_vaults.py" +--- + +## Ejemplo + +```python +import sys, os +sys.path.insert(0, os.path.join("python", "functions")) +from obsidian import list_registered_obsidian_vaults + +vaults = list_registered_obsidian_vaults(config_path="") # ~/.config/obsidian/obsidian.json +for v in vaults: + print(v["id"], v["open"], v["path"]) +# 3f9a1c0b7e2d4a86 True /home/enmanuel/vaults/osint +# a1b2c3d4e5f60718 False /home/enmanuel/vaults/personal +``` + +## Cuando usarla + +Cuando necesites saber que vaults tiene dados de alta la app de escritorio Obsidian (no los que existan en disco): para auditar, desregistrar uno concreto (`unregister_obsidian_vault`), o comprobar si un vault ya esta registrado antes de abrirlo. Distinta de `list_obsidian_vaults`, que escanea un directorio del filesystem en busca de carpetas con `.obsidian/`. + +## Gotchas + +- **Lee la config de la app** (I/O impuro): refleja el estado del archivo `~/.config/obsidian/obsidian.json` en ese instante, no el filesystem real. Un vault listado aqui puede ya no existir en disco (entrada huerfana). +- **Single-instance**: si Obsidian esta abierto y se han registrado vaults despues de arrancar, la lista en disco puede divergir de la que la app tiene en memoria. +- El campo `ts` es epoch en milisegundos (no segundos). Convertir con `datetime.fromtimestamp(ts/1000)` si lo necesitas como fecha. +- Devuelve lista vacia (no lanza) cuando el archivo no existe — util para arranques en limpio. diff --git a/python/functions/obsidian/list_registered_obsidian_vaults.py b/python/functions/obsidian/list_registered_obsidian_vaults.py new file mode 100644 index 00000000..5e64cd79 --- /dev/null +++ b/python/functions/obsidian/list_registered_obsidian_vaults.py @@ -0,0 +1,87 @@ +"""Lista los vaults registrados en la app de escritorio Obsidian. + +Lee el archivo de configuracion de la app (~/.config/obsidian/obsidian.json) y +devuelve las entradas de la clave "vaults". NO inspecciona el sistema de archivos +del vault — solo refleja lo que la app Obsidian conoce. + +Funcion impura: lee un archivo del disco (la config de la app). +""" + +import json +import os + + +def _default_config_path() -> str: + """Ruta por defecto del obsidian.json de la app de escritorio Obsidian.""" + return os.path.expanduser("~/.config/obsidian/obsidian.json") + + +def _load_config(config_path: str) -> dict: + """Carga obsidian.json. Si no existe devuelve la estructura vacia base.""" + if not os.path.exists(config_path): + return {"vaults": {}} + with open(config_path, "r", encoding="utf-8") as f: + data = json.load(f) + if not isinstance(data, dict): + raise ValueError(f"obsidian config is not a JSON object: {config_path}") + if "vaults" not in data or not isinstance(data.get("vaults"), dict): + data["vaults"] = {} + return data + + +def list_registered_obsidian_vaults(config_path: str = "") -> list: + """Devuelve los vaults registrados en la app Obsidian, ordenados por path. + + Args: + config_path: ruta al obsidian.json de la app. Vacio -> default + ~/.config/obsidian/obsidian.json. + + Returns: + Lista de dicts {"id", "path", "open", "ts"}, una por vault registrado, + ordenada por "path". Lista vacia si el archivo no existe o no tiene vaults. + + Raises: + ValueError: si el obsidian.json existente no es un objeto JSON valido. + OSError: si la lectura del archivo falla por I/O. + """ + cfg_path = config_path or _default_config_path() + if not os.path.exists(cfg_path): + return [] + + data = _load_config(cfg_path) + out = [] + for vid, entry in data["vaults"].items(): + if not isinstance(entry, dict): + continue + out.append( + { + "id": vid, + "path": entry.get("path", ""), + "open": entry.get("open", False), + "ts": entry.get("ts", 0), + } + ) + out.sort(key=lambda e: e["path"]) + return out + + +if __name__ == "__main__": + import tempfile + import time + + tmp = tempfile.mkdtemp() + cfg = os.path.join(tmp, "obsidian.json") + payload = { + "extra": "preservar", + "vaults": { + "aaaaaaaaaaaaaaaa": {"path": "/z/Zeta", "ts": int(time.time() * 1000), "open": True}, + "bbbbbbbbbbbbbbbb": {"path": "/a/Alpha", "ts": int(time.time() * 1000), "open": False}, + }, + } + with open(cfg, "w", encoding="utf-8") as f: + json.dump(payload, f) + + rows = list_registered_obsidian_vaults(config_path=cfg) + assert [r["path"] for r in rows] == ["/a/Alpha", "/z/Zeta"], rows + assert list_registered_obsidian_vaults(config_path=os.path.join(tmp, "nope.json")) == [] + print("list_registered_obsidian_vaults smoke OK") diff --git a/python/functions/obsidian/list_registered_obsidian_vaults_test.py b/python/functions/obsidian/list_registered_obsidian_vaults_test.py new file mode 100644 index 00000000..3b3f4351 --- /dev/null +++ b/python/functions/obsidian/list_registered_obsidian_vaults_test.py @@ -0,0 +1,36 @@ +"""Tests para list_registered_obsidian_vaults.""" + +import json + +from list_registered_obsidian_vaults import list_registered_obsidian_vaults + + +def test_lista_vaults_registrados_ordenados_por_path(tmp_path): + # Golden path: dos vaults registrados se devuelven ordenados por path. + cfg = str(tmp_path / "obsidian.json") + with open(cfg, "w", encoding="utf-8") as f: + json.dump( + { + "extra": "preservar", + "vaults": { + "aaaaaaaaaaaaaaaa": {"path": "/z/Zeta", "ts": 200, "open": True}, + "bbbbbbbbbbbbbbbb": {"path": "/a/Alpha", "ts": 100, "open": False}, + }, + }, + f, + ) + + rows = list_registered_obsidian_vaults(config_path=cfg) + assert len(rows) == 2 + assert [r["path"] for r in rows] == ["/a/Alpha", "/z/Zeta"] + + alpha = rows[0] + assert alpha["id"] == "bbbbbbbbbbbbbbbb" + assert alpha["open"] is False + assert alpha["ts"] == 100 + + +def test_config_inexistente_devuelve_lista_vacia(tmp_path): + # Edge: archivo de config ausente -> lista vacia, sin excepcion. + cfg = str(tmp_path / "no_existe.json") + assert list_registered_obsidian_vaults(config_path=cfg) == [] diff --git a/python/functions/obsidian/open_obsidian_vault.md b/python/functions/obsidian/open_obsidian_vault.md new file mode 100644 index 00000000..2bebae54 --- /dev/null +++ b/python/functions/obsidian/open_obsidian_vault.md @@ -0,0 +1,67 @@ +--- +name: open_obsidian_vault +kind: function +lang: py +domain: obsidian +version: "1.0.0" +purity: impure +signature: "def open_obsidian_vault(vault: str, register_if_missing: bool = True, launch: bool = True, config_path: str = '') -> dict" +description: "Abre un vault en la app de escritorio Obsidian construyendo el URI obsidian://open?vault= (name = basename del path o el propio nombre, URL-encodeado) y lanzandolo desacoplado via xdg-open. Si recibe una ruta existente no registrada y register_if_missing=True, la registra antes componiendo register_obsidian_vault (open=True). Con launch=False construye el URI sin lanzar nada (util en tests)." +tags: [obsidian, vault, open, launch, uri, desktop-app, obsidian-json] +uses_functions: ["register_obsidian_vault_py_obsidian"] +uses_types: [] +returns: [] +returns_optional: false +error_type: "error_go_core" +imports: ["os", "subprocess", "urllib.parse"] +params: + - name: vault + desc: "ruta absoluta a un vault o su nombre (basename); si es ruta existente el URI usa su basename" + - name: register_if_missing + desc: "si True (default) y vault es una ruta existente no registrada, la registra (open=True) antes de abrir" + - name: launch + desc: "si True (default) lanza la app via xdg-open desacoplado; False solo construye el URI sin lanzar (tests)" + - name: config_path + desc: "ruta al obsidian.json de la app pasada a register_obsidian_vault; vacio usa ~/.config/obsidian/obsidian.json" +output: "dict con vault (arg original), uri (obsidian://open?vault=), name (usado en el URI), launched (bool) y registered_now (bool, True si se registro en esta llamada)" +tested: true +tests: + - "registra si falta y construye uri sin lanzar gui" + - "por nombre construye uri sin registrar" +test_file_path: "python/functions/obsidian/open_obsidian_vault_test.py" +file_path: "python/functions/obsidian/open_obsidian_vault.py" +--- + +## Ejemplo + +```python +import sys, os +sys.path.insert(0, os.path.join("python", "functions")) +from obsidian import open_obsidian_vault + +# Abrir por ruta: registra si falta y lanza Obsidian +res = open_obsidian_vault( + vault="/home/enmanuel/vaults/osint", + register_if_missing=True, + launch=True, + config_path="", +) +print(res["uri"]) # obsidian://open?vault=osint +print(res["registered_now"])# True si no estaba registrado + +# Solo construir el URI (sin lanzar GUI, p.ej. para inspeccion o tests) +res = open_obsidian_vault("Mi Vault", launch=False) +print(res["uri"]) # obsidian://open?vault=Mi%20Vault +``` + +## Cuando usarla + +Cuando quieras abrir un vault en la app de escritorio Obsidian desde codigo o un agente: tras crear (`create_obsidian_vault`) y registrar (`register_obsidian_vault`) un vault, esta funcion lo abre en un solo paso (registra si hace falta + lanza el URI). Para construir el URI sin lanzar la GUI, usa `launch=False`. + +## Gotchas + +- **Single-instance (lo importante)**: Obsidian es de instancia unica. Si ya hay una instancia corriendo con una config vieja en memoria, el URI `obsidian://open?vault=` puede responder **"unable to find a vault"** para un vault recien registrado, hasta reiniciar la app. Si registras y abres en caliente, reinicia Obsidian para que recargue `obsidian.json`. +- **Lanza un proceso externo** (`xdg-open`, I/O impuro) de forma desacoplada (`start_new_session=True`, stdio a DEVNULL): no bloquea ni captura salida. `launch=False` evita lanzar nada. +- El nombre del URI es el **basename** del path (Obsidian resuelve vaults por nombre, no por ruta completa). Dos vaults con el mismo basename en rutas distintas pueden colisionar en el URI; Obsidian abrira el que tenga ese nombre en su config. +- Necesita un entorno grafico: hereda `DISPLAY`/`WAYLAND_DISPLAY` del entorno. En una sesion headless `xdg-open` puede fallar silenciosamente (el proceso se lanza pero no abre nada). +- Si `register_if_missing=True` y el vault es una ruta existente nueva, **escribe la config de la app** (via `register_obsidian_vault`, con su backup `.bak`). diff --git a/python/functions/obsidian/open_obsidian_vault.py b/python/functions/obsidian/open_obsidian_vault.py new file mode 100644 index 00000000..245fd88c --- /dev/null +++ b/python/functions/obsidian/open_obsidian_vault.py @@ -0,0 +1,107 @@ +"""Abre un vault en la app de escritorio Obsidian via el esquema obsidian://. + +Construye el URI `obsidian://open?vault=` y lanza la app para abrir ese vault. +Opcionalmente lo registra antes en la config de la app si recibe una ruta existente +no registrada (compone register_obsidian_vault del grupo obsidian). + +Funcion impura: puede escribir la config de la app (al registrar) y lanza un proceso +externo (xdg-open) de forma desacoplada para abrir el URI en Obsidian. +""" + +import os +import subprocess +import urllib.parse + +from obsidian import register_obsidian_vault + + +def open_obsidian_vault( + vault: str, + register_if_missing: bool = True, + launch: bool = True, + config_path: str = "", +) -> dict: + """Abre un vault en la app Obsidian construyendo y lanzando un URI obsidian://. + + Args: + vault: ruta absoluta a un vault o su nombre (basename). Si es una ruta + existente, el nombre del URI es su basename; si no parece una ruta + existente se trata como nombre tal cual. + register_if_missing: si True (default) y vault es una ruta existente no + registrada en la app, la registra (open=True) antes de abrir. + launch: si True (default) lanza la app via `xdg-open ` desacoplado. + Si False (util en tests) NO lanza nada, solo construye el URI. + config_path: ruta al obsidian.json de la app. Vacio -> default + ~/.config/obsidian/obsidian.json. Se pasa a register_obsidian_vault. + + Returns: + dict con: + - vault: el argumento original recibido. + - uri: el URI obsidian://open?vault= construido. + - name: nombre del vault usado en el URI (basename o el propio vault). + - launched: True si se lanzo xdg-open. + - registered_now: True si se registro el vault en esta llamada. + + Raises: + OSError: si el lanzamiento del proceso o el registro fallan por I/O. + """ + registered_now = False + is_path = os.path.sep in vault or vault.startswith("~") + abs_path = os.path.abspath(os.path.expanduser(vault)) if is_path else "" + + # Si es una ruta existente, opcionalmente registrarla y usar su basename. + if abs_path and os.path.isdir(abs_path): + name = os.path.basename(abs_path.rstrip(os.path.sep)) + if register_if_missing: + res = register_obsidian_vault(abs_path, open=True, config_path=config_path) + registered_now = bool(res.get("registered")) + elif is_path: + # Parece ruta pero no existe: usar su basename como nombre. + name = os.path.basename(abs_path.rstrip(os.path.sep)) + else: + # Es un nombre, no una ruta. + name = vault + + uri = "obsidian://open?vault=" + urllib.parse.quote(name) + + launched = False + if launch: + env = dict(os.environ) + subprocess.Popen( + ["xdg-open", uri], + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + start_new_session=True, + env=env, + ) + launched = True + + return { + "vault": vault, + "uri": uri, + "name": name, + "launched": launched, + "registered_now": registered_now, + } + + +if __name__ == "__main__": + import tempfile + + tmp = tempfile.mkdtemp() + cfg = os.path.join(tmp, "obsidian.json") + vault = os.path.join(tmp, "Mi Vault") + os.makedirs(vault, exist_ok=True) + + r = open_obsidian_vault(vault, launch=False, config_path=cfg) + assert r["launched"] is False, r + assert r["registered_now"] is True, r + assert r["name"] == "Mi Vault", r + assert r["uri"] == "obsidian://open?vault=Mi%20Vault", r + + # Por nombre, sin lanzar ni registrar. + r2 = open_obsidian_vault("MiVaultPorNombre", launch=False, config_path=cfg) + assert r2["uri"] == "obsidian://open?vault=MiVaultPorNombre", r2 + assert r2["registered_now"] is False, r2 + print("open_obsidian_vault smoke OK") diff --git a/python/functions/obsidian/open_obsidian_vault_test.py b/python/functions/obsidian/open_obsidian_vault_test.py new file mode 100644 index 00000000..b48f304a --- /dev/null +++ b/python/functions/obsidian/open_obsidian_vault_test.py @@ -0,0 +1,40 @@ +"""Tests para open_obsidian_vault.""" + +import json + +from open_obsidian_vault import open_obsidian_vault + + +def test_registra_si_falta_y_construye_uri_sin_lanzar_gui(tmp_path): + # Golden path: ruta existente no registrada -> se registra y se construye el + # URI con el basename URL-encodeado, sin lanzar la app (launch=False). + cfg = str(tmp_path / "obsidian.json") + vault = tmp_path / "Mi Vault" + vault.mkdir() + + res = open_obsidian_vault(str(vault), launch=False, config_path=cfg) + assert res["launched"] is False + assert res["registered_now"] is True + assert res["name"] == "Mi Vault" + assert res["uri"] == "obsidian://open?vault=Mi%20Vault" + assert res["vault"] == str(vault) + + # Quedo registrado en la config de la app. + with open(cfg, "r", encoding="utf-8") as f: + data = json.load(f) + paths = [e["path"] for e in data["vaults"].values()] + assert str(vault) in paths + # Registrado con open=True por open_obsidian_vault. + entry = next(e for e in data["vaults"].values() if e["path"] == str(vault)) + assert entry["open"] is True + + +def test_por_nombre_construye_uri_sin_registrar(tmp_path): + # Edge: vault es un nombre (no ruta) -> URI directo, sin tocar la config. + cfg = str(tmp_path / "obsidian.json") + + res = open_obsidian_vault("MiVaultPorNombre", launch=False, config_path=cfg) + assert res["uri"] == "obsidian://open?vault=MiVaultPorNombre" + assert res["name"] == "MiVaultPorNombre" + assert res["registered_now"] is False + assert res["launched"] is False diff --git a/python/functions/obsidian/register_obsidian_vault.md b/python/functions/obsidian/register_obsidian_vault.md new file mode 100644 index 00000000..86567484 --- /dev/null +++ b/python/functions/obsidian/register_obsidian_vault.md @@ -0,0 +1,64 @@ +--- +name: register_obsidian_vault +kind: function +lang: py +domain: obsidian +version: "1.0.0" +purity: impure +signature: "def register_obsidian_vault(vault_path: str, open: bool = False, config_path: str = '') -> dict" +description: "Registra un vault en la app de escritorio Obsidian anadiendo su entrada a la clave 'vaults' de ~/.config/obsidian/obsidian.json. Opera sobre la config de la app, NO sobre el filesystem del vault. Idempotente por path: no duplica entradas, solo actualiza el flag 'open' si difiere. Genera id hex de 16 chars (secrets.token_hex(8)) y ts en epoch ms. Hace backup .bak antes de escribir y preserva las demas claves del JSON y los demas vaults." +tags: [obsidian, vault, register, config, desktop-app, obsidian-json] +uses_functions: [] +uses_types: [] +returns: [] +returns_optional: false +error_type: "error_go_core" +imports: ["json", "os", "secrets", "shutil", "time"] +params: + - name: vault_path + desc: "ruta al directorio del vault; se normaliza a ruta absoluta antes de registrar" + - name: open + desc: "flag 'open' de la entrada (si Obsidian deberia abrirlo al arrancar); default False" + - name: config_path + desc: "ruta al obsidian.json de la app; vacio usa ~/.config/obsidian/obsidian.json" +output: "dict con id (hex 16), path (abs), registered (bool, True si entrada nueva), already (bool, True si ya existia), open (bool final), config_path y backup_path (ruta del .bak o '' si no habia archivo previo)" +tested: true +tests: + - "registra entrada nueva" + - "segundo registro mismo path no duplica y devuelve already" + - "actualiza flag open de entrada existente" + - "preserva claves extra de nivel superior" + - "crea config y directorios si no existe" + - "hace backup bak antes de sobreescribir" +test_file_path: "python/functions/obsidian/register_obsidian_vault_test.py" +file_path: "python/functions/obsidian/register_obsidian_vault.py" +--- + +## Ejemplo + +```python +import sys, os +sys.path.insert(0, os.path.join("python", "functions")) +from obsidian import register_obsidian_vault + +res = register_obsidian_vault( + vault_path="/home/enmanuel/vaults/osint", + open=True, + config_path="", # usa ~/.config/obsidian/obsidian.json +) +print(res["id"], res["registered"], res["already"]) +# 3f9a1c0b7e2d4a86 True False (primera vez) +# 3f9a1c0b7e2d4a86 False True (segunda vez, ya registrado) +``` + +## Cuando usarla + +Cuando quieras que la app de escritorio Obsidian "conozca" un vault y lo muestre en su selector de vaults (la pantalla de bienvenida / `Open another vault`). Usala despues de `create_obsidian_vault` (que solo crea la carpeta en disco) para dar de alta ese vault en la app. Es el paso previo natural a `open_obsidian_vault`. + +## Gotchas + +- **Escribe la config de la app** (I/O impuro) en `~/.config/obsidian/obsidian.json` y crea un backup `.bak` antes de sobreescribir. Si la entrada ya existe y nada cambia, NO reescribe ni genera backup. +- **NO crea la carpeta del vault**: solo registra la entrada. Si la ruta no existe en disco, Obsidian la mostrara pero no podra abrirla. Usa `create_obsidian_vault` para crear la carpeta. +- **Single-instance**: si Obsidian ya esta corriendo, tiene la lista de vaults cargada en memoria; el vault recien registrado puede no aparecer hasta reiniciar la app. +- Idempotente **por path absoluto**: dos rutas que resuelven al mismo path absoluto se consideran el mismo vault. +- Preserva las demas claves de nivel superior del JSON (`appVersionLastUsed`, `updateDisabled`, etc.) y los demas vaults; solo toca la entrada de este vault. diff --git a/python/functions/obsidian/register_obsidian_vault.py b/python/functions/obsidian/register_obsidian_vault.py new file mode 100644 index 00000000..e2d35bbb --- /dev/null +++ b/python/functions/obsidian/register_obsidian_vault.py @@ -0,0 +1,154 @@ +"""Registra un vault en la app de escritorio Obsidian. + +Opera sobre el archivo de configuracion de la app (~/.config/obsidian/obsidian.json), +NO sobre el sistema de archivos del vault. Anade (o actualiza) la entrada del vault +en la clave "vaults" de ese JSON para que Obsidian lo conozca en su lista de vaults. + +Funcion impura: lee y escribe el archivo de configuracion de la app y crea un +backup .bak antes de sobreescribir. Idempotente: si ya existe una entrada con ese +path no la duplica, solo actualiza su flag "open" si difiere. +""" + +import json +import os +import secrets +import shutil +import time + + +def _default_config_path() -> str: + """Ruta por defecto del obsidian.json de la app de escritorio Obsidian.""" + return os.path.expanduser("~/.config/obsidian/obsidian.json") + + +def _load_config(config_path: str) -> dict: + """Carga obsidian.json. Si no existe devuelve la estructura vacia base.""" + if not os.path.exists(config_path): + return {"vaults": {}} + with open(config_path, "r", encoding="utf-8") as f: + data = json.load(f) + if not isinstance(data, dict): + raise ValueError(f"obsidian config is not a JSON object: {config_path}") + if "vaults" not in data or not isinstance(data.get("vaults"), dict): + data["vaults"] = {} + return data + + +def _save_config(config_path: str, data: dict) -> str: + """Escribe obsidian.json haciendo backup .bak previo si ya existia. + + Crea los directorios padre que falten. Devuelve la ruta del backup creado + (cadena vacia si no habia archivo previo que respaldar). + """ + parent = os.path.dirname(config_path) + if parent: + os.makedirs(parent, exist_ok=True) + + backup_path = "" + if os.path.exists(config_path): + backup_path = config_path + ".bak" + shutil.copy2(config_path, backup_path) + + with open(config_path, "w", encoding="utf-8") as f: + json.dump(data, f, ensure_ascii=False, indent=2) + + return backup_path + + +def register_obsidian_vault( + vault_path: str, + open: bool = False, + config_path: str = "", +) -> dict: + """Registra un vault en la lista de vaults de la app Obsidian. + + Args: + vault_path: ruta al directorio del vault. Se normaliza a ruta absoluta. + open: valor del flag "open" de la entrada del vault (si Obsidian deberia + abrirlo al arrancar). Por defecto False. + config_path: ruta al obsidian.json de la app. Vacio -> default + ~/.config/obsidian/obsidian.json. + + Returns: + dict con: + - id: id hex de 16 chars de la entrada (existente o recien creada). + - path: ruta absoluta normalizada del vault. + - registered: True si se creo una entrada nueva. + - already: True si ya existia una entrada con ese path. + - open: flag "open" final de la entrada. + - config_path: ruta del obsidian.json usado. + - backup_path: ruta del backup .bak escrito (vacio si no habia archivo previo). + + Raises: + ValueError: si el obsidian.json existente no es un objeto JSON valido. + OSError: si la lectura/escritura del archivo falla por I/O. + """ + cfg_path = config_path or _default_config_path() + abs_path = os.path.abspath(os.path.expanduser(vault_path)) + + data = _load_config(cfg_path) + vaults = data["vaults"] + + # Idempotencia: buscar entrada existente por path. + existing_id = None + for vid, entry in vaults.items(): + if isinstance(entry, dict) and entry.get("path") == abs_path: + existing_id = vid + break + + if existing_id is not None: + entry = vaults[existing_id] + changed = entry.get("open") != open + if changed: + entry["open"] = open + backup_path = _save_config(cfg_path, data) if changed else "" + return { + "id": existing_id, + "path": abs_path, + "registered": False, + "already": True, + "open": entry.get("open", open), + "config_path": cfg_path, + "backup_path": backup_path, + } + + # Entrada nueva. + new_id = secrets.token_hex(8) + while new_id in vaults: + new_id = secrets.token_hex(8) + vaults[new_id] = { + "path": abs_path, + "ts": int(time.time() * 1000), + "open": open, + } + backup_path = _save_config(cfg_path, data) + return { + "id": new_id, + "path": abs_path, + "registered": True, + "already": False, + "open": open, + "config_path": cfg_path, + "backup_path": backup_path, + } + + +if __name__ == "__main__": + import tempfile + + tmp = tempfile.mkdtemp() + cfg = os.path.join(tmp, "obsidian.json") + vault = os.path.join(tmp, "MiVault") + os.makedirs(vault, exist_ok=True) + + r1 = register_obsidian_vault(vault, open=True, config_path=cfg) + assert r1["registered"] is True and r1["already"] is False, r1 + assert r1["open"] is True, r1 + assert len(r1["id"]) == 16, r1 + + r2 = register_obsidian_vault(vault, open=True, config_path=cfg) + assert r2["already"] is True and r2["registered"] is False, r2 + assert r2["id"] == r1["id"], (r1, r2) + assert os.path.isfile(cfg + ".bak") is False or True # backup solo si cambio + + print("register_obsidian_vault smoke OK") diff --git a/python/functions/obsidian/register_obsidian_vault_test.py b/python/functions/obsidian/register_obsidian_vault_test.py new file mode 100644 index 00000000..9497e086 --- /dev/null +++ b/python/functions/obsidian/register_obsidian_vault_test.py @@ -0,0 +1,120 @@ +"""Tests para register_obsidian_vault.""" + +import json +import os + +from register_obsidian_vault import register_obsidian_vault + + +def _read(cfg): + with open(cfg, "r", encoding="utf-8") as f: + return json.load(f) + + +def test_registra_entrada_nueva(tmp_path): + # Golden path: registra un vault nuevo en una config inexistente. + cfg = str(tmp_path / "obsidian.json") + vault = tmp_path / "MiVault" + vault.mkdir() + + res = register_obsidian_vault(str(vault), open=True, config_path=cfg) + assert res["registered"] is True + assert res["already"] is False + assert res["open"] is True + assert len(res["id"]) == 16 + assert res["path"] == str(vault) + + data = _read(cfg) + assert res["id"] in data["vaults"] + entry = data["vaults"][res["id"]] + assert entry["path"] == str(vault) + assert entry["open"] is True + assert isinstance(entry["ts"], int) and entry["ts"] > 0 + + +def test_segundo_registro_mismo_path_no_duplica_y_devuelve_already(tmp_path): + # Edge: idempotencia por path. La segunda llamada no crea otra entrada. + cfg = str(tmp_path / "obsidian.json") + vault = tmp_path / "MiVault" + vault.mkdir() + + r1 = register_obsidian_vault(str(vault), open=True, config_path=cfg) + r2 = register_obsidian_vault(str(vault), open=True, config_path=cfg) + + assert r2["already"] is True + assert r2["registered"] is False + assert r2["id"] == r1["id"] + + data = _read(cfg) + assert len(data["vaults"]) == 1 + + +def test_actualiza_flag_open_de_entrada_existente(tmp_path): + # Edge: misma ruta pero flag open distinto -> actualiza y reescribe. + cfg = str(tmp_path / "obsidian.json") + vault = tmp_path / "MiVault" + vault.mkdir() + + register_obsidian_vault(str(vault), open=False, config_path=cfg) + r2 = register_obsidian_vault(str(vault), open=True, config_path=cfg) + + assert r2["already"] is True + assert r2["open"] is True + + data = _read(cfg) + assert data["vaults"][r2["id"]]["open"] is True + + +def test_preserva_claves_extra_de_nivel_superior(tmp_path): + # Edge: claves ajenas a "vaults" deben sobrevivir a la escritura. + cfg = str(tmp_path / "obsidian.json") + with open(cfg, "w", encoding="utf-8") as f: + json.dump( + { + "appVersionLastUsed": "1.5.3", + "updateDisabled": True, + "vaults": { + "0000000000000000": {"path": "/otro/Vault", "ts": 111, "open": False} + }, + }, + f, + ) + + vault = tmp_path / "Nuevo" + vault.mkdir() + register_obsidian_vault(str(vault), open=True, config_path=cfg) + + data = _read(cfg) + assert data["appVersionLastUsed"] == "1.5.3" + assert data["updateDisabled"] is True + # La entrada previa de otro vault se conserva. + assert "0000000000000000" in data["vaults"] + assert len(data["vaults"]) == 2 + + +def test_crea_config_y_directorios_si_no_existe(tmp_path): + # Edge: la config y sus directorios padre no existen y se crean. + cfg = str(tmp_path / "sub" / "dir" / "obsidian.json") + vault = tmp_path / "MiVault" + vault.mkdir() + + res = register_obsidian_vault(str(vault), config_path=cfg) + assert res["registered"] is True + assert os.path.isfile(cfg) + # Sin archivo previo no hay backup. + assert res["backup_path"] == "" + + +def test_hace_backup_bak_antes_de_sobreescribir(tmp_path): + # Edge: una segunda escritura sobre config existente genera backup .bak. + cfg = str(tmp_path / "obsidian.json") + v1 = tmp_path / "V1" + v1.mkdir() + v2 = tmp_path / "V2" + v2.mkdir() + + register_obsidian_vault(str(v1), config_path=cfg) # crea config + res = register_obsidian_vault(str(v2), config_path=cfg) # ahora si hay backup + + assert res["backup_path"] == cfg + ".bak" + assert os.path.isfile(cfg + ".bak") diff --git a/python/functions/obsidian/unregister_obsidian_vault.md b/python/functions/obsidian/unregister_obsidian_vault.md new file mode 100644 index 00000000..093c33c2 --- /dev/null +++ b/python/functions/obsidian/unregister_obsidian_vault.md @@ -0,0 +1,58 @@ +--- +name: unregister_obsidian_vault +kind: function +lang: py +domain: obsidian +version: "1.0.0" +purity: impure +signature: "def unregister_obsidian_vault(vault_ref: str, config_path: str = '') -> dict" +description: "Desregistra un vault de la app de escritorio Obsidian quitando su entrada de la clave 'vaults' de ~/.config/obsidian/obsidian.json. Opera sobre la config de la app: NO borra la carpeta del vault en disco, solo hace que Obsidian deje de conocerlo. Acepta vault_ref como id exacto (hex 16) o como ruta (se normaliza a absoluta y se compara con path). Hace backup .bak antes de escribir y preserva el resto del JSON." +tags: [obsidian, vault, unregister, config, desktop-app, obsidian-json] +uses_functions: [] +uses_types: [] +returns: [] +returns_optional: false +error_type: "error_go_core" +imports: ["json", "os", "shutil"] +params: + - name: vault_ref + desc: "id exacto de la entrada (hex 16 chars) O ruta al vault (se normaliza a absoluta y se compara con 'path'); primero intenta match por id, luego por path" + - name: config_path + desc: "ruta al obsidian.json de la app; vacio usa ~/.config/obsidian/obsidian.json" +output: "dict con removed (bool), id (de la entrada quitada o ''), path (de la entrada quitada o ''), config_path y backup_path (ruta del .bak o '')" +tested: true +tests: + - "desregistra por path" + - "desregistra por id" + - "preserva resto del json al quitar entrada" + - "ref inexistente devuelve removed false" +test_file_path: "python/functions/obsidian/unregister_obsidian_vault_test.py" +file_path: "python/functions/obsidian/unregister_obsidian_vault.py" +--- + +## Ejemplo + +```python +import sys, os +sys.path.insert(0, os.path.join("python", "functions")) +from obsidian import unregister_obsidian_vault + +# Por ruta (se normaliza a absoluta) +res = unregister_obsidian_vault("/home/enmanuel/vaults/viejo", config_path="") +print(res["removed"], res["id"]) # True 3f9a1c0b7e2d4a86 + +# O por id exacto +unregister_obsidian_vault("a1b2c3d4e5f60718", config_path="") +``` + +## Cuando usarla + +Cuando quieras que la app de escritorio Obsidian deje de mostrar un vault en su selector sin tocar los archivos del vault en disco: limpiar entradas obsoletas, quitar un vault movido a otra ruta, o sanear `obsidian.json`. Para listar las entradas y obtener sus ids usa `list_registered_obsidian_vaults`. + +## Gotchas + +- **NO borra la carpeta del vault**: solo elimina la entrada de la config de la app. Los archivos `.md` y el `.obsidian/` del vault siguen en disco. +- **Escribe la config de la app** (I/O impuro) y crea backup `.bak` antes de sobreescribir, pero **solo si encontro la entrada** (`removed=True`). Si no la encuentra, no escribe ni hace backup. +- Resolucion de `vault_ref`: primero intenta match exacto por id; si no, normaliza el ref a ruta absoluta y compara con `path`. Un id que coincide por azar con un path nunca pasara: el match por id va primero. +- **Single-instance**: si Obsidian esta corriendo, sigue teniendo el vault en memoria hasta reiniciar; el desregistro solo afecta al archivo de config. +- Preserva las demas claves de nivel superior del JSON y los demas vaults. diff --git a/python/functions/obsidian/unregister_obsidian_vault.py b/python/functions/obsidian/unregister_obsidian_vault.py new file mode 100644 index 00000000..bb2179e3 --- /dev/null +++ b/python/functions/obsidian/unregister_obsidian_vault.py @@ -0,0 +1,151 @@ +"""Desregistra un vault de la app de escritorio Obsidian. + +Quita la entrada del vault de la clave "vaults" del archivo de configuracion de la +app (~/.config/obsidian/obsidian.json). NO borra la carpeta del vault en disco: solo +hace que la app Obsidian deje de conocerlo. Preserva el resto del JSON intacto. + +Funcion impura: lee y escribe la config de la app y hace backup .bak antes de +sobreescribir. +""" + +import json +import os +import shutil + + +def _default_config_path() -> str: + """Ruta por defecto del obsidian.json de la app de escritorio Obsidian.""" + return os.path.expanduser("~/.config/obsidian/obsidian.json") + + +def _load_config(config_path: str) -> dict: + """Carga obsidian.json. Si no existe devuelve la estructura vacia base.""" + if not os.path.exists(config_path): + return {"vaults": {}} + with open(config_path, "r", encoding="utf-8") as f: + data = json.load(f) + if not isinstance(data, dict): + raise ValueError(f"obsidian config is not a JSON object: {config_path}") + if "vaults" not in data or not isinstance(data.get("vaults"), dict): + data["vaults"] = {} + return data + + +def _save_config(config_path: str, data: dict) -> str: + """Escribe obsidian.json haciendo backup .bak previo si ya existia. + + Devuelve la ruta del backup creado (cadena vacia si no habia archivo previo). + """ + parent = os.path.dirname(config_path) + if parent: + os.makedirs(parent, exist_ok=True) + + backup_path = "" + if os.path.exists(config_path): + backup_path = config_path + ".bak" + shutil.copy2(config_path, backup_path) + + with open(config_path, "w", encoding="utf-8") as f: + json.dump(data, f, ensure_ascii=False, indent=2) + + return backup_path + + +def unregister_obsidian_vault(vault_ref: str, config_path: str = "") -> dict: + """Quita un vault de la lista de vaults conocidos por la app Obsidian. + + Args: + vault_ref: id exacto de la entrada (hex de 16 chars) O una ruta al vault. + Si parece una ruta se normaliza a absoluta y se compara con "path". + Primero se intenta match por id; si no hay, por path. + config_path: ruta al obsidian.json de la app. Vacio -> default + ~/.config/obsidian/obsidian.json. + + Returns: + dict con: + - removed: True si se quito una entrada, False si no se encontro. + - id: id de la entrada quitada ("" si no se encontro). + - path: path de la entrada quitada ("" si no se encontro). + - config_path: ruta del obsidian.json usado. + - backup_path: ruta del backup .bak escrito ("" si no se escribio nada). + + Raises: + ValueError: si el obsidian.json existente no es un objeto JSON valido. + OSError: si la lectura/escritura del archivo falla por I/O. + """ + cfg_path = config_path or _default_config_path() + not_found = { + "removed": False, + "id": "", + "path": "", + "config_path": cfg_path, + "backup_path": "", + } + + if not os.path.exists(cfg_path): + return not_found + + data = _load_config(cfg_path) + vaults = data["vaults"] + + target_id = None + + # 1) Match directo por id. + if vault_ref in vaults: + target_id = vault_ref + else: + # 2) Match por path normalizado. + abs_ref = os.path.abspath(os.path.expanduser(vault_ref)) + for vid, entry in vaults.items(): + if isinstance(entry, dict) and entry.get("path") == abs_ref: + target_id = vid + break + + if target_id is None: + return not_found + + removed_path = "" + entry = vaults.get(target_id) + if isinstance(entry, dict): + removed_path = entry.get("path", "") + del vaults[target_id] + + backup_path = _save_config(cfg_path, data) + return { + "removed": True, + "id": target_id, + "path": removed_path, + "config_path": cfg_path, + "backup_path": backup_path, + } + + +if __name__ == "__main__": + import tempfile + import time + + tmp = tempfile.mkdtemp() + cfg = os.path.join(tmp, "obsidian.json") + payload = { + "extra": "preservar", + "vaults": { + "aaaaaaaaaaaaaaaa": {"path": "/a/Alpha", "ts": int(time.time() * 1000), "open": True}, + "bbbbbbbbbbbbbbbb": {"path": "/b/Beta", "ts": int(time.time() * 1000), "open": False}, + }, + } + with open(cfg, "w", encoding="utf-8") as f: + json.dump(payload, f) + + r = unregister_obsidian_vault("/a/Alpha", config_path=cfg) + assert r["removed"] is True and r["id"] == "aaaaaaaaaaaaaaaa", r + + r2 = unregister_obsidian_vault("bbbbbbbbbbbbbbbb", config_path=cfg) + assert r2["removed"] is True and r2["path"] == "/b/Beta", r2 + + with open(cfg, "r", encoding="utf-8") as f: + final = json.load(f) + assert final["extra"] == "preservar", final + assert final["vaults"] == {}, final + + assert unregister_obsidian_vault("nope", config_path=cfg)["removed"] is False + print("unregister_obsidian_vault smoke OK") diff --git a/python/functions/obsidian/unregister_obsidian_vault_test.py b/python/functions/obsidian/unregister_obsidian_vault_test.py new file mode 100644 index 00000000..9773a3ca --- /dev/null +++ b/python/functions/obsidian/unregister_obsidian_vault_test.py @@ -0,0 +1,83 @@ +"""Tests para unregister_obsidian_vault.""" + +import json + +from unregister_obsidian_vault import unregister_obsidian_vault + + +def _write_cfg(cfg): + with open(cfg, "w", encoding="utf-8") as f: + json.dump( + { + "extra": "preservar", + "appVersionLastUsed": "1.5.3", + "vaults": { + "aaaaaaaaaaaaaaaa": {"path": "/a/Alpha", "ts": 100, "open": True}, + "bbbbbbbbbbbbbbbb": {"path": "/b/Beta", "ts": 200, "open": False}, + }, + }, + f, + ) + + +def _read(cfg): + with open(cfg, "r", encoding="utf-8") as f: + return json.load(f) + + +def test_desregistra_por_path(tmp_path): + # Golden path: quita la entrada cuya 'path' coincide con la ruta dada. + cfg = str(tmp_path / "obsidian.json") + _write_cfg(cfg) + + res = unregister_obsidian_vault("/a/Alpha", config_path=cfg) + assert res["removed"] is True + assert res["id"] == "aaaaaaaaaaaaaaaa" + assert res["path"] == "/a/Alpha" + assert res["backup_path"] == cfg + ".bak" + + data = _read(cfg) + assert "aaaaaaaaaaaaaaaa" not in data["vaults"] + assert "bbbbbbbbbbbbbbbb" in data["vaults"] + + +def test_desregistra_por_id(tmp_path): + # Edge: quita la entrada por su id hex exacto. + cfg = str(tmp_path / "obsidian.json") + _write_cfg(cfg) + + res = unregister_obsidian_vault("bbbbbbbbbbbbbbbb", config_path=cfg) + assert res["removed"] is True + assert res["id"] == "bbbbbbbbbbbbbbbb" + assert res["path"] == "/b/Beta" + + data = _read(cfg) + assert "bbbbbbbbbbbbbbbb" not in data["vaults"] + + +def test_preserva_resto_del_json_al_quitar_entrada(tmp_path): + # Edge: las demas claves y vaults sobreviven al desregistro. + cfg = str(tmp_path / "obsidian.json") + _write_cfg(cfg) + + unregister_obsidian_vault("/a/Alpha", config_path=cfg) + + data = _read(cfg) + assert data["extra"] == "preservar" + assert data["appVersionLastUsed"] == "1.5.3" + assert "bbbbbbbbbbbbbbbb" in data["vaults"] + assert len(data["vaults"]) == 1 + + +def test_ref_inexistente_devuelve_removed_false(tmp_path): + # Error path: ni id ni path coinciden -> removed False, sin reescritura. + cfg = str(tmp_path / "obsidian.json") + _write_cfg(cfg) + + res = unregister_obsidian_vault("/no/existe", config_path=cfg) + assert res["removed"] is False + assert res["id"] == "" + assert res["backup_path"] == "" + + data = _read(cfg) + assert len(data["vaults"]) == 2