gendesign/tradein-mvp/backend/app/services/matching/listings.py

"""Listing cross-source matching — 3-tier algorithm.

Tier 0 (confidence 1.0): cadastral_number exact match on listings table.
Tier 1 (confidence 1.0): ext_source + ext_id already in listing_sources.
Tier 2 (confidence 0.85): description_minhash exact match within same house.
Tier 3 (confidence 0.75): composite match — house + floor + area ±2% + rooms.
New   (sentinel 0):       caller is responsible for INSERT + call _upsert_listing_source.

Algorithm reference: decisions/Cross_Source_Matching_Strategy.md sec 4
"""

import json
import logging

from sqlalchemy import text
from sqlalchemy.orm import Session

logger = logging.getLogger(__name__)


def match_or_create_listing(
    db: Session,
    ext_source: str,
    ext_id: str,
    house_id: int,
    floor: int | None = None,
    rooms_count: int | None = None,
    area_m2: float | None = None,
    price_rub: float | None = None,
    *,
    cadastral_number: str | None = None,
    description_minhash: str | None = None,
    source_url: str | None = None,
    source_data: dict | None = None,
) -> tuple[int, float, str]:
    """Match existing listing or signal that caller should create a new one.

    Returns:
        (listing_id, confidence in [0.0, 1.0], method name)

    Method values:
        'cadastr_exact'  — matched by cadastral number (confidence 1.0)
        'source_exact'   — already in listing_sources for this source+ext_id (confidence 1.0)
        'minhash'        — matched by description minhash within house (confidence 0.85)
        'composite'      — matched by house+floor+area±2%+rooms (confidence 0.75)
        'new'            — no match; caller must INSERT listing then call upsert_listing_source()

    When method == 'new', listing_id is 0 (sentinel). Caller flow:
        listing_id, conf, method = match_or_create_listing(db, ...)
        if method == 'new':
            listing_id = <INSERT canonical listing row>
            upsert_listing_source(db, listing_id=listing_id, ...)
    """
    # Tier 0: cadastral number exact match
    if cadastral_number:
        row = (
            db.execute(
                text(
                    "SELECT id FROM listings WHERE cadastral_number = :cad ORDER BY id ASC LIMIT 1"
                ),
                {"cad": cadastral_number},
            )
            .mappings()
            .first()
        )
        if row:
            listing_id = int(row["id"])
            _upsert_listing_source(
                db,
                listing_id=listing_id,
                ext_source=ext_source,
                ext_id=ext_id,
                method="cadastr_exact",
                confidence=1.0,
                price_rub=price_rub,
                area_m2=area_m2,
                floor=floor,
                rooms_count=rooms_count,
                source_url=source_url,
                source_data=source_data,
            )
            logger.info("listing match cadastr_exact id=%s cad=%s", listing_id, cadastral_number)
            return (listing_id, 1.0, "cadastr_exact")

    # Tier 1: source+ext_id already registered
    row = (
        db.execute(
            text(
                "SELECT listing_id FROM listing_sources "
                "WHERE ext_source = :s AND ext_id = :e LIMIT 1"
            ),
            {"s": ext_source, "e": str(ext_id)},
        )
        .mappings()
        .first()
    )
    if row:
        listing_id = int(row["listing_id"])
        logger.info(
            "listing match source_exact id=%s src=%s ext_id=%s", listing_id, ext_source, ext_id
        )
        return (listing_id, 1.0, "source_exact")

    # Tier 2: description minhash match within same house
    # Stage 8 v1: exact match on pre-computed minhash string.
    # Future (Stage 8.x): replace with datasketch MinHash LSH for approximate Jaccard.
    if description_minhash:
        row = (
            db.execute(
                text("""
                SELECT id FROM listings
                WHERE house_id = CAST(:hid AS bigint)
                  AND description_minhash = :hash
                LIMIT 1
            """),
                {"hid": house_id, "hash": description_minhash},
            )
            .mappings()
            .first()
        )
        if row:
            listing_id = int(row["id"])
            _upsert_listing_source(
                db,
                listing_id=listing_id,
                ext_source=ext_source,
                ext_id=ext_id,
                method="minhash",
                confidence=0.85,
                price_rub=price_rub,
                area_m2=area_m2,
                floor=floor,
                rooms_count=rooms_count,
                source_url=source_url,
                source_data=source_data,
            )
            logger.info("listing match minhash id=%s hash=%s", listing_id, description_minhash)
            return (listing_id, 0.85, "minhash")

    # Tier 3: composite match — house + floor + area ±2% + rooms
    if house_id and floor is not None and area_m2 is not None and rooms_count is not None:
        row = (
            db.execute(
                text("""
                SELECT id,
                       ABS(area_m2 - CAST(:area AS numeric)) AS diff
                FROM listings
                WHERE house_id = CAST(:hid AS bigint)
                  AND floor = :fl
                  AND rooms = :rc
                  AND area_m2 BETWEEN CAST(:area AS numeric) * 0.98
                              AND     CAST(:area AS numeric) * 1.02
                ORDER BY diff ASC
                LIMIT 1
            """),
                {"hid": house_id, "fl": floor, "rc": rooms_count, "area": area_m2},
            )
            .mappings()
            .first()
        )
        if row:
            listing_id = int(row["id"])
            _upsert_listing_source(
                db,
                listing_id=listing_id,
                ext_source=ext_source,
                ext_id=ext_id,
                method="composite",
                confidence=0.75,
                price_rub=price_rub,
                area_m2=area_m2,
                floor=floor,
                rooms_count=rooms_count,
                source_url=source_url,
                source_data=source_data,
            )
            logger.info(
                "listing match composite id=%s house=%s floor=%s area=%.1f rooms=%s",
                listing_id,
                house_id,
                floor,
                area_m2,
                rooms_count,
            )
            return (listing_id, 0.75, "composite")

    # No match — signal caller to create new listing
    # NOTE (Part 4 / #849): match_or_create_listing returns (listing_id, confidence, method)
    # but the main scraper ingestion path (_link_listing_to_house in scrapers/base.py) does NOT
    # call this function — it uses match_or_create_house + upsert_listing_source('source_link')
    # directly. Wiring the matcher confidence/method from this function into listing_sources on
    # the source_link path would require restructuring the ingestion flow non-trivially.
    # listing_sources already has confidence + matched_method columns (028_matching_tables.sql),
    # but they are written as confidence=1.0 / method='source_link' by the scraper path.
    # Accurate per-tier confidence wiring is deferred to #774-3.
    return (0, 1.0, "new")


