Перейти до змісту

🌍 Мова: English version

Створення агента - мультисайтового парсера оголошень квартир: Патерни проєктування та архітектура

Written by:

Igor Gorovyy
DevOps Engineer Lead & Senior Solutions Architect

LinkedIn

The Mechanical Apartment Hunter Mk. III

Як я побудував sophisticated веб-скрапер використовуючи патерни проєктування Python для автоматизації пошуку квартир у Варшаві

The Mechanical Apartment Hunter Mk. III

Проблема

Знайти відповідну квартиру у Варшаві - це як шукати голку в стозі сіна. З тисячами оголошень, розкиданих по різних платформах нерухомості (OLX.pl, Otodom.pl), щоденна ручна перевірка кожного сайту стає роботою на повний робочий день.

Мені потрібне було автоматизоване рішення, яке могло б:
- Моніторити декілька сайтів нерухомості одночасно
- Застосовувати складні критерії фільтрації (район, ціна, кількість кімнат, меблі, домашні тварини)
- Надсилати сповіщення в реальному часі через Slack
- Запобігати дублюванню повідомлень
- Працювати з різними архітектурами сайтів та API

Рішення: Архітектура мульти-парсера

Замість створення монолітного скрапера, я спроєктував гнучку, розширювану систему з використанням декількох патернів проєктування, що зробило код підтримуваним, тестованим та легко розширюваним.

Використані патерни проєктування

1. Патерн Template Method (Шаблонний Метод) - Клас BaseParser

Фундаментом нашої архітектури є патерн Template Method через абстрактний базовий клас:

from abc import ABC, abstractmethod

class BaseParser(ABC):
    def __init__(self, source_name: str):
        self.source_name = source_name
        self._init_database()

    def process_new_listings(self) -> None:
        """Шаблонний метод, що визначає структуру алгоритму"""
        listings = self.fetch_listings()  # Абстрактний метод
        new_listings = self._filter_new_listings(listings)

        for listing in new_listings:
            if not listing.get('image_url'):
                listing['image_url'] = self._fetch_photo_from_detail_page(listing['url'])

            self.send_to_slack(listing)
            self._save_listing(listing)

    @abstractmethod
    def fetch_listings(self) -> List[Dict]:
        """Кожен парсер повинен реалізувати свою логіку отримання даних"""
        pass

Переваги:
- Узгодженість: Всі парсери слідують одному робочому процесу
- Повторне використання коду: Спільна функціональність (база даних, Slack, rate limiting) використовується разом
- Розширюваність: Легко додати нові парсери, реалізувавши один метод

2. Патерн Strategy (Стратегія) - Платформо-специфічний парсинг

Кожна платформа нерухомості вимагає різних стратегій парсингу:

class OLXParser(BaseParser):
    def fetch_listings(self) -> List[Dict]:
        """OLX-специфічна стратегія парсингу"""
        # HTML парсинг з BeautifulSoup
        # URL-базова фільтрація
        # Клієнтська фільтрація району
        pass

class OtodomParser(BaseParser):
    def fetch_listings(self) -> List[Dict]:
        """Otodom-специфічна стратегія парсингу"""
        # Витягування JSON з __NEXT_DATA__
        # Підтримка багатосторінковості
        # Клієнтська фільтрація приватних оголошень
        pass

Переваги:
- Платформна незалежність: Кожен парсер обробляє особливості своєї платформи
- Легке тестування: Можливість мокати різні стратегії для юніт-тестів
- Підтримуваність: Зміни в одній платформі не впливають на інші

3. Патерн Factory (Фабрика) - Створення парсерів

Оркестратор використовує патерн Factory для створення відповідних парсерів:

class ParserFactory:
    @staticmethod
    def create_parser(source: str) -> BaseParser:
        parsers = {
            'OLX': OLXParser,
            'Otodom': OtodomParser
        }

        if source not in parsers:
            raise ValueError(f"Невідомий парсер: {source}")

        return parsers[source]()

4. Патерн Observer (Спостерігач) - Slack Сповіщення

Система діє як суб'єкт, сповіщаючи спостерігачів (Slack канали) про нові оголошення:

class SlackNotifier:
    def __init__(self, webhook_url: str):
        self.webhook_url = webhook_url

    def notify(self, listing: Dict) -> None:
        """Відправка сповіщення в Slack"""
        message = self._format_message(listing)
        self._send_to_slack(message)

    def _format_message(self, listing: Dict) -> Dict:
        """Форматування даних оголошення в Slack Block Kit"""
        return {
            "blocks": [
                {
                    "type": "header",
                    "text": {"type": "plain_text", "text": f"🏠 Нове Оголошення [{listing['source']}]"}
                },
                # ... більше блоків
            ]
        }

