lekss361/gendesign
1
0
Fork 0
Code Issues 143 Pull requests Projects Releases Packages Wiki Activity Actions

feat(msk-collector): Циан как вторая площадка сбора по Москве и МО

Browse source 
Сборщик получает ключ --platform {avito,cian}: платформо-зависимые куски
(URL коридора, счётчик, парс, потолок пагинации, целевая таблица) вынесены
в PlatformAdapter, общая часть — бисекция по цене, guard на блок, накопитель
и заливка — одна на обе площадки.

Скоуп Циан проверен живыми запросами 10.09, а не взят из документации:

- region=1 и region=4593 двумя параметрами НЕ объединяются, сервер берёт
  последний. Прежний базовый URL собрал бы одну Московскую область и молча
  потерял Москву целиком (92 817 объявлений). Правильный скоуп — region=-1.
- object_type[0]=1 обязателен: без него счётчик считает новостройки, которые
  дальше не сохраняются, и расходится с числом карточек вдвое.
- Итоговый скоуп: 62 548 объявлений вторички по Москве и МО.
- Потолок пагинации 54 страницы подтверждён: страница 55 пуста.

Фильтра новостроек в адаптере нет, и это сознательное отличие от кита. Кит
считает новостройкой всё, у чего есть offer.newbuilding.id (serp.py:401-402),
потому что для ЕКБ выдача бралась без object_type. На московской странице из
28 карточек 16 имеют этот блок, и у всех шестнадцати isFromBuilder=false,
isFromLeadFactory=false — это вторичка в ЖК, а не лоты застройщика. С китовым
фильтром прогон терял бы 57 % корпуса безвозвратно. msk_raw — сырьё, payload
несёт listing_segment целиком, сегмент отделяется на импорте в listings.

Детект блока Циан. Капча приходит как HTTP 200 с обычной на вид страницей:
поймана живьём, 40 КБ, title «Captcha - база объявлений ЦИАН», без
window._cianConfig. По статусу её не отличить, поэтому маркер ищется в первых
4 КБ (в нормальной выдаче на 2,6 МБ там ни одного вхождения). Вторая сеть в
parse_page: счётчик None при нуле сырых карточек — тоже стоп. Без этого блок
засчитывался бы как пустая страница, коридор уходил в done, а --resume его
уже не перебрал бы.

Гвард empty_page считает карточки ДО фильтра: иначе штатный ноль после
фильтрации был бы неотличим от блока. У Авито атрибута нет, дефолт — длина
итогового списка, поведение прежнее.

Заливка. Целевая таблица берётся из адаптера, staging создаётся с
INCLUDING IDENTITY (без него identity-колонка не переносится, а NOT NULL
переносится всегда, и \copy падал бы на каждом батче). Наличие таблицы
проверяется на старте одним SELECT to_regclass — иначе прогон умирал бы
на первой заливке, после часов планирования.

Миграция 299 заводит cian_cards, domclick_cards, yandex_cards и три вью
по образцу avito_cards/avito_latest. id — bigserial, а не IDENTITY, ровно
по причине выше.

Заодно: _wt-mskcol/ выведен из индекса и закрыт в .gitignore. Копия-worktree
попала в репозиторий 09.09, и три фикса ушли в дубликат мимо канонического
collect.py — дефект нашёлся только при сверке размера страницы.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VQ8jqr4SFirX5tFLwdSrXh
This commit is contained in:
bot-backend 2026-09-10 13:23:46 +03:00
parent 455766dcc4
commit 8fcef103c2
5 changed files with 666 additions and 907 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
.gitignore vendored
View file

@ -109,3 +109,10 @@ ops/metrics/alertmanager/alertmanager.yml
# и Prometheus не видел ни одного приёмника (#3155). Производный файл убирает
# сам зазор — правится только там же, где принимается решение о профиле.
ops/metrics/prometheus/alertmanager_targets.gen.yml
# Временные рабочие копии-worktree вида _wt-<тема>/ живут рядом с репозиторием
# и в него попадать не должны. 09.09 копия _wt-mskcol попала в индекс одним
# файлом сборщика, и три следующих фикса ушли в дубликат мимо канонического
# tradein-mvp/scripts/local-avito-msk/collect.py — дефект нашёлся только при
# сверке page size.
_wt-*/

785
_wt-mskcol/tradein-mvp/scripts/local-avito-msk/collect.py
View file

