Skip to content

Commit

Permalink
integrations: Add ClickUp integration script.
Browse files Browse the repository at this point in the history
This script is also intended to be downloaded and run locally
on the user terminal. So, urlopen is used instead of the usual
requests library to avoid dependency.

Unlike zulip_trello.py, this script will have to use some input()
to gather some user input instead of argsparse because some
datas are only available while the script is running. This script
can be run multiple times to re-configure the ClickUp integration.
  • Loading branch information
PieterCK committed Apr 3, 2024
1 parent 20ccb22 commit 48020ab
Show file tree
Hide file tree
Showing 3 changed files with 383 additions and 0 deletions.
18 changes: 18 additions & 0 deletions zulip/integrations/clickup/README.md
Original file line number Diff line number Diff line change
@@ -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).
Empty file.
365 changes: 365 additions & 0 deletions zulip/integrations/clickup/zulip_clickup.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,365 @@
#!/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()

0 comments on commit 48020ab

Please sign in to comment.