odoo-expertos.com/directory/services/apify.py

"""
Apify service for Google Maps scraping.
"""

import logging
from typing import List, Optional, Dict, Any
import time

import httpx

from ..config import get_settings, APIFY_GOOGLE_MAPS_ACTOR, SEARCH_QUERIES
from ..models.schemas import ApifyGoogleMapsResult
from ..utils.retry import retry_with_backoff

logger = logging.getLogger(__name__)


class ApifyService:
    """Service for interacting with Apify Google Maps scraper."""

    BASE_URL = "https://api.apify.com/v2"

    def __init__(self):
        """Initialize Apify client."""
        settings = get_settings()
        self.token = settings.apify_token
        self.headers = {"Authorization": f"Bearer {self.token}"}

    def _get_search_queries(self, city: str, language: str = "de") -> List[str]:
        """
        Generate search queries for a city.

        Args:
            city: City name
            language: Language code for queries

        Returns:
            List of search query strings
        """
        templates = SEARCH_QUERIES.get(language, SEARCH_QUERIES["en"])
        return [template.format(city=city) for template in templates]

    @retry_with_backoff(max_attempts=3, exceptions=(httpx.HTTPError,))
    def run_google_maps_scraper(
        self,
        search_queries: List[str],
        max_results_per_query: int = 20,
        language: str = "de",
        country: str = "DE",
    ) -> str:
        """
        Start a Google Maps scraper run.

        Args:
            search_queries: List of search queries
            max_results_per_query: Max results per query
            language: Language for results
            country: Country code for search

        Returns:
            Run ID for tracking
        """
        url = f"{self.BASE_URL}/acts/{APIFY_GOOGLE_MAPS_ACTOR}/runs"

        payload = {
            "searchStringsArray": search_queries,
            "maxCrawledPlaces": max_results_per_query * len(search_queries),
            "language": language,
            "countryCode": country,
            "maxReviews": 0,  # Don't fetch reviews to save credits
            "maxImages": 0,  # Don't fetch images
            "includeWebResults": False,
        }

        with httpx.Client(timeout=60.0) as client:
            response = client.post(url, json=payload, headers=self.headers)
            response.raise_for_status()
            data = response.json()
            run_id = data["data"]["id"]
            logger.info(f"Started Apify run: {run_id}")
            return run_id

    def wait_for_run(
        self, run_id: str, timeout: int = 300, poll_interval: int = 10
    ) -> str:
        """
        Wait for an Apify run to complete.

        Args:
            run_id: Run ID to wait for
            timeout: Maximum wait time in seconds
            poll_interval: Time between status checks

        Returns:
            Final status of the run
        """
        url = f"{self.BASE_URL}/actor-runs/{run_id}"
        start_time = time.time()

        while True:
            if time.time() - start_time > timeout:
                raise TimeoutError(f"Apify run {run_id} timed out after {timeout}s")

            with httpx.Client(timeout=30.0) as client:
                response = client.get(url, headers=self.headers)
                response.raise_for_status()
                data = response.json()

            status = data["data"]["status"]
            logger.debug(f"Run {run_id} status: {status}")

            if status in ("SUCCEEDED", "FAILED", "ABORTED", "TIMED-OUT"):
                return status

            time.sleep(poll_interval)

    def get_run_results(self, run_id: str) -> List[ApifyGoogleMapsResult]:
        """
        Get results from a completed Apify run.

        Args:
            run_id: Run ID to get results for

        Returns:
            List of Google Maps results
        """
        url = f"{self.BASE_URL}/actor-runs/{run_id}/dataset/items"

        with httpx.Client(timeout=60.0) as client:
            response = client.get(url, headers=self.headers)
            response.raise_for_status()
            data = response.json()

        results = []
        for item in data:
            try:
                result = ApifyGoogleMapsResult(
                    title=item.get("title", ""),
                    address=item.get("address"),
                    phone=item.get("phone"),
                    website=item.get("website"),
                    rating=item.get("totalScore"),
                    reviewsCount=item.get("reviewsCount"),
                    placeId=item.get("placeId"),
                    url=item.get("url"),
                    location=item.get("location"),
                )
                results.append(result)
            except Exception as e:
                logger.warning(f"Failed to parse result: {e}")

        logger.info(f"Retrieved {len(results)} results from run {run_id}")
        return results

    def scrape_city(
        self,
        city_name: str,
        language: str = "de",
        country: str = "DE",
        max_per_query: int = 20,
    ) -> List[ApifyGoogleMapsResult]:
        """
        Scrape Google Maps for Odoo partners in a city.

        This is the main high-level method that combines query generation,
        running the scraper, and retrieving results.

        Args:
            city_name: Name of the city to scrape
            language: Language code
            country: Country code
            max_per_query: Max results per search query

        Returns:
            List of Google Maps results
        """
        queries = self._get_search_queries(city_name, language)
        logger.info(f"Scraping {city_name} with {len(queries)} queries")

        # Start the run
        run_id = self.run_google_maps_scraper(
            search_queries=queries,
            max_results_per_query=max_per_query,
            language=language,
            country=country,
        )

        # Wait for completion
        status = self.wait_for_run(run_id)

        if status != "SUCCEEDED":
            raise RuntimeError(f"Apify run failed with status: {status}")

        # Get results
        return self.get_run_results(run_id)

    def deduplicate_results(
        self, results: List[ApifyGoogleMapsResult]
    ) -> List[ApifyGoogleMapsResult]:
        """
        Remove duplicate results based on place ID.

        Args:
            results: List of results to deduplicate

        Returns:
            Deduplicated list
        """
        seen_place_ids = set()
        unique_results = []

        for result in results:
            if result.placeId and result.placeId not in seen_place_ids:
                seen_place_ids.add(result.placeId)
                unique_results.append(result)
            elif not result.placeId:
                # Keep results without place ID but log warning
                logger.warning(f"Result without place ID: {result.title}")
                unique_results.append(result)

        logger.info(
            f"Deduplicated {len(results)} results to {len(unique_results)} unique"
        )
        return unique_results