"""
TickTick API Wrapper - OAuth and API client classes
"""
import os
import requests
import urllib.parse
import urllib3
from dotenv import load_dotenv
# Suppress SSL warnings when behind corporate proxy
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
load_dotenv()
class TickTickOAuth:
"""TickTick OAuth 2.0 authentication handler"""
def __init__(self):
self.client_id = os.getenv('TICKTICK_CLIENT_ID')
self.client_secret = os.getenv('TICKTICK_CLIENT_SECRET')
self.redirect_uri = os.getenv('TICKTICK_REDIRECT_URI')
# TickTick OAuth endpoints
self.auth_url = "https://ticktick.com/oauth/authorize"
self.token_url = "https://ticktick.com/oauth/token"
self.api_base = "https://api.ticktick.com"
def get_authorization_url(self, scope="tasks:write tasks:read"):
"""
Generate the authorization URL to redirect user to TickTick
Args:
scope: OAuth scope permissions
Returns:
Authorization URL string
"""
params = {
'client_id': self.client_id,
'response_type': 'code',
'redirect_uri': self.redirect_uri,
'scope': scope,
'state': 'random_state_string' # Use a random string for security
}
query_string = urllib.parse.urlencode(params)
authorization_url = f"{self.auth_url}?{query_string}"
print(f"Visit this URL to authorize: {authorization_url}")
return authorization_url
def exchange_code_for_token(self, authorization_code):
"""
Exchange authorization code for access token
Args:
authorization_code: Code from OAuth callback
Returns:
Dictionary with access_token and other OAuth data
"""
# Prepare the request data
data = {
'grant_type': 'authorization_code',
'client_id': self.client_id,
'client_secret': self.client_secret,
'redirect_uri': self.redirect_uri,
'code': authorization_code
}
# Make the token request
response = requests.post(self.token_url, data=data, verify=False)
if response.status_code == 200:
token_data = response.json()
return {
'access_token': token_data['access_token'],
'token_type': token_data['token_type'],
'expires_in': token_data.get('expires_in'),
'refresh_token': token_data.get('refresh_token')
}
else:
print(f"Error getting token: {response.status_code}")
print(response.text)
return None
def refresh_access_token(self, refresh_token):
"""
Refresh an expired access token
Args:
refresh_token: Refresh token from previous OAuth flow
Returns:
New token data dictionary
"""
data = {
'grant_type': 'refresh_token',
'refresh_token': refresh_token,
'client_id': self.client_id,
'client_secret': self.client_secret
}
response = requests.post(self.token_url, data=data, verify=False)
if response.status_code == 200:
return response.json()
else:
print(f"Error refreshing token: {response.status_code}")
return None
class TickTickAPI:
"""TickTick API client for making authenticated requests"""
def __init__(self, access_token):
"""
Initialize API client
Args:
access_token: Valid TickTick OAuth access token
"""
self.access_token = access_token
self.api_base = "https://api.ticktick.com"
self.headers = {
'Authorization': f'Bearer {access_token}',
'Content-Type': 'application/json'
}
def get_user_info(self):
"""
Validate token by fetching projects.
Note: The /open/v1/user endpoint does not exist in TickTick's API.
We use /open/v1/project as a proxy to validate the token works.
Returns a dict with 'authenticated': True if successful.
"""
response = requests.get(f"{self.api_base}/open/v1/project", headers=self.headers, verify=False)
if response.status_code == 200:
return {'authenticated': True, 'projects_count': len(response.json())}
return None
def get_projects(self):
"""Get all projects (lists)"""
response = requests.get(f"{self.api_base}/open/v1/project", headers=self.headers, verify=False)
return response.json() if response.status_code == 200 else None
def get_projects_data(self, projectId):
"""
Get all data for a specific project including tasks
Args:
projectId: Project ID to fetch
Returns:
Project data dictionary with tasks
"""
response = requests.get(f"{self.api_base}/open/v1/project/{projectId}/data", headers=self.headers, verify=False)
return response.json() if response.status_code == 200 else None
def get_tasks(self, project_id=None, task_id=None):
"""
Get tasks from a specific project or all tasks
Args:
project_id: Optional project ID filter
task_id: Optional specific task ID
Returns:
Task data
"""
if project_id:
url = f"{self.api_base}/open/v1/project/{project_id}/task/{task_id}"
response = requests.get(url, headers=self.headers, verify=False)
print(response.json())
return response.json() if response.status_code == 200 else None
def create_task(self, title, project_id=None, content=None, due_date=None, tags=None):
"""
Create a new task
Args:
title: Task title
project_id: Optional project ID to add task to
content: Optional task description/content
due_date: Optional due date (format: "2024-12-31T23:59:59.000Z")
tags: Optional list of tag strings (e.g., ["work", "urgent"])
Returns:
Created task data or None if failed
"""
task_data = {
'title': title,
'content': content or '',
}
if project_id:
task_data['projectId'] = project_id
if due_date:
task_data['dueDate'] = due_date
if tags:
task_data['tags'] = tags if isinstance(tags, list) else [tags]
response = requests.post(
f"{self.api_base}/open/v1/task",
json=task_data,
headers=self.headers,
verify=False
)
if response.status_code == 200:
return response.json()
else:
print(f"Error creating task: {response.status_code}")
print(f"Response: {response.text}")
return None
def update_task(self, task_id, project_id, title=None, content=None, due_date=None, tags=None):
"""
Update an existing task
Args:
task_id: ID of the task to update
project_id: Project ID the task belongs to
title: Optional new title
content: Optional new description/content
due_date: Optional due date (format: "2024-12-31T23:59:59.000Z")
tags: Optional list of tag strings
Returns:
Updated task data or None if failed
"""
task_data = {
'id': task_id,
'projectId': project_id,
}
if title:
task_data['title'] = title
if content is not None:
task_data['content'] = content
if due_date:
task_data['dueDate'] = due_date
if tags:
task_data['tags'] = tags if isinstance(tags, list) else [tags]
response = requests.post(
f"{self.api_base}/open/v1/task/{task_id}",
json=task_data,
headers=self.headers,
verify=False
)
if response.status_code == 200:
return response.json()
else:
print(f"Error updating task: {response.status_code}")
print(f"Response: {response.text}")
return None
