Yandex: Mesin Pencari Raksasa Rusia & API Lengkapnya

# Simpan API key di environment variable, JANGAN hardcode di source code!
import os

# Set environment variable di terminal:
# export YANDEX_API_KEY="your-api-key-here"  (Linux/Mac)
# set YANDEX_API_KEY=your-api-key-here       (Windows)

API_KEY = os.environ.get("YANDEX_API_KEY")

if not API_KEY:
    print("API key belum diset! Bikin environment variable YANDEX_API_KEY dulu.")
    exit(1)

API key berhasil dimuat: AIza...XXXX

pip install requests

pip install yandex-cloud

import requests
import json
import os

# Konfirmasi library terinstall dengan benar
print(f"requests version: {requests.__version__}")

requests version: 2.31.0

import requests
import os

API_KEY = os.environ.get("YANDEX_API_KEY")
BASE_URL = "https://translate.yandex.net/api/v1.5/tr.json/translate"

def translate_text(text, target_lang="en"):
    """
    Terjemahkan teks ke bahasa target menggunakan Yandex Translate API.
    
    Args:
        text (str): Teks yang ingin diterjemahkan
        target_lang (str): Kode bahasa tujuan (mis. 'en', 'id', 'ru')
    
    Returns:
        str: Teks hasil terjemahan
    """
    params = {
        "key": API_KEY,
        "text": text,
        "lang": target_lang,
        "format": "plain"
    }
    
    response = requests.get(BASE_URL, params=params)
    
    if response.status_code == 200:
        data = response.json()
        translated = data["text"][0]
        return translated
    else:
        return f"Error {response.status_code}: {response.text}"

# Tes translasi dari Rusia ke Inggris
hasil = translate_text("Привет, мир!", "en")
print(f"Hasil terjemahan: {hasil}")

Hasil terjemahan: Hello, world!

import requests
import os
import json

API_KEY = os.environ.get("YANDEX_API_KEY")
FOLDER_ID = os.environ.get("YANDEX_FOLDER_ID")  # Diperlukan untuk Yandex Cloud

def yandex_search(query, limit=10):
    """
    Lakukan pencarian web menggunakan Yandex Search API.
    
    Args:
        query (str): Query pencarian
        limit (int): Jumlah hasil maksimal
    
    Returns:
        list: Daftar hasil pencarian
    """
    url = "https://searchapi.cloud.yandex.net/v2/web/searchAsync"
    
    headers = {
        "Authorization": f"Api-Key {API_KEY}",
        "Content-Type": "application/json"
    }
    
    body = {
        "query": query,
        "folderId": FOLDER_ID,
        "maxPassages": limit
    }
    
    # Step 1: Kirim request search (async)
    response = requests.post(url, headers=headers, json=body)
    
    if response.status_code != 200:
        print(f"Error mengirim search: {response.status_code}")
        print(response.text)
        return []
    
    operation_id = response.json().get("id")
    print(f"Search operation ID: {operation_id}")
    
    # Step 2: Polling untuk hasil
    result_url = f"https://operation.api.cloud.yandex.net/operations/{operation_id}"
    
    import time
    for attempt in range(30):
        time.sleep(2)
        result_response = requests.get(result_url, headers=headers)
        
        if result_response.status_code == 200:
            result_data = result_response.json()
            if result_data.get("done"):
                return result_data.get("response", {}).get("rawData", [])
        
        print(f"Menunggu hasil... attempt {attempt + 1}")
    
    print("Timeout: hasil belum tersedia setelah 60 detik")
    return []

# Contoh penggunaan
hasil_search = yandex_search("belajar python indonesia", limit=5)
for i, item in enumerate(hasil_search, 1):
    print(f"{i}. {item.get('title', 'No title')}")
    print(f"   URL: {item.get('url', 'No URL')}")
    print(f"   Snippet: {item.get('snippet', '')[:100]}...")
    print()

Search operation ID: operation-abc123def456
Menunggu hasil... attempt 1
Menunggu hasil... attempt 2
1. Belajar Python dari Nol: Panduan Lengkap
   URL: https://www.contoh-artikel.com/belajar-python
   Snippet: Panduan lengkap belajar Python dari nol untuk bahasa Indonesia. Cocok untuk yang baru mengenal programming...

import requests
import os

MAPS_API_KEY = os.environ.get("YANDEX_MAPS_API_KEY")

