diff --git a/backend/app/services/analytics_queries.py b/backend/app/services/analytics_queries.py index 4fe96955..7239d698 100644 --- a/backend/app/services/analytics_queries.py +++ b/backend/app/services/analytics_queries.py @@ -504,6 +504,15 @@ def developer_history( def developer_portfolio(db: Session, developer_id: str) -> list[dict[str, Any]]: + # #2464 cluster E: без дедупа каждый ЖК возвращался ×N раз (одна строка на + # snapshot_date, ретенции нет) — портфель девелопера раздувался в разы (прод-замер + # ~7.6-9.0× per-developer). DISTINCT ON (obj_id) + snapshot_date DESC NULLS LAST — + # latest-snapshot-per-obj_id (тот же паттерн, что cmp_rows/latest_obj CTE в + # recommend_mix ниже в этом файле). dev_id — стабильный идентификатор + # (не меняется между снапшотами одного ЖК) → фильтруется ДО DISTINCT ON, как + # region_cd/district_name в сиблингах. Итоговый ORDER BY ready_dt вынесен во внешний + # SELECT: DISTINCT ON требует, чтобы его собственный ORDER BY начинался с ключа + # дедупа (obj_id, snapshot_date), не с ready_dt. rows = ( db.execute( text( @@ -511,8 +520,15 @@ def developer_portfolio(db: Session, developer_id: str) -> list[dict[str, Any]]: SELECT obj_id, comm_name, addr, region_cd, flat_count, square_living, ready_dt, obj_class, escrow, problem_flag, latitude, longitude, is_ekb - FROM domrf_kn_objects - WHERE dev_id = :dev + FROM ( + SELECT DISTINCT ON (obj_id) + obj_id, comm_name, addr, region_cd, flat_count, + square_living, ready_dt, obj_class, escrow, + problem_flag, latitude, longitude, is_ekb + FROM domrf_kn_objects + WHERE dev_id = :dev + ORDER BY obj_id, snapshot_date DESC NULLS LAST + ) latest ORDER BY ready_dt DESC NULLS LAST """ ), @@ -1797,12 +1813,38 @@ def _active_competitors_count( Возвращает (count, scope_used). Min 1 чтобы не делить на 0.""" def _q(where_extras: str, params: dict[str, Any]) -> int: + # #2464 cluster E: domrf_kn_objects хранит МНОЖЕСТВО snapshot_date на obj_id + # (UNIQUE(obj_id, snapshot_date), ретенции нет) — прод-замер 4090 строк / 482 + # distinct obj_id для site_status='Строящиеся' (8.49× инфляция). Наивный + # COUNT(*) считал каждый исторический снапшот отдельно. Fix: DISTINCT ON + # (obj_id) + snapshot_date DESC NULLS LAST даёт latest-snapshot-per-obj_id + # (сиблинг-паттерн: _L3_FUTURE_SQL #1212 в site_finder/supply_layers.py). + # + # ВСЕ волатильные предикаты применяются ПОСЛЕ DISTINCT ON (во внешнем WHERE), + # а не внутри CTE — иначе DISTINCT ON вернул бы «последний снапшот, ПРОШЕДШИЙ + # фильтр», а не истинно последний снапшот объекта. Прод-замер per-obj_id + # variance across snapshots (authoritative, coordinator 2026-07-08): + # dev_id=0, region_cd=0 (СТАБИЛЬНЫ) · district_name=1, obj_class=29, + # site_status=11 (ВОЛАТИЛЬНЫ). + # → В CTE остаётся ТОЛЬКО region_cd (стабилен + ограничивает стоимость дедупа + # как partition-scope). {where_extras} (district_name/obj_class) и + # site_status='Строящиеся' — во внешнем WHERE, на истинно-последней строке. + # Пример бага, который это чинит: ЖК сменил класс Комфорт→Бизнес; при + # фильтре obj_class внутри CTE взяли бы старый Комфорт-снапшот и посчитали + # его активным Комфортом, хотя его актуальный класс уже Бизнес. n = db.execute( text( f""" - SELECT COUNT(*) FROM domrf_kn_objects - WHERE region_cd = :rc - AND site_status = 'Строящиеся' + WITH latest AS ( + SELECT DISTINCT ON (obj_id) + obj_id, site_status, district_name, + obj_class, obj_class_fallback + FROM domrf_kn_objects + WHERE region_cd = :rc + ORDER BY obj_id, snapshot_date DESC NULLS LAST + ) + SELECT COUNT(*) FROM latest + WHERE site_status = 'Строящиеся' {where_extras} """ ), @@ -2233,14 +2275,28 @@ def _competitors_two_dim( db.execute( text( f""" - WITH active AS ( - SELECT DISTINCT ON (obj_id) obj_id, latitude, longitude, district_name + WITH latest AS ( + -- #2464 cluster E: сначала истинно-последний снапшот на obj_id, + -- ТОЛЬКО по стабильному region_cd (variance=0). Волатильные + -- предикаты (site_status=11, district_name=1, obj_class=29 per + -- прод-замер) — НЕ внутри DISTINCT ON, иначе взяли бы «последний + -- снапшот, прошедший фильтр», а не истинно последний (ЖК, + -- сменивший класс/район/статус, засчитался бы по устаревшему + -- снапшоту). Зеркалит _q() в _active_competitors_count выше. + SELECT DISTINCT ON (obj_id) + obj_id, latitude, longitude, district_name, + site_status, obj_class, obj_class_fallback FROM domrf_kn_objects WHERE region_cd = :rc - AND site_status = 'Строящиеся' + ORDER BY obj_id, snapshot_date DESC NULLS LAST + ), + active AS ( + -- Волатильные фильтры на истинно-последней строке. + SELECT obj_id, latitude, longitude, district_name + FROM latest + WHERE site_status = 'Строящиеся' AND district_name = :dn {class_filter} - ORDER BY obj_id, snapshot_date DESC NULLS LAST ), centroid AS ( SELECT ST_SetSRID(ST_GeomFromText(:centroid), 4326)::geography AS pt diff --git a/backend/app/services/scrapers/domrf_kn.py b/backend/app/services/scrapers/domrf_kn.py index 3aa23056..87875301 100644 --- a/backend/app/services/scrapers/domrf_kn.py +++ b/backend/app/services/scrapers/domrf_kn.py @@ -535,14 +535,29 @@ UPSERT_OBJECT_SQL = text( :snapshot_date ) ON CONFLICT (obj_id, snapshot_date) DO UPDATE SET + -- #2464 cluster E: hobj_id/short_addr/dev_inn/region_cd/floor_min/floor_max/ + -- problem_flag/green_house были в INSERT (populated every call by + -- _norm_object) но пропущены в DO UPDATE SET → re-run скрапера для + -- уже существующего (obj_id, snapshot_date) оставлял эти 8 колонок stale. + -- Все 8 — direct EXCLUDED (не COALESCE): freshly computed каждый вызов, + -- в т.ч. problem_flag должен ОБНУЛЯТЬСЯ, если проблема на объекте снята + -- (COALESCE зафиксировал бы устаревший "проблемный" статус навсегда). + hobj_id = EXCLUDED.hobj_id, comm_name = EXCLUDED.comm_name, addr = EXCLUDED.addr, + short_addr = EXCLUDED.short_addr, + region_cd = EXCLUDED.region_cd, dev_id = EXCLUDED.dev_id, dev_name = EXCLUDED.dev_name, + dev_inn = EXCLUDED.dev_inn, + floor_min = EXCLUDED.floor_min, + floor_max = EXCLUDED.floor_max, flat_count = EXCLUDED.flat_count, square_living = EXCLUDED.square_living, ready_dt = EXCLUDED.ready_dt, + problem_flag = EXCLUDED.problem_flag, site_status = EXCLUDED.site_status, + green_house = EXCLUDED.green_house, escrow = EXCLUDED.escrow, obj_class = COALESCE(EXCLUDED.obj_class, domrf_kn_objects.obj_class), wall_type = COALESCE(EXCLUDED.wall_type, domrf_kn_objects.wall_type), @@ -921,6 +936,9 @@ UPSERT_PHOTO_SQL = text( :local_path, :thumb_path, :downloaded_at ) ON CONFLICT (obj_id, obj_file_id) DO UPDATE SET + -- #2464 cluster F: build_type в INSERT column list (populated every call) + -- но пропущен здесь → re-run скрапера оставлял build_type stale. + build_type = EXCLUDED.build_type, ord_num = EXCLUDED.ord_num, photo_url = EXCLUDED.photo_url, photo_dttm = EXCLUDED.photo_dttm, diff --git a/backend/tests/services/scrapers/test_domrf_kn_upsert_sql.py b/backend/tests/services/scrapers/test_domrf_kn_upsert_sql.py new file mode 100644 index 00000000..c0f7de00 --- /dev/null +++ b/backend/tests/services/scrapers/test_domrf_kn_upsert_sql.py @@ -0,0 +1,107 @@ +"""#2464 cluster E + F — UPSERT DO UPDATE SET column-omission regression tests. + +Both UPSERT_OBJECT_SQL and UPSERT_PHOTO_SQL had columns populated on every +INSERT (fresh scrape values, see _norm_object / catalog scraper) but silently +omitted from `ON CONFLICT ... DO UPDATE SET`. Re-running the scraper for an +already-seen (obj_id, snapshot_date) / (obj_id, obj_file_id) row left those +columns stale forever (INSERT wins once, DO UPDATE never refreshes them). + +SQL-string assertions against the module-level `text(...)` constants — no live +DB needed; the INSERT column list is the source of truth for "must this column +also be kept fresh on conflict", so each test also cross-checks against the +INSERT list to prevent the same class of omission from creeping back in for a +newly-added column. +""" + +from __future__ import annotations + +import re + +from app.services.scrapers.domrf_kn import UPSERT_OBJECT_SQL, UPSERT_PHOTO_SQL + +_OBJECT_PREVIOUSLY_OMITTED = [ + "problem_flag", + "green_house", + "floor_min", + "floor_max", + "hobj_id", + "short_addr", + "dev_inn", + "region_cd", +] + + +def _insert_columns(sql: str, table: str) -> list[str]: + """Extract the `INSERT INTO (col1, col2, ...)` column list.""" + m = re.search(rf"INSERT INTO {table}\s*\(([^)]+)\)", sql) + assert m, f"could not locate INSERT INTO {table} (...) column list" + return [c.strip() for c in m.group(1).split(",")] + + +def _do_update_set_columns(sql: str) -> set[str]: + """Extract every `col = ` LHS from the `DO UPDATE SET` clause.""" + m = re.search(r"DO UPDATE SET(.*)", sql, re.DOTALL) + assert m, "could not locate ON CONFLICT ... DO UPDATE SET clause" + body = m.group(1) + return set(re.findall(r"(?:^|,)\s*(\w+)\s*=", body, re.MULTILINE)) + + +class TestUpsertObjectSqlDoUpdateSet: + def test_previously_omitted_columns_now_present(self) -> None: + sql = str(UPSERT_OBJECT_SQL) + updated = _do_update_set_columns(sql) + missing = [c for c in _OBJECT_PREVIOUSLY_OMITTED if c not in updated] + assert not missing, f"still missing from DO UPDATE SET: {missing}" + + def test_previously_omitted_columns_use_direct_excluded_not_coalesce(self) -> None: + # These 8 are freshly computed every scrape call (per _norm_object) -- + # COALESCE would permanently pin a stale value once ever set (most + # visibly wrong for problem_flag: a resolved problem must be able to + # clear back to NULL, not get stuck "still problematic" forever). + sql = " ".join(str(UPSERT_OBJECT_SQL).split()) + for col in _OBJECT_PREVIOUSLY_OMITTED: + assert ( + f"{col} = EXCLUDED.{col}" in sql + ), f"{col} should be direct `= EXCLUDED.{col}` (not COALESCE):\n{sql}" + # guard against a COALESCE(EXCLUDED.col, ...) formulation, which would + # permanently pin a stale value once ever set instead of refreshing it. + assert ( + f"COALESCE(EXCLUDED.{col}," not in sql + ), f"{col} must not be wrapped in COALESCE (needs to be able to clear to NULL)" + + def test_all_insert_columns_covered_by_do_update_set_or_conflict_target(self) -> None: + # Locks in the class of bug (INSERT populates a column, DO UPDATE SET + # silently drops it) so it can't recur for a column added later. + sql = str(UPSERT_OBJECT_SQL) + insert_cols = set(_insert_columns(sql, "domrf_kn_objects")) + updated_cols = _do_update_set_columns(sql) + conflict_target = {"obj_id", "snapshot_date"} # immutable conflict key, not updated + missing = insert_cols - updated_cols - conflict_target + assert not missing, f"INSERT columns missing from DO UPDATE SET: {sorted(missing)}" + + def test_no_double_colon_cast(self) -> None: + sql = str(UPSERT_OBJECT_SQL) + assert re.search(r":[a-z_]+::[a-z]", sql) is None + + +class TestUpsertPhotoSqlDoUpdateSet: + def test_build_type_now_present(self) -> None: + sql = str(UPSERT_PHOTO_SQL) + updated = _do_update_set_columns(sql) + assert "build_type" in updated + + def test_build_type_uses_direct_excluded(self) -> None: + sql = " ".join(str(UPSERT_PHOTO_SQL).split()) + assert "build_type = EXCLUDED.build_type" in sql + + def test_all_insert_columns_covered_by_do_update_set_or_conflict_target(self) -> None: + sql = str(UPSERT_PHOTO_SQL) + insert_cols = set(_insert_columns(sql, "domrf_kn_photos")) + updated_cols = _do_update_set_columns(sql) + conflict_target = {"obj_id", "obj_file_id"} + missing = insert_cols - updated_cols - conflict_target + assert not missing, f"INSERT columns missing from DO UPDATE SET: {sorted(missing)}" + + def test_no_double_colon_cast(self) -> None: + sql = str(UPSERT_PHOTO_SQL) + assert re.search(r":[a-z_]+::[a-z]", sql) is None diff --git a/backend/tests/services/test_analytics_queries_domrf_dedup.py b/backend/tests/services/test_analytics_queries_domrf_dedup.py new file mode 100644 index 00000000..a6db1805 --- /dev/null +++ b/backend/tests/services/test_analytics_queries_domrf_dedup.py @@ -0,0 +1,380 @@ +"""#2464 cluster E — domrf_kn_objects snapshot dedup regression tests. + +domrf_kn_objects retains MULTIPLE historical rows per obj_id (one per +snapshot_date, UNIQUE(obj_id, snapshot_date), no retention). Prod measurement: +4090 rows / 482 distinct obj_id for site_status='Строящиеся' (8.49x inflation). + +Three read-side queries counted/returned every retained snapshot instead of the +latest one per obj_id: + - _active_competitors_count (via its _q closure) + - developer_portfolio + - _competitors_two_dim (its `active` CTE) + +All fixed with the established repo dedup pattern: DISTINCT ON (obj_id) ... +ORDER BY obj_id, snapshot_date DESC NULLS LAST (sibling: _L3_FUTURE_SQL #1212 in +app/services/site_finder/supply_layers.py, cmp_rows/latest_obj CTE in +analytics_queries.recommend_mix). + +VOLATILE-FILTER PLACEMENT (authoritative prod measurement, coordinator +2026-07-08 — per-obj_id variance across snapshots): dev_id=0, region_cd=0 +(STABLE) vs district_name=1, obj_class=29, site_status=11 (VOLATILE). Volatile +predicates MUST be applied to the true-latest row (outer/downstream WHERE), NOT +pre-filtered inside the DISTINCT ON CTE — else DISTINCT ON returns "the latest +snapshot that still MATCHED the filter", not the object's true latest snapshot +(a ЖК that sold out / changed class / moved districts would be miscounted by a +stale snapshot). Only the stable region_cd stays inside the CTE as the dedup +partition-scope. dev_id (developer_portfolio) is likewise stable → kept in-CTE. + +Mock-based — mirrors the SQL-shape assertion convention used in +tests/services/site_finder/test_supply_layers.py (TestLayer2Hidden. +test_sql_dedups_latest_snapshot) since there's no SQLite stand-in for +Postgres-only `DISTINCT ON` syntax and no throwaway-schema DML fixture in this +repo's integration-test conventions (tests/integration/conftest.py is +EXPLAIN-only / read-only against a real tunnel DB — safe for plan-checks, not +for INSERT fixtures). A second class (`TestDedupAlgorithmSpec`) hand-replicates +the exact dedup+filter semantics the SQL implements in pure Python and asserts +the two behaviors the epic calls out explicitly: multi-snapshot obj_id counted +once, and a sold-out object with only an OLD 'Строящиеся' snapshot excluded. +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import MagicMock + +from app.services.analytics_queries import _active_competitors_count, developer_portfolio + +# ── shared helpers (mirrors test_supply_layers.py _mock_db/_executed_sql) ───── + + +def _mock_scalar_db(value: int) -> MagicMock: + """Session stand-in: every db.execute(...).scalar() returns `value`.""" + db = MagicMock() + db.execute.return_value.scalar.return_value = value + return db + + +def _mock_mapping_db(rows: list[dict[str, Any]]) -> MagicMock: + db = MagicMock() + db.execute.return_value.mappings.return_value.all.return_value = rows + return db + + +def _executed_sql(db: MagicMock, call_index: int = 0) -> str: + args, _kwargs = db.execute.call_args_list[call_index] + return str(args[0]) + + +def _executed_params(db: MagicMock, call_index: int = 0) -> dict: + args, _kwargs = db.execute.call_args_list[call_index] + return args[1] + + +def _split_latest_cte(sql: str) -> tuple[str, str]: + """Split whitespace-normalized SQL into (CTE body, outer body). + + CTE body = `WITH latest AS ( ... )` up to the `)` that closes the CTE, + which is the first `)` after the ORDER BY feeding DISTINCT ON. Outer body = + everything after. Lets tests assert WHICH side a predicate lives on. + """ + norm = " ".join(sql.split()) + cte_open = norm.index("WITH latest AS (") + order_by_idx = norm.index("ORDER BY obj_id, snapshot_date DESC NULLS LAST") + cte_close = norm.index(")", order_by_idx) + return norm[cte_open:cte_close], norm[cte_close:] + + +# ── _active_competitors_count SQL shape ──────────────────────────────────── + + +class TestActiveCompetitorsCountSqlShape: + def test_dedups_via_distinct_on_obj_id_latest_snapshot(self) -> None: + # n never reaches the >=2 early-return threshold with scalar()==0, so + # all 3 cascade tiers execute — call_index 0 is tier1 (district+class). + db = _mock_scalar_db(0) + _active_competitors_count( + db, region_code=66, district_name="Ленинский", target_class="Комфорт" + ) + sql = _executed_sql(db, 0) + norm = " ".join(sql.split()) + assert "DISTINCT ON (obj_id)" in norm + assert "ORDER BY obj_id, snapshot_date DESC NULLS LAST" in norm + + def test_site_status_filter_is_volatile_applied_after_distinct_on(self) -> None: + # Regression guard for the exact bug shape #2464 warns about: a + # sold-out object whose ONLY 'Строящиеся' row is an old snapshot must + # not be counted. That requires site_status to be filtered on the + # OUTER (post-DISTINCT-ON) query, never inside the `latest` CTE's + # WHERE — else DISTINCT ON would pick "the latest snapshot that still + # matched site_status='Строящиеся'" instead of the object's true + # latest snapshot. (Prod per-obj_id variance: site_status=11.) + db = _mock_scalar_db(0) + _active_competitors_count( + db, region_code=66, district_name="Ленинский", target_class="Комфорт" + ) + cte_body, outer_body = _split_latest_cte(_executed_sql(db, 0)) + # site_status legitimately appears in the CTE's SELECT list (it's the + # column DISTINCT ON needs to expose) -- what must NOT appear is a + # filter predicate on it inside the CTE's WHERE. + assert ( + "site_status = 'Строящиеся'" not in cte_body + ), f"site_status must not pre-filter the DISTINCT ON CTE (volatile field):\n{cte_body}" + assert "site_status = 'Строящиеся'" in outer_body + + def test_district_and_class_are_volatile_applied_after_distinct_on(self) -> None: + # Companion to the site_status guard: authoritative prod measurement + # (coordinator 2026-07-08) — per-obj_id variance across snapshots is + # district_name=1 and obj_class=29 (NON-zero → VOLATILE), unlike + # dev_id/region_cd (0 → stable). So district_name/obj_class predicates + # must ALSO be applied on the true-latest row (outer WHERE), never + # pre-filtered inside the DISTINCT ON CTE — else a ЖК that changed + # class/district gets counted by a stale snapshot. Tier-1 + # (district+class) is call_index 0 when target_class is given. + db = _mock_scalar_db(0) + _active_competitors_count( + db, region_code=66, district_name="Ленинский", target_class="Комфорт" + ) + cte_body, outer_body = _split_latest_cte(_executed_sql(db, 0)) + # CTE WHERE must scope on ONLY the stable region_cd — no volatile predicate. + assert ( + "district_name = :dn" not in cte_body + ), f"district_name (volatile) must not pre-filter the DISTINCT ON CTE:\n{cte_body}" + assert ( + "COALESCE(obj_class, obj_class_fallback) = :cls" not in cte_body + ), f"obj_class (volatile) must not pre-filter the DISTINCT ON CTE:\n{cte_body}" + # Both live in the outer WHERE, applied to the deduped true-latest row. + assert "district_name = :dn" in outer_body + assert "COALESCE(obj_class, obj_class_fallback) = :cls" in outer_body + + def test_cte_where_scopes_only_stable_region_cd(self) -> None: + # Belt-and-suspenders: the DISTINCT ON CTE's WHERE partition-scope is + # exactly `region_cd = :rc` and nothing else (all volatile predicates + # relocated). Guards against a future re-introduction of a volatile + # pre-filter that would silently pick a stale-but-matching snapshot. + db = _mock_scalar_db(0) + _active_competitors_count( + db, region_code=66, district_name="Ленинский", target_class="Комфорт" + ) + cte_body, _outer = _split_latest_cte(_executed_sql(db, 0)) + # extract the CTE's WHERE clause (between WHERE and the feeding ORDER BY, + # which _split_latest_cte leaves at the tail of the CTE body). + where_start = cte_body.index("WHERE ") + len("WHERE ") + where_end = cte_body.index("ORDER BY", where_start) + where_clause = cte_body[where_start:where_end].strip() + assert ( + where_clause == "region_cd = :rc" + ), f"CTE WHERE must scope on ONLY stable region_cd, got: {where_clause!r}" + + def test_no_double_colon_cast(self) -> None: + import re + + db = _mock_scalar_db(0) + _active_competitors_count(db, region_code=66, district_name="Ленинский", target_class=None) + sql = _executed_sql(db, 0) + assert re.search(r":[a-z_]+::[a-z]", sql) is None + + def test_cascade_still_stops_early_when_tier_hits_threshold(self) -> None: + # Sanity: fallback cascade logic (n>=2 -> stop) is untouched by the dedup fix. + db = _mock_scalar_db(5) + n, scope = _active_competitors_count( + db, region_code=66, district_name="Ленинский", target_class="Комфорт" + ) + assert n == 5 + assert scope == "district+class" + assert db.execute.call_count == 1 + + +# ── developer_portfolio SQL shape ────────────────────────────────────────── + + +class TestDeveloperPortfolioSqlShape: + def test_dedups_via_distinct_on_obj_id_latest_snapshot(self) -> None: + db = _mock_mapping_db([]) + developer_portfolio(db, "6208_0") + sql = _executed_sql(db, 0) + norm = " ".join(sql.split()) + assert "DISTINCT ON (obj_id)" in norm + assert "ORDER BY obj_id, snapshot_date DESC NULLS LAST" in norm + + def test_preserves_final_ready_dt_ordering(self) -> None: + # DISTINCT ON's own ORDER BY must start with (obj_id, snapshot_date) — + # the caller-facing ready_dt ordering has to live in the OUTER SELECT. + db = _mock_mapping_db([]) + developer_portfolio(db, "6208_0") + sql = _executed_sql(db, 0) + norm = " ".join(sql.split()) + assert norm.rstrip().endswith("ORDER BY ready_dt DESC NULLS LAST") + + def test_dev_id_filter_unchanged(self) -> None: + db = _mock_mapping_db([]) + developer_portfolio(db, "6208_0") + params = _executed_params(db, 0) + assert params == {"dev": "6208_0"} + + def test_returned_columns_unchanged(self) -> None: + # Fix must not add/drop columns from the caller-facing shape. + rows = [ + { + "obj_id": 1, + "comm_name": "ЖК Тест", + "addr": "ул. Тест, 1", + "region_cd": 66, + "flat_count": 100, + "square_living": 5000.0, + "ready_dt": None, + "obj_class": "Комфорт", + "escrow": True, + "problem_flag": None, + "latitude": 56.8, + "longitude": 60.6, + "is_ekb": True, + } + ] + db = _mock_mapping_db(rows) + out = developer_portfolio(db, "6208_0") + assert len(out) == 1 + assert set(out[0].keys()) == { + "obj_id", + "comm_name", + "addr", + "region_cd", + "flat_count", + "square_living", + "ready_dt", + "obj_class", + "escrow", + "problem_flag", + "lat", + "lon", + "is_ekb", + } + + +# ── _competitors_two_dim `active`/`latest` CTE shape ─────────────────────── + + +def _mock_two_dim_db(radius_n: int, district_only_n: int) -> MagicMock: + """Session stand-in for _competitors_two_dim: call 0 = centroid lookup + (returns a non-null WKT so the main query runs), call 1 = the radius/ + district aggregate. radius_n/district_only_n chosen so total_weighted >= 1 + keeps execution off the single-dim fallback path (which would add calls).""" + db = MagicMock() + centroid_res = MagicMock() + centroid_res.mappings.return_value.first.return_value = {"centroid_wkt": "POINT(60.6 56.8)"} + main_res = MagicMock() + main_res.mappings.return_value.first.return_value = { + "radius_n": radius_n, + "district_only_n": district_only_n, + } + db.execute.side_effect = [centroid_res, main_res] + return db + + +class TestCompetitorsTwoDimActiveCte: + """#2464 sibling: _competitors_two_dim's `active` CTE had site_status / + district_name / obj_class ALL inside its DISTINCT ON — same latest-matching- + not-latest bug. Fix dedups to true-latest per obj_id first (region_cd-only + CTE scope), then applies the volatile predicates.""" + + def _main_sql(self, db: MagicMock) -> str: + # call_index 1 is the radius/district aggregate (0 is the centroid lookup). + return " ".join(_executed_sql(db, 1).split()) + + def test_dedups_via_distinct_on_before_volatile_filters(self) -> None: + from app.services.analytics_queries import _competitors_two_dim + + db = _mock_two_dim_db(radius_n=3, district_only_n=0) + _competitors_two_dim(db, region_code=66, district_name="Ленинский", target_class="Комфорт") + sql = self._main_sql(db) + assert "DISTINCT ON (obj_id)" in sql + assert "ORDER BY obj_id, snapshot_date DESC NULLS LAST" in sql + + def test_volatile_filters_are_post_distinct_on(self) -> None: + from app.services.analytics_queries import _competitors_two_dim + + db = _mock_two_dim_db(radius_n=3, district_only_n=0) + _competitors_two_dim(db, region_code=66, district_name="Ленинский", target_class="Комфорт") + sql = self._main_sql(db) + # Split the DISTINCT ON CTE (`latest`) from everything downstream. Its + # WHERE must scope on ONLY stable region_cd; site_status/district_name/ + # obj_class predicates live in the downstream `active` CTE. + latest_open = sql.index("WITH latest AS (") + order_by_idx = sql.index("ORDER BY obj_id, snapshot_date DESC NULLS LAST") + latest_close = sql.index(")", order_by_idx) + latest_body = sql[latest_open:latest_close] + downstream = sql[latest_close:] + + assert "site_status = 'Строящиеся'" not in latest_body + assert "district_name = :dn" not in latest_body + assert "COALESCE(obj_class, obj_class_fallback) = :cls" not in latest_body + + assert "site_status = 'Строящиеся'" in downstream + assert "district_name = :dn" in downstream + assert "COALESCE(obj_class, obj_class_fallback) = :cls" in downstream + + def test_output_shape_preserved(self) -> None: + from app.services.analytics_queries import _competitors_two_dim + + db = _mock_two_dim_db(radius_n=3, district_only_n=2) + radius_n, district_only_n, total_weighted, scope = _competitors_two_dim( + db, region_code=66, district_name="Ленинский", target_class="Комфорт" + ) + assert radius_n == 3 + assert district_only_n == 2 + # 3*1.0 + 2*0.6 = 4.2 + assert total_weighted == 4.2 + assert scope == "district_2d" + + +# ── Pure-Python spec replica: exact dedup + volatile-filter algorithm ───── +# +# MagicMock can't execute real SQL (and DISTINCT ON has no SQLite equivalent), +# so this class locks in the ALGORITHM the SQL is required to implement: +# "take the row with MAX(snapshot_date) per obj_id, THEN filter on that row's +# site_status" — as opposed to the buggy "filter rows on site_status, then +# take MAX(snapshot_date) among survivors". It's a spec/regression guard, not +# a live-DB execution test (see module docstring for why). + + +def _latest_snapshot_active_count(rows: list[dict[str, Any]]) -> int: + """Reference implementation of what the fixed `_q` SQL computes.""" + latest_by_obj: dict[int, dict[str, Any]] = {} + for row in rows: + obj_id = row["obj_id"] + cur = latest_by_obj.get(obj_id) + if cur is None or (row["snapshot_date"] or "") > (cur["snapshot_date"] or ""): + latest_by_obj[obj_id] = row + return sum(1 for r in latest_by_obj.values() if r["site_status"] == "Строящиеся") + + +class TestDedupAlgorithmSpec: + def test_multi_snapshot_same_obj_id_counted_once(self) -> None: + rows = [ + {"obj_id": 1, "snapshot_date": "2026-04-27", "site_status": "Строящиеся"}, + {"obj_id": 1, "snapshot_date": "2026-05-25", "site_status": "Строящиеся"}, + {"obj_id": 1, "snapshot_date": "2026-06-28", "site_status": "Строящиеся"}, + ] + assert _latest_snapshot_active_count(rows) == 1 + + def test_sold_out_object_with_old_active_snapshot_not_counted(self) -> None: + # obj_id=2's LATEST snapshot says "Реализован" (sold out) even though + # older snapshots said "Строящиеся" — must NOT count as active. + rows = [ + {"obj_id": 2, "snapshot_date": "2026-04-27", "site_status": "Строящиеся"}, + {"obj_id": 2, "snapshot_date": "2026-05-25", "site_status": "Строящиеся"}, + {"obj_id": 2, "snapshot_date": "2026-06-28", "site_status": "Реализован"}, + ] + assert _latest_snapshot_active_count(rows) == 0 + + def test_mixed_objects_only_latest_active_ones_counted(self) -> None: + rows = [ + # obj 1: still active across all 3 snapshots -> counts once. + {"obj_id": 1, "snapshot_date": "2026-04-27", "site_status": "Строящиеся"}, + {"obj_id": 1, "snapshot_date": "2026-06-28", "site_status": "Строящиеся"}, + # obj 2: sold out on the latest snapshot -> excluded. + {"obj_id": 2, "snapshot_date": "2026-04-27", "site_status": "Строящиеся"}, + {"obj_id": 2, "snapshot_date": "2026-06-28", "site_status": "Реализован"}, + # obj 3: single snapshot, active -> counts once. + {"obj_id": 3, "snapshot_date": "2026-06-28", "site_status": "Строящиеся"}, + ] + assert _latest_snapshot_active_count(rows) == 2