2018-lpon-site/_prj_blueprint/parser_architecture.md

Архитектура парсера и валидации данных

Дата: 2026-07-15
Последний обновленно: 2026-07-15
Статус: В разработке ✏️

Обзор проблемы

Система должна поддерживать импорт данных из различных источников:

• Excel/CSV файлы от продавцов и издателей
• Парсинг URL-страниц с каталогами
• Ручной ввод данных в админке
• Будущие API интеграции

Основной вызов: Валидация и обнаружение конфликтов (дубликаты, совпадения в синонимах) требует пользовательского участия для принятия решений. Нельзя просто блокировать или автоматически удалять/объединять данные.

Текущее состояние (MVP)

Модели и структура

┌─ TbSource (Excel, CSV, URL) ──┐
│  s_source_name                │
│  l_source_type (excel/csv/url)│
│  k_source_to_seller ──────────┼──► TbSeller
│  source_file (FilerFileField) │
│  s_source_url                 │
│  j_source_metadata (struktura)│    (metadata описывает столбцы, вкладки и т.д.)
└─────────────────────────────────┘

TbOffer ◄──┐
           │
TbSource ──┴─► TbItem ──────► TbArticle (текст, SEO, slug, картинка)
           │
           └──► TbLabel, TbArtist, TbMusicStyle
                (все через TbArticle)

Валидация (текущая реализация)

Файл: lpon_site/frontend/utils_validators.py

Функция: validate_for_duplicates()

• Проверяет основное поле (s_label, s_artist и т.д.)
• Проверяет синонимы в j_*_metadata[SYN_EN]
• Возвращает список найденных дубликатов с типом совпадения

Типы совпадений:

1. EXACT_MATCH — точное совпадение основного поля (s_label == s_label)
2. FIND_IN_SYNONYM — основное поле текущей записи найдено в синонимах другой
3. EXACT_SYNONYM_MATCH — синонимы текущей записи совпадают с синонимами другой

Где вызывается:

• ✅ В админке: admin.py (переопределение clean() формы) → показывает ошибку пользователю
• ✅ В моделях: save() методы → выбрасывают ValidationError
• ❌ В парсере: НЕТ (парсер еще не создан)

Конфликты синонимов (TODO)

Сценарий 1: FIND_IN_SYNONYM

# В БД уже есть
Artist(pk=1, s_artist="Beatles", j_artist_metadata={"SYN_EN": ["The Beatles", "Beatles"]})

# Парсер хочет создать
Artist(pk=None, s_artist="The Beatles", j_artist_metadata={"SYN_EN": []})

# Валидатор: основное поле "The Beatles" найдено в синонимах существующей записи!
# Текущее поведение: ValidationError (блокировка)
# TODO: Отправить в очередь, ждать решения админа

Сценарий 2: EXACT_SYNONYM_MATCH

# В БД
Label(pk=1, s_label="Sony", j_label_metadata={"SYN_EN": ["Sony Records", "Sony Music"]})

# Парсер хочет
Label(pk=None, s_label="Sony Music", j_label_metadata={"SYN_EN": ["Sony Records", "Sony Music"]})

# Валидатор: синонимы совпадают!
# Текущее поведение: ValidationError
# TODO: Отправить в очередь, ждать решения админа

Архитектура парсера (требуемая)

Компоненты

┌──────────────────────────────────────────────────────────────┐
│                      DJANGO ADMIN                            │
│  ┌─────────────────────────────────────────────────────────┐ │
│  │ Приложение "Parser" (новое приложение)                 │ │
│  │ ┌─────────────────────────────────────────────────────┐ │ │
│  │ │ Очередь валидации (ValidatorQueue model)            │ │ │
│  │ │ Содержит: состояние экземпляра, ошибки, решение... │ │ │
│  │ └─────────────────────────────────────────────────────┘ │ │
│  │ ┌─────────────────────────────────────────────────────┐ │ │
│  │ │ Admin list view + inline actions                   │ │ │
│  │ │ - Просмотр конфликта (дубликаты)                  │ │ │
│  │ │ - Кнопка "Создать новую запись"                   │ │ │
│  │ │ - Кнопка "Объединить/обновить синонимы"           │ │ │
│  │ │ - Кнопка "Пропустить"                             │ │ │
│  │ └─────────────────────────────────────────────────────┘ │ │
│  └─────────────────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────┘

┌──────────────────────────────────────────────────────────────┐
│                    БРОКЕР ОЧЕРЕДИ (Redis)                    │
│  ┌─────────────────────────────────────────────────────────┐ │
│  │ parser:validation:pending (ZSET)                        │ │
│  │ parser:validation:{queue_id}:data (STRING/JSON)         │ │
│  │ parser:validation:{queue_id}:duplicates (JSON)          │ │
│  │ parser:validation:{queue_id}:solution (STRING)          │ │
│  └─────────────────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────┘

