WordPress 代码块批量转换为 EnlighterJS

WordPress 使用时间久了，文章中的代码块格式可能比较混乱。如果后来改用 EnlighterJS，逐篇修改成本比较高。

这里通过一个 Python 脚本，直接处理 WordPress 导出的 SQL 文件，批量完成转换。

整个过程在本地完成，不需要连接 WordPress 数据库，也不需要上传 SQL 文件。

1. 导出数据库

先从 phpMyAdmin、宝塔等工具导出 WordPress 数据库，得到 .sql 文件。

建议保留原始 SQL，不要直接修改。

2. 运行 Python 脚本

运行转换脚本，选择刚才导出的 SQL 文件。

脚本会自动找到 wp_posts 表中的 post_content 字段，并处理文章内容。

主要转换三种情况：

• wp:preformatted → EnlighterJS
• wp:code → EnlighterJS
• 普通 <code> → EnlighterJS 行内代码

例如：

<!-- wp:preformatted -->
<pre class="wp-block-preformatted">apt-get update -y &amp;&amp; apt-get install curl -y</pre>
<!-- /wp:preformatted -->

转换后：

<!-- wp:enlighter/codeblock -->
<pre class="EnlighterJSRAW" data-enlighter-language="generic" data-enlighter-theme="" data-enlighter-highlight="" data-enlighter-linenumbers="" data-enlighter-lineoffset="" data-enlighter-title="" data-enlighter-group="">apt-get update -y &amp;&amp; apt-get install curl -y</pre>
<!-- /wp:enlighter/codeblock -->

普通行内代码：

<code>functions.php</code>

会转换成：

<code data-enlighter-language="generic" class="EnlighterJSRAW">functions.php</code>

已经是 EnlighterJS 的代码不会重复转换。

3. 查看转换结果

脚本完成后会自动生成：

原始备份_时间.sql

和：

enlighter_时间.sql

例如：

扫描文章：397
修改文章：61

wp:preformatted：31
wp:code：1
普通 <code>：166

4. 导入前检查

建议先用编辑器打开转换后的 SQL，搜索：

wp:enlighter/codeblock

确认代码块已经转换。

同时检查特殊字符，例如：

&&

不要变成：

&&

确认无误后，先导入测试站验证，没问题再用于正式站。

总结

整个过程就是：

WordPress 导出 SQL → Python 本地批量转换 → 检查 → 测试导入 → 正式使用

对于已经有大量历史文章的 WordPress 网站，可以一次性完成代码块格式迁移，避免逐篇手动修改。

代码

# -*- coding: utf-8 -*-
"""
WordPress EnlighterJS 本地 SQL 批量转换工具
================================================
完全本地运行：
    SQL备份
       ↓
    Python
       ↓
    自动解析 wp_posts
       ↓
    转换 post_content
       ↓
    生成新的 SQL
不会连接 WordPress
不会连接 MySQL
不会修改线上数据库
支持：
1. wp:preformatted
2. wp:code
3. 普通 <code>
4. 已有 Enlighter 自动跳过
5. 保留 &amp;，避免 &amp;amp;
6. 自动识别 wp_posts 字段顺序
7. 支持：
       INSERT INTO wp_posts VALUES (...)
   以及：
       INSERT INTO wp_posts (...) VALUES (...)
原 SQL 永远不会被修改。
"""
import os
import re
import shutil
import datetime
import tkinter as tk
from tkinter import filedialog

# ============================================================
# 配置
# ============================================================
# True：
# 只处理 post_type = 'post' 的文章
#
# False：
# 处理 wp_posts 中所有 post_content
#
# 如果你的页面、附件等也希望处理，可以改成 False。
ONLY_POST = True

# ============================================================
# 选择 SQL
# ============================================================
def choose_sql_file():
    root = tk.Tk()
    root.withdraw()
    path = filedialog.askopenfilename(
        title="请选择 WordPress 数据库 SQL 文件",
        filetypes=[
            ("SQL 文件", "*.sql"),
            ("所有文件", "*.*")
        ]
    )
    root.destroy()
    return path

