home-automation-backend/src/util/ticktick.py

from __future__ import annotations

import urllib.parse
from dataclasses import asdict, dataclass
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from datetime import datetime

import httpx

from src.config import Config


class TickTick:
    @dataclass
    class Task:
        projectId: str  # noqa: N815
        title: str
        dueDate: str | None = None  # noqa: N815
        content: str | None = None
        desc: str | None = None

    def __init__(self) -> None:
        print("Initializing TickTick...")
        if Config.get_env("TICKTICK_ACCESS_TOKEN") is None:
            self._begin_auth()
        else:
            self._access_token = Config.get_env("TICKTICK_ACCESS_TOKEN")

    def _begin_auth(self) -> None:
        ticktick_code_auth_url = "https://ticktick.com/oauth/authorize?"
        ticktick_code_auth_params = {
            "client_id": Config.get_env("TICKTICK_CLIENT_ID"),
            "scope": "tasks:read tasks:write",
            "state": "begin_auth",
            "redirect_uri": Config.get_env("TICKTICK_CODE_REDIRECT_URI"),
            "response_type": "code",
        }
        ticktick_auth_url_encoded = urllib.parse.urlencode(ticktick_code_auth_params)
        print("Visit: ", ticktick_code_auth_url + ticktick_auth_url_encoded, " to authenticate.")

    async def retrieve_access_token(self, code: str, state: str) -> bool:
        if state != "begin_auth":
            print("Invalid state.")
            return False
        ticktick_token_url = "https://ticktick.com/oauth/token"  # noqa: S105
        ticktick_token_auth_params: dict[str, str] = {
            "code": code,
            "grant_type": "authorization_code",
            "scope": "tasks:write tasks:read",
            "redirect_uri": Config.get_env("TICKTICK_CODE_REDIRECT_URI"),
        }
        client_id = Config.get_env("TICKTICK_CLIENT_ID")
        client_secret = Config.get_env("TICKTICK_CLIENT_SECRET")
        response = await httpx.AsyncClient().post(
            ticktick_token_url,
            data=ticktick_token_auth_params,
            auth=httpx.BasicAuth(username=client_id, password=client_secret),
            timeout=10,
        )
        Config.update_env("TICKTICK_ACCESS_TOKEN", response.json()["access_token"])
        return True

    async def get_tasks(self, project_id: str) -> list[dict]:
        ticktick_get_tasks_url = "https://api.ticktick.com/open/v1/project/" + project_id + "/data"
        header: dict[str, str] = {"Authorization": f"Bearer {self._access_token}"}
        response = await httpx.AsyncClient().get(ticktick_get_tasks_url, headers=header, timeout=10)
        return response.json()["tasks"]

    async def has_duplicate_task(self, project_id: str, task_title: str) -> bool:
        tasks = await self.get_tasks(project_id=project_id)
        return any(task["title"] == task_title for task in tasks)

    async def create_task(self, task: TickTick.Task) -> dict[str, str]:
        if not await self.has_duplicate_task(project_id=task.projectId, task_title=task.title):
            ticktick_task_creation_url = "https://api.ticktick.com/open/v1/task"
            header: dict[str, str] = {"Authorization": f"Bearer {self._access_token}"}
            await httpx.AsyncClient().post(ticktick_task_creation_url, headers=header, json=asdict(task), timeout=10)
        return {"title": task.title}

    @staticmethod
    def datetime_to_ticktick_format(datetime: datetime) -> str:
        return datetime.strftime("%Y-%m-%dT%H:%M:%S") + "+0000"