rename

src/util/ticktick.py

@@ -0,0 +1,84 @@
+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"