302 lines
10 KiB
Python
302 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import secrets
|
|
import base64
|
|
from dataclasses import asdict, dataclass, field, fields
|
|
from typing import Any
|
|
from urllib import error, parse, request
|
|
|
|
from app.config import Settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
TICKTICK_AUTH_URL = "https://ticktick.com/oauth/authorize"
|
|
TICKTICK_TOKEN_URL = "https://ticktick.com/oauth/token"
|
|
TICKTICK_OPEN_API_BASE_URL = "https://api.ticktick.com/open/v1"
|
|
TICKTICK_DATETIME_FORMAT = "%Y-%m-%dT%H:%M:%S%z"
|
|
AUTH_SCOPE = "tasks:read tasks:write"
|
|
|
|
|
|
class TickTickConfigError(RuntimeError):
|
|
"""Raised when TickTick is missing required runtime configuration."""
|
|
|
|
|
|
class TickTickAuthError(RuntimeError):
|
|
"""Raised when TickTick OAuth state validation fails."""
|
|
|
|
|
|
class TickTickRequestError(RuntimeError):
|
|
"""Raised when a TickTick API request fails."""
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class TickTickProject:
|
|
id: str
|
|
name: str
|
|
color: str | None = None
|
|
sortOrder: int | None = None
|
|
closed: bool | None = None
|
|
groupId: str | None = None
|
|
viewMode: str | None = None
|
|
permission: str | None = None
|
|
kind: str | None = None
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class TickTickTask:
|
|
projectId: str
|
|
title: str
|
|
id: str | None = None
|
|
isAllDay: bool | None = None
|
|
completedTime: str | None = None
|
|
content: str | None = None
|
|
desc: str | None = None
|
|
dueDate: str | None = None
|
|
items: list[Any] | None = None
|
|
priority: int | None = None
|
|
reminders: list[str] | None = None
|
|
repeatFlag: str | None = None
|
|
sortOrder: int | None = None
|
|
startDate: str | None = None
|
|
status: int | None = None
|
|
timeZone: str | None = None
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class TickTickAuthStateStore:
|
|
pending_state: str | None = None
|
|
|
|
def issue_state(self) -> str:
|
|
self.pending_state = secrets.token_hex(6)
|
|
return self.pending_state
|
|
|
|
def matches_state(self, state: str) -> bool:
|
|
return bool(self.pending_state and state == self.pending_state)
|
|
|
|
def consume_state(self, state: str) -> bool:
|
|
if not self.pending_state or state != self.pending_state:
|
|
return False
|
|
self.pending_state = None
|
|
return True
|
|
|
|
def clear(self) -> None:
|
|
self.pending_state = None
|
|
|
|
|
|
default_auth_state_store = TickTickAuthStateStore()
|
|
|
|
|
|
def _coerce_dataclass_payload(model_type: type, payload: dict[str, Any]) -> Any:
|
|
allowed_field_names = {item.name for item in fields(model_type)}
|
|
filtered_payload = {
|
|
key: value for key, value in payload.items() if key in allowed_field_names
|
|
}
|
|
return model_type(**filtered_payload)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class TickTickClient:
|
|
settings: Settings
|
|
auth_state_store: TickTickAuthStateStore = field(default_factory=lambda: default_auth_state_store)
|
|
timeout_seconds: float = 10.0
|
|
|
|
def is_configured(self) -> bool:
|
|
return bool(self._client_id() and self._client_secret())
|
|
|
|
def has_token(self) -> bool:
|
|
return bool(self.settings.ticktick_token)
|
|
|
|
def build_authorization_url(self) -> str:
|
|
self._require_auth_config()
|
|
state = self.auth_state_store.issue_state()
|
|
params = parse.urlencode(
|
|
{
|
|
"client_id": self._client_id(),
|
|
"response_type": "code",
|
|
"redirect_uri": self._redirect_uri(),
|
|
"state": state,
|
|
"scope": AUTH_SCOPE,
|
|
}
|
|
)
|
|
return f"{TICKTICK_AUTH_URL}?{params}"
|
|
|
|
def exchange_authorization_code(self, *, code: str, state: str) -> str:
|
|
self._require_auth_config()
|
|
if not code:
|
|
raise ValueError("code must not be empty")
|
|
if not state:
|
|
raise ValueError("state must not be empty")
|
|
if not self.auth_state_store.matches_state(state):
|
|
raise TickTickAuthError("Invalid state")
|
|
|
|
body = parse.urlencode(
|
|
{
|
|
"code": code,
|
|
"grant_type": "authorization_code",
|
|
"scope": AUTH_SCOPE,
|
|
"redirect_uri": self._redirect_uri(),
|
|
}
|
|
).encode("utf-8")
|
|
req = request.Request(TICKTICK_TOKEN_URL, data=body, method="POST")
|
|
req.add_header("Content-Type", "application/x-www-form-urlencoded")
|
|
req.add_header("Authorization", self._basic_auth_header())
|
|
payload = self._send_json_request(req, operation="exchange_authorization_code")
|
|
self.auth_state_store.clear()
|
|
token = payload.get("access_token")
|
|
if not isinstance(token, str) or not token:
|
|
raise TickTickRequestError("TickTick token response did not include access_token")
|
|
return token
|
|
|
|
def get_projects(self) -> list[TickTickProject]:
|
|
self._require_token()
|
|
payload = self._authorized_json_request(
|
|
method="GET",
|
|
path="/project/",
|
|
operation="get_projects",
|
|
)
|
|
if not isinstance(payload, list):
|
|
raise TickTickRequestError("TickTick get_projects returned an unexpected payload")
|
|
return [_coerce_dataclass_payload(TickTickProject, project) for project in payload]
|
|
|
|
def get_tasks(self, project_id: str) -> list[TickTickTask]:
|
|
self._require_token()
|
|
if not project_id:
|
|
raise ValueError("project_id must not be empty")
|
|
payload = self._authorized_json_request(
|
|
method="GET",
|
|
path=f"/project/{parse.quote(project_id, safe='')}/data",
|
|
operation="get_tasks",
|
|
accepted_status_codes={200, 404},
|
|
)
|
|
if payload is None:
|
|
return []
|
|
if not isinstance(payload, dict):
|
|
raise TickTickRequestError("TickTick get_tasks returned an unexpected payload")
|
|
tasks = payload.get("tasks", [])
|
|
if not isinstance(tasks, list):
|
|
raise TickTickRequestError("TickTick get_tasks returned an invalid tasks payload")
|
|
return [_coerce_dataclass_payload(TickTickTask, task) for task in tasks]
|
|
|
|
def has_duplicate_task(self, *, project_id: str, task_title: str) -> bool:
|
|
if not task_title:
|
|
raise ValueError("task_title must not be empty")
|
|
return any(task.title == task_title for task in self.get_tasks(project_id))
|
|
|
|
def create_task(self, task: TickTickTask) -> None:
|
|
self._require_token()
|
|
if not task.projectId:
|
|
raise ValueError("task.projectId must not be empty")
|
|
if not task.title:
|
|
raise ValueError("task.title must not be empty")
|
|
if self.has_duplicate_task(project_id=task.projectId, task_title=task.title):
|
|
return
|
|
|
|
payload = {key: value for key, value in asdict(task).items() if value is not None}
|
|
self._authorized_json_request(
|
|
method="POST",
|
|
path="/task",
|
|
operation="create_task",
|
|
body=payload,
|
|
accepted_status_codes={200},
|
|
)
|
|
|
|
def _authorized_json_request(
|
|
self,
|
|
*,
|
|
method: str,
|
|
path: str,
|
|
operation: str,
|
|
body: Any | None = None,
|
|
accepted_status_codes: set[int] | None = None,
|
|
) -> Any:
|
|
url = f"{TICKTICK_OPEN_API_BASE_URL}{path}"
|
|
encoded_body = None if body is None else json.dumps(body).encode("utf-8")
|
|
req = request.Request(url, data=encoded_body, method=method)
|
|
req.add_header("Authorization", f"Bearer {self.settings.ticktick_token}")
|
|
if body is not None:
|
|
req.add_header("Content-Type", "application/json")
|
|
return self._send_json_request(
|
|
req,
|
|
operation=operation,
|
|
accepted_status_codes=accepted_status_codes,
|
|
)
|
|
|
|
def _send_json_request(
|
|
self,
|
|
req: request.Request,
|
|
*,
|
|
operation: str,
|
|
accepted_status_codes: set[int] | None = None,
|
|
) -> Any:
|
|
accepted_codes = accepted_status_codes or {200}
|
|
try:
|
|
with request.urlopen(req, timeout=self.timeout_seconds) as response:
|
|
status_code = response.getcode()
|
|
if status_code not in accepted_codes:
|
|
raise TickTickRequestError(
|
|
f"TickTick {operation} returned unexpected status {status_code}"
|
|
)
|
|
raw_body = response.read()
|
|
except error.HTTPError as exc:
|
|
if exc.code in accepted_codes:
|
|
raw_body = exc.read()
|
|
else:
|
|
logger.warning(
|
|
"TickTick %s failed with HTTP %s for %s",
|
|
operation,
|
|
exc.code,
|
|
req.full_url,
|
|
)
|
|
raise TickTickRequestError(
|
|
f"TickTick {operation} failed with HTTP {exc.code}"
|
|
) from exc
|
|
except error.URLError as exc:
|
|
logger.warning("TickTick %s failed for %s: %s", operation, req.full_url, exc)
|
|
raise TickTickRequestError(
|
|
f"TickTick {operation} failed to reach TickTick API"
|
|
) from exc
|
|
|
|
if not raw_body:
|
|
return None
|
|
try:
|
|
return json.loads(raw_body)
|
|
except json.JSONDecodeError as exc:
|
|
raise TickTickRequestError(
|
|
f"TickTick {operation} returned invalid JSON"
|
|
) from exc
|
|
|
|
def _basic_auth_header(self) -> str:
|
|
raw_credentials = f"{self._client_id()}:{self._client_secret()}"
|
|
token = base64.b64encode(raw_credentials.encode("utf-8")).decode("ascii")
|
|
return f"Basic {token}"
|
|
|
|
def _client_id(self) -> str:
|
|
return self.settings.ticktick_client_id.strip()
|
|
|
|
def _client_secret(self) -> str:
|
|
return self.settings.ticktick_client_secret.strip()
|
|
|
|
def _redirect_uri(self) -> str:
|
|
return self.settings.ticktick_redirect_uri
|
|
|
|
def _require_auth_config(self) -> None:
|
|
if not self.is_configured():
|
|
raise TickTickConfigError(
|
|
"TickTick integration is not configured. Set TICKTICK_CLIENT_ID and "
|
|
"TICKTICK_CLIENT_SECRET."
|
|
)
|
|
if not self._redirect_uri():
|
|
raise TickTickConfigError(
|
|
"TickTick integration is missing APP_HOSTNAME for OAuth callback generation."
|
|
)
|
|
|
|
def _require_token(self) -> None:
|
|
self._require_auth_config()
|
|
if self.has_token():
|
|
return
|
|
raise TickTickConfigError(
|
|
"TickTick integration is missing TICKTICK_TOKEN. Complete the OAuth flow first."
|
|
)
|
|
|