feat(tradein): scrape_pipeline.py — Avito full orchestrator (search → houses → detail) #451

Merged

lekss361 merged 1 commit from feat/tradein-scrape-pipeline into main

2026-05-23 12:55:14 +00:00

lekss361 commented

2026-05-23 12:49:50 +00:00

Owner

Summary

Stage 2e of Avito Scraper v2 — orchestrator модуль координирующий полный pipeline: search (Stage 2a) → group_by_house → enrich houses (Stage 2c) → enrich top-N details (Stage 2b). IMV (Stage 2d) НЕ здесь — он on-demand из estimator (Stage 3).

LOC: scrape_pipeline.py 229 lines, test_scrape_pipeline.py 86 lines (3 tests).

Changes

NEW tradein-mvp/backend/app/services/scrape_pipeline.py
- run_avito_pipeline(db, lat, lon, radius_m, *, enrich_houses, enrich_detail_top_n) — 5-step async flow:
  1. SEARCH: AvitoScraper().fetch_around (uses Stage 2a refactored)
  2. SAVE: save_listings (с house_ext_id linking из Stage 2a)
  3. GROUP: dedupe house_ext_id → unique house URLs
  4. ENRICH_HOUSES: fetch_house_catalog + save_house_catalog_enrichment per unique house (Stage 2c)
  5. ENRICH_DETAIL: top-N priority listings (detail_enriched_at IS NULL в радиусе) → fetch_detail + save_detail_enrichment (Stage 2b)
- PipelineCounters dataclass: lots/houses/detail counters + errors[]
- PipelineResult dataclass
- Graceful degradation: per-house + per-listing try/except — single failure не валит pipeline
NEW tradein-mvp/backend/tests/test_scrape_pipeline.py (3 tests, 53/53 passed суммарно с existing)

Smoke

✅ ruff clean
✅ pytest: 53/53 passed (3 новых + 50 existing, no regressions)
⏸ Network smoke deferred to deploy

Bug fix in spec

Worker поймал SQL bug в моём inline spec: WHERE ... AND condition1 OR condition2 без скобок → AND приоритет выше OR → condition2 применялся бы ко всем строкам без source='avito' filter. Обёрнут в (...) OR (...). Хорошая catch.

Tooling adds

pytest-asyncio>=0.24.0 добавлен в pyproject.toml (отсутствовал в tradein-mvp) + asyncio_mode = "auto"

Test plan

ruff
pytest (3 new tests covering counters/search-failure/group-by-house)
Deploy → trigger pipeline manually на ЕКБ anchor → verify counters в logs (lots/houses/detail)
Future: wire cron-scrape.sh или новый admin endpoint вызывает run_avito_pipeline вместо raw /admin/scrape (Stage 4a)

Behavior

Step	Action	Failure mode
Search	fetch_around 50 lots	Critical — return empty result
Save	upsert listings	Per-batch error logged, continue
Group	dedupe house_ext_id from lots	Pure logic, no failure
Enrich houses	per-house fetch+save	Per-house try/except, counter++ failed
Enrich detail	top-N priority listings	Per-listing try/except, counter++ failed

