- Şehirler
- Ormanlar
- Nehirler
- Denizler
- Göller
- Dağlar
- Çöller
- Bölgeler
- Ovalar
Bu oldukça kapsamlı bir veri toplama projesi. Ben ona yardımcı olabilirim. Öncelikle bir Python scripti oluşturabilirim ki bu script:
- Ülkelerin listesini alsın
- Her ülke için coğrafi verileri toplasın
- JSON formatında kaydetsin
Bu Python scripti size 200 ülke için kapsamlı coğrafi veri toplama konusunda yardımcı olacak. İşte scriptin özellikleri ve kullanımı:
🔑 Ana Özellikler:
- Çoklu Veri Kaynağı:
- REST Countries API (ülke listesi ve temel bilgiler)
- GeoNames API (detaylı coğrafi veriler)
- OpenStreetMap Nominatim API (ek coğrafi veriler)
- Paralel İşleme: ThreadPoolExecutor ile aynı anda 5 ülkenin verisi toplanıyor
- Hata Yönetimi: API hatalarında script durmaz, diğer ülkelerle devam eder
- Esnek Kayıt:
- Tek bir büyük JSON dosyası
- Her ülke için ayrı JSON dosyaları
📋 Kullanım Adımları:
- Gerekli kütüphaneleri yükleyin:
pip install requests
- GeoNames hesabı açın(ücretsiz):
- http://www.geonames.org/login adresinden kayıt olun
- Script'teki username = "demo" satırını kendi kullanıcı adınızla değiştirin
- Script'i çalıştırın:
python country_geo_collector.py
🛠️ Özelleştirme:
- Daha fazla veri kaynağı eklemek için:
- Wikidata API
- Google Places API (ücretli)
- MapBox API
- Veri kalitesini artırmak için:
- Her kategori için minimum veri sayısı belirleyin
- Koordinat doğrulama ekleyin
- Türkçe Wikipedia'dan ek bilgiler çekin
- Eksik verileri tamamlamak için manuel veri dosyası:
# manual_data.json
{
"TUR": {
"deserts": [
{
"name": "Karapınar Çölü",
"lat": 37.7,
"lon": 33.5,
"details": "Türkiye'nin tek çöl bölgesi"
}
]
}
}
💡 İpuçları:
- Rate Limiting: API'lerin istek limitine dikkat edin
- Önbellek: Toplanan verileri önbellekte saklayın
- İncremental Update: Sadece eksik verileri güncelleyin
import json
import requests
import time
from typing import Dict, List, Optional
import logging
from dataclasses import dataclass, asdict
from concurrent.futures import ThreadPoolExecutor, as_completed
# Logging ayarları
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
@dataclass
class GeographicFeature:
name: str
lat: float
lon: float
details: str
@dataclass
class CountryData:
name: str
code: str
lat: float
lon: float
features: Dict[str, List[Dict]]
class CountryGeographyCollector:
def __init__(self):
self.session = requests.Session()
self.session.headers.update({
'User-Agent': 'CountryGeographyCollector/1.0'
})
def get_countries_list(self) -> List[Dict]:
"""Tüm ülkelerin listesini al"""
try:
# REST Countries API'den ülke listesi
response = self.session.get('https://restcountries.com/v3.1/all')
response.raise_for_status()
countries = response.json()
country_list = []
for country in countries:
country_info = {
'name_tr': country.get('translations', {}).get('tur', {}).get('common', country['name']['common']),
'name_en': country['name']['common'],
'code': country.get('cca3', ''),
'lat': country.get('latlng', [0, 0])[0],
'lon': country.get('latlng', [0, 0])[1],
'capital': country.get('capital', [''])[0] if country.get('capital') else '',
'region': country.get('region', ''),
'subregion': country.get('subregion', '')
}
country_list.append(country_info)
return sorted(country_list, key=lambda x: x['name_tr'])
except Exception as e:
logger.error(f"Ülke listesi alınırken hata: {e}")
return []
def search_geonames(self, country_code: str, feature_type: str, limit: int = 5) -> List[Dict]:
"""GeoNames API'den coğrafi özellikleri ara"""
# GeoNames username'i (ücretsiz hesap açarak alabilirsiniz)
username = "demo" # Kendi username'inizi kullanın
feature_codes = {
'cities': 'PPL',
'mountains': 'MT',
'rivers': 'STM',
'lakes': 'LK',
'forests': 'FRST',
'plains': 'PLN',
'regions': 'RGN'
}
if feature_type not in feature_codes:
return []
try:
params = {
'country': country_code[:2], # 2 harfli ülke kodu
'featureCode': feature_codes[feature_type],
'maxRows': limit,
'username': username,
'lang': 'tr'
}
response = self.session.get(
'http://api.geonames.org/searchJSON',
params=params,
timeout=10
)
response.raise_for_status()
data = response.json()
results = []
for item in data.get('geonames', []):
results.append({
'name': item.get('name', ''),
'lat': float(item.get('lat', 0)),
'lon': float(item.get('lng', 0)),
'details': item.get('toponymName', '')
})
return results
except Exception as e:
logger.warning(f"GeoNames araması başarısız ({feature_type}): {e}")
return []
def search_nominatim(self, country: str, feature_type: str, limit: int = 5) -> List[Dict]:
"""OpenStreetMap Nominatim API'den veri ara"""
feature_queries = {
'cities': 'city',
'mountains': 'mountain',
'rivers': 'river',
'lakes': 'water lake',
'forests': 'forest',
'plains': 'plain',
'seas': 'sea'
}
if feature_type not in feature_queries:
return []
try:
params = {
'country': country,
'q': feature_queries[feature_type],
'format': 'json',
'limit': limit,
'accept-language': 'tr'
}
response = self.session.get(
'https://nominatim.openstreetmap.org/search',
params=params,
timeout=10
)
response.raise_for_status()
data = response.json()
results = []
for item in data:
results.append({
'name': item.get('display_name', '').split(',')[0],
'lat': float(item.get('lat', 0)),
'lon': float(item.get('lon', 0)),
'details': item.get('type', '')
})
time.sleep(1) # Nominatim rate limit
return results
except Exception as e:
logger.warning(f"Nominatim araması başarısız ({feature_type}): {e}")
return []
def get_country_features(self, country_info: Dict) -> Dict:
"""Bir ülke için tüm coğrafi özellikleri topla"""
logger.info(f"İşleniyor: {country_info['name_tr']}")
features = {
'cities': [],
'forests': [],
'rivers': [],
'seas': [],
'lakes': [],
'mountains': [],
'deserts': [],
'regions': [],
'plains': []
}
# Başkenti ekle
if country_info['capital']:
features['cities'].append({
'name': country_info['capital'],
'lat': country_info['lat'],
'lon': country_info['lon'],
'details': f"{country_info['name_tr']} başkenti"
})
# Her özellik türü için veri topla
for feature_type in features.keys():
# GeoNames'den ara
geonames_results = self.search_geonames(country_info['code'], feature_type)
features[feature_type].extend(geonames_results[:3])
# Nominatim'den ara
if len(features[feature_type]) < 3:
nominatim_results = self.search_nominatim(country_info['name_en'], feature_type)
features[feature_type].extend(nominatim_results[:2])
# Tekrarları kaldır
for feature_type in features:
seen = set()
unique_features = []
for feature in features[feature_type]:
if feature['name'] not in seen:
seen.add(feature['name'])
unique_features.append(feature)
features[feature_type] = unique_features[:5] # En fazla 5 öğe
return {
'name': country_info['name_tr'],
'code': country_info['code'],
'lat': country_info['lat'],
'lon': country_info['lon'],
'features': features
}
def collect_all_countries(self, limit: Optional[int] = None) -> List[Dict]:
"""Tüm ülkeler için veri topla"""
countries = self.get_countries_list()
if limit:
countries = countries[:limit]
all_country_data = []
# Paralel işleme için ThreadPoolExecutor kullan
with ThreadPoolExecutor(max_workers=5) as executor:
future_to_country = {
executor.submit(self.get_country_features, country): country
for country in countries
}
for future in as_completed(future_to_country):
try:
country_data = future.result()
all_country_data.append(country_data)
logger.info(f"Tamamlandı: {country_data['name']}")
except Exception as e:
country = future_to_country[future]
logger.error(f"Hata ({country['name_tr']}): {e}")
return sorted(all_country_data, key=lambda x: x['name'])
def save_to_json(self, data: List[Dict], filename: str = 'countries_geographic_data.json'):
"""Verileri JSON dosyasına kaydet"""
try:
with open(filename, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
logger.info(f"Veriler {filename} dosyasına kaydedildi")
except Exception as e:
logger.error(f"Dosya kaydetme hatası: {e}")
def save_individual_countries(self, data: List[Dict], directory: str = 'countries'):
"""Her ülkeyi ayrı dosyaya kaydet"""
import os
if not os.path.exists(directory):
os.makedirs(directory)
for country in data:
filename = f"{directory}/{country['code']}.json"
try:
with open(filename, 'w', encoding='utf-8') as f:
json.dump(country, f, ensure_ascii=False, indent=2)
except Exception as e:
logger.error(f"Dosya kaydetme hatası ({country['name']}): {e}")
def main():
collector = CountryGeographyCollector()
# Önce birkaç ülke ile test et
logger.info("Test modu: İlk 5 ülke için veri toplanıyor...")
test_data = collector.collect_all_countries(limit=5)
collector.save_to_json(test_data, 'test_countries.json')
# Tüm ülkeler için veri toplamak isterseniz:
# all_data = collector.collect_all_countries()
# collector.save_to_json(all_data, 'all_countries_geographic_data.json')
# collector.save_individual_countries(all_data)
logger.info("İşlem tamamlandı!")
if __name__ == "__main__":
main()