412 lines
17 KiB
Python
412 lines
17 KiB
Python
# -*- coding: utf-8 -*-
|
||
import os
|
||
import re
|
||
import sqlite3
|
||
from typing import Optional, Dict, Any, Tuple, List
|
||
|
||
# 1) Чтение OSM .osm.pbf и извлечение addr:* через pyosmium
|
||
import osmium # pip install pyosmium
|
||
|
||
# 2) Парсинг/нормализация адресов через libpostal
|
||
# Требуется установленная системная библиотека libpostal и Python-биндинги "postal"
|
||
from postal.parser import parse_address
|
||
from postal.expand import expand_address
|
||
import json
|
||
# 3) Нечеткий поиск (для автокоррекции)
|
||
from rapidfuzz import process, fuzz, utils # pip install rapidfuzz
|
||
import math
|
||
# (Опционально) геометрический центроид для way, если доступен shapely
|
||
try:
|
||
from shapely.geometry import LineString, Polygon
|
||
HAVE_SHAPELY = True
|
||
except Exception:
|
||
HAVE_SHAPELY = False
|
||
|
||
DB_SCHEMA = """
|
||
PRAGMA journal_mode=WAL;
|
||
CREATE TABLE IF NOT EXISTS street_names (
|
||
street_norm TEXT PRIMARY KEY,
|
||
street_original TEXT
|
||
);
|
||
CREATE TABLE IF NOT EXISTS address_index (
|
||
street_norm TEXT,
|
||
housenumber_norm TEXT,
|
||
street_original TEXT,
|
||
housenumber_original TEXT,
|
||
city TEXT,
|
||
postcode TEXT,
|
||
lat REAL,
|
||
lon REAL
|
||
);
|
||
CREATE INDEX IF NOT EXISTS idx_address_street ON address_index(street_norm);
|
||
CREATE INDEX IF NOT EXISTS idx_address_street_house ON address_index(street_norm,housenumber_norm);
|
||
"""
|
||
|
||
# ---------------------------
|
||
# Минимальная предобработка
|
||
# ---------------------------
|
||
|
||
_DASHES = {"–": "-", "—": "-", "−": "-"}
|
||
|
||
def _preclean_text(s: str) -> str:
|
||
"""Минимальная очистка для устойчивости парсера: регистр, дефисы, пробелы, буква-цифра разделение."""
|
||
s = (s or "").strip().lower().replace("ё", "е")
|
||
for k, v in _DASHES.items():
|
||
s = s.replace(k, v)
|
||
s = re.sub(r"\s+", " ", s)
|
||
# Вставить пробел на границе буква<->цифра, чтобы "дом43" -> "дом 43"
|
||
s = re.sub(r"(?<=[A-Za-zА-Яа-я])(?=\d)|(?<=\d)(?=[A-Za-zА-Яа-я])", " ", s)
|
||
return s.strip()
|
||
|
||
def _parts_from_libpostal(addr_text: str) -> Dict[str, str]:
|
||
"""Собирает метка->значение из parse_address без сложных правил."""
|
||
parts: Dict[str, str] = {}
|
||
for comp, label in parse_address(addr_text):
|
||
comp = " ".join(comp.split())
|
||
if not comp:
|
||
continue
|
||
parts[label] = (parts[label] + " " + comp).strip() if label in parts else comp
|
||
return parts
|
||
|
||
def _prefer_cyrillic_min(expansions: List[str]) -> Optional[str]:
|
||
"""Выбрать «лучший» вариант из expand_address: предпочесть кириллицу и минимальную длину."""
|
||
if not expansions:
|
||
return None
|
||
cyr = [e for e in expansions if re.search(r"[А-Яа-я]", e)]
|
||
pool = cyr if cyr else expansions
|
||
pool = [" ".join(e.split()) for e in pool]
|
||
pool.sort(key=len)
|
||
return pool[0] if pool else None
|
||
|
||
def canonicalize_road(road: str) -> str:
|
||
"""Канонизация названия улицы через expand_address с минимальной эвристикой выбора."""
|
||
if not road:
|
||
return ""
|
||
exp = expand_address(road)
|
||
return _prefer_cyrillic_min(exp) or road
|
||
|
||
def _clean_token(s: str) -> str:
|
||
return " ".join((s or "").strip().lower().replace("ё", "е").split())
|
||
|
||
def _unit_suffix_and_value(text: str) -> str:
|
||
"""
|
||
Преобразует текст unit/доп.компонента в компактный русскоязычный суффикс:
|
||
- 'корпус 1' / 'к 1' -> 'К1'
|
||
- 'строение 2' / 'стр 2' -> 'С2'
|
||
- 'лит а' -> 'ЛА'
|
||
- 'влад 5' -> 'ВЛ5'
|
||
Если тип не распознан, по умолчанию считаем корпус 'К' (без тяжёлых регексов).
|
||
"""
|
||
t = _clean_token(text)
|
||
# выделяем буквы/цифры (минимум регексов: только удалить пробелы)
|
||
val = t.replace(" ", "")
|
||
# эвристики по подстрокам
|
||
if "строен" in t or t.startswith("стр"):
|
||
# строение
|
||
val = val.replace("строение", "").replace("стр", "")
|
||
return f"С{val}".upper() if val else "С"
|
||
if "корп" in t or t == "к" or t.startswith("к"):
|
||
# корпус
|
||
val = val.replace("корпус", "").replace("корп", "").replace("к", "")
|
||
return f"К{val}".upper() if val else "К"
|
||
if t.startswith("лит"):
|
||
# литера
|
||
val = val.replace("литера", "").replace("лит", "")
|
||
return f"Л{val}".upper() if val else "Л"
|
||
if t.startswith("влад"):
|
||
# владение
|
||
val = val.replace("владение", "").replace("влад", "")
|
||
return f"ВЛ{val}".upper() if val else "ВЛ"
|
||
# вход/подъезд/лестница можно при желании кодировать как 'П', 'ЛС', но обычно это не часть addr:housenumber в OSM
|
||
# по умолчанию — трактуем как корпус
|
||
# оставим только цифро-буквенную часть после удаления пробелов
|
||
return f"К{val}".upper() if val else ""
|
||
|
||
def build_house_number_from_parts(parts: Dict[str, str], original_line: str) -> str:
|
||
"""
|
||
Формирует финальный housenumber из libpostal:
|
||
- base = house_number
|
||
- добавляет суффиксы из unit/entrance/staircase/level, конвертируя их в 'К..'/'С..'/'Л..'/'ВЛ..'
|
||
"""
|
||
base = (parts.get("house_number") or "").strip()
|
||
# защитимся от «дом43» и т.п. минимальной разделиловкой (сделано ранее в _preclean_text)
|
||
suffixes = []
|
||
|
||
# unit часто содержит корпус/строение
|
||
if parts.get("unit"):
|
||
suffixes.append(_unit_suffix_and_value(parts["unit"]))
|
||
|
||
# некоторые адреса могут класть строение/литеру в entrance/staircase
|
||
for key in ("entrance", "staircase"):
|
||
if parts.get(key):
|
||
sfx = _unit_suffix_and_value(parts[key])
|
||
if sfx:
|
||
suffixes.append(sfx)
|
||
|
||
# level — это этаж, обычно не часть housenumber в OSM; по умолчанию не добавляем
|
||
|
||
# если базовый номер пуст — попробуем через expand_address всей строки (без сложных правил)
|
||
if not base:
|
||
for e in expand_address(original_line):
|
||
p2 = _parts_from_libpostal(e)
|
||
b2 = (p2.get("house_number") or "").strip()
|
||
if b2:
|
||
base = b2
|
||
if p2.get("unit") and not suffixes:
|
||
suffixes.append(_unit_suffix_and_value(p2["unit"]))
|
||
break
|
||
|
||
# склеиваем: пробелы убираем, верхний регистр для стабильности
|
||
hn = "".join(base.split()).upper() + "".join(suffixes)
|
||
return hn
|
||
|
||
def normalize_street_for_index(street_original: str) -> str:
|
||
"""Нормализация ключа улицы для индекса (регистронезависимая, с лексической очисткой)."""
|
||
# Используем обработчик RapidFuzz по умолчанию + канонизацию libpostal
|
||
street_clean = _preclean_text(street_original)
|
||
street_canon = canonicalize_road(street_clean)
|
||
return utils.default_process(street_canon) or street_canon
|
||
|
||
def normalize_house_for_index(house_original: str) -> str:
|
||
"""Простая нормализация номера для индекса: убрать пробелы, привести к верхнему регистру."""
|
||
return "".join((house_original or "").split()).upper()
|
||
|
||
|
||
|
||
# ---------------------------
|
||
# Поиск/валидация адреса
|
||
# ---------------------------
|
||
|
||
def parse_with_libpostal(addr: str) -> Dict[str, str]:
|
||
"""
|
||
Минимально-шумоустойчивая раскладка через libpostal.
|
||
"""
|
||
txt = _preclean_text(addr)
|
||
parts = _parts_from_libpostal(txt)
|
||
road = parts.get("road") or parts.get("pedestrian") or parts.get("footway") or parts.get("residential") or ""
|
||
house = build_house_number_from_parts(parts, txt)
|
||
city = parts.get("city") or parts.get("city_district") or parts.get("suburb") or ""
|
||
postcode = parts.get("postcode", "")
|
||
# Канонизируем улицу для сравнения
|
||
road_canon = canonicalize_road(road)
|
||
return {
|
||
"road": road_canon or road or "",
|
||
"house_number": house or "",
|
||
"city": city or "",
|
||
"postcode": postcode or ""
|
||
}
|
||
|
||
def best_street_match(conn: sqlite3.Connection, street_query: str, cutoff: int = 86) -> Optional[Tuple[str, str]]:
|
||
"""
|
||
Ищет с учётом опечаток: exact по нормализованному ключу, иначе WRatio к ближайшему.
|
||
"""
|
||
s_key = normalize_street_for_index(street_query)
|
||
cur = conn.execute("SELECT street_original FROM street_names WHERE street_norm = ?", (s_key,))
|
||
row = cur.fetchone()
|
||
if row:
|
||
return (s_key, row[0])
|
||
|
||
cur = conn.execute("SELECT street_norm, street_original FROM street_names")
|
||
rows = cur.fetchall()
|
||
if not rows:
|
||
return None
|
||
choices = [r[0] for r in rows]
|
||
match = process.extractOne(s_key, choices, scorer=fuzz.WRatio, processor=None, score_cutoff=cutoff)
|
||
if match:
|
||
matched_norm = match[0]
|
||
for s_n, s_o in rows:
|
||
if s_n == matched_norm:
|
||
return (s_n, s_o)
|
||
return None
|
||
|
||
def _base_digits(s: str) -> str:
|
||
m = re.search(r"\d+", s or "")
|
||
return m.group(0) if m else ""
|
||
|
||
def best_house_match(conn, street_norm: str, house_query: str, cutoff: int = 82) -> Optional[Tuple[str, str]]:
|
||
# Простой канон запроса: убрать пробелы, верхний регистр
|
||
q_norm = ("".join((house_query or "").split())).upper()
|
||
q_base = _base_digits(q_norm)
|
||
|
||
# Вытаскиваем все номера по улице
|
||
cur = conn.execute(
|
||
"SELECT housenumber_norm, housenumber_original FROM address_index WHERE street_norm = ?",
|
||
(street_norm,)
|
||
)
|
||
rows = cur.fetchall()
|
||
if not rows:
|
||
return None
|
||
|
||
# 1) Точное совпадение целиком
|
||
for hn, ho in rows:
|
||
if hn == q_norm:
|
||
return (hn, ho)
|
||
|
||
# 2) Жёсткий фильтр по базовому числу (обязательное совпадение)
|
||
if q_base:
|
||
filtered: List[Tuple[str, str]] = [(hn, ho) for hn, ho in rows if _base_digits(hn) == q_base]
|
||
else:
|
||
filtered = rows
|
||
|
||
if not filtered:
|
||
return None
|
||
|
||
# 3) Нечёткий матч только среди отфильтрованных кандидатов
|
||
choices = [hn for hn, _ in filtered]
|
||
match = process.extractOne(q_norm, choices, scorer=fuzz.WRatio, processor=None, score_cutoff=cutoff)
|
||
if match:
|
||
best_hn = match[0]
|
||
for hn, ho in filtered:
|
||
if hn == best_hn:
|
||
return (hn, ho)
|
||
|
||
# 4) Fallback: если остался один кандидат с таким же базовым числом — принять его
|
||
if len(filtered) == 1:
|
||
return filtered[0]
|
||
|
||
return None
|
||
|
||
def _interp_point_along(coords: List[List[float]], t: float) -> Dict[str,float]:
|
||
# coords: [[lon,lat], ...], t∈[0,1] по длине полилинии
|
||
def dist(a,b):
|
||
dx=a[0]-b[0]; dy=a[1]-b[1]; return math.hypot(dx, dy)
|
||
seg_len=[]; total=0.0
|
||
for i in range(len(coords)-1):
|
||
d=dist(coords[i], coords[i+1]); seg_len.append(d); total += d
|
||
if total==0.0:
|
||
return {"lat": coords[0][1], "lon": coords[0][0]}
|
||
target = t*total; acc=0.0
|
||
for i,d in enumerate(seg_len):
|
||
if acc + d >= target:
|
||
ratio = 0.0 if d==0 else (target-acc)/d
|
||
lon = coords[i][0] + (coords[i+1][0]-coords[i][0])*ratio
|
||
lat = coords[i][1] + (coords[i+1][1]-coords[i][1])*ratio
|
||
return {"lat": float(lat), "lon": float(lon)}
|
||
acc += d
|
||
return {"lat": coords[-1][1], "lon": coords[-1][0]}
|
||
|
||
def _fits_type(n:int, itype:str)->bool:
|
||
itype=(itype or "").lower()
|
||
if itype=="odd": return n%2==1
|
||
if itype=="even": return n%2==0
|
||
return True # 'all' или неизвестный тип
|
||
|
||
def find_interpolation_coord(conn: sqlite3.Connection, street_norm: str, house_query: str) -> Optional[Dict[str,float]]:
|
||
m = re.search(r"\d+", house_query or "")
|
||
if not m:
|
||
return None
|
||
n = int(m.group(0))
|
||
cur = conn.execute(
|
||
"SELECT itype, start_num, end_num, coords_json FROM interpolation_ways WHERE street_norm=?",
|
||
(street_norm,)
|
||
)
|
||
best=None
|
||
for itype, s_i, e_i, coords_json in cur.fetchall():
|
||
if n < min(s_i,e_i) or n > max(s_i,e_i):
|
||
continue
|
||
if not _fits_type(n, itype):
|
||
continue
|
||
coords = json.loads(coords_json)
|
||
# позиция вдоль диапазона с плавающей точкой
|
||
span = (e_i - s_i) if (e_i - s_i)!=0 else 1
|
||
t = (n - s_i) / span
|
||
t = max(0.0, min(1.0, t))
|
||
return _interp_point_along(coords, t)
|
||
return best
|
||
|
||
def get_street_centroid_any(conn: sqlite3.Connection, street_norm: str) -> Optional[Dict[str, float]]:
|
||
# 1) по адресным точкам
|
||
cur = conn.execute(
|
||
"SELECT AVG(lat), AVG(lon) FROM address_index WHERE street_norm=? AND lat IS NOT NULL AND lon IS NOT NULL",
|
||
(street_norm,)
|
||
)
|
||
lat, lon = cur.fetchone() or (None,None)
|
||
if lat is not None and lon is not None:
|
||
return {"lat": float(lat), "lon": float(lon)}
|
||
# 2) по геометрии улиц (highway)
|
||
cur = conn.execute("SELECT lat, lon FROM street_center WHERE street_norm=?",(street_norm,))
|
||
row = cur.fetchone()
|
||
if row:
|
||
return {"lat": float(row[0]), "lon": float(row[1])}
|
||
return None
|
||
|
||
def validate_address(sqlite_path: str, address_str: str) -> Dict[str, Any]:
|
||
comps = parse_with_libpostal(address_str)
|
||
street_in = comps.get("road", "")
|
||
house_in = comps.get("house_number", "")
|
||
|
||
conn = sqlite3.connect(sqlite_path)
|
||
try:
|
||
# улица обязательна
|
||
if not street_in:
|
||
return {
|
||
"valid": False,
|
||
"corrected": None,
|
||
"components": comps,
|
||
"coordinates": None,
|
||
"reason": "Не удалось распознать улицу"
|
||
}
|
||
|
||
st = best_street_match(conn, street_in)
|
||
if not st:
|
||
return {
|
||
"valid": False,
|
||
"corrected": None,
|
||
"components": comps,
|
||
"coordinates": None,
|
||
"reason": "Улица не найдена в локальной базе"
|
||
}
|
||
street_norm, street_orig = st
|
||
|
||
# 1) дом не введён: вернуть координату улицы (centroid)
|
||
if not house_in:
|
||
cent = get_street_centroid_any(conn, street_norm)
|
||
return {
|
||
"valid": True if cent else False,
|
||
"corrected": street_orig,
|
||
"components": comps,
|
||
"coordinates": cent,
|
||
"reason": None if cent else "Для улицы нет координатной опоры"
|
||
}
|
||
|
||
# 2) дом введён: пробуем точное сопоставление
|
||
hn = best_house_match(conn, street_norm, house_in)
|
||
if hn:
|
||
house_norm, house_orig = hn
|
||
cur = conn.execute(
|
||
"SELECT city, postcode, lat, lon FROM address_index WHERE street_norm=? AND housenumber_norm=? LIMIT 1",
|
||
(street_norm, house_norm)
|
||
)
|
||
row = cur.fetchone()
|
||
city, postcode, lat, lon = (row or (None,None,None,None))
|
||
coords = {"lat": float(lat), "lon": float(lon)} if lat is not None and lon is not None else None
|
||
# если почему-то lat/lon пустые (редкий случай), дополнительно попробуем интерполяцию/центроид
|
||
if coords is None:
|
||
coords = find_interpolation_coord(conn, street_norm, house_norm) or get_street_centroid_any(conn, street_norm)
|
||
corrected = ", ".join(x for x in [f"{street_orig} {house_orig}", city, postcode] if x and str(x).strip())
|
||
return {
|
||
"valid": True,
|
||
"corrected": corrected,
|
||
"components": comps,
|
||
"coordinates": coords,
|
||
"reason": None
|
||
}
|
||
|
||
# 3) дом введён, но не найден: invalid, однако координаты улицы — обязательно
|
||
fallback = find_interpolation_coord(conn, street_norm, house_in) or get_street_centroid_any(conn, street_norm)
|
||
return {
|
||
"valid": False,
|
||
"corrected": None,
|
||
"components": comps,
|
||
"coordinates": fallback,
|
||
"reason": "Номер дома отсутствует на найденной улице"
|
||
}
|
||
finally:
|
||
conn.close()
|
||
|
||
if __name__ == "__main__":
|
||
|
||
res = validate_address("addresses.sqlite", "ул. Большая Покровская, 15")
|
||
print(res)
|
||
pass
|