Custom actions

custom_components/rohlikcz/__init__.py

@@ -1,34 +1,41 @@
 """Rohlík CZ custom component."""
 from __future__ import annotations
 
+import logging
+
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
 from homeassistant.core import HomeAssistant
 
 from .const import DOMAIN
 from .hub import RohlikAccount
+from .services import register_services
+
+_LOGGER = logging.getLogger(__name__)
 
 PLATFORMS: list[str] = ["sensor", "binary_sensor"]
 
 
 async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     """Set up Rohlik integration from a config entry flow."""
-    account = RohlikAccount(hass, entry.data[CONF_EMAIL], entry.data[CONF_PASSWORD])  # type: ignore[Any]
+    account = RohlikAccount(hass, entry.data[CONF_EMAIL], entry.data[CONF_PASSWORD])
     await account.async_update()
 
-    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = account  # type: ignore[Any]
+    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = account
+
+    # Register services
+    register_services(hass)
 
     await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
     return True
 
 
 async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     """Unload a config entry."""
-    # This is called when an entry/configured device is to be removed. The class
-    # needs to unload itself, and remove callbacks. See the classes for further
-    # details
     unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
     if unload_ok:
-        hass.data[DOMAIN].pop(entry.entry_id)  # type: ignore[Any]
+        hass.data[DOMAIN].pop(entry.entry_id)
 
     return unload_ok
+
+

custom_components/rohlikcz/const.py

@@ -1,6 +1,8 @@
 """
 Defining constants for the project.
 """
+from __future__ import annotations
+
 from aiohttp import ClientTimeout
 from typing import Final
 
@@ -31,3 +33,17 @@
 ICON_CALENDAR_CHECK = "mdi:calendar-check"
 ICON_CALENDAR_REMOVE = "mdi:calendar-remove"
 ICON_INFO = "mdi:information-outline"
+
+""" Service attributes """
+ATTR_CONFIG_ENTRY_ID = "config_entry_id"
+ATTR_PRODUCT_ID = "product_id"
+ATTR_QUANTITY = "quantity"
+ATTR_PRODUCT_NAME = "product_name"
+ATTR_SHOPPING_LIST_ID = "shopping_list_id"
+
+""" Service names """
+SERVICE_ADD_TO_CART = "add_to_cart"
+SERVICE_SEARCH_PRODUCT = "search_product"
+SERVICE_GET_SHOPPING_LIST = "get_shopping_list"
+SERVICE_GET_CART_CONTENT = "get_cart_content"
+SERVICE_SEARCH_AND_ADD_PRODUCT = "search_and_add_to_cart"

custom_components/rohlikcz/hub.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 from collections.abc import Callable
-from typing import Any, cast
+from typing import Any, cast, List, Optional, Dict
 
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.entity import DeviceInfo
@@ -32,6 +32,11 @@ def name(self) -> str:
         """Provides name for account."""
         return self.data["login"]["data"]["user"]["name"]
 
+    @property
+    def unique_id(self) -> str:
+        """Return the unique ID for this account."""
+        return self.data["login"]["data"]["user"]["id"]
+
     async def async_update(self) -> None:
         """ Updates the data from API."""
 
@@ -52,4 +57,34 @@ async def publish_updates(self) -> None:
         for callback in self._callbacks:
             callback()
 
+    # New service methods
+    async def add_to_cart(self, product_id: int, quantity: int) -> Dict:
+        """Add a product to the shopping cart."""
+        product_list = [{"product_id": product_id, "quantity": quantity}]
+        result = await self._rohlik_api.add_to_cart(product_list)
+        return result
+
+    async def search_product(self, product_name: str) -> Optional[Dict[str, Any]]:
+        """Search for a product by name."""
+        result = await self._rohlik_api.search_product(product_name)
+        return result
+
+    async def get_shopping_list(self, shopping_list_id: str) -> Dict[str, Any]:
+        """Get a shopping list by ID."""
+        result = await self._rohlik_api.get_shopping_list(shopping_list_id)
+        return result
+
+    async def get_cart_content(self) -> Dict:
+        """ Retrieves cart content. """
+        result = await self._rohlik_api.get_cart_content()
+        return result
+
+    async def search_and_add(self, product_name: str, quantity: int) -> Dict | None:
+        """ Searches for product by name and adds to cart"""
+        searched_product = await self.search_product(product_name)
+        added_product: dict = await self.add_to_cart(searched_product["id"], quantity)
 
+        if added_product:
+            return {"added_to_cart": searched_product}
+        else:
+            return None

custom_components/rohlikcz/icons.json

@@ -0,0 +1,9 @@
+{
+  "services": {
+    "add_to_cart": {"service": "mdi:cart-plus"},
+    "search_product": {"service": "mdi:magnify"},
+    "get_shopping_list": {"service": "mdi:clipboard-list"},
+    "get_cart_content": {"service": "mdi:cart"},
+    "search_and_add_to_cart": {"service": "mdi:cart-arrow-right"}
+  }
+}

