VMK 360 Data Collector — Архитектура

Общая архитектура

Сервис построен по принципу слоистой архитектуры (Layered Architecture) с разделением на слои:

1. API Layer — FastAPI роутеры, Pydantic схемы, middleware
2. Service Layer — бизнес-логика, пайплайны
3. Repository Layer — абстракция доступа к БД
4. Domain Layer — чистые сущности и правила
5. Infrastructure Layer — HTTP клиенты, БД, хранилище файлов, логирование

┌─────────────────────────────────────────────────────────────┐
│                        API Layer                             │
│  POST /api/v1/properties/ingest                              │
│  Pydantic validation → FastAPI router                        │
└─────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────┐
│                     Service Layer                            │
│  PropertyPipeline                                            │
│  ├── AiNormalizer (Ollama)                                  │
│  ├── PropertyNormalizer                                     │
│  ├── ImageDownloader                                        │
│  ├── AiImageAnalyzer (Ollama Vision)                        │
│  └── AiEnricher (Ollama)                                    │
└─────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────┐
│                   Repository Layer                           │
│  RawDataRepository          PropertyRepository              │
│  ImageRepository            CustomFieldRepository           │
│  SnapshotRepository         AiEnrichmentRepository          │
└─────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────┐
│                   Infrastructure Layer                       │
│  PostgreSQL (asyncpg)        Ollama API (httpx)             │
│  Local filesystem (images)   structlog                     │
└─────────────────────────────────────────────────────────────┘

Поток данных

Приём нового объявления

Parser ──► POST /api/v1/properties/ingest
              │
              ▼
         RawDataIngestSchema (Pydantic validate)
              │
              ▼
         [RawDataRepository] save(status=pending)
              │
              ▼
         PropertyPipeline.process(raw_data_id)
              │
              ├─► [AiNormalizer] LLM validates & structures
              │   ├──► reject → raw_data.status = invalid
              │   └──► accept → NormalizedPropertySchema
              │
              ├─► [PropertyRepository] upsert
              │   ├──► new → INSERT property_listings
              │   └──► existing → INSERT snapshot + UPDATE listing
              │
              ├─► [ImageDownloader] async download
              │   └──► local filesystem + property_images records
              │
              ├─► [AiImageAnalyzer] vision analysis per image
              │   └──► ai_description + analysis_status
              │
              ├─► [AiEnricher] text + image_analysis
              │   └──► ai_enrichments record
              │
              └─► [RawDataRepository] status = completed
              │
              ▼
         Response 202 Accepted

Обновление существующего объявления

Same external_id from same source
              │
              ▼
         Load existing property_listings
              │
              ▼
         [SnapshotRepository] save current state as JSONB
              │
              ▼
         [PropertyRepository] update fields
              │
              ▼
         [CustomFieldRepository] delete old + insert new
              │
              ▼
         [ImageDownloader] skip existing (hash match), download new
              │
              ▼
         Re-run AiImageAnalyzer + AiEnricher
              │
              ▼
         Response 202 { property_id, snapshot_id }

Модель данных (ER диаграмма)

┌──────────────────┐       ┌─────────────────────┐
│  data_sources    │       │    property_types   │
├──────────────────┤       ├─────────────────────┤
│ id (PK)          │       │ id (PK)             │
│ slug (UQ)        │       │ slug (UQ)           │
│ name             │       │ name                │
│ url_pattern      │       │ description         │
│ description      │       └─────────────────────┘
│ created_at       │
└──────────────────┘
         │
         │ 1:N
         ▼
┌──────────────────────────────────────────────┐
│           raw_parsing_data                     │
├──────────────────────────────────────────────┤
│ id (PK)                                      │
│ source_id (FK) ──────────────────────────────┐
│ external_id                                  │
│ payload (JSONB)                              │
│ status: pending/processing/completed/failed/invalid
│ validation_result                            │
│ error_message                                │
│ received_at                                  │
│ processed_at                                 │
└──────────────────────────────────────────────┘
         │
         │ 1:1
         ▼
┌──────────────────────────────────────────────┐
│          property_listings                     │
├──────────────────────────────────────────────┤
│ id (PK)                                      │
│ raw_data_id (FK, UQ) ◄───────────────────────┘
│ source_id (FK)                               │
│ external_id                                  │
│ title, description, generated_description    │
│ deal_type, property_type_id (FK)             │
│ price, currency, original_price, original_currency
│ price_per_sqm                                │
│ total_area, living_area, kitchen_area, land_area
│ rooms_count, bedrooms_count, bathrooms_count   │
│ layout, floor, floors_total                  │
│ building_year, building_type                 │
│ renovation_status, ceiling_height, material   │
│ has_balcony, has_loggia, balcony_count       │
│ loggia_count, bathroom_type                  │
│ elevator_count, has_freight_elevator           │
│ parking_type, heating_type                   │
│ internet, security                           │
│ windows_direction, window_view               │
│ address_raw, city, district, micro_district  │
│ street, house_number                         │
│ metro_station, metro_distance_min            │
│ metro_distance_type                          │
│ latitude, longitude                          │
│ contact_phone, contact_name, contact_email    │
│ is_agent, agency_name                        │
│ publish_date, url_source                     │
│ listing_status                               │
│ images_count                                 │
│ listing_quality_score                        │
│ reliability_rating                           │
│ sentiment_score                              │
│ created_at, updated_at                       │
└──────────────────────────────────────────────┘
         │
    ┌────┼────┐
    │    │    │
    ▼    ▼    ▼
