feat: enhance jupyter notebook functions with auto-init and kernel management

Auto-create notebooks y sesiones en jupyter_exec (append y cell).
Auto-create en jupyter_write (append_code, append_markdown, batch).
Nuevos subcomandos cleanup y shutdown-all en jupyter_kernel.
README.md renombrado a README.txt para evitar error de parseo del indexer.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-04-08 00:10:23 +02:00
parent 0bdf35a461
commit 72c0379c63
7 changed files with 215 additions and 7 deletions
@@ -196,6 +196,80 @@ def jupyter_kernel_sessions(
return sessions
def jupyter_kernel_cleanup(
server_url: str = "http://localhost:8888",
token: str = "",
idle_seconds: int = 3600,
) -> list[dict]:
"""Apaga todos los kernels que llevan mas de idle_seconds sin actividad.
Util para liberar recursos en servidores con muchos notebooks abiertos.
Por defecto cierra kernels inactivos desde hace mas de 1 hora.
Args:
server_url: URL base del servidor Jupyter.
token: Token de autenticacion. Vacio si el servidor no requiere auth.
idle_seconds: Segundos de inactividad para considerar un kernel ocioso.
Returns:
Lista de dicts con los kernels apagados (id, name, last_activity, idle_seconds).
Raises:
urllib.error.HTTPError: Si la respuesta HTTP indica un error.
urllib.error.URLError: Si no se puede conectar al servidor.
"""
from datetime import datetime, timezone
kernels = jupyter_kernel_list(server_url, token)
now = datetime.now(timezone.utc)
shutdown_list = []
for k in kernels:
last_activity = k.get("last_activity", "")
if not last_activity:
continue
try:
last_dt = datetime.fromisoformat(last_activity.replace("Z", "+00:00"))
idle = (now - last_dt).total_seconds()
except (ValueError, TypeError):
continue
if idle >= idle_seconds:
jupyter_kernel_shutdown(server_url, token, k["id"])
shutdown_list.append({
"id": k["id"],
"name": k.get("name", ""),
"last_activity": last_activity,
"idle_seconds": int(idle),
})
return shutdown_list
def jupyter_kernel_shutdown_all(
server_url: str = "http://localhost:8888",
token: str = "",
) -> list[dict]:
"""Apaga todos los kernels activos del servidor.
Args:
server_url: URL base del servidor Jupyter.
token: Token de autenticacion. Vacio si el servidor no requiere auth.
Returns:
Lista de dicts con los kernels apagados (id, name).
Raises:
urllib.error.HTTPError: Si la respuesta HTTP indica un error.
urllib.error.URLError: Si no se puede conectar al servidor.
"""
kernels = jupyter_kernel_list(server_url, token)
shutdown_list = []
for k in kernels:
jupyter_kernel_shutdown(server_url, token, k["id"])
shutdown_list.append({"id": k["id"], "name": k.get("name", "")})
return shutdown_list
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
@@ -248,6 +322,18 @@ if __name__ == "__main__":
# sessions
subparsers.add_parser("sessions", help="Lista las sesiones activas.")
# cleanup
sp_cleanup = subparsers.add_parser("cleanup", help="Apaga kernels inactivos.")
sp_cleanup.add_argument(
"--idle-seconds",
type=int,
default=3600,
help="Segundos de inactividad para considerar ocioso (default: 3600)",
)
# shutdown-all
subparsers.add_parser("shutdown-all", help="Apaga todos los kernels activos.")
args = parser.parse_args()
try:
@@ -267,6 +353,10 @@ if __name__ == "__main__":
result = {"status": "shutdown", "kernel_id": args.kernel_id}
elif args.command == "sessions":
result = jupyter_kernel_sessions(args.server, args.token)
elif args.command == "cleanup":
result = jupyter_kernel_cleanup(args.server, args.token, args.idle_seconds)
elif args.command == "shutdown-all":
result = jupyter_kernel_shutdown_all(args.server, args.token)
else:
parser.print_help()
sys.exit(1)