From d77941a9a755e2577911c5b381dde6b024fbf0a2 Mon Sep 17 00:00:00 2001 From: q Date: Mon, 9 Mar 2026 11:04:49 +0300 Subject: [PATCH] first commit --- .dockerignore | 19 + .gitignore | 17 + Dockerfile | 16 + README.md | 142 ++++++ docker-compose.yml | 23 + magnit_api.py | 652 +++++++++++++++++++++++++ nginx/magnit-api.location.conf.example | 28 ++ requirements.txt | 1 + 8 files changed, 898 insertions(+) create mode 100644 .dockerignore create mode 100644 .gitignore create mode 100644 Dockerfile create mode 100644 README.md create mode 100644 docker-compose.yml create mode 100644 magnit_api.py create mode 100644 nginx/magnit-api.location.conf.example create mode 100644 requirements.txt diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..cb7bcb4 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,19 @@ +# Secrets and local runtime state +magnit_session.json +.env +.env.* +*.log + +# Python +__pycache__/ +*.py[cod] +*.pyo +*.pyd +.venv/ +venv/ +.pytest_cache/ +.mypy_cache/ + +# VCS and local tooling +.git/ +.gitignore diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..071d1db --- /dev/null +++ b/.gitignore @@ -0,0 +1,17 @@ +# Secrets and local runtime state +magnit_session.json +.env +.env.* + +# Python +__pycache__/ +*.py[cod] +*.pyo +*.pyd +.venv/ +venv/ +.pytest_cache/ +.mypy_cache/ + +# Logs +*.log diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..9cd155a --- /dev/null +++ b/Dockerfile @@ -0,0 +1,16 @@ +FROM python:3.13-slim + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 + +WORKDIR /app + +COPY requirements.txt /app/requirements.txt +RUN pip install --no-cache-dir -r /app/requirements.txt + +COPY magnit_api.py /app/magnit_api.py +COPY README.md /app/README.md + +EXPOSE 8080 + +CMD ["python3", "magnit_api.py", "--session-file", "/app/magnit_session.json", "serve", "--host", "0.0.0.0", "--port", "8080"] diff --git a/README.md b/README.md new file mode 100644 index 0000000..84c2fbf --- /dev/null +++ b/README.md @@ -0,0 +1,142 @@ +# Magnit price API + +Локальный HTTP API и CLI для `magnit.ru`, который: + +- принимает `store_code` и `query`; +- ищет товары через backend `POST /webgate/v2/goods/search`; +- сам обновляет `access token` через `POST /magnit-id/v1/auth/token/refresh`; +- хранит рабочую сессию в `magnit_session.json`. + +## Быстрый старт + +Запуск API: + +```bash +python3 magnit_api.py serve --host 127.0.0.1 --port 8080 +``` + +## Docker Compose + +Сборка и запуск: + +```bash +cp .env.example .env +docker compose up -d --build +``` + +По умолчанию сервис публикуется только в loopback на `127.0.0.1:18080`. + +Проверка: + +```bash +curl "http://127.0.0.1:18080/health" +curl "http://127.0.0.1:18080/price?store_code=618224&query=Red%20Bull" +``` + +Если нужен другой внешний порт: + +```bash +MAGNIT_API_PORT=18081 docker compose up -d --build +``` + +Проверка: + +```bash +curl "http://127.0.0.1:18081/price?store_code=618224&query=Red%20Bull" +``` + +CLI без HTTP-сервера: + +```bash +python3 magnit_api.py price --store-code 618224 --query "Red Bull" +python3 magnit_api.py search --store-code 618224 --query "Red Bull" +``` + +## Endpoint'ы + +`GET /health` + +`GET /session` + +`POST /session/refresh` + +`POST /session/bootstrap` + +```json +{ + "mg_at": "%7B%22access%22%3A%22...%22%2C%22refresh%22%3A%22...%22%7D", + "device_id": "761fa244-97f7-4b6e-b93a-8c258f357580", + "device_tag": "8894ea27-7a01-42f0-8500-20eb63bbc7c9_1" +} +``` + +`GET /search?store_code=618224&query=Red%20Bull` + +`GET /price?store_code=618224&query=Red%20Bull` + +## Nginx proxy_pass + +Готовый пример лежит в `nginx/magnit-api.location.conf.example`. + +Если `nginx` стоит на хосте и `docker compose` публикует API на `127.0.0.1:18080`, используйте: + +```nginx +location /magnit-api/ { + proxy_pass http://127.0.0.1:18080/; + proxy_http_version 1.1; + + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + proxy_set_header X-Forwarded-Host $host; +} +``` + +Тогда внешний запрос будет таким: + +```bash +curl "https://your-domain.example/magnit-api/price?store_code=618224&query=Red%20Bull" +``` + +Если `nginx` контейнерный и находится в одной сети с `magnit-api`, замените upstream на `http://magnit-api:8080/`. + +## Что хранится в сессии + +`magnit_session.json` содержит: + +- `refresh_token` +- `access_token` +- `device_id` (`mg_udi`) +- `device_tag` (`mg_ksid`) +- служебные заголовки браузера + +Скрипт сам обновляет `access_token`, когда JWT подходит к истечению. + +## Как перебутстрапить сессию + +Если `refresh_token` станет невалидным, откройте `magnit.ru`, где вы уже авторизованы, и в DevTools Console выполните: + +```js +copy(JSON.stringify({ + mg_at: decodeURIComponent(document.cookie.split('; ').find(v => v.startsWith('mg_at=')).split('=').slice(1).join('=')), + device_id: document.cookie.split('; ').find(v => v.startsWith('mg_udi=')).split('=').slice(1).join('='), + device_tag: document.cookie.split('; ').find(v => v.startsWith('mg_ksid=')).split('=').slice(1).join('=') +}, null, 2)); +``` + +Потом отправьте это в локальный API: + +```bash +curl -X POST "http://127.0.0.1:8080/session/bootstrap" \ + -H "Content-Type: application/json" \ + -d @payload.json +``` + +Если API запущен через `docker compose`, используйте опубликованный порт, по умолчанию `18080`: + +```bash +curl -X POST "http://127.0.0.1:18080/session/bootstrap" \ + -H "Content-Type: application/json" \ + -d @payload.json +``` diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..d74d28c --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,23 @@ +services: + magnit-api: + build: + context: . + dockerfile: Dockerfile + container_name: magnit-api + restart: unless-stopped + ports: + - "127.0.0.1:${MAGNIT_API_PORT:-18080}:8080" + volumes: + - ./magnit_session.json:/app/magnit_session.json:rw + healthcheck: + test: + [ + "CMD", + "python3", + "-c", + "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8080/health', timeout=5).read()", + ] + interval: 30s + timeout: 5s + retries: 3 + start_period: 10s diff --git a/magnit_api.py b/magnit_api.py new file mode 100644 index 0000000..8a8c1de --- /dev/null +++ b/magnit_api.py @@ -0,0 +1,652 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import base64 +import json +import re +import time +from dataclasses import asdict, dataclass +from difflib import SequenceMatcher +from http import HTTPStatus +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from typing import Any +from urllib.parse import parse_qs, quote, unquote, urlparse + +import requests + +DEFAULT_BASE_URL = "https://magnit.ru" +DEFAULT_AUDIENCE = "loyalty-web" +DEFAULT_APP_VERSION = "2026.3.6-16.4" +DEFAULT_PLATFORM_VERSION = "Linux Chrome 144" +DEFAULT_USER_AGENT = ( + "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/144.0.0.0 Safari/537.36" +) +DEFAULT_TIMEOUT_SECONDS = 20 +TOKEN_REFRESH_SKEW_SECONDS = 90 +DEFAULT_STORE_TYPE = "express" +DEFAULT_CATALOG_TYPE = "3" +SESSION_FILE = Path(__file__).with_name("magnit_session.json") + + +class MagnitError(RuntimeError): + """Raised when Magnit backend rejects the request or the local session is invalid.""" + + +@dataclass +class SessionConfig: + access_token: str = "" + refresh_token: str = "" + device_id: str = "" + device_tag: str = "" + app_version: str = DEFAULT_APP_VERSION + platform_version: str = DEFAULT_PLATFORM_VERSION + user_agent: str = DEFAULT_USER_AGENT + audience: str = DEFAULT_AUDIENCE + base_url: str = DEFAULT_BASE_URL + + @classmethod + def load(cls, path: Path) -> "SessionConfig": + if not path.exists(): + raise MagnitError( + f"Файл сессии не найден: {path}. " + "Создайте его через POST /session/bootstrap или команду bootstrap." + ) + return cls(**json.loads(path.read_text(encoding="utf-8"))) + + @classmethod + def load_or_default(cls, path: Path) -> "SessionConfig": + if not path.exists(): + return cls() + return cls.load(path) + + def save(self, path: Path) -> None: + path.write_text( + json.dumps(asdict(self), ensure_ascii=False, indent=2) + "\n", + encoding="utf-8", + ) + + def sanitized(self) -> dict[str, Any]: + return { + "has_access_token": bool(self.access_token), + "has_refresh_token": bool(self.refresh_token), + "access_token_expires_at": decode_jwt_exp(self.access_token), + "device_id": self.device_id, + "device_tag": self.device_tag, + "app_version": self.app_version, + "platform_version": self.platform_version, + "audience": self.audience, + "base_url": self.base_url, + } + + +def decode_jwt_exp(token: str) -> int | None: + if not token or token.count(".") != 2: + return None + try: + payload_part = token.split(".")[1] + payload_part += "=" * (-len(payload_part) % 4) + payload = json.loads(base64.urlsafe_b64decode(payload_part.encode("ascii"))) + except Exception: + return None + exp = payload.get("exp") + return int(exp) if isinstance(exp, (int, float)) else None + + +def token_is_expiring(token: str, skew_seconds: int = TOKEN_REFRESH_SKEW_SECONDS) -> bool: + exp = decode_jwt_exp(token) + if exp is None: + return True + return exp <= int(time.time()) + skew_seconds + + +def parse_bool(value: Any, default: bool = False) -> bool: + if value is None: + return default + if isinstance(value, bool): + return value + return str(value).strip().lower() in {"1", "true", "yes", "y", "on"} + + +def normalize_text(value: str) -> str: + return re.sub(r"\s+", " ", re.sub(r"[^0-9a-zа-яё]+", " ", value.lower())).strip() + + +def money_minor_to_major(value: int | None) -> float | None: + if value is None: + return None + return round(value / 100, 2) + + +def normalize_item(item: dict[str, Any], store_type: str) -> dict[str, Any]: + promotion = item.get("promotion") or {} + store_code = item.get("storeCode") + product_id = item.get("productId") or item.get("id") + return { + "product_id": product_id, + "name": item.get("name"), + "price_minor": item.get("price"), + "price": money_minor_to_major(item.get("price")), + "old_price_minor": promotion.get("oldPrice"), + "old_price": money_minor_to_major(promotion.get("oldPrice")), + "discount_percent": promotion.get("discountPercent"), + "promotion_active": promotion.get("isPromotion"), + "promotion_end": promotion.get("endDate"), + "store_code": store_code, + "store_type": store_type, + "catalog_type": item.get("catalogType"), + "quantity": item.get("quantity"), + "pickup_only": item.get("pickupOnly"), + "is_for_adults": item.get("isForAdults"), + "need_passport": item.get("needPassport"), + "seo_code": item.get("seoCode"), + "service": item.get("service"), + "url": ( + f"{DEFAULT_BASE_URL}/product/{product_id}?shopCode={store_code}&shopType={store_type}" + if product_id and store_code + else None + ), + "rating": (item.get("ratings") or {}).get("rating"), + "comments_count": (item.get("ratings") or {}).get("commentsCount"), + } + + +def score_item(query: str, item_name: str) -> float: + query_norm = normalize_text(query) + item_norm = normalize_text(item_name) + if not query_norm or not item_norm: + return 0.0 + score = SequenceMatcher(None, query_norm, item_norm).ratio() + if query_norm in item_norm: + score += 1.0 + query_tokens = set(query_norm.split()) + item_tokens = set(item_norm.split()) + score += 0.15 * len(query_tokens & item_tokens) + return score + + +def build_best_match(query: str, items: list[dict[str, Any]]) -> dict[str, Any] | None: + if not items: + return None + ranked = sorted(items, key=lambda item: score_item(query, item.get("name", "")), reverse=True) + return ranked[0] + + +def parse_mg_at_cookie(mg_at_value: str) -> dict[str, str]: + raw = unquote(mg_at_value.strip()) + if not raw: + raise MagnitError("Передан пустой mg_at.") + try: + payload = json.loads(raw) + except json.JSONDecodeError as exc: + raise MagnitError("Не удалось распарсить mg_at. Нужен raw cookie value или JSON.") from exc + access_token = payload.get("access", "") + refresh_token = payload.get("refresh", "") + if not refresh_token: + raise MagnitError("В mg_at отсутствует refresh token.") + return {"access_token": access_token, "refresh_token": refresh_token} + + +class MagnitClient: + def __init__(self, session_path: Path = SESSION_FILE, timeout_seconds: int = DEFAULT_TIMEOUT_SECONDS): + self.session_path = session_path + self.timeout_seconds = timeout_seconds + self.session_config = SessionConfig.load(session_path) + self.http = requests.Session() + self._sync_cookies() + + def session_status(self) -> dict[str, Any]: + return self.session_config.sanitized() + + def bootstrap_session( + self, + *, + mg_at: str | None = None, + access_token: str | None = None, + refresh_token: str | None = None, + device_id: str | None = None, + device_tag: str | None = None, + app_version: str | None = None, + platform_version: str | None = None, + user_agent: str | None = None, + audience: str | None = None, + base_url: str | None = None, + ) -> dict[str, Any]: + updates: dict[str, Any] = {} + if mg_at: + updates.update(parse_mg_at_cookie(mg_at)) + if access_token is not None: + updates["access_token"] = access_token + if refresh_token is not None: + updates["refresh_token"] = refresh_token + if device_id is not None: + updates["device_id"] = device_id + if device_tag is not None: + updates["device_tag"] = device_tag + if app_version is not None: + updates["app_version"] = app_version + if platform_version is not None: + updates["platform_version"] = platform_version + if user_agent is not None: + updates["user_agent"] = user_agent + if audience is not None: + updates["audience"] = audience + if base_url is not None: + updates["base_url"] = base_url.rstrip("/") + + for key, value in updates.items(): + setattr(self.session_config, key, value) + + if not self.session_config.refresh_token: + raise MagnitError("Нужен refresh token: передайте его напрямую или через mg_at.") + if not self.session_config.device_id or not self.session_config.device_tag: + raise MagnitError("Нужны device_id (mg_udi) и device_tag (mg_ksid).") + + self._save_session() + self._sync_cookies() + return self.session_status() + + def refresh_session(self) -> dict[str, Any]: + if not self.session_config.refresh_token: + raise MagnitError("В локальной сессии нет refresh token.") + previous_refresh_token = self.session_config.refresh_token + self._sync_cookies() + response = self.http.post( + f"{self.session_config.base_url}/magnit-id/v1/auth/token/refresh", + headers=self._base_headers(include_auth=False), + json={ + "aud": self.session_config.audience, + "refreshToken": self.session_config.refresh_token, + }, + timeout=self.timeout_seconds, + ) + data = self._decode_json(response) + if response.status_code != HTTPStatus.OK: + raise MagnitError(f"Не удалось обновить сессию: {response.status_code} {data}") + + access_token = data.get("accessToken") + refresh_token = data.get("refreshToken") or self.session_config.refresh_token + if not access_token: + raise MagnitError(f"Backend не вернул accessToken: {data}") + + self.session_config.access_token = access_token + self.session_config.refresh_token = refresh_token + self._save_session() + self._sync_cookies() + return { + "refreshed": True, + "access_token_expires_at": decode_jwt_exp(access_token), + "refresh_token_changed": refresh_token != previous_refresh_token, + } + + def ensure_access_token(self) -> None: + if not self.session_config.access_token or token_is_expiring(self.session_config.access_token): + self.refresh_session() + + def search_goods( + self, + *, + store_code: str, + query: str, + limit: int = 36, + offset: int = 0, + store_type: str = DEFAULT_STORE_TYPE, + catalog_type: str = DEFAULT_CATALOG_TYPE, + include_adult_goods: bool = True, + ) -> dict[str, Any]: + if not store_code: + raise MagnitError("Пустой store_code.") + if not query: + raise MagnitError("Пустой query.") + + self.ensure_access_token() + payload = { + "term": query, + "pagination": {"offset": offset, "limit": limit}, + "sort": {"order": "desc", "type": "popularity"}, + "includeAdultGoods": include_adult_goods, + "storeCode": str(store_code), + "storeType": store_type, + "catalogType": str(catalog_type), + } + self._sync_cookies(store_code=str(store_code)) + response = self.http.post( + f"{self.session_config.base_url}/webgate/v2/goods/search", + headers=self._base_headers(include_auth=True), + json=payload, + timeout=self.timeout_seconds, + ) + + if response.status_code == HTTPStatus.UNAUTHORIZED: + self.refresh_session() + self._sync_cookies(store_code=str(store_code)) + response = self.http.post( + f"{self.session_config.base_url}/webgate/v2/goods/search", + headers=self._base_headers(include_auth=True), + json=payload, + timeout=self.timeout_seconds, + ) + + data = self._decode_json(response) + if response.status_code != HTTPStatus.OK: + raise MagnitError(f"Поиск завершился ошибкой: {response.status_code} {data}") + + normalized_items = [ + normalize_item(item, store_type=store_type) + for item in data.get("items", []) + ] + return { + "query": query, + "store_code": str(store_code), + "store_type": store_type, + "catalog_type": str(catalog_type), + "count": len(normalized_items), + "total_count": ((data.get("pagination") or {}).get("totalCount")), + "corrected_term": data.get("correctedTerm"), + "items": normalized_items, + "best_match": build_best_match(query, normalized_items), + "raw": data, + } + + def _save_session(self) -> None: + self.session_config.save(self.session_path) + + def _cookie_mg_at(self) -> str: + payload = { + "access": self.session_config.access_token, + "refresh": self.session_config.refresh_token, + } + return quote(json.dumps(payload, ensure_ascii=False, separators=(",", ":"))) + + def _sync_cookies(self, store_code: str | None = None) -> None: + cfg = self.session_config + if cfg.device_id: + self.http.cookies.set("mg_udi", cfg.device_id, domain=".magnit.ru", path="/") + if cfg.device_tag: + self.http.cookies.set("mg_ksid", cfg.device_tag, domain=".magnit.ru", path="/") + if cfg.refresh_token: + self.http.cookies.set("mg_at", self._cookie_mg_at(), domain=".magnit.ru", path="/") + self.http.cookies.set("mg_icp", "valid", domain=".magnit.ru", path="/") + if store_code: + self.http.cookies.set("shopCode", store_code, domain="magnit.ru", path="/") + self.http.cookies.set("nmg_sp", "Y", domain="magnit.ru", path="/") + self.http.cookies.set("nmg_dt", "DELIVERY_TYPE_PICKUP", domain="magnit.ru", path="/") + self.http.cookies.set("x_shop_type", "MM", domain="magnit.ru", path="/") + self.http.cookies.set("mg_uac", "1", domain="magnit.ru", path="/") + + def _base_headers(self, *, include_auth: bool) -> dict[str, str]: + headers = { + "accept": "application/json", + "content-type": "application/json", + "origin": self.session_config.base_url, + "referer": f"{self.session_config.base_url}/", + "user-agent": self.session_config.user_agent, + "accept-language": "ru-RU,ru;q=0.9,en;q=0.8", + "x-app-version": self.session_config.app_version, + "x-client-name": "magnit", + "x-device-id": self.session_config.device_id, + "x-device-platform": "Web", + "x-device-tag": self.session_config.device_tag, + "x-new-magnit": "true", + "x-platform-version": self.session_config.platform_version, + } + if include_auth and self.session_config.access_token: + headers["authorization"] = f"Bearer {self.session_config.access_token}" + return headers + + @staticmethod + def _decode_json(response: requests.Response) -> Any: + try: + return response.json() + except ValueError: + return response.text + + +class ApiHandler(BaseHTTPRequestHandler): + server_version = "MagnitAPI/1.0" + + @property + def client(self) -> MagnitClient: + return self.server.client # type: ignore[attr-defined] + + def do_GET(self) -> None: # noqa: N802 + self._dispatch() + + def do_POST(self) -> None: # noqa: N802 + self._dispatch() + + def log_message(self, format: str, *args: Any) -> None: + return + + def _dispatch(self) -> None: + try: + parsed = urlparse(self.path) + path = parsed.path.rstrip("/") or "/" + query_params = { + key: values[-1] + for key, values in parse_qs(parsed.query, keep_blank_values=True).items() + } + body = self._read_json_body() if self.command == "POST" else {} + + if path == "/health": + self._send_json(HTTPStatus.OK, {"ok": True, "timestamp": int(time.time())}) + return + if path == "/session": + self._send_json(HTTPStatus.OK, self.client.session_status()) + return + if path == "/session/refresh": + self._send_json(HTTPStatus.OK, self.client.refresh_session()) + return + if path == "/session/bootstrap": + payload = body or query_params + result = self.client.bootstrap_session( + mg_at=payload.get("mg_at"), + access_token=payload.get("access_token"), + refresh_token=payload.get("refresh_token"), + device_id=payload.get("device_id"), + device_tag=payload.get("device_tag"), + app_version=payload.get("app_version"), + platform_version=payload.get("platform_version"), + user_agent=payload.get("user_agent"), + audience=payload.get("audience"), + base_url=payload.get("base_url"), + ) + self._send_json(HTTPStatus.OK, result) + return + if path == "/search": + payload = {**query_params, **body} + result = self.client.search_goods( + store_code=str(payload.get("store_code", "")), + query=str(payload.get("query", "")), + limit=int(payload.get("limit", 36)), + offset=int(payload.get("offset", 0)), + store_type=str(payload.get("store_type", DEFAULT_STORE_TYPE)), + catalog_type=str(payload.get("catalog_type", DEFAULT_CATALOG_TYPE)), + include_adult_goods=parse_bool(payload.get("include_adult_goods"), True), + ) + self._send_json(HTTPStatus.OK, result) + return + if path == "/price": + payload = {**query_params, **body} + result = self.client.search_goods( + store_code=str(payload.get("store_code", "")), + query=str(payload.get("query", "")), + limit=int(payload.get("limit", 36)), + offset=int(payload.get("offset", 0)), + store_type=str(payload.get("store_type", DEFAULT_STORE_TYPE)), + catalog_type=str(payload.get("catalog_type", DEFAULT_CATALOG_TYPE)), + include_adult_goods=parse_bool(payload.get("include_adult_goods"), True), + ) + self._send_json( + HTTPStatus.OK, + { + "query": result["query"], + "store_code": result["store_code"], + "count": result["count"], + "best_match": result["best_match"], + "items": result["items"], + }, + ) + return + + self._send_json(HTTPStatus.NOT_FOUND, {"error": f"Неизвестный путь: {path}"}) + except MagnitError as exc: + self._send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)}) + except Exception as exc: # pragma: no cover - fallback for runtime debugging + self._send_json( + HTTPStatus.INTERNAL_SERVER_ERROR, + {"error": f"{type(exc).__name__}: {exc}"}, + ) + + def _read_json_body(self) -> dict[str, Any]: + content_length = int(self.headers.get("Content-Length", "0")) + if content_length <= 0: + return {} + raw_body = self.rfile.read(content_length) + if not raw_body: + return {} + try: + return json.loads(raw_body.decode("utf-8")) + except json.JSONDecodeError as exc: + raise MagnitError("Body должен быть валидным JSON.") from exc + + def _send_json(self, status: int, payload: Any) -> None: + body = json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "application/json; charset=utf-8") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + +def run_server(host: str, port: int, session_file: Path) -> None: + server = ThreadingHTTPServer((host, port), ApiHandler) + server.client = MagnitClient(session_file) # type: ignore[attr-defined] + print(f"Magnit API listening on http://{host}:{port}") + server.serve_forever() + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="Локальный API и CLI для цен Magnit.") + parser.add_argument( + "--session-file", + default=str(SESSION_FILE), + help="Путь до JSON-файла с access/refresh и device headers.", + ) + subparsers = parser.add_subparsers(dest="command", required=True) + + serve = subparsers.add_parser("serve", help="Поднять локальный HTTP API.") + serve.add_argument("--host", default="127.0.0.1") + serve.add_argument("--port", type=int, default=8080) + + for command_name in ("search", "price"): + command = subparsers.add_parser(command_name, help=f"{command_name} по store_code и query.") + command.add_argument("--store-code", required=True) + command.add_argument("--query", required=True) + command.add_argument("--limit", type=int, default=36) + command.add_argument("--offset", type=int, default=0) + command.add_argument("--store-type", default=DEFAULT_STORE_TYPE) + command.add_argument("--catalog-type", default=DEFAULT_CATALOG_TYPE) + command.add_argument("--include-adult-goods", action="store_true", default=True) + + subparsers.add_parser("refresh-session", help="Принудительно обновить access token.") + subparsers.add_parser("show-session", help="Показать текущее состояние локальной сессии.") + + bootstrap = subparsers.add_parser("bootstrap", help="Сохранить cookies/токены в локальную сессию.") + bootstrap.add_argument("--mg-at", default="") + bootstrap.add_argument("--access-token", default="") + bootstrap.add_argument("--refresh-token", default="") + bootstrap.add_argument("--device-id", required=True) + bootstrap.add_argument("--device-tag", required=True) + bootstrap.add_argument("--app-version", default=DEFAULT_APP_VERSION) + bootstrap.add_argument("--platform-version", default=DEFAULT_PLATFORM_VERSION) + bootstrap.add_argument("--user-agent", default=DEFAULT_USER_AGENT) + bootstrap.add_argument("--audience", default=DEFAULT_AUDIENCE) + bootstrap.add_argument("--base-url", default=DEFAULT_BASE_URL) + + return parser + + +def main() -> None: + parser = build_parser() + args = parser.parse_args() + session_file = Path(args.session_file) + + if args.command == "serve": + run_server(args.host, args.port, session_file) + return + + if args.command == "bootstrap": + config = SessionConfig.load_or_default(session_file) + config.save(session_file) + client = MagnitClient(session_file) + result = client.bootstrap_session( + mg_at=args.mg_at or None, + access_token=args.access_token or None, + refresh_token=args.refresh_token or None, + device_id=args.device_id, + device_tag=args.device_tag, + app_version=args.app_version, + platform_version=args.platform_version, + user_agent=args.user_agent, + audience=args.audience, + base_url=args.base_url, + ) + print(json.dumps(result, ensure_ascii=False, indent=2)) + return + + client = MagnitClient(session_file) + + if args.command == "refresh-session": + print(json.dumps(client.refresh_session(), ensure_ascii=False, indent=2)) + return + + if args.command == "show-session": + print(json.dumps(client.session_status(), ensure_ascii=False, indent=2)) + return + + if args.command == "search": + result = client.search_goods( + store_code=args.store_code, + query=args.query, + limit=args.limit, + offset=args.offset, + store_type=args.store_type, + catalog_type=args.catalog_type, + include_adult_goods=args.include_adult_goods, + ) + print(json.dumps(result, ensure_ascii=False, indent=2)) + return + + if args.command == "price": + result = client.search_goods( + store_code=args.store_code, + query=args.query, + limit=args.limit, + offset=args.offset, + store_type=args.store_type, + catalog_type=args.catalog_type, + include_adult_goods=args.include_adult_goods, + ) + print( + json.dumps( + { + "query": result["query"], + "store_code": result["store_code"], + "count": result["count"], + "best_match": result["best_match"], + "items": result["items"], + }, + ensure_ascii=False, + indent=2, + ) + ) + return + + parser.error(f"Неизвестная команда: {args.command}") + + +if __name__ == "__main__": + main() diff --git a/nginx/magnit-api.location.conf.example b/nginx/magnit-api.location.conf.example new file mode 100644 index 0000000..68fd453 --- /dev/null +++ b/nginx/magnit-api.location.conf.example @@ -0,0 +1,28 @@ +# Вариант 1: nginx работает на хосте, а docker-compose публикует API только в loopback. +location /magnit-api/ { + proxy_pass http://127.0.0.1:18080/; + proxy_http_version 1.1; + + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + proxy_set_header X-Forwarded-Host $host; +} + +# После этого запросы выглядят так: +# /magnit-api/price?store_code=618224&query=Red%20Bull + +# Вариант 2: nginx тоже в Docker и находится в одной сети с сервисом magnit-api. +# Тогда вместо loopback можно проксировать по имени сервиса: +# +# location /magnit-api/ { +# proxy_pass http://magnit-api:8080/; +# proxy_http_version 1.1; +# +# proxy_set_header Host $host; +# proxy_set_header X-Real-IP $remote_addr; +# proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; +# proxy_set_header X-Forwarded-Proto $scheme; +# proxy_set_header X-Forwarded-Host $host; +# } diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..d80d9fc --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +requests==2.32.3