┌────────┐ ┌──────────────┐ ┌─────────────────┐
│property│ │property_custom│ │  ai_enrichments │
│_images │ │    _fields     │ │                 │
├────────┤ ├──────────────┤ ├─────────────────┤
│id (PK) │ │id (PK)        │ │id (PK)          │
│property│ │property_id(FK) │ │property_id(FK,UQ)│
│_id(FK) │ │field_name     │ │extracted_features│
│url     │ │field_value    │ │price_assessment │
│local_  │ │field_type     │ │listing_quality_score│
│path    │ │               │ │reliability_rating│
│hash    │ │UQ(property_id,│ │sentiment_score  │
│file_size│ │  field_name)  │ │classification   │
│width   │ │               │ │image_analysis_  │
│height  │ │               │ │  results         │
│download│ │               │ │generated_description│
│_status │ │               │ │summary          │
│ai_     │ │               │ │model_version    │
│description│              │ │processing_time_ms│
│analysis│ │               │ │created_at       │
│_status │ │               │ └─────────────────┘
│order_  │ │               │
│index   │ │               │
└────────┘ └──────────────┘
    │
    ▼
┌──────────────────────┐
│  property_snapshots  │
├──────────────────────┤
│ id (PK)              │
│ property_id (FK)    │
│ snapshot_data (JSONB)│
│ changed_fields (JSONB)│
│ created_at           │
└──────────────────────┘

Компоненты

AiNormalizer

• Роль: Первый фильтр. LLM решает: недвижимость это или нет.
• Вход: raw_parsing_data.payload (title, description, params).
• Выход: NormalizedPropertySchema или ValidationError.
• Особенности: Работает с непредсказуемыми данными. Извлекает параметры даже из каши текста.

PropertyNormalizer

• Роль: Маппинг результата LLM в property_listings + property_custom_fields.
• Особенности: Все поля, которых нет в core-схеме, идут в custom_fields.

ImageDownloader

• Роль: Асинхронное скачивание изображений.
• Процесс:
  1. httpx async GET по URL.
  2. Сохранение в IMAGE_STORAGE_PATH / {property_id} / {hash}.{ext}.
  3. SHA-256 hash для дедупликации.
  4. Pillow для width/height.
  5. Обновление property_images записей.

AiImageAnalyzer

• Роль: Анализ каждого изображения через Ollama Vision.
• Вход: base64 изображение.
• Выход: ImageAnalysisSchema (состояние, комнаты, проблемы, плюсы).
• Модель: llava или llama3.2-vision.

AiEnricher

• Роль: Финальное обогащение текста + агрегированный image_analysis.
• Вход: Нормализованный текст + image_analysis_results.
• Выход: AiEnrichmentSchema.
• Модель: llama3.2 или аналог.

PropertyPipeline

• Роль: Оркестратор всего пайплайна.
• Поведение: Все шаги выполняются последовательно в рамках одного HTTP-запроса (средняя нагрузка). При росте — можно вынести в Celery/RQ без изменения интерфейсов.

Хранение изображений

IMAGE_STORAGE_PATH/
└── properties/
    └── {property_id}/
        ├── {hash1}.jpg
        ├── {hash2}.png
        └── {hash3}.jpg

• Дедупликация по SHA-256 hash на уровне БД.
• При upsert: новые картинки добавляются, существующие по hash пропускаются.

AI-валидация: reject cases

Примеры payload, которые должны получить status = invalid:

// Автомобиль
{ "title": "Продам Toyota Camry 2020", "price": 1500000, ... }

// Телефон
{ "title": "iPhone 15 Pro Max 256GB", ... }

// Животное
{ "title": "Котёнок британский даром", ... }

// Вакансия
{ "title": "Требуется водитель категории B", ... }

Примеры, которые должны пройти (status = completed):

// Квартира (даже если неструктурированно)
{ "title": "Продам 2-к квартиру 55м² на 5 этаже" }

// Дом
{ "title": "Дом 120м² на участке 6 соток" }

// Гараж
{ "title": "Продам гараж 18м² в ГСК Маяк" }

// Земля
{ "title": "Участок 10 соток ИЖС в с. Вешки" }

Масштабируемость

Текущая архитектура рассчитана на среднюю нагрузку (синхронная обработка в HTTP-запросе).

Для масштабирования:

1. Вынести PropertyPipeline.process() в фоновую задачу (Celery / RQ / arq).
2. API возвращает 202 Accepted с job_id и сразу отдаёт ответ.
3. Фоновый воркер обрабатывает: normalize → download images → AI analyze → save.
4. Добавить webhook callback по завершению.

Технологические решения