Автоматизированная проверка целостности данных после миграции

Реализация проверки целостности данных после миграции

Миграция данных — всегда стресс. Вы перенесли тысячи записей, но уверены ли, что ничего не потерялось? Связи не оборваны? URL не битые? Без автоматизированной проверки остаётся только гадать. Ручная выборочная проверка — это лотерея: вы проверяете 5% записей, а ошибки прячутся в оставшихся 95%. Особенно когда речь идёт о вложенных комментариях, мета-полях или SEO-тегах — каждый пропуск грозит потерей трафика или функциональности.

Недавно мы работали с клиентом: 10 000 постов, 50 000 комментариев на WordPress, миграция на Laravel. Ручная выборочная проверка (3 дня работы двух инженеров) нашла 30 битых ссылок. Автоматизированная — за 1 день — выявила ещё 200 потерянных комментариев, 5% дубликатов URL и 150 страниц без SEO-заголовков. Разница очевидна: автоматизация в 3 раза быстрее и в 20 раз точнее.

Мы — команда инженеров с 5+ годами опыта в сложных миграциях — разработали набор скриптов, который за 1–2 дня даёт полную картину целостности. Скрипты адаптируются под вашу CMS и структуру БД, будь то PostgreSQL, MySQL или MongoDB.

Почему автоматизированная проверка целостности необходима?

Ручная выборочная проверка неэффективна. Вы рискуете пропустить потерю 2–5% записей (особенно вложенных комментариев или мета-полей), разрыв связей «родитель-потомок», дубликаты URL и SEO-тегов, которые убьют ранжирование. Автоматизация снимает эти риски. Сравним два подхода:

Типичные последствия пропущенных ошибок:

Что мы проверяем: детальный чек-лист

Наши инженеры реализовали скрипты для пяти ключевых проверок. Каждая — с конкретными метриками.

Как работает проверка контрольных сумм?

Сравниваем агрегированный MD5 по критичным полям. Выявляет даже незначительные изменения в данных. Источник: Wikipedia, MD5.

def checksum_check(source_db, target_db):
    """Сравнение контрольных сумм по критичным полям"""
    # PostgreSQL
    source_hash = source_db.query_one("""
        SELECT md5(string_agg(
            md5(id::text || coalesce(email,'') || coalesce(slug,'')),
            ',' ORDER BY id
        )) as hash
        FROM articles
        WHERE status = 'published'
    """)
    target_hash = target_db.query_one("""
        SELECT md5(string_agg(
            md5(legacy_id || coalesce(email,'') || coalesce(slug,'')),
            ',' ORDER BY CAST(legacy_id AS INTEGER)
        )) as hash
        FROM articles
        WHERE status = 'published'
    """)
    return source_hash == target_hash

Количественная проверка

Сравниваем количество записей по каждому типу контента. Учитываем фильтры статусов (published/draft). Пример кода:

class MigrationValidator:
    def __init__(self, source_db, target_db):
        self.source = source_db
        self.target = target_db
        self.results = []

    def check_counts(self):
        tables = [
            ('posts', 'articles', "status='publish'", "status='published'"),
            ('users', 'users', None, None),
            ('comments', 'comments', "approved=1", "status='approved'"),
            ('categories', 'categories', None, None),
        ]
        for src_table, tgt_table, src_where, tgt_where in tables:
            src_count = self.source.count(src_table, src_where)
            tgt_count = self.target.count(tgt_table, tgt_where)
            status = 'OK' if src_count == tgt_count else 'MISMATCH'
            self.results.append({
                'check': f'count_{src_table}',
                'status': status,
                'source': src_count,
                'target': tgt_count,
                'diff': tgt_count - src_count
            })
"}

Проверка ссылочной целостности

Ищем сиротские записи: статьи без автора, комментарии без родительской статьи или родительского комментария.

def check_referential_integrity(target_db):
    issues = []
    # Статьи без автора
    orphaned_posts = target_db.query("""
        SELECT a.id, a.title
        FROM articles a
        LEFT JOIN users u ON a.author_id = u.id
        WHERE a.author_id IS NOT NULL AND u.id IS NULL
    """)
    if orphaned_posts:
        issues.append(f"Articles without valid author: {len(orphaned_posts)}")

    # Комментарии к несуществующим постам
    orphaned_comments = target_db.query("""
        SELECT c.id
        FROM comments c
        LEFT JOIN articles a ON c.post_id = a.id
        WHERE a.id IS NULL
    """)
    if orphaned_comments:
        issues.append(f"Orphaned comments: {len(orphaned_comments)}")

    # Дочерние комментарии без родителя
    broken_threads = target_db.query("""
        SELECT c.id
        FROM comments c
        LEFT JOIN comments p ON c.parent_id = p.id
        WHERE c.parent_id IS NOT NULL AND p.id IS NULL
    """)
    if broken_threads:
        issues.append(f"Comments with missing parent: {len(broken_threads)}")

    return issues