def geocode_address(address):
    """
    Konversi alamat menjadi koordinat latitude/longitude menggunakan Yandex Geocoder API.
    
    Args:
        address (str): Alamat yang ingin di-geocode
    
    Returns:
        dict: Koordinat dan metadata lokasi
    """
    url = "https://geocode-maps.yandex.ru/1.x/"
    
    params = {
        "apikey": MAPS_API_KEY,
        "geocode": address,
        "format": "json",
        "results": 1,
        "lang": "id_ID"
    }
    
    response = requests.get(url, params=params)
    
    if response.status_code == 200:
        data = response.json()
        try:
            member = data["response"]["GeoObjectCollection"]["featureMember"][0]
            geo_obj = member["GeoObject"]
            point = geo_obj["Point"]["pos"].split()
            
            return {
                "address": geo_obj["metaDataProperty"]["GeocoderMetaData"]["text"],
                "coordinates": {
                    "longitude": float(point[0]),
                    "latitude": float(point[1])
                },
                "kind": geo_obj["metaDataProperty"]["GeocoderMetaData"]["kind"]
            }
        except (IndexError, KeyError) as e:
            return {"error": f"Lokasi tidak ditemukan: {e}"}
    else:
        return {"error": f"HTTP {response.status_code}: {response.text}"}

# Tes geocoding
lokasi = geocode_address("Jakarta, Indonesia")
print(json.dumps(lokasi, indent=2, ensure_ascii=False))

{
  "address": "Jakarta, Indonesia",
  "coordinates": {
    "longitude": 106.84513,
    "latitude": -6.21462
  },
  "kind": "locality"
}

import requests
import os
import base64

SPEECHKIT_API_KEY = os.environ.get("YANDEX_SPEECHKIT_KEY")
FOLDER_ID = os.environ.get("YANDEX_FOLDER_ID")

def text_to_speech(text, lang="ru-RU", voice="alena"):
    """
    Konversi teks menjadi audio menggunakan Yandex SpeechKit TTS.
    
    Args:
        text (str): Teks yang ingin dikonversi
        lang (str): Kode bahasa (mis. 'ru-RU', 'en-US', 'tr-TR')
        voice (str): Nama voice (mis. 'alena', 'filipp', 'jane')
    
    Returns:
        bytes: Data audio dalam format OggOpus
    """
    url = "https://tts.api.cloud.yandex.net/speech/v1/tts:synthesize"
    
    headers = {
        "Authorization": f"Api-Key {SPEECHKIT_API_KEY}"
    }
    
    data = {
        "text": text,
        "lang": lang,
        "voice": voice,
        "folderId": FOLDER_ID,
        "format": "oggopus"
    }
    
    response = requests.post(url, headers=headers, data=data)
    
    if response.status_code == 200:
        return response.content
    else:
        print(f"Error TTS: {response.status_code} - {response.text}")
        return None

# Konversi teks ke audio
audio_data = text_to_speech("Halo, ini adalah tes text to speech.", lang="id-ID")

if audio_data:
    with open("output.ogg", "wb") as f:
        f.write(audio_data)
    print("Audio berhasil disimpan ke output.ogg")
    print(f"Ukuran file: {len(audio_data)} bytes")

Audio berhasil disimpan ke output.ogg
Ukuran file: 15234 bytes

import requests
import os
import time

class YandexAPIError(Exception):
    """Custom exception untuk error Yandex API."""
    def __init__(self, status_code, message):
        self.status_code = status_code
        self.message = message
        super().__init__(f"Yandex API Error {status_code}: {message}")

def safe_api_call(url, params=None, headers=None, method="GET", max_retries=3):
    """
    Lakukan API call dengan retry logic dan error handling lengkap.
    
    Args:
        url (str): Endpoint URL
        params (dict): Query parameters
        headers (dict): Request headers
        method (str): HTTP method (GET atau POST)
        max_retries (int): Maksimum retry attempts
    
    Returns:
        dict: Response JSON
    
    Raises:
        YandexAPIError: Jika semua retry gagal
    """
    for attempt in range(max_retries):
        try:
            if method == "GET":
                response = requests.get(url, params=params, headers=headers, timeout=30)
            else:
                response = requests.post(url, json=params, headers=headers, timeout=30)
            
            # Cek status code
            if response.status_code == 200:
                return response.json()
            
            # Handle error berdasarkan status code
            error_messages = {
                400: "Request tidak valid - cek parameter kamu",
                401: "API key tidak valid atau kedaluwarsa",
                403: "Akses ditolak - cek permission API key kamu",
                404: "Endpoint tidak ditemukan - cek URL",
                429: "Rate limit exceeded - terlalu banyak request",
                500: "Server error Yandex - coba lagi nanti",
                503: "Service unavailable - sedang maintenance"
            }
            
            error_msg = error_messages.get(
                response.status_code, 
                f"Unknown error: {response.text}"
            )
            
            # Retry untuk error server dan rate limit
            if response.status_code in [429, 500, 503]:
                wait_time = 2 ** attempt  # Exponential backoff
                print(f"Attempt {attempt + 1} gagal: {error_msg}")
                print(f"Menunggu {wait_time} detik sebelum retry...")
                time.sleep(wait_time)
                continue
            
            # Untuk error klien (4xx), jangan retry
            raise YandexAPIError(response.status_code, error_msg)
            
        except requests.exceptions. Timeout:
            print(f"Timeout pada attempt {attempt + 1}")
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)
                continue
            raise YandexAPIError(0, "Request timeout setelah semua retry")
            
        except requests.exceptions. ConnectionError:
            print(f"Connection error pada attempt {attempt + 1}")
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)
                continue
            raise YandexAPIError(0, "Tidak bisa connect ke server Yandex")
    
    raise YandexAPIError(0, f"Gagal setelah {max_retries} retry")

