models.py

"""Data models for Marktplaats scraper."""

import csv
import json
from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone
from typing import Optional


@dataclass
class Listing:
    """Represents a single Marktplaats listing."""

    title: str
    price: str
    description: str
    location: str
    listing_url: str
    category: str
    condition: str = ""
    shipping: str = ""
    seller_name: str = ""
    date_posted: str = ""
    scraped_at: str = field(
        default_factory=lambda: datetime.now(timezone.utc).isoformat()
    )

    def to_dict(self) -> dict:
        """Convert listing to dictionary."""
        return asdict(self)

    def __hash__(self):
        """Hash based on unique listing URL."""
        return hash(self.listing_url)

    def __eq__(self, other):
        """Two listings are equal if they have the same URL."""
        if not isinstance(other, Listing):
            return False
        return self.listing_url == other.listing_url


class ListingCollection:
    """Collection of listings with deduplication and export functionality."""

    def __init__(self):
        self._listings: dict[str, Listing] = {}

    def add(self, listing: Listing) -> bool:
        """
        Add a listing to the collection.

        Returns True if listing was added (new), False if duplicate.
        """
        if listing.listing_url in self._listings:
            return False
        self._listings[listing.listing_url] = listing
        return True

    def add_many(self, listings: list[Listing]) -> int:
        """
        Add multiple listings to the collection.

        Returns the number of new listings added.
        """
        added = 0
        for listing in listings:
            if self.add(listing):
                added += 1
        return added

    @property
    def listings(self) -> list[Listing]:
        """Get all listings as a list."""
        return list(self._listings.values())

    def __len__(self) -> int:
        return len(self._listings)

    def __iter__(self):
        return iter(self._listings.values())

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

        Args:
            filepath: Path to the output CSV file.

        Returns:
            The filepath where data was saved.
        """
        if not self._listings:
            return filepath

        fieldnames = [
            "title",
            "price",
            "description",
            "location",
            "listing_url",
            "category",
            "condition",
            "shipping",
            "seller_name",
            "date_posted",
            "scraped_at",
        ]

        with open(filepath, "w", newline="", encoding="utf-8") as f:
            writer = csv.DictWriter(f, fieldnames=fieldnames)
            writer.writeheader()
            for listing in self._listings.values():
                writer.writerow(listing.to_dict())

        return filepath

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

        Args:
            filepath: Path to the output JSON file.

        Returns:
            The filepath where data was saved.
        """
        data = {
            "scraped_at": datetime.now(timezone.utc).isoformat(),
            "total_listings": len(self._listings),
            "listings": [listing.to_dict() for listing in self._listings.values()],
        }

        with open(filepath, "w", encoding="utf-8") as f:
            json.dump(data, f, indent=2, ensure_ascii=False)

        return filepath