fix(tradein/geocode): прошить city_hint в deals-скрипт + развести счётчики гейта (#2603)
All checks were successful
CI Trade-In / changes (pull_request) Successful in 8s
CI / changes (pull_request) Successful in 8s
CI Trade-In / frontend-checks (pull_request) Has been skipped
CI / backend-tests (pull_request) Has been skipped
CI / frontend-tests (pull_request) Has been skipped
CI / openapi-codegen-check (pull_request) Has been skipped
CI Trade-In / backend-tests (pull_request) Successful in 2m46s
All checks were successful
CI Trade-In / changes (pull_request) Successful in 8s
CI / changes (pull_request) Successful in 8s
CI Trade-In / frontend-checks (pull_request) Has been skipped
CI / backend-tests (pull_request) Has been skipped
CI / frontend-tests (pull_request) Has been skipped
CI / openapi-codegen-check (pull_request) Has been skipped
CI Trade-In / backend-tests (pull_request) Successful in 2m46s
Хвосты после #2601 (замыкание петли «город → геокодер»). 1. scripts/geocode_deals_nominatim.py — непрошитый sibling-caller. Скрипт группировал `GROUP BY address` и звал `geocode(address, db)` без города, хотя deals.city (миграция 177) заполнена на 100%: один и тот же текст адреса из разных городов схлопывался в одну группу, один geocode-вызов и один UPDATE по тексту адреса. Теперь — та же форма, что в #2601: группировка по паре (address, city), city_hint в geocode(), UPDATE и mark-tried через `city IS NOT DISTINCT FROM` (обычное `=` не ловит NULL-город → NULL-группа не обновлялась бы вовсе). Хинт передаётся ТОЛЬКО для значений из geocoder.SVERDLOVSK_OBLAST_CITIES: deals.city росреестровое, в хвосте лежит мусор («Бессонова», «Бердюгина», «Билейский рыбопитомник»), а любой не-ЕКБ хинт жёстко закрывает EKB-локальные тиры и подставляется в запрос провайдеру — мусорный хинт хуже отсутствия хинта. Словарь переиспользован, а не заведён свой: тот же набор уже питает гейты самого геокодера (_names_non_ekb_city / _ekb_local_tiers_allowed) и estimator._resolve_target_city. 2. tasks/backfill_listings_coords_geoportal.py — наблюдаемость городского гейта. Добавлен skipped_non_ekb_by_column (+ в to_counters и в DONE-логи): колоночный гейт стоит перед парсером адреса, поэтому по мере раскатки областных развёрток (#2598) строки потекут из no_address в skipped_non_ekb и общий счётчик поменяет смысл ровно тогда, когда по нему валидируют раскатку. Старый счётчик не тронут — остаётся суммой обоих гейтов, вклад текстового считается разностью. 3. tests: test_admin_geocode_missing_passes_city_hint параметризован на target="deals" (колонка city есть в обеих таблицах, ветка была не покрыта). 4. tasks/geocode_missing.py: dry-run лог печатает city — он с #2594 часть ключа группы, без него две строки dry-run неотличимы. Refs #2603
This commit is contained in:
parent
d362b16d7c
commit
ec17886d60
6 changed files with 382 additions and 86 deletions
|
|
@ -85,7 +85,17 @@ class BackfillCoordsResult:
|
|||
updated: int = 0 # реально обновлено (UPDATE rowcount)
|
||||
no_address: int = 0 # listing.address IS NULL / не распарсился
|
||||
no_match: int = 0 # адрес распарсился, но в реестре здания нет
|
||||
skipped_non_ekb: int = 0 # non-ЕКБ гейт: колонка city (#2594) ИЛИ текст адреса (#2583)
|
||||
skipped_non_ekb: int = 0 # non-ЕКБ гейт ВСЕГО: колонка city (#2594) ИЛИ текст (#2583)
|
||||
# Подмножество skipped_non_ekb — только те, кого отсёк гейт по КОЛОНКЕ city
|
||||
# (#2603). Зачем отдельный счётчик: колоночный гейт стоит ПЕРЕД парсером
|
||||
# адреса, поэтому по мере раскатки областных развёрток (#2598) строки, которые
|
||||
# сейчас падают в no_address (город неизвестен, адрес не парсится), начнут
|
||||
# перетекать в skipped_non_ekb — и общий счётчик поменяет смысл ровно тогда,
|
||||
# когда по нему хотят валидировать раскатку. Разность
|
||||
# skipped_non_ekb - skipped_non_ekb_by_column = вклад ТЕКСТОВОГО гейта, т.е.
|
||||
# старая метрика #2583 остаётся вычислимой. Обратная совместимость:
|
||||
# skipped_non_ekb продолжает означать то же, что и раньше (гейт целиком).
|
||||
skipped_non_ekb_by_column: int = 0
|
||||
errors: int = 0 # исключения при обработке отдельной записи
|
||||
duration_sec: float = field(default=0.0)
|
||||
|
||||
|
|
@ -97,6 +107,7 @@ class BackfillCoordsResult:
|
|||
"no_address": self.no_address,
|
||||
"no_match": self.no_match,
|
||||
"skipped_non_ekb": self.skipped_non_ekb,
|
||||
"skipped_non_ekb_by_column": self.skipped_non_ekb_by_column,
|
||||
"errors": self.errors,
|
||||
"duration_sec": int(self.duration_sec),
|
||||
}
|
||||
|
|
@ -227,6 +238,10 @@ def backfill_coords_from_geoportal(
|
|||
city: str | None = row.get("city")
|
||||
if city is not None and city != "Екатеринбург":
|
||||
res.skipped_non_ekb += 1
|
||||
# Отдельный срез (#2603) — общий skipped_non_ekb смешивает
|
||||
# колоночный и текстовый гейты, а по мере раскатки #2598
|
||||
# колоночный будет забирать строки из no_address.
|
||||
res.skipped_non_ekb_by_column += 1
|
||||
continue
|
||||
|
||||
# Текстовый гейт (#2583, H3) — fallback для листингов, у которых
|
||||
|
|
@ -313,13 +328,15 @@ def backfill_coords_from_geoportal(
|
|||
|
||||
logger.info(
|
||||
"backfill_coords: DONE — candidates=%d matched=%d updated=%d "
|
||||
"no_address=%d no_match=%d skipped_non_ekb=%d errors=%d duration=%.1fs",
|
||||
"no_address=%d no_match=%d skipped_non_ekb=%d (by_column=%d) "
|
||||
"errors=%d duration=%.1fs",
|
||||
res.candidates,
|
||||
res.matched,
|
||||
res.updated,
|
||||
res.no_address,
|
||||
res.no_match,
|
||||
res.skipped_non_ekb,
|
||||
res.skipped_non_ekb_by_column,
|
||||
res.errors,
|
||||
res.duration_sec,
|
||||
)
|
||||
|
|
@ -369,7 +386,8 @@ def run_geoportal_coords_backfill(
|
|||
runs_mod.mark_done(db, run_id, counters)
|
||||
logger.info(
|
||||
"run_geoportal_coords_backfill: run_id=%d DONE candidates=%d matched=%d "
|
||||
"updated=%d no_address=%d no_match=%d skipped_non_ekb=%d errors=%d duration=%.1fs",
|
||||
"updated=%d no_address=%d no_match=%d skipped_non_ekb=%d (by_column=%d) "
|
||||
"errors=%d duration=%.1fs",
|
||||
run_id,
|
||||
res.candidates,
|
||||
res.matched,
|
||||
|
|
@ -377,6 +395,7 @@ def run_geoportal_coords_backfill(
|
|||
res.no_address,
|
||||
res.no_match,
|
||||
res.skipped_non_ekb,
|
||||
res.skipped_non_ekb_by_column,
|
||||
res.errors,
|
||||
res.duration_sec,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -223,9 +223,13 @@ async def geocode_missing_listings(
|
|||
result.addresses_geocoded += 1
|
||||
|
||||
if dry_run:
|
||||
# city в логе (#2603) — с #2594 это часть ключа группы: без него две
|
||||
# строки dry-run с одинаковым текстом адреса неотличимы друг от друга.
|
||||
logger.info(
|
||||
"geocode_missing[dry]: '%s' → (%.5f, %.5f) provider=%s would update %d listings",
|
||||
"geocode_missing[dry]: '%s' city=%r → (%.5f, %.5f) provider=%s "
|
||||
"would update %d listings",
|
||||
address[:60],
|
||||
city,
|
||||
geo.lat,
|
||||
geo.lon,
|
||||
geo.provider,
|
||||
|
|
|
|||
|
|
@ -18,12 +18,30 @@ result into `geocode_cache`. We do NOT call Nominatim directly — that keeps a
|
|||
single source of truth for provider order, ЕКБ bbox filtering, and the 1 req/sec
|
||||
policy.
|
||||
|
||||
Why dedup by address (not one call per row)
|
||||
-------------------------------------------
|
||||
`deals.address` is street-only ('Екатеринбург, <street>'), so thousands of rows
|
||||
share the same address. The geocoder caches by normalized address, but we also
|
||||
GROUP BY address up front so the real call count is driven by DISTINCT streets,
|
||||
not the ~6,951 row backlog. One geocode call → UPDATE every deal on that street.
|
||||
Why dedup by (address, city) — not one call per row
|
||||
---------------------------------------------------
|
||||
`deals.address` is street-only ('<city>, <street>'), so thousands of rows share
|
||||
the same address. The geocoder caches by normalized address, but we also GROUP
|
||||
BY up front so the real call count is driven by DISTINCT streets, not the
|
||||
~6,951 row backlog. One geocode call → UPDATE every deal on that street.
|
||||
|
||||
Grouping is by the PAIR (address, city), not by address alone (#2603, same
|
||||
shape as #2601 fixed in app/tasks/geocode_missing.py): the same address text in
|
||||
two different cities must not collapse into one geocode call, and the UPDATE
|
||||
must not spill onto the other city's rows — hence `city IS NOT DISTINCT FROM`
|
||||
(plain `=` never matches a NULL city, so a NULL-city group would update nothing).
|
||||
|
||||
City hint (and why it is filtered)
|
||||
----------------------------------
|
||||
`deals.city` comes from Rosreestr (migration 177) and is filled on 100% of the
|
||||
rows, so a hint is available for every group — but its long tail holds values
|
||||
that are not cities at all ('Бессонова', 'Бердюгина', 'Билейский рыбопитомник').
|
||||
A non-EKB hint is a HARD signal inside the geocoder: it closes the EKB-only
|
||||
local tiers (`_ekb_local_tiers_allowed`) and gets prefixed into the provider
|
||||
query, so a junk hint makes the result strictly WORSE than no hint at all.
|
||||
We therefore pass the hint only for values that are recognised cities of oblast
|
||||
66 (`_city_hint` below); everything else degrades to the previous behaviour
|
||||
(no hint, address text only).
|
||||
|
||||
Street-level precision is accepted: the estimator's comparable search uses a
|
||||
1000-2000 m radius, so a street-level point lands every deal on that street in
|
||||
|
|
@ -87,6 +105,7 @@ except ImportError: # pragma: no cover — fallback for adhoc invocation
|
|||
|
||||
try:
|
||||
from app.services.geocoder import ( # type: ignore[import-not-found]
|
||||
SVERDLOVSK_OBLAST_CITIES,
|
||||
GeocodeResult,
|
||||
geocode,
|
||||
)
|
||||
|
|
@ -94,7 +113,7 @@ except ImportError: # pragma: no cover
|
|||
import sys
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||||
from app.services.geocoder import GeocodeResult, geocode
|
||||
from app.services.geocoder import SVERDLOVSK_OBLAST_CITIES, GeocodeResult, geocode
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
|
|
@ -122,17 +141,43 @@ _LOG_EVERY = 25
|
|||
|
||||
@dataclass
|
||||
class AddressGroup:
|
||||
"""One distinct deals.address and how many lat-IS-NULL rows share it."""
|
||||
"""One distinct (deals.address, deals.city) pair + its lat-IS-NULL row count.
|
||||
|
||||
`city` is the raw Rosreestr value (may be NULL for non-rosreestr sources, and
|
||||
may be junk — see `_city_hint`), kept verbatim so the UPDATE can target
|
||||
exactly the rows this group came from.
|
||||
"""
|
||||
|
||||
address: str
|
||||
deals_count: int
|
||||
city: str | None = None
|
||||
|
||||
|
||||
def _city_hint(city: str | None) -> str | None:
|
||||
"""`city` if it is a recognised oblast-66 city name, else None (#2603).
|
||||
|
||||
Dictionary = `geocoder.SVERDLOVSK_OBLAST_CITIES` — the single source of truth
|
||||
already used by the geocoder's own city gates (`_names_non_ekb_city`,
|
||||
`_ekb_local_tiers_allowed`) and by `estimator._resolve_target_city`. No local
|
||||
list: a second one would drift from the gates the hint actually feeds.
|
||||
|
||||
Filtering is not optional politeness: `deals.city` is Rosreestr-sourced and
|
||||
its tail holds non-cities ('Бессонова', 'Бердюгина', 'Билейский
|
||||
рыбопитомник'). Any non-EKB hint closes the EKB-only local tiers and is
|
||||
prefixed into the provider query, so an unrecognised value would make the
|
||||
geocode worse than passing nothing.
|
||||
"""
|
||||
if not city:
|
||||
return None
|
||||
normalized = " ".join(city.lower().split())
|
||||
return city if normalized in SVERDLOVSK_OBLAST_CITIES else None
|
||||
|
||||
|
||||
@dataclass
|
||||
class Stats:
|
||||
"""Final-summary counters.
|
||||
|
||||
- processed distinct addresses fed to the geocoder this run.
|
||||
- processed distinct (address, city) groups fed to the geocoder this run.
|
||||
- geocoded addresses the geocoder resolved to coords.
|
||||
- geocode_failed addresses the geocoder returned None for (still stamped).
|
||||
- skipped distinct addresses skipped before any geocode call
|
||||
|
|
@ -156,10 +201,8 @@ class Stats:
|
|||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _select_pending_addresses(
|
||||
db: Session, *, limit: int, stale_days: int
|
||||
) -> list[AddressGroup]:
|
||||
"""Distinct deals.address still needing coords — resume-safe candidate set.
|
||||
def _select_pending_addresses(db: Session, *, limit: int, stale_days: int) -> list[AddressGroup]:
|
||||
"""Distinct (address, city) pairs still needing coords — resume-safe set.
|
||||
|
||||
Combines the `deals_geocode_pending_idx` partial index predicate
|
||||
(`lat IS NULL`) with a staleness filter so failed/un-geocodable addresses
|
||||
|
|
@ -167,44 +210,68 @@ def _select_pending_addresses(
|
|||
|
||||
The GROUP BY collapses the ~6,951-row backlog into its distinct streets;
|
||||
`:limit` caps the deal ROWS fanned out, computed from the running SUM of
|
||||
per-address counts so a single huge street can't blow past the cap. We
|
||||
order by occurrence DESC (biggest ROI per geocode call first) then address
|
||||
for a deterministic resume order.
|
||||
per-group counts so a single huge street can't blow past the cap. We order
|
||||
by occurrence DESC (biggest ROI per geocode call first) then address, city
|
||||
for a deterministic resume order (the window ORDER BY must match the outer
|
||||
one, otherwise the running_rows cap slices a different ordering).
|
||||
|
||||
GROUP BY address, city — NOT address alone (#2603, the shape #2601 fixed in
|
||||
app/tasks/geocode_missing.py): the same street name in two cities is two
|
||||
geocode calls with two different hints, not one call whose result lands on
|
||||
both. SQL groups NULL cities together, so a NULL-city group stays its own
|
||||
group rather than merging into an arbitrary city.
|
||||
"""
|
||||
rows = db.execute(
|
||||
text(
|
||||
"SELECT address, deals_count FROM ("
|
||||
" SELECT address, "
|
||||
" COUNT(*) AS deals_count, "
|
||||
" SUM(COUNT(*)) OVER ("
|
||||
" ORDER BY COUNT(*) DESC, address ASC"
|
||||
" ) AS running_rows "
|
||||
" FROM deals "
|
||||
" WHERE lat IS NULL "
|
||||
" AND address IS NOT NULL "
|
||||
" AND length(trim(address)) >= 3 "
|
||||
" AND (geocode_tried_at IS NULL "
|
||||
" OR geocode_tried_at < NOW() "
|
||||
" - make_interval(days => CAST(:stale_days AS int))) "
|
||||
" GROUP BY address "
|
||||
") g "
|
||||
"WHERE running_rows - deals_count < CAST(:limit AS int) "
|
||||
"ORDER BY deals_count DESC, address ASC"
|
||||
),
|
||||
{"limit": limit, "stale_days": stale_days},
|
||||
).mappings().all()
|
||||
return [AddressGroup(address=r["address"], deals_count=r["deals_count"]) for r in rows]
|
||||
rows = (
|
||||
db.execute(
|
||||
text(
|
||||
"SELECT address, city, deals_count FROM ("
|
||||
" SELECT address, city, "
|
||||
" COUNT(*) AS deals_count, "
|
||||
" SUM(COUNT(*)) OVER ("
|
||||
" ORDER BY COUNT(*) DESC, address ASC, city ASC NULLS FIRST"
|
||||
" ) AS running_rows "
|
||||
" FROM deals "
|
||||
" WHERE lat IS NULL "
|
||||
" AND address IS NOT NULL "
|
||||
" AND length(trim(address)) >= 3 "
|
||||
" AND (geocode_tried_at IS NULL "
|
||||
" OR geocode_tried_at < NOW() "
|
||||
" - make_interval(days => CAST(:stale_days AS int))) "
|
||||
" GROUP BY address, city "
|
||||
") g "
|
||||
"WHERE running_rows - deals_count < CAST(:limit AS int) "
|
||||
"ORDER BY deals_count DESC, address ASC, city ASC NULLS FIRST"
|
||||
),
|
||||
{"limit": limit, "stale_days": stale_days},
|
||||
)
|
||||
.mappings()
|
||||
.all()
|
||||
)
|
||||
return [
|
||||
AddressGroup(
|
||||
address=r["address"],
|
||||
city=r.get("city"),
|
||||
deals_count=r["deals_count"],
|
||||
)
|
||||
for r in rows
|
||||
]
|
||||
|
||||
|
||||
def _count_pending_total(db: Session, *, stale_days: int) -> tuple[int, int]:
|
||||
"""Full backlog: (distinct addresses, total rows) eligible this pass.
|
||||
"""Full backlog: (distinct (address, city) pairs, total rows) this pass.
|
||||
|
||||
Denominators for the dry-run projection — counts every lat-IS-NULL deal
|
||||
that passes the staleness filter, ignoring --limit.
|
||||
that passes the staleness filter, ignoring --limit. Counts PAIRS, matching
|
||||
what `_select_pending_addresses` actually feeds the geocoder (#2603) —
|
||||
counting distinct addresses here would understate the denominator and the
|
||||
projection would read above 100%. COALESCE(city, '') keeps a NULL-city pair
|
||||
countable: `count()` skips NULL inputs, and a bare row expression with a
|
||||
NULL field invites exactly that argument — the empty string can't collide
|
||||
with a real city name, so the pair count stays honest either way.
|
||||
"""
|
||||
row = db.execute(
|
||||
text(
|
||||
"SELECT COUNT(DISTINCT address) AS streets, COUNT(*) AS rows "
|
||||
"SELECT COUNT(DISTINCT (address, COALESCE(city, ''))) AS streets, COUNT(*) AS rows "
|
||||
"FROM deals "
|
||||
"WHERE lat IS NULL "
|
||||
" AND address IS NOT NULL "
|
||||
|
|
@ -225,8 +292,10 @@ def _count_pending_total(db: Session, *, stale_days: int) -> tuple[int, int]:
|
|||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _update_deals_geocoded(db: Session, *, address: str, lat: float, lon: float) -> int:
|
||||
"""UPDATE every lat-IS-NULL deal on `address`; geom auto-fills via trigger.
|
||||
def _update_deals_geocoded(
|
||||
db: Session, *, address: str, city: str | None, lat: float, lon: float
|
||||
) -> int:
|
||||
"""UPDATE every lat-IS-NULL deal on (address, city); geom via trigger.
|
||||
|
||||
The `deals_set_geom_trg` BEFORE UPDATE OF lat, lon trigger
|
||||
(002_core_tables.sql, reuses listings_set_geom()) populates geom from the
|
||||
|
|
@ -234,6 +303,12 @@ def _update_deals_geocoded(db: Session, *, address: str, lat: float, lon: float)
|
|||
so the row drops out of the candidate set. The `AND lat IS NULL` guard
|
||||
keeps this idempotent and avoids clobbering rows another pass already set.
|
||||
|
||||
`city IS NOT DISTINCT FROM` (not `=`) scopes the write to the group the
|
||||
geocode was made for: plain `=` is never true for a NULL city, so a
|
||||
NULL-city group would update zero rows and re-run forever; `IS NOT DISTINCT
|
||||
FROM` treats NULL=NULL as a match while staying strict for a real city, so
|
||||
the same street text in another city keeps its own coords (#2603).
|
||||
|
||||
Returns the number of deal rows updated.
|
||||
"""
|
||||
result = db.execute(
|
||||
|
|
@ -243,14 +318,15 @@ def _update_deals_geocoded(db: Session, *, address: str, lat: float, lon: float)
|
|||
" lon = CAST(:lon AS double precision), "
|
||||
" geocode_tried_at = NOW() "
|
||||
" WHERE address = CAST(:addr AS text) "
|
||||
" AND city IS NOT DISTINCT FROM CAST(:city AS text) "
|
||||
" AND lat IS NULL"
|
||||
),
|
||||
{"addr": address, "lat": lat, "lon": lon},
|
||||
{"addr": address, "city": city, "lat": lat, "lon": lon},
|
||||
)
|
||||
return result.rowcount
|
||||
|
||||
|
||||
def _mark_deals_tried(db: Session, *, address: str) -> int:
|
||||
def _mark_deals_tried(db: Session, *, address: str, city: str | None) -> int:
|
||||
"""Stamp `geocode_tried_at = NOW()` WITHOUT touching lat/lon (geocode miss).
|
||||
|
||||
Critical for resume: an address the geocoder can't resolve must still drop
|
||||
|
|
@ -259,6 +335,9 @@ def _mark_deals_tried(db: Session, *, address: str) -> int:
|
|||
it for `stale_days`. The `AND lat IS NULL` guard means a concurrent success
|
||||
can't be downgraded.
|
||||
|
||||
Scoped to the (address, city) pair for the same reason as the coords write
|
||||
— a miss in one city must not defer the other city's retry (#2603).
|
||||
|
||||
Returns the number of deal rows stamped.
|
||||
"""
|
||||
result = db.execute(
|
||||
|
|
@ -266,9 +345,10 @@ def _mark_deals_tried(db: Session, *, address: str) -> int:
|
|||
"UPDATE deals "
|
||||
" SET geocode_tried_at = NOW() "
|
||||
" WHERE address = CAST(:addr AS text) "
|
||||
" AND city IS NOT DISTINCT FROM CAST(:city AS text) "
|
||||
" AND lat IS NULL"
|
||||
),
|
||||
{"addr": address},
|
||||
{"addr": address, "city": city},
|
||||
)
|
||||
return result.rowcount
|
||||
|
||||
|
|
@ -285,7 +365,7 @@ async def _run_backfill(
|
|||
batch: str,
|
||||
dry_run: bool,
|
||||
) -> Stats:
|
||||
"""For each distinct address: geocode once, then UPDATE all its deals.
|
||||
"""For each (address, city) pair: geocode once, then UPDATE all its deals.
|
||||
|
||||
Per-address SAVEPOINT (`db.begin_nested()`) so one bad UPDATE can't abort
|
||||
the batch (backend.md SAVEPOINT rule). Per-address commit on success → a
|
||||
|
|
@ -301,6 +381,11 @@ async def _run_backfill(
|
|||
|
||||
for i, group in enumerate(groups, start=1):
|
||||
address = group.address
|
||||
city = group.city
|
||||
# Only a recognised oblast-66 city is fed to the geocoder; junk Rosreestr
|
||||
# values degrade to None (see `_city_hint`). The raw `city` is still used
|
||||
# for the UPDATE scope — it identifies the group either way.
|
||||
hint = _city_hint(city)
|
||||
|
||||
# The geocoder itself rejects <3 chars, but skip here too so the dry-run
|
||||
# report and counters stay honest (no phantom "processed" address).
|
||||
|
|
@ -310,13 +395,14 @@ async def _run_backfill(
|
|||
|
||||
result: GeocodeResult | None = None
|
||||
try:
|
||||
result = await geocode(address, db)
|
||||
result = await geocode(address, db, city_hint=hint)
|
||||
except Exception as exc: # defensive — one geocode error must not kill batch
|
||||
stats.geocode_failed += 1
|
||||
stats.processed += 1
|
||||
logger.warning(
|
||||
"geocode raised for addr=%r (%d deals): %s",
|
||||
"geocode raised for addr=%r city=%r (%d deals): %s",
|
||||
address[:60],
|
||||
city,
|
||||
group.deals_count,
|
||||
exc,
|
||||
)
|
||||
|
|
@ -331,18 +417,21 @@ async def _run_backfill(
|
|||
stats.processed += 1
|
||||
if dry_run:
|
||||
logger.info(
|
||||
"DRY-RUN addr=%r (%d deals) → NOT FOUND (would stamp tried)",
|
||||
"DRY-RUN addr=%r city=%r (%d deals) → NOT FOUND (would stamp tried)",
|
||||
address[:60],
|
||||
city,
|
||||
group.deals_count,
|
||||
)
|
||||
else:
|
||||
try:
|
||||
with db.begin_nested():
|
||||
_mark_deals_tried(db, address=address)
|
||||
_mark_deals_tried(db, address=address, city=city)
|
||||
db.commit()
|
||||
except Exception as exc: # defensive — isolate one bad UPDATE
|
||||
db.rollback()
|
||||
logger.warning("mark_tried failed for addr=%r: %s", address[:60], exc)
|
||||
logger.warning(
|
||||
"mark_tried failed for addr=%r city=%r: %s", address[:60], city, exc
|
||||
)
|
||||
_maybe_log_progress(i, groups, batch, stats)
|
||||
continue
|
||||
|
||||
|
|
@ -356,8 +445,9 @@ async def _run_backfill(
|
|||
|
||||
if dry_run:
|
||||
logger.info(
|
||||
"DRY-RUN addr=%r → (%.5f, %.5f) provider=%s would update %d deals",
|
||||
"DRY-RUN addr=%r city=%r → (%.5f, %.5f) provider=%s would update %d deals",
|
||||
address[:60],
|
||||
city,
|
||||
result.lat,
|
||||
result.lon,
|
||||
result.provider,
|
||||
|
|
@ -369,7 +459,7 @@ async def _run_backfill(
|
|||
try:
|
||||
with db.begin_nested():
|
||||
n = _update_deals_geocoded(
|
||||
db, address=address, lat=result.lat, lon=result.lon
|
||||
db, address=address, city=city, lat=result.lat, lon=result.lon
|
||||
)
|
||||
# Per-address commit so resume picks up exactly where we crashed.
|
||||
db.commit()
|
||||
|
|
@ -378,7 +468,7 @@ async def _run_backfill(
|
|||
db.rollback()
|
||||
# The geocode itself succeeded (and is cached); only the write
|
||||
# failed. Count the address as geocoded but log the write failure.
|
||||
logger.warning("db_write failed for addr=%r: %s", address[:60], exc)
|
||||
logger.warning("db_write failed for addr=%r city=%r: %s", address[:60], city, exc)
|
||||
|
||||
_maybe_log_progress(i, groups, batch, stats)
|
||||
|
||||
|
|
@ -513,12 +603,10 @@ async def main(argv: list[str] | None = None) -> int:
|
|||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
groups = _select_pending_addresses(
|
||||
db, limit=args.limit, stale_days=args.stale_days
|
||||
)
|
||||
groups = _select_pending_addresses(db, limit=args.limit, stale_days=args.stale_days)
|
||||
total_rows = sum(g.deals_count for g in groups)
|
||||
logger.info(
|
||||
"loaded %d distinct addresses (%d deal rows) needing coords",
|
||||
"loaded %d distinct (address, city) groups (%d deal rows) needing coords",
|
||||
len(groups),
|
||||
total_rows,
|
||||
)
|
||||
|
|
@ -533,9 +621,7 @@ async def main(argv: list[str] | None = None) -> int:
|
|||
stats = await _run_backfill(db, groups, batch=args.batch, dry_run=args.dry_run)
|
||||
|
||||
if args.dry_run:
|
||||
total_streets, backlog_rows = _count_pending_total(
|
||||
db, stale_days=args.stale_days
|
||||
)
|
||||
total_streets, backlog_rows = _count_pending_total(db, stale_days=args.stale_days)
|
||||
_report_dry_run(
|
||||
stats,
|
||||
total_streets=total_streets,
|
||||
|
|
|
|||
|
|
@ -12,6 +12,9 @@ Coverage (per the issue's test plan):
|
|||
batch.
|
||||
- dedup: distinct addresses drive geocode call count (one call per address).
|
||||
- main() wiring: SessionLocal, --limit bind, returns geocoded count.
|
||||
- #2603: grouping by the (address, city) PAIR, city_hint filtered against
|
||||
geocoder.SVERDLOVSK_OBLAST_CITIES (junk Rosreestr city → no hint), and
|
||||
writes scoped with `city IS NOT DISTINCT FROM`.
|
||||
|
||||
No real Postgres. `geocode()` is async → patched with AsyncMock (mirrors
|
||||
tests/tasks/test_geocode_missing.py). The Session is a MagicMock that routes
|
||||
|
|
@ -31,6 +34,7 @@ from app.services.geocoder import GeocodeResult
|
|||
from scripts.geocode_deals_nominatim import (
|
||||
AddressGroup,
|
||||
Stats,
|
||||
_city_hint,
|
||||
_mark_deals_tried,
|
||||
_run_backfill,
|
||||
_select_pending_addresses,
|
||||
|
|
@ -93,7 +97,7 @@ def _make_db_mock(
|
|||
tried_updates.append(dict(params) if params else {})
|
||||
result.rowcount = update_rowcount
|
||||
return result
|
||||
if "COUNT(DISTINCT address)" in sql_str:
|
||||
if "COUNT(DISTINCT" in sql_str:
|
||||
result.first.return_value = (total_streets, total_rows)
|
||||
return result
|
||||
if "FROM deals" in sql_str:
|
||||
|
|
@ -176,13 +180,16 @@ async def test_mark_deals_tried_sql_stamps_tried_at_only():
|
|||
"""The mark-tried writer sets geocode_tried_at and nothing else (no lat/lon)."""
|
||||
db = MagicMock()
|
||||
db.execute.return_value = MagicMock(rowcount=4)
|
||||
n = _mark_deals_tried(db, address="Екатеринбург, X")
|
||||
n = _mark_deals_tried(db, address="Екатеринбург, X", city="Екатеринбург")
|
||||
args, _kw = db.execute.call_args
|
||||
sql_str = str(args[0])
|
||||
assert "UPDATE deals" in sql_str
|
||||
assert "geocode_tried_at = NOW()" in sql_str
|
||||
assert "SET lat" not in sql_str
|
||||
assert "lat IS NULL" in sql_str # idempotency / no-clobber guard
|
||||
# #2603: a miss in one city must not defer the other city's retry.
|
||||
assert "city IS NOT DISTINCT FROM CAST(:city AS text)" in sql_str
|
||||
assert args[1]["city"] == "Екатеринбург"
|
||||
assert n == 4
|
||||
|
||||
|
||||
|
|
@ -312,7 +319,9 @@ async def test_run_backfill_db_write_failure_isolated_to_address():
|
|||
def test_update_deals_geocoded_sets_lat_lon_tried_at_not_geom():
|
||||
db = MagicMock()
|
||||
db.execute.return_value = MagicMock(rowcount=7)
|
||||
n = _update_deals_geocoded(db, address="Екатеринбург, Y", lat=56.1, lon=60.2)
|
||||
n = _update_deals_geocoded(
|
||||
db, address="Екатеринбург, Y", city="Екатеринбург", lat=56.1, lon=60.2
|
||||
)
|
||||
args, _kw = db.execute.call_args
|
||||
sql_str = str(args[0])
|
||||
binds = args[1]
|
||||
|
|
@ -321,10 +330,28 @@ def test_update_deals_geocoded_sets_lat_lon_tried_at_not_geom():
|
|||
# geom must NOT be set manually — the deals_set_geom_trg trigger fills it.
|
||||
assert "geom" not in sql_str
|
||||
assert "lat IS NULL" in sql_str # no-clobber guard
|
||||
assert binds == {"addr": "Екатеринбург, Y", "lat": 56.1, "lon": 60.2}
|
||||
# #2603: the write is scoped to the (address, city) group it was made for.
|
||||
assert "city IS NOT DISTINCT FROM CAST(:city AS text)" in sql_str
|
||||
assert binds == {
|
||||
"addr": "Екатеринбург, Y",
|
||||
"city": "Екатеринбург",
|
||||
"lat": 56.1,
|
||||
"lon": 60.2,
|
||||
}
|
||||
assert n == 7
|
||||
|
||||
|
||||
def test_update_deals_geocoded_null_city_group_uses_null_safe_predicate():
|
||||
"""A NULL-city group must still update its rows: plain `city = NULL` is never
|
||||
true, so `IS NOT DISTINCT FROM` is what keeps those rows reachable."""
|
||||
db = MagicMock()
|
||||
db.execute.return_value = MagicMock(rowcount=2)
|
||||
_update_deals_geocoded(db, address="Тестовая, 1", city=None, lat=56.1, lon=60.2)
|
||||
args, _kw = db.execute.call_args
|
||||
assert "city IS NOT DISTINCT FROM CAST(:city AS text)" in str(args[0])
|
||||
assert args[1]["city"] is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# resume filter — candidate SQL carries lat IS NULL + staleness predicate
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
@ -333,23 +360,19 @@ def test_update_deals_geocoded_sets_lat_lon_tried_at_not_geom():
|
|||
def test_select_pending_addresses_filters_null_and_stale():
|
||||
"""The candidate query must combine lat IS NULL with the staleness window
|
||||
and bind --stale-days, so already-tried-recently rows are excluded."""
|
||||
db, _, _ = _make_db_mock(
|
||||
address_rows=[{"address": "Екатеринбург, Z", "deals_count": 3}]
|
||||
)
|
||||
db, _, _ = _make_db_mock(address_rows=[{"address": "Екатеринбург, Z", "deals_count": 3}])
|
||||
|
||||
groups = _select_pending_addresses(db, limit=2000, stale_days=30)
|
||||
|
||||
assert groups == [AddressGroup(address="Екатеринбург, Z", deals_count=3)]
|
||||
# Inspect the SQL + binds of the SELECT.
|
||||
select_call = next(
|
||||
c for c in db.execute.call_args_list if "FROM deals" in str(c[0][0])
|
||||
)
|
||||
select_call = next(c for c in db.execute.call_args_list if "FROM deals" in str(c[0][0]))
|
||||
sql_str = str(select_call[0][0])
|
||||
binds = select_call[0][1]
|
||||
assert "lat IS NULL" in sql_str
|
||||
assert "geocode_tried_at IS NULL" in sql_str
|
||||
assert "make_interval(days => CAST(:stale_days AS int))" in sql_str
|
||||
assert "GROUP BY address" in sql_str
|
||||
assert "GROUP BY address, city" in sql_str # #2603 — pair, not address alone
|
||||
assert binds["stale_days"] == 30
|
||||
assert binds["limit"] == 2000
|
||||
|
||||
|
|
@ -357,9 +380,7 @@ def test_select_pending_addresses_filters_null_and_stale():
|
|||
def test_select_pending_addresses_passes_custom_stale_days():
|
||||
db, _, _ = _make_db_mock(address_rows=[])
|
||||
_select_pending_addresses(db, limit=500, stale_days=7)
|
||||
select_call = next(
|
||||
c for c in db.execute.call_args_list if "FROM deals" in str(c[0][0])
|
||||
)
|
||||
select_call = next(c for c in db.execute.call_args_list if "FROM deals" in str(c[0][0]))
|
||||
binds = select_call[0][1]
|
||||
assert binds["stale_days"] == 7
|
||||
assert binds["limit"] == 500
|
||||
|
|
@ -419,9 +440,7 @@ async def test_main_respects_limit_bind():
|
|||
):
|
||||
await main(["--limit", "50"])
|
||||
|
||||
select_call = next(
|
||||
c for c in db.execute.call_args_list if "FROM deals" in str(c[0][0])
|
||||
)
|
||||
select_call = next(c for c in db.execute.call_args_list if "FROM deals" in str(c[0][0]))
|
||||
assert select_call[0][1]["limit"] == 50
|
||||
|
||||
|
||||
|
|
@ -440,6 +459,117 @@ async def test_main_no_pending_returns_zero():
|
|||
db.close.assert_called_once()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# #2603 — (address, city) pair grouping + filtered city_hint
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_select_pending_addresses_returns_city_in_group():
|
||||
"""The candidate SELECT reads `city` and carries it into the group."""
|
||||
db, _, _ = _make_db_mock(
|
||||
address_rows=[
|
||||
{"address": "Нижний Тагил, Победы", "city": "Нижний Тагил", "deals_count": 3},
|
||||
{"address": "Тестовая, 1", "city": None, "deals_count": 1},
|
||||
]
|
||||
)
|
||||
|
||||
groups = _select_pending_addresses(db, limit=2000, stale_days=30)
|
||||
|
||||
assert groups == [
|
||||
AddressGroup(address="Нижний Тагил, Победы", deals_count=3, city="Нижний Тагил"),
|
||||
AddressGroup(address="Тестовая, 1", deals_count=1, city=None),
|
||||
]
|
||||
sql_str = str(next(c for c in db.execute.call_args_list if "FROM deals" in str(c[0][0]))[0][0])
|
||||
assert "SELECT address, city, deals_count" in sql_str
|
||||
|
||||
|
||||
def test_city_hint_keeps_known_oblast_city():
|
||||
"""A recognised oblast-66 city passes through as the hint (case/spacing-insensitive)."""
|
||||
assert _city_hint("Нижний Тагил") == "Нижний Тагил"
|
||||
assert _city_hint("екатеринбург") == "екатеринбург"
|
||||
assert _city_hint(" Каменск-Уральский ") == " Каменск-Уральский " # raw value kept
|
||||
|
||||
|
||||
def test_city_hint_drops_junk_rosreestr_value():
|
||||
"""Rosreestr tail values are NOT cities — a junk hint is worse than none.
|
||||
|
||||
Any non-EKB hint closes the geocoder's EKB-only local tiers and is prefixed
|
||||
into the provider query, so 'Бессонова' would actively degrade the result.
|
||||
"""
|
||||
assert _city_hint("Бессонова") is None
|
||||
assert _city_hint("Бердюгина") is None
|
||||
assert _city_hint("Билейский рыбопитомник") is None
|
||||
assert _city_hint(None) is None
|
||||
assert _city_hint("") is None
|
||||
|
||||
|
||||
async def test_run_backfill_passes_known_city_as_hint():
|
||||
groups = [AddressGroup(address="Победы, 30", deals_count=2, city="Нижний Тагил")]
|
||||
db, coord_updates, _ = _make_db_mock(update_rowcount=2)
|
||||
|
||||
with patch(
|
||||
_GEOCODE_PATH, new_callable=AsyncMock, return_value=_result("nominatim")
|
||||
) as mock_geo:
|
||||
await _run_backfill(db, groups, batch="b", dry_run=False)
|
||||
|
||||
mock_geo.assert_called_once_with("Победы, 30", db, city_hint="Нижний Тагил")
|
||||
assert coord_updates[0]["city"] == "Нижний Тагил"
|
||||
|
||||
|
||||
async def test_run_backfill_junk_city_geocodes_without_hint_but_scopes_update():
|
||||
"""Junk city → geocode WITHOUT a hint (old behaviour), yet the UPDATE is still
|
||||
scoped to that exact group so it can't spill onto another city's rows."""
|
||||
groups = [AddressGroup(address="Бессонова, 11", deals_count=1, city="Бессонова")]
|
||||
db, coord_updates, _ = _make_db_mock(update_rowcount=1)
|
||||
|
||||
with patch(
|
||||
_GEOCODE_PATH, new_callable=AsyncMock, return_value=_result("nominatim")
|
||||
) as mock_geo:
|
||||
await _run_backfill(db, groups, batch="b", dry_run=False)
|
||||
|
||||
mock_geo.assert_called_once_with("Бессонова, 11", db, city_hint=None)
|
||||
assert coord_updates[0]["city"] == "Бессонова"
|
||||
|
||||
|
||||
async def test_run_backfill_same_address_two_cities_independent():
|
||||
"""Key #2603 scenario: identical address text in two cities → two geocode
|
||||
calls with their own hints, and two UPDATEs scoped to their own city."""
|
||||
groups = [
|
||||
AddressGroup(address="Победы, 30", deals_count=4, city="Екатеринбург"),
|
||||
AddressGroup(address="Победы, 30", deals_count=2, city="Нижний Тагил"),
|
||||
]
|
||||
db, coord_updates, _ = _make_db_mock(update_rowcount=1)
|
||||
|
||||
with patch(
|
||||
_GEOCODE_PATH,
|
||||
new_callable=AsyncMock,
|
||||
side_effect=[_result("nominatim"), _result("nominatim", lat=57.910, lon=59.985)],
|
||||
) as mock_geo:
|
||||
stats = await _run_backfill(db, groups, batch="b", dry_run=False)
|
||||
|
||||
assert mock_geo.call_count == 2
|
||||
assert [c.kwargs["city_hint"] for c in mock_geo.call_args_list] == [
|
||||
"Екатеринбург",
|
||||
"Нижний Тагил",
|
||||
]
|
||||
assert stats.geocoded == 2
|
||||
assert [(u["addr"], u["city"]) for u in coord_updates] == [
|
||||
("Победы, 30", "Екатеринбург"),
|
||||
("Победы, 30", "Нижний Тагил"),
|
||||
]
|
||||
|
||||
|
||||
async def test_run_backfill_miss_marks_only_its_own_city():
|
||||
groups = [AddressGroup(address="Победы, 30", deals_count=1, city="Нижний Тагил")]
|
||||
db, coord_updates, tried_updates = _make_db_mock()
|
||||
|
||||
with patch(_GEOCODE_PATH, new_callable=AsyncMock, return_value=None):
|
||||
await _run_backfill(db, groups, batch="b", dry_run=False)
|
||||
|
||||
assert coord_updates == []
|
||||
assert tried_updates == [{"addr": "Победы, 30", "city": "Нижний Тагил"}]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Stats dataclass
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
|
|||
|
|
@ -359,6 +359,51 @@ def test_city_column_null_falls_back_to_text_gate() -> None:
|
|||
mock_geo.assert_not_called()
|
||||
|
||||
|
||||
# ── наблюдаемость гейта: отдельный срез по колонке (#2603) ───────────────────
|
||||
|
||||
|
||||
def test_skipped_non_ekb_by_column_counts_only_column_gate() -> None:
|
||||
"""Колоночный и текстовый гейты различимы: by_column считает ТОЛЬКО первый.
|
||||
|
||||
Общий skipped_non_ekb остаётся суммой обоих (обратная совместимость), но по
|
||||
мере раскатки областных развёрток (#2598) колоночный гейт начнёт забирать
|
||||
строки из no_address — без отдельного среза общий счётчик поменял бы смысл
|
||||
ровно тогда, когда по нему валидируют раскатку.
|
||||
"""
|
||||
rows = [
|
||||
# 1. Гейт по колонке: город в колонке, в тексте адреса города НЕТ.
|
||||
{"id": 300, "address": "ул. Победы, 30", "city": "Нижний Тагил"},
|
||||
# 2. Текстовый гейт: колонка пуста, город назван в тексте.
|
||||
{"id": 301, "address": "г. Нижний Тагил, проспект Ленина, 1", "city": None},
|
||||
]
|
||||
db = _make_db([rows, []])
|
||||
|
||||
with (
|
||||
patch(
|
||||
"app.tasks.backfill_listings_coords_geoportal._geoportal_house_match",
|
||||
return_value=_HIT,
|
||||
),
|
||||
patch("app.tasks.backfill_listings_coords_geoportal._parse_street_house"),
|
||||
):
|
||||
res = backfill_coords_from_geoportal(db, batch_size=500)
|
||||
|
||||
assert res.candidates == 2
|
||||
assert res.skipped_non_ekb == 2 # оба гейта, как и раньше
|
||||
assert res.skipped_non_ekb_by_column == 1 # только колоночный
|
||||
# Вклад текстового гейта вычисляется разностью — старая метрика #2583 жива.
|
||||
assert res.skipped_non_ekb - res.skipped_non_ekb_by_column == 1
|
||||
|
||||
|
||||
def test_to_counters_exposes_skipped_non_ekb_by_column() -> None:
|
||||
"""Новый счётчик уходит в scrape_runs.counters, старый ключ на месте."""
|
||||
counters = BackfillCoordsResult(skipped_non_ekb=7, skipped_non_ekb_by_column=5).to_counters()
|
||||
|
||||
assert counters["skipped_non_ekb"] == 7
|
||||
assert counters["skipped_non_ekb_by_column"] == 5
|
||||
# Дефолт нулевой — прогон без колоночных скипов пишет 0, а не теряет ключ.
|
||||
assert BackfillCoordsResult().to_counters()["skipped_non_ekb_by_column"] == 0
|
||||
|
||||
|
||||
# ── idempotency ───────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -761,11 +761,17 @@ def test_admin_geocode_missing_post_dry_run_endpoint_exists() -> None:
|
|||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_admin_geocode_missing_passes_city_hint() -> None:
|
||||
@pytest.mark.parametrize("target", ["listings", "deals"])
|
||||
async def test_admin_geocode_missing_passes_city_hint(target: str) -> None:
|
||||
"""POST /admin/geocode-missing читает city из SELECT и передаёт как city_hint.
|
||||
|
||||
Раньше endpoint читал только row["address"] и звал geocode(clean, db) без
|
||||
города — голый тагильский адрес без города в тексте уходил в Екатеринбург.
|
||||
|
||||
Оба target'а (#2603): колонка city есть и в listings (миграция 196), и в
|
||||
deals (миграция 177) — SELECT + прокидывание хинта общие для обеих веток,
|
||||
расходится только extra_filter, поэтому deals обязан покрываться тем же
|
||||
контрактом.
|
||||
"""
|
||||
from app.api.v1 import admin as admin_module
|
||||
|
||||
|
|
@ -792,11 +798,17 @@ async def test_admin_geocode_missing_passes_city_hint() -> None:
|
|||
new_callable=AsyncMock,
|
||||
return_value=geo,
|
||||
) as mock_geo:
|
||||
result = await admin_module.geocode_missing(db, limit=100, target="listings")
|
||||
result = await admin_module.geocode_missing(
|
||||
db,
|
||||
limit=100,
|
||||
target=target, # type: ignore[arg-type]
|
||||
)
|
||||
|
||||
mock_geo.assert_called_once_with("ул. Победы, 30", db, city_hint="Нижний Тагил")
|
||||
assert result["geocoded"] == 1
|
||||
assert result["skipped"] == 0
|
||||
# SELECT адресован именно запрошенной таблице (обе несут колонку city).
|
||||
assert f"FROM {target}" in str(db.execute.call_args_list[0][0][0])
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue