backend/app/api/v1/parcels.py

@@ -150,13 +150,6 @@ NOISE_L_BASE: dict[str, float] = {
 }
 
 
-def _wind_label(deg: float) -> str:
-    """Перевести угол направления ветра (0-360) в 8-позиционную розу на русском."""
-    rose = ["Север", "С-В", "Восток", "Ю-В", "Юг", "Ю-З", "Запад", "С-З"]
-    idx = round(deg / 45) % 8
-    return rose[idx]
-
-
 # Координаты центра ЕКБ — Площадь 1905 года
 EKB_CENTER_LAT: float = 56.838011
 EKB_CENTER_LON: float = 60.597474

backend/app/schemas/parcel.py

@@ -6,29 +6,6 @@ from pydantic import BaseModel, ConfigDict, Field
 # ── #105 Phase 5: Recent permits schemas ──────────────────────────────────────
 
 
-class RecentPermit(BaseModel):
-    """Одно строительное разрешение (РНС или РВЭ) из ekburg_construction_permits."""
-
-    permit_type: str
-    permit_number: str
-    issue_date: str | None
-    developer_name: str | None
-    developer_inn: str | None
-    object_name: str | None
-    object_type: str | None
-    construction_address: str | None
-    total_area_sqm: float | None
-
-
-class PermitsSummary(BaseModel):
-    """Агрегированная сводка по разрешениям в квартале."""
-
-    rns_count: int
-    rve_count: int
-    rns_total_area_sqm: float
-    by_developer: list[dict[str, Any]]
-
-
 # ── Connection points schemas (issue #115) ────────────────────────────────────
 
 
@@ -564,16 +541,6 @@ class DeveloperAttributionResult(BaseModel):
 # ── Layout analysis (Issue #113) ───────────────────────────────────────────
 
 
-class LayoutSignature(BaseModel):
-    """Минимальная сигнатура планировки = (room_bucket, area_bin).
-
-    Phase 2.1: layout_type/balcony_count в БД нет, ждут B2B Объектив (#52).
-    """
-
-    room_bucket: Literal["studio", "1", "2", "3", "4+"]
-    area_bin: Literal["<25", "25-40", "40-60", "60-80", "80-100", "100+"]
-
-
 class BestLayoutsRequest(BaseModel):
     """Параметры запроса top-планировок в радиусе вокруг участка."""
 

backend/app/services/scrapers/documents.py

@@ -181,13 +181,3 @@ def upsert_documents(db: Session, obj_id: int, docs: list[dict[str, Any]]) -> tu
 
 # ── Stub for future Celery download task ──────────────────────────────────────
 
-
-def download_document_stub(obj_id: int, doc_id: int, file_url: str) -> None:
-    """Placeholder для будущей Celery-задачи download_domrf_documents.
-
-    Скачивает PDF, сохраняет в data/raw/domrf_docs/{obj_id}/{filename},
-    обновляет domrf_kn_documents.local_path + downloaded_at.
-
-    Реализация — отдельный PR (future task).
-    """
-    raise NotImplementedError("PDF download not implemented in 22i. See issue #297 future PR.")

backend/app/services/scrapers/domrf_kn.py

