VMK 360 Data Collector

Описание

Сервис приёма, нормализации и ИИ-обогащения данных об объектах недвижимости.

• Приём: Принимает полусырые данные от парсеров через REST API
• Валидация: AI определяет, является ли payload объявлением о недвижимости
• Нормализация: Приводит неструктурированные данные к единому формату
• Обогащение: Анализ изображений (vision) + текстовый анализ (NER, summary, оценка цены)
• Хранение: Сохраняет в PostgreSQL с полной историей изменений (snapshots)

Быстрый старт (Docker)

# 1. Клонировать и перейти в директорию
cd vmk_data_collector

# 2. Скопировать конфиг
#   Отредактируй .env под себя (OLLAMA_BASE_URL, порты и т.д.)
cp .env.example .env

# 3. Запуск всего стека (PostgreSQL + FastAPI)
docker compose up -d --build

# 4. Проверка
#   Документация API: http://localhost:8020/docs
#   Health check:      http://localhost:8020/api/v1/health

Стоп/очистка

docker compose down            # остановить
docker compose down -v        # остановить + удалить данные БД

Ручной старт (для разработки)

# 1. PostgreSQL должен быть доступен (локально или в Docker)
#    Порт по умолчанию: 5432 (хост) / 5433 (если через docker-compose)

# 2. Установка зависимостей
pip install -e ".[dev]"

# 3. Применение миграций
alembic upgrade head

# 4. Запуск приложения
uvicorn vmk_data_collector.main:app --reload --port 8020

API для парсеров

Подробный гайд для разработчиков парсеров с примерами кода, retry-логикой и FAQ:
📄 docs/PARSER_INTEGRATION.md

POST /api/v1/ingest

Принимает сырые данные от парсеров, валидирует payload и ставит задачу в очередь на обработку.

Заголовки

Тело запроса

{
  "source_slug": "avito",
  "external_id": "avito-12345678",
  "payload": {
    "title": "2-комнатная квартира, 65 м², 5/25 этаж",
    "description": "Продается просторная двухкомнатная квартира в новостройке. Рядом метро.",
    "price": 8500000,
    "url": "https://avito.ru/item/12345678",
    "images": [
      "https://avito.ru/img1.jpg",
      "https://avito.ru/img2.jpg"
    ],
    "contact_phone": "+7 (999) 123-45-67",
    "address": "Москва, Тверская ул., 1",
    "area": 65.5,
    "rooms": 2,
    "floor": 5
  }
}

Поля payload

Важно: хотя бы одно из полей title или description должно присутствовать.

Ответ 202 Accepted (успех)

{
  "job_id": 42,
  "property_id": null,
  "status": "pending",
  "reason": null,
  "message": "Queued for processing",
  "snapshot_id": null
}

Ответ 422 Unprocessable Entity (ошибка валидации)

{
  "detail": "Invalid payload: 1 validation error for PayloadSchema..."
}

POST /api/v1/ingest/with-images

Принимает бинарные изображения от парсера напрямую. Парсер сам выкачивает фото со своего источника и передаёт их нам — сервер не пытается обходить чужие CDN.

Когда использовать

Заголовки

Тело запроса (multipart)

metadata: {"source_slug": "domria", "external_id": "34329468", "payload": {"title": "...", "description": "...", "price": 125000}}
images:   [binary-photo-1.jpg]
images:   [binary-photo-2.jpg]
images:   [binary-photo-3.jpg]

Поле metadata — это JSON-строка с тем же форматом, что и в /ingest. Поле images повторяется для каждого файла (FastAPI принимает list[UploadFile]).

Лимиты

• Макс. файлов за запрос: 50
• Макс. размер одного файла: 10 МБ

Ответ 202 Accepted (успех)

Тот же формат, что и у /ingest, но pipeline отрабатывает синхронно в рамках запроса — потому что изображения уже локальные и не нужно ждать очереди.

{
  "job_id": 42,
  "property_id": 7,
  "status": "completed",
  "reason": null,
  "message": "Property ingested successfully",
  "snapshot_id": 3
}

Примеры (curl)

