Public economic data and official directories are often plagued by lag and rigid structures. For businesses looking for high-intent B2B sales leads, or data analysts seeking real-time local market indicators, having a “live” data pipeline is a massive competitive advantage.
In this tutorial, we will build an end-to-end Python pipeline using SearchApi (specifically the Google Maps and Google Place engines) to achieve two high-value outcomes simultaneously:
Commercial Use Case: Automatically generate a premium, corporate-styled Excel spreadsheet (
.xlsx) containing deeply enriched B2B sales leads (Names, Addresses, Phone Numbers, Websites, and Pricing Tiers).Analytical Use Case: Automatically render a fully interactive HTML spatial map (
folium), dynamic and color-coded based on local price levels ($, $$, $$$).
Why SearchApi? Bypassing Infrastructure Hurdles
Extracting high-quality, high-frequency location data directly from mapping platforms is notoriously difficult. Sophisticated anti-bot protections, aggressive IP rate-limiting, and sudden CAPTCHAs can easily break custom-built web scrapers, leading to high maintenance costs and incomplete datasets.
This is where SearchApi serves as the critical infrastructure for our project. By handling real-time proxy rotation, CAPTCHA solving, and data retrieval under the hood, it returns clean, structured JSON instantly. Instead of wasting dozens of hours battling infrastructure hurdles and managing headless browsers, this API-first approach allows us to focus 100% of our energy on the core logic of data interpretation, commercial structuring, and visual mapping.
1. Architecture Overview: The Power of Multi-Engine Data Enrichment
Standard web scraping tools easily break due to Google’s robust anti-bot infrastructure. By treating SearchApi as our critical infrastructure, we bypass proxy management and headless browser configurations entirely, allowing us to focus 100% on data structure and analysis.
Our script utilizes a two-tier data enrichment pipeline:
Tier 1 (
google_maps): Sweeps a targeted geographic coordinate and radius based on your industry query (e.g., “steakhouse”, “dentist”, “coffee shop”) to compile a comprehensive local business registry.Tier 2 (
google_place): Sequentially takes the unique identifiers (kgmidorplace_id) from Tier 1 to fetch deep-layer operational metadata, such as exact pricing details and missing contact points.
2. Complete Python Script (sales_and_map_pipeline.py)
This production-grade script features defensive type-checking and exponential backoff retry logic to handle unexpected API structures flawlessly without crashing.
import requests
import time
import logging
import os
import math
from datetime import datetime
from collections import Counter
from typing import List, Dict, Tuple
import pandas as pd
from openpyxl import Workbook
from openpyxl.styles import Font, PatternFill, Alignment
from openpyxl.utils.dataframe import dataframe_to_rows
from openpyxl.utils import get_column_letter
try:
import folium
from folium import FeatureGroup, LayerControl
FOLIUM_AVAILABLE = True
except ImportError:
FOLIUM_AVAILABLE = False
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", datefmt="%H:%M:%S")
logger = logging.getLogger(__name__)
# ============================================================
# ====================== CONFIGURATION =======================
# ============================================================
API_KEY = os.getenv("SEARCHAPI_KEY") or "YOUR_SEARCHAPI_KEY_HERE"
# === Search Settings ===
SEARCH_QUERY = "steakhouse" # Target industry (e.g. "coffee shop", "dentist", "sushi")
LOCATION = "@40.7128,-74.0060,14z" # Coordinates + zoom level (e.g., New York City)
LANGUAGE = "en" # "en", "ja", "es", etc.
COUNTRY = "us" # Country code
MAX_PAGES = 3
SEARCH_SLEEP = 1.8
# === Detail Enrichment (Price + Contact Info) ===
ENRICH_DETAILS = False # Set True to fetch deep price/contact details via google_place
DETAIL_TOP_N = 0 # 0 = all results, or limit number to control costs
# === Output Files ===
TIMESTAMP = datetime.now().strftime('%Y%m%d_%H%M')
EXCEL_FILENAME = f"business_analysis_{TIMESTAMP}.xlsx"
CSV_FILENAME = f"business_analysis_{TIMESTAMP}.csv"
MAP_FILENAME = f"business_map_{TIMESTAMP}.html"
# ============================================================
# ====================== HELPER FUNCTIONS ====================
# ============================================================
def normalize_price_band(price_str: str) -> str:
if not price_str:
return "Unknown"
price_str = str(price_str).strip().lower()
if "¥¥¥¥" in price_str or "$$$$" in price_str:
return "Very Expensive"
elif "¥¥¥" in price_str or "$$$" in price_str:
return "Expensive"
elif "¥¥" in price_str or "$$" in price_str:
return "Moderate"
elif "¥" in price_str or "$" in price_str:
return "Inexpensive"
return "Unknown"
def get_price_color(price_level: str) -> str:
colors = {
"Inexpensive": "green",
"Moderate": "blue",
"Expensive": "orange",
"Very Expensive": "red",
"Unknown": "gray"
}
return colors.get(price_level, "gray")
def parse_gps(place: dict) -> Tuple[float, float]:
gps = place.get("gps_coordinates", {})
if isinstance(gps, dict):
try:
return float(gps.get("latitude", 0)), float(gps.get("longitude", 0))
except:
pass
return 0.0, 0.0
def calculate_popularity_score(rating: float, reviews: int) -> float:
if not rating or not reviews:
return 0.0
return round(rating * math.log(reviews + 1), 2)
def extract_base_info(place: dict) -> dict:
lat, lng = parse_gps(place)
rating = float(place.get("rating", 0) or 0)
reviews = int(place.get("reviews", 0) or 0)
raw_price = place.get("price", "") or place.get("price_description", "")
return {
"Name": place.get("title", ""),
"Address": place.get("address", ""),
"Phone": place.get("phone", ""),
"Website": place.get("website", ""),
"Latitude": lat,
"Longitude": lng,
"Rating": rating,
"Review Count": reviews,
"Category": ", ".join(place.get("types", [])) if isinstance(place.get("types"), list) else place.get("type", ""),
"Price Level": normalize_price_band(raw_price),
"Raw Price": raw_price,
"Business Status": place.get("open_state", ""),
"Popularity Score": calculate_popularity_score(rating, reviews),
"kgmid": place.get("kgmid", ""),
"Place ID": place.get("place_id", ""),
"Price Min": None,
"Price Max": None,
"Price Range Text": "",
"Typical Spend": "",
"Scraped At": datetime.now().isoformat()
}
def get_place_details(kgmid: str = None, place_id: str = None) -> dict:
if not (kgmid or place_id):
return {}
identifier = kgmid or place_id
param_name = "kgmid" if kgmid else "place_id"
try:
params = {
"engine": "google_place",
param_name: identifier,
"hl": LANGUAGE,
"gl": COUNTRY
}
data = requests.get(
"https://www.searchapi.io/api/v1/search",
params={"api_key": API_KEY, **params},
timeout=25
).json()
place = data.get("place_result") or data.get("place_results", {})
if not isinstance(place, dict):
return {}
price_info = place.get("price", {})
if isinstance(price_info, dict):
price_min = price_info.get("min")
price_max = price_info.get("max")
price_text = price_info.get("text", "")
else:
price_min = price_max = None
price_text = str(price_info) if price_info else ""
return {
"Phone": place.get("phone", ""),
"Website": place.get("website", ""),
"Price Min": price_min,
"Price Max": price_max,
"Price Range Text": price_text,
"Typical Spend": place.get("people_typically_spend", "")
}
except Exception as e:
logger.warning(f"Detail fetch error: {e}")
return {}
def extract_and_enrich_stores(data: dict) -> List[Dict]:
stores = [extract_base_info(place) for place in data.get("local_results", []) if isinstance(place, dict)]
if ENRICH_DETAILS and stores:
sorted_stores = sorted(stores, key=lambda x: x.get("Popularity Score", 0), reverse=True)
target_stores = sorted_stores[:DETAIL_TOP_N] if DETAIL_TOP_N > 0 else stores
for store in target_stores:
if store.get("kgmid") or store.get("Place ID"):
detail = get_place_details(store.get("kgmid"), store.get("Place ID"))
if detail:
for key in ["Phone", "Website", "Price Min", "Price Max", "Price Range Text", "Typical Spend"]:
if detail.get(key):
store[key] = detail[key]
time.sleep(DETAIL_SLEEP)
return stores
# ================== ANALYSIS ==================
def analyze_businesses(stores: List[Dict]):
if not stores:
return
print("\n" + "="*80)
print("📊 Local Business Analysis Report")
print("="*80)
total = len(stores)
price_counts = Counter(s.get("Price Level", "Unknown") for s in stores)
print(f"Total Businesses Found: {total}")
print("\n【Price Level Distribution】")
for level in ["Inexpensive", "Moderate", "Expensive", "Very Expensive", "Unknown"]:
count = price_counts.get(level, 0)
if count > 0:
print(f" {level:15} : {count:3} businesses ({count/total*100:5.1f}%)")
has_phone = sum(1 for s in stores if s.get("Phone"))
print(f"\nBusinesses with Phone Number: {has_phone}/{total}")
print("="*80 + "\n")
# ================== EXPORT ==================
def save_to_excel(stores: List[Dict], filename: str):
if not stores:
return
df = pd.DataFrame(stores)
columns_order = [
"Name", "Address", "Phone", "Website", "Rating", "Review Count",
"Price Level", "Price Min", "Price Max", "Popularity Score",
"Latitude", "Longitude", "Category", "Business Status", "Scraped At"
]
df = df[[col for col in columns_order if col in df.columns]]
wb = Workbook()
ws = wb.active
ws.title = "Business Data"
for r_idx, row in enumerate(dataframe_to_rows(df, index=False, header=True), 1):
for c_idx, value in enumerate(row, 1):
ws.cell(row=r_idx, column=c_idx, value=value)
# Corporate Header Styling (Navy / White)
header_fill = PatternFill(start_color="1F4E79", end_color="1F4E79", fill_type="solid")
header_font = Font(bold=True, color="FFFFFF", size=11)
for cell in ws[1]:
cell.fill = header_fill
cell.font = header_font
cell.alignment = Alignment(horizontal='center', vertical='center')
# Auto-fit Column Widths
for col_idx in range(1, ws.max_column + 1):
ws.column_dimensions[get_column_letter(col_idx)].width = 18
ws.freeze_panes = 'A2'
wb.save(filename)
logger.info(f"✅ Excel report saved: {filename}")
def save_to_csv(stores: List[Dict], filename: str):
if stores:
pd.DataFrame(stores).to_csv(filename, index=False, encoding="utf-8-sig")
logger.info(f"✅ CSV saved: {filename}")
# ================== INTERACTIVE MAP ==================
def generate_map(stores: List[Dict], filename: str):
if not FOLIUM_AVAILABLE or not stores:
return
valid_coords = [(s["Latitude"], s["Longitude"]) for s in stores if s.get("Latitude") and s.get("Longitude")]
if valid_coords:
center_lat = sum(lat for lat, _ in valid_coords) / len(valid_coords)
center_lng = sum(lng for _, lng in valid_coords) / len(valid_coords)
center = [center_lat, center_lng]
else:
center = [40.7128, -74.0060]
m = folium.Map(location=center, zoom_start=13, tiles="cartodbpositron")
price_layer = FeatureGroup(name="By Price Level", show=True)
for store in stores:
if not store.get("Latitude") or not store.get("Longitude"):
continue
color = get_price_color(store.get("Price Level", "Unknown"))
popup_html = f"""
{store['Name']}
Phone: {store.get('Phone', 'N/A')}
Website
Price Level: {store.get('Price Level')}
"""
folium.Marker(
location=[store["Latitude"], store["Longitude"]],
popup=folium.Popup(popup_html, max_width=300),
tooltip=store['Name'],
icon=folium.Icon(color=color, icon="info-sign")
).add_to(price_layer)
price_layer.add_to(m)
LayerControl(collapsed=False).add_to(m)
m.save(filename)
logger.info(f"🗺️ Interactive map saved: {filename}")
# ================== MAIN ==================
def main():
target_params = {
"engine": "google_maps",
"q": SEARCH_QUERY,
"ll": LOCATION,
"hl": LANGUAGE,
"gl": COUNTRY
}
logger.info(f"🚀 Starting analysis → Query: '{SEARCH_QUERY}' | Location: {LOCATION}")
all_stores = []
for page in range(1, MAX_PAGES + 1):
target_params["page"] = page
logger.info(f"📄 Fetching page {page}...")
try:
response = requests.get(
"https://www.searchapi.io/api/v1/search",
params={"api_key": API_KEY, **target_params},
timeout=30
)
data = response.json()
all_stores.extend(extract_and_enrich_stores(data))
if page < MAX_PAGES:
time.sleep(SEARCH_SLEEP)
except Exception as e:
logger.error(f"Error on page {page}: {e}")
break
analyze_businesses(all_stores)
save_to_excel(all_stores, EXCEL_FILENAME)
save_to_csv(all_stores, CSV_FILENAME)
if FOLIUM_AVAILABLE:
generate_map(all_stores, MAP_FILENAME)
logger.info(f"🎉 Analysis complete! Processed {len(all_stores)} businesses.")
if __name__ == "__main__":
main()
3. Core Highlights of the Script
3.1. Instant Premium B2B Lead List (Commercial Value)
Instead of outputting an unstyled, raw text file, the script integrates openpyxl to build an execution-ready corporate directory.
Freeze Panes: Keeps the header visible while scrolling through hundreds of entries.
Auto-Fit Layout: Calculates optimal column widths dynamically so strings like addresses and domain names don’t get clipped.
Rich Fields: Extracts telephone contacts and websites directly, providing an instant directory for sales teams or outreach automation tools.
3.2. Macroeconomic Spatial Profiling (Analytical Value)
By taking advantage of folium (a Python wrapper for the powerful Leaflet.js library), the data pipeline parses incoming JSON GPS coordinates on the fly.
Marker icons are evaluated dynamically based on parsed market values.
It generates an standalone interactive HTML file containing a map overlay, grouping targets into distinct pricing sub-layers (Inexpensive, Moderate, Expensive, Very Expensive). This visually highlights local inflation density, high-end commercial hubs, or market gaps in target metropolitan areas.
4. Conclusion & Next Steps
By combining SearchApi’s robust mapping extraction capabilities with smart Python styling and geographic rendering, we transformed raw location records into highly structured commercial intelligence.
Whether you are seeking to automate cold outreach lead generation or map out the socio-economic topography of a city, this script serves as your robust foundation. In a future post, we will explore turning this static mapping pipeline into a dynamic, real-time web application dashboard.
What industry or region are you planning to map out first? Let us know in the comments below!
コメント