scraper.py

"""Marktplaats scraper using ScrapingAnt API."""

import os
import re
import time
from typing import Optional
from urllib.parse import urlencode

import requests
from bs4 import BeautifulSoup

from config import (
    SCRAPINGANT_BASE_URL,
    MARKTPLAATS_BASE_URL,
    REQUEST_CONFIG,
    CATEGORIES,
    SELECTORS,
)
from models import Listing, ListingCollection
from utils import (
    clean_text,
    parse_listing_text,
    build_category_url,
    build_listing_url,
    format_price,
    truncate_text,
    extract_condition,
    extract_shipping,
    extract_date,
)


class MarktplaatsScraper:
    """Scraper for Marktplaats listings using ScrapingAnt API."""

    def __init__(self, api_key: Optional[str] = None, verbose: bool = False):
        """
        Initialize the scraper.

        Args:
            api_key: ScrapingAnt API key. If not provided, uses SCRAPINGANT_API_KEY env var.
            verbose: Enable verbose output.
        """
        self.api_key = api_key or os.environ.get("SCRAPINGANT_API_KEY")
        if not self.api_key:
            raise ValueError(
                "ScrapingAnt API key is required. "
                "Set SCRAPINGANT_API_KEY environment variable or pass api_key parameter."
            )
        self.verbose = verbose
        self.collection = ListingCollection()

    def _log(self, message: str):
        """Print message if verbose mode is enabled."""
        if self.verbose:
            print(message)

    def _build_url(self, target_url: str) -> str:
        """Build ScrapingAnt API URL with parameters."""
        params = {
            "url": target_url,
            "x-api-key": self.api_key,
            "browser": "true",
            "wait_for_selector": SELECTORS["wait_for_content"],
            "return_page_source": "true",
            "proxy_country": "NL",  # Use Dutch proxy for Marktplaats
            "proxy_type": "residential",
        }
        return f"{SCRAPINGANT_BASE_URL}?{urlencode(params)}"

    def _fetch_page(self, category: str, page: int = 1) -> Optional[str]:
        """
        Fetch a page from Marktplaats.

        Args:
            category: Category path (e.g., 'computers-en-software/windows-laptops')
            page: Page number (1-indexed)

        Returns:
            HTML content or None if request failed.
        """
        target_url = build_category_url(MARKTPLAATS_BASE_URL, category, page)
        api_url = self._build_url(target_url)

        self._log(f"Fetching: {target_url} (page {page})")

        try:
            response = requests.get(api_url, timeout=REQUEST_CONFIG["timeout"] / 1000)
            response.raise_for_status()
            return response.text
        except requests.RequestException as e:
            self._log(f"Error fetching {target_url}: {e}")
            return None

    def _parse_listing(self, listing_elem: BeautifulSoup, category: str) -> Optional[Listing]:
        """
        Parse a single listing from a listing element.

        Args:
            listing_elem: BeautifulSoup element containing listing
            category: Category name for the listing

        Returns:
            Listing object or None if parsing failed.
        """
        try:
            # Get listing URL
            link = listing_elem.select_one(SELECTORS["listing_link"])
            if not link:
                return None

            href = link.get("href", "")
            if not href or "/v/" not in href:
                return None

            listing_url = build_listing_url(MARKTPLAATS_BASE_URL, href)

            # Get title from h3
            title_elem = listing_elem.select_one(SELECTORS["title"])
            title = title_elem.get_text(strip=True) if title_elem else ""

            # Get full text content for parsing
            full_text = listing_elem.get_text(" ", strip=True)

            # Parse structured data
            parsed = parse_listing_text(full_text, title)

            # Skip if no title
            if not title:
                return None

            # Get category display name
            category_name = CATEGORIES.get(category, category)

            # Try to extract location from seller section
            location = ""
            seller_section = listing_elem.select_one('[class*="seller"], [class*="Seller"]')
            if seller_section:
                seller_text = seller_section.get_text(" ", strip=True)
                # Location is usually a city name in the seller section
                # Common Dutch cities or "Heel Nederland"
                location_match = re.search(
                    r"(Heel Nederland|Amsterdam|Rotterdam|Den Haag|Utrecht|Eindhoven|"
                    r"Groningen|Tilburg|Almere|Breda|Nijmegen|Apeldoorn|Haarlem|"
                    r"Arnhem|Enschede|Amersfoort|Zaanstad|Haarlemmermeer|"
                    r"[A-Z][a-z]+(?:\s[A-Z][a-z]+)?)",
                    seller_text
                )
                if location_match:
                    location = location_match.group(1)

            # Extract seller name (usually at the beginning of seller section)
            seller_name = ""
            if seller_section:
                seller_text = seller_section.get_text(" ", strip=True)
                # Seller name is often at the start, before location
                parts = seller_text.split()
                if parts:
                    # Take first 1-3 words as seller name
                    seller_name = " ".join(parts[:3])
                    # Clean up common suffixes
                    seller_name = re.sub(r"(Heel Nederland|Vandaag|Topadvertentie|Bezoek website).*", "", seller_name).strip()

            return Listing(
                title=clean_text(title),
                price=format_price(parsed["price"]),
                description=truncate_text(parsed["description"], 500),
                location=location,
                listing_url=listing_url,
                category=category_name,
                condition=parsed["condition"],
                shipping=parsed["shipping"],
                seller_name=seller_name[:100],
                date_posted=parsed["date_posted"],
            )

        except Exception as e:
            self._log(f"Error parsing listing: {e}")
            return None

    def _parse_listings(self, html: str, category: str) -> list[Listing]:
        """
        Parse all listings from HTML content.

        Args:
            html: HTML content of the page
            category: Category path

        Returns:
            List of Listing objects.
        """
        listings = []
        soup = BeautifulSoup(html, "lxml")

        # Find all listing containers
        listing_elems = soup.select(SELECTORS["listing_container"])

        self._log(f"Found {len(listing_elems)} listing elements")

        seen_urls = set()
        for elem in listing_elems:
            listing = self._parse_listing(elem, category)
            if listing and listing.listing_url not in seen_urls:
                listings.append(listing)
                seen_urls.add(listing.listing_url)

        return listings

    def scrape_category(self, category: str, max_pages: int = 2) -> int:
        """
        Scrape listings from a specific category.

        Args:
            category: Category path (e.g., 'computers-en-software/windows-laptops')
            max_pages: Maximum number of pages to scrape

        Returns:
            Number of new listings added.
        """
        category_name = CATEGORIES.get(category, category)
        total_added = 0

        for page in range(1, max_pages + 1):
            html = self._fetch_page(category, page)
            if not html:
                self._log(f"Failed to fetch page {page} for {category_name}")
                break

            listings = self._parse_listings(html, category)
            self._log(f"Parsed {len(listings)} listings from {category_name} (page {page})")

            if not listings:
                self._log(f"No listings found on page {page}, stopping pagination")
                break

            added = self.collection.add_many(listings)
            total_added += added
            self._log(f"Added {added} new listings (total: {len(self.collection)})")

            # Small delay between pages
            if page < max_pages:
                time.sleep(2)

        return total_added

    def scrape_categories(self, categories: list[str], max_pages: int = 2) -> int:
        """
        Scrape listings from multiple categories.

        Args:
            categories: List of category paths
            max_pages: Maximum number of pages per category

        Returns:
            Total number of listings in collection.
        """
        for i, category in enumerate(categories):
            self.scrape_category(category, max_pages)
            self._log("")  # Blank line between categories

            # Small delay between categories
            if i < len(categories) - 1:
                time.sleep(2)

        return len(self.collection)

    def get_listings(self) -> list[Listing]:
        """Get all scraped listings."""
        return self.collection.listings

    def export_csv(self, filepath: str) -> str:
        """Export listings to CSV file."""
        return self.collection.to_csv(filepath)

    def export_json(self, filepath: str) -> str:
        """Export listings to JSON file."""
        return self.collection.to_json(filepath)