Merge pull request #10 from Open-Inflation/dev

Dev

Author: Miskler

chizhik_api/__init__.py

@@ -1,4 +1,4 @@
 from .manager import ChizhikAPI
 
 __all__ = ["ChizhikAPI"]
-__version__ = "0.2.4"
+__version__ = "0.2.5"

chizhik_api/abstraction.py

@@ -0,0 +1,4 @@
+class DeliveryMode:
+    """Я хз что это такое"""
+
+    STORE = "store"

chizhik_api/endpoints/advertising.py

@@ -22,5 +22,5 @@ async def active_inout(self) -> FetchResponse:
         """Получить активные рекламные баннеры."""
         return await self._parent._request(
             HttpMethod.GET,
-            f"{self._parent.CATALOG_URL}/catalog/unauthorized/active_inout/",
+            f"{self._parent.API_URL}/v1/catalog/unauthorized/active_inout/",
         )

chizhik_api/endpoints/catalog.py

@@ -9,6 +9,8 @@
 from human_requests import ApiChild, ApiParent, api_child_field, autotest
 from human_requests.abstraction import FetchResponse, HttpMethod
 
+from ..abstraction import DeliveryMode
+
 if TYPE_CHECKING:
     from ..manager import ChizhikAPI  # noqa: F401
 
@@ -33,11 +35,43 @@ def __init__(self, parent: "ChizhikAPI"):
     @autotest
     async def tree(self, city_id: Optional[str] = None) -> FetchResponse:
         """Получить дерево категорий."""
-        url = f"{self._parent.CATALOG_URL}/catalog/unauthorized/categories/"
+        url = f"{self._parent.API_URL}/v1/catalog/unauthorized/categories/"
         if city_id:
             url += f"?city_id={city_id}"
         return await self._parent._request(HttpMethod.GET, url)
 
