bot-backend
dffe1ccc18
Хвост #769 (A1+D смержены #798; E отложен — geo-radius client-risk). A2 (#10 index): 086 — pg_trgm GIN индекс на deals.address (был seq-scan на street-match). Идемпот (IF NOT EXISTS), plain CREATE INDEX (txn-safe). B (#13 FDW): 087 — ALTER SERVER gendesign_remote OPTIONS fetch_size=10000 + use_remote_estimate=true (лучше планируется 6.8M-row FDW import). DO-block читает srvoptions → ADD-or-SET, идемпотентно. C (#14 matview): listings_search_mv активно читается search_query.py но НИКОГДА не рефрешился (refresh_search_matview.py существует, в beat/scheduler не было). scheduler.py: + trigger_refresh_search_matview_run (по образцу asking_to_sold, через _claim_run advisory-lock #750, run_in_executor — refresh CONCURRENTLY с autocommit). 088 — seed scrape_schedules (03:00-04:00 UTC ночь, ON CONFLICT DO NOTHING). ruff clean. Post-deploy QA: EXPLAIN deals street-query → Bitmap Index Scan; FDW remote-estimate.
838 lines
36 KiB
Python
838 lines
36 KiB
Python
"""In-app scheduler — periodically triggers configured scrapes based on scrape_schedules table.
|
||
|
||
Запускается в FastAPI lifespan (app.main lifespan context manager) при startup.
|
||
Логика:
|
||
1. Каждые 60 sec query scrape_schedules WHERE enabled=true AND next_run_at <= NOW().
|
||
2. Для каждого: проверить нет ли уже running run этого source — если нет, trigger.
|
||
3. После trigger — recompute next_run_at в следующих суток.
|
||
4. На old runs (orphaned status='running' > 6h без heartbeat) — пометить как 'zombie'.
|
||
|
||
Sources:
|
||
- avito_city_sweep → run_avito_city_sweep (scrape_pipeline.py)
|
||
- yandex_city_sweep → run_yandex_city_sweep (scrape_pipeline.py, #561; shipped DORMANT)
|
||
- n1_city_sweep → run_n1_city_sweep (scrape_pipeline.py, #575)
|
||
- cian_history_backfill → run_cian_backfill (this module, #560)
|
||
- rosreestr_dkp_import → import_rosreestr_dkp (this module, #563)
|
||
- listing_source_snapshot → snapshot_listing_sources (tasks/listing_source_snapshot.py, #570;
|
||
pure internal DB snapshot — no external calls, enabled by default)
|
||
- asking_to_sold_ratio_refresh → recompute_asking_to_sold_ratios
|
||
(tasks/asking_to_sold_ratio.py, #648 Stage 4; pure internal DB
|
||
re-derivation of the asking→sold ratios — no external calls,
|
||
enabled by default; window after rosreestr_dkp_import)
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import logging
|
||
import random
|
||
from datetime import UTC, datetime, time, timedelta
|
||
from typing import Any
|
||
|
||
import sentry_sdk
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.db import SessionLocal
|
||
from app.services import scrape_runs as runs_mod
|
||
from app.services.scrape_pipeline import (
|
||
run_avito_city_sweep,
|
||
run_n1_city_sweep,
|
||
run_yandex_city_sweep,
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Loop interval — check каждую минуту
|
||
SCHEDULER_TICK_SEC = 60
|
||
ZOMBIE_THRESHOLD_HOURS = 6
|
||
|
||
|
||
def compute_next_run_at(
|
||
window_start_hour: int, window_end_hour: int, *, now: datetime | None = None
|
||
) -> datetime:
|
||
"""Pick random datetime в window [start, end) UTC, для СЛЕДУЮЩИХ суток после now.
|
||
|
||
Если window_end_hour <= window_start_hour → cross-midnight window
|
||
(например 22→3 → окно 22:00-23:59 ИЛИ 00:00-02:59).
|
||
"""
|
||
now = now or datetime.now(tz=UTC)
|
||
tomorrow = (now + timedelta(days=1)).date()
|
||
|
||
if window_end_hour > window_start_hour:
|
||
# Обычное окно (например 2..5 → 02:00-04:59)
|
||
start_seconds = window_start_hour * 3600
|
||
end_seconds = window_end_hour * 3600
|
||
rand_seconds = random.randint(start_seconds, end_seconds - 1)
|
||
return datetime.combine(tomorrow, time(0, 0), tzinfo=UTC) + timedelta(seconds=rand_seconds)
|
||
else:
|
||
# Cross-midnight (22..3 → 22:00-23:59 + 00:00-02:59)
|
||
# Длина окна = (24-start) + end часов
|
||
total_seconds = ((24 - window_start_hour) + window_end_hour) * 3600
|
||
rand_seconds = random.randint(0, total_seconds - 1)
|
||
# Если rand попадает в первую часть (start..24)
|
||
first_half = (24 - window_start_hour) * 3600
|
||
if rand_seconds < first_half:
|
||
# Текущая дата (если ещё не наступило окно) или next day
|
||
base_date = now.date() if now.hour < window_start_hour else tomorrow
|
||
return datetime.combine(base_date, time(0, 0), tzinfo=UTC) + timedelta(
|
||
seconds=window_start_hour * 3600 + rand_seconds
|
||
)
|
||
else:
|
||
# Во второй части (0..end), следующего дня
|
||
offset = rand_seconds - first_half
|
||
return datetime.combine(tomorrow, time(0, 0), tzinfo=UTC) + timedelta(seconds=offset)
|
||
|
||
|
||
def has_running_run(db: Session, source: str) -> bool:
|
||
"""Есть ли активный run для source (status='running')."""
|
||
row = db.execute(
|
||
text(
|
||
"""
|
||
SELECT 1 FROM scrape_runs
|
||
WHERE source = :source AND status = 'running'
|
||
LIMIT 1
|
||
"""
|
||
),
|
||
{"source": source},
|
||
).fetchone()
|
||
return row is not None
|
||
|
||
|
||
def reap_zombies(db: Session) -> int:
|
||
"""Mark scrape_runs as 'zombie' если heartbeat не обновлялся > ZOMBIE_THRESHOLD_HOURS hours."""
|
||
zombie_interval = f"{ZOMBIE_THRESHOLD_HOURS} hours"
|
||
result = db.execute(
|
||
text(
|
||
"""
|
||
UPDATE scrape_runs
|
||
SET status = 'zombie', finished_at = NOW()
|
||
WHERE status = 'running'
|
||
AND (heartbeat_at IS NULL
|
||
OR heartbeat_at < NOW() - CAST(:interval AS interval))
|
||
RETURNING id
|
||
"""
|
||
),
|
||
{"interval": zombie_interval},
|
||
)
|
||
rows = result.fetchall()
|
||
db.commit()
|
||
if rows:
|
||
logger.warning("scheduler: reaped %d zombie runs: %s", len(rows), [r.id for r in rows])
|
||
return len(rows)
|
||
|
||
|
||
def _claim_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
|
||
"""Claim run: INSERT scrape_runs + UPDATE scrape_schedules next_run_at.
|
||
|
||
Returns run_id или None если уже есть running run для этого source.
|
||
Общий helper для всех source-handler'ов.
|
||
"""
|
||
source = schedule_row["source"]
|
||
if has_running_run(db, source):
|
||
logger.info("scheduler: skip — already running for source=%s", source)
|
||
return None
|
||
|
||
# #750: атомарный claim через transaction-scoped advisory lock. Сериализует
|
||
# конкурентные _claim_run одного source (redeploy-overlap старого+нового контейнера
|
||
# / 2 реплики), иначе оба пройдут has_running_run-pre-check и запустят ДВОЙНОЙ sweep
|
||
# → 2× rate к Avito/Cian → бан. pg_try_advisory_xact_lock НЕ блокирует: lock занят
|
||
# другим тиком → false → return None. Лок авто-снимается на commit/rollback этой
|
||
# транзакции (has_running_run выше — быстрый pre-check, обычно без взятия лока).
|
||
got_lock = db.execute(
|
||
text("SELECT pg_try_advisory_xact_lock(hashtext(:source))"),
|
||
{"source": source},
|
||
).scalar()
|
||
if not got_lock:
|
||
logger.info("scheduler: skip — concurrent claim in progress for source=%s", source)
|
||
return None
|
||
|
||
# Double-checked под локом: конкурентный тик мог закоммитить running-run МЕЖДУ
|
||
# pre-check и взятием лока (затем освободить lock на commit). Перепроверяем под
|
||
# локом — иначе словим двойной run на узком окне.
|
||
if has_running_run(db, source):
|
||
logger.info("scheduler: skip — running appeared under lock for source=%s", source)
|
||
db.rollback() # освобождаем advisory lock (claim не состоялся)
|
||
return None
|
||
|
||
params = schedule_row.get("default_params") or {}
|
||
run_id = runs_mod.create_run(db, source=source, params=params)
|
||
|
||
next_at = compute_next_run_at(
|
||
schedule_row["window_start_hour"],
|
||
schedule_row["window_end_hour"],
|
||
)
|
||
db.execute(
|
||
text(
|
||
"""
|
||
UPDATE scrape_schedules
|
||
SET last_run_id = :run_id, last_run_at = NOW(),
|
||
next_run_at = :next_at, updated_at = NOW()
|
||
WHERE source = :source
|
||
"""
|
||
),
|
||
{"run_id": run_id, "next_at": next_at, "source": source},
|
||
)
|
||
db.commit()
|
||
logger.info(
|
||
"scheduler: claimed run_id=%d source=%s next_run_at=%s",
|
||
run_id,
|
||
source,
|
||
next_at.isoformat(),
|
||
)
|
||
return run_id
|
||
|
||
|
||
async def trigger_avito_city_sweep_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
|
||
"""Создать новый scrape_runs + launch run_avito_city_sweep в asyncio.create_task.
|
||
|
||
Returns run_id (или None если skip — есть running run).
|
||
"""
|
||
run_id = _claim_run(db, schedule_row)
|
||
if run_id is None:
|
||
return None
|
||
|
||
params = schedule_row.get("default_params") or {}
|
||
|
||
# Spawn asyncio task — pass NEW session (avoid sharing with this scheduler tick)
|
||
async def _run() -> None:
|
||
run_db = SessionLocal()
|
||
try:
|
||
await run_avito_city_sweep(
|
||
run_db,
|
||
run_id=run_id,
|
||
pages_per_anchor=int(params.get("pages_per_anchor", 3)),
|
||
detail_top_n=int(params.get("detail_top_n", 20)),
|
||
request_delay_sec=float(params.get("request_delay_sec", 7.0)),
|
||
enrich_houses=bool(params.get("enrich_houses", True)),
|
||
radius_m=int(params.get("radius_m", 1500)),
|
||
)
|
||
except Exception:
|
||
logger.exception("scheduler: run_avito_city_sweep crashed run_id=%d", run_id)
|
||
finally:
|
||
run_db.close()
|
||
|
||
task = asyncio.create_task(_run())
|
||
# Keep reference to avoid GC before task completes (RUF006)
|
||
task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None)
|
||
logger.info("scheduler: triggered city_sweep run_id=%d", run_id)
|
||
return run_id
|
||
|
||
|
||
async def trigger_yandex_city_sweep_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
|
||
"""Создать новый scrape_runs + launch run_yandex_city_sweep в asyncio.create_task.
|
||
|
||
Зеркало trigger_avito_city_sweep_run, но проще (#561): у Yandex sweep нет
|
||
houses/detail-параметров. Shipped DORMANT — schedule seed enabled=false (078),
|
||
включается оператором вручную после #623 (egress-IP rotation).
|
||
|
||
Returns run_id (или None если skip — есть running run).
|
||
"""
|
||
run_id = _claim_run(db, schedule_row)
|
||
if run_id is None:
|
||
return None
|
||
|
||
params = schedule_row.get("default_params") or {}
|
||
|
||
# Spawn asyncio task — pass NEW session (avoid sharing with this scheduler tick)
|
||
async def _run() -> None:
|
||
run_db = SessionLocal()
|
||
try:
|
||
await run_yandex_city_sweep(
|
||
run_db,
|
||
run_id=run_id,
|
||
pages_per_anchor=int(params.get("pages_per_anchor", 2)),
|
||
request_delay_sec=float(params.get("request_delay_sec", 9.0)),
|
||
radius_m=int(params.get("radius_m", 1500)),
|
||
)
|
||
except Exception:
|
||
logger.exception("scheduler: run_yandex_city_sweep crashed run_id=%d", run_id)
|
||
finally:
|
||
run_db.close()
|
||
|
||
task = asyncio.create_task(_run())
|
||
# Keep reference to avoid GC before task completes (RUF006)
|
||
task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None)
|
||
logger.info("scheduler: triggered yandex_city_sweep run_id=%d", run_id)
|
||
return run_id
|
||
|
||
|
||
async def trigger_n1_city_sweep_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
|
||
"""Создать новый scrape_runs + launch run_n1_city_sweep в asyncio.create_task (#575).
|
||
|
||
Зеркало trigger_yandex_city_sweep_run — N1 sweep простой (search → save, без
|
||
houses/detail). Returns run_id (или None если skip — есть running run).
|
||
"""
|
||
run_id = _claim_run(db, schedule_row)
|
||
if run_id is None:
|
||
return None
|
||
|
||
params = schedule_row.get("default_params") or {}
|
||
|
||
async def _run() -> None:
|
||
run_db = SessionLocal()
|
||
try:
|
||
await run_n1_city_sweep(
|
||
run_db,
|
||
run_id=run_id,
|
||
pages_per_anchor=int(params.get("pages_per_anchor", 2)),
|
||
request_delay_sec=float(params.get("request_delay_sec", 7.0)),
|
||
radius_m=int(params.get("radius_m", 1500)),
|
||
)
|
||
except Exception:
|
||
logger.exception("scheduler: run_n1_city_sweep crashed run_id=%d", run_id)
|
||
finally:
|
||
run_db.close()
|
||
|
||
task = asyncio.create_task(_run())
|
||
task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None)
|
||
logger.info("scheduler: triggered n1_city_sweep run_id=%d", run_id)
|
||
return run_id
|
||
|
||
|
||
async def trigger_cian_backfill_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
|
||
"""Создать scrape_runs + launch run_cian_backfill в asyncio.create_task.
|
||
|
||
Pre-checks Cian session validity before claiming the run.
|
||
Marks nothing as failed if cookies are absent/expired — just logs + GlitchTip warning.
|
||
Returns run_id или None (skip / expired cookies).
|
||
"""
|
||
from app.services.cian_session import load_session, verify_session
|
||
|
||
# Pre-check: load and verify cookies before claiming the run slot
|
||
cookies = load_session(db)
|
||
if cookies is None:
|
||
logger.warning("scheduler: cian_history_backfill skipped — no valid session cookies in DB")
|
||
return None
|
||
|
||
state = await verify_session(cookies)
|
||
if state is None:
|
||
logger.warning(
|
||
"scheduler: cian_history_backfill — cookies expired or invalid, skipping run"
|
||
)
|
||
try:
|
||
sentry_sdk.capture_message(
|
||
"cian_history_backfill skipped: Cian session cookies expired — "
|
||
"please re-upload via admin UI",
|
||
level="warning",
|
||
)
|
||
except Exception:
|
||
pass # sentry_sdk not initialised in dev
|
||
return None
|
||
|
||
run_id = _claim_run(db, schedule_row)
|
||
if run_id is None:
|
||
return None
|
||
|
||
params = schedule_row.get("default_params") or {}
|
||
|
||
async def _run() -> None:
|
||
run_db = SessionLocal()
|
||
try:
|
||
await _execute_cian_backfill(run_db, run_id=run_id, params=params)
|
||
except Exception:
|
||
logger.exception("scheduler: _execute_cian_backfill crashed run_id=%d", run_id)
|
||
finally:
|
||
run_db.close()
|
||
|
||
task = asyncio.create_task(_run())
|
||
task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None)
|
||
logger.info("scheduler: triggered cian_history_backfill run_id=%d", run_id)
|
||
return run_id
|
||
|
||
|
||
async def _execute_cian_backfill(
|
||
db: Session,
|
||
*,
|
||
run_id: int,
|
||
params: dict[str, Any],
|
||
) -> None:
|
||
"""Orchestrate Cian history backfill with heartbeat + checkpoint.
|
||
|
||
Wraps backfill_cian_history(), updating scrape_runs counters (via update_heartbeat)
|
||
before and after the batch call for zombie-detection visibility.
|
||
|
||
Checkpoint/resume semantics: backfill_cian_history() queries rows WHERE history IS
|
||
NULL via LEFT JOIN — so re-running after a partial completion naturally skips
|
||
already-processed rows (idempotent by design).
|
||
|
||
Params (from default_params jsonb):
|
||
batch_size: int — rows per run (listings + houses counted separately).
|
||
"""
|
||
from app.tasks.cian_history_backfill import backfill_cian_history
|
||
|
||
batch_size = int(params.get("batch_size", 100))
|
||
counters: dict[str, int] = {
|
||
"listings_processed": 0,
|
||
"listings_succeeded": 0,
|
||
"listings_failed": 0,
|
||
"houses_processed": 0,
|
||
"houses_succeeded": 0,
|
||
"houses_failed": 0,
|
||
}
|
||
|
||
try:
|
||
runs_mod.update_heartbeat(db, run_id, counters)
|
||
|
||
result = await backfill_cian_history(
|
||
db,
|
||
batch_size=batch_size,
|
||
do_listings=True,
|
||
do_houses=True,
|
||
do_valuations=False,
|
||
)
|
||
|
||
counters = {
|
||
"listings_processed": result.listings_processed,
|
||
"listings_succeeded": result.listings_succeeded,
|
||
"listings_failed": result.listings_failed_fetch + result.listings_failed_save,
|
||
"houses_processed": result.houses_processed,
|
||
"houses_succeeded": result.houses_succeeded,
|
||
"houses_failed": result.houses_failed_fetch + result.houses_failed_save,
|
||
"duration_sec": int(result.duration_sec),
|
||
}
|
||
runs_mod.mark_done(db, run_id, counters)
|
||
logger.info(
|
||
"scheduler: cian_history_backfill run_id=%d done — "
|
||
"listings=%d/%d houses=%d/%d %.1fs",
|
||
run_id,
|
||
result.listings_succeeded,
|
||
result.listings_total,
|
||
result.houses_succeeded,
|
||
result.houses_total,
|
||
result.duration_sec,
|
||
)
|
||
except Exception as exc:
|
||
logger.exception("scheduler: cian_history_backfill run_id=%d failed", run_id)
|
||
runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters)
|
||
raise
|
||
|
||
|
||
async def trigger_rosreestr_dkp_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
|
||
"""Создать scrape_runs + launch import_rosreestr_dkp в executor (sync I/O-bound task)."""
|
||
run_id = _claim_run(db, schedule_row)
|
||
if run_id is None:
|
||
return None
|
||
|
||
params = schedule_row.get("default_params") or {}
|
||
|
||
async def _run() -> None:
|
||
run_db = SessionLocal()
|
||
try:
|
||
# import_rosreestr_dkp is synchronous (DB-only, no async HTTP)
|
||
loop = asyncio.get_event_loop()
|
||
await loop.run_in_executor(None, import_rosreestr_dkp, run_db, run_id, params)
|
||
except Exception:
|
||
logger.exception("scheduler: import_rosreestr_dkp crashed run_id=%d", run_id)
|
||
finally:
|
||
run_db.close()
|
||
|
||
task = asyncio.create_task(_run())
|
||
task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None)
|
||
logger.info("scheduler: triggered rosreestr_dkp_import run_id=%d", run_id)
|
||
return run_id
|
||
|
||
|
||
async def trigger_listing_source_snapshot_run(
|
||
db: Session, schedule_row: dict[str, Any]
|
||
) -> int | None:
|
||
"""Создать scrape_runs + launch snapshot_listing_sources в executor (sync DB-only task).
|
||
|
||
Per-source price-history daily snapshot (#570). Mirror trigger_rosreestr_dkp_run:
|
||
задача синхронная (чистый internal DB-snapshot, БЕЗ внешних HTTP-вызовов / анти-бота),
|
||
поэтому гоняем её в run_in_executor. SAFE to enable — schedule seed enabled=true (079),
|
||
в отличие от avito/yandex sweep (dormant).
|
||
|
||
Returns run_id (или None если skip — есть running run).
|
||
"""
|
||
run_id = _claim_run(db, schedule_row)
|
||
if run_id is None:
|
||
return None
|
||
|
||
async def _run() -> None:
|
||
run_db = SessionLocal()
|
||
try:
|
||
# snapshot_listing_sources is synchronous (DB-only, no async HTTP)
|
||
from app.tasks.listing_source_snapshot import snapshot_listing_sources
|
||
|
||
loop = asyncio.get_event_loop()
|
||
await loop.run_in_executor(None, snapshot_listing_sources, run_db, run_id)
|
||
except Exception:
|
||
logger.exception("scheduler: snapshot_listing_sources crashed run_id=%d", run_id)
|
||
finally:
|
||
run_db.close()
|
||
|
||
task = asyncio.create_task(_run())
|
||
task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None)
|
||
logger.info("scheduler: triggered listing_source_snapshot run_id=%d", run_id)
|
||
return run_id
|
||
|
||
|
||
async def trigger_refresh_search_matview_run(
|
||
db: Session, schedule_row: dict[str, Any]
|
||
) -> int | None:
|
||
"""Создать scrape_runs + запустить REFRESH MATERIALIZED VIEW CONCURRENTLY listings_search_mv.
|
||
|
||
listings_search_mv активно читается search_query.py (SELECT FROM listings_search_mv).
|
||
Без периодического refresh матвью устаревает по мере появления новых listings.
|
||
Задача синхронная (DB-only, БЕЗ внешних HTTP-вызовов) — гоняем в run_in_executor
|
||
по образцу trigger_asking_to_sold_ratio_run (#648).
|
||
|
||
refresh_search_matview() использует собственное psycopg-соединение с autocommit=True
|
||
(REFRESH CONCURRENTLY не может работать внутри транзакции). scrape_runs lifecycle
|
||
(mark_done/mark_failed) управляется через отдельную SessionLocal.
|
||
|
||
SAFE to enable — schedule seed enabled=true (088), pure internal DB op.
|
||
Окно 03:00-04:00 UTC — ночное, до rosreestr_dkp_import (04:00-06:00 UTC).
|
||
|
||
Returns run_id (или None если skip — есть running run).
|
||
"""
|
||
run_id = _claim_run(db, schedule_row)
|
||
if run_id is None:
|
||
return None
|
||
|
||
async def _run() -> None:
|
||
run_db = SessionLocal()
|
||
try:
|
||
from app.tasks.refresh_search_matview import refresh_search_matview
|
||
|
||
loop = asyncio.get_event_loop()
|
||
# refresh_search_matview() manages its own connection with autocommit=True
|
||
# (required for REFRESH MATERIALIZED VIEW CONCURRENTLY outside a transaction).
|
||
await loop.run_in_executor(None, refresh_search_matview)
|
||
runs_mod.mark_done(run_db, run_id, {})
|
||
except Exception:
|
||
logger.exception("scheduler: refresh_search_matview crashed run_id=%d", run_id)
|
||
runs_mod.mark_failed(run_db, run_id, "refresh_search_matview failed", {})
|
||
finally:
|
||
run_db.close()
|
||
|
||
task = asyncio.create_task(_run())
|
||
task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None)
|
||
logger.info("scheduler: triggered refresh_search_matview run_id=%d", run_id)
|
||
return run_id
|
||
|
||
|
||
async def trigger_asking_to_sold_ratio_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
|
||
"""Создать scrape_runs + launch recompute_asking_to_sold_ratios в executor (sync DB-only task).
|
||
|
||
Daily TRUE-MIRROR refresh асking→sold коэффициентов (#648 Stage 4): DELETE WHERE
|
||
district='' + re-derive INSERT (та же 080-derivation) в одной транзакции, чтобы ratio
|
||
не устаревал по мере ночного импорта ДКП. Mirror trigger_listing_source_snapshot_run:
|
||
задача синхронная (чистый internal DB re-derive, БЕЗ внешних HTTP-вызовов / анти-бота),
|
||
поэтому гоняем её в run_in_executor. SAFE to enable — schedule seed enabled=true (082),
|
||
в отличие от avito/yandex sweep (dormant).
|
||
|
||
Returns run_id (или None если skip — есть running run).
|
||
"""
|
||
run_id = _claim_run(db, schedule_row)
|
||
if run_id is None:
|
||
return None
|
||
|
||
async def _run() -> None:
|
||
run_db = SessionLocal()
|
||
try:
|
||
# recompute_asking_to_sold_ratios is synchronous (DB-only, no async HTTP)
|
||
from app.tasks.asking_to_sold_ratio import recompute_asking_to_sold_ratios
|
||
|
||
loop = asyncio.get_event_loop()
|
||
await loop.run_in_executor(None, recompute_asking_to_sold_ratios, run_db, run_id)
|
||
except Exception:
|
||
logger.exception("scheduler: recompute_asking_to_sold_ratios crashed run_id=%d", run_id)
|
||
finally:
|
||
run_db.close()
|
||
|
||
task = asyncio.create_task(_run())
|
||
task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None)
|
||
logger.info("scheduler: triggered asking_to_sold_ratio_refresh run_id=%d", run_id)
|
||
return run_id
|
||
|
||
|
||
def import_rosreestr_dkp(
|
||
db: Session,
|
||
run_id: int,
|
||
params: dict[str, Any],
|
||
) -> None:
|
||
"""Import ДКП-сделок из gendesign rosreestr_deals через postgres_fdw.
|
||
|
||
Python-порт import-rosreestr.sh (Variant C из #563).
|
||
|
||
Источник: foreign table gendesign_rosreestr_deals (создана в migration 072).
|
||
SERVER gendesign_remote настроен в 060_postgres_fdw_extension.sql.
|
||
USER MAPPING создаётся при startup в core/fdw.py (tradein_fdw_reader).
|
||
|
||
Фильтры (совпадают с import-rosreestr.sh + Fix_Rosreestr_Dkp_Filter_May24):
|
||
- region_code = 66 (Свердловская область)
|
||
- city ILIKE '%катеринбург%'
|
||
- realestate_type_code = '002001003000' (квартира)
|
||
- area BETWEEN 18 AND 200
|
||
- deal_price BETWEEN 1000000 AND 100000000
|
||
- street IS NOT NULL AND trim(street) != ''
|
||
- doc_type = 'ДКП' (только вторичка — #549 / Fix_Rosreestr_Dkp_Filter_May24)
|
||
- period_start_date >= since (default '2024-01-01')
|
||
|
||
dedup_hash: 'ros:dkp:' || id — плоский натуральный ключ (инъективный, без коллизий,
|
||
human-readable). До #576 здесь был md5('ros:dkp:' || id); миграция 077 конвертировала
|
||
существующие строки. source_id хранит исходный rosreestr id (дедуп переустанавливаем).
|
||
Rooms: выводятся из площади (Росреестр не отдаёт кол-во комнат).
|
||
|
||
Batch-процессинг: читаем из FDW батчами по batch_size через cursor-based пагинацию
|
||
(WHERE id > last_id ORDER BY id). Heartbeat обновляется каждый батч (= checkpoint).
|
||
SAVEPOINT per row — один сбойный row не откатывает батч.
|
||
|
||
Координаты: NULL после импорта — геокодинг остаётся follow-up (geocode-deals).
|
||
TODO (follow-up): запустить geocode backfill после import.
|
||
|
||
Cleanup: удаляет legacy строки address='Екатеринбург, реальная сделка' (pre-#549).
|
||
"""
|
||
since: str = str(params.get("since", "2024-01-01"))
|
||
batch_size: int = int(params.get("batch_size", 2000))
|
||
|
||
counters: dict[str, int] = {
|
||
"rows_fetched": 0,
|
||
"rows_inserted": 0,
|
||
"rows_skipped": 0,
|
||
"batches_done": 0,
|
||
}
|
||
|
||
# Cleanup legacy synthetic rows (pre-#549, idempotent)
|
||
try:
|
||
deleted = db.execute(
|
||
text("DELETE FROM deals WHERE address = CAST(:addr AS text) RETURNING id"),
|
||
{"addr": "Екатеринбург, реальная сделка"},
|
||
).fetchall()
|
||
db.commit()
|
||
if deleted:
|
||
logger.info(
|
||
"rosreestr_dkp_import run_id=%d: removed %d legacy synthetic rows",
|
||
run_id,
|
||
len(deleted),
|
||
)
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"rosreestr_dkp_import run_id=%d: cleanup failed (non-fatal): %s", run_id, exc
|
||
)
|
||
db.rollback()
|
||
|
||
last_id: int = 0
|
||
total_batches = 0
|
||
|
||
try:
|
||
while True:
|
||
if runs_mod.is_cancelled(db, run_id):
|
||
logger.info("rosreestr_dkp_import run_id=%d: cancelled by user", run_id)
|
||
runs_mod.mark_cancelled(db, run_id)
|
||
return
|
||
|
||
# Cursor-based pagination via foreign table gendesign_rosreestr_deals.
|
||
# FDW pushes WHERE + ORDER BY + LIMIT to gendesign-postgres automatically
|
||
# (postgres_fdw 'use_remote_estimate' is off by default but clause push-down
|
||
# still happens for simple predicates on the remote side).
|
||
batch_rows = (
|
||
db.execute(
|
||
text("""
|
||
SELECT
|
||
id,
|
||
id AS source_id_src,
|
||
'ros:dkp:' || CAST(id AS text) AS dedup_hash,
|
||
'Екатеринбург, ' || trim(street) AS address,
|
||
CASE
|
||
WHEN area < 30 THEN 0
|
||
WHEN area < 44 THEN 1
|
||
WHEN area < 62 THEN 2
|
||
WHEN area < 85 THEN 3
|
||
ELSE 4
|
||
END AS rooms,
|
||
round(area, 2) AS area_m2,
|
||
LEAST(
|
||
NULLIF(
|
||
substring(floor FROM '[0-9]+'), ''
|
||
)::int,
|
||
100
|
||
) AS floor_num,
|
||
year_build AS year_built,
|
||
round(deal_price)::bigint AS price_rub,
|
||
round(price_per_sqm)::int AS price_per_m2,
|
||
period_start_date AS deal_date
|
||
FROM gendesign_rosreestr_deals
|
||
WHERE region_code = 66
|
||
AND city ILIKE '%катеринбург%'
|
||
AND realestate_type_code = '002001003000'
|
||
AND area BETWEEN 18 AND 200
|
||
AND deal_price BETWEEN 1000000 AND 100000000
|
||
AND street IS NOT NULL AND trim(street) <> ''
|
||
AND doc_type = 'ДКП'
|
||
AND period_start_date >= CAST(:since AS date)
|
||
AND id > CAST(:last_id AS bigint)
|
||
ORDER BY id
|
||
LIMIT CAST(:batch_size AS int)
|
||
"""),
|
||
{"since": since, "last_id": last_id, "batch_size": batch_size},
|
||
)
|
||
.mappings()
|
||
.all()
|
||
)
|
||
|
||
if not batch_rows:
|
||
break
|
||
|
||
total_batches += 1
|
||
batch_inserted = 0
|
||
batch_skipped = 0
|
||
batch_max_id = last_id
|
||
|
||
for row in batch_rows:
|
||
row_id: int = int(row["id"])
|
||
if row_id > batch_max_id:
|
||
batch_max_id = row_id
|
||
|
||
try:
|
||
with db.begin_nested(): # SAVEPOINT per row
|
||
inserted = db.execute(
|
||
text("""
|
||
INSERT INTO deals (
|
||
source, dedup_hash, source_id, address, rooms, area_m2,
|
||
floor, year_built, price_rub, price_per_m2, deal_date
|
||
)
|
||
VALUES (
|
||
'rosreestr',
|
||
CAST(:dedup_hash AS text),
|
||
CAST(:source_id AS text),
|
||
CAST(:address AS text),
|
||
CAST(:rooms AS int),
|
||
CAST(:area_m2 AS numeric),
|
||
CAST(:floor_num AS int),
|
||
CAST(:year_built AS int),
|
||
CAST(:price_rub AS bigint),
|
||
CAST(:price_per_m2 AS int),
|
||
CAST(:deal_date AS date)
|
||
)
|
||
ON CONFLICT (dedup_hash) DO NOTHING
|
||
RETURNING id
|
||
"""),
|
||
{
|
||
"dedup_hash": row["dedup_hash"],
|
||
"source_id": str(row["source_id_src"]),
|
||
"address": row["address"],
|
||
"rooms": row["rooms"],
|
||
"area_m2": row["area_m2"],
|
||
"floor_num": row["floor_num"],
|
||
"year_built": row["year_built"],
|
||
"price_rub": row["price_rub"],
|
||
"price_per_m2": row["price_per_m2"],
|
||
"deal_date": row["deal_date"],
|
||
},
|
||
).fetchone()
|
||
if inserted is not None:
|
||
batch_inserted += 1
|
||
else:
|
||
batch_skipped += 1
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"rosreestr_dkp_import run_id=%d: row id=%d INSERT failed: %s",
|
||
run_id,
|
||
row_id,
|
||
exc,
|
||
)
|
||
batch_skipped += 1
|
||
|
||
last_id = batch_max_id
|
||
db.commit()
|
||
|
||
counters["rows_fetched"] += len(batch_rows)
|
||
counters["rows_inserted"] += batch_inserted
|
||
counters["rows_skipped"] += batch_skipped
|
||
counters["batches_done"] = total_batches
|
||
counters["last_id"] = last_id # type: ignore[assignment]
|
||
|
||
# Heartbeat = checkpoint: allows zombie detection + resume visibility
|
||
runs_mod.update_heartbeat(db, run_id, counters)
|
||
logger.info(
|
||
"rosreestr_dkp_import run_id=%d: batch=%d fetched=%d "
|
||
"inserted=%d skipped=%d last_id=%d",
|
||
run_id,
|
||
total_batches,
|
||
len(batch_rows),
|
||
batch_inserted,
|
||
batch_skipped,
|
||
last_id,
|
||
)
|
||
|
||
if len(batch_rows) < batch_size:
|
||
break # Last partial batch — no more rows
|
||
|
||
runs_mod.mark_done(db, run_id, counters)
|
||
logger.info(
|
||
"rosreestr_dkp_import run_id=%d done: "
|
||
"total_fetched=%d inserted=%d skipped=%d batches=%d",
|
||
run_id,
|
||
counters["rows_fetched"],
|
||
counters["rows_inserted"],
|
||
counters["rows_skipped"],
|
||
total_batches,
|
||
)
|
||
|
||
except Exception as exc:
|
||
logger.exception("rosreestr_dkp_import run_id=%d failed at last_id=%d", run_id, last_id)
|
||
runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters)
|
||
raise
|
||
|
||
|
||
def get_due_schedules(db: Session) -> list[dict[str, Any]]:
|
||
"""SELECT scrape_schedules WHERE enabled AND (next_run_at IS NULL OR next_run_at <= NOW())."""
|
||
rows = (
|
||
db.execute(
|
||
text(
|
||
"""
|
||
SELECT id, source, enabled, window_start_hour, window_end_hour,
|
||
default_params, last_run_id, last_run_at, next_run_at
|
||
FROM scrape_schedules
|
||
WHERE enabled = true
|
||
AND (next_run_at IS NULL OR next_run_at <= NOW())
|
||
"""
|
||
),
|
||
)
|
||
.mappings()
|
||
.all()
|
||
)
|
||
return [dict(r) for r in rows]
|
||
|
||
|
||
async def scheduler_loop() -> None:
|
||
"""Бесконечный async loop — tick каждые SCHEDULER_TICK_SEC секунд."""
|
||
logger.info("scheduler: started (tick=%ds)", SCHEDULER_TICK_SEC)
|
||
# Initial sleep 30s чтобы дать FastAPI startup завершиться
|
||
await asyncio.sleep(30)
|
||
while True:
|
||
try:
|
||
db = SessionLocal()
|
||
try:
|
||
# Reap zombies first
|
||
reap_zombies(db)
|
||
# Process due schedules
|
||
due = get_due_schedules(db)
|
||
for sch in due:
|
||
source = sch["source"]
|
||
if source == "avito_city_sweep":
|
||
await trigger_avito_city_sweep_run(db, sch)
|
||
elif source == "yandex_city_sweep":
|
||
await trigger_yandex_city_sweep_run(db, sch)
|
||
elif source == "n1_city_sweep":
|
||
await trigger_n1_city_sweep_run(db, sch)
|
||
elif source == "cian_history_backfill":
|
||
await trigger_cian_backfill_run(db, sch)
|
||
elif source == "rosreestr_dkp_import":
|
||
await trigger_rosreestr_dkp_run(db, sch)
|
||
elif source == "listing_source_snapshot":
|
||
await trigger_listing_source_snapshot_run(db, sch)
|
||
elif source == "asking_to_sold_ratio_refresh":
|
||
await trigger_asking_to_sold_ratio_run(db, sch)
|
||
elif source == "refresh_search_matview":
|
||
await trigger_refresh_search_matview_run(db, sch)
|
||
else:
|
||
logger.warning("scheduler: unknown source=%s, skip", source)
|
||
finally:
|
||
db.close()
|
||
except Exception:
|
||
logger.exception("scheduler: tick failed")
|
||
await asyncio.sleep(SCHEDULER_TICK_SEC)
|