il ilçe mahalle ve sokak verilerini çeken python kodu hazırlamıştım kendim için burada sizde faydalanın istedim.
kod sonunda output klasörüne ilk olarak ilkodu.json olarak verileri kaydediyor. sonra da tüm verileri birleştirip sql dosyası haline getiriyor.
2 adet sql hazırlıyor biri il,ilçe,mahalle diğeri ise il,ilçe,mahalle,sokak olarak hazırlanıyor.
kullanım cmd yi açıp pyhon dosyaadi.py --iller all yazmanız gerekiyor tüm kodları almak için yada pyhon dosyaadi.py --iller 35 yazarsanız da sadece izmir i çeker
import argparse
import json
import sys
import time
from pathlib import Path
from playwright.sync_api import sync_playwright
PTT_PAGE = "https://www.ptt.gov.tr/posta-kodu"
ROOT = Path(__file__).resolve().parent
OUTPUT = ROOT / "output"
PROGRESS = OUTPUT / "progress.json"
MAHALLE_SQL = ROOT / "il_ilce_mahalle.sql"
SOKAK_SQL = ROOT / "il_ilce_mahalle_sokak.sql"
MAHALLE_TABLE = "ililcemahalle"
SOKAK_TABLE = "ililcemahallesokak" # Tablo adınız farklıysa değiştirin.
BATCH_SIZE = 500
def read_json(path: Path, default):
try:
return json.loads(path.read_text(encoding="utf-8")) if path.exists() else default
except (OSError, json.JSONDecodeError):
return default
def save_json(path: Path, value):
path.parent.mkdir(parents=True, exist_ok=True)
temporary = path.with_suffix(path.suffix + ".tmp")
temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2), encoding="utf-8")
temporary.replace(path)
def records_from(response):
if isinstance(response, list):
return response
if not isinstance(response, dict):
return []
for key in ("data", "sonuc", "results", "result"):
item = response.get(key)
if isinstance(item, list):
return item
if isinstance(item, dict):
nested = records_from(item)
if nested:
return nested
return []
def value(record, names):
for name in names:
if record.get(name) is not None:
return str(record[name]).strip()
return ""
def ptt_request(page, payload):
"""İstek sayfanın kendi JavaScript ortamından çıkar; hosting IP'si kullanılmaz."""
for attempt in range(1, 5):
print(f" Deneme {attempt}/4: {payload}", flush=True)
result = page.evaluate(
"""async (payload) => {
const response = await fetch('/api/posta-kodu', {
method: 'POST', credentials: 'include',
headers: {
'accept': '*/*', 'accept-language': 'tr-TR,tr;q=0.9,en-US;q=0.8,en;q=0.7',
'cache-control': 'no-cache', 'content-type': 'application/json', 'pragma': 'no-cache'
}, body: JSON.stringify(payload)
});
return {status: response.status, text: await response.text()};
}""",
payload,
)
if result["status"] == 200:
try:
return json.loads(result["text"])
except json.JSONDecodeError as error:
raise RuntimeError(f"PTT geçersiz JSON döndürdü: {result['text'][:300]}") from error
print(f" HTTP {result['status']}: {result['text'][:220]}", flush=True)
if attempt == 4:
raise RuntimeError(f"PTT isteği reddedildi: HTTP {result['status']}")
time.sleep(attempt * 2)
def sql_text(text):
"""MySQL tek tırnaklı metin kaçışlama."""
return "'" + str(text).replace("\\", "\\\\").replace("'", "\\'").replace("\0", "\\0").replace("\n", "\\n").replace("\r", "\\r").replace("\x1a", "\\Z") + "'"
def write_batch(handle, table, columns, rows):
if rows:
handle.write(f"INSERT INTO `{table}` ({columns}) VALUES\n" + ",\n".join(rows) + ";\n")
rows.clear()
def generate_sql():
"""output/1.json ... output/81.json dosyalarından iki SQL dosyası üretir."""
print("\nJSON → SQL aktarımı başladı...", flush=True)
seen_neighborhoods = set()
mahalle_rows, sokak_rows = [], []
street_total = neighborhood_total = missing = invalid = 0
with MAHALLE_SQL.open("w", encoding="utf-8", newline="\n") as mahalle_file, \
SOKAK_SQL.open("w", encoding="utf-8", newline="\n") as sokak_file:
header = "SET NAMES utf8mb4;\nSTART TRANSACTION;\n\n"
mahalle_file.write("-- İl, ilçe, mahalle (tekil)\n" + header)
sokak_file.write("-- İl, ilçe, mahalle, sokak\n" + header)
for city_id in range(1, 82):
path = OUTPUT / f"{city_id}.json"
print(f" SQL [{city_id}/81] {path.name} taranıyor...", flush=True)
if not path.is_file():
missing += 1
print(" Dosya yok, atlandı.", flush=True)
continue
try:
records = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as error:
invalid += 1
print(f" Okunamadı, atlandı: {error}", flush=True)
continue
if not isinstance(records, list):
invalid += 1
print(" JSON dizi formatında değil, atlandı.", flush=True)
continue
city_streets = city_neighborhoods = 0
for record in records:
if not isinstance(record, dict):
continue
city = str(record.get("il_adi", "")).strip()
district = str(record.get("ilce_adi", "")).strip()
neighborhood = str(record.get("mahalle_adi", "")).strip()
street = str(record.get("sokak_adi", "")).strip()
if not city or not district or not neighborhood:
continue
key = (city, district, neighborhood)
if key not in seen_neighborhoods:
seen_neighborhoods.add(key)
mahalle_rows.append("(" + ",".join(map(sql_text, key)) + ")")
if len(mahalle_rows) >= BATCH_SIZE:
write_batch(mahalle_file, MAHALLE_TABLE, "`il`, `ilce`, `mahalle`", mahalle_rows)
neighborhood_total += 1
city_neighborhoods += 1
sokak_rows.append("(" + ",".join(map(sql_text, (city, district, neighborhood, street))) + ")")
if len(sokak_rows) >= BATCH_SIZE:
write_batch(sokak_file, SOKAK_TABLE, "`il`, `ilce`, `mahalle`, `sokak`", sokak_rows)
street_total += 1
city_streets += 1
print(f" Tamam: {city_streets} sokak kaydı, {city_neighborhoods} yeni tekil mahalle.", flush=True)
write_batch(mahalle_file, MAHALLE_TABLE, "`il`, `ilce`, `mahalle`", mahalle_rows)
write_batch(sokak_file, SOKAK_TABLE, "`il`, `ilce`, `mahalle`, `sokak`", sokak_rows)
mahalle_file.write("\nCOMMIT;\n")
sokak_file.write("\nCOMMIT;\n")
print(f"\nSQL hazır: {neighborhood_total} tekil mahalle, {street_total} sokak kaydı.", flush=True)
print(f" {MAHALLE_SQL}", flush=True)
print(f" {SOKAK_SQL}", flush=True)
print(f"Atlanan dosya: {missing} | Hatalı JSON: {invalid}", flush=True)
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--iller", required=True, help="35 | 6,34,35 | all")
parser.add_argument("--reset", action="store_true", help="Seçili illeri sıfırdan çeker")
args = parser.parse_args()
OUTPUT.mkdir(exist_ok=True)
with sync_playwright() as playwright:
browser = playwright.chromium.launch(headless=False)
try:
page = browser.new_page(locale="tr-TR")
print("PTT sayfası açılıyor...", flush=True)
page.goto(PTT_PAGE, wait_until="domcontentloaded", timeout=90_000)
page.locator("#il-select").wait_for(timeout=60_000)
cities = dict(page.locator("#il-select option[value]").evaluate_all(
"options => options.filter(o => o.value).map(o => [o.value, o.textContent.trim()])"
))
requested = list(cities) if args.iller.lower() == "all" else list(dict.fromkeys(
item.strip() for item in args.iller.split(",") if item.strip()
))
invalid = [city for city in requested if city not in cities]
if invalid:
raise SystemExit("Geçersiz il kodu: " + ", ".join(invalid))
state = read_json(PROGRESS, {"done": {}, "completed": []})
if args.reset:
for city_id in requested:
state["done"].pop(city_id, None)
state["completed"] = [code for code in state["completed"] if code != city_id]
(OUTPUT / f"{city_id}.json").unlink(missing_ok=True)
save_json(PROGRESS, state)
for city_id in requested:
city_name = cities[city_id]
if city_id in state["completed"]:
print(f"{city_name}: daha önce tamamlandı.", flush=True)
continue
print(f"\n{city_id} - {city_name}: ilçeler alınıyor...", flush=True)
districts = records_from(ptt_request(page, {"action": "ilceler", "il_kodu": city_id}))
if not districts:
raise RuntimeError(f"{city_name}: ilçe listesi boş geldi.")
city_file = OUTPUT / f"{city_id}.json"
rows = read_json(city_file, [])
done = set(state.setdefault("done", {}).setdefault(city_id, []))
known = {json.dumps(row, ensure_ascii=False, sort_keys=True) for row in rows}
for district in districts:
district_id = value(district, ["ilce_kodu", "ilce_id", "id", "value", "kod"])
district_name = value(district, ["ilce_adi", "ilceAdi", "ilce", "text", "name", "ad"])
if not district_id or district_id in {"-1", "0"} or district_name.lower() in {"seçiniz", "ilçe seçiniz"} or district_id in done:
continue
print(f"{city_name} / {district_name}: mahalle-sokak-posta kodu alınıyor...", flush=True)
response_rows = records_from(ptt_request(page, {"action": "postakodu", "il_kodu": city_id, "ilce_kodu": district_id}))
for record in response_rows:
row = {"il_id": city_id, "il_adi": city_name,
"ilce_adi": value(record, ["ilce_adi", "ilceAdi", "ilce"]) or district_name,
"posta_kodu": value(record, ["posta_kodu", "postaKodu", "postakodu", "pk"]),
"mahalle_adi": value(record, ["mahalle_adi", "mahalleAdi", "mahalle"]),
"sokak_adi": value(record, ["sokak_adi", "sokakAdi", "sokak", "cadde_sokak"])}
key = json.dumps(row, ensure_ascii=False, sort_keys=True)
if key not in known:
rows.append(row)
known.add(key)
save_json(city_file, rows)
done.add(district_id)
state["done"][city_id] = sorted(done)
save_json(PROGRESS, state)
print(f" Tamam: {len(response_rows)} kayıt; il toplamı: {len(rows)}", flush=True)
time.sleep(0.35)
if city_id not in state["completed"]:
state["completed"].append(city_id)
save_json(PROGRESS, state)
print(f"{city_name} tamamlandı → {city_file}", flush=True)
finally:
browser.close()
# PTT çekimi başarıyla bittiğinde iki SQL dosyası otomatik oluşturulur.
generate_sql()
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
print("\nDurduruldu. Aynı komutu tekrar çalıştırdığınızda kaldığı yerden devam eder.")
sys.exit(130)