kanboard/kanboard_utils.py

# kanboard_utils.py

import requests
from requests.auth import HTTPBasicAuth
import random


def listar_proyectos(api_url: str, usuario: str, token: str) -> list[dict]:
    """
    Devuelve la lista completa de proyectos disponibles en Kanboard.
    """
    auth = HTTPBasicAuth(usuario, token)
    payload = {"jsonrpc": "2.0", "method": "getAllProjects", "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 proyectos: {data['error']}")
    return data.get("result", [])


def listar_usuarios(api_url: str, usuario: str, token: str) -> list[dict]:
    """
    Devuelve la lista completa de usuarios disponibles en Kanboard.
    """
    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", [])


def listar_tareas(api_url: str, usuario: str, token: str, project_id: int) -> list[dict]:
    """
    Lista todas las tareas de un proyecto en Kanboard.
    """
    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", [])


def listar_subtareas(api_url: str, usuario: str, token: str, task_id: int) -> list[dict]:
    """
    Lista todas las subtareas de una tarea principal.
    """
    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", [])


def listar_tareas_por_columna(api_url: str, usuario: str, token: str, project_id: int) -> dict:
    """
    Devuelve todas las tareas agrupadas por columna dentro de un proyecto Kanboard.
    """
    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


def mover_tarea_columna(api_url: str, usuario: str, token: str,
                        project_id: int, task_id: int, column_id: int,
                        position: int = 1, swimlane_id: int = 0) -> bool:
    """
    Mueve una tarea a otra columna dentro de un proyecto Kanboard.
    """
    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
        }
    }

    response = requests.post(api_url, json=payload, auth=auth,
                             headers={"Content-Type": "application/json"}, 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


def crear_tarea(api_url: str, usuario: str, token: str, project_id: int,
                titulo: str, descripcion: str = "", swimlane_id: int | None = None,
                assignee_id: int | None = None, tags: list[str] | None = None,
                priority: int | None = None) -> dict:
    """
    Crea una nueva tarea en un proyecto Kanboard.
    """
    auth = HTTPBasicAuth(usuario, token)
    colores = ["yellow", "blue", "green", "purple", "red", "orange",
               "grey", "brown", "deep_orange", "dark_grey", "lime", "cyan"]
    color_aleatorio = random.choice(colores)

    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"}
    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}")

    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


def editar_tarea(api_url: str, usuario: str, token: str, task_id: int,
                 titulo: str | None = None, descripcion: str | None = None,
                 swimlane_id: int | None = None, assignee_id: int | None = None,
                 tags: list[str] | None = None, priority: int | None = None) -> dict:
    """
    Edita una tarea existente en Kanboard. Solo modifica los campos proporcionados.
    """
    auth = HTTPBasicAuth(usuario, token)
    headers = {"Content-Type": "application/json"}

    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

    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}
    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.")

    if tags:
        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