# Contoh penggunaan dengan error handling
try:
    result = safe_api_call(
        "https://translate.yandex.net/api/v1.5/tr.json/translate",
        params={
            "key": os.environ.get("YANDEX_API_KEY"),
            "text": "Halo dunia",
            "lang": "id-en"
        }
    )
    print(f"Terjemahan: {result['text'][0]}")
except YandexAPIError as e:
    print(f"API gagal: {e}")

Terjemahan: Hello world

API gagal: Yandex API Error 401: API key tidak valid atau kedaluwarsa

import requests
import os
import json
import time

class YandexToolkit:
    """
    Toolkit untuk mengakses berbagai layanan Yandex API dalam satu class.
    """
    
    def __init__(self):
        self.api_key = os.environ.get("YANDEX_API_KEY")
        self.maps_key = os.environ.get("YANDEX_MAPS_API_KEY")
        self.folder_id = os.environ.get("YANDEX_FOLDER_ID")
        
        if not self.api_key:
            raise ValueError("YANDEX_API_KEY belum diset di environment variable")
    
    def translate(self, text, target_lang="en", source_lang=None):
        """Terjemahkan teks menggunakan Yandex Translate."""
        url = "https://translate.yandex.net/api/v1.5/tr.json/translate"
        
        params = {
            "key": self.api_key,
            "text": text,
            "lang": target_lang,
            "format": "plain"
        }
        
        if source_lang:
            params["lang"] = f"{source_lang}-{target_lang}"
        
        response = requests.get(url, params=params, timeout=30)
        
        if response.status_code == 200:
            data = response.json()
            return {
                "translated_text": data["text"][0],
                "detected_lang": data.get("lang", "unknown"),
                "status": "success"
            }
        return {
            "status": "error",
            "code": response.status_code,
            "message": response.text
        }
    
    def detect_language(self, text):
        """Deteksi bahasa dari teks."""
        url = "https://translate.yandex.net/api/v1.5/tr.json/detect"
        
        params = {
            "key": self.api_key,
            "text": text
        }
        
        response = requests.get(url, params=params, timeout=30)
        
        if response.status_code == 200:
            data = response.json()
            return {
                "language": data.get("lang", "unknown"),
                "status": "success"
            }
        return {
            "status": "error",
            "code": response.status_code,
            "message": response.text
        }
    
    def geocode(self, address, lang="en_US"):
        """Geocode alamat menjadi koordinat."""
        if not self.maps_key:
            return {"error": "YANDEX_MAPS_API_KEY belum diset"}
        
        url = "https://geocode-maps.yandex.ru/1.x/"
        
        params = {
            "apikey": self.maps_key,
            "geocode": address,
            "format": "json",
            "results": 1,
            "lang": lang
        }
        
        response = requests.get(url, params=params, timeout=30)
        
        if response.status_code == 200:
            data = response.json()
            try:
                member = data["response"]["GeoObjectCollection"]["featureMember"][0]
                geo_obj = member["GeoObject"]
                point = geo_obj["Point"]["pos"].split()
                
                return {
                    "address": geo_obj["metaDataProperty"]["GeocoderMetaData"]["text"],
                    "latitude": float(point[1]),
                    "longitude": float(point[0]),
                    "status": "success"
                }
            except (IndexError, KeyError):
                return {"status": "error", "message": "Lokasi tidak ditemukan"}
        return {
            "status": "error",
            "code": response.status_code,
            "message": response.text
        }
    
    def get_supported_languages(self):
        """Dapatkan daftar bahasa yang didukung oleh Translate API."""
        url = "https://translate.yandex.net/api/v1.5/tr.json/getLangs"
        
        params = {
            "key": self.api_key,
            "ui": "en"
        }
        
        response = requests.get(url, params=params, timeout=30)
        
        if response.status_code == 200:
            return {
                "languages": response.json().get("langs", {}),
                "status": "success"
            }
        return {
            "status": "error",
            "code": response.status_code,
            "message": response.text
        }

# === CONTOH PENGGUNAAN LENGKAP ===

