diff --git a/zulip/integrations/clickup/README.md b/zulip/integrations/clickup/README.md new file mode 100644 index 000000000..0cff0a065 --- /dev/null +++ b/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-client-id \ + --clickup-client-secret \ + +For more information, please see Zulip's documentation on how to set up +a ClickUp integration [here](https://zulip.com/integrations/doc/clickup). diff --git a/zulip/integrations/clickup/__init__.py b/zulip/integrations/clickup/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/zulip/integrations/clickup/zulip_clickup.py b/zulip/integrations/clickup/zulip_clickup.py new file mode 100644 index 000000000..79e2a30a2 --- /dev/null +++ b/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 . + """ + + 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()