"""Async batch processor for high-throughput Google Maps scraping."""
from __future__ import annotations
import asyncio
import csv
import json
import logging
import random
from datetime import datetime
from pathlib import Path
from typing import Callable, Optional
from gmaps_scraper.models import ScrapeConfig, ScrapeResult
from gmaps_scraper.scraper import GoogleMapsScraper
logger = logging.getLogger("gmaps_scraper")
async def scrape_batch(
urls: list[str],
config: ScrapeConfig | None = None,
output_path: str | Path | None = None,
on_result: Optional[Callable[[ScrapeResult, int, int], None]] = None,
resume: bool = True,
) -> list[ScrapeResult]:
"""Scrape a batch of Google Maps URLs concurrently.
Args:
urls: List of Google Maps URLs to scrape.
config: Scraper configuration (concurrency, delays, etc.).
output_path: Path to save results incrementally (JSON or CSV).
on_result: Callback called after each result (result, index, total).
resume: If True, skip URLs already present in the output file.
Returns:
List of ScrapeResult objects.
Example:
>>> import asyncio
>>> from gmaps_scraper import scrape_batch, ScrapeConfig
>>> urls = ["https://www.google.com/maps/search/...", ...]
>>> config = ScrapeConfig(concurrency=5, headless=True)
>>> results = asyncio.run(scrape_batch(urls, config))
"""
config = config or ScrapeConfig()
output_path = Path(output_path) if output_path else None
# Load already scraped URLs for resume
completed_urls: set[str] = set()
existing_results: list[ScrapeResult] = []
if resume and output_path and output_path.exists():
existing_results, completed_urls = _load_existing_results(output_path)
if completed_urls:
logger.info(
"Resuming: %d URLs already scraped, %d remaining",
len(completed_urls),
len(urls) - len(completed_urls),
)
# Filter out already completed URLs
remaining_urls = [u for u in urls if u.strip() not in completed_urls]
total = len(urls)
completed_count = len(completed_urls)
if not remaining_urls:
logger.info("All %d URLs already scraped!", total)
return existing_results
results: list[ScrapeResult] = list(existing_results)
semaphore = asyncio.Semaphore(config.concurrency)
save_lock = asyncio.Lock()
async with GoogleMapsScraper(config) as scraper:
async def _scrape_one(url: str) -> ScrapeResult:
nonlocal completed_count
async with semaphore:
# Random delay to avoid rate limiting
delay = random.uniform(config.delay_min, config.delay_max)
await asyncio.sleep(delay)
try:
result = await scraper.scrape(url)
except Exception as e:
result = ScrapeResult(
input_url=url,
success=False,
error=str(e),
)
completed_count += 1
# Callback
if on_result:
try:
on_result(result, completed_count, total)
except Exception:
pass
# Periodic save
async with save_lock:
results.append(result)
if (
output_path
and completed_count % config.save_interval == 0
):
_save_results(results, output_path)
logger.info(
"Progress saved: %d/%d (%.1f%%)",
completed_count,
total,
completed_count / total * 100,
)
return result
# Run all tasks with concurrency control
tasks = [_scrape_one(url) for url in remaining_urls]
await asyncio.gather(*tasks, return_exceptions=True)
# Final save
if output_path:
_save_results(results, output_path)
logger.info("Final results saved to %s", output_path)
return results
def _load_existing_results(
path: Path,
) -> tuple[list[ScrapeResult], set[str]]:
"""Load existing results for resume capability."""
results: list[ScrapeResult] = []
urls: set[str] = set()
try:
if path.suffix == ".json":
with open(path, "r", encoding="utf-8") as f:
data = json.load(f)
for item in data:
result = ScrapeResult(**item)
results.append(result)
urls.add(result.input_url)
elif path.suffix == ".csv":
with open(path, "r", encoding="utf-8") as f:
reader = csv.DictReader(f)
for row in reader:
urls.add(row.get("input_url", ""))
# For CSV resume, we just track URLs, not full reconstruction
except Exception as e:
logger.warning("Failed to load existing results from %s: %s", path, e)
return results, urls
def _save_results(results: list[ScrapeResult], path: Path) -> None:
"""Save results to JSON or CSV file."""
path.parent.mkdir(parents=True, exist_ok=True)
if path.suffix == ".csv":
_save_csv(results, path)
else:
_save_json(results, path)
def _save_json(results: list[ScrapeResult], path: Path) -> None:
"""Save results as JSON."""
data = []
for r in results:
item = r.model_dump(mode="json")
data.append(item)
with open(path, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2, default=str)
def _save_csv(results: list[ScrapeResult], path: Path) -> None:
"""Save results as flat CSV (place details only, no reviews)."""
if not results:
return
fieldnames = [
"input_url",
"success",
"error",
"scraped_at",
"name",
"place_id",
"address",
"rating",
"review_count",
"phone",
"website",
"category",
"price_level",
"latitude",
"longitude",
"plus_code",
"url",
"google_maps_url",
"description",
"photos_count",
"permanently_closed",
"temporarily_closed",
]
with open(path, "w", encoding="utf-8-sig", newline="") as f:
writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction="ignore")
writer.writeheader()
for r in results:
row = {
"input_url": r.input_url,
"success": r.success,
"error": r.error,
"scraped_at": str(r.scraped_at),
}
if r.place:
place_dict = r.place.model_dump()
# Flatten hours to a single string
if place_dict.get("hours"):
place_dict["hours"] = " | ".join(place_dict["hours"])
row.update(place_dict)
writer.writerow(row)
def load_urls_from_file(path: str | Path) -> list[str]:
"""Load URLs from a text file or CSV file.
Supports:
- Plain text file (one URL per line)
- CSV file (looks for 'url' column, or uses first column)
Args:
path: Path to the input file.
Returns:
List of URL strings.
"""
path = Path(path)
if path.suffix == ".csv":
urls = []
with open(path, "r", encoding="utf-8") as f:
reader = csv.DictReader(f)
# Try 'url' column first, then first column
for row in reader:
url = row.get("url") or row.get("URL") or row.get("link")
if not url:
# Use first column value
url = next(iter(row.values()), None)
if url and url.strip():
urls.append(url.strip())
return urls
else:
# Plain text, one URL per line
with open(path, "r", encoding="utf-8") as f:
return [
line.strip()
for line in f
if line.strip() and not line.startswith("#")
]