从 MySQL 搬迁到本地/远程 D1处理逻辑

这套方案涵盖了数据从 MySQL 搬迁到本地/远程 D1 的全自动处理逻辑。

第一步：更新核心导出脚本 (migrate_articles.py)

使用这个版本，它能完美处理文章中的复杂换行符（如代码块、长文章），确保生成的 SQL 不会断行。

#!/usr/bin/env python3
import pymysql, sys
from datetime import datetime

MYSQL_CONFIG = {
    'host': 'localhost', 'port': 3306, 'user': 'yys',
    'password': '123456', 'db': 'yys', 'charset': 'utf8mb4'
}

OUTPUT_FILE = 'articles_import.sql'

# 使用 Hex 编码安全转换文本
# 此方法能完全保留换行、分号、单引号等特殊字符，且生成的目标 SQL 语句天生单行，避免 awk 分割错误
def esc_hex(val):
    if val is None:
        return 'NULL'
    s = str(val)
    # 统一换行（保留换行意图，但规范化为 \n）
    s = s.replace('\r\n', '\n').replace('\r', '\n')
    
    # 转换为 utf-8 的 hex 字符串
    hex_str = s.encode('utf-8').hex()
    
    # 使用 SQLite 的 CAST 语法将其转回 TEXT
    return f"CAST(X'{hex_str}' AS TEXT)"

# 拆分 content 文本，避免 D1 的单条 SQL 长度限制
def split_chunks(text, size=8000):
    return [text[i:i+size] for i in range(0, len(text), size)]

def dt(val):
    if val is None:
        return 'NULL'
    if isinstance(val, datetime):
        return f"'{val.strftime('%Y-%m-%d %H:%M:%S')}'"
    return "'" + str(val).split('.')[0] + "'"

def main():
    try:
        conn = pymysql.connect(**MYSQL_CONFIG)
        cur = conn.cursor(pymysql.cursors.DictCursor)
        cur.execute(
            "SELECT id, avatar, title, content, total_views, created, "
            "updated, author_id, category_id, previous_content, is_encrypted "
            "FROM tb_article ORDER BY id"
        )
        rows = cur.fetchall()
        conn.close()
    except Exception as e:
        print(f"❌ MySQL连接失败: {e}")
        sys.exit(1)

    lines = []

    for r in rows:
        id = r["id"]
        # 对文本类型的字段统一采用 esc_hex 处理
        avatar = esc_hex(r["avatar"])
        title = esc_hex(r["title"])
        content = r["content"] or ""
        prev = esc_hex(r["previous_content"])
        
        total = int(r["total_views"] or 0)
        created = dt(r["created"])
        updated = dt(r["updated"])
        author = int(r["author_id"] or 1)
        category = r["category_id"] if r["category_id"] else "NULL"
        encrypted = 1 if r["is_encrypted"] else 0

        # 1. 插入一条初始数据（content 留空）
        lines.append(
            f"INSERT OR REPLACE INTO tb_article "
            f"(id, avatar, title, content, total_views, created, updated, "
            f"author_id, category_id, previous_content, is_encrypted) "
            f"VALUES ({id}, {avatar}, {title}, '', {total}, {created}, "
            f"{updated}, {author}, {category}, {prev}, {encrypted});"
        )

        # 2. 分段追加 content 
        chunks = split_chunks(content)
        for chunk in chunks:
            # 同样使用 esc_hex 处理分段
            lines.append(
                f"UPDATE tb_article SET content = content || {esc_hex(chunk)} "
                f"WHERE id = {id};"
            )

    with open(OUTPUT_FILE, 'w', encoding='utf-8', newline='\n') as f:
        f.write("\n".join(lines))

    print(f"✅ SQL 已导出: {OUTPUT_FILE}（共 {len(rows)} 篇文章）")

if __name__ == '__main__':
    main()

---

第二步：一键导入工作流 (import_all.sh) （不需要省略）

将此脚本保存为 import_all.sh，赋予权限 chmod +x import_all.sh。它负责清理旧数据、重置 ID 序列、并执行导入。

#!/bin/bash
# 配置路径
DB_PATH=".wrangler/state/v3/d1/miniflare-D1DatabaseObject/cbead4c0cbcc98a701e95d7be92a1a77970bb0062aa5d52916341ad4d8399224.sqlite"
SQL_FILE="articles_import.sql"

echo "=== 1. 导出 MySQL 数据 ==="
python3 migrate_articles.py

echo "=== 2. 清理并重置本地 D1 ==="
# 清空数据并重置 SQLite 自增序列，确保 ID 从 1 开始
sqlite3 "$DB_PATH" "DELETE FROM tb_article; DELETE FROM sqlite_sequence WHERE name='tb_article';"

echo "=== 3. 导入 SQL 到本地 ==="
sqlite3 "$DB_PATH" < "$SQL_FILE"

echo "=== 4. 验证 ==="
COUNT=$(sqlite3 "$DB_PATH" "SELECT count(*) FROM tb_article;")
echo "成功导入 $COUNT 篇文章。"
sqlite3 "$DB_PATH" "SELECT id, title FROM tb_article ORDER BY id ASC LIMIT 5;"

---

第三步：远程部署同步脚本 (sync_remote.sh)

由于 Cloudflare 远程限制，导入需要切片：

#!/bin/bash

SQL_FILE="/home/yys/test/articles_import.sql"
PART_DIR="parts"
RETRY=10

rm -rf "$PART_DIR"
mkdir -p "$PART_DIR"

echo "=== 按 200 条 SQL 切割文件 ==="

awk -v dir="$PART_DIR" '
  BEGIN { count = 0; fileIndex = 1 }
  /;$/ {
    print >> sprintf("%s/part_%05d.sql", dir, fileIndex)
    count++
    if (count >= 200) {
      fileIndex++
      count = 0
    }
    next
  }
  { print >> sprintf("%s/part_%05d.sql", dir, fileIndex) }
' "$SQL_FILE"

echo "=== 开始远程同步（支持断点续传） ==="

for f in $PART_DIR/*.sql; do
  echo "上传 $f ..."

  # 如果之前成功过，跳过
  if grep -q "$f" sync_success.log 2>/dev/null; then
    echo "跳过（之前已成功）: $f"
    continue
  fi

  # 自动重试
  for i in $(seq 1 $RETRY); do
    yes | wrangler d1 execute blog --remote --file="$f" && {
      echo "$f" >> sync_success.log
      echo "成功: $f"
      break
    }

    echo "⚠️ 第 $i 次失败，3 秒后重试..."
    sleep 3
  done
done

echo "=== 远程同步完成 ==="

---

总结操作指南

1. 准备环境：在 ~/node 目录下放入 migrate_articles.py, import_all.sh, sync_remote.sh。
2. 执行本地：运行 ./import_all.sh。这会更新你的本地 .sqlite 文件。如果此时在 npx wrangler dev 下看到数据不全，请刷新页面。
3. 执行远程：当你确认本地数据完美后，执行 ./sync_remote.sh，脚本会自动分块推送到 Cloudflare 线上 D1。

为什么这个流程完整？

• 兼容性：解决了 SQL 包含换行符会导致导入失败的问题 (char(10) 技巧)。
• 一致性：sqlite_sequence 的重置确保了 ID 逻辑与 MySQL 完全一致。
• 稳定性：远程分块 (split) 绕过了 Cloudflare API 对单次 SQL 请求的大小限制。

现在你可以直接通过运行 ./import_all.sh 来完成全流程。