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

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

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

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

# 数据库配置
MYSQL_CONFIG = {
    '；host'；: '；localhost'；, '；port'；: 3306, '；user'；: '；yys'；, 
    '；password'；: '；123456'；, '；db'；: '；yys'；, '；charset'；: '；utf8mb4'；
}
OUTPUT_FILE = '；articles_import.sql'；

def esc(val):
    if val is None: return '；NULL'；
    # 核心技巧：将换行符转为 SQLite 函数，防止 SQL 语句在文本中被断开
    s = str(val).replace("；'；"；, "；'；'；"；)
    s = s.replace('；\r\n'；, "；'； || char(10) || '；"；).replace('；\n'；, "；'； || char(10) || '；"；)
    return "；'；"； + s + "；'；"；

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 = ["；PRAGMA journal_mode=WAL；"；]
    cols = "；id, avatar, title, content, total_views, created, updated, author_id, category_id, previous_content, is_encrypted"；
    
    for r in rows:
        vals = f"；{r['；id'；]}, {esc(r['；avatar'；])}, {esc(r['；title'；])}, {esc(r['；content'；])}, {int(r['；total_views'；] or 0)}, {dt(r['；created'；])}, {dt(r['；updated'；])}, {int(r['；author_id'；] or 1)}, {r['；category_id'；] if r['；category_id'；] else '；NULL'；}, {esc(r['；previous_content'；])}, {1 if r['；is_encrypted'；] else 0}"；
        lines.append(f"；INSERT OR REPLACE INTO tb_article ({cols}) VALUES ({vals})；"；)

    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 文件，每 50 条一个文件
mkdir -p parts
split -l 50 articles_import.sql parts/part_

echo "；=== 开始远程同步 ==="；
for f in parts/part_*； do
  echo "；正在上传 $f ..."；
  npx wrangler d1 execute blog --remote --file="；$f"；
done
rm -rf parts
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 来完成全流程。