# ============================================================
# SQL 字符串解析
# ============================================================
def parse_sql_string(text, pos):
    if pos >= len(text) or text[pos] != "'":
        raise ValueError("当前位置不是 SQL 字符串")
    pos += 1
    result = []
    while pos < len(text):
        ch = text[pos]
        # ----------------------------------------------------
        # 字符串结束
        # ----------------------------------------------------
        if ch == "'":
            # SQL 中：
            #
            # ''
            #
            # 表示一个单引号
            if (
                pos + 1 < len(text)
                and text[pos + 1] == "'"
            ):
                result.append("'")
                pos += 2
                continue
            pos += 1
            return ''.join(result), pos
        # ----------------------------------------------------
        # MySQL 反斜杠转义
        # ----------------------------------------------------
        if ch == '\\' and pos + 1 < len(text):
            nxt = text[pos + 1]
            mapping = {
                '0': '\0',
                'b': '\b',
                'n': '\n',
                'r': '\r',
                't': '\t',
                'Z': '\x1a',
                '\\': '\\',
                "'": "'",
                '"': '"'
            }
            if nxt in mapping:
                result.append(mapping[nxt])
            else:
                # 未知转义，保留
                result.append('\\')
                result.append(nxt)
            pos += 2
            continue
        result.append(ch)
        pos += 1
    raise ValueError("SQL 字符串没有找到结束引号")

# ============================================================
# SQL 字符串重新编码
# ============================================================
def sql_escape_string(value):
    value = value.replace('\\', '\\\\')
    value = value.replace('\0', '\\0')
    value = value.replace('\n', '\\n')
    value = value.replace('\r', '\\r')
    value = value.replace('\b', '\\b')
    value = value.replace('\t', '\\t')
    value = value.replace('\x1a', '\\Z')
    value = value.replace("'", "\\'")
    return "'" + value + "'"

# ============================================================
# 解析 CREATE TABLE 字段
# ============================================================
def find_posts_columns(sql_text, table_name):
    """
    从：
        CREATE TABLE `wp_posts` (
            `ID` bigint...
            `post_author` bigint...
            ...
            `post_content` longtext...
        )
    中读取字段顺序。
    返回：
        [
            'ID',
            'post_author',
            ...
            'post_content',
            ...
        ]
    """
    pattern = re.compile(
        r'CREATE\s+TABLE\s+'
        r'(?:IF\s+NOT\s+EXISTS\s+)?'
        r'[`"]?'
        + re.escape(table_name) +
        r'[`"]?'
        r'\s*\((.*?)\)'
        r'\s*(?:ENGINE|;)',
        re.IGNORECASE | re.DOTALL
    )
    match = pattern.search(sql_text)
    if not match:
        return []
    body = match.group(1)
    columns = []
    for line in body.splitlines():
        line = line.strip()
        if not line.startswith('`'):
            continue
        m = re.match(
            r'`([^`]+)`',
            line
        )
        if m:
            columns.append(m.group(1))
    return columns

# ============================================================
# 寻找 posts 表
# ============================================================
def find_posts_table(sql_text):
    patterns = [
        r'CREATE\s+TABLE\s+'
        r'(?:IF\s+NOT\s+EXISTS\s+)?'
        r'[`"]?([A-Za-z0-9_]+_posts)[`"]?',
        r'INSERT\s+INTO\s+'
        r'[`"]?([A-Za-z0-9_]+_posts)[`"]?'
    ]
    for pattern in patterns:
        match = re.search(
            pattern,
            sql_text,
            re.IGNORECASE
        )
        if match:
            return match.group(1)
    return None

# ============================================================
# Enlighter
# ============================================================
def make_enlighter_block(code):
    return (
        '<!-- wp:enlighter/codeblock -->\n'
        '<pre class="EnlighterJSRAW" '
        'data-enlighter-language="generic" '
        'data-enlighter-theme="" '
        'data-enlighter-highlight="" '
        'data-enlighter-linenumbers="" '
        'data-enlighter-lineoffset="" '
        'data-enlighter-title="" '
        'data-enlighter-group="">'
        + code +
        '</pre>\n'
        '<!-- /wp:enlighter/codeblock -->'
    )