if __name__ == "__main__":
    try:
        toolkit = YandexToolkit()
        
        # 1. Deteksi bahasa
        print("=== Deteksi Bahasa ===")
        sample_text = "Halo, apa kabar hari ini?"
        result = toolkit.detect_language(sample_text)
        print(f"Teks: {sample_text}")
        print(f"Bahasa terdeteksi: {result.get('language', 'N/A')}")
        print()
        
        # 2. Terjemahkan
        print("=== Terjemahan ===")
        translation = toolkit.translate(sample_text, target_lang="en")
        print(f"Original: {sample_text}")
        print(f"Terjemahan: {translation.get('translated_text', 'N/A')}")
        print()
        
        # 3. Daftar bahasa yang didukung
        print("=== Bahasa yang Didukung ===")
        langs = toolkit.get_supported_languages()
        if langs["status"] == "success":
            supported = langs["languages"]
            print(f"Total bahasa: {len(supported)}")
            print(f"Contoh: {dict(list(supported.items())[:5])}")
        print()
        
        # 4. Geocode
        print("=== Geocoding ===")
        location = toolkit.geocode("Moscow, Red Square")
        if location["status"] == "success":
            print(f"Alamat: {location['address']}")
            print(f"Latitude: {location['latitude']}")
            print(f"Longitude: {location['longitude']}")
        
    except ValueError as e:
        print(f"Setup error: {e}")
    except Exception as e:
        print(f"Unexpected error: {e}")

=== Deteksi Bahasa ===
Teks: Halo, apa kabar hari ini?
Bahasa terdeteksi: id

=== Terjemahan ===
Original: Halo, apa kabar hari ini?
Terjemahan: Hello, how are you today?

=== Bahasa yang Didukung ===
Total bahasa: 95
Contoh: {'af': 'Afrikaans', 'sq': 'Albanian', 'am': 'Amharic', 'ar': 'Arabic', 'hy': 'Armenian'}

=== Geocoding ===
Alamat: Russia, Moscow, Red Square
Latitude: 55.75391
Longitude: 37.62055

import hashlib
import json
import os
import time
from functools import wraps

class SimpleCache:
    """
    Cache sederhana berbasis file untuk menyimpan hasil API call.
    """
    
    def __init__(self, cache_dir=".cache", ttl=3600):
        self.cache_dir = cache_dir
        self.ttl = ttl  # Time to live dalam detik (default 1 jam)
        os.makedirs(cache_dir, exist_ok=True)
    
    def _get_cache_key(self, *args, **kwargs):
        """Buat cache key unik dari parameter."""
        key_str = json.dumps({"args": args, "kwargs": kwargs}, sort_keys=True)
        return hashlib.md5(key_str.encode()).hexdigest()
    
    def get(self, *args, **kwargs):
        """Ambil data dari cache jika masih valid."""
        key = self._get_cache_key(*args, **kwargs)
        cache_file = os.path.join(self.cache_dir, f"{key}.json")
        
        if os.path.exists(cache_file):
            with open(cache_file, "r") as f:
                cached = json.load(f)
            
            # Cek apakah cache masih dalam TTL
            if time.time() - cached["timestamp"] < self.ttl:
                print(f"[CACHE HIT] Mengambil dari cache: {key[:8]}...")
                return cached["data"]
            else:
                print(f"[CACHE EXPIRED] Cache kedaluwarsa: {key[:8]}...")
                os.remove(cache_file)
        
        return None
    
    def set(self, data, *args, **kwargs):
        """Simpan data ke cache."""
        key = self._get_cache_key(*args, **kwargs)
        cache_file = os.path.join(self.cache_dir, f"{key}.json")
        
        with open(cache_file, "w") as f:
            json.dump({
                "timestamp": time.time(),
                "data": data
            }, f)
        
        print(f"[CACHE SET] Menyimpan ke cache: {key[:8]}...")

# Decorator untuk caching otomatis
cache = SimpleCache(ttl=3600)  # Cache 1 jam