┌──────────────────────────────────────────────────────────────┐
│                      ПАРСЕР (Celery Task)                    │
│  ┌─────────────────────────────────────────────────────────┐ │
│  │ parse_excel() / parse_csv() / parse_url()              │ │
│  │                                                          │ │
│  │ Логика:                                                 │ │
│  │ 1. Читать источник (файл/URL)                          │ │
│  │ 2. Нормализовать данные                                │ │
│  │ 3. Валидировать через validate_for_duplicates()       │ │
│  │ 4. Если конфликт → создать запись в ValidatorQueue   │ │
│  │ 5. Если OK → создать модель (TbLabel, TbArtist и т.д.)│ │
│  │ 6. Отправить уведомление админу (Redis/сигнал)        │ │
│  │ 7. Продолжить обработку следующей строки              │ │
│  └─────────────────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────┘

┌──────────────────────────────────────────────────────────────┐
│              СИГНАЛЫ И УВЕДОМЛЕНИЯ (Django signals)          │
│  ┌─────────────────────────────────────────────────────────┐ │
│  │ post_validate_conflict → Redis уведомление админу      │ │
│  │ post_admin_decision → Обновление очереди              │ │
│  └─────────────────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────┘

Таблица ValidatorQueue (новая модель)

class ValidatorQueue(models.Model):
    """
    Очередь конфликтов, требующих решения админа.
    """
    class Status(TextChoices):
        PENDING = 'pending'      # Ждёт решения админа
        APPROVED = 'approved'    # Админ одобрил - выполнить действие
        REJECTED = 'rejected'    # Админ отклонил
        MERGED = 'merged'        # Данные объединены с существующей записью
        SKIPPED = 'skipped'      # Заметка: пропущено

    class ConflictType(TextChoices):
        FIND_IN_SYNONYM = 'find_in_synonym'          # Основное поле в синонимах
        EXACT_SYNONYM_MATCH = 'exact_synonym_match'  # Синонимы совпадают
        OTHER = 'other'

    # Основные поля
    id = BigAutoField(primary_key=True)
    k_source = ForeignKey(TbSource, ...)            # Откуда пришли данные
    l_conflict_type = CharField(choices=ConflictType)

    # Данные для конфликта
    l_model_type = CharField()                       # Какая модель (TbLabel, TbArtist и т.д.)
    model_pk = IntegerField(null=True)               # PK существующей записи в БД
    j_incoming_data = JSONField()                    # Данные от парсера (весь объект)
    j_duplicates = JSONField()                       # Результат validate_for_duplicates()

    # Решение админа
    l_status = CharField(choices=Status, default=Status.PENDING)
    s_admin_decision = CharField()                   # 'create_new' / 'merge' / 'skip'
    s_admin_notes = TextField()                      # Комментарий админа

    # Сервисные поля
    t_created = DateTimeField(auto_now_add=True)
    t_resolved = DateTimeField(null=True)
    k_resolved_by = ForeignKey(User, null=True)     # Какой админ принял решение

    class Meta:
        verbose_name = 'Конфликт валидации'
        ordering = ['-t_created']
        indexes = [
            Index(fields=['l_status', '-t_created']),
            Index(fields=['k_source', 'l_status']),
        ]

Процесс работы

Фаза 1: Импорт данных (парсер)

# tasks.py (Celery)
@celery.task
def parse_and_validate_excel(source_id: int):
    """
    1. Читает Excel из TbSource
    2. Парсит строки
    3. Валидирует каждую запись
    4. Создаёт или отправляет в очередь
    """
    source = TbSource.objects.get(pk=source_id)
    
    for row in read_excel(source.source_file):
        try:
            # Нормализуем данные
            artist_data = {
                's_artist': row['Artist Name'],
                'j_artist_metadata': {'SYN_EN': [row.get('Aliases', '')]}
            }
            
            # Валидируем
            duplicates = validate_for_duplicates(
                model_class=TbArtist,
                instance_pk=None,
                main_field_value=artist_data['s_artist'],
                metadata_dict=artist_data['j_artist_metadata'],
                main_field_name='s_artist',
                metadata_field_name='j_artist_metadata',
            )
            
            if duplicates:
                # КОНФЛИКТ! Отправляем в очередь
                queue_item = ValidatorQueue.objects.create(
                    k_source=source,
                    l_model_type='TbArtist',
                    l_conflict_type=duplicates[0][VALIDATE_KEY__MATCH_TYPE],
                    j_incoming_data=artist_data,
                    j_duplicates=duplicates,
                )
                # Отправляем сигнал (Redis уведомление админу)
                send_to_redis_queue('parser:conflicts:new', {
                    'queue_id': queue_item.id,
                    'model': 'TbArtist',
                    'conflict': duplicates[0]['match_type'],
                    'data': artist_data,
                })
            else:
                # Нет конфликта - создаём запись
                artist = TbArtist.objects.create(**artist_data)
                logger.info(f"Artist created: {artist.id}")
                
        except Exception as e:
            logger.error(f"Parse error in row {row}: {e}")
            continue

Фаза 2: Админка (приложение Parser)

