Automated content transfer between CMS platforms

Implementing Automatic Content Transfer Between CMS

Automatic content transfer is developing scripts and pipelines that extract data from source CMS, transform and import into target. Eliminates manual copying hundreds or thousands of items.

ETL Pipeline Architecture

Extract (source CMS)
    ↓
Transform (field mapping, data cleanup)
    ↓
Load (target CMS via API or direct DB write)
    ↓
Verify (check result)

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 Migration (Safe Rerun)

def safe_create_post(api, data):
    """Create post only if not already migrated"""
    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 Capability

def rollback_migration(api, dry_run=True):
    """Delete all posts with legacy_wp_id (migration rollback)"""
    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']}")

Execution Time

ETL script for transferring 1000–5000 items between two CMS — 3–5 working days. With media files and SEO metadata — 5–7 days.