@ -1,785 +0,0 @@
#!/usr/bin/env python3
"""Локальный ручной сборщик SERP Авито по Москве и МО (эпик #2989, трек 1).
Запускается ВРУЧНУЮ с машины владельца. Прод-скрейпер, его расписания и
прокси-пул не задействованы вообще: браузер — уже открытый Chrome владельца
(подключение по CDP), парсер — импорт из scraper-kit, заливка — поток в psql
через ssh. Скрипт ничего не устанавливает и своего профиля не поднимает.
Дефолтный режим — --measure 100 (замер): полный проход только по явному --full.
"""
from __future__ import annotations
import argparse
import asyncio
import csv
import io
import json
import math
import os
import random
import re
import subprocess
import sys
import time
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from types import SimpleNamespace
from typing import Any, Iterable
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
# --- импорт парсера из scraper-kit без установки backend -------------------
_KIT_SRC = Path(__file__).resolve().parents[2] / "packages" / "scraper-kit" / "src"
if str(_KIT_SRC) not in sys.path:
sys.path.insert(0, str(_KIT_SRC))
from scraper_kit.providers.avito.serp import ( # noqa: E402
AvitoScraper,
_is_firewall_page,
)
# Вкладка, открытая у владельца: вторичка, Москва + МО.
DEFAULT_BASE_URL = (
"https://www.avito.ru/moskva_i_mo/kvartiry/prodam/vtorichka-ASgBAgICAkSSA8YQ5geMUg"
"?f=ASgBAgICA0SSA8YQ5geMUuDC3c0CgOkP"
)
# Замерено живым проходом (не из документации): выдача Москва+МО отдаёт 50 карточек
# на страницу. Пока здесь стояло 60, planned_pages считал count/60 и не запрашивал
# последние ~17% каждого коридора — 34 753 по счётчику против 28 352 собранных.
# Молчаливое усечение читается как «покрыто всё», поэтому число проверяется живьём.
PAGE_SIZE = 50 # карточек на странице выдачи
MAX_PAGES = 30 # потолок пагинации Авито → 30*50 = 1500 на один запрос
HARD_CAP = PAGE_SIZE * MAX_PAGES
PRICE_FLOOR = 500_000 # нижняя граница осмысленного коридора, ₽
PRICE_PROBE_START = 8_000_000 # старт удвоения при поиске верхней границы
PRICE_CEIL = 2_000_000_000
MIN_WIDTH_RATIO = 1.05 # уже этого коридор не делим (геометрическая ширина)
MAX_DEPTH = 12
_BATCH_ID_RE = re.compile(r"^[A-Za-z0-9._-]+$")
_POW_MARKERS = ("startpow", "доступ ограничен: проверка безопасности")
# Заполняется при входе в Loader.__aenter__ (playwright импортируется лениво,
# чтобы --help работал без установленного пакета).
PlaywrightTimeoutError: type[BaseException] = TimeoutError
# Через столько загрузок вкладка сборщика пересоздаётся (см. _recycle_if_needed).
PAGE_RECYCLE_EVERY = 75
class Blocked(Exception):
"""Первый признак блока. Ретраев нет — только немедленный стоп."""
def __init__(self, reason: str, detail: str = "") -> None:
super().__init__(f"{reason}: {detail}" if detail else reason)
self.reason = reason
self.detail = detail
class BudgetExhausted(Exception):
"""Потолок --measure выбран: штатный выход, не ошибка."""
# --- план коридоров --------------------------------------------------------
@dataclass
class Corridor:
lo: int | None
hi: int | None
count: int | None = None
truncated: bool = False
pages_done: int = 0
status: str = "pending" # pending | done
missed: int = 0 # заведомо недобрано (count - HARD_CAP), если truncated
def label(self) -> str:
lo = "-" if self.lo is None else f"{self.lo:_}"
hi = "-" if self.hi is None else f"{self.hi:_}"
return f"[{lo} .. {hi}]"
def to_json(self) -> dict[str, Any]:
return {
"lo": self.lo, "hi": self.hi, "count": self.count,
"truncated": self.truncated, "pages_done": self.pages_done,
"status": self.status, "missed": self.missed,
}
@staticmethod
def from_json(d: dict[str, Any]) -> "Corridor":
return Corridor(
lo=d.get("lo"), hi=d.get("hi"), count=d.get("count"),
truncated=bool(d.get("truncated")),
pages_done=int(d.get("pages_done") or 0),
status=d.get("status") or "pending", missed=int(d.get("missed") or 0),
)
def planned_pages(self) -> int:
if not self.count:
return 1
return max(1, min(MAX_PAGES, math.ceil(self.count / PAGE_SIZE)))
@dataclass
class Plan:
base_url: str
target: int
batch_id: str
corridors: list[Corridor] = field(default_factory=list)
created_at: str = ""
def save(self, path: Path) -> None:
path.write_text(
json.dumps(
{
"version": 1, "base_url": self.base_url, "target": self.target,
"batch_id": self.batch_id, "created_at": self.created_at,
"corridors": [c.to_json() for c in self.corridors],
},
ensure_ascii=False, indent=1,
),
encoding="utf-8",
)
@staticmethod
def load(path: Path) -> "Plan":
d = json.loads(path.read_text(encoding="utf-8"))
return Plan(
base_url=d["base_url"], target=int(d["target"]), batch_id=d["batch_id"],
created_at=d.get("created_at", ""),
corridors=[Corridor.from_json(c) for c in d.get("corridors", [])],
)
def build_url(base_url: str, page: int, lo: int | None, hi: int | None) -> str:
"""URL коридора: pmin/pmax + пагинация. Гео-параметры не добавляем — #3043."""
parts = urlsplit(base_url)
q = [(k, v) for k, v in parse_qsl(parts.query, keep_blank_values=True)
if k not in {"p", "pmin", "pmax"}]
if lo is not None:
q.append(("pmin", str(int(lo))))
if hi is not None:
q.append(("pmax", str(int(hi))))
if page > 1:
q.append(("p", str(page)))
return urlunsplit(
(parts.scheme, parts.netloc, parts.path, urlencode(q), parts.fragment)
)
def geometric_mid(lo: int | None, hi: int) -> int:
"""Геометрическая середина коридора.
Цены логнормальны: арифметическая середина 1 млн..100 млн (≈50 млн)
отрезает вырожденно-пустую верхнюю половину. sqrt(lo*hi) делит выборку
заметно ровнее.
"""
low = max(int(lo or PRICE_FLOOR), 1)
mid = int(math.sqrt(low * float(hi)))
return max(low + 1, min(hi - 1, mid))
def width_ratio(lo: int | None, hi: int | None) -> float:
if hi is None:
return float("inf")
return float(hi) / max(float(lo or PRICE_FLOOR), 1.0)
# --- загрузка страницы -----------------------------------------------------
def _guard(html: str, status: int | None) -> None:
"""Порядок проверок фиксирован заданием; первое срабатывание = стоп."""
if status in (403, 439):
raise Blocked("platform", f"HTTP {status}")
if status == 429:
raise Blocked("ratelimit", "HTTP 429")
if _is_firewall_page(html):
raise Blocked("firewall", "firewall-страница на HTTP 200")
head = html[:4096].lower()
if any(m in head for m in _POW_MARKERS):
raise Blocked("challenge", "PoW / проверка безопасности")
class Loader:
"""Одна СВОЯ вкладка в уже открытом Chrome владельца (CDP).
Ни браузер, ни контекст, ни чужие вкладки не закрываются и не трогаются:
это рабочий Chrome с залогиненным техаккаунтом.
"""
def __init__(self, delay: float, page_budget: int | None) -> None:
self._delay = delay
self._budget = page_budget
self.loads = 0
self._page: Any = None
self._pw: Any = None
self._browser: Any = None
self._ctx: Any = None
self._last_load = 0.0
self._loads_on_page = 0
async def __aenter__(self) -> "Loader":
from playwright.async_api import async_playwright
from playwright.async_api import TimeoutError as _PwTimeout
global PlaywrightTimeoutError
PlaywrightTimeoutError = _PwTimeout
endpoint = os.environ.get("AVITO_CDP", "http://localhost:9222")
self._pw = await async_playwright().start()
try:
self._browser = await self._pw.chromium.connect_over_cdp(endpoint)
except Exception as exc: # noqa: BLE001 — подсказка важнее типа
await self._pw.stop()
raise SystemExit(
f"Не удалось подключиться по CDP к {endpoint}: {exc}\n"
"Запусти Chrome с залогиненным техаккаунтом Авито и ключом "
"--remote-debugging-port=9222, либо укажи адрес в AVITO_CDP."
) from exc
if not self._browser.contexts:
await self._pw.stop()
raise SystemExit(
"В подключённом Chrome нет ни одного контекста. Открой обычное окно "
"Chrome, запущенное с --remote-debugging-port=9222."
)
self._ctx = self._browser.contexts[0]
self._page = await self._ctx.new_page()
return self
async def __aexit__(self, *exc: object) -> None:
if self._page is not None:
try:
await self._page.close() # ТОЛЬКО своя вкладка
except Exception: # noqa: BLE001
pass
if self._pw is not None:
try:
await self._pw.stop()
except Exception: # noqa: BLE001
pass
def budget_left(self) -> bool:
return self._budget is None or self.loads < self._budget
async def _pause(self) -> None:
if self._last_load == 0.0:
return
jitter = self._delay * random.uniform(-0.2, 0.2)
wait = max(0.0, self._delay + jitter - (time.monotonic() - self._last_load))
if wait > 0:
print(f" пауза {wait:.1f} с", flush=True)
await asyncio.sleep(wait)
async def fetch(self, url: str) -> tuple[str, int | None]:
if not self.budget_left():
raise BudgetExhausted()
await self._recycle_if_needed()
await self._pause()
resp = await self._goto(url)
self._loads_on_page += 1
self.loads += 1
self._last_load = time.monotonic()
status = resp.status if resp is not None else None
try:
await self._page.wait_for_selector('[data-marker="item"]', timeout=7_000)
except Exception: # noqa: BLE001 — пустая/блочная страница разбирается ниже
pass
html = await self._read_content()
_guard(html, status)
return html, status
async def _recycle_if_needed(self) -> None:
"""Каждые PAGE_RECYCLE_EVERY загрузок пересоздаём свою вкладку.
Рендерер Chrome накапливает память по всем навигациям вкладки, а проход
по Москве — под тысячу страниц в одной. Наблюдалось живьём: вкладка
падала с «Опаньки… Код ошибки: Out of Memory» при 34 ГБ свободных в
системе, то есть упирался именно рендерер, а не машина. Свежая вкладка
стоит одну навигацию и обнуляет счёт.
Закрывается ТОЛЬКО своя вкладка; контекст и чужие вкладки владельца не
трогаются — это его рабочий Chrome.
"""
if self._loads_on_page < PAGE_RECYCLE_EVERY:
return
print(f" вкладка пересоздаётся после {self._loads_on_page} загрузок "
"(память рендерера)", flush=True)
old = self._page
self._page = await self._ctx.new_page()
self._loads_on_page = 0
try:
await old.close()
except Exception: # noqa: BLE001 — старая вкладка могла уже умереть
pass
async def _goto(self, url: str, attempts: int = 3):
"""goto с ограниченным ретраем на таймаут навигации.
Авито изредка держит соединение до упора и goto падает по timeout. Это
НЕ признак отказа: в наблюдавшемся случае вкладка показывала нормальную
выдачу, а маркеров фаервола/PoW не было. Но и молча ретраить бесконечно
нельзя — тихий отказ выглядит ровно так же. Поэтому: перед каждым
повтором пробуем прочитать то, что в документе, и прогнать через _guard,
чтобы настоящий блок остановил прогон с правильной причиной; исчерпали
попытки — жёсткий стоп с причиной nav_timeout.
"""
for i in range(attempts):
try:
return await self._page.goto(url, wait_until="domcontentloaded",
timeout=90_000)
except PlaywrightTimeoutError:
try:
partial = await self._page.content()
except Exception: # noqa: BLE001 — документа может не быть вовсе
partial = ""
if partial:
_guard(partial, None) # настоящий блок остановит прогон здесь
if i == attempts - 1:
raise Blocked("nav_timeout") from None
print(f" таймаут навигации, попытка {i + 2}/{attempts}", flush=True)
await asyncio.sleep(10.0)
async def _read_content(self, attempts: int = 4) -> str:
"""page.content() с узким ретраем на гонку клиентской перенавигации.
Авито дорисовывает выдачу после domcontentloaded, и content() иногда
попадает ровно в момент смены документа: "Unable to retrieve content
because the page is navigating and changing the content". Это НЕ отказ
площадки — гвардов не касается, поэтому ретраим только эту ошибку и
только её, а любую другую поднимаем как есть.
"""
last: Exception | None = None
for i in range(attempts):
try:
return await self._page.content()
except Exception as exc: # noqa: BLE001 — сузили проверкой текста ниже
if "page is navigating" not in str(exc):
raise
last = exc
print(f" content() поймал перенавигацию, попытка {i + 2}/{attempts}",
flush=True)
await asyncio.sleep(1.5)
raise RuntimeError(f"page.content() не отдал документ за {attempts} попыток") from last
# --- заливка в msk_raw -----------------------------------------------------
def _sql_str(value: str) -> str:
return "'" + value.replace("'", "''") + "'"
def _csv_rows(rows: Iterable[dict[str, Any]]) -> str:
buf = io.StringIO()
writer = csv.writer(buf, lineterminator="\n")
for r in rows:
writer.writerow([
r["source_id"], r["observed_at"], r["batch_id"], r["kind"],
r["url"], r["price"], r["payload"],
])
return buf.getvalue()
def build_sql(batch_id: str, query: str, rows: list[dict[str, Any]],
started_at: str, kind: str = "serp") -> str:
"""Один поток на `psql -f -`: batch (FK!) → TEMP staging → \\copy → INSERT.
Одиночный `psql -c` через ssh ломается на квотинге скобок и кавычек, поэтому
только поток. rows_new = разница count(*) по batch_id до и после вставки.
"""
bid = _sql_str(batch_id)
return (
"BEGIN;\n"
"INSERT INTO msk_raw.batches (batch_id, kind, query, started_at)\n"
f"VALUES ({bid}, {_sql_str(kind)}, {_sql_str(query)}, "
f"CAST({_sql_str(started_at)} AS timestamptz))\n"
"ON CONFLICT (batch_id) DO NOTHING;\n"
"CREATE TEMP TABLE _stg (LIKE msk_raw.avito_cards INCLUDING DEFAULTS) "
"ON COMMIT DROP;\n"
"CREATE TEMP TABLE _before ON COMMIT DROP AS\n"
f" SELECT count(*) AS n FROM msk_raw.avito_cards WHERE batch_id = {bid};\n"
"\\copy _stg (source_id,observed_at,batch_id,kind,url,price,payload) "
"FROM STDIN WITH (FORMAT csv)\n"
+ _csv_rows(rows)
+ "\\.\n"
"INSERT INTO msk_raw.avito_cards "
"(source_id,observed_at,batch_id,kind,url,price,payload)\n"
"SELECT source_id,observed_at,batch_id,kind,url,price,payload FROM _stg\n"
"ON CONFLICT (source_id,batch_id,kind) DO NOTHING;\n"
"UPDATE msk_raw.batches b SET\n"
" rows_sent = coalesce(b.rows_sent,0) + (SELECT count(*) FROM _stg),\n"
" rows_new = coalesce(b.rows_new,0) +\n"
f" ((SELECT count(*) FROM msk_raw.avito_cards WHERE batch_id = {bid})\n"
" - (SELECT n FROM _before))\n"
f"WHERE b.batch_id = {bid};\n"
"COMMIT;\n"
)
def build_finalize_sql(batch_id: str, query: str, notes: str) -> str:
bid = _sql_str(batch_id)
return (
"INSERT INTO msk_raw.batches (batch_id, kind, query, started_at)\n"
f"VALUES ({bid}, 'serp', {_sql_str(query)}, now())\n"
"ON CONFLICT (batch_id) DO NOTHING;\n"
f"UPDATE msk_raw.batches SET finished_at = now(), notes = {_sql_str(notes)}\n"
f"WHERE batch_id = {bid};\n"
)
def run_psql(sql: str, ssh_host: str, container: str, db_user: str, db_name: str,
attempts: int = 4) -> None:
"""Заливка батча через ssh с ретраем на обрыв транспорта.
Прогон длится часами, и ssh рвётся: живьём поймано «Connection reset by peer»
(ssh возвращает 255) прямо посреди заливки — весь прогон умирал, а несброшенный
батч терялся. Ретраить безопасно: SQL идемпотентен (batch через ON CONFLICT DO
NOTHING, карточки через ON CONFLICT (source_id,batch_id,kind) DO NOTHING).
Ретраится ТОЛЬКО транспорт (ssh 255). Ошибка самого psql (ON_ERROR_STOP, любой
другой код) — это дефект данных или SQL, её повтор не лечит: поднимаем сразу.
"""
cmd = [
"ssh", ssh_host,
f"docker exec -i {container} psql -U {db_user} -d {db_name} "
"-v ON_ERROR_STOP=1 -f -",
]
for i in range(attempts):
proc = subprocess.run(cmd, input=sql.encode("utf-8"), capture_output=True)
out = (proc.stdout + proc.stderr).decode("utf-8", "replace").strip()
if proc.returncode == 0:
if out:
print(f" psql: {out}", flush=True)
return
if proc.returncode != 255 or i == attempts - 1:
raise RuntimeError(f"psql через ssh вернул {proc.returncode}:\n{out}")
tail = out.splitlines()[-1] if out else "без вывода"
print(f" ssh оборвался ({tail}), повтор заливки {i + 2}/{attempts}", flush=True)
time.sleep(15.0 * (i + 1))
# --- накопитель карточек ---------------------------------------------------
@dataclass
class Sink:
"""Батчами на прод (ssh+psql) или в локальный CSV при --dry-run."""
batch_id: str
started_at: str
query: str
batch_size: int
dry_run: bool
csv_path: Path
ssh_host: str
container: str
db_user: str
db_name: str
buffer: list[dict[str, Any]] = field(default_factory=list)
sent: int = 0
skipped_non_numeric: int = 0
def add(self, lot: Any) -> None:
raw_id = str(getattr(lot, "source_id", "") or "")
try:
source_id = int(raw_id) # в БД bigint, у ScrapedLot — строка
except (TypeError, ValueError):
self.skipped_non_numeric += 1
return
payload = lot.model_dump(mode="json")
self.buffer.append({
"source_id": source_id,
"observed_at": datetime.now(timezone.utc).isoformat(),
"batch_id": self.batch_id,
"kind": "serp",
"url": payload.get("source_url"),
"price": payload.get("price_rub"),
"payload": json.dumps(payload, ensure_ascii=False),
})
def maybe_flush(self) -> None:
if len(self.buffer) >= self.batch_size:
self.flush()
def flush(self) -> None:
if not self.buffer:
return
rows, self.buffer = self.buffer, []
if self.dry_run:
fresh = not self.csv_path.exists()
with self.csv_path.open("a", encoding="utf-8", newline="") as fh:
if fresh:
fh.write("source_id,observed_at,batch_id,kind,url,price,payload\n")
fh.write(_csv_rows(rows))
print(f" [dry-run] {len(rows)} строк → {self.csv_path}", flush=True)
else:
run_psql(
build_sql(self.batch_id, self.query, rows, self.started_at),
self.ssh_host, self.container, self.db_user, self.db_name,
)
print(f" залито {len(rows)} строк в msk_raw.avito_cards", flush=True)
self.sent += len(rows)
def finalize(self, notes: str) -> None:
self.flush()
if self.dry_run:
print(f" [dry-run] finalize: {notes}", flush=True)
return
run_psql(
build_finalize_sql(self.batch_id, self.query, notes),
self.ssh_host, self.container, self.db_user, self.db_name,
)
# --- сбор ------------------------------------------------------------------
def parse_page(scraper: AvitoScraper, html: str, url: str) -> tuple[int | None, list[Any]]:
count = scraper._extract_total_count(html)
lots = scraper._parse_html(html, "https://www.avito.ru")
if not lots and count:
raise Blocked("empty_page", f"0 карточек при счётчике {count}: {url}")
return count, lots
async def probe(loader: Loader, scraper: AvitoScraper, base_url: str,
lo: int | None, hi: int | None) -> tuple[int | None, list[Any]]:
url = build_url(base_url, 1, lo, hi)
html, _ = await loader.fetch(url)
return parse_page(scraper, html, url)
async def build_plan(loader: Loader, scraper: AvitoScraper, base_url: str, target: int,
cache: dict[tuple[int | None, int | None], list[Any]]
) -> list[Corridor]:
"""Адаптивная бисекция по цене; страница 1 каждого коридора кэшируется."""
corridors: list[Corridor] = []
def emit(lo: int | None, hi: int | None, count: int | None,
truncated: bool, lots: list[Any]) -> None:
missed = max(0, (count or 0) - HARD_CAP) if truncated else 0
c = Corridor(lo=lo, hi=hi, count=count, truncated=truncated, missed=missed)
corridors.append(c)
cache[(lo, hi)] = lots
flag = " TRUNCATED" if truncated else ""
print(f" коридор {c.label()} count={count} "
f"страниц={c.planned_pages()}{flag}", flush=True)
if truncated:
print(f" ВНИМАНИЕ: коридор {c.label()} не влезает в потолок "
f"{HARD_CAP}; заведомо не добрано ~{missed} объявлений", flush=True)
async def find_upper(lo: int | None) -> int:
"""Верхнюю границу открытого коридора ищем удвоением от разумного старта."""
cand = max(int(lo or PRICE_FLOOR) * 2, PRICE_PROBE_START)
while cand < PRICE_CEIL:
cnt, _ = await probe(loader, scraper, base_url, cand, None)
print(f" проба хвоста pmin={cand:_} count={cnt}", flush=True)
if cnt is not None and cnt <= target:
return cand
cand *= 2
return cand
async def split(lo: int | None, hi: int | None, depth: int,
count: int | None, lots: list[Any]) -> None:
if count is None:
url = build_url(base_url, 1, lo, hi)
raise Blocked("empty_page", f"счётчик не прочитался: {url}")
if count <= target:
emit(lo, hi, count, False, lots)
return
if depth >= MAX_DEPTH or width_ratio(lo, hi) <= MIN_WIDTH_RATIO:
# Предохранитель: не молчим — помечаем truncated и считаем недобор.
emit(lo, hi, count, count > HARD_CAP, lots)
return
upper = hi if hi is not None else await find_upper(lo)
if hi is None:
tail_cnt, tail_lots = await probe(loader, scraper, base_url, upper, None)
emit(upper, None, tail_cnt,
bool(tail_cnt and tail_cnt > HARD_CAP), tail_lots)
mid = geometric_mid(lo, upper)
for sub_lo, sub_hi in ((lo, mid), (mid, upper)):
sub_cnt, sub_lots = await probe(loader, scraper, base_url, sub_lo, sub_hi)
print(f" проба {sub_lo or '-'}..{sub_hi} count={sub_cnt}", flush=True)
await split(sub_lo, sub_hi, depth + 1, sub_cnt, sub_lots)
root_cnt, root_lots = await probe(loader, scraper, base_url, None, None)
print(f"Всего по базовому запросу: {root_cnt}", flush=True)
await split(None, None, 0, root_cnt, root_lots)
return corridors
async def collect(args: argparse.Namespace) -> int:
# avito_serp_ekb_only=False обязателен: с True парсер выбрасывает всё, где в
# URL нет /ekaterinburg/ — то есть все подмосковные слаги (serp.py:2154).
scraper = AvitoScraper(
SimpleNamespace(avito_serp_ekb_only=False), # type: ignore[arg-type]
target_city_slug="moskva",
)
out_dir = Path(args.out_dir).resolve()
out_dir.mkdir(parents=True, exist_ok=True)
plan_path = out_dir / f"plan-{args.batch_id}.json"
csv_path = out_dir / f"cards-{args.batch_id}.csv"
started_at = datetime.now(timezone.utc).isoformat()
plan: Plan | None = None
if args.resume:
if not plan_path.exists():
print(f"--resume: плана нет — {plan_path}", file=sys.stderr)
return 1
plan = Plan.load(plan_path)
done = sum(1 for c in plan.corridors if c.status == "done")
print(f"Resume по {plan_path}: коридоров {len(plan.corridors)}, "
f"готово {done}", flush=True)
# Resume: URL берём из сохранённого плана, а не из CLI — коридоры посчитаны
# именно под него. Расхождение = молчаливая заливка чужой выдачи под тем же
# batch_id, поэтому это ошибка, а не тихий приоритет одного из двух.
if plan is not None and plan.base_url != args.base_url:
raise SystemExit(
"--resume: план построен для другого URL."
f" В плане {plan.base_url}, в аргументах {args.base_url}."
" Убери --base-url (возьмётся из плана) либо начни новый batch_id."
)
base_url = plan.base_url if plan is not None else args.base_url
page_budget = None if args.full else args.measure
mode = "FULL" if args.full else f"MEASURE<={page_budget}"
print(f"Режим: {mode}; batch_id={args.batch_id}; delay={args.delay}s; "
f"target={args.target_count}; dry_run={args.dry_run}", flush=True)
sink = Sink(
batch_id=args.batch_id, started_at=started_at, query=base_url,
batch_size=args.batch_size, dry_run=args.dry_run, csv_path=csv_path,
ssh_host=args.ssh_host, container=args.container,
db_user=args.db_user, db_name=args.db_name,
)
cache: dict[tuple[int | None, int | None], list[Any]] = {}
total = 0
stop_reason = ""
rc = 0
loads = 0
async with Loader(args.delay, page_budget) as loader:
try:
if plan is None:
print("Строю план коридоров...", flush=True)
corridors = await build_plan(loader, scraper, base_url,
args.target_count, cache)
plan = Plan(base_url=base_url, target=args.target_count,
batch_id=args.batch_id, corridors=corridors,
created_at=started_at)
plan.save(plan_path)
print(f"План сохранён: {plan_path} ({len(corridors)} коридоров)",
flush=True)
for corridor in plan.corridors:
if corridor.status == "done":
continue
pages = corridor.planned_pages()
print(f"Коридор {corridor.label()} count={corridor.count} "
f"страниц={pages} (с {corridor.pages_done + 1})", flush=True)
for page in range(corridor.pages_done + 1, pages + 1):
key = (corridor.lo, corridor.hi)
if page == 1 and key in cache:
lots = cache.pop(key) # страница 1 уже скачана при планировании
else:
url = build_url(base_url, page, corridor.lo, corridor.hi)
html, _ = await loader.fetch(url)
_, lots = parse_page(scraper, html, url)
for lot in lots:
sink.add(lot)
total += len(lots)
corridor.pages_done = page
print(f" стр.{page}/{pages}: карточек {len(lots)}, "
f"итого {total}", flush=True)
sink.maybe_flush()
plan.save(plan_path)
if not lots:
print(" пустая страница — конец коридора", flush=True)
break
corridor.status = "done"
plan.save(plan_path)
except BudgetExhausted:
stop_reason = "потолок --measure исчерпан"
print(f"Стоп: {stop_reason}", flush=True)
except Blocked as exc:
stop_reason = f"BLOCKED/{exc.reason}: {exc.detail}"
print(f"СТОП: {stop_reason}", file=sys.stderr, flush=True)
rc = 2
finally:
loads = loader.loads
if plan is not None:
plan.save(plan_path)
truncated = [c for c in (plan.corridors if plan else []) if c.truncated]
missed = sum(c.missed for c in truncated)
notes = "; ".join(x for x in [
f"mode={mode}", f"loads={loads}", f"cards={total}",
f"skipped_non_numeric={sink.skipped_non_numeric}",
(f"truncated_corridors={len(truncated)} missed~{missed}" if truncated else ""),
stop_reason,
] if x)
try:
sink.finalize(notes)
except Exception as exc: # noqa: BLE001 — не прятать исходную причину стопа
print(f"finalize провалился: {exc}", file=sys.stderr)
rc = rc or 1
print(f"Готово. Загрузок: {loads}; карточек: {total}; отправлено: {sink.sent}; "
f"пропущено нечисловых source_id: {sink.skipped_non_numeric}; "
f"notes: {notes}", flush=True)
return rc
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
p = argparse.ArgumentParser(
prog="collect.py",
description="Ручной сбор SERP Авито (вторичка, Москва+МО) в прод-схему msk_raw.",
)
p.add_argument("--base-url", default=DEFAULT_BASE_URL,
help="базовый URL выдачи (дефолт — вкладка владельца)")
p.add_argument("--measure", type=int, default=100, metavar="N",
help="режим замера: не больше N загрузок страниц (дефолт 100)")
p.add_argument("--full", action="store_true",
help="полный проход без потолка страниц (включается только явно)")
p.add_argument("--dry-run", action="store_true",
help="ничего не слать на прод, писать CSV локально")
p.add_argument("--resume", action="store_true",
help="продолжить по сохранённому плану коридоров")
p.add_argument("--delay", type=float, default=8.0,
help="пауза между загрузками, с (±20%% джиттер, дефолт 8.0)")
p.add_argument("--batch-size", type=int, default=1000,
help="карточек в одной заливке (дефолт 1000)")
p.add_argument("--target-count", type=int, default=1500,
help="целевой размер коридора; больше — делим (дефолт 1500)")
p.add_argument("--batch-id", default=None,
help="batch_id в msk_raw.batches (дефолт msk-serp-<UTC>)")
p.add_argument("--out-dir", default=str(Path(__file__).resolve().parent / "runs"),
help="каталог плана/CSV")
p.add_argument("--ssh-host", default="selectel", help="ssh-хост прода")
p.add_argument("--container", default="tradein-postgres",
help="имя контейнера Postgres на проде")
p.add_argument("--db-user", default="tradein")
p.add_argument("--db-name", default="tradein")
args = p.parse_args(argv)
if args.batch_id is None:
args.batch_id = "msk-serp-" + datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
if not _BATCH_ID_RE.match(args.batch_id):
p.error("--batch-id: допустимы только символы [A-Za-z0-9._-]")
if args.measure < 1:
p.error("--measure должен быть >= 1")
return args
def main(argv: list[str] | None = None) -> int:
return asyncio.run(collect(parse_args(argv)))
if __name__ == "__main__":
raise SystemExit(main())

