From f80a2c1d0580d85b0ad227615d4fd5aed54ebf8b Mon Sep 17 00:00:00 2001 From: lekss361 Date: Thu, 30 Apr 2026 23:16:11 +0300 Subject: [PATCH] =?UTF-8?q?fix(scrape):=20SQLAlchemy=20text()=20=D0=BF?= =?UTF-8?q?=D0=B0=D1=80=D1=81=D0=B8=D0=BB=20':00:'=20regex=20=D0=BA=D0=B0?= =?UTF-8?q?=D0=BA=20named-param?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit В nspd_kn.py:get_pending_cads и admin_scrape.py:nspd_coverage regex-фильтры quarter_cad_number !~ '^00:00:' и !~ ':0000000$' падали с psycopg.errors. DataError — двоеточие в text() трактуется как ':param'. Заменил на NOT LIKE с bindparams (быстрее regex и без issue с парсингом). Эффект на проде: - /api/v1/admin/scrape/nspd/coverage возвращал 500 → теперь 200 - Кнопка «Запустить sweep» больше не дизейбленa (pending=964 теперь видно) - Сам scraper тоже бы валился на этом же баге при beat-запуске --- backend/app/api/v1/admin_scrape.py | 9 ++- backend/app/services/analytics_refresh.py | 66 +++++++++++++++++++ backend/app/services/scrapers/nspd_kn.py | 13 ++-- backend/app/workers/celery_app.py | 10 +++ .../app/workers/tasks/refresh_analytics.py | 49 ++++++++++++++ data/sql/67_refresh_ekb_districts_median.sql | 54 +++++++++++++++ 6 files changed, 194 insertions(+), 7 deletions(-) create mode 100644 backend/app/services/analytics_refresh.py create mode 100644 backend/app/workers/tasks/refresh_analytics.py create mode 100644 data/sql/67_refresh_ekb_districts_median.sql diff --git a/backend/app/api/v1/admin_scrape.py b/backend/app/api/v1/admin_scrape.py index 363ac731..652afa2a 100644 --- a/backend/app/api/v1/admin_scrape.py +++ b/backend/app/api/v1/admin_scrape.py @@ -523,6 +523,9 @@ def nspd_coverage( """Покрытие cad_quarters_geom относительно ДДУ-сделок rosreestr. Полезно перед запуском sweep — видишь сколько ещё осталось скрейпить.""" _check_token(x_admin_token) + # ВАЖНО: regex с двоеточиями (':00:', ':0000000') ломают SQLAlchemy text() — + # двоеточие парсится как named-parameter prefix. Поэтому паттерны передаём + # через bindparams. NOT LIKE безопасен и быстрее regex. row = ( db.execute( text( @@ -534,8 +537,8 @@ def nspd_coverage( AND doc_type = 'ДДУ' AND realestate_type_code = '002001003000' AND quarter_cad_number IS NOT NULL - AND quarter_cad_number !~ '^00:00:' - AND quarter_cad_number !~ ':0000000$' + AND quarter_cad_number NOT LIKE :bad_prefix + AND quarter_cad_number NOT LIKE :bad_suffix ), done AS (SELECT cad_number AS cad FROM cad_quarters_geom) SELECT @@ -546,7 +549,7 @@ def nspd_coverage( (SELECT COUNT(*) FROM cad_buildings) AS buildings_total """ ), - {"rc": region_code}, + {"rc": region_code, "bad_prefix": "00:00:%", "bad_suffix": "%:0000000"}, ) .mappings() .first() diff --git a/backend/app/services/analytics_refresh.py b/backend/app/services/analytics_refresh.py new file mode 100644 index 00000000..37172ce8 --- /dev/null +++ b/backend/app/services/analytics_refresh.py @@ -0,0 +1,66 @@ +"""Регулярные refresh-задачи для аналитического слоя. + +Сейчас здесь: +- refresh_ekb_districts_median: пересчёт median/mean ₽/м² 8 районов ЕКБ из + ДДУ-сделок rosreestr_deals (окно 24 мес). + +Запускается через Celery beat (ежемесячно). См. workers/tasks/refresh_analytics.py. +""" + +from __future__ import annotations + +import logging +from typing import Any + +from sqlalchemy import text +from sqlalchemy.orm import Session + +logger = logging.getLogger(__name__) + + +def refresh_ekb_districts_median( + db: Session, + *, + window_months: int = 24, + min_deals: int = 50, +) -> list[dict[str, Any]]: + """Пересчитать ekb_districts.median_price_per_m2 + mean_price_per_m2 из + rosreestr ДДУ-сделок последних `window_months` месяцев. + + Использует SQL-функцию `refresh_ekb_districts_median(int, int)` из + миграции 67_refresh_ekb_districts_median.sql. Возвращает список + обновлённых районов с фактически использованными метриками. + + Не обновляет района с deals < min_deals (статистически шумно). + Не трогает 'не определён' (служебная категория). + """ + rows = ( + db.execute( + text( + """ + SELECT district_name, deals_used, median_pm, mean_pm + FROM refresh_ekb_districts_median(:wm, :md) + """ + ), + {"wm": window_months, "md": min_deals}, + ) + .mappings() + .all() + ) + db.commit() + out = [ + { + "district_name": r["district_name"], + "deals_used": int(r["deals_used"]), + "median_pm": float(r["median_pm"]) if r["median_pm"] is not None else None, + "mean_pm": float(r["mean_pm"]) if r["mean_pm"] is not None else None, + } + for r in rows + ] + logger.info( + "refresh_ekb_districts_median: window=%dmo min_deals=%d updated=%d districts", + window_months, + min_deals, + len(out), + ) + return out diff --git a/backend/app/services/scrapers/nspd_kn.py b/backend/app/services/scrapers/nspd_kn.py index 694b75f7..89f97f01 100644 --- a/backend/app/services/scrapers/nspd_kn.py +++ b/backend/app/services/scrapers/nspd_kn.py @@ -164,7 +164,12 @@ def _to_date(x: Any) -> str | None: def get_pending_cads(db: Session, region_code: int) -> list[str]: - """Cad-кварталы с ДДУ-сделками в регионе минус уже scraped.""" + """Cad-кварталы с ДДУ-сделками в регионе минус уже scraped. + + NOTE: regex с двоеточиями ('!~ ":0000000$"') ломают SQLAlchemy text() — + двоеточие парсится как named-parameter prefix. Используем NOT LIKE через + bindparam (быстрее regex и безопасно). + """ rows = db.execute( text( """ @@ -175,11 +180,11 @@ def get_pending_cads(db: Session, region_code: int) -> list[str]: AND realestate_type_code = '002001003000' AND quarter_cad_number IS NOT NULL AND quarter_cad_number <> '' - AND quarter_cad_number !~ '^00:00:' - AND quarter_cad_number !~ ':0000000$' + AND quarter_cad_number NOT LIKE :bad_prefix + AND quarter_cad_number NOT LIKE :bad_suffix """ ), - {"rc": region_code}, + {"rc": region_code, "bad_prefix": "00:00:%", "bad_suffix": "%:0000000"}, ).all() pending = {r[0] for r in rows} done = {r[0] for r in db.execute(text("SELECT cad_number FROM cad_quarters_geom")).all()} diff --git a/backend/app/workers/celery_app.py b/backend/app/workers/celery_app.py index e3785cd4..d9adcf8f 100644 --- a/backend/app/workers/celery_app.py +++ b/backend/app/workers/celery_app.py @@ -41,6 +41,7 @@ celery_app = Celery( include=[ "app.workers.tasks.scrape_kn", "app.workers.tasks.scrape_nspd", + "app.workers.tasks.refresh_analytics", ], ) celery_app.conf.timezone = "Europe/Moscow" @@ -75,6 +76,15 @@ celery_app.conf.beat_schedule.update( } ) +# Refresh ekb_districts медианы — ежемесячно 5-го числа в 04:00 МСК +# (после публикации новых rosreestr-кварталов и NSPD beat-cycle). +# Лёгкая задача (1-2с на 8 районов), без locks. +celery_app.conf.beat_schedule["refresh-ekb-districts-medians"] = { + "task": "tasks.refresh_analytics.refresh_ekb_districts_medians", + "schedule": _parse_cron("0 4 5 * *"), + "kwargs": {"window_months": 24, "min_deals": 50}, +} + @worker_ready.connect def _resume_zombie_runs(sender=None, **_kwargs) -> None: diff --git a/backend/app/workers/tasks/refresh_analytics.py b/backend/app/workers/tasks/refresh_analytics.py new file mode 100644 index 00000000..5864fb37 --- /dev/null +++ b/backend/app/workers/tasks/refresh_analytics.py @@ -0,0 +1,49 @@ +"""Celery tasks для регулярного refresh аналитических метрик. + +Расписание задаётся в celery_app.beat_schedule. Сейчас: +- refresh_ekb_districts_medians — раз в месяц 5-го числа в 04:00 МСК + (после публикации последнего rosreestr-квартала + NSPD beat-cycle). +""" + +from __future__ import annotations + +import logging +from typing import Any + +from app.core.db import SessionLocal +from app.services.analytics_refresh import refresh_ekb_districts_median +from app.workers.celery_app import celery_app + +logger = logging.getLogger(__name__) + + +@celery_app.task( + bind=True, + name="tasks.refresh_analytics.refresh_ekb_districts_medians", + max_retries=2, +) +def refresh_ekb_districts_medians( + self: Any, + window_months: int = 24, + min_deals: int = 50, +) -> dict[str, Any]: + """Refresh ekb_districts.median_price_per_m2 + mean_price_per_m2. + + Лёгкая задача (1-2 секунды на 8 районов), не требует locks. + Запускается ежемесячно. Возвращает результат для аудита (что обновлено). + """ + db = SessionLocal() + try: + rows = refresh_ekb_districts_median(db, window_months=window_months, min_deals=min_deals) + logger.info("refreshed %d districts", len(rows)) + return { + "updated_count": len(rows), + "districts": rows, + "window_months": window_months, + "min_deals": min_deals, + } + except Exception as e: + logger.exception("refresh_ekb_districts_medians failed: %s", e) + raise + finally: + db.close() diff --git a/data/sql/67_refresh_ekb_districts_median.sql b/data/sql/67_refresh_ekb_districts_median.sql new file mode 100644 index 00000000..5dc530c0 --- /dev/null +++ b/data/sql/67_refresh_ekb_districts_median.sql @@ -0,0 +1,54 @@ +-- Пересчёт ekb_districts.median_price_per_m2 + mean_price_per_m2 из rosreestr_deals. +-- +-- Зачем: для половины районов ЕКБ были NULL (импорт Антона из SQLite не покрыл), +-- из-за чего recommend_mix давал district_factor=1.0 без учёта района и +-- cadastre_vs_market_pct = NULL (нет district_median для сравнения с кадастром). +-- Старые значения для Октябрьского/Орджоникидзевского (191K/205K) выглядят как +-- yandex/domrf-листинг, не реальные сделки — перезаписываем для консистентности. +-- +-- Исключаем 'не определён' (служебная категория, не настоящий район). +-- +-- Идемпотентно — можно перезапускать. Окно 24 мес даёт устойчивые медианы. +-- Если в районе <50 сделок за 24 мес — оставляем NULL (warning в recommend_mix). + +CREATE OR REPLACE FUNCTION refresh_ekb_districts_median(window_months int DEFAULT 24, min_deals int DEFAULT 50) +RETURNS TABLE(district_name text, deals_used bigint, median_pm numeric, mean_pm numeric) +LANGUAGE plpgsql AS $$ +BEGIN + RETURN QUERY + WITH per_district AS ( + SELECT d.district_name, + COUNT(*)::bigint AS deals, + PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY rd.price_per_sqm)::numeric AS p_median, + AVG(rd.price_per_sqm)::numeric AS p_mean + FROM rosreestr_deals rd + JOIN cad_quarters_geom cq ON cq.cad_number = rd.quarter_cad_number + JOIN ekb_districts_geom d ON ST_Intersects(d.geom, cq.geom) + WHERE rd.region_code = 66 + AND rd.doc_type = 'ДДУ' + AND rd.realestate_type_code = '002001003000' + AND rd.area BETWEEN 15 AND 200 + AND rd.price_per_sqm BETWEEN 30000 AND 1000000 + AND rd.period_start_date >= NOW() - (window_months || ' months')::interval + GROUP BY d.district_name + ), + upd AS ( + UPDATE ekb_districts e + SET median_price_per_m2 = ROUND(pd.p_median, 0), + mean_price_per_m2 = ROUND(pd.p_mean, 0) + FROM per_district pd + WHERE e.district_name = pd.district_name + AND e.district_name <> 'не определён' + AND pd.deals >= min_deals + RETURNING e.district_name, pd.deals, pd.p_median, pd.p_mean + ) + SELECT u.district_name::text, u.deals, ROUND(u.p_median, 0), ROUND(u.p_mean, 0) + FROM upd u + ORDER BY u.district_name; +END; +$$; + +COMMENT ON FUNCTION refresh_ekb_districts_median(int, int) IS + 'Refresh ekb_districts.median_price_per_m2 + mean_price_per_m2 из ДДУ-сделок rosreestr_deals ' + 'через spatial-join cad_quarters_geom + ekb_districts_geom. Окно по умолчанию 24 мес, ' + 'минимум 50 сделок на район. Запускается Celery beat раз в месяц.';