#!/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()