Add ticktick auth

requirements.txt

@@ -1,6 +1,7 @@
 annotated-types==0.7.0
 anyio==4.4.0
 certifi==2024.7.4
+charset-normalizer==3.3.2
 click==8.1.7
 dnspython==2.6.1
 email_validator==2.2.0
@@ -28,12 +29,14 @@ pytest==8.3.2
 python-dotenv==1.0.1
 python-multipart==0.0.9
 PyYAML==6.0.1
+requests==2.32.3
 rich==13.7.1
 shellingham==1.5.4
 sniffio==1.3.1
 starlette==0.37.2
 typer==0.12.3
 typing_extensions==4.12.2
+urllib3==2.2.2
 uvicorn==0.30.1
 uvloop==0.19.0
 watchfiles==0.22.0

ruff.toml

@@ -4,7 +4,7 @@ line-length = 144
 [lint]
 select = ["ALL"]
 fixable = ["UP034", "I001"]
-ignore = ["T201", "D", "ANN101"]
+ignore = ["T201", "D", "ANN101", "TD002", "TD003"]
 
 [lint.extend-per-file-ignores]
 "test*.py" = ["S101"]

src/cloud_util/ticktick.py

@@ -0,0 +1,41 @@
+import urllib.parse
+
+import requests
+
+from src.config import Config
+
+
+class TickTick:
+    def __init__(self) -> None:
+        print("Initializing TickTick...")
+        if Config.get_env("TICKTICK_ACCESS_TOKEN") is None:
+            self._begin_auth()
+
+    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)
+        Config.update_env("TICKTICK_ACCESS_TOKEN", response.json().get("access_token"))
+        return True

src/config.py

@@ -16,7 +16,7 @@ DOT_ENV_PATH.touch(mode=0o600, exist_ok=True)
 class Config:
     env_dict: ClassVar[OrderedDict[str, str]] = {}
     dot_env_path = DOT_ENV_PATH
-    VERSION = "1.5"
+    VERSION = "1.6"
 
     @staticmethod
     def init(dotenv_path: str = DOT_ENV_PATH) -> None:

src/main.py

@@ -4,11 +4,13 @@ from fastapi import FastAPI
 from pydantic import BaseModel
 
 from src.cloud_util.mqtt import MQTT
+from src.cloud_util.ticktick import TickTick
 from src.config import Config
 from src.recorder.poo import PooRecorder
 
 Config.init()
 
+ticktick = TickTick()
 mqtt = MQTT()
 poo_recorder = PooRecorder(mqtt)
 
@@ -31,3 +33,10 @@ app = FastAPI(lifespan=_lifespan)
 async def record(record_detail: PooRecordField) -> PooRecordField:
     await poo_recorder.record(record_detail.status)
     return record_detail
+
+
+@app.get("/ticktick/auth/code")
+async def ticktick_auth(code: str, state: str) -> dict:
+    if ticktick.retrieve_access_token(code, state):
+        return {"State": "Token Retrieved"}
+    return {"State": "Token Retrieval Failed"}