# Отправить объявление с 3 фото
curl -X POST http://localhost:8020/api/v1/ingest/with-images \
  -F 'metadata={"source_slug": "domria", "external_id": "34329468", "payload": {"title": "2-комнатная квартира", "description": "Продается квартира", "price": 125000}}' \
  -F 'images=@/path/to/photo1.jpg' \
  -F 'images=@/path/to/photo2.jpg' \
  -F 'images=@/path/to/photo3.jpg'

Примеры (Python requests)

import requests

metadata = {
    "source_slug": "domria",
    "external_id": "34329468",
    "payload": {
        "title": "2-комнатная квартира",
        "description": "Продается квартира",
        "price": 125000,
    },
}

files = [
    ("images", open("photo1.jpg", "rb")),
    ("images", open("photo2.jpg", "rb")),
    ("images", open("photo3.jpg", "rb")),
]

response = requests.post(
    "http://localhost:8020/api/v1/ingest/with-images",
    data={"metadata": json.dumps(metadata)},
    files=files,
)
print(response.json())

Статусы обработки

После приёма задача (job_id) проходит через pipeline:

Примеры (curl)

Успешный ingest

curl -X POST http://localhost:8020/api/v1/ingest \
  -H "Content-Type: application/json" \
  -d '{
    "source_slug": "avito",
    "external_id": "avito-99999",
    "payload": {
      "title": "3-комнатная квартира, 120 м²",
      "description": "Элитная квартира в центре",
      "price": 25000000,
      "url": "https://avito.ru/item/99999",
      "address": "Москва, Сити"
    }
  }'

Проверка архивации (404/410 = объявление снято)

curl -X POST http://localhost:8020/api/v1/listings/1/archive-check

Конфигурация

Все настройки задаются через переменные окружения (.env или docker-compose.yml):

Архитектура

┌──────────────┐     ┌──────────────┐     ┌──────────────────┐
│   Парсеры    │────▶│  FastAPI     │────▶│  PostgreSQL      │
│  (curl/HTTP) │     │  /api/v1/    │     │  (raw_data +     │
└──────────────┘     │  ingest      │     │   property_...)   │
                     └──────┬───────┘     └──────────────────┘
                            │
                            ▼
                     ┌──────────────┐
                     │  QueueWorker │
                     │  (async bg)  │
                     └──────┬───────┘
                            │
              ┌─────────────┼─────────────┐
              ▼             ▼             ▼
        ┌─────────┐   ┌─────────┐   ┌─────────┐
        │AI Норма-│   │AI Анализ│   │AI Обога-│
        │лизатор  │   │изображ. │   │щение    │
        └────┬────┘   └────┬────┘   └────┬────┘
             │             │             │
             └─────────────┴─────────────┘
                           │
                           ▼
                     ┌──────────────┐
                     │   Ollama     │
                     │ 192.168.1.75 │
                     └──────────────┘

• API Layer — FastAPI, Pydantic валидация, structured logging (structlog)
• Service Layer — PropertyPipeline, AI-нормализация, обогащение, QueueWorker
• Repository Layer — абстракция доступа к PostgreSQL (async SQLAlchemy)
• Domain Layer — чистые сущности недвижимости
• Infrastructure Layer — Ollama client, image downloader, rate limiter

Документация

• Техническое задание — полное описание модели данных, API, AI-слоя
• Архитектура — диаграммы, потоки данных, компоненты
• OpenAPI (Swagger): http://localhost:8020/docs
• ReDoc: http://localhost:8020/redoc

Логирование

Все логи выводятся в stdout в формате JSON (structlog), что удобно для сбора через docker compose logs:

# Следить за логами в реальном времени
docker compose logs -f app

# Посмотреть последние 50 строк
docker compose logs --tail=50 app

Ключевые события для мониторинга:

• ingest_request — новый запрос от парсера
• ingest_accepted — задача поставлена в очередь
• ingest_validation_failed — ошибка валидации payload
• pipeline_start — начало обработки задачи
• pipeline_completed — задача успешно завершена
• pipeline_not_real_estate — AI отбросил объект
• ollama_chat_request/response — запросы к Ollama