86 lines
3.5 KiB
Python
86 lines
3.5 KiB
Python
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import text
|
|
from typing import Optional, List
|
|
import uuid
|
|
|
|
class GeoService:
|
|
@staticmethod
|
|
async def get_street_suggestions(db: AsyncSession, zip_code: str, q: str) -> List[str]:
|
|
"""Azonnali utca-kiegészítés (Autocomplete) támogatása."""
|
|
query = text("""
|
|
SELECT s.name
|
|
FROM data.geo_streets s
|
|
JOIN data.geo_postal_codes p ON s.postal_code_id = p.id
|
|
WHERE p.zip_code = :zip AND s.name ILIKE :q
|
|
ORDER BY s.name ASC LIMIT 10
|
|
""")
|
|
res = await db.execute(query, {"zip": zip_code, "q": f"{q}%"})
|
|
return [row[0] for row in res.fetchall()]
|
|
|
|
@staticmethod
|
|
async def get_or_create_full_address(
|
|
db: AsyncSession,
|
|
zip_code: str, city: str, street_name: str,
|
|
street_type: str, house_number: str,
|
|
parcel_id: Optional[str] = None
|
|
) -> uuid.UUID:
|
|
"""Hibrid címrögzítés: ellenőrzi a szótárakat és létrehozza a központi címet."""
|
|
# 1. Zip/City szótár frissítése (Auto-learning)
|
|
zip_id_res = await db.execute(text("""
|
|
INSERT INTO data.geo_postal_codes (zip_code, city) VALUES (:z, :c)
|
|
ON CONFLICT (country_code, zip_code, city) DO UPDATE SET city = EXCLUDED.city
|
|
RETURNING id
|
|
"""), {"z": zip_code, "c": city})
|
|
zip_id = zip_id_res.scalar()
|
|
|
|
# 2. Utca szótár frissítése (Auto-learning)
|
|
await db.execute(text("""
|
|
INSERT INTO data.geo_streets (postal_code_id, name) VALUES (:zid, :n)
|
|
ON CONFLICT (postal_code_id, name) DO NOTHING
|
|
"""), {"zid": zip_id, "n": street_name})
|
|
|
|
# 3. Közterület típus (út, utca...) szótár
|
|
await db.execute(text("""
|
|
INSERT INTO data.geo_street_types (name) VALUES (:n) ON CONFLICT DO NOTHING
|
|
"""), {"n": street_type.lower()})
|
|
|
|
# 4. Központi Address rekord rögzítése
|
|
full_text = f"{zip_code} {city}, {street_name} {street_type} {house_number}."
|
|
if stairwell: full_text += f" {stairwell}. lph,"
|
|
if floor: full_text += f" {floor}. em,"
|
|
if door: full_text += f" {door}. ajtó"
|
|
|
|
query = text("""
|
|
INSERT INTO data.addresses (
|
|
postal_code_id, street_name, street_type, house_number,
|
|
stairwell, floor, door, parcel_id, full_address_text
|
|
)
|
|
VALUES (
|
|
(SELECT id FROM data.geo_postal_codes WHERE zip_code = :z AND city = :c LIMIT 1),
|
|
:sn, :st, :hn, :sw, :fl, :dr, :pid, :txt
|
|
)
|
|
ON CONFLICT DO NOTHING
|
|
RETURNING id
|
|
""")
|
|
|
|
params = {
|
|
"z": zip_code, "c": city, "sn": street_name, "st": street_type,
|
|
"hn": house_number, "sw": stairwell, "fl": floor, "dr": door,
|
|
"pid": parcel_id, "txt": full_text
|
|
}
|
|
|
|
res = await db.execute(query, params)
|
|
addr_id = res.scalar()
|
|
|
|
if not addr_id:
|
|
# Ha már létezett ilyen részletes cím, lekérjük
|
|
addr_id = (await db.execute(text("""
|
|
SELECT id FROM data.addresses
|
|
WHERE street_name = :sn AND house_number = :hn
|
|
AND (stairwell IS NOT DISTINCT FROM :sw)
|
|
AND (floor IS NOT DISTINCT FROM :fl)
|
|
AND (door IS NOT DISTINCT FROM :dr)
|
|
LIMIT 1
|
|
"""), params)).scalar()
|
|
|
|
return addr_id |