def cached_api_call(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        # Coba ambil dari cache dulu
        cached_result = cache.get(func.__name__, *args, **kwargs)
        if cached_result is not None:
            return cached_result
        
        # Kalau nggak ada di cache, panggil fungsi asli
        result = func(*args, **kwargs)
        
        # Simpan ke cache
        if result and isinstance(result, dict) and result.get("status") == "success":
            cache.set(result, func.__name__, *args, **kwargs)
        
        return result
    return wrapper

# Pakai decorator di method toolkit
class CachedYandexToolkit(YandexToolkit):
    
    @cached_api_call
    def translate(self, text, target_lang="en", source_lang=None):
        return super().translate(text, target_lang, source_lang)
    
    @cached_api_call
    def geocode(self, address, lang="en_US"):
        return super().geocode(address, lang)

# Tes caching
toolkit = CachedYandexToolkit()

print("=== Panggilan Pertama (API Call) ===")
result1 = toolkit.translate("Привет мир", target_lang="en")
print(f"Hasil: {result1.get('translated_text')}")
print()

print("=== Panggilan Kedua (Cache Hit) ===")
result2 = toolkit.translate("Привет мир", target_lang="en")
print(f"Hasil: {result2.get('translated_text')}")

=== Panggilan Pertama (API Call) ===
Hasil: Hello world

=== Panggilan Kedua (Cache Hit) ===
[CACHE HIT] Mengambil dari cache: a3b2c1d4...
Hasil: Hello world

import time
import logging
from datetime import datetime

# Setup logging
logging.basicConfig(
    level=logging. INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    handlers=[
        logging. FileHandler("yandex_api.log"),
        logging. StreamHandler()
    ]
)

logger = logging.getLogger("YandexAPI")

class APIMonitor:
    """
    Monitor sederhana untuk track performa API call.
    """
    
    def __init__(self):
        self.stats = {
            "total_calls": 0,
            "success": 0,
            "errors": 0,
            "total_time_ms": 0,
            "errors_detail": []
        }
    
    def track(self, func):
        """Decorator untuk track performa setiap API call."""
        def wrapper(*args, **kwargs):
            self.stats["total_calls"] += 1
            start_time = time.time()
            
            try:
                result = func(*args, **kwargs)
                elapsed_ms = (time.time() - start_time) * 1000
                self.stats["success"] += 1
                self.stats["total_time_ms"] += elapsed_ms
                
                logger.info(
                    f"{func.__name__} - Success - {elapsed_ms:.0f}ms"
                )
                return result
                
            except Exception as e:
                elapsed_ms = (time.time() - start_time) * 1000
                self.stats["errors"] += 1
                self.stats["errors_detail"].append({
                    "function": func.__name__,
                    "error": str(e),
                    "timestamp": datetime.now().isoformat()
                })
                
                logger.error(
                    f"{func.__name__} - Error: {e} - {elapsed_ms:.0f}ms"
                )
                raise
        
        return wrapper
    
    def get_stats(self):
        """Dapatkan statistik performa."""
        avg_time = (
            self.stats["total_time_ms"] / self.stats["total_calls"]
            if self.stats["total_calls"] > 0 else 0
        )
        success_rate = (
            (self.stats["success"] / self.stats["total_calls"]) * 100
            if self.stats["total_calls"] > 0 else 0
        )
        
        return {
            "total_calls": self.stats["total_calls"],
            "success_rate": f"{success_rate:.1f}%",
            "avg_response_time": f"{avg_time:.0f}ms",
            "total_errors": self.stats["errors"],
            "recent_errors": self.stats["errors_detail"][-5:]
        }

monitor = APIMonitor()

# Contoh penggunaan
@monitor.track
def tracked_translate(toolkit, text, lang="en"):
    return toolkit.translate(text, target_lang=lang)

# Simulasi beberapa panggilan
toolkit = YandexToolkit()

for text in ["Привет", "Спасибо", "Доброе утро"]:
    try:
        tracked_translate(toolkit, text, "en")
    except Exception:
        pass

# Cek statistik
print(json.dumps(monitor.get_stats(), indent=2))

{
  "total_calls": 3,
  "success_rate": "100.0%",
  "avg_response_time": "245ms",
  "total_errors": 0,
  "recent_errors": []
}

# Solusi: cek API key di environment variable
import os
key = os.environ.get("YANDEX_API_KEY")
print(f"Key length: {len(key) if key else 0}")
print(f"Key prefix: {key[:8]}..." if key else "Key tidak ditemukan!")

import time

def rate_limited_call(func, delay=1.0):
    """Tambahkan delay antara API call untuk hindari rate limit."""
    def wrapper(*args, **kwargs):
        result = func(*args, **kwargs)
        time.sleep(delay)
        return result
    return wrapper

import requests

# HANYA untuk debugging, JANGAN pakai di production!
response = requests.get(url, params=params, verify=False)

pip install --upgrade certifi

response = requests.get(url, params=params, timeout=60)  # Naikkan dari 30 ke 60

import queue
import threading

class BatchTranslator:
    def __init__(self, toolkit, max_workers=3):
        self.toolkit = toolkit
        self.task_queue = queue. Queue()
        self.results = {}
        self.max_workers = max_workers
    
    def add_task(self, product_id, text, target_langs):
        """Tambahkan task translasi ke queue."""
        self.task_queue.put({
            "product_id": product_id,
            "text": text,
            "target_langs": target_langs
        })
    
    def worker(self):
        """Worker thread untuk memproses translasi."""
        while True:
            try:
                task = self.task_queue.get(timeout=5)
            except queue. Empty:
                break
            
            product_id = task["product_id"]
            text = task["text"]
            
            translations = {}
            for lang in task["target_langs"]:
                result = self.toolkit.translate(text, target_lang=lang)
                if result["status"] == "success":
                    translations[lang] = result["translated_text"]
            
            self.results[product_id] = translations
            self.task_queue.task_done()
    
    def run(self):
        """Jalankan semua worker."""
        threads = []
        for _ in range(self.max_workers):
            t = threading. Thread(target=self.worker)
            t.start()
            threads.append(t)
        
        for t in threads:
            t.join()
        
        return self.results

# Simulasi penggunaan
toolkit = YandexToolkit()
batch = BatchTranslator(toolkit, max_workers=3)

# Tambahkan beberapa produk
batch.add_task("P001", "Sepatu lari ringan untuk pria", ["ru", "kk", "tr"])
batch.add_task("P002", "Tas ransel waterproof 30L", ["ru", "kk", "tr"])
batch.add_task("P003", "Jaket hoodie cotton premium", ["ru", "kk", "tr"])

results = batch.run()

for pid, translations in results.items():
    print(f"\nProduk {pid}:")
    for lang, text in translations.items():
        print(f"  {lang}: {text}")

# Contoh batch translate
def batch_translate(toolkit, texts, target_lang="en"):
    """Terjemahkan multiple teks dalam satu request."""
    url = "https://translate.yandex.net/api/v1.5/tr.json/translate"
    
    params = {
        "key": toolkit.api_key,
        "lang": target_lang,
        "format": "plain"
    }
    
    # Tambahkan setiap teks sebagai parameter terpisah
    for text in texts:
        params.setdefault("text", [])
        if isinstance(params["text"], list):
            params["text"].append(text)
        else:
            params["text"] = [params["text"], text]
    
    response = requests.get(url, params=params, timeout=30)
    
    if response.status_code == 200:
        data = response.json()
        return {
            "translations": data["text"],
            "status": "success"
        }
    return {"status": "error", "code": response.status_code}

# Bandingkan: 10 teks = 10 request vs 1 request
texts = [f"Teks contoh nomor {i}" for i in range(10)]

# Cara lama: 10 API call
# for text in texts:
#     toolkit.translate(text, "en")

# Cara batch: 1 API call
result = batch_translate(toolkit, texts, "en")
print(f"Jumlah hasil: {len(result.get('translations', []))}")

# Geocoder untuk region berbeda
REGIONS = {
    "russia": "https://geocode-maps.yandex.ru/1.x/",
    "turkey": "https://geocode-maps.yandex.com.tr/1.x/",
    "international": "https://geocode-maps.yandex.com/1.x/"
}

def geocode_by_region(address, region="russia", api_key=None):
    """Geocode dengan endpoint yang sesuai region."""
    url = REGIONS.get(region, REGIONS["russia"])
    
    params = {
        "apikey": api_key,
        "geocode": address,
        "format": "json"
    }
    
    response = requests.get(url, params=params, timeout=30)
    return response.json() if response.status_code == 200 else None

class HybridSearchClient:
    """
    Client yang bisa switch antara Yandex dan Google berdasarkan region.
    """
    
    def __init__(self, yandex_key, google_key):
        self.yandex_key = yandex_key
        self.google_key = google_key
        self.default_provider = "yandex"
    
    def search(self, query, region="global"):
        """
        Pilih provider berdasarkan region.
        """
        if region in ["ru", "kz", "uz", "tm", "ge", "tr"]:
            return self._yandex_search(query)
        else:
            return self._google_search(query)
    
    def _yandex_search(self, query):
        """Implementasi Yandex Search."""
        # ... kode Yandex search
        pass
    
    def _google_search(self, query):
        """Implementasi Google Search."""
        # ... kode Google search
        pass

def call_with_backoff(func, max_retries=5):
    """Panggil API dengan exponential backoff."""
    for attempt in range(max_retries):
        try:
            return func()
        except Exception as e:
            if "429" in str(e) and attempt < max_retries - 1:
                wait = 2 ** attempt
                print(f"Rate limited. Menunggu {wait}s...")
                time.sleep(wait)
                continue
            raise

# Benar
with open("cache.json", "w") as f:
    json.dump(data, f)

# Salah (bisa bikin file handle leak)
f = open("cache.json", "w")
json.dump(data, f)
# lupa f.close()

import unittest
from unittest.mock import patch, MagicMock
import json

class TestYandexToolkit(unittest. TestCase):
    
    def setUp(self):
        """Setup sebelum setiap test."""
        os.environ["YANDEX_API_KEY"] = "test-api-key-12345"
        self.toolkit = YandexToolkit()
    
    @patch('requests.get')
    def test_translate_success(self, mock_get):
        """Test translasi berhasil."""
        mock_response = MagicMock()
        mock_response.status_code = 200
        mock_response.json.return_value = {
            "code": 200,
            "lang": "id-en",
            "text": ["Hello world"]
        }
        mock_get.return_value = mock_response
        
        result = self.toolkit.translate("Halo dunia", "en")
        
        self.assertEqual(result["status"], "success")
        self.assertEqual(result["translated_text"], "Hello world")
        self.assertEqual(result["detected_lang"], "id-en")
    
    @patch('requests.get')
    def test_translate_api_error(self, mock_get):
        """Test translasi gagal (API error)."""
        mock_response = MagicMock()
        mock_response.status_code = 401
        mock_response.text = "Invalid API key"
        mock_get.return_value = mock_response
        
        result = self.toolkit.translate("Halo dunia", "en")
        
        self.assertEqual(result["status"], "error")
        self.assertEqual(result["code"], 401)
    
    @patch('requests.get')
    def test_detect_language(self, mock_get):
        """Test deteksi bahasa."""
        mock_response = MagicMock()
        mock_response.status_code = 200
        mock_response.json.return_value = {
            "code": 200,
            "lang": "id"
        }
        mock_get.return_value = mock_response
        
        result = self.toolkit.detect_language("Halo, apa kabar?")
        
        self.assertEqual(result["status"], "success")
        self.assertEqual(result["language"], "id")
    
    @patch('requests.get')
    def test_geocode_not_found(self, mock_get):
        """Test geocode lokasi tidak ditemukan."""
        mock_response = MagicMock()
        mock_response.status_code = 200
        mock_response.json.return_value = {
            "response": {
                "GeoObjectCollection": {
                    "featureMember": []
                }
            }
        }
        mock_get.return_value = mock_response
        
        result = self.toolkit.geocode("lokasi yang tidak ada")
        
        self.assertEqual(result["status"], "error")
        self.assertIn("tidak ditemukan", result["message"])

# Jalankan test
if __name__ == "__main__":
    unittest.main(verbosity=2)

test_detect_language ... ok
test_geocode_not_found ... ok
test_translate_api_error ... ok
test_translate_success ... ok

----------------------------------------------------------------------
Ran 4 tests in 0.02s

OK

yandex-project/
├── .env                    # Environment variables (JANGAN commit!)
├── .env.example            # Template env variable
├── .gitignore
├── requirements.txt        # Dependencies
├── README.md
├── src/
│   ├── __init__.py
│   ├── main.py             # Entry point
│   ├── config.py           # Konfigurasi
│   ├── api/
│   │   ├── __init__.py
│   │   ├── yandex_client.py     # Yandex API client
│   │   ├── translate.py         # Translate module
│   │   ├── search.py            # Search module
│   │   └── maps.py              # Maps module
│   ├── utils/
│   │   ├── __init__.py
│   │   ├── cache.py             # Caching utility
│   │   ├── logger.py            # Logging utility
│   │   └── monitor.py          # Performance monitor
│   └── tests/
│       ├── __init__.py
│       ├── test_translate.py
│       ├── test_search.py
│       └── test_maps.py
└── docs/
    └── api_notes.md

# src/config.py
import os
from dataclasses import dataclass

@dataclass
class Config:
    """Konfigurasi aplikasi dari environment variables."""
    yandex_api_key: str = os.environ.get("YANDEX_API_KEY", "")
    yandex_maps_key: str = os.environ.get("YANDEX_MAPS_API_KEY", "")
    yandex_folder_id: str = os.environ.get("YANDEX_FOLDER_ID", "")
    cache_ttl: int = int(os.environ.get("CACHE_TTL", "3600"))
    max_retries: int = int(os.environ.get("MAX_RETRIES", "3"))
    request_timeout: int = int(os.environ.get("REQUEST_TIMEOUT", "30"))
    
    def validate(self):
        """Validasi konfigurasi."""
        if not self.yandex_api_key:
            raise ValueError("YANDEX_API_KEY wajib diset")
        return True

# .env.example
"""
YANDEX_API_KEY=your-api-key-here
YANDEX_MAPS_API_KEY=your-maps-key-here
YANDEX_FOLDER_ID=your-folder-id-here
CACHE_TTL=3600
MAX_RETRIES=3
REQUEST_TIMEOUT=30
"""

# Tes endpoint dengan curl sebelum bikin kode
curl -X GET "https://translate.yandex.net/api/v1.5/tr.json/translate" \
  -d "key=YOUR_API_KEY" \
  -d "text=Halo dunia" \
  -d "lang=id-en"

{
    "code": 200,
    "lang": "id-en",
    "text": ["Hello world"]
}

# Reverse geocoding: dari koordinat ke alamat
def reverse_geocode(latitude, longitude, api_key):
    """Konversi koordinat ke alamat."""
    url = "https://geocode-maps.yandex.ru/1.x/"
    
    coords = f"{longitude},{latitude}"
    
    params = {
        "apikey": api_key,
        "geocode": coords,
        "format": "json",
        "lang": "en_US",
        "results": 1,
        "kind": "house"
    }
    
    response = requests.get(url, params=params, timeout=30)
    
    if response.status_code == 200:
        data = response.json()
        try:
            member = data["response"]["GeoObjectCollection"]["featureMember"][0]
            return member["GeoObject"]["metaDataProperty"]["GeocoderMetaData"]["text"]
        except (IndexError, KeyError):
            return "Alamat tidak ditemukan untuk koordinat ini"
    
    return f"Error: {response.status_code}"

# Tes reverse geocoding
alamat = reverse_geocode(55.75391, 37.62055, os.environ.get("YANDEX_MAPS_API_KEY"))
print(f"Alamat: {alamat}")

Alamat: Russia, Moscow, Red Square

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

class YandexSession:
    """
    HTTP session dengan connection pooling dan retry otomatis.
    """
    
    def __init__(self, api_key):
        self.api_key = api_key
        self.session = requests. Session()
        
        # Retry strategy
        retry_strategy = Retry(
            total=3,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503],
            allowed_methods=["GET", "POST"]
        )
        
        adapter = HTTPAdapter(
            max_retries=retry_strategy,
            pool_connections=10,
            pool_maxsize=100
        )
        
        self.session.mount("https://", adapter)
        self.session.mount("http://", adapter)
    
    def get(self, url, params=None):
        """GET request dengan session pooling."""
        return self.session.get(url, params=params, timeout=30)
    
    def post(self, url, json=None, data=None, headers=None):
        """POST request dengan session pooling."""
        return self.session.post(url, json=json, data=data, headers=headers, timeout=30)
    
    def close(self):
        """Tutup session."""
        self.session.close()