## Summary **Stage 2e of Avito Scraper v2** — orchestrator модуль координирующий полный pipeline: search (Stage 2a) → group_by_house → enrich houses (Stage 2c) → enrich top-N details (Stage 2b). IMV (Stage 2d) НЕ здесь — он on-demand из estimator (Stage 3). LOC: `scrape_pipeline.py` 229 lines, `test_scrape_pipeline.py` 86 lines (3 tests). ## Changes - **NEW** `tradein-mvp/backend/app/services/scrape_pipeline.py` - `run_avito_pipeline(db, lat, lon, radius_m, *, enrich_houses, enrich_detail_top_n)` — 5-step async flow: 1. SEARCH: `AvitoScraper().fetch_around` (uses Stage 2a refactored) 2. SAVE: `save_listings` (с `house_ext_id` linking из Stage 2a) 3. GROUP: dedupe `house_ext_id` → unique house URLs 4. ENRICH_HOUSES: `fetch_house_catalog` + `save_house_catalog_enrichment` per unique house (Stage 2c) 5. ENRICH_DETAIL: top-N priority listings (`detail_enriched_at IS NULL` в радиусе) → `fetch_detail` + `save_detail_enrichment` (Stage 2b) - `PipelineCounters` dataclass: lots/houses/detail counters + errors[] - `PipelineResult` dataclass - Graceful degradation: per-house + per-listing try/except — single failure не валит pipeline - **NEW** `tradein-mvp/backend/tests/test_scrape_pipeline.py` (3 tests, 53/53 passed суммарно с existing) ## Smoke - ✅ ruff clean - ✅ pytest: 53/53 passed (3 новых + 50 existing, no regressions) - ⏸ Network smoke deferred to deploy ## Bug fix in spec Worker поймал SQL bug в моём inline spec: `WHERE ... AND condition1 OR condition2` без скобок → AND приоритет выше OR → `condition2` применялся бы ко всем строкам без `source='avito'` filter. Обёрнут в `(...) OR (...)`. Хорошая catch. ## Tooling adds - `pytest-asyncio>=0.24.0` добавлен в `pyproject.toml` (отсутствовал в tradein-mvp) + `asyncio_mode = "auto"` ## Test plan - [x] ruff - [x] pytest (3 new tests covering counters/search-failure/group-by-house) - [ ] Deploy → trigger pipeline manually на ЕКБ anchor → verify counters в logs (lots/houses/detail) - [ ] Future: wire `cron-scrape.sh` или новый admin endpoint вызывает `run_avito_pipeline` вместо raw `/admin/scrape` (Stage 4a) ## Behavior | Step | Action | Failure mode | |---|---|---| | Search | fetch_around 50 lots | Critical — return empty result | | Save | upsert listings | Per-batch error logged, continue | | Group | dedupe house_ext_id from lots | Pure logic, no failure | | Enrich houses | per-house fetch+save | Per-house try/except, counter++ failed | | Enrich detail | top-N priority listings | Per-listing try/except, counter++ failed |

lekss361 added 1 commit 2026-05-23 12:49:51 +00:00

feat(tradein): scrape_pipeline.py — Avito full orchestrator (search → houses → detail) 0154c2914a

Stage 2e of AvitoScraper_v2.

- run_avito_pipeline(db, lat, lon, radius_m, *, enrich_houses, enrich_detail_top_n)
- 5-step flow:
  1. SEARCH: AvitoScraper().fetch_around (uses Stage 2a refactored search)
  2. SAVE: save_listings (inserts + tracks house_ext_id from Stage 2a)
  3. GROUP: dedupe house path → unique set (path-only для consistency с fetch_house_catalog)
  4. ENRICH_HOUSES: fetch_house_catalog + save_house_catalog_enrichment (Stage 2c) per unique house
  5. ENRICH_DETAIL: top-N priority listings (detail_enriched_at IS NULL within radius)
     → fetch_detail + save_detail_enrichment (Stage 2b)
- PipelineCounters dataclass: lots_fetched/inserted/updated, unique_houses, houses_enriched/failed,
  detail_attempted/enriched/failed, errors[]
- Graceful degradation: per-house and per-detail try/except — single failure не валит pipeline
- IMV ОТСУТСТВУЕТ — он on-demand (Stage 3 estimator integration), не cron
- pytest offline smoke (3 tests: counters default + search failure graceful + group_by_house dedupe)
- Add pytest-asyncio>=0.24.0 to dev deps + asyncio_mode=auto в pyproject.toml

Refs: AvitoScraper_v2_Implementation_Plan Stage 2e.

lekss361 merged commit eb4764f75a into main

2026-05-23 12:55:14 +00:00

lekss361 referenced this pull request from a commit

2026-05-23 12:55:16 +00:00

