752cb1044b
All checks were successful
CI Trade-In / changes (pull_request) Successful in 7s
CI / changes (pull_request) Successful in 7s
CI Trade-In / frontend-checks (pull_request) Has been skipped
CI / backend-tests (pull_request) Has been skipped
CI / frontend-tests (pull_request) Has been skipped
CI / openapi-codegen-check (pull_request) Has been skipped
CI Trade-In / backend-tests (pull_request) Successful in 51s
- scheduler.py import_rosreestr_dkp: снят фильтр city ILIKE '%катеринбург%', address из реального city источника, deals.region_code + новая deals.city заполняются (было: хардкод 'Екатеринбург,' + region_code NULL) - migration 177: deals.city + индекс + бэкфилл существующих EKB-строк region_code=66/city - guard city IS NOT NULL → address не NULL для ~15 null-city строк источника - sync deploy/import-rosreestr.sh (ops-fallback) под тот же oblast-scope - split dedup-теста: 077 (историческая) хранит EKB-фильтр, живой импорт — нет Открывает +47183 не-ЕКБ сделок region 66, уже сидящих в источнике, ранее резавшихся на импорте.
432 lines
20 KiB
Python
432 lines
20 KiB
Python
"""Shared scheduler helpers (#2397 Part C — legacy scheduler_loop removed).
|
||
|
||
История: до #2192 этот модуль нёс полный in-app asyncio scheduler (`scheduler_loop`,
|
||
27-веточный `if/elif` dispatch, `trigger_*_run` per-source launchers, `get_due_schedules`,
|
||
`reap_zombies`, `_claim_run` advisory-lock claim, `_spawn_tracked`/`_drain_inflight`
|
||
graceful-drain machinery). #2192 представил kit-путь (`scraper_kit.orchestration.scheduler`
|
||
+ `app.services.product_handlers`) как ship-dark alternative; прод переключился на него
|
||
(`USE_KIT_SCHEDULER=true` в tradein-scraper, см. docker-compose.prod.yml). #2397 Part C
|
||
убрал legacy `scheduler_loop` и все `trigger_*`/dispatch-функции — kit теперь единственный
|
||
scheduling-путь (`app/scheduler_main.py` безусловно запускает `_run_kit_scheduler()`).
|
||
|
||
Что осталось в этом модуле — НЕ scheduler-loop, а функции с живыми потребителями вне
|
||
удалённой machinery:
|
||
- `compute_next_run_at` — читается admin.py (операторский предпросмотр "next run").
|
||
- `has_running_run` — читается admin.py (UI-индикатор "уже бежит").
|
||
- `import_rosreestr_dkp` — job-тело, вызываемое kit-handler'ом
|
||
product_handlers._job_rosreestr_dkp (lazy import).
|
||
- `_execute_cian_backfill` — job-тело, вызываемое kit-handler'ом
|
||
product_handlers._job_cian_history_backfill (lazy import).
|
||
|
||
Zombie-reap, advisory-lock claim и tick-loop теперь целиком в
|
||
`scraper_kit/orchestration/scheduler.py` (см. test_scraper_kit_scheduler_parity.py).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import random
|
||
from datetime import UTC, datetime, time, timedelta
|
||
from typing import Any
|
||
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.shutdown import shutdown_requested
|
||
from app.services import scrape_runs as runs_mod
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def compute_next_run_at(
|
||
window_start_hour: int,
|
||
window_end_hour: int,
|
||
*,
|
||
now: datetime | None = None,
|
||
interval_days: int = 1,
|
||
) -> datetime:
|
||
"""Pick random datetime в window [start, end) UTC, через interval_days суток после now.
|
||
|
||
interval_days задаёт каденс источника: 1 (default) = daily (back-compat), 7 = weekly.
|
||
Берётся из schedule.default_params["interval_days"] вызывающим кодом; отсутствие ключа
|
||
→ 1 → прежнее ежедневное поведение.
|
||
|
||
Если 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)
|
||
interval_days = max(1, int(interval_days))
|
||
# Целевая дата = now + interval_days суток (interval_days=1 → завтра, как раньше).
|
||
target = (now + timedelta(days=interval_days)).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(target, 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:
|
||
# interval_days=1: текущая дата (если окно ещё не наступило сегодня) или next day.
|
||
# interval_days>1: всегда целевая дата (стаггер на N суток вперёд).
|
||
today_ok = interval_days == 1 and now.hour < window_start_hour
|
||
base_date = now.date() if today_ok else target
|
||
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(target, 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
|
||
|
||
|
||
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
|
||
|
||
|
||
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).
|
||
|
||
Область покрытия: вся Свердловская область (region_code=66), не только Екатеринбург —
|
||
прежний ILIKE-фильтр по подстроке города (ограничивавший импорт одним Екатеринбургом)
|
||
снят (Mera trade-in расширяется на весь регион, unlocks +47183 сделок вне ЕКБ уже
|
||
сидящих в source foreign table). address и deals.city строятся из реального city
|
||
источника (не хардкод "Екатеринбург"), deals.region_code заполняется из строки
|
||
источника (= 66 при текущем фильтре).
|
||
|
||
Фильтры (совпадают с import-rosreestr.sh + Fix_Rosreestr_Dkp_Filter_May24):
|
||
- region_code = 66 (вся Свердловская область, все города)
|
||
- city IS NOT NULL AND trim(city) != '' (непустой город → корректный address)
|
||
- 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:
|
||
cancelled = runs_mod.is_cancelled(db, run_id)
|
||
if cancelled or shutdown_requested():
|
||
if cancelled:
|
||
# User-cancel: семантика без изменений — mark_cancelled.
|
||
logger.info("rosreestr_dkp_import run_id=%d: cancelled by user", run_id)
|
||
runs_mod.mark_cancelled(db, run_id)
|
||
return
|
||
# #1182 Phase 2: кооперативный SIGTERM-drain (деплой recreate scraper).
|
||
# Это НЕ user-cancel → mark_done (partial), не mark_cancelled. Курсор
|
||
# уже зафиксирован update_heartbeat'ом каждый батч; следующий run
|
||
# пере-сканирует с id=0 (INSERT ... ON CONFLICT DO NOTHING идемпотентен)
|
||
# → уже вставленные строки пропускаются, остаток до-импортируется.
|
||
# mark_done выводит run из 'running' → reap_zombies его не тронет.
|
||
logger.info(
|
||
"rosreestr_dkp_import run_id=%d: SIGTERM-drain — committing partial "
|
||
"(last_id=%d, batches=%d) and exiting",
|
||
run_id,
|
||
last_id,
|
||
total_batches,
|
||
)
|
||
runs_mod.update_heartbeat(db, run_id, counters)
|
||
runs_mod.mark_done(db, run_id, counters)
|
||
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(city) || ', ' || trim(street) AS address,
|
||
region_code,
|
||
trim(city) AS city,
|
||
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(
|
||
-- #1525: '-?[0-9]+' сохраняет знак минус, иначе цоколь/подвал
|
||
-- '-1' импортируется как 1. Для '5/9' по-прежнему берётся 5.
|
||
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 IS NOT NULL AND trim(city) <> ''
|
||
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, region_code, city,
|
||
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(:region_code AS int),
|
||
CAST(:city 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"],
|
||
"region_code": row["region_code"],
|
||
"city": row["city"],
|
||
"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
|