5. Патерн Singleton (Одинак) - Підключення до бази даних

З'єднання з базою даних SQLite керується як singleton для забезпечення узгодженості:

import sqlite3
from functools import lru_cache

class DatabaseManager:
    _instance = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    @lru_cache(maxsize=1)
    def get_connection(self) -> sqlite3.Connection:
        """Кешоване підключення до бази даних"""
        return sqlite3.connect('listings.db')

6. Патерн Decorator (Декоратор) - Rate Limiting

Rate limiting реалізований як декоратор для уникнення перевантаження цільових веб-сайтів:

import time
from functools import wraps

def rate_limit(seconds: int = 60):
    """Декоратор для обмеження частоти запитів"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            time.sleep(seconds)
            return func(*args, **kwargs)
        return wrapper
    return decorator

class BaseParser:
    @rate_limit(60)  # 1 хвилина між запитами
    def fetch_listings(self) -> List[Dict]:
        # Реалізація отримання даних
        pass

Архітектура системи

graph TB
    subgraph "Клієнтський Рівень"
        A[multi_parser.py<br/>Оркестратор]
    end

    subgraph "Рівень Парсерів"
        B[OLXParser]
        C[OtodomParser]
        D[BaseParser<br/>Абстрактний Клас]
    end

    subgraph "Рівень Даних"
        E[SQLite База Даних]
        F[Slack API]
    end

    subgraph "Зовнішні API"
        G[OLX.pl]
        H[Otodom.pl]
    end

    A --> B
    A --> C
    B --> D
    C --> D
    D --> E
    D --> F
    B --> G
    C --> H

    style A fill:#e1f5fe
    style D fill:#fff3e0
    style E fill:#e8f5e8
    style F fill:#fce4ec

Архітектура потоку даних

sequenceDiagram
    participant MP as MultiParser
    participant BP as BaseParser
    participant OLX as OLXParser
    participant OTD as OtodomParser
    participant DB as SQLite БД
    participant SL as Slack API
    participant WS as Веб-сайти

    MP->>BP: Ініціалізація парсерів
    BP->>DB: Створення таблиць якщо потрібно

    loop Кожну хвилину (режим Loop)
        MP->>OLX: fetch_listings()
        OLX->>WS: HTTP Запит (OLX.pl)
        WS-->>OLX: HTML Відповідь
        OLX->>OLX: Парсинг HTML/Витягування даних
        OLX->>OLX: Клієнтська фільтрація
        OLX-->>BP: Дані оголошень

        MP->>OTD: fetch_listings()
        OTD->>WS: HTTP Запит (Otodom.pl)
        WS-->>OTD: JSON Відповідь
        OTD->>OTD: Парсинг JSON/Витягування даних
        OTD->>OTD: Обробка багатьох сторінок
        OTD->>OTD: Клієнтська фільтрація
        OTD-->>BP: Дані оголошень

        BP->>DB: Перевірка унікальності
        DB-->>BP: Тільки нові оголошення

        loop Для кожного нового оголошення
            BP->>WS: Отримати фото зі сторінки деталей
            WS-->>BP: URL фото
            BP->>SL: Відправити Slack сповіщення
            BP->>DB: Зберегти оголошення
        end
    end

Патерн Стратегії фільтрації

Різні платформи вимагають різних підходів до фільтрації:

graph TD
    A[Дані Оголошення] --> B{Платформа?}

    B -->|OLX| C[Фільтрація через URL]
    B -->|Otodom| D[Фільтрація через JSON]

    C --> E[Клієнтська Перевірка Району]
    D --> F[Клієнтська Перевірка Приватних]

    E --> G[Фінальні Відфільтровані Оголошення]
    F --> G

    style C fill:#e3f2fd
    style D fill:#f3e5f5
    style G fill:#e8f5e8

Патерн Управління конфігурацією

Конфігурація на основі змінних середовища з використанням Configuration Object Pattern:

from dataclasses import dataclass
from typing import List

@dataclass
class SearchConfig:
    """Об'єкт конфігурації для параметрів пошуку"""
    district_name: str
    rooms: str
    price_from: int
    price_to: int
    furniture: str
    pets: str
    listing_type: str

    @classmethod
    def from_env(cls) -> 'SearchConfig':
        """Фабричний метод для створення конфігурації з оточення"""
        return cls(
            district_name=os.getenv('DISTRICT_NAME', 'wola'),
            rooms=os.getenv('ROOMS', 'three'),
            price_from=int(os.getenv('PRICE_FROM', '4000')),
            price_to=int(os.getenv('PRICE_TO', '8000')),
            furniture=os.getenv('FURNITURE', 'yes'),
            pets=os.getenv('PETS', 'Tak'),
            listing_type=os.getenv('LISTING_TYPE', 'private')
        )

