mirror of
https://github.com/wasrusgen/zov-tech.git
synced 2026-06-03 15:44:47 +00:00
1a57374020
CITILINK: - Now reads data-src / data-original / srcset / src in priority order - srcset → picks largest size variant (last in comma-list) - Filters only _next/static/images (placeholder) and 'placeholder' in URL - Accepts cs.citilink.ru / c.citilink.ru / images.citilink.ru product photos ЯНДЕКС.МАРКЕТ: - Collects all img attrs (data-src, data-original, srcset, data-srcset, src) - Prefers avatars.mds.yandex.net (real product CDN), skips yastatic (icons/logos) - Auto-appends /300x300 suffix to avatars.mds URLs without size ENRICH_ONE (aggregator): - Image picked by source priority: yamarket > wb > ozon > citilink > dns - Yamarket photos are cleanest (avatars.mds.yandex.net) - WB has product photos via basket-XX.wbbasket.ru
279 lines
11 KiB
Python
279 lines
11 KiB
Python
"""Парсер Я.Маркета — через Playwright (рендер JS).
|
||
|
||
Я.Маркет — SPA на React, товары подгружаются через XHR после первой загрузки.
|
||
Простой HTTP-запрос не вернёт каталог. Поэтому используем headless Chromium.
|
||
|
||
Ждём пока в DOM появятся карточки `[data-zone-name="snippet-card"]` или
|
||
`a[href*="/product--"]`, потом извлекаем данные.
|
||
"""
|
||
from __future__ import annotations
|
||
import logging
|
||
import re
|
||
from typing import Any
|
||
from urllib.parse import quote_plus
|
||
|
||
from bs4 import BeautifulSoup
|
||
|
||
from . import playwright_engine
|
||
|
||
log = logging.getLogger("zov.parser.yamarket")
|
||
|
||
_BASE_URL = "https://market.yandex.ru"
|
||
_PRICE_RE = re.compile(r"([\d\s]+)\s*₽")
|
||
|
||
|
||
def search_yamarket(query: str, limit: int = 3, timeout: float = 30.0,
|
||
max_retries: int = 1) -> list[dict[str, Any]]:
|
||
"""Поиск товара в Я.Маркете через headless Chromium + residential proxy.
|
||
|
||
Я.Маркет (2025-2026) использует URL pattern `/card/{slug}/{productId}`.
|
||
Старые URL `/product--` больше не применяются.
|
||
"""
|
||
url = f"{_BASE_URL}/search?text={quote_plus(query)}"
|
||
|
||
html = None
|
||
for attempt in range(max_retries + 1):
|
||
html = playwright_engine.fetch_page(
|
||
url,
|
||
# Ждём появления товарных ссылок /card/...
|
||
wait_selector="a[href*='/card/']",
|
||
wait_ms=5000,
|
||
timeout_ms=int(timeout * 1000),
|
||
)
|
||
if html:
|
||
break
|
||
|
||
if not html:
|
||
log.warning("YaMarket: no HTML for query=%r", query)
|
||
return []
|
||
|
||
if "showcaptcha" in html.lower() or "qrator" in html.lower()[:5000]:
|
||
log.warning("YaMarket: Qrator/captcha for query=%r", query)
|
||
return []
|
||
if "Похоже, вы используете" in html[:30000] or "используете VPN" in html[:30000]:
|
||
log.warning("YaMarket: VPN warning page for query=%r", query)
|
||
return []
|
||
|
||
return _parse_html(html, limit=limit)
|
||
|
||
|
||
def _parse_html(html: str, limit: int) -> list[dict[str, Any]]:
|
||
"""Парсим товары через URL pattern /card/{slug}/{productId} (Я.Маркет 2026)."""
|
||
soup = BeautifulSoup(html, "html.parser")
|
||
results: list[dict[str, Any]] = []
|
||
seen_ids = set()
|
||
|
||
for link in soup.select("a[href*='/card/']"):
|
||
if len(results) >= limit:
|
||
break
|
||
href = link.get("href") or ""
|
||
# Пропускаем sponsored — их URL'ы с CPC-токенами истекают через несколько часов
|
||
if "sponsored=1" in href or "cpc=" in href or "advUuid" in href:
|
||
continue
|
||
m_id = re.search(r"/card/[^/]+/(\d+)", href)
|
||
if not m_id:
|
||
continue
|
||
product_id = m_id.group(1)
|
||
if product_id in seen_ids:
|
||
continue
|
||
seen_ids.add(product_id)
|
||
|
||
full_url = href if href.startswith("http") else f"{_BASE_URL}{href}"
|
||
clean_url = full_url.split("?")[0]
|
||
|
||
# Карточка-родитель — article, div с data-zone-name или просто ближайший div
|
||
card = (
|
||
link.find_parent("article")
|
||
or link.find_parent("div", attrs={"data-zone-name": True})
|
||
or link.find_parent("div")
|
||
)
|
||
if not card:
|
||
continue
|
||
item = _extract_card(card, link, clean_url)
|
||
if item:
|
||
results.append(item)
|
||
|
||
return results
|
||
|
||
|
||
def _extract_card(card, link_el, url: str) -> dict[str, Any] | None:
|
||
"""Достаём title, price, image, rating, reviews, stores из карточки."""
|
||
# Удаляем JSON-стейт виджетов React/Apiary — они захламляют get_text()
|
||
import copy
|
||
card_clean = copy.copy(card)
|
||
for tag in card_clean.find_all(["script", "noscript", "noframes", "template"]):
|
||
tag.decompose()
|
||
full_text = card_clean.get_text(" ", strip=True)
|
||
|
||
# Title — приоритет: из URL slug (всегда чистый, есть всегда)
|
||
title = ""
|
||
m_slug = re.search(r"/card/([^/]+)/\d+", url)
|
||
if m_slug:
|
||
slug = m_slug.group(1)
|
||
# bosch-kgn39ul30u-dvukhkamernyy-kholodilnik → Bosch Kgn39ul30u двухкамерный холодильник
|
||
title = _slug_to_title(slug)
|
||
# Запасной — из <a title="..."> или h3/h2
|
||
if not title or len(title) < 5:
|
||
title = (link_el.get("title") or "").strip()
|
||
if not title or len(title) < 5:
|
||
for sel in ["h3", "h2", "[data-auto='snippet-title']", "span[itemprop='name']"]:
|
||
el = card_clean.select_one(sel)
|
||
if el:
|
||
t = (el.get("title") or el.get_text(strip=True)).strip()
|
||
if t and len(t) > 5 and "{" not in t[:5]:
|
||
title = t
|
||
break
|
||
if not title or len(title) < 5:
|
||
return None
|
||
|
||
# Цена — минимальная в карточке
|
||
price_min = None
|
||
for m in _PRICE_RE.finditer(full_text):
|
||
raw = m.group(1).replace(" ", "").replace(" ", "").replace(" ", "")
|
||
try:
|
||
v = int(raw)
|
||
if 100 < v < 10_000_000:
|
||
if price_min is None or v < price_min:
|
||
price_min = v
|
||
except ValueError:
|
||
pass
|
||
|
||
# Картинка — приоритет: avatars.mds.yandex.net (реальные товарные фото)
|
||
img_url = None
|
||
candidates = []
|
||
for img_el in card.find_all("img"):
|
||
for attr in ("data-src", "data-original", "srcset", "data-srcset", "src"):
|
||
v = img_el.get(attr) or ""
|
||
if not v:
|
||
continue
|
||
if attr in ("srcset", "data-srcset"):
|
||
# Берём самый большой размер
|
||
parts = v.split(",")
|
||
if parts:
|
||
candidates.append(parts[-1].strip().split(" ")[0])
|
||
else:
|
||
candidates.append(v)
|
||
# Чистим и приоритизируем
|
||
for src in candidates:
|
||
if not src or "data:image" in src:
|
||
continue
|
||
if src.startswith("//"):
|
||
src = "https:" + src
|
||
if "yastatic" in src: # это иконки/логотипы
|
||
continue
|
||
# avatars.mds.yandex.net — высший приоритет
|
||
if "avatars.mds.yandex" in src:
|
||
img_url = src
|
||
break
|
||
# Любой полноценный http(s) URL — fallback
|
||
if src.startswith("http") and not img_url:
|
||
img_url = src
|
||
|
||
# Если URL не содержит размерного суффикса — добавим /300x300
|
||
if img_url and "avatars.mds.yandex" in img_url and not re.search(r"/\d+x\d+/?$|/orig/?$", img_url):
|
||
img_url = img_url.rstrip("/") + "/300x300"
|
||
|
||
# Рейтинг
|
||
rating = None
|
||
m = re.search(r"(\d[.,]\d)(?:\s*★|\s*\(?\d+\s*оцен)", full_text)
|
||
if m:
|
||
try:
|
||
r = float(m.group(1).replace(",", "."))
|
||
if 0 < r <= 5.0:
|
||
rating = r
|
||
except ValueError:
|
||
pass
|
||
|
||
# Отзывы
|
||
reviews = None
|
||
m = re.search(r"(\d[\d\s ]*)\s*(?:отзыв|оценок|review)", full_text, re.I)
|
||
if m:
|
||
try:
|
||
reviews = int(m.group(1).replace(" ", "").replace(" ", "").replace(" ", ""))
|
||
except ValueError:
|
||
pass
|
||
|
||
# Кол-во магазинов / предложений
|
||
stores = None
|
||
m = re.search(r"(?:от|в)\s+(\d+)\s+(?:магазин|предложен)", full_text)
|
||
if m:
|
||
try:
|
||
stores = int(m.group(1))
|
||
except ValueError:
|
||
pass
|
||
|
||
return {
|
||
"title": title[:250],
|
||
"url": url,
|
||
"image_url": img_url,
|
||
"price_min_rub": price_min,
|
||
"price_max_rub": None,
|
||
"rating": rating,
|
||
"reviews_count": reviews,
|
||
"stores_count": stores,
|
||
"specs": {},
|
||
"source": "yamarket",
|
||
}
|
||
|
||
|
||
_TRANSLIT_MAP = {
|
||
"kholodilnik": "холодильник", "khol": "хол",
|
||
"dvukhkamernyy": "двухкамерный", "odnokamernyy": "однокамерный",
|
||
"morozilnaya": "морозильная", "kamera": "камера", "kolonna": "колонна",
|
||
"varochnaya": "варочная", "panel": "панель", "dukhovoy": "духовой", "shkaf": "шкаф",
|
||
"posudomoechnaya": "посудомоечная", "mashina": "машина",
|
||
"vytyazhka": "вытяжка", "kupolnaya": "купольная", "naklonnaya": "наклонная", "ostrovnaya": "островная",
|
||
"mikrovolnovaya": "микроволновая", "pech": "печь",
|
||
"kofemashina": "кофемашина", "kofevarka": "кофеварка",
|
||
"stiralnaya": "стиральная", "sushilnaya": "сушильная",
|
||
"induktsionnaya": "индукционная", "gazovaya": "газовая", "elektricheskaya": "электрическая",
|
||
"vstraivaemyy": "встраиваемый", "vstraivaemaya": "встраиваемая",
|
||
"no-frost": "No Frost", "nofrost": "NoFrost", "side-by-side": "Side-by-Side",
|
||
"belyy": "белый", "chernyy": "чёрный", "seryy": "серый", "metallik": "металлик",
|
||
"nerzhaveyushchaya": "нержавеющая", "stal": "сталь",
|
||
"serebristyy": "серебристый", "korichnevyy": "коричневый",
|
||
"diapazon": "диапазон", "obem": "объём", "shirina": "ширина",
|
||
}
|
||
|
||
|
||
def _slug_to_title(slug: str) -> str:
|
||
"""Конвертирует URL-slug в читаемое название.
|
||
'bosch-kgn39ul30u-dvukhkamernyy-kholodilnik-no-frost-seryy-metallik' →
|
||
'Bosch KGN39UL30U Двухкамерный Холодильник NoFrost Серый Металлик'
|
||
"""
|
||
# Сначала заменяем известные транслитерированные слова
|
||
s = slug.lower()
|
||
# Сложные многословные паттерны
|
||
s = s.replace("no-frost", "NoFrost")
|
||
s = s.replace("side-by-side", "Side-by-Side")
|
||
parts = s.replace("-", " ").split()
|
||
result = []
|
||
for p in parts:
|
||
replaced = _TRANSLIT_MAP.get(p, p)
|
||
# Капитализируем первой буквой, если это слово (не модель типа kgn39ul30u)
|
||
if any(c.isdigit() for c in replaced) and re.search(r"[a-z]", replaced):
|
||
# Модель — uppercase: kgn39ul30u → KGN39UL30U
|
||
replaced = replaced.upper()
|
||
elif replaced.isascii() and replaced.isalpha():
|
||
replaced = replaced.capitalize() # Bosch, NoFrost оставляем
|
||
result.append(replaced)
|
||
return " ".join(result).strip()
|
||
|
||
|
||
def _try_int(v: Any) -> int | None:
|
||
if v is None:
|
||
return None
|
||
try:
|
||
return int(float(str(v).replace(" ", "").replace(",", ".")))
|
||
except (ValueError, TypeError):
|
||
return None
|
||
|
||
|
||
def _try_float(v: Any) -> float | None:
|
||
if v is None:
|
||
return None
|
||
try:
|
||
return float(str(v).replace(" ", "").replace(",", "."))
|
||
except (ValueError, TypeError):
|
||
return None
|