lekss361/gendesign

Fork 0

feat(nspd): harvest_quarter Celery task + beat + admin endpoint (#94 pt.4 part A) #111

Merged

lekss361 merged 2 commits from feat/nspd-harvest-quarter-task into main

2026-05-12 15:45:58 +00:00

lekss361 commented

2026-05-12 15:34:06 +00:00

(Migrated from github.com)

Summary

Sprint 1.1 item #3 из #94 part 2. Замыкает harvest cycle:

PR	Что	Status
#109	`QuarterDump` + `NSPDClient.search_by_quarter()`	✅ merged
#110	`nspd_quarter_dumps` table schema	✅ merged
#111 (этот)	Celery harvest_quarter task + beat + admin endpoint	🔄
Next	`analyze_parcel` reads dump cache	⏳

После этого PR data harvesting автоматизирован. После следующего — analyze_parcel перестанет делать NSPD HTTP в request-цикле (читает из nspd_quarter_dumps cache).

Backend changes

`backend/app/workers/tasks/nspd_sync.py` (new)

@celery_app.task(
    bind=True, name="tasks.nspd_sync.harvest_quarter",
    autoretry_for=(NspdLiteWafError,),
    retry_backoff=True, retry_backoff_max=120, max_retries=3,
    soft_time_limit=120,
)
def harvest_quarter(self, quarter_cad, region_code=66, include_zouit=True, include_risks=False):
    """Single-quarter harvest: search_by_quarter → UPSERT в nspd_quarter_dumps."""

Error handling:

NspdLiteWafError (403/429) → autoretry с exponential backoff. Maximum 3 retries.
Other exceptions → upsert error row с harvest_error=str(exc), counts=0. НЕ re-raise (избегаем silent task failure).
Worst case: после max_retries failed → task fails, no row written. Future beat run подхватит снова.

@celery_app.task(name="tasks.nspd_sync.harvest_stale_quarters")
def harvest_stale_quarters(region_code=66, max_age_days=90, batch_size=50):
    """Beat fanout: find stale quarters → apply_async per cad. Returns {enqueued: N}."""

SQL для поиска stale:

SELECT q.cad_number FROM cad_quarters_geom q
 WHERE q.region_code = :rc
   AND NOT EXISTS (
       SELECT 1 FROM nspd_quarter_dumps d
        WHERE d.quarter_cad = q.cad_number
          AND d.region_code = :rc
          AND d.fetched_at_utc > NOW() - INTERVAL '1 day' * :days
          AND d.harvest_error IS NULL
   )
 LIMIT :batch

`backend/app/workers/celery_app.py`

Add nspd_sync в include
Beat entry nspd-harvest-stale-quarters: Mon 04:00 МСК (crontab(hour=4, minute=0, day_of_week="mon")), batch_size=50, max_age_days=90
worker_ready sanity check SELECT 1 FROM nspd_quarter_dumps LIMIT 0 (non-fatal critical log per Bug_Worker_Ready_EarlyReturn_Fixed)

`backend/app/api/v1/admin_scrape.py`

class HarvestQuarterRequest(BaseModel):
    quarter_cad: str = Field(min_length=5, pattern=r"^\d+:\d+:\d+$")
    region_code: int = Field(default=66)
    include_zouit: bool = Field(default=True)
    include_risks: bool = Field(default=False)


@router.post("/nspd/harvest-quarter")
def trigger_harvest_quarter(...) -> dict:
    """Manual trigger. Requires X-Admin-Token. Returns task_id."""

Pattern ^\d+:\d+:\d+$ rejects 4-сегментный cad (only quarters allowed для этого task).

Critical contracts

Contract	Implementation
Schema MultiPolygon (PR #110)	`ST_Multi(ST_Transform(ST_SetSRID(ST_GeomFromGeoJSON(:geom), 3857), 4326))` на ingest
bbox storage	`ST_MakeEnvelope(:xmin, :ymin, :xmax, :ymax, 3857)`
layers_fetched TEXT[]	PostgreSQL array literal `{a,b,c}` через string concat
Empty quarter	Row с `quarter_geom=NULL`, `bbox_3857=NULL`, `features_json='[]'`, counts=0, no error
Failed harvest	Row с `harvest_error=str(exc)`, counts=0 — re-tried в следующий beat

Tests: `backend/tests/test_nspd_sync.py` (14 tests)

Mock-based, no real NSPD/DB:

test_harvest_quarter_happy_path — search_by_quarter mock → verify UPSERT SQL
test_harvest_quarter_empty_quarter — NSPD пуст → row with null geometry
test_harvest_quarter_waf_retry — NspdLiteWafError → re-raise (autoretry)
test_harvest_quarter_generic_error_writes_row — generic exc → upsert error row, no re-raise
test_harvest_stale_quarters_fanout — 50 stale → 50 apply_async calls
test_harvest_stale_quarters_batch_limit — batch_size=10 → 10 calls maximum
test_features_json_zouit_groups — include_zouit=True → all 5 zouit keys present
test_features_json_risks_groups — include_risks=True → all 11 risk keys
test_features_json_layer_assignment — каждый feature имеет правильный layer name
test_upsert_with_geom_uses_st_multi — verify SQL uses ST_Multi(ST_Transform(...))
test_upsert_without_geom_uses_null — empty quarter → CASE WHEN NULL THEN NULL
... (всего 14)

uv run pytest tests/test_nspd_sync.py -v → 14/14 passed.

Code review (pre-push, code-reviewer agent)

Verdict: APPROVE, 0 blocking, 5 minor (all acceptable defer):

inline import datetime in error else-branch (style, technical debt)
layers_fetched array literal via string concat — works for current layer names; risky only if future layer name contains , or {
harvest_stale_quarters re-tries error rows infinitely — intended behavior (manual cleanup or retry-count cap later)
Test call_args.kwargs["args"] fragile if call site changes
Beat collision with refresh-ekb-districts-medians once every ~2 years (negligible)

Acceptance (per #94 plan)

harvest_quarter task implemented с WAF retry
harvest_stale_quarters beat fanout
Mon 04:00 МСК beat schedule
Admin endpoint для manual trigger
Tests: 14/14 pass
Sanity check в worker_ready
Code-reviewer APPROVE

Test plan

ruff check + format — clean
pytest tests/test_nspd_sync.py -v — 14/14 passed

After deploy — smoke:

curl -X POST https://gendsgn.ru/api/v1/admin/scrape/nspd/harvest-quarter \
  -H "X-Admin-Token: $TOKEN" \
  -d '{"quarter_cad": "66:41:0204016"}'
# → {"task_id": "...", "quarter_cad": "66:41:0204016"}

# 10s later
psql -c "SELECT quarter_cad, total_features, harvest_error FROM nspd_quarter_dumps WHERE quarter_cad='66:41:0204016'"
# → total_features > 0, harvest_error IS NULL

Verify Celery Beat подхватил nspd-harvest-stale-quarters в schedule (logs)
После первого Mon 04:00 — verify N rows added в nspd_quarter_dumps

Sequential rule — следующий PR (analyze_parcel integration) только после merge этого.

Part of #94.

## Summary **Sprint 1.1 item #3** из #94 part 2. Замыкает harvest cycle: | PR | Что | Status | |---|---|---| | #109 | `QuarterDump` + `NSPDClient.search_by_quarter()` | ✅ merged | | #110 | `nspd_quarter_dumps` table schema | ✅ merged | | **#111 (этот)** | **Celery harvest_quarter task + beat + admin endpoint** | 🔄 | | Next | `analyze_parcel` reads dump cache | ⏳ | После этого PR data harvesting автоматизирован. После следующего — `analyze_parcel` перестанет делать NSPD HTTP в request-цикле (читает из `nspd_quarter_dumps` cache). ## Backend changes ### `backend/app/workers/tasks/nspd_sync.py` (new) ```python @celery_app.task( bind=True, name="tasks.nspd_sync.harvest_quarter", autoretry_for=(NspdLiteWafError,), retry_backoff=True, retry_backoff_max=120, max_retries=3, soft_time_limit=120, ) def harvest_quarter(self, quarter_cad, region_code=66, include_zouit=True, include_risks=False): """Single-quarter harvest: search_by_quarter → UPSERT в nspd_quarter_dumps.""" ``` **Error handling:** - `NspdLiteWafError` (403/429) → autoretry с exponential backoff. Maximum 3 retries. - Other exceptions → upsert error row с `harvest_error=str(exc)`, counts=0. НЕ re-raise (избегаем silent task failure). - Worst case: после max_retries failed → task fails, no row written. Future beat run подхватит снова. ```python @celery_app.task(name="tasks.nspd_sync.harvest_stale_quarters") def harvest_stale_quarters(region_code=66, max_age_days=90, batch_size=50): """Beat fanout: find stale quarters → apply_async per cad. Returns {enqueued: N}.""" ``` **SQL для поиска stale:** ```sql SELECT q.cad_number FROM cad_quarters_geom q WHERE q.region_code = :rc AND NOT EXISTS ( SELECT 1 FROM nspd_quarter_dumps d WHERE d.quarter_cad = q.cad_number AND d.region_code = :rc AND d.fetched_at_utc > NOW() - INTERVAL '1 day' * :days AND d.harvest_error IS NULL ) LIMIT :batch ``` ### `backend/app/workers/celery_app.py` - Add `nspd_sync` в `include` - Beat entry `nspd-harvest-stale-quarters`: Mon 04:00 МСК (`crontab(hour=4, minute=0, day_of_week="mon")`), batch_size=50, max_age_days=90 - `worker_ready` sanity check `SELECT 1 FROM nspd_quarter_dumps LIMIT 0` (non-fatal critical log per `Bug_Worker_Ready_EarlyReturn_Fixed`) ### `backend/app/api/v1/admin_scrape.py` ```python class HarvestQuarterRequest(BaseModel): quarter_cad: str = Field(min_length=5, pattern=r"^\d+:\d+:\d+$") region_code: int = Field(default=66) include_zouit: bool = Field(default=True) include_risks: bool = Field(default=False) @router.post("/nspd/harvest-quarter") def trigger_harvest_quarter(...) -> dict: """Manual trigger. Requires X-Admin-Token. Returns task_id.""" ``` Pattern `^\d+:\d+:\d+$` rejects 4-сегментный cad (only quarters allowed для этого task). ## Critical contracts | Contract | Implementation | |---|---| | Schema MultiPolygon (PR #110) | `ST_Multi(ST_Transform(ST_SetSRID(ST_GeomFromGeoJSON(:geom), 3857), 4326))` на ingest | | bbox storage | `ST_MakeEnvelope(:xmin, :ymin, :xmax, :ymax, 3857)` | | layers_fetched TEXT[] | PostgreSQL array literal `{a,b,c}` через string concat | | Empty quarter | Row с `quarter_geom=NULL`, `bbox_3857=NULL`, `features_json='[]'`, counts=0, no error | | Failed harvest | Row с `harvest_error=str(exc)`, counts=0 — re-tried в следующий beat | ## Tests: `backend/tests/test_nspd_sync.py` (14 tests) Mock-based, no real NSPD/DB: - `test_harvest_quarter_happy_path` — search_by_quarter mock → verify UPSERT SQL - `test_harvest_quarter_empty_quarter` — NSPD пуст → row with null geometry - `test_harvest_quarter_waf_retry` — `NspdLiteWafError` → re-raise (autoretry) - `test_harvest_quarter_generic_error_writes_row` — generic exc → upsert error row, no re-raise - `test_harvest_stale_quarters_fanout` — 50 stale → 50 apply_async calls - `test_harvest_stale_quarters_batch_limit` — `batch_size=10` → 10 calls maximum - `test_features_json_zouit_groups` — `include_zouit=True` → all 5 zouit keys present - `test_features_json_risks_groups` — `include_risks=True` → all 11 risk keys - `test_features_json_layer_assignment` — каждый feature имеет правильный layer name - `test_upsert_with_geom_uses_st_multi` — verify SQL uses `ST_Multi(ST_Transform(...))` - `test_upsert_without_geom_uses_null` — empty quarter → CASE WHEN NULL THEN NULL - ... (всего 14) `uv run pytest tests/test_nspd_sync.py -v` → **14/14 passed**. ## Code review (pre-push, code-reviewer agent) Verdict: **APPROVE**, 0 blocking, 5 minor (all acceptable defer): 1. inline `import datetime` in error else-branch (style, technical debt) 2. `layers_fetched` array literal via string concat — works for current layer names; risky only if future layer name contains `,` or `{` 3. `harvest_stale_quarters` re-tries error rows infinitely — intended behavior (manual cleanup or retry-count cap later) 4. Test `call_args.kwargs["args"]` fragile if call site changes 5. Beat collision with `refresh-ekb-districts-medians` once every ~2 years (negligible) ## Acceptance (per #94 plan) - [x] `harvest_quarter` task implemented с WAF retry - [x] `harvest_stale_quarters` beat fanout - [x] Mon 04:00 МСК beat schedule - [x] Admin endpoint для manual trigger - [x] Tests: 14/14 pass - [x] Sanity check в worker_ready - [x] Code-reviewer APPROVE ## Test plan - [x] `ruff check + format` — clean - [x] `pytest tests/test_nspd_sync.py -v` — 14/14 passed - [ ] After deploy — smoke: ```bash curl -X POST https://gendsgn.ru/api/v1/admin/scrape/nspd/harvest-quarter \ -H "X-Admin-Token: $TOKEN" \ -d '{"quarter_cad": "66:41:0204016"}' # → {"task_id": "...", "quarter_cad": "66:41:0204016"} # 10s later psql -c "SELECT quarter_cad, total_features, harvest_error FROM nspd_quarter_dumps WHERE quarter_cad='66:41:0204016'" # → total_features > 0, harvest_error IS NULL ``` - [ ] Verify Celery Beat подхватил `nspd-harvest-stale-quarters` в schedule (logs) - [ ] После первого Mon 04:00 — verify N rows added в nspd_quarter_dumps Sequential rule — следующий PR (analyze_parcel integration) только после merge этого. Part of #94.

lekss361 commented

2026-05-12 15:37:48 +00:00

(Migrated from github.com)

⚠️ Auto-review: minor (non-blocking, но M1 — runtime bug)

Backend — backend/app/workers/tasks/nspd_sync.py:

M1 (functional bug, fix first): :477 harvest_stale_quarters запрашивает region_code из cad_quarters_geom, но этой колонки нет в схеме. Prod schema: (cad_number, geom, raw_props, fetched_at, source). При первом beat tick'е получим column "region_code" does not exist. Fix: убрать WHERE region_code = :rc из outer SELECT по cad_quarters_geom (inner subquery по nspd_quarter_dumps корректно использует region_code).
M2 (correctness): _upsert_dump :273-335 не делает rollback на exception. finally: db.close() без except: rollback() → PendingRollbackError при PostGIS отбраковке geometry или JSON cast failure. harvest_stale_quarters :497-500 уже имеет explicit rollback — повторить паттерн в _upsert_dump.
M4: :299 layers_fetched array literal через string join — "{" + ",".join(...) + "}" обходит psycopg v3 native array binding. SQL уже имеет CAST(:layers_fetched AS text[]) — заменить на "layers_fetched": list(dump.layers_fetched).
M5: :307 import datetime as _dt внутри else branch — top-of-file import convention. Поднять наверх.

backend/app/workers/celery_app.py:

M3: :94-111 worker_ready sanity-check — inner try/finally без rollback() перед _db.close(). Low priority (SELECT 1), но консистентно с проектом стоит добавить.

backend/tests/test_nspd_sync.py:

M7 (test dead code): :819-844 mock __enter__/__exit__ на SessionLocal() — но harvest_stale_quarters не использует context manager. Эти моки no-op. Test работает потому что mock_db.execute.return_value.all.return_value задан напрямую. Cleanup для clarity.

Positive:

_check_token(x_admin_token) :34 — auth correct per project pattern.
HarvestQuarterRequest Pydantic regex ^\d+:\d+:\d+$ + range guards — input validated before enqueue.
autoretry_for=(NspdLiteWafError,), retry_backoff=True, retry_jitter=True, max_retries=3 — корректный Celery pattern для WAF.
UPSERT idempotent через ON CONFLICT (quarter_cad) DO UPDATE, region_code исключён из UPDATE (immutable).
worker_ready НЕ повторяет Bug_Worker_Ready_EarlyReturn_Fixed — nested try/finally без return.
No print(), requests, psycopg2. logger.* всюду.

M6 deferred import в admin_scrape.py — by design per file convention, not an issue.

Fix and push — bot will re-review automatically.

⚠️ Auto-review: minor (non-blocking, но M1 — runtime bug) **Backend — `backend/app/workers/tasks/nspd_sync.py`:** - **M1 (functional bug, fix first): `:477` `harvest_stale_quarters` запрашивает `region_code` из `cad_quarters_geom`, но этой колонки нет в схеме.** Prod schema: `(cad_number, geom, raw_props, fetched_at, source)`. При первом beat tick'е получим `column "region_code" does not exist`. Fix: убрать `WHERE region_code = :rc` из outer SELECT по `cad_quarters_geom` (inner subquery по `nspd_quarter_dumps` корректно использует `region_code`). - **M2 (correctness): `_upsert_dump :273-335` не делает rollback на exception.** `finally: db.close()` без `except: rollback()` → `PendingRollbackError` при PostGIS отбраковке geometry или JSON cast failure. `harvest_stale_quarters :497-500` уже имеет explicit rollback — повторить паттерн в `_upsert_dump`. - **M4: `:299` `layers_fetched` array literal через string join** — `"{" + ",".join(...) + "}"` обходит psycopg v3 native array binding. SQL уже имеет `CAST(:layers_fetched AS text[])` — заменить на `"layers_fetched": list(dump.layers_fetched)`. - **M5: `:307` `import datetime as _dt` внутри else branch** — top-of-file import convention. Поднять наверх. **`backend/app/workers/celery_app.py`:** - **M3: `:94-111` worker_ready sanity-check** — inner try/finally без `rollback()` перед `_db.close()`. Low priority (SELECT 1), но консистентно с проектом стоит добавить. **`backend/tests/test_nspd_sync.py`:** - **M7 (test dead code): `:819-844` mock `__enter__`/`__exit__`** на `SessionLocal()` — но `harvest_stale_quarters` не использует context manager. Эти моки no-op. Test работает потому что `mock_db.execute.return_value.all.return_value` задан напрямую. Cleanup для clarity. **Positive:** - `_check_token(x_admin_token) :34` — auth correct per project pattern. - `HarvestQuarterRequest` Pydantic regex `^\d+:\d+:\d+$` + range guards — input validated before enqueue. - `autoretry_for=(NspdLiteWafError,)`, `retry_backoff=True`, `retry_jitter=True`, `max_retries=3` — корректный Celery pattern для WAF. - UPSERT idempotent через `ON CONFLICT (quarter_cad) DO UPDATE`, `region_code` исключён из UPDATE (immutable). - `worker_ready` НЕ повторяет `Bug_Worker_Ready_EarlyReturn_Fixed` — nested try/finally без `return`. - No `print()`, `requests`, `psycopg2`. `logger.*` всюду. **M6 deferred import в admin_scrape.py — by design per file convention, not an issue.** Fix and push — bot will re-review automatically.

lekss361 commented

2026-05-12 15:43:25 +00:00

(Migrated from github.com)

Все 6 findings применены в `aca4b44`:

M1 (runtime bug) ✅ — `harvest_stale_quarters` убрал `WHERE region_code` из outer SELECT по `cad_quarters_geom` (колонки нет в prod schema). Inner subquery по `nspd_quarter_dumps` сохраняет фильтр.

M2 (correctness) ✅ — `_upsert_dump` теперь `except Exception: db.rollback(); raise` перед finally close. PendingRollbackError prevented.

M3 (low) ✅ — `worker_ready` breadcrumb INSERT — same rollback pattern для consistency.

M4 (style) ✅ — `layers_fetched` — native Python `list` вместо string concat `'{' + ','.join(...) + '}'`. psycopg v3 сам сериализует.

M5 (style) ✅ — `import datetime as _dt` → top-level imports.

M7 (test cleanup) ✅ — dead `enter`/`exit` mock setup удалён.

`ruff` + `format` + 14/14 tests pass. Ждём re-review.

Все 6 findings применены в \`aca4b44\`: **M1 (runtime bug)** ✅ — \`harvest_stale_quarters\` убрал \`WHERE region_code\` из outer SELECT по \`cad_quarters_geom\` (колонки нет в prod schema). Inner subquery по \`nspd_quarter_dumps\` сохраняет фильтр. **M2 (correctness)** ✅ — \`_upsert_dump\` теперь \`except Exception: db.rollback(); raise\` перед finally close. PendingRollbackError prevented. **M3 (low)** ✅ — \`worker_ready\` breadcrumb INSERT — same rollback pattern для consistency. **M4 (style)** ✅ — \`layers_fetched\` — native Python \`list\` вместо string concat \`'{' + ','.join(...) + '}'\`. psycopg v3 сам сериализует. **M5 (style)** ✅ — \`import datetime as _dt\` → top-level imports. **M7 (test cleanup)** ✅ — dead \`__enter__\`/\`__exit__\` mock setup удалён. \`ruff\` + \`format\` + 14/14 tests pass. Ждём re-review.

lekss361 commented

2026-05-12 15:45:48 +00:00

(Migrated from github.com)

✅ Auto-review passed (LGTM) — все 5 actionable findings корректно применены.

Fixes verified:

M1 fixed: outer SELECT cad_number FROM cad_quarters_geom :492-503 — без WHERE region_code. region_code = :rc остался только в inner subquery по nspd_quarter_dumps где колонка существует. No more runtime crash.
M2 fixed: _upsert_dump :343-347 — except Exception: db.rollback(); raise перед finally: db.close(). Pattern correct.
M3 fixed: celery_app.py :90-92 — except Exception: _db.rollback(); raise в _resume_zombie_runs inner try/finally. Outer try/except :95 catches+logs.
M4 fixed: layers_fetched теперь list(dump.layers_fetched) (lines 310 + 414). CAST(:layers_fetched AS text[]) получает Python list, psycopg v3 нативно binds — no string-join bypass.
M5 fixed: import datetime as _dt на top of file (line 148). Error-only path :335 использует _dt.datetime.now(_dt.UTC).

Bonus cleanup:

M7 dead code (__enter__/__exit__ mock) удалён из теста.
harvest_stale_quarters :508-511 — explicit try: db.rollback() except: pass перед return error dict — корректный pattern.

No regressions. Admin endpoint _check_token :34 intact. Deferred import :35 (M6) by-design per file convention.

Merging.

✅ Auto-review passed (LGTM) — все 5 actionable findings корректно применены. **Fixes verified:** - **M1 fixed:** outer `SELECT cad_number FROM cad_quarters_geom :492-503` — без `WHERE region_code`. `region_code = :rc` остался только в inner subquery по `nspd_quarter_dumps` где колонка существует. No more runtime crash. - **M2 fixed:** `_upsert_dump :343-347` — `except Exception: db.rollback(); raise` перед `finally: db.close()`. Pattern correct. - **M3 fixed:** `celery_app.py :90-92` — `except Exception: _db.rollback(); raise` в `_resume_zombie_runs` inner try/finally. Outer try/except :95 catches+logs. - **M4 fixed:** `layers_fetched` теперь `list(dump.layers_fetched)` (lines 310 + 414). `CAST(:layers_fetched AS text[])` получает Python list, psycopg v3 нативно binds — no string-join bypass. - **M5 fixed:** `import datetime as _dt` на top of file (line 148). Error-only path :335 использует `_dt.datetime.now(_dt.UTC)`. **Bonus cleanup:** - M7 dead code (`__enter__`/`__exit__` mock) удалён из теста. - `harvest_stale_quarters` :508-511 — explicit `try: db.rollback() except: pass` перед return error dict — корректный pattern. No regressions. Admin endpoint `_check_token` :34 intact. Deferred import :35 (M6) by-design per file convention. Merging.