Add homeassistant

src/cloud_util/homeassistant.py

@@ -0,0 +1,31 @@
+from pydantic import BaseModel
+
+from src.cloud_util.ticktick import TickTick
+from src.config import Config
+
+
+class HomeAssistant:
+    class PublishMessage(BaseModel):
+        target: str
+        action: str
+        content: str
+
+    def __init__(self, ticktick: TickTick) -> None:
+        self._ticktick = ticktick
+
+    def process_publish_message(self, message: PublishMessage) -> dict[str, str]:
+        if message.target == "ticktick":
+            return self._process_ticktick_message(message=message)
+
+        return {"Status": "Unknown target"}
+
+    def _process_ticktick_message(self, message: PublishMessage) -> dict[str, str]:
+        if message.action == "create_shopping_list":
+            return self._create_shopping_list(item=message.content)
+
+        return {"Status": "Unknown action"}
+
+    def _create_shopping_list(self, item: str) -> dict[str, str]:
+        project_id = Config.get_env("TICKTICK_SHOPPING_LIST")
+        task = TickTick.Task(projectId=project_id, title=item)
+        return self._ticktick.create_task(task=task)

src/cloud_util/ticktick.py

@@ -1,4 +1,11 @@
+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
 
@@ -6,10 +13,20 @@ 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?"
@@ -37,5 +54,26 @@ class TickTick:
         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"

src/main.py

@@ -3,6 +3,7 @@ from contextlib import asynccontextmanager
 from fastapi import FastAPI
 from pydantic import BaseModel
 
+from src.cloud_util.homeassistant import HomeAssistant
 from src.cloud_util.mqtt import MQTT
 from src.cloud_util.notion import NotionAsync
 from src.cloud_util.ticktick import TickTick
@@ -15,6 +16,7 @@ ticktick = TickTick()
 notion = NotionAsync(token=Config.get_env(key="NOTION_TOKEN"))
 mqtt = MQTT()
 poo_recorder = PooRecorder(mqtt=mqtt, notion=notion)
+homeassistant = HomeAssistant(ticktick=ticktick)
 
 
 @asynccontextmanager
@@ -31,6 +33,16 @@ class PooRecordField(BaseModel):
 app = FastAPI(lifespan=_lifespan)
 
 
+@app.get("/homeassistant/status")
+async def get_status() -> dict:
+    return {"Status": "Ok"}
+
+
+@app.post("/homeassistant/publish")
+async def homeassistant_publish(payload: HomeAssistant.PublishMessage) -> dict:
+    return homeassistant.process_publish_message(message=payload)
+
+
 # Poo recorder
 @app.post("/poo/record")
 async def record(record_detail: PooRecordField) -> PooRecordField: