fix(tradein-estimator): address review-bot items on PR #507
- _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.
This commit is contained in:
lekss361
parent
fa43ea8dc8
commit
8e0e80cf58
2 changed files with 105 additions and 51 deletions
|
|
@ -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
|
||||
"""
|
||||
),
|
||||
|
|
|
|||
49
tradein-mvp/backend/tests/test_extract_short_addr.py
Normal file
49
tradein-mvp/backend/tests/test_extract_short_addr.py
Normal file
|
|
@ -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
|
||||
Loading…
Add table
Reference in a new issue