integrations: Add ClickUp integration script.

Add a helper python script to help integrate
Zulip with Clickup.

Urlopen is used instead of the usual requests
library inorder to make the script standalone.

Fixes zulip#26529

zulip/integrations/clickup/README.md

@@ -0,0 +1,18 @@
+# A script that automates setting up a webhook with ClickUp
+
+Usage :
+
+1. Make sure you have all of the relevant ClickUp credentials before
+   executing the script:
+    - The ClickUp Team ID
+    - The ClickUp Client ID
+    - The ClickUp Client Secret
+
+2. Execute the script :
+
+    $ python zulip_clickup.py --clickup-team-id <clickup_team_id> \
+                             --clickup-client-id <clickup_board_name> \
+                             --clickup-client-secret <clickup_board_id> \
+
+For more information, please see Zulip's documentation on how to set up
+a ClickUp integration [here](https://zulip.com/integrations/doc/clickup).

zulip/integrations/clickup/zulip_clickup.py

@@ -0,0 +1,368 @@
+#!/usr/bin/env python3  # noqa: EXE001
+#
+# A ClickUp integration script for Zulip.
+
+import argparse
+import json
+import os
+import re
+import sys
+import time
+import urllib.request
+import webbrowser
+from typing import Any, Callable, ClassVar, Dict, List, Tuple, Union
+from urllib.parse import parse_qs, urlparse
+from urllib.request import Request, urlopen
+
+
+def clear_terminal_and_sleep(sleep_duration: int = 3) -> Callable[[Any], Callable[..., Any]]:
+    """
+    Decorator to clear the terminal and sleep for a specified duration before and after the execution of the decorated function.
+    """
+    cmd = "cls" if os.name == "nt" else "clear"
+
+    def decorator(func: Any) -> Any:
+        def wrapper(*args: Any, **kwargs: Any) -> Any:
+            os.system(cmd)  # noqa: S605
+            result = func(*args, **kwargs)
+            time.sleep(sleep_duration)
+            os.system(cmd)  # noqa: S605
+            return result
+
+        return wrapper
+
+    return decorator
+
+
+def process_url(input_url: str, base_url: str) -> str:
+    """
+    Makes sure the input URL is the same the users zulip app URL.
+    Returns the authorization code from the URL query
+    """
+    parsed_input_url = urlparse(input_url)
+    parsed_base_url = urlparse(base_url)
+
+    same_domain: bool = parsed_input_url.netloc == parsed_base_url.netloc
+    auth_code = parse_qs(parsed_input_url.query).get("code")
+
+    if same_domain and auth_code:
+        return auth_code[0]
+    else:
+        print("Unable to fetch the auth code. exiting")
+        sys.exit(1)
+
+
+class ClickUpAPI:
+    def __init__(
+        self,
+        client_id: str,
+        client_secret: str,
+        team_id: str,
+    ) -> None:
+        self.client_id: str = client_id
+        self.client_secret: str = client_secret
+        self.team_id: str = team_id
+        self.API_KEY: str = ""
+
+    # To avoid dependency, urlopen is used instead of requests library
+    # since the script is inteded to be downloaded and run locally
+
+    def get_access_token(self, auth_code: str) -> str:
+        """
+        POST request to retrieve ClickUp's API KEY
+
+        https://clickup.com/api/clickupreference/operation/GetAccessToken/
+        """
+
+        query: Dict[str, str] = {
+            "client_id": self.client_id,
+            "client_secret": self.client_secret,
+            "code": auth_code,
+        }
+        encoded_data = urllib.parse.urlencode(query).encode("utf-8")
+
+        with urlopen("https://api.clickup.com/api/v2/oauth/token", data=encoded_data) as response:
+            if response.status != 200:
+                print(f"Error getting access token: {response.status}")
+                sys.exit(1)
+            data: Dict[str, str] = json.loads(response.read().decode("utf-8"))
+        api_key = data.get("access_token")
+        if api_key:
+            return api_key
+        else:
+            print("Unable to fetch the API key. exiting")
+            sys.exit(1)
+
+    def create_webhook(self, end_point: str, events: List[str]) -> Dict[str, Any]:
+        """
+        POST request to create ClickUp webhooks
+
+        https://clickup.com/api/clickupreference/operation/CreateWebhook/
+        """
+        url: str = f"https://api.clickup.com/api/v2/team/{self.team_id}/webhook"
+
+        payload: Dict[str, Union[str, List[str]]] = {
+            "endpoint": end_point,
+            "events": events,
+        }
+        encoded_payload = json.dumps(payload).encode("utf-8")
+
+        headers: Dict[str, str] = {
+            "Content-Type": "application/json",
+            "Authorization": self.API_KEY,
+        }
+
+        req = Request(url, data=encoded_payload, headers=headers, method="POST")  # noqa: S310
+        with urlopen(req) as response:  # noqa: S310
+            if response.status != 200:
+                print(f"Error creating webhook: {response.status}")
+                sys.exit(1)
+            data: Dict[str, Any] = json.loads(response.read().decode("utf-8"))
+
+        return data
+
+    def get_webhooks(self) -> Dict[str, Any]:
+        """
+        GET request to retrieve ClickUp webhooks
+
+        https://clickup.com/api/clickupreference/operation/GetWebhooks/
+        """
+        url: str = f"https://api.clickup.com/api/v2/team/{self.team_id}/webhook"
+
+        headers: Dict[str, str] = {"Authorization": self.API_KEY}
+
+        req = Request(url, headers=headers, method="GET")  # noqa: S310
+        with urlopen(req) as response:  # noqa: S310
+            if response.getcode() != 200:
+                print(f"Error getting webhooks: {response.getcode()}")
+                sys.exit(1)
+            data: Dict[str, Any] = json.loads(response.read().decode("utf-8"))
+
+        return data
+
+    def delete_webhook(self, webhook_id: str) -> None:
+        """
+        DELETE request to delete a ClickUp webhook
+
+        https://clickup.com/api/clickupreference/operation/DeleteWebhook/
+        """
+        url: str = f"https://api.clickup.com/api/v2/webhook/{webhook_id}"
+
+        headers: Dict[str, str] = {"Authorization": self.API_KEY}
+
+        req = Request(url, headers=headers, method="DELETE")  # noqa: S310
+        with urlopen(req) as response:  # noqa: S310
+            if response.getcode() != 200:
+                print(f"Error deleting webhook: {response.getcode()}")
+                sys.exit(1)
+
+
+class ZulipClickUpIntegration(ClickUpAPI):
+    EVENT_CHOICES: ClassVar[Dict[str, Tuple[str, ...]]] = {
+        "1": ("taskCreated", "taskUpdated", "taskDeleted"),
+        "2": ("listCreated", "listUpdated", "listDeleted"),
+        "3": ("folderCreated", "folderUpdated", "folderDeleted"),
+        "4": ("spaceCreated", "spaceUpdated", "spaceDeleted"),
+        "5": ("goalCreated", "goalUpdated", "goalDeleted"),
+    }
+
+    def __init__(
+        self,
+        client_id: str,
+        client_secret: str,
+        team_id: str,
+    ) -> None:
+        super().__init__(client_id, client_secret, team_id)
+
+    @clear_terminal_and_sleep(1)
+    def query_for_integration_url(self) -> None:
+        print(
+            """
+        STEP 1
+        ----
+        Please enter the integration URL you've just generated
+        from your Zulip app settings.
+
+        It should look similar to this:
+        e.g. http://YourZulipApp.com/api/v1/external/clickup?api_key=TJ9DnIiNqt51bpfyPll5n2uT4iYxMBW9
+        """
+        )
+        while True:
+            input_url: str = input("INTEGRATION URL: ")
+            if input_url:
+                break
+        self.zulip_integration_url = input_url
+
+    @clear_terminal_and_sleep(4)
+    def authorize_clickup_workspace(self) -> None:
+        print(
+            """
+        STEP 2
+        ----
+        ClickUp authorization page will open in your browser.
+        Please authorize your workspace(s).
+
+        Click 'Connect Workspace' on the page to proceed...
+        """
+        )
+        parsed_url = urlparse(self.zulip_integration_url)
+        base_url: str = f"{parsed_url.scheme}://{parsed_url.netloc}"
+        url: str = f"https://app.clickup.com/api?client_id={self.client_id}&redirect_uri={base_url}"
+        time.sleep(1)
+        webbrowser.open(url)
+
+    @clear_terminal_and_sleep(1)
+    def query_for_authorization_code(self) -> str:
+        print(
+            """
+        STEP 3
+        ----
+        After you've authorized your workspace,
+        you should be redirected to your home URL.
+        Please copy your home URL and paste it below.
+        It should contain a code, and look similar to this:
+
+        e.g. https://YourZulipDomain.com/?code=332KKA3321NNAK3MADS
+        """
+        )
+        input_url: str = input("YOUR HOME URL: ")
+
+        auth_code: str = process_url(input_url=input_url, base_url=self.zulip_integration_url)
+
+        return auth_code
+
+    @clear_terminal_and_sleep(1)
+    def query_for_notification_events(self) -> List[str]:
+        print(
+            """
+        STEP 4
+        ----
+        Please select which ClickUp event notification(s) you'd
+        like to receive in your Zulip app.
+        EVENT CODES:
+              1 = task
+              2 = list
+              3 = folder
+              4 = space
+              5 = goals
+
+        Here's an example input if you intend to only receive notifications
+        related to task, list and folder: 1,2,3
+        """
+        )
+        querying_user_input: bool = True
+        selected_events: List[str] = []
+
+        while querying_user_input:
+            input_codes: str = input("EVENT CODE(s): ")
+            user_input: List[str] = re.split(",", input_codes)
+
+            input_is_valid: bool = len(user_input) > 0
+            exhausted_options: List[str] = []
+
+            for event_code in user_input:
+                if event_code in self.EVENT_CHOICES and event_code not in exhausted_options:
+                    selected_events += self.EVENT_CHOICES[event_code]
+                    exhausted_options.append(event_code)
+                else:
+                    input_is_valid = False
+
+            if not input_is_valid:
+                print("Please enter a valid set of options and only select each option once")
+
+            querying_user_input = not input_is_valid
+
+        return selected_events
+
+    def delete_old_webhooks(self) -> None:
+        """
+        Checks for existing webhooks, and deletes them if found.
+        """
+        data: Dict[str, Any] = self.get_webhooks()
+        for webhook in data["webhooks"]:
+            zulip_url_domain = urlparse(self.zulip_integration_url).netloc
+            registered_webhook_domain = urlparse(webhook["endpoint"]).netloc
+
+            if zulip_url_domain in registered_webhook_domain:
+                self.delete_webhook(webhook["id"])
+
+    def run(self) -> None:
+        self.query_for_integration_url()
+        self.authorize_clickup_workspace()
+        auth_code: str = self.query_for_authorization_code()
+        self.API_KEY: str = self.get_access_token(auth_code)
+        events_payload: List[str] = self.query_for_notification_events()
+        self.delete_old_webhooks()
+
+        zulip_webhook_url = (
+            self.zulip_integration_url
+            + "&clickup_api_key="
+            + self.API_KEY
+            + "&team_id="
+            + self.team_id
+        )
+        create_webhook_resp: Dict[str, Any] = self.create_webhook(
+            events=events_payload, end_point=zulip_webhook_url
+        )
+
+        success_msg = """
+        SUCCESS: Registered your zulip app to ClickUp webhook!
+        webhook_id: {webhook_id}
+
+        You may delete this script or run it again to reconfigure
+        your integration.
+        """.format(webhook_id=create_webhook_resp["id"])
+
+        print(success_msg)
+
+
+def main() -> None:
+    description = """
+    zulip_clickup.py is a handy little script that allows Zulip users to
+    quickly set up a ClickUp webhook.
+
+    Note: The ClickUp webhook instructions available on your Zulip server
+    may be outdated. Please make sure you follow the updated instructions
+    at <https://zulip.com/integrations/doc/clickup>.
+    """
+
+    parser = argparse.ArgumentParser(description=description)
+
+    parser.add_argument(
+        "--clickup-team-id",
+        required=True,
+        help=(
+            "Your team_id is the numbers immediately following the base ClickUp URL"
+            "https://app.clickup.com/25567147/home"
+            "For instance, the team_id for the URL above would be 25567147"
+        ),
+    )
+
+    parser.add_argument(
+        "--clickup-client-id",
+        required=True,
+        help=(
+            "Visit https://clickup.com/api/developer-portal/authentication/#step-1-create-an-oauth-app"
+            "and follow 'Step 1: Create an OAuth app' to generate client_id & client_secret."
+        ),
+    )
+    parser.add_argument(
+        "--clickup-client-secret",
+        required=True,
+        help=(
+            "Visit https://clickup.com/api/developer-portal/authentication/#step-1-create-an-oauth-app"
+            "and follow 'Step 1: Create an OAuth app' to generate client_id & client_secret."
+        ),
+    )
+
+    options = parser.parse_args()
+    zulip_clickup_integration = ZulipClickUpIntegration(
+        options.clickup_client_id,
+        options.clickup_client_secret,
+        options.clickup_team_id,
+    )
+    zulip_clickup_integration.run()
+
+
+if __name__ == "__main__":
+    main()