Реализация маппинга контента (старая структура → новая) при миграции

Реализация маппинга контента (старая структура → новая) при миграции

Маппинг контента — создание соответствия между полями, типами и структурами старой CMS и новой платформы. Без явного маппинга данные либо теряются, либо попадают не в те поля.

Что включает маппинг

• Типы контента: page → post, article → news, product → catalog_item
• Поля: post_title → title, post_content → body, post_date → published_at
• Таксономии: categories/tags → одна или разные таксономии
• Медиафайлы: attachment → media_library
• Пользователи: wp_users.user_login → users.username

Документация маппинга в виде схемы

# content-mapping.yml
content_types:
  - source: "post"
    target: "article"
    fields:
      - source: "ID"
        target: "legacy_id"
        transform: "int_to_string"
      - source: "post_title"
        target: "title"
        transform: null
      - source: "post_content"
        target: "body"
        transform: "wp_shortcodes_to_html"
      - source: "post_excerpt"
        target: "summary"
        transform: "strip_tags"
      - source: "post_date"
        target: "published_at"
        transform: "datetime_utc"
      - source: "post_status"
        target: "status"
        transform: "map_status"  # publish→published, draft→draft, private→hidden
      - source: "_yoast_wpseo_title"
        target: "seo_title"
        source_type: "meta"
      - source: "_yoast_wpseo_metadesc"
        target: "seo_description"
        source_type: "meta"
      - source: "featured_image"
        target: "cover_image_id"
        transform: "resolve_attachment_id"

  - source: "page"
    target: "page"
    fields:
      - source: "post_title"
        target: "name"
      - source: "post_content"
        target: "content"
      - source: "post_name"  # slug
        target: "slug"

taxonomies:
  - source: "category"
    target: "category"
    preserve_hierarchy: true
  - source: "post_tag"
    target: "tag"
    preserve_hierarchy: false

Python-скрипт маппинга WordPress → кастомная CMS

import mysql.connector
import requests
import json
from datetime import datetime

class WordPressMapper:
    def __init__(self, wp_conn, target_api):
        self.wp = wp_conn
        self.api = target_api
        self.attachment_map = {}  # wp_id → new_id
        self.user_map = {}
        self.category_map = {}

    def map_post(self, wp_post):
        # Получить мета-данные поста
        cursor = self.wp.cursor(dictionary=True)
        cursor.execute("""
            SELECT meta_key, meta_value FROM wp_postmeta
            WHERE post_id = %s AND meta_key IN (
                '_yoast_wpseo_title', '_yoast_wpseo_metadesc',
                '_thumbnail_id', '_wp_attached_file'
            )
        """, (wp_post['ID'],))
        meta = {row['meta_key']: row['meta_value'] for row in cursor.fetchall()}

        # Получить категории и теги
        cursor.execute("""
            SELECT t.name, t.slug, tt.taxonomy
            FROM wp_terms t
            JOIN wp_term_taxonomy tt ON t.term_id = tt.term_id
            JOIN wp_term_relationships tr ON tt.term_taxonomy_id = tr.term_taxonomy_id
            WHERE tr.object_id = %s
        """, (wp_post['ID'],))
        terms = cursor.fetchall()

        return {
            'legacy_id': str(wp_post['ID']),
            'title': wp_post['post_title'],
            'body': self.transform_content(wp_post['post_content']),
            'summary': self.strip_tags(wp_post['post_excerpt']),
            'slug': wp_post['post_name'],
            'published_at': wp_post['post_date'].isoformat() + 'Z',
            'status': self.map_status(wp_post['post_status']),
            'author_id': self.user_map.get(wp_post['post_author']),
            'seo_title': meta.get('_yoast_wpseo_title', ''),
            'seo_description': meta.get('_yoast_wpseo_metadesc', ''),
            'cover_image_id': self.attachment_map.get(meta.get('_thumbnail_id')),
            'categories': [
                self.category_map.get(t['slug'])
                for t in terms if t['taxonomy'] == 'category'
            ],
            'tags': [t['slug'] for t in terms if t['taxonomy'] == 'post_tag'],
        }

    def map_status(self, wp_status):
        return {
            'publish': 'published',
            'draft': 'draft',
            'private': 'hidden',
            'trash': None  # пропустить
        }.get(wp_status, 'draft')

    def transform_content(self, content):
        # Заменить WordPress shortcodes на HTML
        import re
        # [gallery ids="1,2,3"] → <div class="gallery">...</div>
        content = re.sub(
            r'\[gallery ids="([^"]+)"\]',
            lambda m: self.render_gallery(m.group(1)),
            content
        )
        # Заменить внутренние ссылки
        content = content.replace('https://old-site.com/', '/')
        return content

Маппинг из Drupal в WordPress

Drupal использует более сложную структуру (field API):

def map_drupal_node(node_row, field_data):
    return {
        'post_title': node_row['title'],
        'post_content': field_data.get('body_value', ''),
        'post_status': 'published' if node_row['status'] == 1 else 'draft',
        'post_date': datetime.fromtimestamp(node_row['created']).strftime('%Y-%m-%d %H:%M:%S'),
        'post_name': node_row['alias'] or slugify(node_row['title']),
        '_yoast_wpseo_title': field_data.get('field_meta_title_value', ''),
        '_yoast_wpseo_metadesc': field_data.get('field_meta_desc_value', ''),
    }

Маппинг таксономий с сохранением иерархии

def migrate_categories(wp_cursor, target_api):
    # Получить категории с родителями
    wp_cursor.execute("""
        SELECT t.term_id, t.name, t.slug, tt.parent
        FROM wp_terms t
        JOIN wp_term_taxonomy tt ON t.term_id = tt.term_id
        WHERE tt.taxonomy = 'category'
        ORDER BY tt.parent ASC  # сначала родители
    """)

    category_map = {}  # wp_term_id → new_id

    for cat in wp_cursor.fetchall():
        response = target_api.post('/categories', {
            'name': cat['name'],
            'slug': cat['slug'],
            'parent_id': category_map.get(cat['parent'])
        })
        category_map[cat['term_id']] = response['id']

    return category_map

Верификация маппинга

# Проверить: все посты имеют все обязательные поля
def validate_mapped_post(post):
    required = ['title', 'body', 'slug', 'published_at']
    missing = [f for f in required if not post.get(f)]
    if missing:
        print(f"WARNING: Post {post.get('legacy_id')} missing: {missing}")
    return len(missing) == 0

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

Разработка маппинга и скриптов трансформации для сайта до 5000 страниц — 3–5 рабочих дней.