lekss361/gendesign

Fork 0

feat(scrape-kn): parallel Phase B/C + wire obj_checks + documents + is_ekb auto-derive #322

Merged

lekss361 merged 1 commit from feat/domrf-speedup-wire-extras into main

2026-05-17 19:40:17 +00:00

lekss361 commented

2026-05-17 19:36:37 +00:00

Owner

4-в-1 PR для `domrf_kn` scraper

После активации Objective live API (PR #311/313/315/318/320/321) этот PR закрывает 4 dangling задачи DOM.РФ scraper'а:

1. ⚡ Speedup Phase B/C (~6× быстрее)

backend/app/services/scrapers/domrf_kn.py:1669-1787

Было (run #14 stats: 12 526s / 1 516 obj = 3.5 ч, 8.3 sec/obj):

for i in range(start_index, total):
    obj_id = ...
    flats = await fetch_flats_for_object(sess, obj_id)        # seq req 1
    for type_ in SALE_GRAPH_TYPES:
        rows = await fetch_sale_graph(sess, obj_id, type_)    # seq req 2,3
    agg   = await fetch_sales_agg(sess, obj_id)               # seq req 4
    pois  = await fetch_infrastructure(sess, obj_id)          # seq req 5
    photos = await fetch_photos(sess, obj_id)                 # seq req 6

Стало — asyncio.gather всех 6 (теперь 8) endpoints одного obj_id параллельно через _fetch_*_safe wrappers. BrowserSession._sem(3) уже bounds concurrency. Ожидаемый speedup ~6×: 3.5 ч → ~35 мин для 1 532 объектов.

2. 📄 Wire `documents` extractor

PR #305 создал таблицу domrf_kn_documents + parser services/scrapers/documents.py (готовые extract_documents() + upsert_documents()), но scraper main loop их не импортировал. Таблица 0 rows.

Этот PR:

Импортирует extract_documents + upsert_documents в domrf_kn.py
Добавляет _fetch_documents_safe(sess, obj_id) wrapper
Включает в parallel asyncio.gather
PATH_DOCUMENTS = "/сервисы/api/object/{obj_id}/documents" (подтверждено per vault Fix_BUG22i)
Добавляет documents_rows в extras_counts + progress log

После следующего sweep — заполнятся PDF docs (декларации, разрешения, проектные).

3. ✅ Create + wire `obj_checks` extractor

PR #303 создал таблицу domrf_obj_checks (6 чекбоксов "Проверено на наш.дом.рф"), но extractor не существовал. Таблица 0 rows.

Этот PR создаёт backend/app/services/scrapers/obj_checks.py (~160 строк):

CHECK_TYPES = ["no_problems", "docs", "timing", "photos", "bankruptcy", "declaration"]
extract_obj_checks(payload) — поддерживает dict-format и list-format payloads
upsert_obj_checks(db, obj_id, checks) — SAVEPOINT per-row
Wires в domrf_kn.py parallel gather

⚠️ PATH_CHECKS = "/сервисы/api/object/{obj_id}/checks" — endpoint URL не verified через devtools, derived from same /сервисы/api/object/ namespace что у /infrastructure, /documents, /photos. Если после next sweep всё ещё 0 rows + 404 в kn_scrape_failures для всех objects → нужно devtools-inspect страницу ЖК и update constant. На failure — sweep продолжается, не падает.

4. 🇪🇰🇧 Phase E — `is_ekb` auto-derive

После Phase D finalize (~line 1652+) добавлена Phase E:

UPDATE domrf_kn_objects
   SET is_ekb = TRUE
 WHERE snapshot_date = :snap
   AND (district_name ILIKE '%екатеринбург%' OR district_name ILIKE '%свердловск%')
   AND COALESCE(is_ekb, FALSE) = FALSE

Best-effort: exception caught + logged, sweep всё равно завершается. Раньше требовался manual run после scrape.

Operations note

Run #18 (started 19:20Z, ETA 23:00Z) будет убит при deploy этого PR — --force-recreate worker пересоздаёт контейнер, in-flight task dies. После deploy — manual re-trigger /admin/scrape/kn сразу даст ~35-мин sweep вместо 3.5-час.

User explicitly authorized kill #18 + retrigger с оптимизированной версией.

Acceptance после merge + deploy

domrf_kn_documents count > 0 после первого sweep (or 404 в kn_scrape_failures если documents endpoint missing)
domrf_obj_checks count > 0 (or 404 → нужен fix endpoint URL)
Phase E log message появляется в kn_scrape_log после Phase D
is_ekb=TRUE count в свежем snapshot ~1300 (вся ЕКБ)
Sweep duration ~35-45 мин (vs 3.5 ч до этого PR)
Coverage obj_class / wall_type / energy_eff / flat_number ↑ (применение PR #302 mapping fix)

Tech debt outstanding (не в этом PR)

DOM.РФ price coverage 0.3% — потребует catalog SSR scraper (PR #297 22d) ИЛИ migration на objective_lots.price_per_m2_rub (81%)
Photos binary download — photos=False default, тяжёлая операция (~1+ GB на регион). Опционально вкл через job_settings.

Risk

🟡 obj_checks endpoint placeholder — если 404 → таблица останется 0 rows, но sweep не сломается (graceful logging)
🟢 documents endpoint verified vault entry
🟢 Phase E — простой UPDATE, idempotent (COALESCE guard)
🟢 asyncio.gather — BrowserSession уже имеет _sem(3) semaphore

После deploy и первого sweep — verify endpoint via vault + adjust if needed.

## 4-в-1 PR для `domrf_kn` scraper После активации Objective live API (PR #311/313/315/318/320/321) этот PR закрывает 4 dangling задачи DOM.РФ scraper'а: ### 1. ⚡ Speedup Phase B/C (~6× быстрее) `backend/app/services/scrapers/domrf_kn.py:1669-1787` **Было** (run #14 stats: 12 526s / 1 516 obj = 3.5 ч, 8.3 sec/obj): ```python for i in range(start_index, total): obj_id = ... flats = await fetch_flats_for_object(sess, obj_id) # seq req 1 for type_ in SALE_GRAPH_TYPES: rows = await fetch_sale_graph(sess, obj_id, type_) # seq req 2,3 agg = await fetch_sales_agg(sess, obj_id) # seq req 4 pois = await fetch_infrastructure(sess, obj_id) # seq req 5 photos = await fetch_photos(sess, obj_id) # seq req 6 ``` **Стало** — `asyncio.gather` всех 6 (теперь 8) endpoints одного `obj_id` параллельно через `_fetch_*_safe` wrappers. `BrowserSession._sem(3)` уже bounds concurrency. Ожидаемый speedup ~6×: **3.5 ч → ~35 мин** для 1 532 объектов. ### 2. 📄 Wire `documents` extractor PR #305 создал таблицу `domrf_kn_documents` + parser `services/scrapers/documents.py` (готовые `extract_documents()` + `upsert_documents()`), но scraper main loop их **не импортировал**. Таблица 0 rows. Этот PR: - Импортирует `extract_documents` + `upsert_documents` в `domrf_kn.py` - Добавляет `_fetch_documents_safe(sess, obj_id)` wrapper - Включает в parallel `asyncio.gather` - `PATH_DOCUMENTS = "/сервисы/api/object/{obj_id}/documents"` (подтверждено per vault `Fix_BUG22i`) - Добавляет `documents_rows` в `extras_counts` + progress log После следующего sweep — заполнятся PDF docs (декларации, разрешения, проектные). ### 3. ✅ Create + wire `obj_checks` extractor PR #303 создал таблицу `domrf_obj_checks` (6 чекбоксов "Проверено на наш.дом.рф"), но extractor **не существовал**. Таблица 0 rows. Этот PR создаёт `backend/app/services/scrapers/obj_checks.py` (~160 строк): - `CHECK_TYPES = ["no_problems", "docs", "timing", "photos", "bankruptcy", "declaration"]` - `extract_obj_checks(payload)` — поддерживает dict-format и list-format payloads - `upsert_obj_checks(db, obj_id, checks)` — SAVEPOINT per-row - Wires в `domrf_kn.py` parallel gather ⚠️ **`PATH_CHECKS = "/сервисы/api/object/{obj_id}/checks"`** — endpoint URL **не verified через devtools**, derived from same `/сервисы/api/object/` namespace что у `/infrastructure`, `/documents`, `/photos`. Если после next sweep всё ещё 0 rows + 404 в `kn_scrape_failures` для всех objects → нужно devtools-inspect страницу ЖК и update constant. На failure — sweep продолжается, не падает. ### 4. 🇪🇰🇧 Phase E — `is_ekb` auto-derive После Phase D finalize (~line 1652+) добавлена Phase E: ```python UPDATE domrf_kn_objects SET is_ekb = TRUE WHERE snapshot_date = :snap AND (district_name ILIKE '%екатеринбург%' OR district_name ILIKE '%свердловск%') AND COALESCE(is_ekb, FALSE) = FALSE ``` Best-effort: exception caught + logged, sweep всё равно завершается. Раньше требовался manual run после scrape. ## Operations note **Run #18 (started 19:20Z, ETA 23:00Z) будет убит при deploy этого PR** — `--force-recreate worker` пересоздаёт контейнер, in-flight task dies. После deploy — manual re-trigger `/admin/scrape/kn` сразу даст ~35-мин sweep вместо 3.5-час. User explicitly authorized kill #18 + retrigger с оптимизированной версией. ## Acceptance после merge + deploy - [ ] `domrf_kn_documents` count > 0 после первого sweep (or 404 в `kn_scrape_failures` если documents endpoint missing) - [ ] `domrf_obj_checks` count > 0 (or 404 → нужен fix endpoint URL) - [ ] Phase E log message появляется в `kn_scrape_log` после Phase D - [ ] `is_ekb=TRUE` count в свежем snapshot ~1300 (вся ЕКБ) - [ ] Sweep duration ~35-45 мин (vs 3.5 ч до этого PR) - [ ] Coverage `obj_class` / `wall_type` / `energy_eff` / `flat_number` ↑ (применение PR #302 mapping fix) ## Tech debt outstanding (не в этом PR) - DOM.РФ price coverage 0.3% — потребует catalog SSR scraper (PR #297 22d) ИЛИ migration на `objective_lots.price_per_m2_rub` (81%) - Photos binary download — `photos=False` default, тяжёлая операция (~1+ GB на регион). Опционально вкл через job_settings. ## Risk - 🟡 obj_checks endpoint placeholder — если 404 → таблица останется 0 rows, но sweep не сломается (graceful logging) - 🟢 documents endpoint verified vault entry - 🟢 Phase E — простой UPDATE, idempotent (COALESCE guard) - 🟢 asyncio.gather — BrowserSession уже имеет `_sem(3)` semaphore После deploy и первого sweep — verify endpoint via vault + adjust if needed.

lekss361 added 1 commit 2026-05-17 19:36:38 +00:00

feat(scrape-kn): parallel Phase B/C + wire obj_checks + documents + is_ekb auto-derive 1db19b3b8a

4 changes:
1. asyncio.gather() в Phase B/C — 6 endpoints/obj параллельно
   через _fetch_*_safe wrappers. ~6x speedup (3.5h -> ~35min).
   BrowserSession._sem(3) bounds concurrent requests автоматически.
2. Wire documents extractor (services/scrapers/documents.py) —
   per-object call в parallel gather. Заполняет domrf_kn_documents
   (был пуст после PR #305 schema-only).
3. Create services/scrapers/obj_checks.py extractor + wire в loop.
   Заполняет domrf_obj_checks (был пуст после PR #303 schema-only).
   Endpoint URL /checks не верифицирован devtools — выведен по
   паттерну /infrastructure; при ошибке пишет в kn_scrape_failures.
4. Phase E (после Phase D) — UPDATE is_ekb=TRUE для свежего snapshot
   по district_name ILIKE екатеринбург/свердловск. Best-effort
   (exception caught, scrape не прерывается).

lekss361 commented

2026-05-17 19:40:08 +00:00

Author

Owner

Deep Code Review — verdict: APPROVE

Summary

Status: APPROVE
Files reviewed: 2 (P1: domrf_kn.py, obj_checks.py new)
Lines: +408 / -55
Head SHA: 1db19b3

Correctness — parallel Phase B/C

All seven _fetch_*_safe wrappers catch Exception and return it in the result tuple — no exception bubbles up to abort asyncio.gather. return_exceptions=False is therefore safe.
BrowserSession._sem = asyncio.Semaphore(3) (stealth.py:107) is acquired inside get_json() per-request (stealth.py:152), not at the outer fetcher level. Every parallel coroutine routes through get_json → bounded concurrency confirmed.
DB session usage is sequential after gather completes (domrf_kn.py:1698+) — single Session, no parallel writes. Race-free.
No shared mutable state across the gathered coroutines — each takes (sess, obj_id [, type_]) only; accumulation into extras_counts happens in the main coroutine post-gather.
_classify_and_log(db, run_id, obj_id, kind_tag, full_url, e) signature matches existing usage (domrf_kn.py:888).

Correctness — `obj_checks.py`

SAVEPOINT-per-row pattern via with db.begin_nested(): — compliant with .claude/rules/backend.md.
Parameterised SQL (:obj_id, :check_type, :passed), no :x::type casts, no f-string injection.
extract_obj_checks defensively handles both dict and list payloads with sensible warning + empty return on unknown shape — won't crash sweep.
ON CONFLICT (obj_id, check_type) DO UPDATE is idempotent.
Module-level db.commit() after the loop is consistent with documents.upsert_documents.

Correctness — documents wiring

Import + _fetch_documents_safe + result dispatch correctly call upsert_documents(db, obj_id, docs) -> (int, int) and unpack only ins.
Accumulator extras_counts["documents_rows"] initialised at line 1585, incremented at result dispatch — clean.

Correctness — Phase E `is_ekb`

:snap placeholder, no ::date cast (already date type) — psycopg v3 clean.
COALESCE(is_ekb, FALSE) = FALSE guard makes it idempotent across reruns.
try/except + db.rollback() + log_progress(level="warn") — sweep continues on failure.
Commits before logging the result, but that is the established pattern in this file.

Conventions

No psycopg2, no requests, no print(), no hardcoded creds, no f-string SQL.
Line length under 100. Type hints present on every public symbol.

Cross-file impact

extras_counts dict gets two new keys (documents_rows, checks_rows) — local to run_region_sweep, returned in result dict additively. No consumer breakage.
No removed callers / no renamed symbols. Backward compat preserved.

Minor observations (non-blocking)

_fetch_sale_graph_safe etc. return tuples like (rows, url) even though the dispatcher only needs rows — url is captured both in the wrapper tuple and the inner result tuple. Cosmetic only.
PATH_CHECKS = "/сервисы/api/object/{obj_id}/checks" is unverified per author note. Acceptable trade-off: graceful 404 logged into kn_scrape_failures via _classify_and_log, sweep continues. If endpoint is wrong, first sweep will produce 1500+ WARN entries in kn_scrape_failures — verify post-deploy via SQL and adjust the constant if needed.

Risk

Risk: Low for parallelism (semaphore proven, exceptions contained, sequential DB).
Reversibility: Easy — pure code change, no migration.
Recommended merge window: anytime; just retrigger /admin/scrape/kn after deploy as author noted.

Merging.

## Deep Code Review — verdict: APPROVE ### Summary - Status: APPROVE - Files reviewed: 2 (P1: `domrf_kn.py`, `obj_checks.py` new) - Lines: +408 / -55 - Head SHA: `1db19b3` ### Correctness — parallel Phase B/C - All seven `_fetch_*_safe` wrappers catch `Exception` and return it in the result tuple — no exception bubbles up to abort `asyncio.gather`. `return_exceptions=False` is therefore safe. - `BrowserSession._sem = asyncio.Semaphore(3)` (`stealth.py:107`) is acquired inside `get_json()` per-request (`stealth.py:152`), not at the outer fetcher level. Every parallel coroutine routes through `get_json` → bounded concurrency confirmed. - DB session usage is sequential after `gather` completes (`domrf_kn.py:1698`+) — single Session, no parallel writes. Race-free. - No shared mutable state across the gathered coroutines — each takes `(sess, obj_id [, type_])` only; accumulation into `extras_counts` happens in the main coroutine post-gather. - `_classify_and_log(db, run_id, obj_id, kind_tag, full_url, e)` signature matches existing usage (`domrf_kn.py:888`). ### Correctness — `obj_checks.py` - SAVEPOINT-per-row pattern via `with db.begin_nested():` — compliant with `.claude/rules/backend.md`. - Parameterised SQL (`:obj_id`, `:check_type`, `:passed`), no `:x::type` casts, no f-string injection. - `extract_obj_checks` defensively handles both dict and list payloads with sensible warning + empty return on unknown shape — won't crash sweep. - `ON CONFLICT (obj_id, check_type) DO UPDATE` is idempotent. - Module-level `db.commit()` after the loop is consistent with `documents.upsert_documents`. ### Correctness — documents wiring - Import + `_fetch_documents_safe` + result dispatch correctly call `upsert_documents(db, obj_id, docs) -> (int, int)` and unpack only `ins`. - Accumulator `extras_counts["documents_rows"]` initialised at line 1585, incremented at result dispatch — clean. ### Correctness — Phase E `is_ekb` - `:snap` placeholder, no `::date` cast (already date type) — psycopg v3 clean. - `COALESCE(is_ekb, FALSE) = FALSE` guard makes it idempotent across reruns. - `try/except + db.rollback() + log_progress(level="warn")` — sweep continues on failure. - Commits before logging the result, but that is the established pattern in this file. ### Conventions - No `psycopg2`, no `requests`, no `print()`, no hardcoded creds, no f-string SQL. - Line length under 100. Type hints present on every public symbol. ### Cross-file impact - `extras_counts` dict gets two new keys (`documents_rows`, `checks_rows`) — local to `run_region_sweep`, returned in result dict additively. No consumer breakage. - No removed callers / no renamed symbols. Backward compat preserved. ### Minor observations (non-blocking) - `_fetch_sale_graph_safe` etc. return tuples like `(rows, url)` even though the dispatcher only needs `rows` — `url` is captured both in the wrapper tuple and the inner result tuple. Cosmetic only. - `PATH_CHECKS = "/сервисы/api/object/{obj_id}/checks"` is unverified per author note. Acceptable trade-off: graceful 404 logged into `kn_scrape_failures` via `_classify_and_log`, sweep continues. If endpoint is wrong, first sweep will produce 1500+ WARN entries in `kn_scrape_failures` — verify post-deploy via SQL and adjust the constant if needed. ### Risk - Risk: **Low** for parallelism (semaphore proven, exceptions contained, sequential DB). - Reversibility: **Easy** — pure code change, no migration. - Recommended merge window: anytime; just retrigger `/admin/scrape/kn` after deploy as author noted. Merging.