# 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`: ```json // Автомобиль { "title": "Продам Toyota Camry 2020", "price": 1500000, ... } // Телефон { "title": "iPhone 15 Pro Max 256GB", ... } // Животное { "title": "Котёнок британский даром", ... } // Вакансия { "title": "Требуется водитель категории B", ... } ``` Примеры, которые должны пройти (`status = completed`): ```json // Квартира (даже если неструктурированно) { "title": "Продам 2-к квартиру 55м² на 5 этаже" } // Дом { "title": "Дом 120м² на участке 6 соток" } // Гараж { "title": "Продам гараж 18м² в ГСК Маяк" } // Земля { "title": "Участок 10 соток ИЖС в с. Вешки" } ``` ## Поиск (Search) Два complementary endpoint'а для поиска объявлений: ### Semantic Search — `POST /api/v1/search/similar` - **Технология:** pgvector (768d embeddings) + cosine similarity (HNSW index) - **Поток:** Query → Ollama `nomic-embed-text:latest` → `embedding <=> query_vector` в PostgreSQL - **Преимущества:** Находит "близкие по смыслу" объявления даже при других словах - **Недостатки:** Требует Ollama call per query (~200-500ms) ### Full-Text Search — `POST /api/v1/search/fulltext` - **Технология:** PostgreSQL tsvector + tsquery + GIN index - **Конфигурация:** `ukrainian` (копия `simple` + `ukrainian_stop` словарь) - **Стоп-слова:** ~120 украинских слов из `scripts/ukrainian.stop` - **Преимущества:** Быстрый (< 50ms), точный keyword match - **Недостатки:** Нет стемминга (`квартира` ≠ `квартири`). См. [[REVIEW_FOLLOWUP.md]] пункт 15. ### Hybrid Search (идея для будущего) ``` 1. FTS pre-filter: tsquery находит кандидатов (top-100) 2. Vector re-rank: cosine similarity среди кандидатов (top-10) 3. Комбинированный score: α * rank_score + β * similarity_score ``` ## Масштабируемость Текущая архитектура рассчитана на **среднюю нагрузку** (синхронная обработка в HTTP-запросе). Для масштабирования: 1. Вынести `PropertyPipeline.process()` в фоновую задачу (Celery / RQ / arq). 2. API возвращает `202 Accepted` с `job_id` и сразу отдаёт ответ. 3. Фоновый воркер обрабатывает: normalize → download images → AI analyze → save. 4. Добавить webhook callback по завершению. ## Технологические решения | Решение | Обоснование | |---------|-------------| | **Async SQLAlchemy + asyncpg** | FastAPI async. Лучшее использование I/O-bound операций (БД + HTTP к Ollama). | | **JSONB для payload / snapshot** | Парсеры присылают разные структуры. Гибкость без ALTER TABLE. | | **Structured LLM output** | Ollama поддерживает JSON mode. Надёжнее парсинга свободного текста. | | **Repository pattern** | Легко мокировать в тестах. Замена БД без изменения бизнес-логики. | | **SHA-256 дедупликация изображений** | Одна и та же картинка может быть по разным URL. | | **Custom fields таблица** | Парсеры постоянно добавляют новые атрибуты. Не нужно менять схему БД. | | **Snapshots** | Требование бизнеса: история изменений объявлений. | | **Ollama (локальный LLM)** | Независимость от внешних API. Конфиденциальность данных. Контроль затрат. | | **Pillow для изображений** | Лёгкая зависимость. Извлечение метаданных без тяжёлых библиотек. |