122
tradein-mvp/backend/data/sql/299_msk_raw_cian_domclick_yandex_cards.sql Normal file
View file

@ -0,0 +1,122 @@
-- 299_msk_raw_cian_domclick_yandex_cards.sql
--
-- Контекст: msk_raw.avito_cards уже хранит сырые SERP-карточки Авито (Москва+МО,
-- вторичка), заливаемые ручным локальным сборщиком scripts/local-avito-msk/collect.py.
-- Эта миграция заводит зеркальные raw-таблицы под ЦИАН, Домклик и Яндекс.Недвижимость
-- по той же структуре и с тем же набором индексов (префикс площадки вместо avito_),
-- под будущие аналогичные локальные сборщики для этих площадок.
--
-- batch_id у всех трёх таблиц ссылается на существующую msk_raw.batches(batch_id)
-- (как и у avito_cards) — сборщик обязан сначала вставить строку в batches,
-- потом заливать карточки в рамках этого batch_id.
--
-- Тип source_id решён отдельно на каждую площадку по факту из парсеров
-- (packages/scraper-kit/src/scraper_kit/providers/{cian,domclick,yandex}/serp.py):
-- * cian -> bigint: source_id = str(offer.get("cianId") or offer.get("id")),
-- cianId у ЦИАН всегда числовой, используется как есть в URL
-- (/sale/flat/<cianId>/).
-- * domclick -> bigint: source_id = str(item.get("id")), id у Домклика — числовой
-- offer id (как в URL карточки), аналогично Авито/ЦИАН.
-- * yandex -> text: source_id = offer_id = str(entity.get("offerId") or ""),
-- в парсере есть отдельный _to_bigint() для price-полей, но offerId
-- сознательно НЕ приводится через него и остаётся строкой — числовой
-- формат offerId у Яндекс.Недвижимости не гарантирован.
--
-- id — именно bigserial, а НЕ 'GENERATED ALWAYS AS IDENTITY'. Сборщик заливает
-- карточки через CREATE TEMP TABLE _stg (LIKE msk_raw.<table> INCLUDING DEFAULTS INCLUDING IDENTITY)
-- и \copy без колонки id: LIKE ... INCLUDING DEFAULTS переносит в _stg обычный
-- DEFAULT (nextval у bigserial), но НЕ переносит identity-свойство, а NOT NULL
-- переносится всегда — с identity каждый батч падал бы на not-null violation по id.
-- Прод-таблица msk_raw.avito_cards заведена так же:
-- id bigint NOT NULL DEFAULT nextval('msk_raw.avito_cards_id_seq'::regclass).
--
-- Идемпотентно: повторный запуск безопасен (IF NOT EXISTS везде).
BEGIN;
CREATE TABLE IF NOT EXISTS msk_raw.cian_cards (
id bigserial PRIMARY KEY,
source_id bigint NOT NULL,
observed_at timestamptz NOT NULL,
batch_id text NOT NULL REFERENCES msk_raw.batches(batch_id),
kind text NOT NULL DEFAULT 'serp',
url text,
price numeric,
payload jsonb NOT NULL
);
CREATE INDEX IF NOT EXISTS cian_cards_observed_at_idx
ON msk_raw.cian_cards (observed_at);
CREATE INDEX IF NOT EXISTS cian_cards_payload_gin_idx
ON msk_raw.cian_cards USING gin (payload jsonb_path_ops);
CREATE UNIQUE INDEX IF NOT EXISTS cian_cards_source_id_batch_id_kind_uidx
ON msk_raw.cian_cards (source_id, batch_id, kind);
CREATE INDEX IF NOT EXISTS cian_cards_source_id_observed_at_idx
ON msk_raw.cian_cards (source_id, observed_at DESC);
CREATE TABLE IF NOT EXISTS msk_raw.domclick_cards (
id bigserial PRIMARY KEY,
source_id bigint NOT NULL,
observed_at timestamptz NOT NULL,
batch_id text NOT NULL REFERENCES msk_raw.batches(batch_id),
kind text NOT NULL DEFAULT 'serp',
url text,
price numeric,
payload jsonb NOT NULL
);
CREATE INDEX IF NOT EXISTS domclick_cards_observed_at_idx
ON msk_raw.domclick_cards (observed_at);
CREATE INDEX IF NOT EXISTS domclick_cards_payload_gin_idx
ON msk_raw.domclick_cards USING gin (payload jsonb_path_ops);
CREATE UNIQUE INDEX IF NOT EXISTS domclick_cards_source_id_batch_id_kind_uidx
ON msk_raw.domclick_cards (source_id, batch_id, kind);
CREATE INDEX IF NOT EXISTS domclick_cards_source_id_observed_at_idx
ON msk_raw.domclick_cards (source_id, observed_at DESC);
CREATE TABLE IF NOT EXISTS msk_raw.yandex_cards (
id bigserial PRIMARY KEY,
source_id text NOT NULL,
observed_at timestamptz NOT NULL,
batch_id text NOT NULL REFERENCES msk_raw.batches(batch_id),
kind text NOT NULL DEFAULT 'serp',
url text,
price numeric,
payload jsonb NOT NULL
);
CREATE INDEX IF NOT EXISTS yandex_cards_observed_at_idx
ON msk_raw.yandex_cards (observed_at);
CREATE INDEX IF NOT EXISTS yandex_cards_payload_gin_idx
ON msk_raw.yandex_cards USING gin (payload jsonb_path_ops);
CREATE UNIQUE INDEX IF NOT EXISTS yandex_cards_source_id_batch_id_kind_uidx
ON msk_raw.yandex_cards (source_id, batch_id, kind);
CREATE INDEX IF NOT EXISTS yandex_cards_source_id_observed_at_idx
ON msk_raw.yandex_cards (source_id, observed_at DESC);
-- Вью «последнее наблюдение по объявлению» — по образцу существующей
-- msk_raw.avito_latest (тот же DISTINCT ON и тот же порядок колонок).
CREATE OR REPLACE VIEW msk_raw.cian_latest AS
SELECT DISTINCT ON (source_id) id, source_id, observed_at, batch_id, kind, url, price, payload
FROM msk_raw.cian_cards ORDER BY source_id, observed_at DESC;
CREATE OR REPLACE VIEW msk_raw.domclick_latest AS
SELECT DISTINCT ON (source_id) id, source_id, observed_at, batch_id, kind, url, price, payload
FROM msk_raw.domclick_cards ORDER BY source_id, observed_at DESC;
CREATE OR REPLACE VIEW msk_raw.yandex_latest AS
SELECT DISTINCT ON (source_id) id, source_id, observed_at, batch_id, kind, url, price, payload
FROM msk_raw.yandex_cards ORDER BY source_id, observed_at DESC;
COMMIT;

