65 lines
2.4 KiB
Python
Executable File
65 lines
2.4 KiB
Python
Executable File
import psycopg2
|
|
import json
|
|
import urllib.request
|
|
import urllib.parse
|
|
import os
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv()
|
|
|
|
HUNGARY_BBOX = "45.7,16.1,48.6,22.9"
|
|
|
|
def fetch_osm_data(query_part):
|
|
print(f"🔍 Adatgyűjtés: {query_part[:30]}...")
|
|
query = f'[out:json][timeout:90];(node{query_part}({HUNGARY_BBOX});way{query_part}({HUNGARY_BBOX}););out center;'
|
|
url = "http://overpass-api.de/api/interpreter?data=" + urllib.parse.quote(query)
|
|
try:
|
|
with urllib.request.urlopen(url) as response:
|
|
return json.loads(response.read())['elements']
|
|
except: return []
|
|
|
|
def get_service_type(tags, name):
|
|
name = name.lower()
|
|
shop = tags.get('shop', '')
|
|
amenity = tags.get('amenity', '')
|
|
|
|
if shop == 'tyres' or 'gumi' in name: return 'tire_shop'
|
|
if amenity == 'car_wash' or 'mosó' in name: return 'car_wash'
|
|
if 'villamoss' in name or 'autóvill' in name: return 'electrician'
|
|
if 'fényez' in name or 'lakatos' in name: return 'body_shop'
|
|
if 'dízel' in name or 'diesel' in name: return 'diesel_specialist'
|
|
if tags.get('service:vehicle:electric') == 'yes': return 'ev_specialist'
|
|
return 'mechanic'
|
|
|
|
def sync():
|
|
conn = psycopg2.connect(dbname=os.getenv("POSTGRES_DB"), user=os.getenv("POSTGRES_USER"),
|
|
password=os.getenv("POSTGRES_PASSWORD"), host="localhost")
|
|
cur = conn.cursor()
|
|
|
|
# 1. Szervizek és Gumisok (Összetett keresés)
|
|
search_query = '["shop"~"car_repair|tyres"]'
|
|
results = fetch_osm_data(search_query)
|
|
|
|
# 2. Autómosók külön
|
|
washes = fetch_osm_data('["amenity"="car_wash"]')
|
|
|
|
for node in results + washes:
|
|
tags = node.get('tags', {})
|
|
lat = node.get('lat', node.get('center', {}).get('lat'))
|
|
lon = node.get('lon', node.get('center', {}).get('lon'))
|
|
if not lat or not lon: continue
|
|
|
|
name = tags.get('name', tags.get('operator', 'Ismeretlen szerviz'))
|
|
s_type = get_service_type(tags, name)
|
|
city = tags.get('addr:city', 'Ismeretlen')
|
|
|
|
cur.execute("""
|
|
INSERT INTO data.service_providers (name, service_type, location_city, latitude, longitude, search_tags, is_active)
|
|
VALUES (%s, %s, %s, %s, %s, %s, true) ON CONFLICT DO NOTHING
|
|
""", (name, s_type, city, lat, lon, json.dumps(tags)))
|
|
|
|
conn.commit()
|
|
print("✅ Országos szerviz adatbázis frissítve!")
|
|
|
|
if __name__ == "__main__":
|
|
sync() |