667 lines
26 KiB
Python
667 lines
26 KiB
Python
#!/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."
|
||
)
|
||
if path.is_dir():
|
||
raise MagnitError(
|
||
f"Ожидался JSON-файл сессии, но путь является директорией: {path}. "
|
||
"Проверьте docker volume mount."
|
||
)
|
||
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()
|
||
if path.is_dir():
|
||
raise MagnitError(
|
||
f"Ожидался JSON-файл сессии, но путь является директорией: {path}. "
|
||
"Проверьте docker volume mount."
|
||
)
|
||
return cls.load(path)
|
||
|
||
def save(self, path: Path) -> None:
|
||
if path.exists() and path.is_dir():
|
||
raise MagnitError(
|
||
f"Невозможно сохранить сессию: путь является директорией, а не файлом: {path}."
|
||
)
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
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_or_default(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()
|