149
tradein-mvp/scripts/local-avito-msk/README.md
View file

@ -1,7 +1,11 @@
# Локальный сбор SERP Авито по Москве и МО (эпик #2989, трек 1)
# Локальный сбор SERP Авито/Циан по Москве и МО (эпик #2989, трек 1)
`collect.py` — ручной скрипт **с машины владельца**. Собирает карточки выдачи Авито
(вторичка, Москва + МО) и заливает их в прод-схему `msk_raw`.
или Циан (вторичка, Москва + МО) и заливает их в прод-схему `msk_raw`. Площадка —
ключ `--platform {avito,cian}` (дефолт `avito`). Платформо-зависимые куски (URL
коридора, счётчик, парс карточек, потолок пагинации, целевая таблица) вынесены в
`PlatformAdapter` / `ADAPTERS` в `collect.py` — общая часть (бисекция по цене, guard
на блок, накопитель/заливка в psql) одна на обе платформы.
Прод-скрейпер, его расписания, прокси-пул и сайдкар **не задействованы вообще**.
Браузер — уже открытый Chrome владельца (подключение по CDP), парсер — импорт из
@ -41,62 +45,102 @@
Больше ничего ставить не нужно: только stdlib + playwright + импорт `scraper_kit`
(путь `packages/scraper-kit/src` скрипт добавляет в `sys.path` сам, от `__file__`).
4. **Миграция под Циан применена на проде** — до первого не-`--dry-run` прогона
`--platform cian` выполнить
`tradein-mvp/backend/data/sql/299_msk_raw_cian_domclick_yandex_cards.sql`
(таблицы `msk_raw.cian_cards` в проде пока нет). Путь указан именно так:
каталога `backend/data/sql` в корне репозитория не существует, файл лежит внутри
`tradein-mvp/`. Сборщик проверяет наличие целевой таблицы на старте и падает
сразу, не начиная обход выдачи.
## Запуск (PowerShell)
```powershell
cd D:\prjct\gendesign\tradein-mvp\scripts\local-avito-msk
# 0) сухой прогон: ничего не шлём на прод, карточки пишем в runs\cards-<batch>.csv
# 0) сухой прогон Авито: ничего не шлём на прод, карточки пишем в runs\cards-<batch>.csv
python .\collect.py --dry-run --measure 5
# 0б) то же для Циан
python .\collect.py --platform cian --dry-run --measure 5
# 1) обязательный первый прогон — замер (дефолт, 100 загрузок страниц)
python .\collect.py
python .\collect.py --platform cian
# 2) полный проход — только явно
python .\collect.py --full --batch-id msk-serp-20260908
python .\collect.py --full --batch-id msk-serp-avito-20260908
python .\collect.py --platform cian --full --batch-id msk-serp-cian-20260908
# 3) продолжить прерванный прогон по сохранённому плану коридоров
python .\collect.py --full --resume --batch-id msk-serp-20260908
python .\collect.py --full --resume --batch-id msk-serp-avito-20260908
```
Без аргументов скрипт работает в режиме `--measure 100` и полный проход **не начинает**.
Дефолт — `--platform avito`.
Ключи: `--delay` (пауза между загрузками, дефолт 8.0 с ±20 % джиттера — сознательно
совпадает с прод-расписаниями `request_delay_sec` 7–10 с), `--batch-size` (карточек в
одной заливке, дефолт 1000), `--target-count` (целевой размер коридора, дефолт 1500),
`--base-url`, `--batch-id`, `--out-dir`, `--ssh-host/--container/--db-user/--db-name`.
Ключи: `--platform {avito,cian}` (дефолт avito), `--delay` (пауза между загрузками,
дефолт 8.0 с ±20 % джиттера — сознательно совпадает с прод-расписаниями
`request_delay_sec` 7–10 с), `--batch-size` (карточек в одной заливке, дефолт 1000),
`--target-count` (целевой размер коридора, дефолт 1500), `--base-url` (дефолт зависит
от `--platform`), `--batch-id` (дефолт `msk-serp-<platform>-<UTC>` — платформа в имени,
чтобы avito- и cian-прогоны не затирали друг друга план/CSV), `--out-dir`,
`--ssh-host/--container/--db-user/--db-name`.
`AVITO_CDP` (адрес CDP, дефолт `http://localhost:9222`) общий для обеих платформ —
это адрес браузера владельца, а не площадки; имя переменной оставлено историческим.
## Как режется выдача
Потолок пагинации Авито — 30 страниц по 60 = **1800 объявлений на запрос**. Любой
запрос с `count > 1800` целиком не добирается, поэтому строится план ценовых коридоров:
Обе площадки режутся одной и той же адаптивной бисекцией по цене, но с **разным
потолком пагинации на запрос**:
* читаем счётчик «N объявлений» со страницы 1 (`page-title/count`);
| Платформа | Карточек/страница | Потолок страниц | Потолок объявлений/запрос |
|---|---|---|---|
| Авито | 50 (`AVITO_PAGE_SIZE`) | 30 (`AVITO_MAX_PAGES`) | **1500** |
| Циан | 28 (`_CIAN_OFFERS_PER_PAGE`) | 54 (см. `CianScraper._paginate_leaf_bucket`, "hard cap ~54") | **1512** |
Любой коридор, где `count` больше потолка платформы, целиком не добирается, поэтому
строится план ценовых коридоров:
* читаем счётчик найденных объявлений со страницы 1 (Авито — `page-title/count`;
Циан — `results.totalOffers` из Redux-state SSR-страницы, `_extract_total_offers`);
* `count > --target-count` → делим коридор пополам **по геометрической середине**
(`sqrt(lo*hi)`): цены логнормальны, арифметическая середина диапазона 1 млн … 100 млн
даёт вырожденно-пустую верхнюю половину;
* верхняя граница открытого коридора подбирается удвоением от 8 млн ₽;
* предохранители: глубина рекурсии ≤ 12 и минимальная ширина коридора (отношение
границ ≤ 1.05). Если коридор уже узкий, а `count` всё ещё > 1800 — он помечается
`truncated: true` в плане, а в лог и в `msk_raw.batches.notes` пишется, сколько
объявлений заведомо не добрано;
* гео-параметры `radius`/`geoCoords` не используются: сервером они не применяются (#3043).
границ ≤ 1.05). Если коридор уже узкий, а `count` всё ещё больше потолка платформы —
он помечается `truncated: true` в плане, а в лог и в `msk_raw.batches.notes` пишется,
сколько объявлений заведомо не добрано;
* гео-параметры типа `radius`/`geoCoords` не используются: для Авито сервером они не
применяются (#3043); для Циан фильтрация по гео вообще не поддерживается на уровне
SERP (сервер отдаёт весь регион, `region=` — единственный гео-скоуп).
План лежит в `runs/plan-<batch_id>.json` и обновляется после каждой страницы — отсюда
работает `--resume`.
План лежит в `runs/plan-<batch_id>.json` (включает `platform`, чтобы `--resume` не
перепутал план Авито с планом Циан) и обновляется после каждой страницы.
## Стоп на первом признаке блока
Проверки в фиксированном порядке, первое срабатывание = немедленный стоп
(никаких ретраев и никакого «продолжим со следующего коридора»):
| # | Признак | Причина в `notes` |
|---|---|---|
| 1 | HTTP 403 / 439 | `platform` |
| 2 | HTTP 429 | `ratelimit` |
| 3 | `_is_firewall_page(html)` — «доступ ограничен», «проблема с ip», `firewall-container` | `firewall` |
| 4 | `startpow` / «доступ ограничен: проверка безопасности» в первых 4 КБ | `challenge` |
| 5 | 0 карточек при ненулевом счётчике (DOM-drift или тихий блок) | `empty_page` |
| # | Признак | Причина в `notes` | Платформа |
|---|---|---|---|
| 1 | HTTP 403 / 439 | `platform` | обе |
| 2 | HTTP 429 | `ratelimit` | обе |
| 3 | `_is_firewall_page(html)` — «доступ ограничен», «проблема с ip», `firewall-container` | `firewall` | avito |
| 4 | `startpow` / «доступ ограничен: проверка безопасности» в первых 4 КБ | `challenge` | avito |
| 5 | 0 карточек при ненулевом счётчике (DOM-drift или тихий блок) | `empty_page` | обе |
| 6 | три подряд таймаута навигации (`Blocked("nav_timeout")` в `Loader._goto`) | `nav_timeout` | обе |
Для Циан текстовых маркеров блока (аналог п.3/4) пока нет — не проверялись живым
прогоном (задание такого прогона и не предполагало). Блок на Циан обычно всплывает как
проваленное извлечение Redux-state (`extract_state` вернул `None` — капча/смена
вёрстки), что уже покрыто общим `empty_page` (0 карточек при ненулевом счётчике) и
стопом `счётчик не прочитался` при `count is None` во время построения плана. Если на
практике всплывёт свой текстовый маркер блока Циан — его место в `_cian_detect_block`
в `collect.py`.
При стопе: недоотправленный батч дозаливается, у батча проставляются `finished_at` и
`notes`, скрипт выходит с кодом **2**.
@ -107,24 +151,33 @@ python .\collect.py --full --resume --batch-id msk-serp-20260908
увеличить `--delay`. `ratelimit` — темп слишком высокий: `--delay 15` и выше.
`empty_page` — сначала посмотреть сохранённую страницу в браузере: если выдача
рисуется, значит уехал DOM и чинить надо парсер в `scraper-kit`, а не скрипт.
`nav_timeout` — **не признак бана**: страница трижды подряд не догрузилась за таймаут.
Проверить сеть/VPN и саму вкладку в браузере владельца (жива ли, не висит ли диалог),
при необходимости поднять `--delay`. Циан-страницы тяжелее авитовских (SSR c Redux-state),
поэтому там `nav_timeout` вероятнее — это ожидаемо, паузы на часы он не требует.
После разбора — `--resume` с тем же `--batch-id`, уже собранное не потеряется.
## Куда пишем
Схема `msk_raw` на проде (создана заранее):
Схема `msk_raw` на проде:
* `msk_raw.batches(batch_id PK, kind, query, started_at, finished_at, rows_sent, rows_new, notes, uploaded_at)`
— общая для всех платформ (батч различается только по `batch_id`).
* `msk_raw.avito_cards(id, source_id, observed_at, batch_id → batches, kind, url, price, payload, UNIQUE(source_id,batch_id,kind))`
* `msk_raw.avito_latest` — вью `DISTINCT ON (source_id) … ORDER BY source_id, observed_at DESC`
* `msk_raw.cian_cards` — **та же схема**, что и `avito_cards` (тот же набор колонок и
тот же `UNIQUE(source_id,batch_id,kind)`); таблица создаётся отдельной SQL-миграцией
(database-expert) до первого прогона `--platform cian` — этот скрипт DDL не запускает.
* `msk_raw.avito_latest` / `msk_raw.cian_latest` — вью `DISTINCT ON (source_id) …
ORDER BY source_id, observed_at DESC` (аналогично `cian_latest`, тоже отдельная миграция).
Форма заливки: поток в
`ssh <host> "docker exec -i tradein-postgres psql -U tradein -d tradein -v ON_ERROR_STOP=1 -f -"`.
Внутри одной транзакции: `INSERT` батча (`ON CONFLICT DO NOTHING` — строка обязана
существовать до карточек из-за FK), `CREATE TEMP TABLE _stg (LIKE msk_raw.avito_cards
существовать до карточек из-за FK), `CREATE TEMP TABLE _stg (LIKE msk_raw.<table>
INCLUDING DEFAULTS)`, `\copy _stg (...) FROM STDIN WITH (FORMAT csv)`, затем
`INSERT … SELECT` в `avito_cards` с `ON CONFLICT (source_id,batch_id,kind) DO NOTHING`
и `UPDATE batches SET rows_sent/rows_new` (`rows_new` = разница `count(*)` по batch_id
до и после вставки).
`INSERT … SELECT` в `msk_raw.<table>` (`avito_cards` или `cian_cards`, по `--platform`)
с `ON CONFLICT (source_id,batch_id,kind) DO NOTHING` и `UPDATE batches SET
rows_sent/rows_new` (`rows_new` = разница `count(*)` по batch_id до и после вставки).
Одиночный `psql -c` через ssh ломается на квотинге скобок и кавычек — поэтому только поток.
@ -139,9 +192,37 @@ ssh selectel "docker exec -i tradein-postgres psql -U tradein -d tradein -c \"SE
ssh selectel "docker exec -i tradein-postgres psql -U tradein -d tradein -c \"SELECT batch_id, rows_sent, rows_new, started_at, finished_at, notes FROM msk_raw.batches ORDER BY uploaded_at DESC LIMIT 5;\""
```
## Про фильтр городов
## Про фильтр городов и площадок
Парсер конструируется как `AvitoScraper(SimpleNamespace(avito_serp_ekb_only=False),
Авито: парсер конструируется как `AvitoScraper(SimpleNamespace(avito_serp_ekb_only=False),
target_city_slug="moskva")`. `avito_serp_ekb_only=False` **обязателен**: с `True`
`_parse_html` выбрасывает всё, у чего в URL нет `/ekaterinburg/`, включая все
подмосковные слаги — из московской выдачи не осталось бы ничего.
подмосковные слаги — из московской выдачи не осталось бы ничего. URL берётся с
готовым слагом `vtorichka` — фильтр «только вторичка» задан самим URL.
Циан: `CianScraper(SimpleNamespace(glitchtip_dsn=None))`. Домен URL — `www.cian.ru`,
базовый скоуп — `region=-1` (Москва и МО **одним** параметром) плюс `object_type[0]=1`
(вторичка). Два параметра `region=` не объединяются — сервер берёт последний, поэтому
прежняя пара `region=1&region=4593` отдавала только область. Живой замер 10.09 по
`region=-1` + `object_type[0]=1`: **62 548 объявлений**. У `_build_url` из
`CianScraper` домен захардкожен под ЕКБ (`self.base_url = "https://ekb.cian.ru"`),
поэтому URL коридора строит свой билдер `_cian_build_url` в `collect.py`, а не метод
скрапера — используются только чистые парс-функции скрапера (`_extract_total_offers`,
`_parse_serp_html`), которым домен запроса не важен.
Вторичка у Циан задана в URL (`object_type[0]=1`), и **постфильтра новостроек в
адаптере нет** — это сознательное отличие от кита. `CianScraper` внутри
`_paginate_leaf_bucket`/`fetch_around` (эти методы скрипт не вызывает) выбрасывает всё,
у чего есть `offer.newbuilding.id`, считая SERP-параметр `object_type=1` ненадёжным
(`serp.py:401-402`). Для ЕКБ, где выдача бралась без этого параметра, так и было
правильно. Здесь — нет.
Замер живой страницы 10.09: из 28 карточек 16 имеют блок `newbuilding`, и у всех
шестнадцати `isFromBuilder=false`, `isFromLeadFactory=false`, `flatType=rooms`, а ЖК —
«Пригород Лесное», «РУСИЧ Новые Котельники» и подобные. Это обычная вторичка в новых
домах, а не лоты застройщика: `newbuilding.id` означает «дом в ЖК». С китовым фильтром
прогон терял бы **57 % корпуса**, причём безвозвратно.
`msk_raw` — сырьё: `payload` несёт `listing_segment` целиком, поэтому сегмент отделяется
позже, на импорте в `listings`. Адаптер только считает карточки со ссылкой на ЖК —
`scraper.last_nb_ref` попадает в `notes` как `nb_ref=N`.

510
tradein-mvp/scripts/local-avito-msk/collect.py
View file

@ -1,11 +1,15 @@
#!/usr/bin/env python3
"""Локальный ручной сборщик SERP Авито по Москве и МО (эпик #2989, трек 1).
"""Локальный ручной сборщик SERP Авито/Циан по Москве и МО (эпик #2989, трек 1).
Запускается ВРУЧНУЮ с машины владельца. Прод-скрейпер, его расписания и
прокси-пул не задействованы вообще: браузер — уже открытый Chrome владельца
(подключение по CDP), парсер — импорт из scraper-kit, заливка — поток в psql
через ssh. Скрипт ничего не устанавливает и своего профиля не поднимает.
Платформа выбирается ключом --platform {avito,cian} (дефолт avito) — см. класс
PlatformAdapter ниже. У каждой платформы свой потолок пагинации, свой билдер
URL коридора и своя целевая таблица в msk_raw.
Дефолтный режим — --measure 100 (замер): полный проход только по явному --full.
"""
@ -20,6 +24,7 @@ import math
import os
import random
import re
import shlex
import subprocess
import sys
import time
@ -27,10 +32,10 @@ from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from types import SimpleNamespace
from typing import Any, Iterable
from typing import Any, Callable, Iterable
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
# --- импорт парсера из scraper-kit без установки backend -------------------
# --- импорт парсеров из scraper-kit без установки backend -------------------
_KIT_SRC = Path(__file__).resolve().parents[2] / "packages" / "scraper-kit" / "src"
if str(_KIT_SRC) not in sys.path:
sys.path.insert(0, str(_KIT_SRC))
@ -39,16 +44,50 @@ from scraper_kit.providers.avito.serp import ( # noqa: E402
AvitoScraper,
_is_firewall_page,
)
from scraper_kit.providers.cian.serp import ( # noqa: E402
CianScraper,
_CIAN_OFFERS_PER_PAGE,
)
# Вкладка, открытая у владельца: вторичка, Москва + МО.
DEFAULT_BASE_URL = (
DEFAULT_AVITO_BASE_URL = (
"https://www.avito.ru/moskva_i_mo/kvartiry/prodam/vtorichka-ASgBAgICAkSSA8YQ5geMUg"
"?f=ASgBAgICA0SSA8YQ5geMUuDC3c0CgOkP"
)
PAGE_SIZE = 60 # карточек на странице выдачи
MAX_PAGES = 30 # потолок пагинации Авито → 30*60 = 1800 на один запрос
HARD_CAP = PAGE_SIZE * MAX_PAGES
# domain www.cian.ru, а не ekb.cian.ru: у CianScraper._build_url домен захардкожен
# под ЕКБ (self.base_url класс-константа), сюда не подходит — поэтому URL
# коридора строим своим билдером (_cian_build_url), а не scraper._build_url.
#
# Параметры проверены живым запросом 10.09 (curl, <title> и totalOffers из SSR):
# region=1 -> «Купить квартиру в Москве — 92 817»
# region=4593 -> «Купить квартиру в Московской области — 60 231»
# region=1&region=4593 -> «Московская область — 60 231»: побеждает ПОСЛЕДНИЙ,
# объединения по нескольким region= НЕТ
# region=-1 -> «Москва и Московская область — 153 049»
# region=-1 + object_type[0]=1 -> «Москва и МО — 62 548», puid3=sale_type_second
# region=1 + object_type[0]=1 -> «Москва — 36 743»
# Отсюда region=-1 — единственный способ получить единый скоуп Москва+МО.
# object_type[0]=1 (вторичка) обязателен: без него totalOffers считает и
# новостройки, которые парсер выбрасывает, — счётчик и карточки расходятся вдвое.
DEFAULT_CIAN_BASE_URL = (
"https://www.cian.ru/cat.php?deal_type=sale&engine_version=2&offer_type=flat"
"&region=-1&object_type%5B0%5D=1&sort=creation_date_desc"
)
# Замерено живым проходом (не из документации): выдача Москва+МО отдаёт 50 карточек
# на страницу. Пока здесь стояло 60, planned_pages считал count/60 и не запрашивал
# последние ~17% каждого коридора — 34 753 по счётчику против 28 352 собранных.
# Молчаливое усечение читается как «покрыто всё», поэтому число проверяется живьём.
AVITO_PAGE_SIZE = 50 # карточек на странице выдачи Авито
AVITO_MAX_PAGES = 30 # потолок пагинации Авито → 30*50 = 1500 на один запрос
# Циан отдаёт ~28 офферов/страницу (_CIAN_OFFERS_PER_PAGE в providers/cian/serp.py).
# Потолок страниц на бакет — 54, см. CianScraper._paginate_leaf_bucket
# (max_pages_per_bucket: int = 54, комментарий "Cian hard cap ~54; не превышать").
# 28*54 = 1512 — заметно меньше авитовских 1800, поэтому свой HARD_CAP обязателен.
CIAN_PAGE_SIZE = _CIAN_OFFERS_PER_PAGE
CIAN_MAX_PAGES = 54
PRICE_FLOOR = 500_000 # нижняя граница осмысленного коридора, ₽
PRICE_PROBE_START = 8_000_000 # старт удвоения при поиске верхней границы
@ -65,6 +104,10 @@ _POW_MARKERS = ("startpow", "доступ ограничен: проверка
PlaywrightTimeoutError: type[BaseException] = TimeoutError
# Через столько загрузок вкладка сборщика пересоздаётся (см. _recycle_if_needed).
PAGE_RECYCLE_EVERY = 75
class Blocked(Exception):
"""Первый признак блока. Ретраев нет — только немедленный стоп."""
@ -112,10 +155,10 @@ class Corridor:
status=d.get("status") or "pending", missed=int(d.get("missed") or 0),
)
def planned_pages(self) -> int:
def planned_pages(self, page_size: int, max_pages: int) -> int:
if not self.count:
return 1
return max(1, min(MAX_PAGES, math.ceil(self.count / PAGE_SIZE)))
return max(1, min(max_pages, math.ceil(self.count / page_size)))
@dataclass
@ -123,6 +166,7 @@ class Plan:
base_url: str
target: int
batch_id: str
platform: str = "avito" # старые планы (до --platform) читаются как avito
corridors: list[Corridor] = field(default_factory=list)
created_at: str = ""
@ -131,7 +175,8 @@ class Plan:
json.dumps(
{
"version": 1, "base_url": self.base_url, "target": self.target,
"batch_id": self.batch_id, "created_at": self.created_at,
"batch_id": self.batch_id, "platform": self.platform,
"created_at": self.created_at,
"corridors": [c.to_json() for c in self.corridors],
},
ensure_ascii=False, indent=1,
@ -144,13 +189,14 @@ class Plan:
d = json.loads(path.read_text(encoding="utf-8"))
return Plan(
base_url=d["base_url"], target=int(d["target"]), batch_id=d["batch_id"],
platform=d.get("platform", "avito"),
created_at=d.get("created_at", ""),
corridors=[Corridor.from_json(c) for c in d.get("corridors", [])],
)
def build_url(base_url: str, page: int, lo: int | None, hi: int | None) -> str:
"""URL коридора: pmin/pmax + пагинация. Гео-параметры не добавляем — #3043."""
def _avito_build_url(base_url: str, page: int, lo: int | None, hi: int | None) -> str:
"""URL коридора Авито: pmin/pmax + пагинация. Гео-параметры не добавляем — #3043."""
parts = urlsplit(base_url)
q = [(k, v) for k, v in parse_qsl(parts.query, keep_blank_values=True)
if k not in {"p", "pmin", "pmax"}]
@ -165,6 +211,26 @@ def build_url(base_url: str, page: int, lo: int | None, hi: int | None) -> str:
)
def _cian_build_url(base_url: str, page: int, lo: int | None, hi: int | None) -> str:
"""URL коридора Циан: minprice/maxprice + пагинация (аналог _avito_build_url).
Названия ключей — как в CianScraper._build_url (minprice/maxprice/p), но домен
и мульти-регион берём из base_url как есть, а не пересобираем через scraper.
"""
parts = urlsplit(base_url)
q = [(k, v) for k, v in parse_qsl(parts.query, keep_blank_values=True)
if k not in {"p", "minprice", "maxprice"}]
if lo is not None:
q.append(("minprice", str(int(lo))))
if hi is not None:
q.append(("maxprice", str(int(hi))))
if page > 1:
q.append(("p", str(page)))
return urlunsplit(
(parts.scheme, parts.netloc, parts.path, urlencode(q), parts.fragment)
)
def geometric_mid(lo: int | None, hi: int) -> int:
"""Геометрическая середина коридора.
@ -183,20 +249,169 @@ def width_ratio(lo: int | None, hi: int | None) -> float:
return float(hi) / max(float(lo or PRICE_FLOOR), 1.0)
# --- платформенные адаптеры --------------------------------------------------
def _avito_detect_block(html: str) -> tuple[str, str] | None:
"""Доп. текстовые маркеры блока Авито поверх общих HTTP-проверок в _guard.
Порядок (firewall перед PoW) идентичен исходному _guard — регресс-нейтрально.
"""
if _is_firewall_page(html):
return "firewall", "firewall-страница на HTTP 200"
head = html[:4096].lower()
if any(m in head for m in _POW_MARKERS):
return "challenge", "PoW / проверка безопасности"
return None
async def _avito_wait_ready(page: Any) -> None:
"""Карточка Авито — обычный DOM-узел, ждём её появления."""
try:
await page.wait_for_selector('[data-marker="item"]', timeout=7_000)
except Exception: # noqa: BLE001 — пустая/блочная страница разбирается ниже
pass
async def _cian_wait_ready(page: Any) -> None:
"""У Циан карточки читаются не из DOM, а из window._cianConfig['frontend-serp'],
поэтому ждать `[data-marker="item"]` (авитовский селектор) бессмысленно: он не
появится никогда, и каждая загрузка стоила бы лишних 7 с таймаута. Ждём сам
state — ровно то, что потом парсится.
"""
try:
await page.wait_for_function("!!window._cianConfig", timeout=10_000)
except Exception: # noqa: BLE001 — пустая/блочная страница разбирается ниже
pass
def _cian_detect_block(html: str) -> tuple[str, str] | None:
"""Капча Циан: HTTP 200 и обычная с виду страница, но без выдачи.
Поймана живьём 10.09: тот же IP, что нормально отдаёт SERP браузеру и curl,
на голом http-клиенте получил 40 КБ с <title>Captcha - база объявлений ЦИАН</title>
и без window._cianConfig вообще. То есть блок Циан НЕ приходит ни 403, ни 429 —
только телом, и по HTTP-статусу его не отличить.
Маркер берём в первых 4 КБ: в блок-странице слово встречается там 4 раза
(title + текст), в нормальной выдаче на 2.6 МБ — ни разу (все восемь вхождений
лежат глубоко в скриптах). Ложных срабатываний на живой выдаче нет.
Вторая сеть на случай другой формы блока — parse_page: count is None и сырых
карточек 0 тоже даёт Blocked("challenge").
"""
if "captcha" in html[:4096].lower():
return "challenge", "капча Циан (HTTP 200, страница без выдачи)"
return None
def _cian_parse_cards(scraper: CianScraper, html: str) -> list[Any]:
"""Парс карточек Циан. Ничего не отбрасываем, только считаем ссылки на ЖК.
Кит фильтрует новостройки по `offer.newbuilding.id` (serp.py:401-402,
listing_segment == "novostroyki"), считая SERP-параметр object_type=1
ненадёжным. Для ЕКБ, где выдача бралась без него, это было верно, здесь —
нет: базовый URL уже несёт object_type[0]=1, и Циан подтверждает скоуп
сам (puid3=sale_type_second).
Замер живой страницы 10.09: из 28 карточек 16 имеют блок newbuilding, и у
ВСЕХ шестнадцати isFromBuilder=false, isFromLeadFactory=false, flatType=rooms,
а ЖК — «Пригород Лесное», «РУСИЧ Новые Котельники» и подобные. То есть это
обычная вторичка в новых домах, а не продажа от застройщика: newbuilding.id
означает «дом в ЖК», а не «лот застройщика».
С фильтром прогон терял бы 57% корпуса безвозвратно, а msk_raw — сырьё:
payload несёт listing_segment целиком, и отделить сегмент можно потом, на
импорте в listings. Поэтому здесь только счётчик last_nb_ref для notes.
"""
lots = scraper._parse_serp_html(html)
# Сырое число карточек — по нему parse_page отличает блок от пустой страницы.
scraper.last_raw_count = len(lots)
nb_ref = sum(1 for lot in lots
if getattr(lot, "listing_segment", None) == "novostroyki")
if nb_ref:
scraper.last_nb_ref = getattr(scraper, "last_nb_ref", 0) + nb_ref
return lots
@dataclass(frozen=True)
class PlatformAdapter:
"""Платформо-зависимые куски сбора. Всё общее (бисекция, guard по HTTP-статусу,
накопитель/заливка) — в основном теле скрипта и от платформы не зависит."""
name: str
table: str # msk_raw.<table>
default_base_url: str
page_size: int
max_pages: int
build_url: Callable[[str, int, int | None, int | None], str]
extract_total_count: Callable[[Any, str], int | None]
parse_cards: Callable[[Any, str], list[Any]]
detect_block: Callable[[str], tuple[str, str] | None]
wait_ready: Callable[[Any], Any] # корутина: дождаться готовности страницы
@property
def hard_cap(self) -> int:
return self.page_size * self.max_pages
def make_scraper(self) -> Any:
if self.name == "avito":
# avito_serp_ekb_only=False обязателен: с True парсер выбрасывает всё,
# где в URL нет /ekaterinburg/ — то есть все подмосковные слаги
# (serp.py:2154).
return AvitoScraper(
SimpleNamespace(avito_serp_ekb_only=False), # type: ignore[arg-type]
target_city_slug="moskva",
)
# glitchtip_dsn=None — отключает попытку sentry-репорта schema-regression
# из _report_schema_regression (нет реального DSN в локальном прогоне).
return CianScraper(SimpleNamespace(glitchtip_dsn=None)) # type: ignore[arg-type]
ADAPTERS: dict[str, PlatformAdapter] = {
"avito": PlatformAdapter(
name="avito",
table="avito_cards",
default_base_url=DEFAULT_AVITO_BASE_URL,
page_size=AVITO_PAGE_SIZE,
max_pages=AVITO_MAX_PAGES,
build_url=_avito_build_url,
extract_total_count=lambda scraper, html: scraper._extract_total_count(html),
parse_cards=lambda scraper, html: scraper._parse_html(html, "https://www.avito.ru"),
detect_block=_avito_detect_block,
wait_ready=_avito_wait_ready,
),
"cian": PlatformAdapter(
name="cian",
table="cian_cards",
default_base_url=DEFAULT_CIAN_BASE_URL,
page_size=CIAN_PAGE_SIZE,
max_pages=CIAN_MAX_PAGES,
build_url=_cian_build_url,
extract_total_count=lambda scraper, html: scraper._extract_total_offers(html),
parse_cards=_cian_parse_cards,
detect_block=_cian_detect_block,
wait_ready=_cian_wait_ready,
),
}
# --- загрузка страницы -----------------------------------------------------
def _guard(html: str, status: int | None) -> None:
"""Порядок проверок фиксирован заданием; первое срабатывание = стоп."""
def _guard(html: str, status: int | None, adapter: PlatformAdapter) -> None:
"""Порядок проверок фиксирован заданием; первое срабатывание = стоп.
HTTP-статус — платформо-независимая проверка. Текстовые маркеры (firewall/
PoW и т.п.) — через adapter.detect_block, у каждой платформы свои.
"""
if status in (403, 439):
raise Blocked("platform", f"HTTP {status}")
if status == 429:
raise Blocked("ratelimit", "HTTP 429")
if _is_firewall_page(html):
raise Blocked("firewall", "firewall-страница на HTTP 200")
head = html[:4096].lower()
if any(m in head for m in _POW_MARKERS):
raise Blocked("challenge", "PoW / проверка безопасности")
extra = adapter.detect_block(html)
if extra is not None:
raise Blocked(*extra)
class Loader:
@ -206,14 +421,17 @@ class Loader:
это рабочий Chrome с залогиненным техаккаунтом.
"""
def __init__(self, delay: float, page_budget: int | None) -> None:
def __init__(self, delay: float, page_budget: int | None, adapter: PlatformAdapter) -> None:
self._delay = delay
self._budget = page_budget
self._adapter = adapter
self.loads = 0
self._page: Any = None
self._pw: Any = None
self._browser: Any = None
self._ctx: Any = None
self._last_load = 0.0
self._loads_on_page = 0
async def __aenter__(self) -> "Loader":
from playwright.async_api import async_playwright
@ -222,6 +440,8 @@ class Loader:
global PlaywrightTimeoutError
PlaywrightTimeoutError = _PwTimeout
# AVITO_CDP — исторически названо под первую платформу, но это адрес
# браузера владельца, а не площадки: используется для обеих платформ.
endpoint = os.environ.get("AVITO_CDP", "http://localhost:9222")
self._pw = await async_playwright().start()
try:
@ -230,7 +450,7 @@ class Loader:
await self._pw.stop()
raise SystemExit(
f"Не удалось подключиться по CDP к {endpoint}: {exc}\n"
"Запусти Chrome с залогиненным техаккаунтом Авито и ключом "
"Запусти Chrome с залогиненным техаккаунтом и ключом "
"--remote-debugging-port=9222, либо укажи адрес в AVITO_CDP."
) from exc
if not self._browser.contexts:
@ -239,8 +459,8 @@ class Loader:
"В подключённом Chrome нет ни одного контекста. Открой обычное окно "
"Chrome, запущенное с --remote-debugging-port=9222."
)
ctx = self._browser.contexts[0]
self._page = await ctx.new_page()
self._ctx = self._browser.contexts[0]
self._page = await self._ctx.new_page()
return self
async def __aexit__(self, *exc: object) -> None:
@ -270,19 +490,42 @@ class Loader:
async def fetch(self, url: str) -> tuple[str, int | None]:
if not self.budget_left():
raise BudgetExhausted()
await self._recycle_if_needed()
await self._pause()
resp = await self._goto(url)
self._loads_on_page += 1
self.loads += 1
self._last_load = time.monotonic()
status = resp.status if resp is not None else None
try:
await self._page.wait_for_selector('[data-marker="item"]', timeout=7_000)
except Exception: # noqa: BLE001 — пустая/блочная страница разбирается ниже
pass
await self._adapter.wait_ready(self._page)
html = await self._read_content()
_guard(html, status)
_guard(html, status, self._adapter)
return html, status
async def _recycle_if_needed(self) -> None:
"""Каждые PAGE_RECYCLE_EVERY загрузок пересоздаём свою вкладку.
Рендерер Chrome накапливает память по всем навигациям вкладки, а проход
по Москве — под тысячу страниц в одной. Наблюдалось живьём: вкладка
падала с «Опаньки… Код ошибки: Out of Memory» при 34 ГБ свободных в
системе, то есть упирался именно рендерер, а не машина. Свежая вкладка
стоит одну навигацию и обнуляет счёт.
Закрывается ТОЛЬКО своя вкладка; контекст и чужие вкладки владельца не
трогаются — это его рабочий Chrome.
"""
if self._loads_on_page < PAGE_RECYCLE_EVERY:
return
print(f" вкладка пересоздаётся после {self._loads_on_page} загрузок "
"(память рендерера)", flush=True)
old = self._page
self._page = await self._ctx.new_page()
self._loads_on_page = 0
try:
await old.close()
except Exception: # noqa: BLE001 — старая вкладка могла уже умереть
pass
async def _goto(self, url: str, attempts: int = 3):
"""goto с ограниченным ретраем на таймаут навигации.
@ -304,7 +547,7 @@ class Loader:
except Exception: # noqa: BLE001 — документа может не быть вовсе
partial = ""
if partial:
_guard(partial, None) # настоящий блок остановит прогон здесь
_guard(partial, None, self._adapter) # настоящий блок остановит прогон здесь
if i == attempts - 1:
raise Blocked("nav_timeout") from None
print(f" таймаут навигации, попытка {i + 2}/{attempts}", flush=True)
@ -352,35 +595,43 @@ def _csv_rows(rows: Iterable[dict[str, Any]]) -> str:
def build_sql(batch_id: str, query: str, rows: list[dict[str, Any]],
started_at: str, kind: str = "serp") -> str:
started_at: str, table: str, kind: str = "serp") -> str:
"""Один поток на `psql -f -`: batch (FK!) → TEMP staging → \\copy → INSERT.
Одиночный `psql -c` через ssh ломается на квотинге скобок и кавычек, поэтому
только поток. rows_new = разница count(*) по batch_id до и после вставки.
table — msk_raw.avito_cards / msk_raw.cian_cards (из PlatformAdapter, не из
пользовательского ввода — подстановка f-строкой безопасна).
У staging обязателен INCLUDING IDENTITY: если целевая таблица объявлена как
GENERATED ALWAYS AS IDENTITY, один INCLUDING DEFAULTS identity не переносит,
а NOT NULL переносится всегда — и \\copy без колонки id падает на каждом
батче.
"""
bid = _sql_str(batch_id)
tbl = f"msk_raw.{table}"
return (
"BEGIN;\n"
"INSERT INTO msk_raw.batches (batch_id, kind, query, started_at)\n"
f"VALUES ({bid}, {_sql_str(kind)}, {_sql_str(query)}, "
f"CAST({_sql_str(started_at)} AS timestamptz))\n"
"ON CONFLICT (batch_id) DO NOTHING;\n"
"CREATE TEMP TABLE _stg (LIKE msk_raw.avito_cards INCLUDING DEFAULTS) "
f"CREATE TEMP TABLE _stg (LIKE {tbl} INCLUDING DEFAULTS INCLUDING IDENTITY) "
"ON COMMIT DROP;\n"
"CREATE TEMP TABLE _before ON COMMIT DROP AS\n"
f" SELECT count(*) AS n FROM msk_raw.avito_cards WHERE batch_id = {bid};\n"
f" SELECT count(*) AS n FROM {tbl} WHERE batch_id = {bid};\n"
"\\copy _stg (source_id,observed_at,batch_id,kind,url,price,payload) "
"FROM STDIN WITH (FORMAT csv)\n"
+ _csv_rows(rows)
+ "\\.\n"
"INSERT INTO msk_raw.avito_cards "
f"INSERT INTO {tbl} "
"(source_id,observed_at,batch_id,kind,url,price,payload)\n"
"SELECT source_id,observed_at,batch_id,kind,url,price,payload FROM _stg\n"
"ON CONFLICT (source_id,batch_id,kind) DO NOTHING;\n"
"UPDATE msk_raw.batches b SET\n"
" rows_sent = coalesce(b.rows_sent,0) + (SELECT count(*) FROM _stg),\n"
" rows_new = coalesce(b.rows_new,0) +\n"
f" ((SELECT count(*) FROM msk_raw.avito_cards WHERE batch_id = {bid})\n"
f" ((SELECT count(*) FROM {tbl} WHERE batch_id = {bid})\n"
" - (SELECT n FROM _before))\n"
f"WHERE b.batch_id = {bid};\n"
"COMMIT;\n"
@ -398,18 +649,54 @@ def build_finalize_sql(batch_id: str, query: str, notes: str) -> str:
)
def run_psql(sql: str, ssh_host: str, container: str, db_user: str, db_name: str) -> None:
def run_psql(sql: str, ssh_host: str, container: str, db_user: str, db_name: str,
attempts: int = 4) -> None:
"""Заливка батча через ssh с ретраем на обрыв транспорта.
Прогон длится часами, и ssh рвётся: живьём поймано «Connection reset by peer»
(ssh возвращает 255) прямо посреди заливки — весь прогон умирал, а несброшенный
батч терялся. Ретраить безопасно: SQL идемпотентен (batch через ON CONFLICT DO
NOTHING, карточки через ON CONFLICT (source_id,batch_id,kind) DO NOTHING).
Ретраится ТОЛЬКО транспорт (ssh 255). Ошибка самого psql (ON_ERROR_STOP, любой
другой код) — это дефект данных или SQL, её повтор не лечит: поднимаем сразу.
"""
cmd = [
"ssh", ssh_host,
f"docker exec -i {container} psql -U {db_user} -d {db_name} "
"-v ON_ERROR_STOP=1 -f -",
]
proc = subprocess.run(cmd, input=sql.encode("utf-8"), capture_output=True)
out = (proc.stdout + proc.stderr).decode("utf-8", "replace").strip()
for i in range(attempts):
proc = subprocess.run(cmd, input=sql.encode("utf-8"), capture_output=True)
out = (proc.stdout + proc.stderr).decode("utf-8", "replace").strip()
if proc.returncode == 0:
if out:
print(f" psql: {out}", flush=True)
return
if proc.returncode != 255 or i == attempts - 1:
raise RuntimeError(f"psql через ssh вернул {proc.returncode}:\n{out}")
tail = out.splitlines()[-1] if out else "без вывода"
print(f" ssh оборвался ({tail}), повтор заливки {i + 2}/{attempts}", flush=True)
time.sleep(15.0 * (i + 1))
def psql_scalar(sql: str, ssh_host: str, container: str, db_user: str,
db_name: str) -> str:
"""Тот же ssh+psql, что и run_psql, но `-Atc` и с возвратом stdout.
Нужен для дешёвых проверок до старта (преflight целевой таблицы в collect):
один SELECT, без ретраев и без потока на stdin.
"""
cmd = [
"ssh", ssh_host,
f"docker exec -i {container} psql -U {db_user} -d {db_name} "
f"-v ON_ERROR_STOP=1 -Atc {shlex.quote(sql)}",
]
proc = subprocess.run(cmd, capture_output=True)
if proc.returncode != 0:
raise RuntimeError(f"psql через ssh вернул {proc.returncode}:\n{out}")
if out:
print(f" psql: {out}", flush=True)
err = (proc.stdout + proc.stderr).decode("utf-8", "replace").strip()
raise RuntimeError(f"psql через ssh вернул {proc.returncode}:\n{err}")
return proc.stdout.decode("utf-8", "replace").strip()
# --- накопитель карточек ---------------------------------------------------
@ -422,6 +709,7 @@ class Sink:
batch_id: str
started_at: str
query: str
table: str
batch_size: int
dry_run: bool
csv_path: Path
@ -468,10 +756,10 @@ class Sink:
print(f" [dry-run] {len(rows)} строк → {self.csv_path}", flush=True)
else:
run_psql(
build_sql(self.batch_id, self.query, rows, self.started_at),
build_sql(self.batch_id, self.query, rows, self.started_at, self.table),
self.ssh_host, self.container, self.db_user, self.db_name,
)
print(f" залито {len(rows)} строк в msk_raw.avito_cards", flush=True)
print(f" залито {len(rows)} строк в msk_raw.{self.table}", flush=True)
self.sent += len(rows)
def finalize(self, notes: str) -> None:
@ -488,45 +776,68 @@ class Sink:
# --- сбор ------------------------------------------------------------------
def parse_page(scraper: AvitoScraper, html: str, url: str) -> tuple[int | None, list[Any]]:
count = scraper._extract_total_count(html)
lots = scraper._parse_html(html, "https://www.avito.ru")
if not lots and count:
def parse_page(scraper: Any, html: str, url: str,
adapter: PlatformAdapter) -> tuple[int | None, list[Any]]:
"""Счётчик и карточки страницы + детект блока по СЫРОМУ числу карточек.
Гвард считает по сырому числу (до отброса новостроек), а не по итоговому
списку: на Циан страница может легально дать 0 карточек после фильтра
вторички, и такой штатный ноль неотличим от блока. scraper.last_raw_count
выставляет _cian_parse_cards; у Авито этого атрибута нет,
поэтому дефолт getattr — len(lots), и авито-ветка ведёт себя как раньше.
Это вторая сеть после текстовых маркеров в adapter.detect_block: блок,
не попавший под маркер, всплывает как проваленный extract_state, то есть
count=None и пустой парс одновременно.
"""
count = adapter.extract_total_count(scraper, html)
lots = adapter.parse_cards(scraper, html)
raw = getattr(scraper, "last_raw_count", len(lots))
if count is None and raw == 0:
raise Blocked(
"challenge",
f"страница без счётчика и без карточек (провал extract_state): {url}",
)
if raw == 0 and count:
raise Blocked("empty_page", f"0 карточек при счётчике {count}: {url}")
return count, lots
async def probe(loader: Loader, scraper: AvitoScraper, base_url: str,
lo: int | None, hi: int | None) -> tuple[int | None, list[Any]]:
url = build_url(base_url, 1, lo, hi)
async def probe(loader: Loader, scraper: Any, base_url: str,
lo: int | None, hi: int | None,
adapter: PlatformAdapter) -> tuple[int | None, list[Any]]:
url = adapter.build_url(base_url, 1, lo, hi)
html, _ = await loader.fetch(url)
return parse_page(scraper, html, url)
return parse_page(scraper, html, url, adapter)
async def build_plan(loader: Loader, scraper: AvitoScraper, base_url: str, target: int,
cache: dict[tuple[int | None, int | None], list[Any]]
async def build_plan(loader: Loader, scraper: Any, base_url: str, target: int,
cache: dict[tuple[int | None, int | None], list[Any]],
adapter: PlatformAdapter,
) -> list[Corridor]:
"""Адаптивная бисекция по цене; страница 1 каждого коридора кэшируется."""
corridors: list[Corridor] = []
hard_cap = adapter.hard_cap
def emit(lo: int | None, hi: int | None, count: int | None,
truncated: bool, lots: list[Any]) -> None:
missed = max(0, (count or 0) - HARD_CAP) if truncated else 0
missed = max(0, (count or 0) - hard_cap) if truncated else 0
c = Corridor(lo=lo, hi=hi, count=count, truncated=truncated, missed=missed)
corridors.append(c)
cache[(lo, hi)] = lots
flag = " TRUNCATED" if truncated else ""
print(f" коридор {c.label()} count={count} "
f"страниц={c.planned_pages()}{flag}", flush=True)
f"страниц={c.planned_pages(adapter.page_size, adapter.max_pages)}{flag}",
flush=True)
if truncated:
print(f" ВНИМАНИЕ: коридор {c.label()} не влезает в потолок "
f"{HARD_CAP}; заведомо не добрано ~{missed} объявлений", flush=True)
f"{hard_cap}; заведомо не добрано ~{missed} объявлений", flush=True)
async def find_upper(lo: int | None) -> int:
"""Верхнюю границу открытого коридора ищем удвоением от разумного старта."""
cand = max(int(lo or PRICE_FLOOR) * 2, PRICE_PROBE_START)
while cand < PRICE_CEIL:
cnt, _ = await probe(loader, scraper, base_url, cand, None)
cnt, _ = await probe(loader, scraper, base_url, cand, None, adapter)
print(f" проба хвоста pmin={cand:_} count={cnt}", flush=True)
if cnt is not None and cnt <= target:
return cand
@ -536,39 +847,35 @@ async def build_plan(loader: Loader, scraper: AvitoScraper, base_url: str, targe
async def split(lo: int | None, hi: int | None, depth: int,
count: int | None, lots: list[Any]) -> None:
if count is None:
url = build_url(base_url, 1, lo, hi)
url = adapter.build_url(base_url, 1, lo, hi)
raise Blocked("empty_page", f"счётчик не прочитался: {url}")
if count <= target:
emit(lo, hi, count, False, lots)
return
if depth >= MAX_DEPTH or width_ratio(lo, hi) <= MIN_WIDTH_RATIO:
# Предохранитель: не молчим — помечаем truncated и считаем недобор.
emit(lo, hi, count, count > HARD_CAP, lots)
emit(lo, hi, count, count > hard_cap, lots)
return
upper = hi if hi is not None else await find_upper(lo)
if hi is None:
tail_cnt, tail_lots = await probe(loader, scraper, base_url, upper, None)
tail_cnt, tail_lots = await probe(loader, scraper, base_url, upper, None, adapter)
emit(upper, None, tail_cnt,
bool(tail_cnt and tail_cnt > HARD_CAP), tail_lots)
bool(tail_cnt and tail_cnt > hard_cap), tail_lots)
mid = geometric_mid(lo, upper)
for sub_lo, sub_hi in ((lo, mid), (mid, upper)):
sub_cnt, sub_lots = await probe(loader, scraper, base_url, sub_lo, sub_hi)
sub_cnt, sub_lots = await probe(loader, scraper, base_url, sub_lo, sub_hi, adapter)
print(f" проба {sub_lo or '-'}..{sub_hi} count={sub_cnt}", flush=True)
await split(sub_lo, sub_hi, depth + 1, sub_cnt, sub_lots)
root_cnt, root_lots = await probe(loader, scraper, base_url, None, None)
root_cnt, root_lots = await probe(loader, scraper, base_url, None, None, adapter)
print(f"Всего по базовому запросу: {root_cnt}", flush=True)
await split(None, None, 0, root_cnt, root_lots)
return corridors
async def collect(args: argparse.Namespace) -> int:
# avito_serp_ekb_only=False обязателен: с True парсер выбрасывает всё, где в
# URL нет /ekaterinburg/ — то есть все подмосковные слаги (serp.py:2154).
scraper = AvitoScraper(
SimpleNamespace(avito_serp_ekb_only=False), # type: ignore[arg-type]
target_city_slug="moskva",
)
adapter = ADAPTERS[args.platform]
scraper = adapter.make_scraper()
out_dir = Path(args.out_dir).resolve()
out_dir.mkdir(parents=True, exist_ok=True)
plan_path = out_dir / f"plan-{args.batch_id}.json"
@ -585,9 +892,14 @@ async def collect(args: argparse.Namespace) -> int:
print(f"Resume по {plan_path}: коридоров {len(plan.corridors)}, "
f"готово {done}", flush=True)
# Resume: URL берём из сохранённого плана, а не из CLI — коридоры посчитаны
# именно под него. Расхождение = молчаливая заливка чужой выдачи под тем же
# batch_id, поэтому это ошибка, а не тихий приоритет одного из двух.
# Resume: URL и платформу берём из сохранённого плана, а не из CLI — коридоры
# посчитаны именно под них. Расхождение = молчаливая заливка чужой выдачи под
# тем же batch_id, поэтому это ошибка, а не тихий приоритет одного из двух.
if plan is not None and plan.platform != args.platform:
raise SystemExit(
"--resume: план построен для другой платформы."
f" В плане {plan.platform}, в аргументах {args.platform}."
)
if plan is not None and plan.base_url != args.base_url:
raise SystemExit(
"--resume: план построен для другого URL."
@ -598,13 +910,27 @@ async def collect(args: argparse.Namespace) -> int:
page_budget = None if args.full else args.measure
mode = "FULL" if args.full else f"MEASURE<={page_budget}"
print(f"Режим: {mode}; batch_id={args.batch_id}; delay={args.delay}s; "
f"target={args.target_count}; dry_run={args.dry_run}", flush=True)
print(f"Режим: {mode}; platform={args.platform}; batch_id={args.batch_id}; "
f"delay={args.delay}s; target={args.target_count}; dry_run={args.dry_run}",
flush=True)
# Преflight целевой таблицы: без неё прогон умирал бы только на первой
# заливке — после часов планирования и сбора. Проверяем на первой секунде.
if not args.dry_run:
if not psql_scalar(f"SELECT to_regclass('msk_raw.{adapter.table}')",
args.ssh_host, args.container,
args.db_user, args.db_name):
raise SystemExit(
f"Таблицы msk_raw.{adapter.table} на проде нет. Применить"
" tradein-mvp/backend/data/sql/"
"299_msk_raw_cian_domclick_yandex_cards.sql"
" либо гонять с --dry-run."
)
sink = Sink(
batch_id=args.batch_id, started_at=started_at, query=base_url,
batch_size=args.batch_size, dry_run=args.dry_run, csv_path=csv_path,
ssh_host=args.ssh_host, container=args.container,
table=adapter.table, batch_size=args.batch_size, dry_run=args.dry_run,
csv_path=csv_path, ssh_host=args.ssh_host, container=args.container,
db_user=args.db_user, db_name=args.db_name,
)
cache: dict[tuple[int | None, int | None], list[Any]] = {}
@ -613,15 +939,15 @@ async def collect(args: argparse.Namespace) -> int:
rc = 0
loads = 0
async with Loader(args.delay, page_budget) as loader:
async with Loader(args.delay, page_budget, adapter) as loader:
try:
if plan is None:
print("Строю план коридоров...", flush=True)
corridors = await build_plan(loader, scraper, base_url,
args.target_count, cache)
args.target_count, cache, adapter)
plan = Plan(base_url=base_url, target=args.target_count,
batch_id=args.batch_id, corridors=corridors,
created_at=started_at)
batch_id=args.batch_id, platform=args.platform,
corridors=corridors, created_at=started_at)
plan.save(plan_path)
print(f"План сохранён: {plan_path} ({len(corridors)} коридоров)",
flush=True)
@ -629,7 +955,7 @@ async def collect(args: argparse.Namespace) -> int:
for corridor in plan.corridors:
if corridor.status == "done":
continue
pages = corridor.planned_pages()
pages = corridor.planned_pages(adapter.page_size, adapter.max_pages)
print(f"Коридор {corridor.label()} count={corridor.count} "
f"страниц={pages} (с {corridor.pages_done + 1})", flush=True)
for page in range(corridor.pages_done + 1, pages + 1):
@ -637,9 +963,9 @@ async def collect(args: argparse.Namespace) -> int:
if page == 1 and key in cache:
lots = cache.pop(key) # страница 1 уже скачана при планировании
else:
url = build_url(base_url, page, corridor.lo, corridor.hi)
url = adapter.build_url(base_url, page, corridor.lo, corridor.hi)
html, _ = await loader.fetch(url)
_, lots = parse_page(scraper, html, url)
_, lots = parse_page(scraper, html, url, adapter)
for lot in lots:
sink.add(lot)
total += len(lots)
@ -667,9 +993,11 @@ async def collect(args: argparse.Namespace) -> int:
truncated = [c for c in (plan.corridors if plan else []) if c.truncated]
missed = sum(c.missed for c in truncated)
nb_ref = getattr(scraper, "last_nb_ref", 0)
notes = "; ".join(x for x in [
f"mode={mode}", f"loads={loads}", f"cards={total}",
f"mode={mode}", f"platform={args.platform}", f"loads={loads}", f"cards={total}",
f"skipped_non_numeric={sink.skipped_non_numeric}",
(f"nb_ref={nb_ref}" if nb_ref else ""),
(f"truncated_corridors={len(truncated)} missed~{missed}" if truncated else ""),
stop_reason,
] if x)
@ -687,10 +1015,12 @@ async def collect(args: argparse.Namespace) -> int:
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
p = argparse.ArgumentParser(
prog="collect.py",
description="Ручной сбор SERP Авито (вторичка, Москва+МО) в прод-схему msk_raw.",
description="Ручной сбор SERP Авито/Циан (вторичка, Москва+МО) в прод-схему msk_raw.",
)
p.add_argument("--base-url", default=DEFAULT_BASE_URL,
help="базовый URL выдачи (дефолт — вкладка владельца)")
p.add_argument("--platform", choices=tuple(ADAPTERS), default="avito",
help="площадка сбора (дефолт avito)")
p.add_argument("--base-url", default=None,
help="базовый URL выдачи (дефолт зависит от --platform)")
p.add_argument("--measure", type=int, default=100, metavar="N",
help="режим замера: не больше N загрузок страниц (дефолт 100)")
p.add_argument("--full", action="store_true",
@ -706,7 +1036,7 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
p.add_argument("--target-count", type=int, default=1500,
help="целевой размер коридора; больше — делим (дефолт 1500)")
p.add_argument("--batch-id", default=None,
help="batch_id в msk_raw.batches (дефолт msk-serp-<UTC>)")
help="batch_id в msk_raw.batches (дефолт msk-serp-<platform>-<UTC>)")
p.add_argument("--out-dir", default=str(Path(__file__).resolve().parent / "runs"),
help="каталог плана/CSV")
p.add_argument("--ssh-host", default="selectel", help="ssh-хост прода")
@ -715,8 +1045,12 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
p.add_argument("--db-user", default="tradein")
p.add_argument("--db-name", default="tradein")
args = p.parse_args(argv)
if args.base_url is None:
args.base_url = ADAPTERS[args.platform].default_base_url
if args.batch_id is None:
args.batch_id = "msk-serp-" + datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
args.batch_id = (
f"msk-serp-{args.platform}-" + datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
)
if not _BATCH_ID_RE.match(args.batch_id):
p.error("--batch-id: допустимы только символы [A-Za-z0-9._-]")
if args.measure < 1: