"""Utility functions for URL parsing, retry logic, and helpers."""
from __future__ import annotations
import asyncio
import logging
import random
import re
from functools import wraps
from typing import Any, Callable
from urllib.parse import parse_qs, urlparse
logger = logging.getLogger("gmaps_scraper")
# Realistic user agents for rotation
USER_AGENTS = [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:133.0) Gecko/20100101 Firefox/133.0",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/18.2 Safari/605.1.15",
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
]
def get_random_user_agent() -> str:
"""Return a random user agent string."""
return random.choice(USER_AGENTS)
def parse_google_maps_url(url: str) -> dict[str, Any]:
"""Parse a Google Maps URL and extract relevant parameters.
Supports formats:
- https://www.google.com/maps/search/?api=1&query=...&query_place_id=...&hl=ja
- https://www.google.com/maps/place/...
- https://maps.google.com/...
Returns:
dict with keys: query, query_place_id, hl, original_url
"""
parsed = urlparse(url)
params = parse_qs(parsed.query)
result: dict[str, Any] = {
"original_url": url,
"query": None,
"query_place_id": None,
"hl": None,
}
# Extract query parameters
if "query" in params:
result["query"] = params["query"][0]
if "query_place_id" in params:
result["query_place_id"] = params["query_place_id"][0]
if "hl" in params:
result["hl"] = params["hl"][0]
# Try to extract place_id from /maps/place/ URLs
if not result["query_place_id"]:
place_id_match = re.search(r"place_id[=:]([A-Za-z0-9_-]+)", url)
if place_id_match:
result["query_place_id"] = place_id_match.group(1)
return result
def normalize_google_maps_url(url: str, language: str | None = None) -> str:
"""Normalize a Google Maps URL to a consistent format.
Ensures the URL is a proper search URL with the specified language.
"""
url = url.strip()
# If it's already a search URL, ensure language param overrides existing ones
parsed = urlparse(url)
params = parse_qs(parsed.query)
if language:
if "hl" in params:
# Replace existing hl param
url = re.sub(r'([?&])hl=[^&]*', rf'\g<1>hl={language}', url)
else:
# Append new hl param
separator = "&" if parsed.query else "?"
url = f"{url}{separator}hl={language}"
return url
def extract_coordinates_from_url(url: str) -> tuple[float, float] | None:
"""Extract latitude/longitude from a Google Maps URL."""
# Match @lat,lng pattern
coord_match = re.search(r"@(-?\d+\.?\d*),(-?\d+\.?\d*)", url)
if coord_match:
return float(coord_match.group(1)), float(coord_match.group(2))
# Match query=lat,lng pattern
parsed = urlparse(url)
params = parse_qs(parsed.query)
if "query" in params:
query = params["query"][0]
coord_match = re.search(r"^(-?\d+\.?\d*),\s*(-?\d+\.?\d*)$", query)
if coord_match:
return float(coord_match.group(1)), float(coord_match.group(2))
return None
def validate_url(url: str) -> bool:
"""Check if a URL is a valid Google Maps URL."""
try:
parsed = urlparse(url)
return (
parsed.scheme in ("http", "https")
and "google" in parsed.netloc
and "map" in parsed.path.lower()
) or (
parsed.scheme in ("http", "https")
and "maps.google" in parsed.netloc
)
except Exception:
return False
def retry_async(
max_retries: int = 3,
base_delay: float = 2.0,
max_delay: float = 30.0,
) -> Callable:
"""Decorator for async retry with exponential backoff."""
def decorator(func: Callable) -> Callable:
@wraps(func)
async def wrapper(*args: Any, **kwargs: Any) -> Any:
last_exception = None
for attempt in range(max_retries + 1):
try:
return await func(*args, **kwargs)
except Exception as e:
last_exception = e
if attempt < max_retries:
delay = min(
base_delay * (2**attempt) + random.uniform(0, 1),
max_delay,
)
logger.warning(
"Attempt %d/%d failed for %s: %s. Retrying in %.1fs",
attempt + 1,
max_retries + 1,
func.__name__,
str(e),
delay,
)
await asyncio.sleep(delay)
raise last_exception # type: ignore[misc]
return wrapper
return decorator
def safe_int(value: str | None) -> int | None:
"""Safely parse a string to int, removing non-numeric characters."""
if not value:
return None
# Remove commas, spaces, and other non-numeric chars (keep digits)
cleaned = re.sub(r"[^\d]", "", value)
return int(cleaned) if cleaned else None
def parse_review_count(value: str | int | None) -> int | None:
"""Parse a review count string that might contain multipliers like K, M, 萬, 万."""
if not value:
return None
if isinstance(value, int):
return value
value = str(value).strip().upper()
multiplier = 1
if "K" in value:
multiplier = 1000
elif "M" in value:
multiplier = 1000000
elif "萬" in value or "万" in value:
multiplier = 10000
value = value.replace("K", "").replace("M", "").replace("萬", "").replace("万", "")
m = re.search(r"(\d+(?:\.\d+)?)", value.replace(",", ""))
if m:
try:
return int(float(m.group(1)) * multiplier)
except ValueError:
pass
return None
def safe_float(value: str | None) -> float | None:
"""Safely parse a string to float."""
if not value:
return None
# Remove all but digits, dots, and minus signs
cleaned = re.sub(r"[^\d.\-]", "", value)
try:
return float(cleaned) if cleaned else None
except ValueError:
return None