def is_enlighter(value):
    lower = value.lower()
    return (
        'enlighterjsraw' in lower
        or 'data-enlighter-language' in lower
        or 'wp:enlighter/codeblock' in lower
    )

# ============================================================
# 转换 wp:preformatted
# ============================================================
def convert_preformatted(content):
    pattern = re.compile(
        r'<!--\s*wp:preformatted\b[^>]*-->'
        r'\s*'
        r'<pre\b([^>]*)>(.*?)</pre>'
        r'\s*'
        r'<!--\s*/wp:preformatted\s*-->',
        re.IGNORECASE | re.DOTALL
    )
    count = 0
    def replace(match):
        nonlocal count
        attrs = match.group(1)
        code = match.group(2)
        if is_enlighter(attrs) or is_enlighter(code):
            return match.group(0)
        count += 1
        return make_enlighter_block(code)
    return pattern.sub(replace, content), count

# ============================================================
# 转换 wp:code
# ============================================================
def convert_code_block(content):
    pattern = re.compile(
        r'<!--\s*wp:code\b[^>]*-->'
        r'\s*'
        r'<pre\b([^>]*)>(.*?)</pre>'
        r'\s*'
        r'<!--\s*/wp:code\s*-->',
        re.IGNORECASE | re.DOTALL
    )
    count = 0
    def replace(match):
        nonlocal count
        attrs = match.group(1)
        inner = match.group(2)
        if is_enlighter(attrs) or is_enlighter(inner):
            return match.group(0)
        code = inner
        # 去掉外层 <code>
        m = re.fullmatch(
            r'\s*<code\b[^>]*>(.*?)</code>\s*',
            inner,
            re.IGNORECASE | re.DOTALL
        )
        if m:
            code = m.group(1)
        count += 1
        return make_enlighter_block(code)
    return pattern.sub(replace, content), count

# ============================================================
# 转换普通 <code>
# ============================================================
def convert_inline_code(content):
    pattern = re.compile(
        r'<code\b([^>]*)>(.*?)</code>',
        re.IGNORECASE | re.DOTALL
    )
    count = 0
    def replace(match):
        nonlocal count
        attrs = match.group(1)
        code = match.group(2)
        attrs_lower = attrs.lower()
        if (
            'enlighterjsraw' in attrs_lower
            or 'data-enlighter-language' in attrs_lower
        ):
            return match.group(0)
        count += 1
        return (
            '<code '
            'data-enlighter-language="generic" '
            'class="EnlighterJSRAW">'
            + code +
            '</code>'
        )
    return pattern.sub(replace, content), count

# ============================================================
# 一篇文章转换
# ============================================================
def convert_content(content):
    original = content
    preformatted_count = 0
    codeblock_count = 0
    inline_count = 0
    # 顺序不能改变
    #
    # 先处理大代码块
    # 再处理普通 code
    content, n = convert_preformatted(content)
    preformatted_count += n
    content, n = convert_code_block(content)
    codeblock_count += n
    content, n = convert_inline_code(content)
    inline_count += n
    return (
        content,
        content != original,
        preformatted_count,
        codeblock_count,
        inline_count
    )

# ============================================================
# 解析一行 INSERT 中的 VALUES
# ============================================================
def parse_values_rows(values_text):
    rows = []
    i = 0
    length = len(values_text)
    while i < length:
        # 找到 (
        while i < length and values_text[i] != '(':
            i += 1
        if i >= length:
            break
        i += 1
        fields = []
        while i < length:
            while i < length and values_text[i].isspace():
                i += 1
            if i >= length:
                break
            # NULL
            if values_text[i:i + 4].upper() == 'NULL':
                fields.append({
                    'type': 'null',
                    'value': None
                })
                i += 4
            # 字符串
            elif values_text[i] == "'":
                value, i = parse_sql_string(
                    values_text,
                    i
                )
                fields.append({
                    'type': 'string',
                    'value': value
                })
            # 数字 / 常量
            else:
                start = i
                while i < length:
                    if values_text[i] in ',)':
                        break
                    i += 1
                raw = values_text[start:i].strip()
                fields.append({
                    'type': 'raw',
                    'value': raw
                })
            while i < length and values_text[i].isspace():
                i += 1
            if i < length and values_text[i] == ',':
                i += 1
                continue
            if i < length and values_text[i] == ')':
                i += 1
                rows.append(fields)
                break
            raise ValueError(
                f"VALUES 解析异常，位置 {i}"
            )
    return rows

