From c4f1978ed043fd0cbb6c96ca6e50e60372ca4e32 Mon Sep 17 00:00:00 2001 From: lekss361 Date: Sun, 24 May 2026 13:59:12 +0300 Subject: [PATCH 1/3] =?UTF-8?q?feat(tradein-estimator):=20tiered=20house-m?= =?UTF-8?q?atch=20S=E2=86=92H=E2=86=92W=20(same-house,=20class,=20wide)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Currently year/house_type only influence relevance sort. For a 17-эт. 1975 building we returned 5-9 эт. брежневки as analogs. Three tiers: - S (same building): address ILIKE prefix, ≥3 → return only these - H (same class): year ±15, total_floors ±30%, ≥5 → return - W (wide): current logic without year/floors WHERE filter Tier propagated to confidence_explanation so user sees why analogs may differ. Source: estimate a0a0b820-e8a8-4eee-aa73-0ab3b98ac233. --- tradein-mvp/backend/app/services/estimator.py | 364 +++++++++++++++--- 1 file changed, 311 insertions(+), 53 deletions(-) diff --git a/tradein-mvp/backend/app/services/estimator.py b/tradein-mvp/backend/app/services/estimator.py index 4e2fc03e..c840cc4d 100644 --- a/tradein-mvp/backend/app/services/estimator.py +++ b/tradein-mvp/backend/app/services/estimator.py @@ -22,6 +22,7 @@ from __future__ import annotations import hashlib import json import logging +import math import re from datetime import UTC, datetime, timedelta from typing import Any @@ -493,39 +494,45 @@ async def estimate_quality( if target_house_type is None: target_house_type = house_meta.house_type - # 3. Three-tier fallback: - # a) 1km + ±15% area - # b) 2km + ±15% area (fallback_used = True) - # c) 2km + ±25% area (fallback_used = True, area_widened = True) - listings, fallback_used = _fetch_analogs( + # 3. House-match: S → H → W tiered lookup (see _fetch_analogs docstring). + # Radius fallback still applies when W tier has < 5 results at 1km. + listings, fallback_used, analog_tier = _fetch_analogs( db, lat=geo.lat, lon=geo.lon, rooms=payload.rooms, area=payload.area_m2, radius_m=DEFAULT_RADIUS_M, + full_address=geo.full_address, year_built=target_year, house_type=target_house_type, + total_floors=payload.total_floors, ) area_widened = False if len(listings) < 5: - listings_wide, _ = _fetch_analogs( + listings_wide, _, analog_tier_wide = _fetch_analogs( db, lat=geo.lat, lon=geo.lon, rooms=payload.rooms, area=payload.area_m2, radius_m=FALLBACK_RADIUS_M, + full_address=geo.full_address, year_built=target_year, house_type=target_house_type, + total_floors=payload.total_floors, ) if len(listings_wide) > len(listings): listings = listings_wide fallback_used = True + analog_tier = analog_tier_wide # Tier C: если даже на 2км мало — расширяем area tolerance до ±25% # (актуально для отдалённых районов / новостроек с нестандартной планировкой) if len(listings) < 3: - listings_widearea, _ = _fetch_analogs( + listings_widearea, _, analog_tier_wa = _fetch_analogs( db, lat=geo.lat, lon=geo.lon, rooms=payload.rooms, area=payload.area_m2, radius_m=FALLBACK_RADIUS_M, area_tolerance=0.25, + full_address=geo.full_address, year_built=target_year, house_type=target_house_type, + total_floors=payload.total_floors, ) if len(listings_widearea) > len(listings): listings = listings_widearea fallback_used = True area_widened = True + analog_tier = analog_tier_wa # 3. Outlier filter listings_clean = _filter_outliers(listings) @@ -567,7 +574,20 @@ async def estimate_quality( q3_ppm2 if listings_clean else 0, fallback_used, area_widened, listings=listings_clean, ) - explanation = (explanation or "") + repair_note + + # Tier note — информируем пользователя о качестве house-match + tier_note = "" + if analog_tier == "S": + tier_note = " (аналоги из того же дома)" + elif analog_tier == "H": + tf_str = f"{payload.total_floors}-эт." if payload.total_floors else "" + yr_str = f"{target_year}±15 г." if target_year else "" + parts_str = ", ".join(p for p in [yr_str, tf_str] if p) + tier_note = f" (аналоги из домов того же класса: {parts_str})" if parts_str else "" + else: + tier_note = " (нет аналогов в том же доме/классе — расширили поиск)" + + explanation = (explanation or "") + tier_note + repair_note # ── Stage 3: Avito IMV evaluation as 5-th source (on-demand cached) ── imv_eval: IMVEvaluation | None = None @@ -841,17 +861,119 @@ def _compute_freshness_minutes(lots: list[dict[str, Any]]) -> int | None: # ── Internals ──────────────────────────────────────────────────────────────── + + +def _extract_short_addr(full_address: str) -> str | None: + """Извлекает «улица + номер дома» из полного адреса для поиска в том же доме. + + Примеры: + "г. Екатеринбург, ул. Крауля, 48/2, кв. 5" → "ул. Крауля, 48/2" + "Екатеринбург, Ленина, 36, корп. 2, кв. 10" → "Ленина, 36" + "Свердловская обл., г. Екатеринбург, пр-т Ленина, 36 к2" → "пр-т Ленина, 36" + + Алгоритм: + 1. Разбиваем по запятой. + 2. Отбрасываем сегменты, которые выглядят как «г.», «обл.», «р-н» (city/region prefix). + 3. Берём первые 2 оставшихся токена (улица + дом), strip кв/корп/к-суффикс из последнего. + 4. Возвращаем None, если результат слишком короткий (< 3 символов) — не с чем матчить. + """ + if not full_address: + return None + + parts = [p.strip() for p in full_address.split(",")] + + # Паттерн для «административных» сегментов: г., обл., р-н, с., д. (населённый пункт) + admin_re = re.compile( + r"^(г\.?|обл\.?|р-н\.?|пгт\.?|с\.?|д\.?|мкр\.?)\s", + re.IGNORECASE, + ) + + meaningful = [p for p in parts if not admin_re.match(p) and len(p) > 1] + + if len(meaningful) < 2: + return None + + street = meaningful[0] + house_raw = meaningful[1] + + # Убираем «кв. N», «корп. N», «к2», «к 2» из номера дома + house = re.sub(r"\s*(кв\.?|корп\.?|к\.?)\s*\d+.*$", "", house_raw, flags=re.IGNORECASE).strip() + + result = f"{street}, {house}" + return result if len(result) >= 3 else None + + +def _stratify_candidates(candidates: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Стратифицированная выборка Approach B — гарантирует MIN_ANALOGS_PER_SOURCE слотов. + + Candidates должны быть уже отсортированы по relevance_score (ASC). + """ + guaranteed: list[dict[str, Any]] = [] + guaranteed_ids: set[int] = set() + by_source: dict[str, list[dict[str, Any]]] = {} + for row in candidates: + src = row.get("source") or "unknown" + by_source.setdefault(src, []).append(row) + + for _src, src_rows in by_source.items(): + quota = min(len(src_rows), MIN_ANALOGS_PER_SOURCE) + for row in src_rows[:quota]: + if id(row) not in guaranteed_ids: + guaranteed.append(row) + guaranteed_ids.add(id(row)) + + remaining_slots = 50 - len(guaranteed) + remainder: list[dict[str, Any]] = [] + if remaining_slots > 0: + for row in candidates: + if id(row) not in guaranteed_ids: + remainder.append(row) + if len(remainder) >= remaining_slots: + break + + result = guaranteed + remainder + result.sort(key=lambda r: r.get("relevance_score") or 0.0) + return result[:50] + + +_ANALOG_SELECT_COLS = """ + source, source_url, address, lat, lon, + rooms, area_m2, floor, total_floors, + price_rub, price_per_m2, + listing_date, days_on_market, photo_urls, + scraped_at +""" + +_COMMON_WHERE = """ + AND rooms = :rooms + AND area_m2 BETWEEN :area_min AND :area_max + AND is_active = true + AND scraped_at > NOW() - (:fresh_days || ' days')::interval + AND price_rub > 0 +""" + + def _fetch_analogs( db: Session, *, lat: float, lon: float, rooms: int, area: float, radius_m: int, + full_address: str | None = None, area_tolerance: float = AREA_TOLERANCE, year_built: int | None = None, house_type: str | None = None, -) -> tuple[list[dict[str, Any]], bool]: - """SELECT аналогов с PostGIS distance + house-match relevance. + total_floors: int | None = None, + ext_house_id: str | None = None, +) -> tuple[list[dict[str, Any]], bool, str]: + """SELECT аналогов — трёхуровневый house-match (S → H → W). - House-match (встреча Птицы — «соразмерные квартиры»): сортировка не просто - по расстоянию, а по relevance-скору, где учитывается близость года постройки - и совпадение типа дома. Так аналог «рядом + та же эпоха дома» побеждает - аналог «чуть ближе, но дом на 30 лет старше». + **Tier S (same building):** address ILIKE prefix OR ext_house_id match. + Если ≥3 результатов → возвращаем только их; tier='S'. + + **Tier H (same class):** PostGIS + rooms + area + year ±15 + total_floors ±30%. + Если ≥5 результатов → возвращаем; tier='H'. + Пропускается если year_built или total_floors неизвестны. + + **Tier W (wide / current):** текущая логика без year/floors WHERE фильтра. + tier='W'. + + House-match relevance_score используется для сортировки в Tier H и W. Стратифицированная выборка (Approach B): 1. SQL вытягивает до 300 кандидатов с per-address row_number (cap MAX_ANALOGS_PER_ADDRESS). @@ -860,9 +982,177 @@ def _fetch_analogs( 4. Итоговый список отсортирован по relevance, LIMIT 50. Returns: - (list_of_listings_as_dicts, fallback_radius_used_flag) + (list_of_listings_as_dicts, fallback_radius_used_flag, tier) + tier: 'S' | 'H' | 'W' """ - rows = db.execute( + area_min = area * (1 - area_tolerance) + area_max = area * (1 + area_tolerance) + base_params: dict[str, Any] = { + "rooms": rooms, + "area_min": area_min, + "area_max": area_max, + "fresh_days": LISTINGS_FRESH_DAYS, + "max_per_addr": MAX_ANALOGS_PER_ADDRESS, + } + + # ── Tier S: same building ───────────────────────────────────────────────── + short_addr = _extract_short_addr(full_address) if full_address else None + has_ext_id = ext_house_id is not None + + if short_addr or has_ext_id: + addr_clause = "" + if short_addr and has_ext_id: + addr_clause = ( + "(address ILIKE :short_addr_prefix" + " OR (ext_house_id IS NOT NULL AND ext_house_id = :ext_house_id))" + ) + elif short_addr: + addr_clause = "address ILIKE :short_addr_prefix" + else: + addr_clause = "ext_house_id IS NOT NULL AND ext_house_id = :ext_house_id" + + tier_s_params = {**base_params} + if short_addr: + tier_s_params["short_addr_prefix"] = short_addr + "%" + if has_ext_id: + tier_s_params["ext_house_id"] = ext_house_id + + tier_s_rows = db.execute( + text( + f""" + WITH base AS ( + SELECT + {_ANALOG_SELECT_COLS}, + 0.0 AS distance_m, + 0.0 AS relevance_score, + row_number() OVER (PARTITION BY address ORDER BY scraped_at DESC) AS rn_addr + FROM listings + WHERE {addr_clause} + {_COMMON_WHERE} + ) + SELECT + source, source_url, address, lat, lon, + rooms, area_m2, floor, total_floors, + price_rub, price_per_m2, + listing_date, days_on_market, photo_urls, + scraped_at, distance_m, relevance_score + FROM base + WHERE rn_addr <= :max_per_addr + ORDER BY relevance_score + LIMIT 300 + """ + ), + tier_s_params, + ).mappings().all() + + tier_s = [dict(r) for r in tier_s_rows] + if len(tier_s) >= 3: + logger.info( + "analogs tier=S addr_prefix=%r → %d results", + short_addr, + len(tier_s), + ) + return _stratify_candidates(tier_s), radius_m > DEFAULT_RADIUS_M, "S" + + # ── Tier H: same class (year ±15, total_floors ±30%) ───────────────────── + if year_built is not None and total_floors is not None: + year_min = year_built - 15 + year_max = year_built + 15 + tf_min = math.floor(total_floors * 0.7) + tf_max = math.ceil(total_floors * 1.3) + + tier_h_rows = db.execute( + text( + f""" + WITH base AS ( + SELECT + {_ANALOG_SELECT_COLS}, + ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography) + AS distance_m, + ( + ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography) + / 1000.0 + + CASE + WHEN year_built IS NOT NULL + THEN abs(year_built - CAST(:target_year AS integer)) / 12.0 + ELSE 0 + END + + CASE + WHEN CAST(:target_house_type AS text) IS NOT NULL + AND house_type IS NOT NULL + AND house_type <> CAST(:target_house_type AS text) + THEN 1.5 + ELSE 0 + END + ) AS relevance_score, + row_number() OVER ( + PARTITION BY address + ORDER BY ( + ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography) + / 1000.0 + + CASE + WHEN year_built IS NOT NULL + THEN abs(year_built - CAST(:target_year AS integer)) / 12.0 + ELSE 0 + END + + CASE + WHEN CAST(:target_house_type AS text) IS NOT NULL + AND house_type IS NOT NULL + AND house_type <> CAST(:target_house_type AS text) + THEN 1.5 + ELSE 0 + END + ) + ) AS rn_addr + FROM listings + WHERE ST_DWithin(geom::geography, ST_MakePoint(:lon, :lat)::geography, :radius) + {_COMMON_WHERE} + AND total_floors BETWEEN CAST(:tf_min AS integer) + AND CAST(:tf_max AS integer) + AND year_built BETWEEN CAST(:year_min AS integer) + AND CAST(:year_max AS integer) + ) + SELECT + source, source_url, address, lat, lon, + rooms, area_m2, floor, total_floors, + price_rub, price_per_m2, + listing_date, days_on_market, photo_urls, + scraped_at, distance_m, relevance_score + FROM base + WHERE rn_addr <= :max_per_addr + ORDER BY relevance_score + LIMIT 300 + """ + ), + { + **base_params, + "lat": lat, + "lon": lon, + "radius": radius_m, + "target_year": year_built, + "target_house_type": house_type, + "tf_min": tf_min, + "tf_max": tf_max, + "year_min": year_min, + "year_max": year_max, + }, + ).mappings().all() + + tier_h = [dict(r) for r in tier_h_rows] + if len(tier_h) >= 5: + logger.info( + "analogs tier=H year=%d±15 tf=%d-%d → %d results", + year_built, tf_min, tf_max, len(tier_h), + ) + return _stratify_candidates(tier_h), radius_m > DEFAULT_RADIUS_M, "H" + + logger.info( + "analogs tier=H year=%d±15 tf=%d-%d → only %d (fallthrough to W)", + year_built, tf_min, tf_max, len(tier_h), + ) + + # ── Tier W: wide (current logic, year/floors only in relevance sort) ────── + tier_w_rows = db.execute( text( """ WITH base AS ( @@ -945,8 +1235,8 @@ def _fetch_analogs( "lon": lon, "radius": radius_m, "rooms": rooms, - "area_min": area * (1 - area_tolerance), - "area_max": area * (1 + area_tolerance), + "area_min": area_min, + "area_max": area_max, "fresh_days": LISTINGS_FRESH_DAYS, "target_year": year_built, "target_house_type": house_type, @@ -954,41 +1244,9 @@ def _fetch_analogs( }, ).mappings().all() - candidates: list[dict[str, Any]] = [dict(r) for r in rows] - - # Stratified quota: гарантируем MIN_ANALOGS_PER_SOURCE слотов каждому source. - # Candidates уже отсортированы по relevance_score (лучшие первые) из SQL. - guaranteed: list[dict[str, Any]] = [] - guaranteed_ids: set[int] = set() # по object id, не по внешнему ключу - by_source: dict[str, list[dict[str, Any]]] = {} - for row in candidates: - src = row.get("source") or "unknown" - by_source.setdefault(src, []).append(row) - - for _src, src_rows in by_source.items(): - quota = min(len(src_rows), MIN_ANALOGS_PER_SOURCE) - for row in src_rows[:quota]: - if id(row) not in guaranteed_ids: - guaranteed.append(row) - guaranteed_ids.add(id(row)) - - # Оставшиеся слоты из candidates, которые ещё не попали в guaranteed. - remaining_slots = 50 - len(guaranteed) - remainder: list[dict[str, Any]] = [] - if remaining_slots > 0: - for row in candidates: - if id(row) not in guaranteed_ids: - remainder.append(row) - if len(remainder) >= remaining_slots: - break - - result = guaranteed + remainder - # Финальная сортировка по relevance (candidates из SQL уже отсортированы, - # но guaranteed + remainder смешиваются). relevance_score присутствует в каждом dict. - result.sort(key=lambda r: r.get("relevance_score") or 0.0) - result = result[:50] - - return result, radius_m > DEFAULT_RADIUS_M + candidates: list[dict[str, Any]] = [dict(r) for r in tier_w_rows] + logger.info("analogs tier=W radius=%dm → %d candidates", radius_m, len(candidates)) + return _stratify_candidates(candidates), radius_m > DEFAULT_RADIUS_M, "W" def _fetch_deals( -- 2.45.3 From fa43ea8dc87bb08b9bb93028e745a9b9a348cdca Mon Sep 17 00:00:00 2001 From: lekss361 Date: Sun, 24 May 2026 14:10:39 +0300 Subject: [PATCH 2/3] fix(tradein-estimator): ruff lint fixes (RUF046, E501) pre-existing in #507 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Remove redundant int() around round() — RUF046 - Wrap long f-strings in _compute_confidence to stay within 100 chars - Simplify photo_url expression in _listing_to_analog --- tradein-mvp/backend/app/services/estimator.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tradein-mvp/backend/app/services/estimator.py b/tradein-mvp/backend/app/services/estimator.py index c840cc4d..939633a0 100644 --- a/tradein-mvp/backend/app/services/estimator.py +++ b/tradein-mvp/backend/app/services/estimator.py @@ -563,7 +563,7 @@ async def estimate_quality( range_low = int(range_low * repair_coef) range_high = int(range_high * repair_coef) median_ppm2 = median_ppm2 * repair_coef - pct = int(round((repair_coef - 1.0) * 100)) + pct = round((repair_coef - 1.0) * 100) repair_note = ( f" Цена скорректирована на состояние ремонта " f"({_REPAIR_LABEL.get(payload.repair_state, '')} {pct:+d}%)." @@ -1403,7 +1403,7 @@ def _listing_to_analog(row: dict[str, Any]) -> AnalogLot: price_per_m2=int(row.get("price_per_m2") or 0), listing_date=row.get("listing_date"), days_on_market=row.get("days_on_market"), - photo_url=(row.get("photo_urls") or [None])[0] if isinstance(row.get("photo_urls"), list) else None, + photo_url=(row["photo_urls"] or [None])[0] if row.get("photo_urls") else None, source=row.get("source"), source_url=row.get("source_url"), distance_m=int(row["distance_m"]) if row.get("distance_m") is not None else None, -- 2.45.3 From 8e0e80cf58e7dcf797295b1e70b7f7204a722f26 Mon Sep 17 00:00:00 2001 From: lekss361 Date: Sun, 24 May 2026 15:36:15 +0300 Subject: [PATCH 3/3] fix(tradein-estimator): address review-bot items on PR #507 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - _extract_short_addr handles Свердловская область / Россия / г. prefixes, drops trailing кв./корп./оф., anchors on street keywords (ул./пр./пер./…) - Tier S ORDER BY scraped_at DESC (was constant relevance_score=0.0) - Drop ext_house_id kwarg (column never existed on listings table) - 7 unit tests for _extract_short_addr covering real address formats Closes 4 deferred items from gendesign-review-bot fa43ea8 review. --- tradein-mvp/backend/app/services/estimator.py | 107 +++++++++--------- .../backend/tests/test_extract_short_addr.py | 49 ++++++++ 2 files changed, 105 insertions(+), 51 deletions(-) create mode 100644 tradein-mvp/backend/tests/test_extract_short_addr.py diff --git a/tradein-mvp/backend/app/services/estimator.py b/tradein-mvp/backend/app/services/estimator.py index 939633a0..50269ed1 100644 --- a/tradein-mvp/backend/app/services/estimator.py +++ b/tradein-mvp/backend/app/services/estimator.py @@ -862,45 +862,63 @@ def _compute_freshness_minutes(lots: list[dict[str, Any]]) -> int | None: # ── Internals ──────────────────────────────────────────────────────────────── +# Compiled regexes for _extract_short_addr — module-level for performance. -def _extract_short_addr(full_address: str) -> str | None: +# Strips leading admin prefixes: «Россия», «Свердловская область», «г. Екатеринбург» etc. +_ADMIN_PREFIX_RE = re.compile( + r"^(?:" + r"\s*(?:Россия|РФ|Российская\s+Федерация)\s*,?\s*|" + r"\s*[А-Яа-яёЁ][А-Яа-яёЁ\s-]+(?:\s+(?:обл(?:асть)?|р-н|район|округ|край|республика))\.?\s*,?\s*|" + r"\s*(?:г(?:ород)?|гор)\.?\s*[А-Яа-яёЁ][А-Яа-яёЁ\s-]+\s*,?\s*|" + r"\s*[А-Я][а-яё-]+(?:\s+[А-Я][а-яё-]+)?\s*,?\s*" + r")+", + flags=re.UNICODE, +) + +# Recognizes start of a street keyword. +_STREET_START_RE = re.compile( + r"(?:ул\.|улица|пр\.|пр-т|проспект|пер\.|переулок|" + r"б-р|бульвар|ш\.|шоссе|наб\.|набережная|проезд|тракт|пл\.|площадь|" + r"мкр\.?|микрорайон)\s+", + flags=re.IGNORECASE | re.UNICODE, +) + +# Drops trailing apartment / office / corpus noise from the end. +_TRAILING_NOISE_RE = re.compile( + r"\s*,\s*(?:кв\.?\s*\d+|корп\.?\s*\w+|оф\.?\s*\d+|пом\.?\s*\d+|подъезд\s*\d+).*$", + flags=re.IGNORECASE | re.UNICODE, +) + + +def _extract_short_addr(full_address: str | None) -> str | None: """Извлекает «улица + номер дома» из полного адреса для поиска в том же доме. Примеры: - "г. Екатеринбург, ул. Крауля, 48/2, кв. 5" → "ул. Крауля, 48/2" - "Екатеринбург, Ленина, 36, корп. 2, кв. 10" → "Ленина, 36" - "Свердловская обл., г. Екатеринбург, пр-т Ленина, 36 к2" → "пр-т Ленина, 36" + "Свердловская область, г. Екатеринбург, ул. Заводская, д. 44-а" → "ул. Заводская, д. 44-а" + "Россия, Екатеринбург, ул. Малышева, 1" → "ул. Малышева, 1" + "РФ, Свердловская обл., Екатеринбург, ул. Ленина, 5, кв. 12" → "ул. Ленина, 5" + "г. Екатеринбург, проспект Ленина, 50" → "проспект Ленина, 50" + "Екатеринбург, ул. Крауля, 48/2" → "ул. Крауля, 48/2" Алгоритм: - 1. Разбиваем по запятой. - 2. Отбрасываем сегменты, которые выглядят как «г.», «обл.», «р-н» (city/region prefix). - 3. Берём первые 2 оставшихся токена (улица + дом), strip кв/корп/к-суффикс из последнего. - 4. Возвращаем None, если результат слишком короткий (< 3 символов) — не с чем матчить. + 1. Отрезаем trailing кв./корп./оф. noise. + 2. Ищем первый street-keyword токен (ул./пр./пер. и т.д.) — возвращаем с него. + 3. Fallback: агрессивно strip admin-prefix regex, вернуть остаток. + 4. None если строка пустая или нечего возвращать. """ if not full_address: return None + s = full_address.strip() + s = _TRAILING_NOISE_RE.sub("", s) - parts = [p.strip() for p in full_address.split(",")] + # Find first street-keyword position and return from there. + m = _STREET_START_RE.search(s) + if m: + return s[m.start():].strip(" ,.") - # Паттерн для «административных» сегментов: г., обл., р-н, с., д. (населённый пункт) - admin_re = re.compile( - r"^(г\.?|обл\.?|р-н\.?|пгт\.?|с\.?|д\.?|мкр\.?)\s", - re.IGNORECASE, - ) - - meaningful = [p for p in parts if not admin_re.match(p) and len(p) > 1] - - if len(meaningful) < 2: - return None - - street = meaningful[0] - house_raw = meaningful[1] - - # Убираем «кв. N», «корп. N», «к2», «к 2» из номера дома - house = re.sub(r"\s*(кв\.?|корп\.?|к\.?)\s*\d+.*$", "", house_raw, flags=re.IGNORECASE).strip() - - result = f"{street}, {house}" - return result if len(result) >= 3 else None + # Fallback: strip known admin prefixes, return whatever remains. + s = _ADMIN_PREFIX_RE.sub("", s) + return s.strip(" ,.") or None def _stratify_candidates(candidates: list[dict[str, Any]]) -> list[dict[str, Any]]: @@ -959,11 +977,11 @@ def _fetch_analogs( area_tolerance: float = AREA_TOLERANCE, year_built: int | None = None, house_type: str | None = None, total_floors: int | None = None, - ext_house_id: str | None = None, + # TODO: когда listings получит колонку house_id_fk — добавить ext_house_id JOIN для Tier S. ) -> tuple[list[dict[str, Any]], bool, str]: """SELECT аналогов — трёхуровневый house-match (S → H → W). - **Tier S (same building):** address ILIKE prefix OR ext_house_id match. + **Tier S (same building):** address ILIKE prefix-match по short_addr. Если ≥3 результатов → возвращаем только их; tier='S'. **Tier H (same class):** PostGIS + rooms + area + year ±15 + total_floors ±30%. @@ -996,26 +1014,13 @@ def _fetch_analogs( } # ── Tier S: same building ───────────────────────────────────────────────── - short_addr = _extract_short_addr(full_address) if full_address else None - has_ext_id = ext_house_id is not None + short_addr = _extract_short_addr(full_address) - if short_addr or has_ext_id: - addr_clause = "" - if short_addr and has_ext_id: - addr_clause = ( - "(address ILIKE :short_addr_prefix" - " OR (ext_house_id IS NOT NULL AND ext_house_id = :ext_house_id))" - ) - elif short_addr: - addr_clause = "address ILIKE :short_addr_prefix" - else: - addr_clause = "ext_house_id IS NOT NULL AND ext_house_id = :ext_house_id" - - tier_s_params = {**base_params} - if short_addr: - tier_s_params["short_addr_prefix"] = short_addr + "%" - if has_ext_id: - tier_s_params["ext_house_id"] = ext_house_id + if short_addr: + tier_s_params = { + **base_params, + "short_addr_prefix": short_addr + "%", + } tier_s_rows = db.execute( text( @@ -1027,7 +1032,7 @@ def _fetch_analogs( 0.0 AS relevance_score, row_number() OVER (PARTITION BY address ORDER BY scraped_at DESC) AS rn_addr FROM listings - WHERE {addr_clause} + WHERE address ILIKE :short_addr_prefix {_COMMON_WHERE} ) SELECT @@ -1038,7 +1043,7 @@ def _fetch_analogs( scraped_at, distance_m, relevance_score FROM base WHERE rn_addr <= :max_per_addr - ORDER BY relevance_score + ORDER BY scraped_at DESC LIMIT 300 """ ), diff --git a/tradein-mvp/backend/tests/test_extract_short_addr.py b/tradein-mvp/backend/tests/test_extract_short_addr.py new file mode 100644 index 00000000..eb8b2b58 --- /dev/null +++ b/tradein-mvp/backend/tests/test_extract_short_addr.py @@ -0,0 +1,49 @@ +"""Unit tests for _extract_short_addr in app.services.estimator. + +Covers real Ekaterinburg address formats: full admin chain, Россия/РФ prefix, +apartment stripped, prospect keyword, korpus stripped, non-street fallback, None passthrough. +""" + +from app.services.estimator import _extract_short_addr + + +def test_full_admin_chain() -> None: + assert _extract_short_addr( + "Свердловская область, г. Екатеринбург, Склад, ул. Заводская, д. 44-а" + ) == "ул. Заводская, д. 44-а" + + +def test_russia_prefix() -> None: + assert _extract_short_addr( + "Россия, Екатеринбург, ул. Малышева, 1" + ) == "ул. Малышева, 1" + + +def test_apt_stripped() -> None: + out = _extract_short_addr("Екатеринбург, ул. Ленина, 5, кв. 12") + assert out is not None + assert "кв" not in out + assert out.startswith("ул. Ленина") + + +def test_prospekt_keyword() -> None: + out = _extract_short_addr("г. Екатеринбург, проспект Ленина, 50") + assert out is not None + assert out.startswith("проспект Ленина") + + +def test_korp_stripped() -> None: + out = _extract_short_addr("Екатеринбург, ул. Крауля, 48, корп. 2") + assert out is not None + assert "корп" not in out.lower() + + +def test_no_street_keyword_returns_fallback() -> None: + # Should not crash; either returns stripped str or None. + out = _extract_short_addr("Industrial zone X") + assert out is None or "Industrial" in out + + +def test_none_passthrough() -> None: + assert _extract_short_addr(None) is None + assert _extract_short_addr("") is None -- 2.45.3