Files
kanboard/conectarse_api_kanboard.py
2025-10-21 20:38:36 +02:00

579 lines
17 KiB
Python

import marimo
__generated_with = "0.17.0"
app = marimo.App(width="columns")
@app.cell(column=0)
def _():
import marimo as mo
return (mo,)
@app.cell
def _(mo):
mo.md(r"""## Listar Proyectos""")
return
@app.cell
def _():
import requests
from requests.auth import HTTPBasicAuth
import json
url = "http://localhost:8080/jsonrpc.php" # o http://host.docker.internal:8080/jsonrpc.php desde WSL
auth = HTTPBasicAuth("jsonrpc", "792d8fdd6cbf69b3a32d800beaf48b958e4490dd9185c72c06c56061a591")
payload = {
"jsonrpc": "2.0",
"method": "getAllProjects",
"id": 1
}
response = requests.post(
url,
json=payload,
auth=auth,
headers={"Content-Type": "application/json"},
timeout=10
)
# 👇 Convierte el cuerpo en JSON (Python dict)
resultado = response.json()
# Muestra el JSON bonito
resultado
return HTTPBasicAuth, requests
@app.cell
def _(mo):
mo.md(r"""## Listar usuarios""")
return
@app.cell
def _(API_URL, HTTPBasicAuth, TOKEN, USER, requests):
def obtener_usuarios(api_url, usuario, token):
"""
Devuelve la lista completa de usuarios disponibles en Kanboard.
Parámetros:
api_url (str): URL del endpoint JSON-RPC (por ejemplo, http://host.docker.internal:8080/jsonrpc.php)
usuario (str): Usuario con permisos API (ej. 'jsonrpc' o 'admin')
token (str): Token de autenticación del usuario
Retorna:
list[dict]: Lista de usuarios, cada uno con su 'id', 'username', 'name', etc.
"""
auth = HTTPBasicAuth(usuario, token)
payload = {
"jsonrpc": "2.0",
"method": "getAllUsers",
"id": 1
}
response = requests.post(
api_url,
json=payload,
auth=auth,
headers={"Content-Type": "application/json"},
timeout=10
)
data = response.json()
if "error" in data:
raise Exception(f"❌ Error al obtener usuarios: {data['error']}")
return data.get("result", [])
# 🧪 Ejemplo de uso
if __name__ == "__main__":
usuarios = obtener_usuarios(API_URL, USER, TOKEN)
print("✅ Usuarios encontrados:")
for u in usuarios:
print(f"ID {u['id']:>2} | Usuario: {u['username']:<15} | Nombre: {u.get('name', '')}")
return
@app.cell(hide_code=True)
def _(mo):
mo.md(r"""## Listar tareas""")
return
@app.cell
def _(API_URL, HTTPBasicAuth, PROJECT_ID, TOKEN, USER, requests):
def listar_tareas(api_url, usuario, token, project_id):
"""
Lista todas las tareas de un proyecto en Kanboard.
Parámetros:
api_url (str): URL del endpoint JSON-RPC
usuario (str): Usuario con permisos API
token (str): Token del usuario
project_id (int): ID del proyecto donde obtener las tareas
Retorna:
list[dict]: Lista de tareas en el proyecto
"""
auth = HTTPBasicAuth(usuario, token)
payload = {
"jsonrpc": "2.0",
"method": "getAllTasks",
"id": 1,
"params": {
"project_id": project_id
}
}
response = requests.post(api_url, json=payload, auth=auth, headers={"Content-Type": "application/json"})
data = response.json()
if "error" in data:
print("❌ Error al obtener tareas:", data["error"])
return []
return data.get("result", [])
# 🧪 Ejemplo de uso:
tareas = listar_tareas(API_URL, USER, TOKEN, PROJECT_ID)
for tarea in tareas:
print(f"Tarea ID {tarea['id']}: {tarea['title']}")
return
@app.cell(hide_code=True)
def _(mo):
mo.md(r"""### Listar subtareas""")
return
@app.cell
def _(API_URL, HTTPBasicAuth, TOKEN, USER, requests):
def listar_subtareas(api_url, usuario, token, task_id):
"""
Lista todas las subtareas de una tarea.
Parámetros:
api_url (str): URL del endpoint JSON-RPC
usuario (str): Usuario con permisos API
token (str): Token del usuario
task_id (int): ID de la tarea principal para obtener sus subtareas
Retorna:
list[dict]: Lista de subtareas asociadas
"""
auth = HTTPBasicAuth(usuario, token)
payload = {
"jsonrpc": "2.0",
"method": "getAllSubtasks",
"id": 1,
"params": {
"task_id": task_id
}
}
response = requests.post(api_url, json=payload, auth=auth, headers={"Content-Type": "application/json"})
data = response.json()
if "error" in data:
print("❌ Error al obtener subtareas:", data["error"])
return []
return data.get("result", [])
# 🧪 Ejemplo de uso:
task_id = 10 # ID de la tarea principal
subtareas = listar_subtareas(API_URL, USER, TOKEN, task_id)
for sub in subtareas:
print(f"Subtarea ID {sub['id']}: {sub['title']}")
return
@app.cell(hide_code=True)
def _(mo):
mo.md(r"""## Listar tareas columnas""")
return
@app.cell
def _(API_URL, HTTPBasicAuth, PROJECT_ID, TOKEN, USER, requests):
def listar_tareas_por_columna(api_url, usuario, token, project_id):
"""
Devuelve todas las _tareas agrupadas por columna dentro de un proyecto Kanboard.
Parámetros:
api_url (str): URL del endpoint JSON-RPC
usuario (str): Usuario con permisos API
token (str): Token del usuario
project_id (int): ID del proyecto
Retorna:
dict: { 'columna': [ {_tarea}, {_tarea} ] }
"""
auth = HTTPBasicAuth(usuario, token)
payload = {
"jsonrpc": "2.0",
"method": "getBoard",
"id": 1,
"params": {
"project_id": project_id
}
}
response = requests.post(api_url, json=payload, auth=auth, headers={"Content-Type": "application/json"})
data = response.json()
if "error" in data:
print("❌ Error al obtener el tablero:", data["error"])
return {}
board = data.get("result", [])
tareas_por_columna = {}
for swimlane in board:
for columna in swimlane.get("columns", []):
nombre_columna = columna["title"]
tareas_por_columna[nombre_columna] = columna.get("tasks", [])
return tareas_por_columna
# 🧪 Ejemplo de uso
if __name__ == "__main__":
tareas_columna = listar_tareas_por_columna(API_URL, USER, TOKEN, PROJECT_ID)
for columna, _tareas in tareas_columna.items():
print(f"\n📦 {columna} ({len(_tareas)} tareas):")
for _tarea in _tareas:
print(f" ├─ [{_tarea['id']}] {_tarea['title']}")
return
@app.cell(hide_code=True)
def _(mo):
mo.md(r"""## Mover tareas entre columnas""")
return
@app.cell
def _(API_URL, HTTPBasicAuth, PROJECT_ID, TOKEN, USER, requests):
def mover_tarea_columna(api_url, usuario, token, project_id, task_id, column_id, position=1, swimlane_id=0):
"""
Mueve una tarea a otra columna dentro de un proyecto Kanboard.
Parámetros:
api_url (str): URL del endpoint JSON-RPC
usuario (str): Usuario con permisos API (por ejemplo 'jsonrpc' o tu bot)
token (str): Token de autenticación
project_id (int): ID del proyecto
task_id (int): ID de la tarea a mover
column_id (int): ID de la columna destino
position (int): Posición dentro de la columna (por defecto 1)
swimlane_id (int): ID de la swimlane (0 = principal)
Retorna:
bool: True si el movimiento fue exitoso, False en caso de error.
"""
auth = HTTPBasicAuth(usuario, token)
payload = {
"jsonrpc": "2.0",
"method": "moveTaskPosition",
"id": 1,
"params": {
"project_id": project_id,
"task_id": task_id,
"column_id": column_id,
"position": position,
"swimlane_id": swimlane_id
}
}
headers = {"Content-Type": "application/json"}
response = requests.post(api_url, json=payload, auth=auth, headers=headers, timeout=10)
data = response.json()
if "error" in data:
print("❌ Error al mover tarea:", data["error"])
return False
result = data.get("result", False)
if result:
print(f"✅ Tarea {task_id} movida correctamente a la columna {column_id}.")
else:
print(f"⚠️ No se pudo mover la tarea {task_id}.")
return result
# 🧪 Ejemplo de uso
if __name__ == "__main__":
# API_URL = "http://host.docker.internal:8080/jsonrpc.php"
# USER = "jsonrpc"
# TOKEN = "792d8fdd6cbf69b3a32d800beaf48b958e4490dd9185c72c06c56061a591"
# PROJECT_ID = 1
TASK_ID = 6 # ID de la tarea a mover
COLUMN_ID = 3 # Nueva columna (por ejemplo "En progreso")
mover_tarea_columna(API_URL, USER, TOKEN, PROJECT_ID, TASK_ID, COLUMN_ID)
return
@app.cell(column=1)
def _():
return
@app.cell
def _(mo):
mo.md(r"""## Crear tarea""")
return
@app.cell
def _(HTTPBasicAuth, requests):
import random
def crear_tarea(api_url, usuario, token, project_id, titulo, descripcion="", swimlane_id=None, assignee_id=None, tags=None, priority=None):
"""
Crea una tarea en el backlog del proyecto indicado en Kanboard,
con un color aleatorio, etiquetas (tags), prioridad
y (opcionalmente) asignada a un usuario.
Parámetros:
api_url (str): URL del endpoint JSON-RPC
usuario (str): Usuario con permisos API
token (str): Token del usuario
project_id (int): ID del proyecto donde crear la tarea
titulo (str): Título de la tarea
descripcion (str): Descripción de la tarea
swimlane_id (int): ID de la swimlane (opcional)
assignee_id (int): ID del usuario a quien se asignará la tarea (opcional)
tags (list[str]): Lista de etiquetas a añadir (opcional)
priority (int): Prioridad numérica de la tarea (opcional)
Retorna:
dict: Respuesta completa de la API
"""
auth = HTTPBasicAuth(usuario, token)
# 🎨 Colores disponibles
colores = [
"yellow", "blue", "green", "purple",
"red", "orange", "grey", "brown",
"deep_orange", "dark_grey", "lime", "cyan"
]
color_aleatorio = random.choice(colores)
# 🧩 Payload principal
payload = {
"jsonrpc": "2.0",
"method": "createTask",
"id": 1,
"params": {
"title": titulo,
"description": descripcion,
"project_id": project_id,
"color_id": color_aleatorio
}
}
if swimlane_id is not None:
payload["params"]["swimlane_id"] = swimlane_id
if assignee_id is not None:
payload["params"]["owner_id"] = assignee_id
if priority is not None:
payload["params"]["priority"] = priority
headers = {"Content-Type": "application/json"}
# 🚀 Crear la tarea
response = requests.post(api_url, json=payload, auth=auth, headers=headers, timeout=10)
data = response.json()
if "error" in data:
print("❌ Error al crear tarea:", data["error"])
return data
task_id = data.get("result")
print(f"✅ Tarea creada correctamente (ID {task_id}) en el proyecto {project_id}")
# 🏷️ Añadir etiquetas (si se proporcionaron)
if tags:
tag_payload = {
"jsonrpc": "2.0",
"method": "setTaskTags",
"id": 1,
"params": {
"project_id": project_id,
"task_id": task_id,
"tags": tags
}
}
tag_response = requests.post(api_url, json=tag_payload, auth=auth, headers=headers, timeout=10)
tag_data = tag_response.json()
if "error" in tag_data:
print("⚠️ Error al añadir etiquetas:", tag_data["error"])
else:
print(f"🏷️ Etiquetas añadidas: {', '.join(tags)}")
return data
# 🧪 Ejemplo de uso:
if __name__ == "__main__":
API_URL = "http://localhost:8080/jsonrpc.php" # desde WSL
USER = "jsonrpc"
TOKEN = "792d8fdd6cbf69b3a32d800beaf48b958e4490dd9185c72c06c56061a591"
PROJECT_ID = 1
ASSIGNEE_ID = 2 # ID de Egutierrez
crear_tarea(
api_url=API_URL,
usuario=USER,
token=TOKEN,
project_id=PROJECT_ID,
titulo="Tarea con maxima prioridad",
descripcion="Esta tarea tiene prioridad 3 y etiquetas automáticas.",
assignee_id=ASSIGNEE_ID,
tags=["etl", "prioridad-alta"],
priority=3
)
return API_URL, PROJECT_ID, TOKEN, USER
@app.cell(hide_code=True)
def _(mo):
mo.md(r"""## Editar tarea""")
return
@app.cell
def _(API_URL, HTTPBasicAuth, TOKEN, USER, requests):
def editar_tarea(api_url, usuario, token, task_id, titulo=None, descripcion=None,
swimlane_id=None, assignee_id=None, tags=None, priority=None):
"""
Edita una tarea existente en Kanboard.
Solo se envían los parámetros que se indiquen explícitamente.
Parámetros:
api_url (str): URL del endpoint JSON-RPC
usuario (str): Usuario con permisos API
token (str): Token del usuario
task_id (int): ID de la tarea a editar
titulo (str): Nuevo título (opcional)
descripcion (str): Nueva descripción (opcional)
swimlane_id (int): Nueva swimlane (opcional)
assignee_id (int): Nuevo responsable (opcional)
tags (list[str]): Lista de etiquetas (opcional)
priority (int): Nueva prioridad (opcional)
Retorna:
dict: Respuesta completa de la API
"""
auth = HTTPBasicAuth(usuario, token)
headers = {"Content-Type": "application/json"}
# 🧩 Construir payload con solo los campos que se quieren cambiar
params = {"id": task_id}
if titulo is not None:
params["title"] = titulo
if descripcion is not None:
params["description"] = descripcion
if swimlane_id is not None:
params["swimlane_id"] = swimlane_id
if assignee_id is not None:
params["owner_id"] = assignee_id
if priority is not None:
params["priority"] = priority
# Si no hay cambios, no se hace nada
if len(params) == 1:
print("⚠️ No se proporcionaron campos a modificar.")
return {"warning": "sin cambios"}
payload = {
"jsonrpc": "2.0",
"method": "updateTask",
"id": 1,
"params": params
}
# 🚀 Ejecutar la actualización
response = requests.post(api_url, json=payload, auth=auth, headers=headers, timeout=10)
data = response.json()
if "error" in data:
print("❌ Error al editar tarea:", data["error"])
return data
print(f"✅ Tarea {task_id} actualizada correctamente en Kanboard.")
# 🏷️ Actualizar etiquetas si se proporcionan
if tags:
# Para setTaskTags se requiere el project_id, lo obtenemos antes
get_task_payload = {
"jsonrpc": "2.0",
"method": "getTask",
"id": 1,
"params": {"task_id": task_id}
}
task_response = requests.post(api_url, json=get_task_payload, auth=auth, headers=headers, timeout=10)
project_id = task_response.json().get("result", {}).get("project_id")
if project_id:
tag_payload = {
"jsonrpc": "2.0",
"method": "setTaskTags",
"id": 1,
"params": {
"project_id": project_id,
"task_id": task_id,
"tags": tags
}
}
tag_response = requests.post(api_url, json=tag_payload, auth=auth, headers=headers, timeout=10)
tag_data = tag_response.json()
if "error" in tag_data:
print("⚠️ Error al actualizar etiquetas:", tag_data["error"])
else:
print(f"🏷️ Etiquetas actualizadas: {', '.join(tags)}")
return data
# 🧪 Ejemplo de uso:
if __name__ == "__main__":
_TASK_ID = 8 # ID de la tarea existente
editar_tarea(
api_url=API_URL,
usuario=USER,
token=TOKEN,
task_id=_TASK_ID,
titulo="🔥 Actualización directa desde Python",
descripcion="Descripción actualizada sin afectar otros campos.",
assignee_id=2,
tags=["automatizada", "revisión", "python"],
priority=2
)
return
if __name__ == "__main__":
app.run()