cleanup

Author: franaudo

src/codebots/bots/_bot.py

@@ -1,35 +1,32 @@
+import os
 import json
 
 
 class BaseBot():
     """BaseBot class for the other bots.
-    """
+    Modern token handling: supports environment variables for CI/CD and falls back to token files."""
 
     def __init__(self, config_file) -> None:
         self._credentials = self._get_token(config_file)
         for k, v in self._credentials.items():
             self.__setattr__(k, v)
 
     def _get_token(self, config_file):
-        """read the access token form a json file.
+        """Read the access token from environment variable or json file.
 
-        Parameters
-        ----------
-        config_file : json
-            json file containing the bot_token and the bot_chatID
-
-        Returns
-        -------
-        dict
-            credential info
+        Priority:
+        1. Environment variable (recommended for CI/CD)
+        2. Token file (for local development)
         """
+        env_var = f'{self.__class__.__name__.upper()}_BOT_TOKEN'
+        env_token = os.getenv(env_var)
+        if env_token:
+            # If only token is needed, return as dict
+            return {'bot_token': env_token}
+        # Fallback to file
         try:
             with open(config_file, "r") as f:
                 token = json.load(f)
             return token
         except FileNotFoundError:
-            raise FileNotFoundError("there is no token for this bot in the tokens folder.\n\n\
-please provide one by running in the command line the following command:\n\
-`{} set-token` and provide the required information\n\n\
-or set a new tokens folder using:\n\
-`codebots set-tokens-path \"path-to-tokens-folder\"`\n".format(self.__name__))
+            raise FileNotFoundError(f"No token found for {self.__class__.__name__}.\nSet the environment variable {env_var} or run the CLI set-token command.")

src/codebots/bots/clickupbot.py

