home-automation/app/integrations/ticktick.py

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."
        )