# ============================================================
# 字段列表
# ============================================================
def parse_insert_columns(statement, table_name):
    pattern = re.compile(
        r'INSERT\s+INTO\s+'
        r'[`"]?'
        + re.escape(table_name) +
        r'[`"]?'
        r'\s*\((.*?)\)'
        r'\s*VALUES',
        re.IGNORECASE | re.DOTALL
    )
    match = pattern.search(statement)
    if not match:
        return None
    fields = []
    for item in match.group(1).split(','):
        item = item.strip()
        item = item.strip('`')
        item = item.strip('"')
        fields.append(item)
    return fields

# ============================================================
# 找 VALUES 开始位置
# ============================================================
def find_values_start(statement):
    match = re.search(
        r'\bVALUES\b',
        statement,
        re.IGNORECASE
    )
    if not match:
        return None
    return match.end()

# ============================================================
# 处理 INSERT
# ============================================================
def process_insert(
    statement,
    table_name,
    table_columns,
    stats
):
    # --------------------------------------------------------
    # 判断是不是 wp_posts
    # --------------------------------------------------------
    if not re.search(
        r'INSERT\s+INTO\s+'
        r'[`"]?'
        + re.escape(table_name) +
        r'[`"]?',
        statement,
        re.IGNORECASE
    ):
        return statement
    # --------------------------------------------------------
    # 获取字段
    #
    # 如果 INSERT 自带字段：
    #
    # INSERT INTO wp_posts (`ID`, `post_content`, ...)
    #
    # 就使用 INSERT 自己的字段。
    #
    # 如果没有：
    #
    # INSERT INTO wp_posts VALUES (...)
    #
    # 就使用 CREATE TABLE 的字段顺序。
    # --------------------------------------------------------
    insert_columns = parse_insert_columns(
        statement,
        table_name
    )
    if insert_columns:
        columns = insert_columns
    else:
        columns = table_columns
    if not columns:
        return statement
    columns_lower = [
        x.lower()
        for x in columns
    ]
    if 'post_content' not in columns_lower:
        return statement
    content_index = columns_lower.index(
        'post_content'
    )
    post_type_index = None
    if 'post_type' in columns_lower:
        post_type_index = columns_lower.index(
            'post_type'
        )
    # --------------------------------------------------------
    # VALUES
    # --------------------------------------------------------
    values_start = find_values_start(statement)
    if values_start is None:
        return statement
    prefix = statement[:values_start]
    values_part = statement[values_start:]
    # 保存分号
    suffix = ''
    stripped = values_part.rstrip()
    if stripped.endswith(';'):
        suffix = ';'
        values_part = stripped[:-1]
    # --------------------------------------------------------
    # 解析 VALUES
    # --------------------------------------------------------
    try:
        rows = parse_values_rows(
            values_part
        )
    except Exception as e:
        print()
        print("⚠️ INSERT 解析失败，已跳过：")
        print(str(e))
        return statement
    new_rows = []
    for row in rows:
        # ----------------------------------------------------
        # 字段数量不匹配
        # ----------------------------------------------------
        if len(row) != len(columns):
            new_rows.append(row)
            continue
        # ----------------------------------------------------
        # 只处理文章
        # ----------------------------------------------------
        if ONLY_POST and post_type_index is not None:
            pt = row[post_type_index]
            if (
                pt['type'] != 'string'
                or pt['value'] != 'post'
            ):
                new_rows.append(row)
                continue
        stats['posts_scanned'] += 1
        # ----------------------------------------------------
        # post_content
        # ----------------------------------------------------
        content_field = row[content_index]
        if content_field['type'] != 'string':
            new_rows.append(row)
            continue
        old_content = content_field['value']
        (
            new_content,
            changed,
            pre_count,
            block_count,
            inline_count
        ) = convert_content(
            old_content
        )
        stats['preformatted'] += pre_count
        stats['codeblock'] += block_count
        stats['inline'] += inline_count
        if changed:
            stats['posts_changed'] += 1
            row[content_index] = {
                'type': 'string',
                'value': new_content
            }
        new_rows.append(row)
    # --------------------------------------------------------
    # 重新生成
    # --------------------------------------------------------
    output_rows = []
    for row in new_rows:
        fields = []
        for field in row:
            if field['type'] == 'null':
                fields.append('NULL')
            elif field['type'] == 'string':
                fields.append(
                    sql_escape_string(
                        field['value']
                    )
                )
            else:
                fields.append(
                    field['value']
                )
        output_rows.append(
            '(' +
            ','.join(fields) +
            ')'
        )
    # MySQL extended INSERT：
    #
    # (...),
    # (...),
    # (...);
    return (
        prefix +
        ',\n'.join(output_rows) +
        suffix
    )