def upsert_listing_source(
    db: Session,
    *,
    listing_id: int,
    ext_source: str,
    ext_id: str,
    method: str,
    confidence: float,
    price_rub: float | None = None,
    area_m2: float | None = None,
    floor: int | None = None,
    rooms_count: int | None = None,
    source_url: str | None = None,
    source_data: dict | None = None,
) -> None:
    """Public helper — register or refresh a listing_sources row.

    Called by external code after creating a new canonical listing.
    """
    _upsert_listing_source(
        db,
        listing_id=listing_id,
        ext_source=ext_source,
        ext_id=ext_id,
        method=method,
        confidence=confidence,
        price_rub=price_rub,
        area_m2=area_m2,
        floor=floor,
        rooms_count=rooms_count,
        source_url=source_url,
        source_data=source_data,
    )


def _upsert_listing_source(
    db: Session,
    *,
    listing_id: int,
    ext_source: str,
    ext_id: str,
    method: str,
    confidence: float,
    price_rub: float | None,
    area_m2: float | None,
    floor: int | None,
    rooms_count: int | None,
    source_url: str | None,
    source_data: dict | None,
) -> None:
    """Insert or refresh listing_sources row for this source+ext_id.

    Отметки времени — statement_timestamp(), НЕ NOW() (#2731). Этот upsert вызывается
    ПОСТРОЧНО из save_listings (hook `_link_listing_to_house`), а транзакция batch'а
    коммитится один раз в конце, поэтому NOW() (== transaction_timestamp) давал одну
    метку на весь вызов: прод-замер 2026-08-06 — 219 строк на 1 метку в 11:00,
    235/1 в 10:00, 297/1 в 09:00, и так каждый час.

    Чинится вместе с listings.scraped_at/last_seen_at, а не отдельно: сегодня
    listings.last_seen_at = listing_sources.last_seen_at у 100% пар (2407 из 2407 за
    сутки) именно потому, что обе колонки берут одну транзакционную метку. Почини
    только одну — вторая осталась бы замороженной на старте batch'а, и расхождение
    выросло бы с миллисекунд (честная разница двух записей) до длительности прогона.

    Все три колонки пишутся ОДНИМ statement'ом, поэтому statement_timestamp() даёт им
    одинаковое значение; clock_timestamp() развёл бы их на микросекунды.
    """
    raw = json.dumps(source_data) if source_data is not None else None
    db.execute(
        text("""
            INSERT INTO listing_sources (
                listing_id, ext_source, ext_id,
                confidence, matched_method, matched_at, last_seen_at,
                source_url, last_scraped_at,
                price_rub, area_m2, floor, rooms_count, raw_payload
            ) VALUES (
                CAST(:lid AS bigint), :s, :e,
                CAST(:c AS real), :m, statement_timestamp(), statement_timestamp(),
                :url, statement_timestamp(),
                CAST(:p AS bigint), CAST(:a AS numeric), :fl, :rc,
                CAST(:raw AS jsonb)
            )
            ON CONFLICT (ext_source, ext_id) DO UPDATE SET
                confidence    = GREATEST(EXCLUDED.confidence, listing_sources.confidence),
                last_seen_at  = statement_timestamp(),
                last_scraped_at = statement_timestamp(),
                price_rub     = COALESCE(EXCLUDED.price_rub,    listing_sources.price_rub),
                area_m2       = COALESCE(EXCLUDED.area_m2,      listing_sources.area_m2),
                floor         = COALESCE(EXCLUDED.floor,        listing_sources.floor),
                rooms_count   = COALESCE(EXCLUDED.rooms_count,  listing_sources.rooms_count),
                raw_payload   = COALESCE(EXCLUDED.raw_payload,  listing_sources.raw_payload)
        """),
        {
            "lid": listing_id,
            "s": ext_source,
            "e": str(ext_id),
            "c": confidence,
            "m": method,
            "url": source_url,
            # Migration 029 declares listing_sources.price_rub bigint — whole rubles only,
            # kopecks truncated. Document expectation in caller.
            "p": int(price_rub) if price_rub is not None else None,
            "a": area_m2,
            "fl": floor,
            "rc": rooms_count,
            "raw": raw,
        },
    )