# admin.py (новое приложение parser)
@admin.register(ValidatorQueue)
class ValidatorQueueAdmin(admin.ModelAdmin):
    """
    Админка для работы с очередью конфликтов.
    """
    list_display = [
        'id', 'l_model_type', 'l_conflict_type', 'l_status',
        'source_link', 'duplicates_summary', 't_created'
    ]
    list_filter = ['l_status', 'l_conflict_type', 'l_model_type', 't_created']
    readonly_fields = ['j_incoming_data', 'j_duplicates', 'k_source']
    
    fieldsets = (
        ('Конфликт', {
            'fields': ['l_model_type', 'l_conflict_type', 'k_source', 'model_pk']
        }),
        ('Входящие данные', {
            'fields': ['j_incoming_data']
        }),
        ('Найденные дубликаты', {
            'fields': ['j_duplicates']
        }),
        ('Решение админа', {
            'fields': ['s_admin_decision', 's_admin_notes', 'l_status']
        }),
    )
    
    actions = ['action_create_new', 'action_merge_synonyms', 'action_skip']
    
    def action_create_new(self, request, queryset):
        """Админ решил: создать новую запись, игнорируя дубликаты"""
        for item in queryset:
            model_class = get_model_class(item.l_model_type)
            instance = model_class(**item.j_incoming_data)
            instance.save(skip_validation=True)  # Пропускаем валидацию
            item.l_status = ValidatorQueue.Status.APPROVED
            item.s_admin_decision = 'create_new'
            item.k_resolved_by = request.user
            item.t_resolved = now()
            item.save()

Фаза 3: Обработка решения админа

# signals.py (новое приложение parser)

@receiver(post_save, sender=ValidatorQueue)
def on_admin_decision(sender, instance, created=False, **kwargs):
    """
    Обработать решение админа и применить действие.
    """
    if created or instance.l_status != ValidatorQueue.Status.APPROVED:
        return
    
    if instance.s_admin_decision == 'create_new':
        # Админ одобрил создание новой записи
        model_class = get_model_class(instance.l_model_type)
        model_instance = model_class(**instance.j_incoming_data)
        model_instance.save(skip_validation=True)
        
    elif instance.s_admin_decision == 'merge':
        # Админ решил объединить (добавить синонимы, и т.д.)
        existing_pk = instance.model_pk
        incoming_data = instance.j_incoming_data
        
        model_class = get_model_class(instance.l_model_type)
        existing = model_class.objects.get(pk=existing_pk)
        
        # Добавляем синонимы из входящих данных
        existing_meta = existing.j_*_metadata or {}
        existing_synonyms = existing_meta.get(KEY_SYNONYM_EN, [])
        incoming_synonyms = incoming_data.get('j_*_metadata', {}).get(KEY_SYNONYM_EN, [])
        
        # Объединяем и сохраняем
        existing_meta[KEY_SYNONYM_EN] = list(set(existing_synonyms + incoming_synonyms))
        existing.j_*_metadata = existing_meta
        existing.save(skip_validation=True)

Файлы для создания

1. Новое приложение parser

lpon_site/parser/
├── __init__.py
├── models.py           # ValidatorQueue
├── admin.py            # ValidatorQueueAdmin + actions
├── signals.py          # Обработка решений админа
├── tasks.py            # Celery tasks для парсинга
├── apps.py
└── migrations/

2. Изменения в existing коде

settings.py:

INSTALLED_APPS = [
    ...,
    'parser',  # Новое приложение
]

# Конфигурация Celery (если еще нет)
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'

frontend/models.py:

# В методе save() моделей добавить параметр
def save(self, *args, skip_validation=False, **kwargs):
    if not skip_validation:
        # Стандартная валидация
        validate_and_raise_for_duplicates(...)
    super().save(*args, **kwargs)

frontend/utils.py:

# Добавить функцию отправки в Redis
def send_to_redis_queue(queue_name: str, data: dict):
    """Отправить уведомление в Redis"""
    import redis
    import json
    
    r = redis.Redis(host='localhost', port=6379, db=0)
    r.lpush(queue_name, json.dumps({
        'timestamp': datetime.now().isoformat(),
        **data
    }))

TODO задачи

• Создать приложение parser с моделью ValidatorQueue
• Реализовать админку ValidatorQueueAdmin с actions
• Написать Celery tasks для парсинга Excel/CSV
• Добавить обработку решений админа через signals
• Интегрировать Redis уведомления админу
• Добавить параметр skip_validation в save() методы
• Написать тесты для парсера и валидации
• Документация для парсера (как использовать)

Примечания

1. Redis vs БД для очереди? → Используем БД (ValidatorQueue модель) как основное хранилище, Redis только для уведомлений админу в реальном времени
2. Async валидация? → Парсер работает в Celery, валидация синхронна внутри задачи
3. Миграция истории? → Поле t_history_created уже editable=True в TbOfferHistory, так что админ может вручную добавлять исторические данные
4. Конфликты в админке? → ValidatorQueue блокирует парсер от создания дубликатов. Админ сам решает что делать.

Ссылки на существующий код

• Валидаторы: lpon_site/frontend/utils_validators.py
• Модели: lpon_site/frontend/models.py (методы save() для TbArtist, TbItem, TbLabel, TbSeller, TbMusicStyle)
• TODO про Redis: lpon_site/frontend/utils.py (в функции create_or_get_related_article())
• TODO про историю: lpon_site/frontend/models.py (в TbOffer.save())