# ============================================================
# SQL 语句拆分
# ============================================================
def split_sql_statements(sql_text):
    statements = []
    start = 0
    i = 0
    in_string = False
    escaped = False
    length = len(sql_text)
    while i < length:
        ch = sql_text[i]
        if in_string:
            if escaped:
                escaped = False
            elif ch == '\\':
                escaped = True
            elif ch == "'":
                # SQL 中：
                #
                # ''
                #
                # 是一个单引号
                if (
                    i + 1 < length
                    and sql_text[i + 1] == "'"
                ):
                    i += 1
                else:
                    in_string = False
        else:
            if ch == "'":
                in_string = True
            elif ch == ';':
                statements.append(
                    sql_text[start:i + 1]
                )
                start = i + 1
        i += 1
    if start < length:
        statements.append(
            sql_text[start:]
        )
    return statements

# ============================================================
# 输出文件
# ============================================================
def make_output_paths(sql_file):
    folder = os.path.dirname(
        os.path.abspath(sql_file)
    )
    filename = os.path.basename(sql_file)
    name, ext = os.path.splitext(filename)
    timestamp = datetime.datetime.now().strftime(
        "%Y%m%d_%H%M%S"
    )
    backup_file = os.path.join(
        folder,
        f"{name}_原始备份_{timestamp}{ext}"
    )
    output_file = os.path.join(
        folder,
        f"{name}_enlighter_{timestamp}{ext}"
    )
    return backup_file, output_file

