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

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

Миграция данных — всегда стресс. Вы перенесли тысячи записей, но уверены ли, что ничего не потерялось? Связи не оборваны? 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.
• Гарантия результата: все проверки проходят на тестовом стенде до основного прогона.

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