443 lines
18 KiB
Python
443 lines
18 KiB
Python
"""Full ЕКБ sweep + houses enrichment — runs locally через residential IP.
|
||
|
||
Why: prod VPS Beget IP заблокирован Avito (datacenter fingerprint → 429).
|
||
Local IP (домашний ISP) = residential → проходит. Скрипт пишет в prod DB
|
||
через SSH tunnel (localhost:15433 → tradein-postgres docker container).
|
||
|
||
Setup:
|
||
ssh -f -N -L 15433:172.20.0.2:5432 gendesign # tradein-postgres tunnel
|
||
cd tradein-mvp/backend
|
||
export DATABASE_URL='postgresql+psycopg://tradein:<PASS>@localhost:15433/tradein'
|
||
|
||
Usage:
|
||
uv run python ../scripts/local-sweep-ekb.py # defaults
|
||
uv run python ../scripts/local-sweep-ekb.py --pages 3 --detail-top 20 --delay 30
|
||
uv run python ../scripts/local-sweep-ekb.py --skip-search # только detail + houses
|
||
uv run python ../scripts/local-sweep-ekb.py --skip-houses # без houses
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
# stdlib
|
||
import argparse
|
||
import asyncio
|
||
import sys
|
||
import time
|
||
import types
|
||
from pathlib import Path
|
||
from urllib.parse import urlparse
|
||
|
||
# Auto-inject backend/ в sys.path чтобы найти app.* package
|
||
# (скрипт лежит в tradein-mvp/scripts/, app — в tradein-mvp/backend/)
|
||
_BACKEND_DIR = Path(__file__).resolve().parent.parent / "backend"
|
||
if str(_BACKEND_DIR) not in sys.path:
|
||
sys.path.insert(0, str(_BACKEND_DIR))
|
||
|
||
# Stub weasyprint (Windows без GTK ломает import chain через trade_in.py)
|
||
for _m in (
|
||
"weasyprint",
|
||
"weasyprint.text",
|
||
"weasyprint.text.ffi",
|
||
"weasyprint.css",
|
||
"weasyprint.css.targets",
|
||
):
|
||
sys.modules[_m] = types.ModuleType(_m)
|
||
sys.modules["weasyprint"].HTML = type("HTML", (), {"__init__": lambda self, *a, **kw: None})
|
||
sys.modules["weasyprint"].CSS = type("CSS", (), {"__init__": lambda self, *a, **kw: None})
|
||
|
||
# app imports после stubs
|
||
from sqlalchemy import text # noqa: E402
|
||
|
||
from app.core.db import SessionLocal # noqa: E402
|
||
from app.services.scrapers.avito import AvitoScraper # noqa: E402
|
||
|
||
|
||
class CityWideAvitoScraper(AvitoScraper):
|
||
"""Subclass без geo-фильтра — search всего города (ЕКБ) через pagination.
|
||
|
||
Override `_build_web_url` чтобы НЕ добавлять `geoCoords`/`radius` — Avito
|
||
отдаёт ВСЕ объявления в городе sorted by date, до 100 pages (cap ~5000).
|
||
"""
|
||
|
||
def _build_web_url(self, lat: float, lon: float, radius_km: int, page: int = 1) -> str: # noqa: ARG002
|
||
from urllib.parse import urlencode
|
||
params = {"s": 104, "p": page} # s=104 sort by date
|
||
return f"{self.base_url}/ekaterinburg/kvartiry/prodam-ASgBAgICAUSSA8YQ?{urlencode(params)}"
|
||
|
||
|
||
# Avito URL slugs для room-filter (extracted из real search page HTML 2026-05-23)
|
||
ROOM_SLUGS: list[tuple[str, str]] = [
|
||
("студии", "studii-ASgBAgICAkSSA8YQygj~WA"),
|
||
("1-комн.", "1-komnatnye-ASgBAgICAkSSA8YQygiAWQ"),
|
||
("2-комн.", "2-komnatnye-ASgBAgICAkSSA8YQygiCWQ"),
|
||
("3-комн.", "3-komnatnye-ASgBAgICAkSSA8YQygiEWQ"),
|
||
("4-комн.", "4-komnatnye-ASgBAgICAkSSA8YQygiGWQ"),
|
||
("5+комн.", "5-komnatnye-ASgBAgICAkSSA8YQygiIWQ"),
|
||
("своб.планир.", "svobodnaya_planirovka-ASgBAgICAkSSA8YQygj8zzI"),
|
||
]
|
||
|
||
|
||
class CityRoomsAvitoScraper(AvitoScraper):
|
||
"""Subclass с URL slug filter по комнатности — для splitting когда total > 5000.
|
||
|
||
URL pattern: /ekaterinburg/kvartiry/prodam/<slug>?s=104&p=N
|
||
"""
|
||
|
||
def __init__(self, slug: str) -> None:
|
||
super().__init__()
|
||
self.room_slug = slug
|
||
|
||
def _build_web_url(self, lat: float, lon: float, radius_km: int, page: int = 1) -> str: # noqa: ARG002
|
||
from urllib.parse import urlencode
|
||
params = {"s": 104, "p": page}
|
||
return f"{self.base_url}/ekaterinburg/kvartiry/prodam/{self.room_slug}?{urlencode(params)}"
|
||
from app.services.scrapers.avito_detail import ( # noqa: E402
|
||
fetch_detail,
|
||
save_detail_enrichment,
|
||
)
|
||
from app.services.scrapers.avito_houses import ( # noqa: E402
|
||
fetch_house_catalog,
|
||
save_house_catalog_enrichment,
|
||
)
|
||
from app.services.scrapers.base import save_listings # noqa: E402
|
||
|
||
# 10 anchors покрывающих центральные/северные/южные/восточные/западные районы ЕКБ
|
||
ANCHORS: list[tuple[str, float, float]] = [
|
||
("Центр", 56.8400, 60.6050),
|
||
("ЮЗ", 56.7950, 60.5300),
|
||
("Уралмаш", 56.8970, 60.6100),
|
||
("Академический", 56.7700, 60.5500),
|
||
("Пионерский", 56.8650, 60.6200),
|
||
("Эльмаш", 56.8567, 60.6700),
|
||
("Втузгородок", 56.8400, 60.6650),
|
||
("ВИЗ", 56.8500, 60.5650),
|
||
("Юго-Запад", 56.7770, 60.5800),
|
||
("Сортировка", 56.8950, 60.5600),
|
||
]
|
||
|
||
|
||
async def phase_citywide(pages: int, delay: float) -> int:
|
||
"""City-wide pagination — ВСЕ объявления ЕКБ (no geo filter), per-page save.
|
||
|
||
Per-page save важно: visible прогресс + не теряем данные если crash mid-way.
|
||
"""
|
||
print(
|
||
f"\n=== PHASE CITYWIDE: full ЕКБ pagination ({pages} pages, delay {delay}s) ===\n",
|
||
flush=True,
|
||
)
|
||
total = 0
|
||
cum_ins = 0
|
||
cum_upd = 0
|
||
async with CityWideAvitoScraper() as s:
|
||
for page in range(1, pages + 1):
|
||
url = s._build_web_url(0.0, 0.0, 1, page=page)
|
||
try:
|
||
assert s._cffi is not None
|
||
response = await s._cffi.get(url)
|
||
if response.status_code != 200:
|
||
print(f" ⚠️ page={page} HTTP {response.status_code} — stop", flush=True)
|
||
break
|
||
lots = s._parse_html(response.text, source_url_base=url)
|
||
except Exception as e: # noqa: BLE001
|
||
print(f" page={page} FAIL fetch: {type(e).__name__}: {e}", flush=True)
|
||
await asyncio.sleep(delay)
|
||
continue
|
||
if not lots:
|
||
print(f" page={page}: 0 lots — end of pagination", flush=True)
|
||
break
|
||
db = SessionLocal()
|
||
try:
|
||
ins, upd = save_listings(db, lots)
|
||
cum_ins += ins
|
||
cum_upd += upd
|
||
total += len(lots)
|
||
print(
|
||
f" page={page}/{pages}: {len(lots)} lots → ins={ins} upd={upd} "
|
||
f"[cum: total={total} new={cum_ins} upd={cum_upd}]",
|
||
flush=True,
|
||
)
|
||
except Exception as e: # noqa: BLE001
|
||
print(f" page={page} FAIL save: {type(e).__name__}: {e}", flush=True)
|
||
finally:
|
||
db.close()
|
||
if page < pages:
|
||
await asyncio.sleep(delay)
|
||
print(f" PHASE CITYWIDE TOTAL: fetched={total} ins={cum_ins} upd={cum_upd}", flush=True)
|
||
return total
|
||
|
||
|
||
async def phase_byrooms(pages: int, delay: float) -> int:
|
||
"""Rooms-split sweep — 7 categories × pages, per-page save.
|
||
|
||
Для total > 5000 (Avito search cap) split по комнатности позволяет получить
|
||
все listings (каждый segment <5000).
|
||
"""
|
||
print(
|
||
f"\n=== PHASE BYROOMS: 7 категорий × {pages} pages, delay {delay}s ===\n",
|
||
flush=True,
|
||
)
|
||
grand_total = 0
|
||
grand_ins = 0
|
||
grand_upd = 0
|
||
import random
|
||
rate_limit_hits = 0
|
||
for ci, (name, slug) in enumerate(ROOM_SLUGS, 1):
|
||
print(f"\n--- [{ci}/{len(ROOM_SLUGS)}] Категория: {name} (slug={slug[:30]}...) ---", flush=True)
|
||
cat_total = 0
|
||
cat_ins = 0
|
||
cat_upd = 0
|
||
async with CityRoomsAvitoScraper(slug) as s:
|
||
for page in range(1, pages + 1):
|
||
url = s._build_web_url(0.0, 0.0, 1, page=page)
|
||
try:
|
||
assert s._cffi is not None
|
||
response = await s._cffi.get(url)
|
||
if response.status_code == 429:
|
||
# Exponential backoff на rate-limit: 60s × 2^N (max 30 min)
|
||
rate_limit_hits += 1
|
||
backoff = min(60 * (2 ** (rate_limit_hits - 1)), 1800)
|
||
print(f" {name} p={page} HTTP 429 (hit #{rate_limit_hits}) → backoff {backoff}s", flush=True)
|
||
await asyncio.sleep(backoff)
|
||
# retry same page after backoff
|
||
try:
|
||
response = await s._cffi.get(url)
|
||
if response.status_code != 200:
|
||
print(f" {name} p={page} retry HTTP {response.status_code} — skip page", flush=True)
|
||
await asyncio.sleep(delay * 2)
|
||
continue
|
||
except Exception as e: # noqa: BLE001
|
||
print(f" {name} p={page} retry FAIL: {type(e).__name__}: {e}", flush=True)
|
||
continue
|
||
elif response.status_code != 200:
|
||
print(f" {name} p={page} HTTP {response.status_code} — skip page", flush=True)
|
||
await asyncio.sleep(delay)
|
||
continue
|
||
lots = s._parse_html(response.text, source_url_base=url)
|
||
except Exception as e: # noqa: BLE001
|
||
print(f" {name} p={page} FAIL fetch: {type(e).__name__}: {e}", flush=True)
|
||
await asyncio.sleep(delay)
|
||
continue
|
||
if not lots:
|
||
print(f" {name} page={page}: 0 lots — end of category", flush=True)
|
||
break
|
||
db = SessionLocal()
|
||
try:
|
||
ins, upd = save_listings(db, lots)
|
||
cat_ins += ins
|
||
cat_upd += upd
|
||
cat_total += len(lots)
|
||
print(
|
||
f" {name} p={page}/{pages}: {len(lots)} lots → ins={ins} upd={upd} "
|
||
f"[cat: tot={cat_total} new={cat_ins}]",
|
||
flush=True,
|
||
)
|
||
except Exception as e: # noqa: BLE001
|
||
print(f" {name} page={page} FAIL save: {type(e).__name__}: {e}", flush=True)
|
||
finally:
|
||
db.close()
|
||
if page < pages:
|
||
# Jitter: ±30% от delay чтобы выглядеть менее robotic
|
||
jittered = delay * random.uniform(0.7, 1.3)
|
||
await asyncio.sleep(jittered)
|
||
print(
|
||
f" --- {name} TOTAL: fetched={cat_total} ins={cat_ins} upd={cat_upd} ---",
|
||
flush=True,
|
||
)
|
||
grand_total += cat_total
|
||
grand_ins += cat_ins
|
||
grand_upd += cat_upd
|
||
print(
|
||
f"\n PHASE BYROOMS GRAND TOTAL: fetched={grand_total} new={grand_ins} updated={grand_upd}",
|
||
flush=True,
|
||
)
|
||
return grand_total
|
||
|
||
|
||
async def phase_a_search(pages: int, delay: float) -> int:
|
||
"""Search всех anchors × pages, save listings."""
|
||
print(
|
||
f"\n=== PHASE A: search ({len(ANCHORS)} anchors × {pages} pages, "
|
||
f"delay {delay}s) ===\n",
|
||
flush=True,
|
||
)
|
||
total = 0
|
||
for i, (name, lat, lon) in enumerate(ANCHORS, 1):
|
||
print(f"[{i}/{len(ANCHORS)}] {name} ({lat},{lon})", flush=True)
|
||
try:
|
||
async with AvitoScraper() as s:
|
||
lots = await s.fetch_around(
|
||
lat, lon, 1500, pages=pages, delay_override_sec=delay
|
||
)
|
||
except Exception as e: # noqa: BLE001
|
||
print(f" FAIL fetch: {type(e).__name__}: {e}", flush=True)
|
||
await asyncio.sleep(delay)
|
||
continue
|
||
print(f" fetched {len(lots)} lots", flush=True)
|
||
if not lots:
|
||
print(" ⚠️ 0 lots — possible 429, stop phase A", flush=True)
|
||
break
|
||
db = SessionLocal()
|
||
try:
|
||
ins, upd = save_listings(db, lots)
|
||
print(f" saved: ins={ins} upd={upd}", flush=True)
|
||
total += len(lots)
|
||
except Exception as e: # noqa: BLE001
|
||
print(f" FAIL save: {type(e).__name__}: {e}", flush=True)
|
||
finally:
|
||
db.close()
|
||
if i < len(ANCHORS):
|
||
await asyncio.sleep(delay)
|
||
return total
|
||
|
||
|
||
async def phase_b_detail(top_n: int, delay: float) -> tuple[int, set[str]]:
|
||
"""Detail enrichment для top-N recent listings без detail.
|
||
|
||
Returns: (enriched_count, set of unique house_catalog_url paths extracted)
|
||
"""
|
||
print(
|
||
f"\n=== PHASE B: detail enrichment (top {top_n}, delay {delay}s) ===\n",
|
||
flush=True,
|
||
)
|
||
db = SessionLocal()
|
||
try:
|
||
rows = db.execute(
|
||
text(
|
||
"""
|
||
SELECT source_url, source_id FROM listings
|
||
WHERE source='avito' AND detail_enriched_at IS NULL
|
||
AND source_url IS NOT NULL AND price_rub > 0
|
||
ORDER BY scraped_at DESC NULLS LAST
|
||
LIMIT :limit
|
||
"""
|
||
),
|
||
{"limit": top_n},
|
||
).mappings().all()
|
||
finally:
|
||
db.close()
|
||
|
||
print(f" queue: {len(rows)} listings to enrich", flush=True)
|
||
enriched = 0
|
||
failed = 0
|
||
house_urls: set[str] = set()
|
||
for i, row in enumerate(rows, 1):
|
||
url = row["source_url"]
|
||
path = urlparse(url).path if url.startswith("http") else url
|
||
print(f"[{i}/{len(rows)}] detail {row['source_id']}", flush=True)
|
||
try:
|
||
enr = await fetch_detail(path)
|
||
db = SessionLocal()
|
||
try:
|
||
ok = save_detail_enrichment(db, enr)
|
||
print(
|
||
f" enriched: ok={ok} "
|
||
f"rooms={enr.rooms} area={enr.area_m2} "
|
||
f"kitchen={enr.kitchen_area_m2} owners={enr.owners_count} "
|
||
f"house_url={enr.house_catalog_url}",
|
||
flush=True,
|
||
)
|
||
if ok:
|
||
enriched += 1
|
||
# Извлекаем house path (strip query string + filter только /catalog/houses/)
|
||
if enr.house_catalog_url:
|
||
p = urlparse(enr.house_catalog_url).path
|
||
if p.startswith("/catalog/houses/"):
|
||
house_urls.add(p)
|
||
finally:
|
||
db.close()
|
||
except Exception as e: # noqa: BLE001
|
||
print(f" FAIL: {type(e).__name__}: {e}", flush=True)
|
||
failed += 1
|
||
if i < len(rows):
|
||
await asyncio.sleep(delay)
|
||
print(f" total: enriched={enriched} failed={failed} unique_houses_extracted={len(house_urls)}", flush=True)
|
||
return enriched, house_urls
|
||
|
||
|
||
async def phase_c_houses(delay: float, extra_urls: set[str] | None = None) -> int:
|
||
"""Houses enrichment — unique house_url из listings (DB) + extra_urls (из Phase B in-memory).
|
||
|
||
save_detail_enrichment не пишет house_url в listings (баг Stage 2b),
|
||
поэтому Phase C принимает extra_urls собранные Phase B вручную.
|
||
"""
|
||
print("\n=== PHASE C: houses enrichment ===\n", flush=True)
|
||
db = SessionLocal()
|
||
try:
|
||
rows = db.execute(
|
||
text(
|
||
"""
|
||
SELECT DISTINCT house_url FROM listings
|
||
WHERE source='avito' AND house_url IS NOT NULL
|
||
AND house_id_fk IS NULL
|
||
ORDER BY house_url
|
||
"""
|
||
)
|
||
).fetchall()
|
||
finally:
|
||
db.close()
|
||
|
||
urls_from_db = {r[0] for r in rows if r[0]}
|
||
urls = urls_from_db | (extra_urls or set())
|
||
print(f" queue: {len(urls)} unique houses (db={len(urls_from_db)} +extras={len(extra_urls or set())})", flush=True)
|
||
if not urls:
|
||
print(" no houses to enrich (skip)", flush=True)
|
||
return 0
|
||
|
||
enriched = 0
|
||
failed = 0
|
||
for i, url in enumerate(sorted(urls), 1):
|
||
path = urlparse(url).path if url.startswith("http") else url
|
||
print(f"[{i}/{len(urls)}] {path}", flush=True)
|
||
try:
|
||
enr = await fetch_house_catalog(path)
|
||
db = SessionLocal()
|
||
try:
|
||
counters = save_house_catalog_enrichment(db, enr)
|
||
print(f" saved: {counters}", flush=True)
|
||
enriched += 1
|
||
finally:
|
||
db.close()
|
||
except Exception as e: # noqa: BLE001
|
||
print(f" FAIL: {type(e).__name__}: {e}", flush=True)
|
||
failed += 1
|
||
if i < len(urls):
|
||
await asyncio.sleep(delay)
|
||
print(f" total: enriched={enriched} failed={failed}", flush=True)
|
||
return enriched
|
||
|
||
|
||
async def main() -> None:
|
||
p = argparse.ArgumentParser()
|
||
p.add_argument(
|
||
"--mode",
|
||
choices=["anchors", "citywide", "byrooms", "both"],
|
||
default="anchors",
|
||
help="anchors=10 anchors; citywide=full pagination без geo; byrooms=7 категорий комнат (для total>5000); both=anchors+citywide",
|
||
)
|
||
p.add_argument("--pages", type=int, default=5, help="pages per anchor (anchors mode) или total pages (citywide)")
|
||
p.add_argument("--detail-top", type=int, default=50, help="top-N для detail enrichment")
|
||
p.add_argument("--delay", type=float, default=30, help="sec между HTTP requests")
|
||
p.add_argument("--skip-search", action="store_true")
|
||
p.add_argument("--skip-detail", action="store_true")
|
||
p.add_argument("--skip-houses", action="store_true")
|
||
args = p.parse_args()
|
||
|
||
t0 = time.perf_counter()
|
||
if not args.skip_search:
|
||
if args.mode in ("anchors", "both"):
|
||
await phase_a_search(args.pages, args.delay)
|
||
if args.mode in ("citywide", "both"):
|
||
cw_pages = args.pages if args.mode == "citywide" else 100
|
||
await phase_citywide(cw_pages, args.delay)
|
||
if args.mode == "byrooms":
|
||
await phase_byrooms(args.pages, args.delay)
|
||
extracted_house_urls: set[str] = set()
|
||
if not args.skip_detail:
|
||
_, extracted_house_urls = await phase_b_detail(args.detail_top, args.delay)
|
||
if not args.skip_houses:
|
||
await phase_c_houses(args.delay, extra_urls=extracted_house_urls)
|
||
elapsed = time.perf_counter() - t0
|
||
print(f"\n=== ALL DONE in {elapsed / 60:.1f} min ===", flush=True)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|