- Ş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()