# Penggunaan
session = YandexSession(os.environ.get("YANDEX_API_KEY"))

# Request pertama (bikin koneksi baru)
r1 = session.get("https://translate.yandex.net/api/v1.5/tr.json/getLangs",
                  params={"key": session.api_key, "ui": "en"})

# Request kedua (reuse koneksi — lebih cepat!)
r2 = session.get("https://translate.yandex.net/api/v1.5/tr.json/translate",
                  params={"key": session.api_key, "text": "test", "lang": "en-ru"})

print(f"Request 1 status: {r1.status_code}")
print(f"Request 2 status: {r2.status_code}")

session.close()

Request 1 status: 200
Request 2 status: 200

# Contoh: tracking biaya API sederhana
class CostTracker:
    """Track estimasi biaya API call."""
    
    PRICING = {
        "translate": 0.00001,      # per character
        "geocode": 0.0002,          # per request
        "search": 0.0001,          # per request
        "tts": 0.0001              # per request
    }
    
    def __init__(self):
        self.costs = {k: 0 for k in self. PRICING}
    
    def add(self, service, units=1):
        """Tambahkan penggunaan ke tracker."""
        cost = self. PRICING.get(service, 0) * units
        self.costs[service] += cost
        return cost
    
    def get_total(self):
        """Total biaya estimasi."""
        return sum(self.costs.values())
    
    def report(self):
        """Laporan biaya per layanan."""
        for service, cost in self.costs.items():
            if cost > 0:
                print(f"  {service}: ${cost:.4f}")
        print(f"  Total: ${self.get_total():.4f}")