Проверка доступности URL

Асинхронно проверяем каждый опубликованный URL на HTTP 200, 404, 500 и цепочки редиректов.

import asyncio
import aiohttp

async def check_urls(urls, base_url, concurrency=20):
    errors = {'404': [], '500': [], 'redirect_chain': []}
    semaphore = asyncio.Semaphore(concurrency)

    async def check_one(session, path):
        async with semaphore:
            url = f"{base_url}{path}"
            try:
                async with session.get(url, allow_redirects=True) as resp:
                    if resp.status == 404:
                        errors['404'].append(path)
                    elif resp.status >= 500:
                        errors['500'].append(path)
                    elif len(resp.history) > 2:
                        errors['redirect_chain'].append(f"{path} ({len(resp.history)} redirects)")
            except Exception as e:
                errors['500'].append(f"{path} (error: {e})")

    async with aiohttp.ClientSession() as session:
        tasks = [check_one(session, url) for url in urls]
        await asyncio.gather(*tasks)
    return errors

# Запуск
urls_to_check = get_all_published_urls(target_db)
results = asyncio.run(check_urls(urls_to_check, 'https://new-site.com'))

Проверка SEO-метаданных

Ищем страницы без title/description, дубликаты title, отсутствие canonical.

def check_seo_completeness(target_db):
    issues = []
    # Страницы без title
    no_title = target_db.query("""
        SELECT slug FROM articles
        WHERE (seo_title IS NULL OR seo_title = '') AND status = 'published'
    """)
    if no_title:
        issues.append(f"Pages without SEO title: {len(no_title)}")

    # Страницы без meta description
    no_desc = target_db.query("""
        SELECT slug FROM articles
        WHERE (seo_description IS NULL OR seo_description = '') AND status = 'published'
    """)
    if no_desc:
        issues.append(f"Pages without meta description: {len(no_desc)}")

    # Дублирующиеся title
    dup_titles = target_db.query("""
        SELECT seo_title, COUNT(*) as count FROM articles
        WHERE status = 'published'
        GROUP BY seo_title
        HAVING COUNT(*) > 1
    """)
    if dup_titles:
        issues.append(f"Duplicate SEO titles: {len(dup_titles)} groups")

    return issues

Как мы это делаем: наш процесс

1. Анализ схемы — изучаем структуру исходной и целевой БД, выявляем критические таблицы и поля.
2. Адаптация скриптов — настраиваем проверки под конкретный стек (PostgreSQL/MySQL, CMS).
3. Запуск и сбор результатов — выполняем скрипты, формируем отчёт в HTML/JSON/JUnit.
4. Анализ и рекомендации — разбираем каждый упавший тест, предлагаем план исправлений.
5. Деплой в CI/CD (опционально) — упаковываем проверку в Docker-контейнер, подключаем к пайплайну.

Что входит в работу

• Скрипты проверки (Python, конфигурируются под ваш проект).
• Детальный отчёт с таблицей результатов и списком проблемных записей.
• Рекомендации по исправлению каждой ошибки.
• Документация по запуску и интерпретации результатов.
• Обучение одного инженера со стороны клиента.
• Гарантия: если после исправления проверка выявит несоответствие, мы бесплатно дорабатываем скрипт.

Ориентировочные сроки

Разработка набора проверок и финальный отчёт — 1–2 рабочих дня. Сложность зависит от числа таблиц и объёма данных. Стоимость рассчитывается индивидуально — свяжитесь с нами, чтобы получить точную оценку.

Почему выбирают нас?

• 5+ лет миграций: от WordPress до кастомных решений на Laravel.
• 200+ проектов по переносу данных.
• Сертифицированные инженеры по PostgreSQL и Docker.
• Гарантия результата: все проверки проходят на тестовом стенде до основного прогона.

Закажите консультацию по проверке целостности данных после миграции — мы оценим объём и подберём оптимальный набор проверок. Свяжитесь с нами, чтобы получить детальный план работ и стоимость.