#!/usr/bin/env python3 """ LANGKAH 1 — Panen jejak digital usaha lewat Google Places API (resmi). Skrip ini memakai Places API (New) "Text Search" — jalur resmi dan sah dari Google, BUKAN scraping halaman Google Maps (yang melanggar Ketentuan Layanan Google). Cara kerja: 1. Wilayah (kotak lintang-bujur) dibagi menjadi sel-sel grid. 2. Untuk setiap sel dan setiap kata kunci, API dipanggil (maksimal 60 hasil per panggilan). 3. Bila satu sel "penuh" (60 hasil), sel itu dipecah empat dan dipanen ulang, supaya usaha di kawasan padat tidak terlewat. 4. Hasil digabung, duplikat place_id dibuang, lalu disimpan ke CSV. Kebutuhan: - Python 3.9+ (tanpa pustaka tambahan). - API key Google Cloud dengan "Places API (New)" aktif dan penagihan (billing) diaktifkan. Panggilan API BERBAYAR per permintaan. Cek harga dan kuota gratis bulanan terbaru di halaman harga Google Maps Platform sebelum panen skala penuh. Contoh pemakaian: export GOOGLE_PLACES_API_KEY="isi-api-key-anda" python3 1_panen_places.py \ --bbox -7.80,110.30,-7.65,110.48 \ --kata-kunci "hotel" "penginapan" "restoran" "kafe" "karaoke" \ --grid 4 --keluaran panen.csv Uji dulu tanpa memanggil API (hanya menampilkan rencana sel dan jumlah panggilan): python3 1_panen_places.py --bbox -7.80,110.30,-7.65,110.48 --kata-kunci hotel --rencana Kolom keluaran: place_id, nama, alamat, lat, lng, tipe_utama, tipe, rating, jumlah_ulasan, tingkat_harga, status_usaha, kueri, sumber """ import argparse import csv import json import os import sys import time import urllib.error import urllib.request URL_API = "https://places.googleapis.com/v1/places:searchText" # Hanya minta kolom yang dipakai — Google menagih berdasarkan kolom yang diminta. FIELD_MASK = ",".join([ "places.id", "places.displayName", "places.formattedAddress", "places.location", "places.primaryType", "places.types", "places.rating", "places.userRatingCount", "places.priceLevel", "places.businessStatus", "nextPageToken", ]) HASIL_MAKS_PER_KUERI = 60 # batas dari Google: 3 halaman × 20 hasil KOLOM = [ "place_id", "nama", "alamat", "lat", "lng", "tipe_utama", "tipe", "rating", "jumlah_ulasan", "tingkat_harga", "status_usaha", "kueri", "sumber", ] # priceLevel dari API berupa teks; diubah ke angka 0–4 supaya mudah diskor. TINGKAT_HARGA = { "PRICE_LEVEL_FREE": 0, "PRICE_LEVEL_INEXPENSIVE": 1, "PRICE_LEVEL_MODERATE": 2, "PRICE_LEVEL_EXPENSIVE": 3, "PRICE_LEVEL_VERY_EXPENSIVE": 4, } def bagi_grid(bbox, n): """Bagi kotak (lat_min, lng_min, lat_maks, lng_maks) menjadi n×n sel.""" lat_min, lng_min, lat_maks, lng_maks = bbox d_lat = (lat_maks - lat_min) / n d_lng = (lng_maks - lng_min) / n sel = [] for i in range(n): for j in range(n): sel.append(( lat_min + i * d_lat, lng_min + j * d_lng, lat_min + (i + 1) * d_lat, lng_min + (j + 1) * d_lng, )) return sel def panggil_api(api_key, kueri, sel, page_token=None, coba=3): badan = { "textQuery": kueri, "pageSize": 20, "languageCode": "id", "regionCode": "ID", "locationRestriction": { "rectangle": { "low": {"latitude": sel[0], "longitude": sel[1]}, "high": {"latitude": sel[2], "longitude": sel[3]}, } }, } if page_token: badan["pageToken"] = page_token req = urllib.request.Request( URL_API, data=json.dumps(badan).encode("utf-8"), headers={ "Content-Type": "application/json", "X-Goog-Api-Key": api_key, "X-Goog-FieldMask": FIELD_MASK, }, method="POST", ) for percobaan in range(coba): try: with urllib.request.urlopen(req, timeout=30) as resp: return json.loads(resp.read().decode("utf-8")) except urllib.error.HTTPError as e: pesan = e.read().decode("utf-8", errors="replace") # 429 = kuota per menit habis; tunggu lalu coba lagi. if e.code in (429, 500, 503) and percobaan < coba - 1: time.sleep(2 ** (percobaan + 1)) continue sys.exit(f"Galat API {e.code}: {pesan}") except urllib.error.URLError as e: if percobaan < coba - 1: time.sleep(2 ** (percobaan + 1)) continue sys.exit(f"Gagal terhubung ke API: {e.reason}") return {} def ubah_baris(place, kueri): lokasi = place.get("location", {}) return { "place_id": place.get("id", ""), "nama": place.get("displayName", {}).get("text", ""), "alamat": place.get("formattedAddress", ""), "lat": lokasi.get("latitude", ""), "lng": lokasi.get("longitude", ""), "tipe_utama": place.get("primaryType", ""), "tipe": "|".join(place.get("types", [])), "rating": place.get("rating", ""), "jumlah_ulasan": place.get("userRatingCount", 0), "tingkat_harga": TINGKAT_HARGA.get(place.get("priceLevel", ""), ""), "status_usaha": place.get("businessStatus", ""), "kueri": kueri, "sumber": "google_places", } def panen_sel(api_key, kueri, sel, kedalaman, kedalaman_maks, jeda): """Panen satu sel; pecah empat bila hasilnya mentok di batas 60.""" hasil = [] token = None while True: data = panggil_api(api_key, kueri, sel, token) hasil.extend(ubah_baris(p, kueri) for p in data.get("places", [])) token = data.get("nextPageToken") if not token: break time.sleep(jeda) if len(hasil) >= HASIL_MAKS_PER_KUERI and kedalaman < kedalaman_maks: print(f" sel penuh ({len(hasil)} hasil) → dipecah 4, kedalaman {kedalaman + 1}") for sub in bagi_grid(sel, 2): hasil.extend(panen_sel(api_key, kueri, sub, kedalaman + 1, kedalaman_maks, jeda)) return hasil def main(): p = argparse.ArgumentParser(description="Panen data usaha lewat Google Places API (New).") p.add_argument("--bbox", required=True, help="lat_min,lng_min,lat_maks,lng_maks — kotak yang menutupi seluruh wilayah") p.add_argument("--kata-kunci", nargs="+", required=True, help='kata kunci pencarian, mis. "hotel" "restoran" "karaoke"') p.add_argument("--grid", type=int, default=4, help="jumlah sel per sisi (default 4 → 16 sel)") p.add_argument("--kedalaman-maks", type=int, default=3, help="berapa kali sel penuh boleh dipecah lagi (default 3)") p.add_argument("--jeda", type=float, default=0.3, help="jeda antarpanggilan, detik") p.add_argument("--keluaran", default="panen.csv") p.add_argument("--rencana", action="store_true", help="tampilkan rencana tanpa memanggil API (gratis)") a = p.parse_args() try: bbox = tuple(float(x) for x in a.bbox.split(",")) assert len(bbox) == 4 and bbox[0] < bbox[2] and bbox[1] < bbox[3] except (ValueError, AssertionError): sys.exit("--bbox harus berformat lat_min,lng_min,lat_maks,lng_maks (mis. -7.80,110.30,-7.65,110.48)") sel_awal = bagi_grid(bbox, a.grid) print(f"Wilayah dibagi {len(sel_awal)} sel × {len(a.kata_kunci)} kata kunci " f"= {len(sel_awal) * len(a.kata_kunci)} kueri dasar (masing-masing 1–3 panggilan API, " f"belum termasuk pemecahan sel padat).") if a.rencana: for i, s in enumerate(sel_awal, 1): print(f" sel {i:>2}: {s[0]:.4f},{s[1]:.4f} → {s[2]:.4f},{s[3]:.4f}") return api_key = os.environ.get("GOOGLE_PLACES_API_KEY") if not api_key: sys.exit("Set dulu variabel lingkungan GOOGLE_PLACES_API_KEY (lihat README).") semua = {} for kueri in a.kata_kunci: for i, sel in enumerate(sel_awal, 1): print(f"[{kueri}] sel {i}/{len(sel_awal)}") for baris in panen_sel(api_key, kueri, sel, 0, a.kedalaman_maks, a.jeda): # Tempat yang sama bisa muncul di beberapa kueri; simpan sekali saja. semua.setdefault(baris["place_id"], baris) with open(a.keluaran, "w", newline="", encoding="utf-8") as f: w = csv.DictWriter(f, fieldnames=KOLOM) w.writeheader() w.writerows(semua.values()) print(f"Selesai: {len(semua)} tempat unik disimpan ke {a.keluaran}") if __name__ == "__main__": main()