Реализация автоматического переноса контента между 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):
        """Перенос медиафайлов через API Strapi"""
        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 AND pm.meta_key = '_wp_attachment_image_alt'
            WHERE p.post_type = 'attachment'
            AND p.post_mime_type LIKE 'image/%'
        """)

        for media in tqdm(cursor.fetchall(), desc="Migrating media"):
            try:
                # Скачать с WP-сервера
                response = requests.get(media['guid'], timeout=30)
                if response.status_code != 200:
                    logger.warning(f"Cannot fetch {media['guid']}")
                    continue

                # Загрузить в Strapi
                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.*, GROUP_CONCAT(
                DISTINCT CASE WHEN tt.taxonomy = 'category' THEN t.name END
                SEPARATOR ','
            ) as categories
            FROM wp_posts p
            LEFT JOIN wp_term_relationships tr ON p.ID = tr.object_id
            LEFT JOIN wp_term_taxonomy tt ON tr.term_taxonomy_id = tt.term_taxonomy_id
            LEFT JOIN wp_terms t ON tt.term_id = t.term_id
            WHERE p.post_type = 'post' AND p.post_status = 'publish'
            GROUP BY p.ID
            ORDER BY p.ID
            LIMIT %s OFFSET %s
        """, (batch_size, offset))

        posts = cursor.fetchall()
        if not posts:
            return 0

        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', '_yoast_wpseo_metadesc')
        """, (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(),
                'seo': {
                    'metaTitle': meta.get('_yoast_wpseo_title', post['post_title']),
                    'metaDescription': meta.get('_yoast_wpseo_metadesc', ''),
                },
                '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()
        logger.info(f"Media map: {len(self.media_map)} files")

        offset = 0
        total = 0
        while True:
            count = self.migrate_posts(batch_size=50, offset=offset)
            total += count
            if count < 50:
                break
            offset += 50

        logger.info(f"Migration complete: {total} posts")

Joomla → WordPress

Используется FG Joomla to WordPress plugin или кастомный скрипт через прямой доступ к БД Joomla:

def migrate_joomla_articles(joomla_conn, wp_conn):
    j_cursor = joomla_conn.cursor(dictionary=True)
    j_cursor.execute("""
        SELECT a.id, a.title, a.alias, a.introtext, a.fulltext,
               a.state, a.created, a.modified, a.metadesc, a.metakey,
               c.title as category_name, c.alias as category_slug
        FROM jos_content a
        LEFT JOIN jos_categories c ON a.catid = c.id
        WHERE a.state = 1
    """)

    for article in j_cursor.fetchall():
        content = article['introtext']
        if article['fulltext']:
            content += '<!--more-->' + article['fulltext']

        wp_cursor = wp_conn.cursor()
        wp_cursor.execute("""
            INSERT INTO wp_posts
            (post_title, post_name, post_content, post_status, post_date, post_type)
            VALUES (%s, %s, %s, 'publish', %s, 'post')
        """, (article['title'], article['alias'], content, article['created']))

        post_id = wp_cursor.lastrowid

        # Добавить meta description
        if article['metadesc']:
            wp_cursor.execute("""
                INSERT INTO wp_postmeta (post_id, meta_key, meta_value)
                VALUES (%s, '_yoast_wpseo_metadesc', %s)
            """, (post_id, article['metadesc']))

    wp_conn.commit()

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&pagination[pageSize]=100")

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

Срок выполнения

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