79 lines
3.2 KiB
Python
79 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
from urllib import error, parse, request
|
|
|
|
from app.config import settings
|
|
|
|
|
|
class KitchenOwlService:
|
|
def __init__(self) -> None:
|
|
self.base_url = settings.kitchenowl_base_url.rstrip("/")
|
|
self.token = settings.kitchenowl_api_token
|
|
self.household_id = settings.kitchenowl_household_id
|
|
|
|
def configured(self) -> bool:
|
|
return bool(self.base_url and self.token and self.household_id)
|
|
|
|
def _request(self, method: str, path: str, *, params: dict[str, Any] | None = None, body: dict[str, Any] | None = None, timeout: int = 30):
|
|
url = f"{self.base_url}{path}"
|
|
if params:
|
|
url += ("&" if "?" in url else "?") + parse.urlencode(params, doseq=True)
|
|
data = None
|
|
headers = {
|
|
"Authorization": f"Bearer {self.token}",
|
|
"Accept": "application/json",
|
|
"User-Agent": "doris-kitchen/1.0",
|
|
}
|
|
if body is not None:
|
|
data = json.dumps(body).encode("utf-8")
|
|
headers["Content-Type"] = "application/json"
|
|
req = request.Request(url, data=data, headers=headers, method=method.upper())
|
|
try:
|
|
with request.urlopen(req, timeout=timeout) as resp:
|
|
raw = resp.read().decode("utf-8", "replace")
|
|
content_type = resp.headers.get("Content-Type", "")
|
|
parsed = json.loads(raw) if "json" in content_type and raw else raw
|
|
return resp.status, parsed
|
|
except error.HTTPError as exc:
|
|
raw = exc.read().decode("utf-8", "replace")
|
|
try:
|
|
parsed = json.loads(raw)
|
|
except Exception:
|
|
parsed = raw
|
|
return exc.code, parsed
|
|
|
|
def list_recipes(self) -> list[dict[str, Any]]:
|
|
if not self.configured():
|
|
return []
|
|
status, payload = self._request("GET", f"/api/household/{self.household_id}/recipe")
|
|
if status == 200 and isinstance(payload, list):
|
|
return payload
|
|
return []
|
|
|
|
def find_existing(self, payload: dict[str, Any]) -> dict[str, Any] | None:
|
|
want_source = str(payload.get("source") or "").strip().lower()
|
|
want_name = str(payload.get("name") or "").strip().lower()
|
|
for recipe in self.list_recipes():
|
|
source = str(recipe.get("source") or "").strip().lower()
|
|
name = str(recipe.get("name") or "").strip().lower()
|
|
if want_source and source and source == want_source:
|
|
return recipe
|
|
if want_name and name == want_name:
|
|
return recipe
|
|
return None
|
|
|
|
def create_recipe(self, payload: dict[str, Any]) -> dict[str, Any]:
|
|
if not self.configured():
|
|
raise RuntimeError("KitchenOwl credentials are not configured.")
|
|
existing = self.find_existing(payload)
|
|
if existing:
|
|
return {"status": "duplicate", "recipe": existing}
|
|
status, created = self._request("POST", f"/api/household/{self.household_id}/recipe", body=payload, timeout=45)
|
|
return {"status": "created" if status == 200 else "error", "http_status": status, "recipe": created}
|
|
|
|
|
|
kitchenowl = KitchenOwlService()
|
|
|