Chris Coutinho
a11ae9c027
Enable ruff PLC0415 rule for all source files (tests excluded via per-file-ignores). Move 136 inline imports to top-level across 33 files. 8 imports suppressed with noqa for legitimate reasons: circular dependencies (client/__init__.py, context.py), optional dependency guards (app.py document processors, auth/userinfo_routes.py), and post-env-setup imports (smithery_main.py). Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
245 lines
7.4 KiB
Python
245 lines
7.4 KiB
Python
"""Client for Nextcloud Cookbook app operations."""
|
|
|
|
import logging
|
|
from typing import Any, Dict, List
|
|
from urllib.parse import quote
|
|
|
|
from httpx import Timeout
|
|
|
|
from .base import BaseNextcloudClient
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class CookbookClient(BaseNextcloudClient):
|
|
"""Client for Nextcloud Cookbook app operations."""
|
|
|
|
app_name = "cookbook"
|
|
|
|
async def get_version(self) -> Dict[str, Any]:
|
|
"""Get Cookbook app and API version."""
|
|
response = await self._make_request("GET", "/apps/cookbook/api/version")
|
|
return response.json()
|
|
|
|
async def get_config(self) -> Dict[str, Any]:
|
|
"""Get current Cookbook app configuration."""
|
|
response = await self._make_request("GET", "/apps/cookbook/api/v1/config")
|
|
return response.json()
|
|
|
|
async def set_config(self, config: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Set Cookbook app configuration.
|
|
|
|
Args:
|
|
config: Configuration dictionary with fields like:
|
|
- folder: Recipe folder path
|
|
- update_interval: Auto-rescan interval in minutes
|
|
- print_image: Whether to print images with recipes
|
|
- visibleInfoBlocks: Visible info blocks configuration
|
|
|
|
Returns:
|
|
Response with status message
|
|
"""
|
|
response = await self._make_request(
|
|
"POST", "/apps/cookbook/api/v1/config", json=config
|
|
)
|
|
return response.json()
|
|
|
|
async def reindex(self) -> str:
|
|
"""Trigger a rescan of all recipes into the caching database.
|
|
|
|
Returns:
|
|
Success message
|
|
"""
|
|
response = await self._make_request("POST", "/apps/cookbook/api/v1/reindex")
|
|
return response.json()
|
|
|
|
async def list_recipes(self) -> List[Dict[str, Any]]:
|
|
"""Get all recipes in the database.
|
|
|
|
Returns:
|
|
List of recipe stubs with basic information
|
|
"""
|
|
response = await self._make_request("GET", "/apps/cookbook/api/v1/recipes")
|
|
return response.json()
|
|
|
|
async def get_recipe(self, recipe_id: int) -> Dict[str, Any]:
|
|
"""Get a single recipe by ID.
|
|
|
|
Args:
|
|
recipe_id: The recipe ID
|
|
|
|
Returns:
|
|
Full recipe data
|
|
"""
|
|
response = await self._make_request(
|
|
"GET", f"/apps/cookbook/api/v1/recipes/{recipe_id}"
|
|
)
|
|
return response.json()
|
|
|
|
async def create_recipe(self, recipe_data: Dict[str, Any]) -> int:
|
|
"""Create a new recipe.
|
|
|
|
Args:
|
|
recipe_data: Recipe data following schema.org/Recipe format.
|
|
Required: name
|
|
Optional: description, ingredients, instructions, etc.
|
|
|
|
Returns:
|
|
ID of the newly created recipe
|
|
"""
|
|
response = await self._make_request(
|
|
"POST", "/apps/cookbook/api/v1/recipes", json=recipe_data
|
|
)
|
|
return response.json()
|
|
|
|
async def update_recipe(self, recipe_id: int, recipe_data: Dict[str, Any]) -> int:
|
|
"""Update an existing recipe.
|
|
|
|
Args:
|
|
recipe_id: The recipe ID to update
|
|
recipe_data: Updated recipe data
|
|
|
|
Returns:
|
|
ID of the updated recipe
|
|
"""
|
|
response = await self._make_request(
|
|
"PUT", f"/apps/cookbook/api/v1/recipes/{recipe_id}", json=recipe_data
|
|
)
|
|
return response.json()
|
|
|
|
async def delete_recipe(self, recipe_id: int) -> str:
|
|
"""Delete a recipe.
|
|
|
|
Args:
|
|
recipe_id: The recipe ID to delete
|
|
|
|
Returns:
|
|
Success message
|
|
"""
|
|
response = await self._make_request(
|
|
"DELETE", f"/apps/cookbook/api/v1/recipes/{recipe_id}"
|
|
)
|
|
return response.json()
|
|
|
|
async def import_recipe(self, url: str) -> Dict[str, Any]:
|
|
"""Import a recipe from a URL using schema.org metadata.
|
|
|
|
Args:
|
|
url: URL of the recipe to import
|
|
|
|
Returns:
|
|
Full imported recipe data
|
|
"""
|
|
logger.info(f"Importing recipe from URL: {url}")
|
|
response = await self._make_request(
|
|
"POST",
|
|
"/apps/cookbook/api/v1/import",
|
|
json={"url": url},
|
|
timeout=Timeout(300.0),
|
|
)
|
|
return response.json()
|
|
|
|
async def get_recipe_image(self, recipe_id: int, size: str = "full") -> bytes:
|
|
"""Get the main image of a recipe.
|
|
|
|
Args:
|
|
recipe_id: The recipe ID
|
|
size: Image size - "full", "thumb" (250px), or "thumb16" (16px)
|
|
|
|
Returns:
|
|
Image bytes
|
|
"""
|
|
response = await self._make_request(
|
|
"GET",
|
|
f"/apps/cookbook/api/v1/recipes/{recipe_id}/image",
|
|
params={"size": size},
|
|
)
|
|
return response.content
|
|
|
|
async def search_recipes(self, query: str) -> List[Dict[str, Any]]:
|
|
"""Search for recipes by keywords, tags, and categories.
|
|
|
|
Args:
|
|
query: Search string (URL-encoded, space/comma separated)
|
|
|
|
Returns:
|
|
List of matching recipe stubs
|
|
"""
|
|
encoded_query = quote(query)
|
|
response = await self._make_request(
|
|
"GET", f"/apps/cookbook/api/v1/search/{encoded_query}"
|
|
)
|
|
return response.json()
|
|
|
|
async def list_categories(self) -> List[Dict[str, Any]]:
|
|
"""Get all known categories.
|
|
|
|
Note: A category name of '*' indicates recipes with no category.
|
|
|
|
Returns:
|
|
List of categories with recipe counts
|
|
"""
|
|
response = await self._make_request("GET", "/apps/cookbook/api/v1/categories")
|
|
return response.json()
|
|
|
|
async def get_recipes_in_category(self, category: str) -> List[Dict[str, Any]]:
|
|
"""Get all recipes in a specific category.
|
|
|
|
Args:
|
|
category: Category name (use "_" for recipes with no category)
|
|
|
|
Returns:
|
|
List of recipe stubs in the category
|
|
"""
|
|
encoded_category = quote(category)
|
|
response = await self._make_request(
|
|
"GET", f"/apps/cookbook/api/v1/category/{encoded_category}"
|
|
)
|
|
return response.json()
|
|
|
|
async def rename_category(self, old_name: str, new_name: str) -> str:
|
|
"""Rename a category.
|
|
|
|
Args:
|
|
old_name: Current category name
|
|
new_name: New category name
|
|
|
|
Returns:
|
|
New category name
|
|
"""
|
|
encoded_old_name = quote(old_name)
|
|
response = await self._make_request(
|
|
"PUT",
|
|
f"/apps/cookbook/api/v1/category/{encoded_old_name}",
|
|
json={"name": new_name},
|
|
)
|
|
return response.json()
|
|
|
|
async def list_keywords(self) -> List[Dict[str, Any]]:
|
|
"""Get all known keywords/tags.
|
|
|
|
Returns:
|
|
List of keywords with recipe counts
|
|
"""
|
|
response = await self._make_request("GET", "/apps/cookbook/api/v1/keywords")
|
|
return response.json()
|
|
|
|
async def get_recipes_with_keywords(
|
|
self, keywords: List[str]
|
|
) -> List[Dict[str, Any]]:
|
|
"""Get all recipes associated with certain keywords.
|
|
|
|
Args:
|
|
keywords: List of keywords to filter by
|
|
|
|
Returns:
|
|
List of recipe stubs matching the keywords
|
|
"""
|
|
# Join keywords with commas
|
|
keywords_str = ",".join(keywords)
|
|
encoded_keywords = quote(keywords_str)
|
|
response = await self._make_request(
|
|
"GET", f"/apps/cookbook/api/v1/tags/{encoded_keywords}"
|
|
)
|
|
return response.json()
|