tracker = CostTracker()

# Simulasi penggunaan
tracker.add("translate", units=5000)    # 5000 karakter
tracker.add("geocode", units=100)       # 100 request
tracker.add("search", units=50)         # 50 request

print("=== Estimasi Biaya API ===")
tracker.report()

=== Estimasi Biaya API ===
  translate: $0.0500
  geocode: $0.0200
  search: $0.0050
  Total: $0.0750

# Contoh: simpan hasil TTS ke Yandex Object Storage (S3 compatible)
import boto3

def upload_to_storage(audio_data, filename, access_key, secret_key):
    """Upload file ke Yandex Object Storage."""
    s3 = boto3.client(
        "s3",
        endpoint_url="https://storage.yandexcloud.net",
        aws_access_key_id=access_key,
        aws_secret_access_key=secret_key,
        region_name="ru-central1"
    )
    
    s3.put_object(
        Bucket="my-audio-bucket",
        Key=filename,
        Body=audio_data,
        ContentType="audio/ogg"
    )
    
    url = f"https://storage.yandexcloud.net/my-audio-bucket/{filename}"
    return url

# Upload hasil TTS
# audio = text_to_speech("Halo, ini tes.", lang="id-ID")
# url = upload_to_storage(audio, "tes_tts.ogg", "ACCESS_KEY", "SECRET_KEY")
# print(f"Audio tersimpan di: {url}")