"""CLI entry point for gmaps-scraper."""
from __future__ import annotations
import argparse
import asyncio
import json
import logging
import sys
from pathlib import Path
from gmaps_scraper.batch import load_urls_from_file, scrape_batch
from gmaps_scraper.models import ScrapeConfig, ScrapeResult
from gmaps_scraper.scraper import GoogleMapsScraper
def main() -> None:
"""Main CLI entry point."""
parser = argparse.ArgumentParser(
prog="gmaps-scraper",
description="Scrape Google Maps place details without API key",
)
parser.add_argument(
"-v", "--verbose",
action="store_true",
help="Enable verbose logging",
)
subparsers = parser.add_subparsers(dest="command", required=True)
# --- scrape command ---
scrape_parser = subparsers.add_parser(
"scrape",
help="Scrape a single Google Maps URL",
)
scrape_parser.add_argument(
"url",
help="Google Maps URL to scrape",
)
scrape_parser.add_argument(
"--lang",
default=None,
help="Language code (e.g., zh-TW, en, ja). Default: none",
)
scrape_parser.add_argument(
"--no-headless",
action="store_true",
help="Show browser window (for debugging)",
)
# --- batch command ---
batch_parser = subparsers.add_parser(
"batch",
help="Batch scrape URLs from a file",
)
batch_parser.add_argument(
"input",
help="Input file (CSV or text, one URL per line)",
)
batch_parser.add_argument(
"-o", "--output",
required=True,
help="Output file path (.json or .csv)",
)
batch_parser.add_argument(
"--concurrency",
type=int,
default=5,
help="Number of concurrent scrapers (default: 5)",
)
batch_parser.add_argument(
"--lang",
default=None,
help="Language code (e.g., zh-TW, en, ja). Default: none",
)
batch_parser.add_argument(
"--no-headless",
action="store_true",
help="Show browser window (for debugging)",
)
batch_parser.add_argument(
"--proxy",
help="Proxy server URL (e.g., http://proxy:8080)",
)
batch_parser.add_argument(
"--delay-min",
type=float,
default=2.0,
help="Minimum delay between requests in seconds (default: 2.0)",
)
batch_parser.add_argument(
"--delay-max",
type=float,
default=5.0,
help="Maximum delay between requests in seconds (default: 5.0)",
)
batch_parser.add_argument(
"--no-resume",
action="store_true",
help="Don't resume from existing output file",
)
batch_parser.add_argument(
"--save-interval",
type=int,
default=50,
help="Save results every N items (default: 50)",
)
args = parser.parse_args()
# Setup logging
log_level = logging.DEBUG if args.verbose else logging.INFO
logging.basicConfig(
level=log_level,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
if args.command == "scrape":
_cmd_scrape(args)
elif args.command == "batch":
_cmd_batch(args)
def _cmd_scrape(args: argparse.Namespace) -> None:
"""Handle `gmaps-scraper scrape` command."""
config = ScrapeConfig(
headless=not args.no_headless,
language=args.lang,
)
async def _run() -> ScrapeResult:
async with GoogleMapsScraper(config) as scraper:
return await scraper.scrape(args.url)
result = asyncio.run(_run())
# Output as JSON
output = result.model_dump(mode="json")
print(json.dumps(output, ensure_ascii=False, indent=2, default=str))
if not result.success:
sys.exit(1)
def _cmd_batch(args: argparse.Namespace) -> None:
"""Handle `gmaps-scraper batch` command."""
input_path = Path(args.input)
if not input_path.exists():
print(f"Error: Input file not found: {input_path}", file=sys.stderr)
sys.exit(1)
urls = load_urls_from_file(input_path)
if not urls:
print("Error: No URLs found in input file", file=sys.stderr)
sys.exit(1)
print(f"Loaded {len(urls)} URLs from {input_path}")
config = ScrapeConfig(
concurrency=args.concurrency,
headless=not args.no_headless,
language=args.lang,
proxy=args.proxy,
delay_min=args.delay_min,
delay_max=args.delay_max,
save_interval=args.save_interval,
)
# Progress tracking
try:
from tqdm import tqdm
pbar = tqdm(total=len(urls), desc="Scraping", unit="place")
def on_result(result: ScrapeResult, idx: int, total: int) -> None:
pbar.update(1)
status = "✓" if result.success else "✗"
name = result.place.name if result.place else "?"
pbar.set_postfix_str(f"{status} {name}")
except ImportError:
def on_result(result: ScrapeResult, idx: int, total: int) -> None:
status = "✓" if result.success else "✗"
name = result.place.name if result.place else "?"
print(f" [{idx}/{total}] {status} {name}")
pbar = None
results = asyncio.run(
scrape_batch(
urls=urls,
config=config,
output_path=args.output,
on_result=on_result,
resume=not args.no_resume,
)
)
if pbar:
pbar.close()
# Summary
success_count = sum(1 for r in results if r.success)
fail_count = len(results) - success_count
print(f"\nDone! {success_count} succeeded, {fail_count} failed")
print(f"Results saved to: {args.output}")
if __name__ == "__main__":
main()