145 lines
4.8 KiB
Python
145 lines
4.8 KiB
Python
"""
|
|
Direct TickTick API client using only OAuth tokens.
|
|
This bypasses the flawed ticktick-py library that incorrectly requires username/password.
|
|
"""
|
|
|
|
import requests
|
|
import urllib3
|
|
from typing import Optional, Dict, List, Any
|
|
from datetime import datetime
|
|
import logging
|
|
|
|
from .auth import load_stored_tokens
|
|
|
|
# Suppress SSL warnings for corporate networks
|
|
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
|
|
|
|
|
class TickTickDirectClient:
|
|
"""Direct TickTick API client using OAuth only."""
|
|
|
|
BASE_URL = "https://api.ticktick.com/open/v1"
|
|
|
|
def __init__(self):
|
|
"""Initialize the client with OAuth token."""
|
|
self.tokens = load_stored_tokens()
|
|
if not self.tokens:
|
|
raise RuntimeError(
|
|
"No OAuth tokens found. Please run authentication first."
|
|
)
|
|
|
|
self.access_token = self.tokens["access_token"]
|
|
self.session = requests.Session()
|
|
self.session.verify = False # Disable SSL verification for corporate networks
|
|
|
|
# Set headers
|
|
self.session.headers.update(
|
|
{
|
|
"Authorization": f"Bearer {self.access_token}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
)
|
|
|
|
def _request(self, method: str, endpoint: str, **kwargs) -> requests.Response:
|
|
"""Make a request to the TickTick API."""
|
|
url = f"{self.BASE_URL}/{endpoint.lstrip('/')}"
|
|
|
|
try:
|
|
response = self.session.request(method, url, **kwargs)
|
|
response.raise_for_status()
|
|
return response
|
|
except requests.exceptions.HTTPError as e:
|
|
if response.status_code == 401:
|
|
raise RuntimeError(
|
|
"OAuth token expired or invalid. Please re-authenticate."
|
|
)
|
|
elif response.status_code == 429:
|
|
raise RuntimeError("Rate limit exceeded. Please try again later.")
|
|
else:
|
|
raise RuntimeError(f"API request failed: {e}")
|
|
|
|
def get_projects(self) -> List[Dict[str, Any]]:
|
|
"""Get all projects."""
|
|
response = self._request("GET", "/project")
|
|
return response.json()
|
|
|
|
def get_tasks(self, project_id: Optional[str] = None) -> List[Dict[str, Any]]:
|
|
"""Get tasks, optionally filtered by project."""
|
|
# NOTE: TickTick's GET /task endpoint appears to have issues (returns 500)
|
|
# This is a known limitation of their API
|
|
# For now, we'll return an empty list and log the issue
|
|
|
|
import logging
|
|
|
|
logging.warning(
|
|
"TickTick GET /task endpoint returns 500 server error - this is a known API issue"
|
|
)
|
|
|
|
# TODO: Implement alternative task fetching when TickTick fixes their API
|
|
# Possible workarounds:
|
|
# 1. Use websocket/sync endpoints
|
|
# 2. Cache created tasks locally
|
|
# 3. Use different API version when available
|
|
|
|
return []
|
|
|
|
def create_task(
|
|
self,
|
|
title: str,
|
|
content: Optional[str] = None,
|
|
project_id: Optional[str] = None,
|
|
due_date: Optional[str] = None,
|
|
priority: Optional[int] = None,
|
|
tags: Optional[List[str]] = None,
|
|
) -> Dict[str, Any]:
|
|
"""Create a new task."""
|
|
task_data = {"title": title}
|
|
|
|
if content:
|
|
task_data["content"] = content
|
|
if project_id:
|
|
task_data["projectId"] = project_id
|
|
if due_date:
|
|
# Convert date string to ISO format if needed
|
|
task_data["dueDate"] = due_date
|
|
if priority is not None:
|
|
task_data["priority"] = priority
|
|
if tags:
|
|
task_data["tags"] = tags
|
|
|
|
response = self._request("POST", "/task", json=task_data)
|
|
return response.json()
|
|
|
|
def update_task(self, task_id: str, **kwargs) -> Dict[str, Any]:
|
|
"""Update an existing task."""
|
|
response = self._request("POST", f"/task/{task_id}", json=kwargs)
|
|
return response.json()
|
|
|
|
def complete_task(self, task_id: str) -> Dict[str, Any]:
|
|
"""Mark a task as completed."""
|
|
return self.update_task(task_id, status=2) # 2 = completed
|
|
|
|
def delete_task(self, task_id: str) -> bool:
|
|
"""Delete a task."""
|
|
try:
|
|
self._request("DELETE", f"/task/{task_id}")
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
def get_task(self, task_id: str) -> Dict[str, Any]:
|
|
"""Get a specific task by ID."""
|
|
# NOTE: TickTick's GET /task/{id} endpoint also returns 500 server error
|
|
import logging
|
|
|
|
logging.warning(
|
|
f"TickTick GET /task/{task_id} endpoint returns 500 server error - this is a known API issue"
|
|
)
|
|
|
|
# Return minimal task info
|
|
return {
|
|
"id": task_id,
|
|
"title": "Task details unavailable (API issue)",
|
|
"status": 0,
|
|
}
|