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

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

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

Замовте консультацію з перевірки цілісності даних після міграції — ми оцінимо обсяг і підберемо оптимальний набір перевірок. Зв'яжіться з нами, щоб отримати детальний план робіт і вартість.