home-automation-backend/src/cloud_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 requests

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

    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 = {
            "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 = requests.post(ticktick_token_url, data=ticktick_token_auth_params, auth=(client_id, client_secret), timeout=10)
        print(response)
        Config.update_env("TICKTICK_ACCESS_TOKEN", response.json().get("access_token"))
        return True

    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 = requests.get(ticktick_get_tasks_url, headers=header, timeout=10)
        return response.json()["tasks"]

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

    def create_task(self, task: TickTick.Task) -> dict[str, str]:
        if not 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}"}
            requests.post(ticktick_task_creation_url, headers=header, json=asdict(task), timeout=10)
        return {"title": task.title}

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