Реалізація автоматичного перенесення контенту між CMS

Реалізація автоматичного переносу контенту між CMS

Автоматичний перенесення контенту — розробка скриптів і пайплайнів, що витягують дані з вихідної CMS, трансформують та імпортують у цільову. Виключає ручне копіювання сотень чи тисяч матеріалів.

Архітектура ETL-пайплайну

Extract (вихідна CMS)
    ↓
Transform (маппінг полів, чистка даних)
    ↓
Load (цільова CMS через API або пряма запис у БД)
    ↓
Verify (перевірка результату)

WordPress → Headless CMS (Contentful/Strapi)

import mysql.connector
import requests
from tqdm import tqdm
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class WordPressToStrapi:
    def __init__(self):
        self.wp_db = mysql.connector.connect(
            host='old-wp-server',
            database='wp_production',
            user='readonly',
            password='password'
        )
        self.strapi_url = 'http://new-strapi:1337/api'
        self.strapi_token = 'strapi-api-token'
        self.media_map = {}  # wp_attachment_id → strapi_media_id

    def migrate_media(self):
        cursor = self.wp_db.cursor(dictionary=True)
        cursor.execute("""
            SELECT p.ID, p.guid, p.post_title, pm.meta_value as alt_text
            FROM wp_posts p
            LEFT JOIN wp_postmeta pm ON p.ID = pm.post_id
            WHERE p.post_type = 'attachment' AND p.post_mime_type LIKE 'image/%'
        """)

        for media in tqdm(cursor.fetchall(), desc="Migrating media"):
            try:
                response = requests.get(media['guid'], timeout=30)
                if response.status_code != 200:
                    logger.warning(f"Cannot fetch {media['guid']}")
                    continue

                filename = media['guid'].split('/')[-1]
                upload_response = requests.post(
                    f"{self.strapi_url}/upload",
                    headers={'Authorization': f"Bearer {self.strapi_token}"},
                    files={'files': (filename, response.content)},
                    data={'fileInfo': f'{{"alternativeText": "{media.get("alt_text", "")}"}}'  }
                )

                if upload_response.status_code == 200:
                    new_id = upload_response.json()[0]['id']
                    self.media_map[media['ID']] = new_id
                    logger.info(f"Media {media['ID']} → {new_id}")

            except Exception as e:
                logger.error(f"Media {media['ID']} failed: {e}")

    def migrate_posts(self, batch_size=50, offset=0):
        cursor = self.wp_db.cursor(dictionary=True)
        cursor.execute("""
            SELECT p.* FROM wp_posts p
            WHERE p.post_type = 'post' AND p.post_status = 'publish'
            ORDER BY p.ID
            LIMIT %s OFFSET %s
        """, (batch_size, offset))

        posts = cursor.fetchall()
        for post in tqdm(posts, desc=f"Posts batch {offset//batch_size + 1}"):
            self._migrate_single_post(post)

        return len(posts)

    def _migrate_single_post(self, post):
        cursor = self.wp_db.cursor(dictionary=True)
        cursor.execute("""
            SELECT meta_key, meta_value FROM wp_postmeta
            WHERE post_id = %s AND meta_key IN ('_thumbnail_id', '_yoast_wpseo_title')
        """, (post['ID'],))
        meta = {r['meta_key']: r['meta_value'] for r in cursor.fetchall()}

        payload = {
            'data': {
                'title': post['post_title'],
                'slug': post['post_name'],
                'content': post['post_content'],
                'publishedAt': post['post_date'].isoformat(),
                'cover': self.media_map.get(meta.get('_thumbnail_id')),
                'legacy_wp_id': post['ID'],
            }
        }

        response = requests.post(
            f"{self.strapi_url}/articles",
            headers={
                'Authorization': f"Bearer {self.strapi_token}",
                'Content-Type': 'application/json'
            },
            json=payload
        )

        if response.status_code not in (200, 201):
            logger.error(f"Post {post['ID']} failed: {response.text}")

    def run(self):
        logger.info("Starting migration...")
        self.migrate_media()
        offset = 0
        while True:
            count = self.migrate_posts(batch_size=50, offset=offset)
            if count < 50:
                break
            offset += 50

Idempotent Міграція (Безпечний повторний запуск)

def safe_create_post(api, data):
    """Створити пост тільки якщо ще не мігрований"""
    legacy_id = data.get('legacy_wp_id')

    check = api.get(f"/articles?filters[legacy_wp_id][$eq]={legacy_id}")
    if check and check.get('data'):
        logger.info(f"Post {legacy_id} already migrated, skipping")
        return check['data'][0]['id']

    response = api.post('/articles', {'data': data})
    return response['data']['id']

Rollback Можливості

def rollback_migration(api, dry_run=True):
    """Видалити всі дописи з legacy_wp_id (откат міграції)"""
    posts = api.get("/articles?filters[legacy_wp_id][$notNull]=true")

    for post in posts['data']:
        logger.info(f"Deleting post {post['id']}")
        if not dry_run:
            api.delete(f"/articles/{post['id']}")

Тривалість виконання

ETL-скрипт для переносу 1000–5000 матеріалів між двома CMS — 3–5 робочих днів. З перенесенням медіафайлів та SEO-метаданих — 5–7 днів.