bot-backend
17d558b18c
All checks were successful
Deploy Trade-In / changes (push) Successful in 12s
Deploy Trade-In / build-frontend (push) Has been skipped
Deploy Trade-In / build-browser (push) Has been skipped
Deploy Trade-In / test (push) Successful in 3m27s
Deploy Trade-In / build-backend (push) Successful in 1m33s
Deploy Trade-In / deploy (push) Successful in 2m12s
296 lines
13 KiB
Python
296 lines
13 KiB
Python
"""Per-field source priority for cross-source canonical merge.
|
||
|
||
Direct port of Cross_Source_Matching_Strategy.md sec 3.5 (houses) + 4.5 (listings).
|
||
Source-of-truth dicts read by merge logic in match_or_create_house/listing.
|
||
|
||
Sources covered: avito (serp/detail/houses_catalog/domoteka/imv),
|
||
cian (serp/bti/detail/stats/valuation), yandex (serp/detail/realty_nb/valuation).
|
||
|
||
STATUS (audit finding, confirmed against vault Decision_774_Matching_Architecture,
|
||
2026-05-31, code-archaeology + live-DB verified): `resolve_house_field` /
|
||
`resolve_listing_field` / `HOUSE_FIELD_PRIORITY` / `LISTING_FIELD_PRIORITY` are
|
||
NOT called anywhere in the production merge path — real house/listing upserts
|
||
in `matching/houses.py` / `matching/listings.py` use a simpler ad hoc
|
||
`COALESCE(EXCLUDED.x, table.x)` (newest-non-null-wins) pattern instead.
|
||
`update_canonical_fields` below is a deliberate Stage-8-v1 no-op stub ("Full
|
||
arbitration deferred to Stage 8.x" — see its own docstring); this is NOT an
|
||
accidentally-orphaned integration, it is an intentionally-staged one that never
|
||
got a Stage-8.x follow-up. Decision_774 already scoped removing this
|
||
(`resolve_*`/stub/`match_or_create_listing`) as an independent "Path 2 / Sub-3"
|
||
cleanup PR, deliberately kept separate from the accuracy-affecting Path 1a work
|
||
(house_id_fk anchor) — do NOT wire this into the live price-calc path without a
|
||
dedicated backtest+A/B (would change client-facing estimates). Left in place
|
||
(not deleted) here because ~30 existing tests in
|
||
`tests/matching/test_conflict_resolution.py` + `tests/test_matching.py` cover
|
||
it in detail; removing both belongs in that separate Sub-3 PR, not bundled with
|
||
an unrelated Tier-S bugfix.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from typing import Any
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# cross_validate: warn if spread between sources exceeds this fraction of the median.
|
||
_CROSS_VALIDATE_DIVERGENCE_THRESHOLD = 0.10
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# HOUSE_FIELD_PRIORITY — vault sec 3.5
|
||
# ---------------------------------------------------------------------------
|
||
HOUSE_FIELD_PRIORITY: dict[str, list[str] | str] = {
|
||
"address": ["cian", "avito", "yandex"],
|
||
"lat": ["cian_serp", "avito_houses_catalog", "yandex_realty_nb"],
|
||
"lon": ["cian_serp", "avito_houses_catalog", "yandex_realty_nb"],
|
||
"year_built": [
|
||
"cian_bti",
|
||
"cian_serp",
|
||
"avito_houses_catalog",
|
||
"yandex_valuation",
|
||
"yandex_realty_nb",
|
||
],
|
||
"house_type": ["cian_bti", "cian_serp", "avito", "yandex_valuation", "yandex_realty_nb"],
|
||
"series_name": ["cian_bti"],
|
||
"passenger_lifts_count": ["cian", "avito"],
|
||
"cargo_lifts_count": ["cian", "avito"],
|
||
"has_concierge": ["cian", "avito"],
|
||
"closed_yard": ["cian", "avito"],
|
||
"parking_type": ["cian"],
|
||
"flat_count": ["cian_bti"],
|
||
"entrances": ["cian_bti"],
|
||
"is_emergency": ["cian_bti"],
|
||
"management_company_id": ["cian_valuation"],
|
||
"house_class": ["avito_houses_catalog", "cian", "yandex_realty_nb"],
|
||
"rating_score": ["avito_houses_catalog", "cian", "yandex_realty_nb"],
|
||
"reviews_count": ["avito_houses_catalog", "cian", "yandex_realty_nb"],
|
||
# Yandex unique fields (newbuilding landing only)
|
||
"text_reviews_count": ["yandex_realty_nb"], # 353 text reviews — Yandex strongpoint
|
||
"corpus_count": ["yandex_realty_nb"], # "три башни" → 3
|
||
"total_area_ha": ["yandex_realty_nb"], # ЖК footprint
|
||
"commission_year": ["cian_serp", "yandex_realty_nb"],
|
||
"commission_month": ["yandex_realty_nb"], # raw RU month name
|
||
"developer_name": ["cian", "yandex_realty_nb"],
|
||
# #2674 (хвост): запись про «панораму» удалена вместе с колонкой (мигр. 259).
|
||
# В отличие от ceiling_height ниже, правило было ИСПОЛНИМО — колонка существовала,
|
||
# источник её писал. Разрешать было нечего: yandex_valuation отдавал False всегда,
|
||
# потому что слова «панорам» на странице оценки нет (0 true из 1536 страниц на
|
||
# проде; живая проверка боевым трактом 13.08.2026 не нашла его и в сыром HTML).
|
||
"yandex_total_listings": ["yandex_valuation"], # "N объектов" в истории
|
||
# Yandex Valuation enrichment (existing house attrs)
|
||
"has_lift": ["cian_bti", "cian_detail", "yandex_valuation"],
|
||
# #2699: запись "ceiling_height" удалена — колонки с таким именем в `houses`
|
||
# никогда не было (проверено на проде: 0 колонок LIKE '%ceiling%'), правило не
|
||
# могло сработать ни разу. Высота потолков — атрибут ОБЪЯВЛЕНИЯ
|
||
# (listings.ceiling_height_m), см. LISTING_FIELD_PRIORITY ниже.
|
||
}
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# LISTING_FIELD_PRIORITY — vault sec 4.5
|
||
# ---------------------------------------------------------------------------
|
||
LISTING_FIELD_PRIORITY: dict[str, list[str] | str] = {
|
||
"address": ["cian_serp", "avito_detail", "avito_serp"],
|
||
"lat": ["cian_serp", "avito_detail"],
|
||
"lon": ["cian_serp", "avito_detail"],
|
||
"area_m2": ["cian_serp", "avito_detail"],
|
||
"living_area_m2": ["cian_serp"],
|
||
"kitchen_area_m2": ["cian_serp", "avito_detail"],
|
||
# #2699: ключ — имя РЕАЛЬНОЙ колонки. Был "ceiling_height" (019), которую
|
||
# эстиматор не читает; канон — ceiling_height_m, и её пишут все три источника,
|
||
# а не только cian_detail.
|
||
"ceiling_height_m": ["cian_detail", "avito_detail", "yandex_detail"],
|
||
"floor": ["cian_serp", "avito_detail"],
|
||
"total_floors": ["cian_serp", "avito_detail"],
|
||
"year_built": ["cian_serp"],
|
||
"house_type": ["cian_serp", "avito_detail", "yandex_detail"],
|
||
"repair_state": ["cian_detail", "avito_detail"],
|
||
"has_balcony": ["cian_serp", "avito_detail"],
|
||
"balconies_count": ["cian_serp"],
|
||
"loggias_count": ["cian_serp"],
|
||
"windows_view_type": ["cian_detail", "avito_detail"],
|
||
"separate_wcs_count": ["cian_detail"],
|
||
"combined_wcs_count": ["cian_detail"],
|
||
"room_type": ["cian_detail", "avito_detail"],
|
||
"has_furniture": ["cian_serp", "avito_detail"],
|
||
"description": ["cian_serp", "avito_detail", "yandex_detail"],
|
||
"photo_urls": "union",
|
||
# Avito Domoteka unique
|
||
"owners_count": ["avito_domoteka"],
|
||
"owners_at_least": ["avito_domoteka"],
|
||
"last_owner_change_date": ["avito_domoteka"],
|
||
"encumbrances_clean": ["avito_domoteka"],
|
||
"registry_match": ["avito_domoteka"],
|
||
# Cian-only
|
||
"is_rosreestr_checked": ["cian_serp"],
|
||
"is_layout_approved": ["cian_serp"],
|
||
"is_commercial_ownership_verified": ["cian_serp"],
|
||
# Cross-validation
|
||
"price_rub": "cross_validate",
|
||
"kadastr_num": "first_non_null",
|
||
# NOTE: task description claims price_rub=['cian','avito','yandex']; vault sec 4.5
|
||
# says 'cross_validate' (flag if diff > 10%). Following vault — change to list if
|
||
# cross_validate semantics aren't desired at caller.
|
||
# Yandex unique (agency block — OfferCardAuthorInfo)
|
||
"agency_name": ["yandex_detail"],
|
||
"agency_founded_year": ["yandex_detail"],
|
||
"agency_objects_count": ["yandex_detail"],
|
||
# Yandex parallel views column (NOT existing `views_total` which is Cian's)
|
||
"views_total_yandex": ["yandex_detail"],
|
||
# Yandex raw publish-date text (relative form)
|
||
"publish_date_relative": ["yandex_detail"],
|
||
}
|
||
|
||
|
||
def _freshest_non_null(
|
||
candidates: dict[str, Any],
|
||
timestamps: dict[str, Any] | None,
|
||
) -> Any:
|
||
"""Pick the non-null candidate from the freshest source (max last_seen_at).
|
||
|
||
Deterministic replacement for dict-insertion-order ``first_non_null`` (#1539).
|
||
Among sources with a non-null value:
|
||
|
||
* if ``timestamps`` is provided, choose the value whose source has the most
|
||
recent ``last_seen_at``; ties (equal / missing timestamps) break by source
|
||
name (lexicographic) so the result is reproducible regardless of dict order;
|
||
* if no usable timestamps are available, fall back to first non-null in
|
||
sorted-key order (still deterministic, unlike raw insertion order).
|
||
"""
|
||
non_null = {src: v for src, v in candidates.items() if v is not None}
|
||
if not non_null:
|
||
return None
|
||
timestamps = timestamps or {}
|
||
|
||
timed = [src for src in non_null if timestamps.get(src) is not None]
|
||
if timed:
|
||
# Freshest last_seen_at wins; equal timestamps tie-break by source name
|
||
# (lexicographically smallest) so the result never depends on dict order.
|
||
latest = max(timestamps[src] for src in timed)
|
||
winner = min(src for src in timed if timestamps[src] == latest)
|
||
return non_null[winner]
|
||
# No timestamps anywhere: deterministic by sorted source name.
|
||
return non_null[min(non_null)]
|
||
|
||
|
||
def _resolve(
|
||
priority: dict[str, list[str] | str],
|
||
field: str,
|
||
candidates: dict[str, Any],
|
||
timestamps: dict[str, Any] | None = None,
|
||
) -> Any:
|
||
"""Pick value from candidates per priority dict semantics.
|
||
|
||
``candidates`` maps source name -> value for ``field``. ``timestamps`` maps
|
||
source name -> ``last_seen_at`` (any comparable, usually ``datetime``); when
|
||
supplied it makes conflict resolution prefer the freshest source instead of
|
||
relying on non-deterministic dict insertion order (#1539).
|
||
"""
|
||
if not candidates:
|
||
return None
|
||
rule = priority.get(field, "first_non_null")
|
||
|
||
if isinstance(rule, str):
|
||
if rule == "union":
|
||
out: list[Any] = []
|
||
for v in candidates.values():
|
||
if v is None:
|
||
continue
|
||
if isinstance(v, list | tuple | set):
|
||
out.extend(v)
|
||
else:
|
||
out.append(v)
|
||
# Preserve order, dedup
|
||
seen: set[Any] = set()
|
||
uniq: list[Any] = []
|
||
for x in out:
|
||
key = repr(x)
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
uniq.append(x)
|
||
return uniq
|
||
if rule == "first_non_null":
|
||
return _freshest_non_null(candidates, timestamps)
|
||
if rule == "max":
|
||
non_null = [v for v in candidates.values() if v is not None]
|
||
return max(non_null) if non_null else None
|
||
if rule == "min":
|
||
non_null = [v for v in candidates.values() if v is not None]
|
||
return min(non_null) if non_null else None
|
||
if rule == "cross_validate":
|
||
# Return the median; flag divergence (vault sec 4.5: warn if diff > 10%).
|
||
nums = [v for v in candidates.values() if isinstance(v, int | float)]
|
||
if not nums:
|
||
return None
|
||
nums.sort()
|
||
n = len(nums)
|
||
if n % 2:
|
||
median = nums[n // 2]
|
||
else:
|
||
# True median for even n: mean of the two central values.
|
||
median = (nums[n // 2 - 1] + nums[n // 2]) / 2
|
||
if median and (max(nums) - min(nums)) > _CROSS_VALIDATE_DIVERGENCE_THRESHOLD * abs(
|
||
median
|
||
):
|
||
logger.warning(
|
||
"cross_validate divergence for field %r: min=%s max=%s median=%s "
|
||
"(spread exceeds %.0f%% of median); candidates=%r",
|
||
field,
|
||
min(nums),
|
||
max(nums),
|
||
median,
|
||
_CROSS_VALIDATE_DIVERGENCE_THRESHOLD * 100,
|
||
candidates,
|
||
)
|
||
return median
|
||
# Unknown rule — same deterministic freshness fallback as first_non_null.
|
||
return _freshest_non_null(candidates, timestamps)
|
||
|
||
# Priority list: pick value from highest-ranked source present (non-null).
|
||
# Explicit per-field ranking is authoritative and beats freshness.
|
||
for src in rule:
|
||
if src in candidates and candidates[src] is not None:
|
||
return candidates[src]
|
||
# Fallback: no ranked source present → freshest non-null (deterministic).
|
||
return _freshest_non_null(candidates, timestamps)
|
||
|
||
|
||
def resolve_house_field(
|
||
field: str,
|
||
candidates: dict[str, Any],
|
||
timestamps: dict[str, Any] | None = None,
|
||
) -> Any:
|
||
"""Pick canonical house field value per HOUSE_FIELD_PRIORITY.
|
||
|
||
``timestamps`` (source -> last_seen_at) makes first_non_null / fallback
|
||
resolution prefer the freshest source deterministically (#1539).
|
||
"""
|
||
return _resolve(HOUSE_FIELD_PRIORITY, field, candidates, timestamps)
|
||
|
||
|
||
def resolve_listing_field(
|
||
field: str,
|
||
candidates: dict[str, Any],
|
||
timestamps: dict[str, Any] | None = None,
|
||
) -> Any:
|
||
"""Pick canonical listing field value per LISTING_FIELD_PRIORITY.
|
||
|
||
``timestamps`` (source -> last_seen_at) makes first_non_null / fallback
|
||
resolution prefer the freshest source deterministically (#1539).
|
||
"""
|
||
return _resolve(LISTING_FIELD_PRIORITY, field, candidates, timestamps)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Legacy stub — kept for backward compat with existing __init__.py and tests
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def update_canonical_fields(
|
||
db: Any,
|
||
listing_id: int,
|
||
ext_source: str,
|
||
lot_data: object,
|
||
) -> None:
|
||
"""Legacy Stage 8 v1 stub — full arbitration deferred to Stage 8.x."""
|
||
pass
|