+    @autotest
+    async def delivery_tree(
+        self,
+        store_id: str,
+        mode: DeliveryMode = DeliveryMode.STORE,
+        include_restrict: bool = True,
+    ):
+        url = f"{self._parent.DELIVERY_API_URL}/catalog/v3/stores/{store_id}/categories?mode={mode}&include_subcategories=1&include_restrict={str(include_restrict).lower()}"
+        return await self._parent._request(HttpMethod.GET, url)
+
+    @autotest
+    async def delivery_tree_extended(
+        self,
+        store_id: str,
+        category_alias: str,
+        mode: DeliveryMode = DeliveryMode.STORE,
+        include_restrict: bool = True,
+    ):
+        url = f"{self._parent.DELIVERY_API_URL}/catalog/v2/stores/{store_id}/categories/{category_alias}/extended?mode={mode}&include_restrict={str(include_restrict).lower()}"
+        return await self._parent._request(HttpMethod.GET, url)
+
+    @autotest
+    async def delivery_tree_ancestors(
+        self,
+        store_id: str,
+        category_alias: str,
+        mode: DeliveryMode = DeliveryMode.STORE,
+        include_restrict: bool = True,
+    ):
+        url = f"{self._parent.DELIVERY_API_URL}/catalog/v3/stores/{store_id}/categories/{category_alias}/ancestors?mode={mode}&include_restrict={str(include_restrict).lower()}"
+        return await self._parent._request(HttpMethod.GET, url)
+
     @autotest
     async def products_list(
         self,
@@ -47,7 +81,7 @@ async def products_list(
         search: Optional[str] = None,
     ) -> FetchResponse:
         """Получить список продуктов в категории."""
-        url = f"{self._parent.CATALOG_URL}/catalog/unauthorized/products/?page={page}"
+        url = f"{self._parent.API_URL}/v1/catalog/unauthorized/products/?page={page}"
         if category_id:
             url += f"&category_id={category_id}"
         if city_id:
@@ -56,6 +90,31 @@ async def products_list(
             url += f"&term={urllib.parse.quote(search)}"
         return await self._parent._request(HttpMethod.GET, url)
 
+    @autotest
+    async def delivery_products_list(
+        self,
+        store_id: str,
+        category_alias: str,
+        offset: int = 0,
+        limit: int = 499,
+        mode: DeliveryMode = DeliveryMode.STORE,
+        include_restrict: bool = True,
+    ):
+        url = f"{self._parent.DELIVERY_API_URL}/catalog/v2/stores/{store_id}/categories/{category_alias}/products?mode={mode}&include_restrict={str(include_restrict).lower()}&limit={limit}&offset={offset}"
+        return await self._parent._request(HttpMethod.GET, url)
+
+    @autotest
+    async def delivery_search(
+        self,
+        store_id: str,
+        query: str,
+        limit: int = 12,
+        mode: DeliveryMode = DeliveryMode.STORE,
+        include_restrict: bool = True,
+    ):
+        url = f"{self._parent.DELIVERY_API_URL}/catalog/v3/stores/{store_id}/search?mode={mode}&include_restrict={str(include_restrict).lower()}&q={query}&limit={limit}"
+        return await self._parent._request(HttpMethod.GET, url)
+
 
 class ProductService(ApiChild["ChizhikAPI"]):
     """Сервис для работы с товарами в каталоге."""
@@ -74,7 +133,19 @@ async def info(
             Response: Ответ от сервера с информацией о товаре.
         """
 
-        url = f"{self._parent.CATALOG_URL}/catalog/unauthorized/products/{product_id}/"
+        url = f"{self._parent.API_URL}/v1/catalog/unauthorized/products/{product_id}/"
         if city_id:
             url += f"?city_id={city_id}"
         return await self._parent._request(HttpMethod.GET, url)
+
+    @autotest
+    async def delivery_info(
+        self,
+        store_id: str,
+        product_id: int,
+        mode: DeliveryMode = DeliveryMode.STORE,
+        include_restrict: bool = True,
+    ):
+        # TODO
+        url = f"{self._parent.DELIVERY_API_URL}/catalog/v2/stores/{store_id}/products/{product_id}?mode={mode}&include_restrict={str(include_restrict).lower()}"
+        return await self._parent._request(HttpMethod.GET, url)

chizhik_api/endpoints/general.py

@@ -28,7 +28,11 @@ async def download_image(
             attempts=retry_attempts, start_timeout=3.0, max_timeout=timeout
         )
 
-        px = self._parent.proxy if isinstance(self._parent.proxy, Proxy) else Proxy(self._parent.proxy)
+        px = (
+            self._parent.proxy
+            if isinstance(self._parent.proxy, Proxy)
+            else Proxy(self._parent.proxy)
+        )
         async with RetryClient(retry_options=retry_options) as retry_client:
             async with retry_client.get(
                 url,

chizhik_api/endpoints/geolocation.py

@@ -2,26 +2,51 @@
 
 from __future__ import annotations
 
+from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
-from human_requests import ApiChild, autotest
+from human_requests import ApiChild, ApiParent, api_child_field, autotest
 from human_requests.abstraction import FetchResponse, HttpMethod
 
 if TYPE_CHECKING:
     from ..manager import ChizhikAPI  # noqa: F401
 
 
+@dataclass(init=False)
 class ClassGeolocation(ApiChild["ChizhikAPI"]):
     """Методы для работы с геолокацией и выбором магазинов.
 
     Включает получение информации о городах, адресах, поиск магазинов
     и управление настройками доставки.
     """
 
+    Shop: ShopService = api_child_field(lambda parent: ShopService(parent.parent))
+    """Сервис для работы с информацией о магазинах."""
+
+    def __init__(self, parent: "ChizhikAPI"):
+        super().__init__(parent)
+        ApiParent.__post_init__(self)
+
     @autotest
     async def cities_list(self, search_name: str, page: int = 1) -> FetchResponse:
         """Получить список городов по частичному совпадению имени."""
         return await self._parent._request(
             HttpMethod.GET,
-            f"{self._parent.CATALOG_URL}/geo/cities/?name={search_name}&page={page}",
+            f"{self._parent.API_URL}/v1/geo/cities/?name={search_name}&page={page}",
         )
+
+
+class ShopService(ApiChild["ChizhikAPI"]):
+    """Сервис для работы с информацией о магазинах."""
+
+    @autotest
+    async def all(self) -> FetchResponse:
+        """Получить список всех точек магазинов."""
+        url = f"{self._parent.API_URL}/v1/shops"
+        return await self._parent._request(HttpMethod.GET, url)
+
+    @autotest
+    async def search(self, query: str) -> FetchResponse:
+        """Получить список всех точек магазинов."""
+        url = f"{self._parent.API_URL}/v1/shops?term={query}"
+        return await self._parent._request(HttpMethod.GET, url)

chizhik_api/manager.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
-import os
+import asyncio
+from collections import defaultdict
 from dataclasses import dataclass, field
 from typing import Any
 
@@ -13,7 +14,7 @@
     api_child_field,
 )
 from human_requests.abstraction import FetchResponse, HttpMethod, Proxy
-from playwright.async_api import TimeoutError as PWTimeoutError
+from human_requests.network_analyzer.anomaly_sniffer import HeaderAnomalySniffer
 
 from .endpoints.advertising import ClassAdvertising
 from .endpoints.catalog import ClassCatalog
@@ -31,12 +32,15 @@ class ChizhikAPI(ApiParent):
     """Время ожидания ответа от сервера в миллисекундах."""
     headless: bool = True
     """Запускать браузер в headless режиме?"""
+    test_mode: bool = False
+    """Режим тестирования предполагает более глубокий _warmup который не требуется для обычного использования"""
     proxy: str | dict | Proxy | None = field(default_factory=Proxy.from_env)
     """Прокси-сервер для всех запросов (если нужен). По умолчанию берет из окружения (если есть).
     Принимает как формат Playwright, так и строчный формат."""
     browser_opts: dict[str, Any] = field(default_factory=dict)
     """Дополнительные опции для браузера (см. https://camoufox.com/python/installation/)"""
-    CATALOG_URL: str = "https://app.chizhik.club/api/v1"
+    API_URL: str = "https://app.chizhik.club/api"
+    DELIVERY_API_URL: str = "https://app.chizhik.club/delivery/api"
     """URL для работы с каталогом."""
     MAIN_SITE_URL: str = "https://chizhik.club/catalog/"
     """URL главной страницы сайта."""
@@ -51,6 +55,11 @@ class ChizhikAPI(ApiParent):
     page: HumanPage = field(init=False, repr=False)
     """Внутренний страница сессии браузера"""
 
+    unstandard_headers: dict[str, str] = field(init=False, repr=False)
+    """Список нестандартных заголовков пойманных при инициализации"""
+    unstandard_urls: dict[str, list[str]] = field(init=False, repr=False)
+    """Список нестандартных заголовков пойманных при инициализации"""
+
     Geolocation: ClassGeolocation = api_child_field(ClassGeolocation)
     """API для работы с геолокацией."""
     Catalog: ClassCatalog = api_child_field(ClassCatalog)
@@ -74,30 +83,63 @@ async def _warmup(self) -> None:
             proxy=px.as_dict(),
             **self.browser_opts,
             block_images=True,
+            i_know_what_im_doing=True,
         ).start()
 
         self.session = HumanBrowser.replace(br)
         self.ctx = await self.session.new_context()
         self.page = await self.ctx.new_page()
         self.page.on_error_screenshot_path = "screenshot.png"
-        await self.page.goto(self.CATALOG_URL, wait_until="networkidle")
-
-        ok = False
-        try_count = 3
-        while not ok and try_count > 0:
-            try_count -= 1
-            try:
-                await self.page.wait_for_selector(
-                    "pre", timeout=self.timeout_ms, state="attached"
-                )
-                ok = True
-            except PWTimeoutError:
-                await self.page.reload()
-        if not ok:
-            raise RuntimeError(await self.page.content())
-
-        # await self.page.wait_for_load_state("networkidle")
-        # await asyncio.sleep(3)
+
+        if self.test_mode:
+            sniffer = HeaderAnomalySniffer(
+                include_subresources=True,  # или False, если интересны только документы
+                url_filter=lambda u: u.startswith(self.API_URL),
+            )
+            await sniffer.start(self.ctx)
+
+            collected = {}
+
+            def on_request(request):
+                if request.url.startswith(self.API_URL):
+                    collected[request.url] = request.headers
+
+            self.ctx.on("request", on_request)
+
+            await self.page.goto(self.MAIN_SITE_URL, wait_until="networkidle")
+            await self.page.wait_for_selector("next-route-announcer", state="attached")
+            await asyncio.sleep(1)
+            await self.page.locator(
+                'main a[data-qa^="sidebar-sub-category-"][data-qa$="-link"]'
+            ).first.click()
+            await self.page.locator(
+                'main div[itemtype="https://schema.org/Product"]'
+            ).first.click()
+            await asyncio.sleep(1)
+            await self.page.wait_for_load_state("load")
+
+            await self.ctx.unroute("**/api/**", on_request)
+            result_sniffer = await sniffer.complete()
+
+            # Результат: {заголовок: [уникальные значения]}
+            result = defaultdict(set)
+
+            # Проходим по всем URL в 'request'
+            for _url, headers in result_sniffer["request"].items():
+                for header, values in headers.items():
+                    result[header].update(
+                        values
+                    )  # добавляем значения, set уберёт дубли
+
+            # Преобразуем set обратно в list
+            self.unstandard_headers = {k: list(v)[0] for k, v in result.items()}
+            self.unstandard_urls = collected
+
+        await self.page.goto(f"{self.API_URL}/v1", wait_until="networkidle")
+
+        await self.page.wait_for_selector(
+            "pre", timeout=self.timeout_ms, state="attached"
+        )
 
     async def __aexit__(self, *exc):
         """Выход из контекстного менеджера с закрытием сессии."""