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