@@ -1807,55 +1807,6 @@ async def fetch_flats_for_object(sess: BrowserSession, obj_id: int) -> list[dict
     return _flatten_table(payload)
 
 
-async def probe_endpoint(
-    url_or_path: str,
-    region_code: int = 66,
-    headed: bool = False,
-    load_state: str | None = None,
-) -> dict[str, Any]:
-    """Quick one-shot probe through a real Playwright session for debugging.
-
-    Returns {status, content_type, body_preview, full_url}. Use to verify that
-    a URL works in our session before/after a failing scraper run.
-    """
-    if url_or_path.startswith("http"):
-        # split into path+query for BrowserSession.get_json
-        from urllib.parse import parse_qsl, urlsplit
-
-        s = urlsplit(url_or_path)
-        path = s.path
-        params = dict(parse_qsl(s.query, keep_blank_values=True))
-    else:
-        path = url_or_path.split("?", 1)[0]
-        params = {}
-        if "?" in url_or_path:
-            from urllib.parse import parse_qsl
-
-            params = dict(parse_qsl(url_or_path.split("?", 1)[1], keep_blank_values=True))
-
-    async with BrowserSession(
-        region_code=region_code, headed=headed, load_state=load_state
-    ) as sess:
-        try:
-            payload = await sess.get_json(path, params)
-            body = json.dumps(payload, ensure_ascii=False)[:2000]
-            qs = "&".join(f"{k}={v}" for k, v in params.items())
-            full_url = f"{BASE_URL}{path}" + (f"?{qs}" if qs else "")
-            return {
-                "status": 200,
-                "content_type": "application/json",
-                "body_preview": body,
-                "full_url": full_url,
-            }
-        except Exception as e:
-            return {
-                "status": _http_status_of(e),
-                "content_type": "error",
-                "body_preview": str(e)[:2000],
-                "full_url": f"{BASE_URL}{path}",
-            }
-
-
 def log_progress(
     db: Session,
     run_id: int,

backend/app/services/scrapers/nspd_lite.py

@@ -133,21 +133,6 @@ def fetch_geoportal(
         raise NspdLiteError(f"Network error: {e}") from e
 
 
-def fetch_quarter(quarter_cad_num: str, **kwargs) -> dict[str, Any]:
-    """Конкретный кадастровый квартал (typed wrapper)."""
-    return fetch_geoportal(quarter_cad_num, thematic_id=THEMATIC["quarter"], **kwargs)
-
-
-def fetch_parcel(parcel_cad_num: str, **kwargs) -> dict[str, Any]:
-    """Конкретный участок ЗУ (typed wrapper)."""
-    return fetch_geoportal(parcel_cad_num, thematic_id=THEMATIC["parcel"], **kwargs)
-
-
-def fetch_building(building_cad_num: str, **kwargs) -> dict[str, Any]:
-    """Конкретное здание ОКС (typed wrapper)."""
-    return fetch_geoportal(building_cad_num, thematic_id=THEMATIC["building"], **kwargs)
-
-
 # ── Bulk через rosreestr2coord library ──────────────────────────────────────
 
 

backend/app/services/scrapers/objective.py

@@ -28,7 +28,7 @@ import logging
 import time
 from collections.abc import Iterator
 from contextlib import contextmanager
-from datetime import date, datetime
+from datetime import date
 from typing import Any
 
 import httpx
@@ -557,36 +557,3 @@ class ObjectiveClient:
 
 # ── удобный one-shot helper ─────────────────────────────────────────────────
 
-
-def fetch_with_raw_log(
-    fn_name: str,
-    *,
-    save_dir: str | None = None,
-    **kwargs: Any,
-) -> tuple[Any, str | None]:
-    """Вызывает client.<fn_name>(**kwargs), сохраняет raw JSON в файл если задан save_dir.
-    Возвращает (data, file_path or None). Удобно для smoke-сценариев и backfill."""
-    from pathlib import Path
-
-    client = ObjectiveClient()
-    try:
-        method = getattr(client, fn_name)
-        data = method(**kwargs)
-    finally:
-        client.close()
-
-    file_path: str | None = None
-    if save_dir:
-        Path(save_dir).mkdir(parents=True, exist_ok=True)
-        ts = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
-        slug = (
-            "_".join(str(v) for k, v in kwargs.items() if k in ("group_name", "complex_name") and v)
-            or "all"
-        )
-        slug = slug.replace(" ", "_").replace("/", "_")
-        file_path = f"{save_dir}/objective_{fn_name}_{slug}_{ts}.json"
-        Path(file_path).write_text(
-            json.dumps(data, ensure_ascii=False, indent=2),
-            encoding="utf-8",
-        )
-    return data, file_path

backend/app/services/site_finder/ird_analyze.py

@@ -61,18 +61,6 @@ _KRT_AT_SQL = text("""
 """)
 
 
-def _functional_zone_at(
-    client: EKBGeoportalClient, lon: float, lat: float
-) -> dict[str, Any] | None:
-    """Функц.зона генплана (#1058) в точке — properties первой WFS-фичи; None при сбое/пусто."""
-    try:
-        feats = client.features_at_point("functional_zone", lon, lat)
-    except Exception as exc:  # внешний геопортал, graceful
-        logger.warning("ird_analyze: functional_zone WFS failed: %s", exc)
-        return None
-    return feats[0].properties if feats else None
-
-
 def _krt_at(db: Session, lon: float, lat: float) -> list[dict[str, Any]]:
     """КРТ-объекты (#1060/#1130) в точке — из БД ekb_krt_geometry.
 

backend/app/services/site_finder/krt_lookup.py

@@ -15,7 +15,6 @@ Graceful: нет таблицы / пустой WKT / исключение БД
 from __future__ import annotations
 
 import logging
-import re
 from typing import Any
 
 from sqlalchemy import text
@@ -98,12 +97,6 @@ _PLANNING_NO_MATCH_SQL = text(
 )
 
 
-def _extract_paga_num(full_name: str) -> str | None:
-    """Извлечь номер ПАГЕ из заголовка planning_project.full_name."""
-    m = re.search(r"ПАГЕ\s*№\s*(\d+)\s+от", full_name)
-    return m.group(1) if m else None
-
-
 def parcel_krt_requisites(db: Session, parcel_wkt: str | None) -> list[dict[str, Any]]:
     """КРТ-реквизиты для участка: застройщик / цена / градпотенциал.
 

backend/app/services/site_finder/utility_infrastructure_loader.py

@@ -472,18 +472,6 @@ def _upsert_elements(elements: list[dict]) -> dict[str, int]:
     return result_dict
 
 
-async def load_utility_infrastructure(db: Session | None = None) -> dict[str, int]:
-    """Async-обёртка для прямого вызова из async-контекста (#1746 spec).
-
-    Тянет Overpass и UPSERT-ит в таблицу. ``db`` принимается для сигнатурной
-    совместимости со спекой, но UPSERT идёт через собственную SessionLocal
-    (как noise/poi sync) — Overpass-fetch async, запись в БД sync. Возвращает
-    те же per-kind counts, что ``sync_utility_infrastructure_to_db``.
-    """
-    elements = await fetch_overpass_utility()
-    return _upsert_elements(elements)
-
-
 # Junk-значения OSM tags, которые не несут смысла для попапа.
 _JUNK_TAG_VALUES = frozenset({"", "yes", "no", "fixme", "unknown", "-"})
 

backend/app/workers/beat_schedule.py

@@ -150,69 +150,6 @@ def _build_beat_schedule_from_db() -> dict:
     return schedule
 
 
-def _build_beat_schedule_fallback() -> dict:
-    """Fallback-расписание из env vars — используется если БД недоступна на старте."""
-    logger.warning("beat_schedule: строим fallback из env vars")
-    schedule: dict = {}
-
-    # scrape_kn
-    try:
-        for rc in _default_regions():
-            schedule[f"kn-region-{rc}"] = {
-                "task": "tasks.scrape_kn.scrape_kn_region",
-                "schedule": _parse_cron(settings.scrape_kn_cron),
-                "args": [rc, None],
-            }
-    except Exception as e:
-        logger.warning("beat_schedule fallback: scrape_kn failed: %s", e)
-
-    # objective_sync
-    try:
-        from app.core.db import SessionLocal
-        from app.services.objective_sync_config import get_cron_schedule_safe
-
-        cron_str = get_cron_schedule_safe(SessionLocal)
-    except Exception:
-        cron_str = settings.objective_sync_cron
-
-    try:
-        schedule["objective-sync"] = {
-            "task": "tasks.scrape_objective.sync_all_groups",
-            "schedule": _parse_cron(cron_str),
-            "kwargs": {"triggered_by": "beat"},
-        }
-    except Exception as e:
-        logger.warning("beat_schedule fallback: objective_sync failed: %s", e)
-
-    # refresh ekb-districts medians — ежемесячно 5-го числа в 04:00 МСК
-    try:
-        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},
-        }
-    except Exception as e:
-        logger.warning("beat_schedule fallback: refresh_analytics failed: %s", e)
-
-    # КРТ-площадки ЕКБ (#1060): реестр обновляется редко (квартальные решения ПАГЕ),
-    # раз в неделю достаточно. Среда 05:00 МСК — не пересекается с permits (вт/ср 02:00).
-    schedule["ekb-krt-sync-weekly"] = {
-        "task": "tasks.ekb_krt_sync.sync_krt_sites",
-        "schedule": _parse_cron("0 5 * * wed"),
-        "options": {"queue": "celery"},
-    }
-
-    # РИАСУРТ Свердл (#108): ежеквартально (1-е число янв/апр/июл/окт, 04:00 МСК).
-    # NB: harvest_all_* на ПЛЕЙСХОЛДЕР-bbox 5 МО (riasurt_sverdl_harvest.MO_BBOXES TODO).
-    schedule["riasurt-sverdl-harvest-quarterly"] = {
-        "task": "tasks.riasurt_sverdl_harvest.harvest_all_riasurt_sverdl",
-        "schedule": _parse_cron("0 4 1 1,4,7,10 *"),
-        "options": {"queue": "celery"},
-    }
-
-    return schedule
-
-
 def build_beat_schedule() -> dict:
     """Строит beat_schedule из DB (job_settings) + добавляет hardcoded entries.