feat(tradein): scrape_pipeline.py — Avito full orchestrator (search → houses → detail) (#451)

lekss361 commented

2026-05-23 12:55:49 +00:00

Author

Owner

Merged via deep-code-reviewer — verdict APPROVE with MINOR.

What was verified

SQL synthesis (step 5 priority query) — parens correct: outer AND source='avito' AND source_url IS NOT NULL AND ( (radius+price branch) OR (recent branch) ). Neither OR branch can leak past the source filter. All columns (source, source_url, detail_enriched_at, price_rub, geom, scraped_at) exist; partial index listings_detail_enrich_idx + GIST listings_geom_idx support the query well.
Transaction strategy — per-step commits inside save_listings / save_house_catalog_enrichment / save_detail_enrichment. Acceptable for orchestrator scope; enables graceful degradation; no long-held transaction across minutes-long enrichment.
Cross-file signatures — fetch_house_catalog, save_house_catalog_enrichment, fetch_detail, save_detail_enrichment, save_listings, AvitoScraper.fetch_around all match call sites.
No conflicts with PR #450 (cian) or PR #452 (IMV integration).

Follow-up nits (non-blocking, do in next PR)

Add db.rollback() inside except Exception for both per-house loop (step 4) and per-listing loop (step 5) — guards against PendingRollbackError cascade if the underlying failure was a DBAPIError/IntegrityError rather than HTTP/parse.
Bound counters.errors[] to last ~20 entries + add error_total int counter — prevents large payloads when enrich_detail_top_n is raised for bigger anchors in Stage 4a wire-up.
Step 5 recent-OR branch (scraped_at > NOW() - INTERVAL '2 hours') has no radius bound — if cron multiplexes anchors in parallel later, add AND ST_DWithin(...) to that branch too.
save_detail_enrichment returns False when listing row not found in DB — currently silently skipped in counters. Consider counters.detail_skipped separate from detail_failed for observability.
Rate limiting: fetch_house_catalog opens its own CffiAsyncSession and bypasses AvitoScraper's request_delay_sec=7.0. For 30 unique houses in a single run that's a burst of 30 reqs → 429 risk. Suggest threading AvitoScraper's session through, or adding await asyncio.sleep(7) between houses.
Test gaps for follow-up: per-house failure isolation (1 of 3 raises, others succeed), per-listing failure isolation, integration test verifying DB call count == enrich_detail_top_n.

Post-merge actions

deploy-tradein.yml should trigger on main.
Module is inert until Stage 4a wires cron-scrape.sh or an admin endpoint to run_avito_pipeline — no behavior change in production yet.
Vault entry code/modules/tradein/Scrape_Pipeline_Orchestrator.md to be created by worker.

Merged via deep-code-reviewer — verdict APPROVE with MINOR. ### What was verified - **SQL synthesis (step 5 priority query)** — parens correct: outer `AND source='avito' AND source_url IS NOT NULL AND ( (radius+price branch) OR (recent branch) )`. Neither OR branch can leak past the source filter. All columns (`source`, `source_url`, `detail_enriched_at`, `price_rub`, `geom`, `scraped_at`) exist; partial index `listings_detail_enrich_idx` + GIST `listings_geom_idx` support the query well. - **Transaction strategy** — per-step commits inside `save_listings` / `save_house_catalog_enrichment` / `save_detail_enrichment`. Acceptable for orchestrator scope; enables graceful degradation; no long-held transaction across minutes-long enrichment. - **Cross-file signatures** — `fetch_house_catalog`, `save_house_catalog_enrichment`, `fetch_detail`, `save_detail_enrichment`, `save_listings`, `AvitoScraper.fetch_around` all match call sites. - **No conflicts** with PR #450 (cian) or PR #452 (IMV integration). ### Follow-up nits (non-blocking, do in next PR) 1. Add `db.rollback()` inside `except Exception` for both per-house loop (step 4) and per-listing loop (step 5) — guards against `PendingRollbackError` cascade if the underlying failure was a `DBAPIError`/`IntegrityError` rather than HTTP/parse. 2. Bound `counters.errors[]` to last ~20 entries + add `error_total` int counter — prevents large payloads when `enrich_detail_top_n` is raised for bigger anchors in Stage 4a wire-up. 3. Step 5 recent-OR branch (`scraped_at > NOW() - INTERVAL '2 hours'`) has no radius bound — if cron multiplexes anchors in parallel later, add `AND ST_DWithin(...)` to that branch too. 4. `save_detail_enrichment` returns `False` when listing row not found in DB — currently silently skipped in counters. Consider `counters.detail_skipped` separate from `detail_failed` for observability. 5. Rate limiting: `fetch_house_catalog` opens its own `CffiAsyncSession` and bypasses `AvitoScraper`'s `request_delay_sec=7.0`. For 30 unique houses in a single run that's a burst of 30 reqs → 429 risk. Suggest threading `AvitoScraper`'s session through, or adding `await asyncio.sleep(7)` between houses. 6. Test gaps for follow-up: per-house failure isolation (1 of 3 raises, others succeed), per-listing failure isolation, integration test verifying DB call count == `enrich_detail_top_n`. ### Post-merge actions - deploy-tradein.yml should trigger on main. - Module is inert until Stage 4a wires `cron-scrape.sh` or an admin endpoint to `run_avito_pipeline` — no behavior change in production yet. - Vault entry `code/modules/tradein/Scrape_Pipeline_Orchestrator.md` to be created by worker.