Стратегія обробки помилок

Надійна обробка помилок з використанням Chain of Responsibility Pattern:

class ErrorHandler:
    def __init__(self):
        self.handlers = [
            NetworkErrorHandler(),
            ParsingErrorHandler(),
            DatabaseErrorHandler(),
            SlackErrorHandler()
        ]

    def handle_error(self, error: Exception, context: Dict) -> None:
        for handler in self.handlers:
            if handler.can_handle(error):
                handler.handle(error, context)
                break
        else:
            # Логування необробленої помилки
            logger.error(f"Необроблена помилка: {error}")

class NetworkErrorHandler:
    def can_handle(self, error: Exception) -> bool:
        return isinstance(error, (requests.RequestException, TimeoutError))

    def handle(self, error: Exception, context: Dict) -> None:
        logger.warning(f"Мережева помилка, повторна спроба: {error}")
        time.sleep(5)  # Стратегія відступу

Оптимізації продуктивності

1. Патерн Lazy Loading

Зображення завантажуються тільки коли потрібно:

class LazyImageLoader:
    def __init__(self, listing: Dict):
        self.listing = listing
        self._image_url = None

    @property
    def image_url(self) -> str:
        if self._image_url is None:
            self._image_url = self._fetch_image()
        return self._image_url

2. **Патерн Caching **

Запити до бази даних кешуються за допомогою functools.lru_cache:

from functools import lru_cache

class DatabaseManager:
    @lru_cache(maxsize=1000)
    def is_listing_seen(self, listing_id: str) -> bool:
        """Кешування пошуку в базі даних для продуктивності"""
        cursor = self.get_connection().cursor()
        cursor.execute("SELECT 1 FROM seen_listings WHERE id = ?", (listing_id,))
        return cursor.fetchone() is not None

Стратегія тестування

Архітектура дозволяє комплексне тестування через Dependency Injection:

class TestableParser(BaseParser):
    def __init__(self, http_client=None, database=None, slack_client=None):
        self.http_client = http_client or requests
        self.database = database or SQLiteManager()
        self.slack_client = slack_client or SlackNotifier()

# В тестах
def test_parser_with_mocks():
    mock_client = MockHTTPClient()
    mock_db = MockDatabase()
    parser = TestableParser(mock_client, mock_db)
    # Тестування з контрольованими залежностями

Результати та метрики

Система успішно:

  • Моніторить 2 платформи одночасно (OLX.pl, Otodom.pl)
  • Обробляє 100+ оголошень за запуск через декілька сторінок
  • Досягає 99.9% uptime з надійною обробкою помилок
  • Надсилає сповіщення в реальному часі з фотографіями та детальною інформацією
  • Запобігає дублюванню з 100% точністю, використовуючи унікальність бази даних
  • Обробляє rate limiting для поваги до цільових веб-сайтів

Ключові висновки

  1. Патерни проєктування - важливі: Використання встановлених патернів зробило код більш підтримуваним та розширюваним
  2. Розділення відповідальностей: Кожен клас має одну відповідальність
  3. Платформна абстракція: Базовий клас обробляє спільну функціональність, дозволяючи платформо-специфічні реалізації
  4. Управління конфігурацією: Конфігурація на основі змінних середовища робить розгортання гнучким
  5. Стійкість до помилок: Комплексна обробка помилок забезпечує безперервну роботу системи
  6. Оптимізація продуктивності: Кешування та ледаче завантаження покращують час відгуку

🔮 Майбутні покращення

Модульна архітектура дозволяє легко додати:

  • Нові платформи: Реалізувати BaseParser для додаткових сайтів нерухомості
  • AI інтеграція: Додати ML моделі для оцінки якості оголошень
  • Покращена фільтрація: Реалізувати нечітке співставлення для районів
  • Аналітична панель: Додати веб-інтерфейс для моніторингу
  • Підтримка багатьох міст: Розширити за межі Варшави

Репозиторій коду

🔗 GitHub Репозиторій: parser-warsaw-appartment

Повна реалізація доступна з:

  • ✅ Комплексною документацією
  • ✅ Діаграмами архітектури
  • ✅ Стратегіями обробки помилок
  • ✅ Оптимізаціями продуктивності

Створення цього парсера навчило мене, що хороша архітектура програмного забезпечення - це не лише вирішення поточної проблеми, а створення фундаменту, який може розвиватися та масштабуватися зі змінними вимогами. Патерни проєктування, використані тут, забезпечують цю гнучкість, зберігаючи при цьому чіткість та надійність коду.