@@ -0,0 +1,237 @@
+import os
+import time
+import requests
+from typing import Any, Dict, Iterable, List, Optional, Union
+
+# Prefer absolute import to avoid relative import issues
+from codebots.bots._bot import BaseBot
+
+__all__ = ["ClickUpBot"]
+
+
+class ClickUpBot(BaseBot):
+    """Bot to interact with ClickUp API v2 using a personal API token.
+
+    Token handling:
+    - Uses environment variable CLICKUPBOT_BOT_TOKEN if set (recommended for CI/CD).
+    - Falls back to ~/.tokens/clickup.json with {"bot_token": "..."} for local dev.
+
+    Parameters
+    ----------
+    config_file : str, optional
+        Path to a JSON token file. Defaults to ~/.tokens/clickup.json
+    base_url : str, optional
+        Override ClickUp API base URL. Defaults to https://api.clickup.com/api/v2
+    timeout : float, optional
+        HTTP request timeout in seconds (default 30.0)
+    """
+
+    def __init__(
+        self,
+        config_file: Optional[str] = None,
+        base_url: str = "https://api.clickup.com/api/v2",
+        timeout: float = 30.0,
+    ) -> None:
+        if not config_file:
+            from .. import TOKENS
+            config_file = os.path.join(TOKENS, "clickup.json")
+        super().__init__(config_file)
+
+        token = getattr(self, "bot_token", None)
+        if not token:
+            raise ValueError(
+                "ClickUp token not found. Set CLICKUPBOT_BOT_TOKEN or provide a token file."
+            )
+
+        self.base_url = base_url.rstrip("/")
+        self.timeout = timeout
+        self._session = requests.Session()
+        # ClickUp expects the token directly in the Authorization header (no 'Bearer' prefix)
+        self._session.headers.update(
+            {
+                "Authorization": token,
+                "Content-Type": "application/json",
+                "Accept": "application/json",
+            }
+        )
+
+    @property
+    def session(self) -> requests.Session:
+        return self._session
+
+    # ---------------------------
+    # Low-level HTTP helper
+    # ---------------------------
+    def _request(
+        self,
+        method: str,
+        path: str,
+        params: Optional[Dict[str, Any]] = None,
+        json: Optional[Dict[str, Any]] = None,
+        max_retries: int = 3,
+    ) -> Dict[str, Any]:
+        """Issue an HTTP request with basic retry for 429/5xx."""
+        url = f"{self.base_url}/{path.lstrip('/')}"
+        attempt = 0
+        while True:
+            attempt += 1
+            resp = self.session.request(
+                method=method.upper(),
+                url=url,
+                params=params,
+                json=json,
+                timeout=self.timeout,
+            )
+
+            # Handle rate limits and transient errors
+            if resp.status_code in (429, 500, 502, 503, 504) and attempt <= max_retries:
+                retry_after = resp.headers.get("Retry-After")
+                delay = float(retry_after) if retry_after else min(2 ** attempt, 10)
+                time.sleep(delay)
+                continue
+
+            # Raise for other non-success responses
+            if not (200 <= resp.status_code < 300):
+                try:
+                    detail = resp.json()
+                except Exception:
+                    detail = resp.text
+                raise RuntimeError(
+                    f"ClickUp API error {resp.status_code} {resp.reason} at {url}: {detail}"
+                )
+
+            # Return JSON payload
+            try:
+                return resp.json()
+            except ValueError:
+                return {}
+
+    # ---------------------------
+    # Team / Space / Folder / List
+    # ---------------------------
+    def get_teams(self) -> List[Dict[str, Any]]:
+        """Return all teams the token has access to."""
+        data = self._request("GET", "/team")
+        return data.get("teams", [])
+
+    def find_team_id(self, name: str) -> Optional[str]:
+        """Find a team id by its name."""
+        for team in self.get_teams():
+            if team.get("name") == name:
+                return str(team.get("id"))
+        return None
+
+    def get_spaces(self, team_id: Union[int, str], archived: bool = False) -> List[Dict[str, Any]]:
+        data = self._request("GET", f"/team/{team_id}/space", params={"archived": str(archived).lower()})
+        return data.get("spaces", [])
+
+    def get_folders(self, space_id: Union[int, str], archived: bool = False) -> List[Dict[str, Any]]:
+        data = self._request("GET", f"/space/{space_id}/folder", params={"archived": str(archived).lower()})
+        return data.get("folders", [])
+
+    def get_lists(self, folder_id: Union[int, str], archived: bool = False) -> List[Dict[str, Any]]:
+        data = self._request("GET", f"/folder/{folder_id}/list", params={"archived": str(archived).lower()})
+        return data.get("lists", [])
+
+    # ---------------------------
+    # Tasks
+    # ---------------------------
+    def list_tasks(
+        self,
+        list_id: Union[int, str],
+        page: Optional[int] = None,
+        archived: bool = False,
+        include_subtasks: Optional[bool] = None,
+    ) -> List[Dict[str, Any]]:
+        params: Dict[str, Any] = {"archived": str(archived).lower()}
+        if page is not None:
+            params["page"] = page
+        if include_subtasks is not None:
+            params["subtasks"] = str(include_subtasks).lower()
+        data = self._request("GET", f"/list/{list_id}/task", params=params)
+        return data.get("tasks", [])
+
+    def get_task(self, task_id: Union[int, str]) -> Dict[str, Any]:
+        return self._request("GET", f"/task/{task_id}")
+
+    def create_task(
+        self,
+        list_id: Union[int, str],
+        name: str,
+        description: Optional[str] = None,
+        status: Optional[str] = None,
+        assignees: Optional[Iterable[Union[int, str]]] = None,
+        tags: Optional[Iterable[str]] = None,
+        priority: Optional[int] = None,
+        due_date: Optional[int] = None,  # Unix ms
+        due_date_time: Optional[bool] = None,
+        start_date: Optional[int] = None,  # Unix ms
+        start_date_time: Optional[bool] = None,
+        notify_all: Optional[bool] = None,
+        parent: Optional[str] = None,
+        time_estimate: Optional[int] = None,
+        custom_fields: Optional[List[Dict[str, Any]]] = None,
+        extra: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        """Create a task in a list. Supply additional API fields via `extra` if needed."""
+        payload: Dict[str, Any] = {"name": name}
+        if description is not None:
+            payload["description"] = description
+        if status is not None:
+            payload["status"] = status
+        if assignees is not None:
+            payload["assignees"] = [int(a) for a in assignees]
+        if tags is not None:
+            payload["tags"] = list(tags)
+        if priority is not None:
+            payload["priority"] = int(priority)
+        if due_date is not None:
+            payload["due_date"] = int(due_date)
+        if due_date_time is not None:
+            payload["due_date_time"] = bool(due_date_time)
+        if start_date is not None:
+            payload["start_date"] = int(start_date)
+        if start_date_time is not None:
+            payload["start_date_time"] = bool(start_date_time)
+        if notify_all is not None:
+            payload["notify_all"] = bool(notify_all)
+        if parent is not None:
+            payload["parent"] = str(parent)
+        if time_estimate is not None:
+            payload["time_estimate"] = int(time_estimate)
+        if custom_fields is not None:
+            payload["custom_fields"] = custom_fields
+        if extra:
+            payload.update(extra)
+
+        return self._request("POST", f"/list/{list_id}/task", json=payload)
+
+    def update_task(
+        self,
+        task_id: Union[int, str],
+        fields: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        """Update a task. Provide API fields in `fields` (e.g., {'status': 'in progress'})."""
+        return self._request("PUT", f"/task/{task_id}", json=fields)
+
+    def add_comment(
+        self,
+        task_id: Union[int, str],
+        comment_text: str,
+        assignee_id: Optional[Union[int, str]] = None,
+        notify_all: bool = False,
+    ) -> Dict[str, Any]:
+        """Add a comment to a task."""
+        payload: Dict[str, Any] = {"comment_text": comment_text, "notify_all": bool(notify_all)}
+        if assignee_id is not None:
+            payload["assignee"] = int(assignee_id)
+        return self._request("POST", f"/task/{task_id}/comment", json=payload)
+
+
+# Debug
+if __name__ == "__main__":
+    # Example usage:
+    # export CLICKUPBOT_BOT_TOKEN="your-token"
+    bot = ClickUpBot()
+    teams = bot.get_teams()
+    print(f"Teams: {[t.get('name') for t in teams]}")

src/codebots/bots/deploybot.py

@@ -54,13 +54,20 @@ class DeployBot(BaseBot):
 
     def __init__(self, project=None, config_file=None) -> None:
         self.__name__ = "deploybot"
+        # Use BaseBot for modern token handling
         if not config_file:
             if not project:
                 raise ValueError("Either a project name or a config_file must be passed")
             from .. import TOKENS
-            config_file = os.path .join(TOKENS, "{}.json".format(project))
+            config_file = os.path.join(TOKENS, f"{project}.json")
         super().__init__(config_file)
-        self.server_complete_path = "{}:{}".format(self.server_address, self.server_repo_path)
+        # Ensure required attributes are set from credentials
+        self.local_repo_path = getattr(self, 'local_repo_path', None)
+        self.server_repo_path = getattr(self, 'server_repo_path', None)
+        self.server_address = getattr(self, 'server_address', None)
+        if not all([self.local_repo_path, self.server_repo_path, self.server_address]):
+            raise ValueError("Missing required deployment configuration. Check your token file or environment variables.")
+        self.server_complete_path = f"{self.server_address}:{self.server_repo_path}"
         self.local_repo = Repo(self.local_repo_path)
 
     def deploy_to_server(self, remote_name="deploy", local_name="master", commit_message="deployed"):

src/codebots/bots/drivebot.py

@@ -33,8 +33,14 @@ class DriveBot():
         if True stores the credentials in the .tokens folder for future use, by default True.
     """
 
-    def __init__(self, authentication, save_credentials=True) -> None:
-        GoogleAuth.DEFAULT_SETTINGS['client_config_file'] = SECRETS
+    def __init__(self, authentication=None, save_credentials=True) -> None:
+        # Modern token handling: env var first, then config file
+        env_token = os.getenv("DRIVEBOT_TOKEN")
+        if env_token:
+            self.token = env_token
+        else:
+            from .. import TOKENS, SECRETS
+            GoogleAuth.DEFAULT_SETTINGS['client_config_file'] = SECRETS
         self._gauth = GoogleAuth()
         self._drive = self._authenticate(authentication, save_credentials)
 
@@ -46,7 +52,7 @@ def gauth(self):
     @property
     def drive(self):
         """obj :  pydrive GoogleDrive object"""
-        return self._gdrive
+        return self._drive
 
     def _authenticate(self, authentication, save_credentials):
         """Sign in your Google Drive.