# ============================================================
# 主转换
# ============================================================
def convert_sql(sql_file):
    print()
    print("=" * 60)
    print("正在读取 SQL 文件……")
    print("=" * 60)
    sql_file = os.path.abspath(sql_file)
    if not os.path.isfile(sql_file):
        raise RuntimeError(
            "选择的路径不是文件：\n"
            + sql_file
        )
    with open(
        sql_file,
        'r',
        encoding='utf-8-sig',
        errors='replace'
    ) as f:
        sql_text = f.read()
    print(
        f"SQL 大小："
        f"{len(sql_text) / 1024 / 1024:.2f} MB"
    )
    # --------------------------------------------------------
    # 找 posts
    # --------------------------------------------------------
    table_name = find_posts_table(
        sql_text
    )
    if not table_name:
        raise RuntimeError(
            "没有找到 WordPress *_posts 表。"
        )
    print(
        f"发现 WordPress 文章表："
        f"{table_name}"
    )
    # --------------------------------------------------------
    # 找字段
    # --------------------------------------------------------
    table_columns = find_posts_columns(
        sql_text,
        table_name
    )
    print(
        f"发现字段数量："
        f"{len(table_columns)}"
    )
    if table_columns:
        if 'post_content' in table_columns:
            print(
                "post_content 字段："
                f"第 {table_columns.index('post_content') + 1} 列"
            )
        else:
            raise RuntimeError(
                "没有在 wp_posts 中找到 post_content。"
            )
    else:
        raise RuntimeError(
            "无法从 CREATE TABLE 中读取 wp_posts 字段。"
        )
    # --------------------------------------------------------
    # 统计
    # --------------------------------------------------------
    stats = {
        'posts_scanned': 0,
        'posts_changed': 0,
        'preformatted': 0,
        'codeblock': 0,
        'inline': 0
    }
    # --------------------------------------------------------
    # SQL 拆分
    # --------------------------------------------------------
    print()
    print("正在解析 SQL……")
    statements = split_sql_statements(
        sql_text
    )
    print(
        f"SQL 语句数量："
        f"{len(statements)}"
    )
    output = []
    total = len(statements)
    for index, statement in enumerate(
        statements,
        1
    ):
        if re.search(
            r'INSERT\s+INTO\s+'
            r'[`"]?'
            + re.escape(table_name) +
            r'[`"]?',
            statement,
            re.IGNORECASE
        ):
            statement = process_insert(
                statement,
                table_name,
                table_columns,
                stats
            )
        output.append(statement)
        if (
            index % 50 == 0
            or index == total
        ):
            print(
                f"\r处理 SQL："
                f"{index}/{total}",
                end='',
                flush=True
            )
    print()
    # --------------------------------------------------------
    # 输出
    # --------------------------------------------------------
    backup_file, output_file = (
        make_output_paths(sql_file)
    )
    print()
    print("正在创建原始备份……")
    shutil.copy2(
        sql_file,
        backup_file
    )
    print(
        f"原始备份：\n"
        f"{backup_file}"
    )
    print()
    print("正在写入转换后的 SQL……")
    with open(
        output_file,
        'w',
        encoding='utf-8',
        newline=''
    ) as f:
        f.write(''.join(output))
    # --------------------------------------------------------
    # 结果
    # --------------------------------------------------------
    print()
    print("=" * 60)
    print("转换完成")
    print("=" * 60)
    print(
        f"扫描文章："
        f"{stats['posts_scanned']}"
    )
    print(
        f"修改文章："
        f"{stats['posts_changed']}"
    )
    print()
    print(
        f"wp:preformatted："
        f"{stats['preformatted']}"
    )
    print(
        f"wp:code        ："
        f"{stats['codeblock']}"
    )
    print(
        f"普通 <code>    ："
        f"{stats['inline']}"
    )
    print()
    print(
        "原始备份："
    )
    print(
        backup_file
    )
    print()
    print(
        "转换结果："
    )
    print(
        output_file
    )
    print("=" * 60)
    return output_file

# ============================================================
# 主程序
# ============================================================
def main():
    print()
    print("=" * 60)
    print(" WordPress EnlighterJS 本地 SQL 转换工具")
    print("=" * 60)
    print()
    print("本工具：")
    print("✓ 不连接线上 WordPress")
    print("✓ 不连接 MySQL")
    print("✓ 不修改原 SQL")
    print("✓ 本地生成新的 SQL")
    print()
    sql_file = choose_sql_file()
    if not sql_file:
        print(
            "没有选择 SQL 文件。"
        )
        return
    print()
    print(
        "选择的 SQL 文件："
    )
    print(
        sql_file
    )
    print()
    # --------------------------------------------------------
    # 特别检查路径
    # --------------------------------------------------------
    if not os.path.isfile(sql_file):
        print(
            "❌ 错误：选择的路径不是普通文件。"
        )
        input(
            "\n按回车退出……"
        )
        return
    # --------------------------------------------------------
    # 确认
    # --------------------------------------------------------
    answer = input(
        "请输入 Y 开始转换，其他键退出："
    ).strip().lower()
    if answer != 'y':
        print(
            "已取消。"
        )
        return
    try:
        convert_sql(
            sql_file
        )
    except Exception as e:
        print()
        print("=" * 60)
        print("❌ 转换失败")
        print("=" * 60)
        print(
            str(e)
        )
        print()
        input(
            "按回车退出……"
        )
        return
    print()
    print("=" * 60)
    print("处理结束")
    print("=" * 60)
    print()
    print(
        "请先使用转换后的 SQL 导入测试数据库，"
    )
    print(
        "确认文章显示正常后，再用于正式数据库。"
    )
    print()
    input(
        "按回车退出……"
    )

if __name__ == '__main__':
    main()