# Copyright (c) 2025 Patrick Motsch # All rights reserved. """ClickUp ProviderConnector — virtual paths for teams → lists → tasks (table rows). Path convention (leading slash, no trailing slash except root): / — authorized workspaces (teams) /team/{teamId} — spaces in the workspace /team/{teamId}/space/{spaceId} — folders + folderless lists /team/{teamId}/space/{spaceId}/folder/{folderId} — lists in folder /team/{teamId}/list/{listId} — tasks in list (rows) /team/{teamId}/list/{listId}/task/{taskId} — single task (download = JSON) """ from __future__ import annotations import json import logging import re from typing import Any, Dict, List, Optional from modules.connectors.connectorProviderBase import ( ProviderConnector, ServiceAdapter, DownloadResult, ) from modules.datamodels.datamodelDataSource import ExternalEntry from modules.serviceCenter.services.serviceClickup.mainServiceClickup import ClickupService logger = logging.getLogger(__name__) # type metadata for ExternalEntry.metadata["cuType"] _CU_TEAM = "team" _CU_SPACE = "space" _CU_FOLDER = "folder" _CU_LIST = "list" _CU_TASK = "task" def _norm(path: str) -> str: p = (path or "").strip() or "/" if not p.startswith("/"): p = "/" + p if p != "/" and p.endswith("/"): p = p.rstrip("/") return p class ClickupListsAdapter(ServiceAdapter): """Maps ClickUp hierarchy + list tasks to browse/download/upload/search.""" def __init__(self, access_token: str): self._token = access_token # Minimal service instance for API calls (no ServiceCenter context) self._svc = ClickupService(context=None, get_service=lambda _: None) self._svc.setAccessToken(access_token) async def browse(self, path: str, filter: Optional[str] = None) -> List[ExternalEntry]: p = _norm(path) out: List[ExternalEntry] = [] if p == "/": data = await self._svc.getAuthorizedTeams() if isinstance(data, dict) and data.get("error"): logger.warning(f"ClickUp browse root: {data.get('error')}") return [] teams = data.get("teams", []) if isinstance(data, dict) else [] for t in teams: tid = str(t.get("id", "")) name = t.get("name") or tid out.append( ExternalEntry( name=name, path=f"/team/{tid}", isFolder=True, metadata={"cuType": _CU_TEAM, "id": tid, "raw": t}, ) ) return out m = re.match(r"^/team/([^/]+)$", p) if m: team_id = m.group(1) data = await self._svc.getSpaces(team_id) if isinstance(data, dict) and data.get("error"): return [] spaces = data.get("spaces", []) if isinstance(data, dict) else [] for s in spaces: sid = str(s.get("id", "")) name = s.get("name") or sid out.append( ExternalEntry( name=name, path=f"/team/{team_id}/space/{sid}", isFolder=True, metadata={"cuType": _CU_SPACE, "id": sid, "raw": s}, ) ) return out m = re.match(r"^/team/([^/]+)/space/([^/]+)$", p) if m: team_id, space_id = m.group(1), m.group(2) folders_r = await self._svc.getFolders(space_id) lists_r = await self._svc.getFolderlessLists(space_id) if isinstance(folders_r, dict) and not folders_r.get("error"): for f in folders_r.get("folders", []) or []: fid = str(f.get("id", "")) name = f.get("name") or fid out.append( ExternalEntry( name=name, path=f"/team/{team_id}/space/{space_id}/folder/{fid}", isFolder=True, metadata={"cuType": _CU_FOLDER, "id": fid, "raw": f}, ) ) if isinstance(lists_r, dict) and not lists_r.get("error"): for lst in lists_r.get("lists", []) or []: lid = str(lst.get("id", "")) name = lst.get("name") or lid out.append( ExternalEntry( name=name, path=f"/team/{team_id}/list/{lid}", isFolder=True, metadata={"cuType": _CU_LIST, "id": lid, "raw": lst}, ) ) return out m = re.match(r"^/team/([^/]+)/space/([^/]+)/folder/([^/]+)$", p) if m: team_id, _space_id, folder_id = m.group(1), m.group(2), m.group(3) data = await self._svc.getListsInFolder(folder_id) if isinstance(data, dict) and data.get("error"): return [] for lst in data.get("lists", []) or []: lid = str(lst.get("id", "")) name = lst.get("name") or lid out.append( ExternalEntry( name=name, path=f"/team/{team_id}/list/{lid}", isFolder=True, metadata={"cuType": _CU_LIST, "id": lid, "raw": lst}, ) ) return out m = re.match(r"^/team/([^/]+)/list/([^/]+)$", p) if m: team_id, list_id = m.group(1), m.group(2) page = 0 while True: data = await self._svc.getTasksInList(list_id, page=page) if isinstance(data, dict) and data.get("error"): break tasks = data.get("tasks", []) if isinstance(data, dict) else [] for task in tasks: tid = str(task.get("id", "")) name = task.get("name") or tid out.append( ExternalEntry( name=name, path=f"/team/{team_id}/list/{list_id}/task/{tid}", isFolder=False, metadata={ "cuType": _CU_TASK, "id": tid, "task": task, }, ) ) if len(tasks) < 100: break page += 1 return out m = re.match(r"^/team/([^/]+)/list/([^/]+)/task/([^/]+)$", p) if m: team_id, list_id, task_id = m.group(1), m.group(2), m.group(3) out.append( ExternalEntry( name=f"task-{task_id}.json", path=p, isFolder=False, metadata={"cuType": _CU_TASK, "id": task_id, "listId": list_id, "teamId": team_id}, ) ) return out logger.warning(f"ClickUp browse: unsupported path {p}") return [] async def download(self, path: str) -> Any: p = _norm(path) m = re.match(r"^/team/([^/]+)/list/([^/]+)/task/([^/]+)$", p) if not m: return b"" task_id = m.group(3) data = await self._svc.getTask(task_id) if isinstance(data, dict) and data.get("error"): return json.dumps(data).encode("utf-8") payload = json.dumps(data, indent=2).encode("utf-8") return DownloadResult(data=payload, fileName=f"task-{task_id}.json", mimeType="application/json") async def upload(self, path: str, data: bytes, fileName: str) -> dict: """Upload attachment to a task. Path must be .../list/{listId}/task/{taskId}.""" p = _norm(path) m = re.match(r"^/team/([^/]+)/list/([^/]+)/task/([^/]+)$", p) if not m: return {"error": "Path must be /team/{teamId}/list/{listId}/task/{taskId} for upload"} task_id = m.group(3) return await self._svc.uploadTaskAttachment(task_id, data, fileName) async def search(self, query: str, path: Optional[str] = None) -> List[ExternalEntry]: base = _norm(path or "/") team_id: Optional[str] = None mt = re.match(r"^/team/([^/]+)", base) if mt: team_id = mt.group(1) if not team_id: teams = await self._svc.getAuthorizedTeams() if not isinstance(teams, dict) or teams.get("error"): return [] tl = teams.get("teams") or [] if not tl: return [] team_id = str(tl[0].get("id", "")) out: List[ExternalEntry] = [] page = 0 while True: data = await self._svc.searchTeamTasks(team_id, query=query, page=page) if isinstance(data, dict) and data.get("error"): break tasks = data.get("tasks", []) if isinstance(data, dict) else [] for task in tasks: tid = str(task.get("id", "")) name = task.get("name") or tid list_obj = task.get("list") or {} lid = str(list_obj.get("id", "")) if list_obj else "" if not lid: continue out.append( ExternalEntry( name=name, path=f"/team/{team_id}/list/{lid}/task/{tid}", isFolder=False, metadata={"cuType": _CU_TASK, "id": tid, "task": task}, ) ) if len(tasks) < 25: break page += 1 return out class ClickupConnector(ProviderConnector): """One ClickUp connection → clickup virtual file service.""" def getAvailableServices(self) -> List[str]: return ["clickup"] def getServiceAdapter(self, service: str) -> ServiceAdapter: if service != "clickup": raise ValueError(f"ClickUp only supports 'clickup' service, got '{service}'") return ClickupListsAdapter(self.accessToken)