120 lines
4.2 KiB
Python
120 lines
4.2 KiB
Python
from typing import Any
|
|
|
|
from app.services.store import JsonStore
|
|
|
|
|
|
DEFAULT_PROFILE = {
|
|
"id": "household",
|
|
"household_name": "Stafford's Kitchen",
|
|
"preference_mode": "household",
|
|
"soft_blacklist": [
|
|
{"name": "lamb", "enabled": True},
|
|
{"name": "mutton", "enabled": True},
|
|
{"name": "offal", "enabled": True},
|
|
{"name": "fish", "enabled": True},
|
|
{"name": "seafood", "enabled": True},
|
|
{"name": "very spicy", "enabled": True},
|
|
{"name": "extra hot", "enabled": True},
|
|
{"name": "ghost pepper", "enabled": True},
|
|
],
|
|
"approved_ingredients": {},
|
|
"rejected_ingredients": {},
|
|
"approved_tags": {},
|
|
"rejected_tags": {},
|
|
"approved_titles": {},
|
|
"rejected_titles": {},
|
|
"auto_import_threshold": 96,
|
|
"auto_import_enabled": True,
|
|
"seed_queries": [
|
|
"budget weeknight dinners with ground beef and potatoes",
|
|
"mild asian inspired ground beef potatoes",
|
|
"easy chicken rice dinner",
|
|
"family pasta bake not spicy",
|
|
"cheap skillet dinner potatoes",
|
|
],
|
|
"notes": "",
|
|
}
|
|
|
|
|
|
class JsonRepository:
|
|
backend = "json"
|
|
|
|
def __init__(self) -> None:
|
|
self.store = JsonStore()
|
|
|
|
def get_profile(self) -> dict[str, Any]:
|
|
profile = self.store.load("profile", None)
|
|
if profile:
|
|
return profile
|
|
return self.store.save("profile", dict(DEFAULT_PROFILE))
|
|
|
|
def save_profile(self, profile: dict[str, Any]) -> dict[str, Any]:
|
|
return self.store.save("profile", profile)
|
|
|
|
def list_suggestions(self) -> list[dict[str, Any]]:
|
|
return self.store.load("suggestions", [])
|
|
|
|
def upsert_suggestion(self, item: dict[str, Any]) -> dict[str, Any]:
|
|
current = self.list_suggestions()
|
|
for idx, existing in enumerate(current):
|
|
if existing.get("url") == item.get("url"):
|
|
merged = {**item}
|
|
for key in (
|
|
"status",
|
|
"decision_at",
|
|
"kitchenowl_status",
|
|
"kitchenowl_result",
|
|
):
|
|
if existing.get(key) is not None:
|
|
merged[key] = existing.get(key)
|
|
if existing.get("id"):
|
|
merged["id"] = existing["id"]
|
|
if existing.get("created_at"):
|
|
merged["created_at"] = existing["created_at"]
|
|
current[idx] = merged
|
|
self.store.save("suggestions", current)
|
|
return merged
|
|
current.append(item)
|
|
self.store.save("suggestions", current)
|
|
return item
|
|
|
|
def update_suggestion(self, suggestion_id: str, updates: dict[str, Any]) -> dict[str, Any] | None:
|
|
return self.store.update_collection_item(
|
|
"suggestions",
|
|
lambda item: str(item.get("id")) == str(suggestion_id),
|
|
lambda item: {**item, **updates},
|
|
)
|
|
|
|
def list_meal_plans(self) -> list[dict[str, Any]]:
|
|
return self.store.load("meal_plans", [])
|
|
|
|
def create_meal_plan(self, item: dict[str, Any]) -> dict[str, Any]:
|
|
return self.store.append_collection("meal_plans", item)
|
|
|
|
def recent_suggestions(self, limit: int = 12) -> list[dict[str, Any]]:
|
|
items = sorted(self.list_suggestions(), key=lambda item: item.get("updated_at") or item.get("created_at") or "", reverse=True)
|
|
return items[:limit]
|
|
|
|
def list_failed_searches(self) -> list[dict[str, Any]]:
|
|
return self.store.load("failed_searches", [])
|
|
|
|
def log_failed_search(self, item: dict[str, Any]) -> dict[str, Any]:
|
|
current = self.list_failed_searches()
|
|
current.append(item)
|
|
current = sorted(current, key=lambda entry: entry.get("created_at") or "", reverse=True)[:50]
|
|
self.store.save("failed_searches", current)
|
|
return item
|
|
|
|
def recent_failed_searches(self, limit: int = 12) -> list[dict[str, Any]]:
|
|
return self.list_failed_searches()[:limit]
|
|
|
|
def update_failed_search(self, failed_search_id: str, updates: dict[str, Any]) -> dict[str, Any] | None:
|
|
return self.store.update_collection_item(
|
|
"failed_searches",
|
|
lambda item: str(item.get("id")) == str(failed_search_id),
|
|
lambda item: {**item, **updates},
|
|
)
|
|
|
|
|
|
repo = JsonRepository()
|