256 lines
11 KiB
Python
256 lines
11 KiB
Python
"""Cian history backfill через real Chrome (Playwright) — обходит anti-bot detection.
|
||
|
||
Iterates Cian listings без offer_price_history, открывает в headless Chrome,
|
||
extract'ит defaultState.offerData.priceChanges через page.evaluate, пишет в БД.
|
||
|
||
Requires:
|
||
pip install playwright "sqlalchemy>=2" "psycopg[binary]"
|
||
playwright install chromium
|
||
|
||
Run:
|
||
$env:TRADEIN_DB_DSN = "postgresql+psycopg://tradein:PASS@localhost:15433/tradein"
|
||
python C:\\Users\\user\\cian-backfill\\playwright_history.py --batch 20 --delay 2
|
||
|
||
Flags:
|
||
--headed показать браузер (debug)
|
||
--batch N листингов за батч (default 20)
|
||
--delay S seconds между listings (default 2)
|
||
--max N hard cap (default 5500)
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import asyncio
|
||
import os
|
||
import signal
|
||
import sys
|
||
import time
|
||
from datetime import datetime, timezone
|
||
|
||
import json
|
||
|
||
import psycopg
|
||
from playwright.async_api import async_playwright
|
||
|
||
EXTRACT_JS = """
|
||
() => {
|
||
if (!window._cianConfig || !window._cianConfig['frontend-offer-card']) {
|
||
return { error: 'no _cianConfig', captcha: location.href.includes('captcha') };
|
||
}
|
||
const arr = window._cianConfig['frontend-offer-card'];
|
||
const def = arr.find(e => e && e.key === 'defaultState');
|
||
if (!def || !def.value) {
|
||
return { error: 'no defaultState' };
|
||
}
|
||
const v = def.value;
|
||
const offerData = v.offerData || {};
|
||
const offer = offerData.offer || {};
|
||
// priceChanges живёт в offerData.priceChanges (НЕ в offerData.offer.priceChanges)
|
||
const priceChanges = offerData.priceChanges || offer.priceChanges || v.priceChanges || [];
|
||
// bargainTerms.price — текущая цена
|
||
const price_rub = (offer.bargainTerms && offer.bargainTerms.price) || null;
|
||
// Также собрать ключи offerData чтобы видеть что доступно
|
||
return {
|
||
cian_id: offer.id || null,
|
||
price_rub,
|
||
price_changes: priceChanges,
|
||
_offerData_keys: Object.keys(offerData).slice(0, 25),
|
||
_offer_keys: Object.keys(offer).slice(0, 25),
|
||
};
|
||
}
|
||
"""
|
||
|
||
|
||
def parse_change_time(s):
|
||
"""Parse 'YYYY-MM-DDTHH:MM:SS' or ISO → timestamptz; return raw string if parse fails."""
|
||
if not s:
|
||
return None
|
||
return s # psycopg позволяет передать строку → postgres парсит как timestamptz
|
||
|
||
|
||
def upsert_changes(cur, listing_id: int, price_rub: int | None, changes: list) -> int:
|
||
"""Insert priceChanges array; return count inserted (ignoring conflicts)."""
|
||
if not changes:
|
||
return 0
|
||
rows = []
|
||
prev_price = None
|
||
for c in changes:
|
||
ct = c.get("changeTime") or c.get("date")
|
||
# priceData.price — основной формат 2026+; fallback на старые поля
|
||
pd = c.get("priceData") if isinstance(c.get("priceData"), dict) else None
|
||
pr = (pd.get("price") if pd else None) or c.get("price") or c.get("priceRub")
|
||
if not ct or not pr:
|
||
continue
|
||
diff_pct = None
|
||
if prev_price is not None and prev_price > 0:
|
||
diff_pct = round((pr - prev_price) / prev_price * 100, 2)
|
||
rows.append((listing_id, ct, pr, diff_pct, "cian"))
|
||
prev_price = pr
|
||
if not rows:
|
||
return 0
|
||
cur.executemany(
|
||
"""
|
||
INSERT INTO offer_price_history (listing_id, change_time, price_rub, diff_percent, source)
|
||
VALUES (%s, %s, %s, %s, %s)
|
||
ON CONFLICT (listing_id, change_time) DO NOTHING
|
||
""",
|
||
rows,
|
||
)
|
||
return cur.rowcount or 0
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--batch", type=int, default=20)
|
||
ap.add_argument("--delay", type=float, default=2.0)
|
||
ap.add_argument("--max", type=int, default=5500)
|
||
ap.add_argument("--headed", action="store_true")
|
||
args = ap.parse_args()
|
||
|
||
dsn = os.environ.get("TRADEIN_DB_DSN")
|
||
if not dsn:
|
||
sys.exit("FATAL: TRADEIN_DB_DSN env required")
|
||
# psycopg native dsn (без +psycopg для sqlalchemy)
|
||
pg_dsn = dsn.replace("postgresql+psycopg://", "postgresql://")
|
||
|
||
interrupted = False
|
||
|
||
def on_sigint(s, f):
|
||
nonlocal interrupted
|
||
print("\n!! Ctrl+C — finishing current item")
|
||
interrupted = True
|
||
signal.signal(signal.SIGINT, on_sigint)
|
||
|
||
async def run() -> None:
|
||
nonlocal interrupted
|
||
cum_ok = cum_fail = cum_changes = cum_with_history = 0
|
||
start = time.time()
|
||
last_id = 0 # paginate forward — избегаем loop когда changes=0
|
||
|
||
# Optional: load cookies из БД для авторизованной сессии
|
||
cookies_from_db: list = []
|
||
try:
|
||
with psycopg.connect(pg_dsn) as conn:
|
||
with conn.cursor() as cur:
|
||
enc_key = os.environ.get("COOKIE_ENCRYPTION_KEY")
|
||
if enc_key:
|
||
cur.execute(
|
||
"""
|
||
SELECT pgp_sym_decrypt(cookies_encrypted, %s)::text
|
||
FROM cian_session_cookies
|
||
WHERE expires_at_estimate > NOW() AND last_invalid_at IS NULL
|
||
ORDER BY uploaded_at DESC LIMIT 1
|
||
""",
|
||
(enc_key,),
|
||
)
|
||
r = cur.fetchone()
|
||
if r:
|
||
jar = json.loads(r[0])
|
||
for name, val in jar.items():
|
||
cookies_from_db.append({
|
||
"name": name, "value": val,
|
||
"domain": ".cian.ru", "path": "/",
|
||
"secure": True, "sameSite": "Lax",
|
||
})
|
||
print(f"[init] loaded {len(cookies_from_db)} cookies from DB")
|
||
except Exception as exc:
|
||
print(f"[init] cookies load skipped: {exc}")
|
||
|
||
async with async_playwright() as p:
|
||
browser = await p.chromium.launch(headless=not args.headed)
|
||
ctx = await browser.new_context(
|
||
user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
|
||
viewport={"width": 1366, "height": 900},
|
||
locale="ru-RU",
|
||
)
|
||
if cookies_from_db:
|
||
await ctx.add_cookies(cookies_from_db)
|
||
page = await ctx.new_page()
|
||
page.set_default_navigation_timeout(45_000)
|
||
page.set_default_timeout(45_000)
|
||
|
||
processed = 0
|
||
while not interrupted and processed < args.max:
|
||
# 1. Fetch batch — paginate id > last_id чтобы не зацикливаться
|
||
with psycopg.connect(pg_dsn) as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
SELECT l.id, l.source_url
|
||
FROM listings l
|
||
LEFT JOIN offer_price_history oph ON oph.listing_id = l.id
|
||
WHERE l.source = 'cian'
|
||
AND l.source_url IS NOT NULL
|
||
AND oph.listing_id IS NULL
|
||
AND l.id > %s
|
||
ORDER BY l.id
|
||
LIMIT %s
|
||
""",
|
||
(last_id, args.batch),
|
||
)
|
||
rows = cur.fetchall()
|
||
if not rows:
|
||
print(f"\n[done] drained — no more listings")
|
||
break
|
||
print(f"\n[batch] {len(rows)} listings to process (cum: ok={cum_ok} fail={cum_fail} changes={cum_changes})")
|
||
|
||
# 2. Process each
|
||
with psycopg.connect(pg_dsn, autocommit=False) as conn:
|
||
for (lid, url) in rows:
|
||
if interrupted:
|
||
break
|
||
processed += 1
|
||
t0 = time.time()
|
||
try:
|
||
await page.goto(url, wait_until="domcontentloaded")
|
||
# wait for _cianConfig populated
|
||
await page.wait_for_function(
|
||
"() => window._cianConfig && window._cianConfig['frontend-offer-card'] && window._cianConfig['frontend-offer-card'].some(e => e.key === 'defaultState')",
|
||
timeout=15_000,
|
||
)
|
||
data = await page.evaluate(EXTRACT_JS)
|
||
if data.get("error"):
|
||
if data.get("captcha"):
|
||
print(f" [{lid}] CAPTCHA — please solve in browser (use --headed)")
|
||
interrupted = True
|
||
break
|
||
cum_fail += 1
|
||
print(f" [{lid}] {data['error']} {time.time() - t0:.1f}s")
|
||
continue
|
||
changes = data.get("price_changes") or []
|
||
with conn.cursor() as cur:
|
||
inserted = upsert_changes(cur, lid, data.get("price_rub"), changes)
|
||
conn.commit()
|
||
cum_ok += 1
|
||
cum_changes += inserted
|
||
if len(changes) > 0:
|
||
cum_with_history += 1
|
||
print(f" [{lid}] OK cian_id={data.get('cian_id')} price={data.get('price_rub')} changes={len(changes)} inserted={inserted} {time.time() - t0:.1f}s")
|
||
# На первом обращении и затем каждые 50 — dump для diagnostics
|
||
if processed == 1 or processed % 50 == 0:
|
||
print(f" [diag] offerData_keys={data.get('_offerData_keys')}")
|
||
print(f" [diag] offer_keys={data.get('_offer_keys')}")
|
||
# Dump shape priceChanges первые 3 раза + при первом успехе
|
||
if (changes and processed <= 3) or (changes and inserted > 0 and cum_changes == inserted):
|
||
print(f" [diag] price_changes_sample={changes[0]}")
|
||
last_id = lid # advance pagination cursor even on 0-change result
|
||
except Exception as exc:
|
||
conn.rollback()
|
||
cum_fail += 1
|
||
print(f" [{lid}] ERR {type(exc).__name__}: {str(exc)[:120]} {time.time() - t0:.1f}s")
|
||
await asyncio.sleep(args.delay)
|
||
|
||
await browser.close()
|
||
|
||
print()
|
||
print(f"== Summary ==")
|
||
print(f" processed_ok={cum_ok} fail={cum_fail}")
|
||
print(f" listings_with_history={cum_with_history} price_changes_inserted={cum_changes}")
|
||
print(f" elapsed={time.time() - start:.1f}s finished={datetime.now(timezone.utc).isoformat(timespec='seconds')}")
|
||
|
||
asyncio.run(run())
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|