custom_components/rohlikcz/rohlik_api.py

@@ -16,7 +16,7 @@ async def example():
 import requests
 from requests import Response
 from requests.exceptions import RequestException
-from typing import TypedDict
+from typing import TypedDict, Dict
 from .errors import InvalidCredentialsError, RohlikczError, AddressNotSetError
 import asyncio
 import functools
@@ -179,12 +179,12 @@ async def get_data(self):
             # Step 3: Close the session
             await self._run_in_executor(session.close)
 
-    async def add_to_cart(self, product_list: list[Product]):
+    async def add_to_cart(self, product_list: list[dict]) -> dict:
         """
         Add multiple products to the shopping cart.
 
         Args:
-            product_list (list[Product]): A list of Product objects containing product_id and quantity for each product to be added to the cart
+            product_list (list[dict]): A list of objects containing product_id and quantity for each product to be added to the cart
         Returns:
             list: A list of product IDs that were successfully added to the cart
         """
@@ -215,15 +215,15 @@ async def add_to_cart(self, product_list: list[Product]):
                 except RequestException as err:
                     _LOGGER.error(f"Error adding {product["product_id"]} due to {err}")
 
-            return added_products
+            return {"added_products":added_products}
 
         except RequestException as err:
             _LOGGER.error(f"Request failed: {err}")
             raise ValueError("Request failed")
         finally:
             await self._run_in_executor(session.close)
 
-    async def search_product(self, product_name):
+    async def search_product(self, product_name: str):
         """
         Search for products by name and return the first matching product.
 
@@ -235,6 +235,7 @@ async def search_product(self, product_name):
         """
 
         session = requests.Session()
+        await self.login(session)
 
         try:
             search_url = "/services/frontend-service/search-metadata"
@@ -256,7 +257,13 @@ async def search_product(self, product_name):
             search_response.raise_for_status()
             search_data = search_response.json()
             if len(search_data["data"]["productList"]) > 0:
-                return search_data["data"]["productList"][0]
+                return {
+                    "id": search_data["data"]["productList"][0]["productId"],
+                    "name": search_data["data"]["productList"][0]["productName"],
+                    "price": f"{search_data["data"]["productList"][0]["price"]["full"]} {search_data["data"]["productList"][0]["price"]["currency"]}",
+                    "brand": search_data["data"]["productList"][0]["brand"],
+                    "amount": search_data["data"]["productList"][0]["textualAmount"]
+                        }
             else:
                 return None
 
@@ -266,7 +273,7 @@ async def search_product(self, product_name):
         finally:
             await self._run_in_executor(session.close)
 
-    async def get_shopping_list(self, shopping_list_id=None):
+    async def get_shopping_list(self, shopping_list_id=None) -> dict:
         """
         Retrieve a shopping list by its ID.
 
@@ -292,10 +299,63 @@ async def get_shopping_list(self, shopping_list_id=None):
             )
             search_response.raise_for_status()
             search_data = search_response.json()
-            return search_data
+            return {"name": search_data["name"], "products_in_list": search_data["products"]}
+
+        except RequestException as err:
+            _LOGGER.error(f"Request failed: {err}")
+            raise ValueError("Request failed")
+        finally:
+            await self._run_in_executor(session.close)
+
+    async def get_cart_content(self) -> Dict:
+        """
+        Fetches the current cart contents
+
+        :return: Dictionary with cart content
+        """
+
+        cart_url = "/services/frontend-service/v2/cart-review/check-cart"
+
+        session = requests.Session()
+
+        try:
+            await self.login(session)
+            cart_response = await self._run_in_executor(
+                session.get,
+                f"{BASE_URL}{cart_url}",
+            )
+            cart_response.raise_for_status()
+            cart_content = cart_response.json()
 
         except RequestException as err:
             _LOGGER.error(f"Request failed: {err}")
             raise ValueError("Request failed")
         finally:
             await self._run_in_executor(session.close)
+
+        data = cart_content.get("data", {})
+
+        # Extract the main cart information
+        cart_info = {
+            "total_price": data.get("totalPrice", 0),
+            "total_items": len(data.get("items", {})),
+            "can_make_order": data.get("submitConditionPassed", False),
+            "products": []
+        }
+
+        # Process each product item
+        for product_id, product_data in data.get("items", {}).items():
+
+            product_info = {
+                "id": product_id,
+                "cart_item_id": product_data.get("orderFieldId", ""),
+                "name": product_data.get("productName", ""),
+                "quantity": product_data.get("quantity", 0),
+                "price": product_data.get("price", 0),
+                "category_name": product_data.get("primaryCategoryName", ""),
+                "brand": product_data.get("brand", "")
+            }
+
+            cart_info["products"].append(product_info)
+
+        return cart_info