迁移LeanCloud到TiDB

今天，突然受到了一封邮件，内容是 LeanCloud 宣布将于 2027 年 1 月关闭其数据库服务。
我点进去看了一下，发现是真的。
LeanCloud 关闭数据库服务通知
毕竟，我博客上评论区、统计阅读量之类的数据都放在了 LeanCloud 上，所以这件事对我来说还是挺重要的。

然后，我就上了 Waline 的官网，发现 Waline 支持 TiDB 作为后端数据库，而且看起来免费额度还挺大，教程也是第一个推荐这个，于是就选了它

创建 TiDB 数据库

第一步，当然是跟着官方教程走，创建数据库。不过，因为官方的文档实在是太旧了，UI界面什么的完全不一样，因此我还花了不少时间摸索了一下。总之，和官方的教程差异如下：

Chat2Query变成了 SQL Editor
链接数据库的时候选择 General选项就好了，里面的几个信息分别填入 Vercel 的 TIDB_HOST、TIDB_USER、TIDB_PASSWORD、TIDB_DB环境变量

导出 LeanCloud 数据

接下来，就是导出 LeanCloud 的数据了。
LeanCloud 提供了数据导出的功能，根据停运公告的教程，我过了 20 分钟左右就收到了包含我数据库所有内容的邮件

不过导出的数据是 JSONL 格式的，而 TiDB 需要的是 CSV (150MB 以下) 格式的，因此还需要进行一些转换。

我最终让 Copilot 帮我写了一个 Python 脚本来完成这个任务，代码如下：

import json
import csv

input_file = "input.jsonl"
output_file = "output.csv"

rows = []

# 读取 JSONL
with open(input_file, "r", encoding="utf-8") as f:
    for line in f:
        if line.strip():
            rows.append(json.loads(line))

# 自动收集所有字段作为表头
headers = set()
for r in rows:
    headers.update(r.keys())
headers = list(headers)

# 写入 CSV
with open(output_file, "w", encoding="utf-8", newline="") as f:
    writer = csv.DictWriter(f, fieldnames=headers)
    writer.writeheader()
    writer.writerows(rows)

print("Done!")

看起来还是挺不错的

导入旧数据到 TiDB

因为原有的 Leancloud 转换出来的 CSV 字段和 TiDB 的表结构并不完全匹配，所以还需要对 CSV 进行一些手动的修改，主要是删除一些多余的字段，和调整字段顺序。
我于是又让他帮我写了下面这个脚本：

import csv
import json
from datetime import datetime

def parse_parse_date(value):
    if not value:
        return r"\N"
    try:
        obj = json.loads(value.replace("'", '"'))
        if isinstance(obj, dict) and obj.get("__type") == "Date":
            iso = obj.get("iso")
            dt = datetime.fromisoformat(iso.replace("Z", "+00:00"))
            return dt.strftime("%Y-%m-%d %H:%M:%S")
    except:
        pass
    return value or r"\N"

def to_int_or_null(value):
    if value is None:
        print(0)
        return r"\N"
    value = str(value).strip()
    print(f"Checking value: '{value}'")
    if value.isdigit():
        print(1)
        return value
    return r"\N"

def to_str_or_null(value):
    if value is None:
        return r""
    v = str(value).strip()
    return v if v else r""

def nullify(value):
    """把空字符串、None、空白都变成 \\N"""
    if value is None:
        return r"\N"
    v = str(value).strip()
    return v if v else r"\N"


input_file = "output.csv"
output_file = "db_ready.csv"

target_fields = [
    "id", "user_id", "comment", "insertedAt", "ip", "link", "mail", "nick",
    "pid", "rid", "sticky", "status", "like", "ua", "url", "createdAt", "updatedAt"
]

# 读取所有行
rows = []
with open(input_file, "r", encoding="utf-8") as f:
    reader = csv.DictReader(f)
    for row in reader:
        rows.append(row)

# 建立 objectId → new_id 映射
objectId_to_newId = {row["objectId"]: i for i, row in enumerate(rows)}

# 生成新 CSV 数据
output_rows = []
for row in rows:
    new_row = {}

    # 新 id
    new_row["id"] = objectId_to_newId[row["objectId"]]

    # user_id
    new_row["user_id"] = to_int_or_null(row.get("user_id", ""))

    # 普通字段
    for field in ["comment", "link", "mail", "nick", "sticky", "like", "ua", "url"]:
        new_row[field] = nullify(row.get(field, ""))
  
    # status and ip use empty string as default (not \N)
    new_row["status"] = to_str_or_null(row.get("status", ""))
    new_row["ip"] = to_str_or_null(row.get("ip", ""))

    # 时间字段
    new_row["insertedAt"] = parse_parse_date(row.get("insertedAt", ""))
    new_row["createdAt"]  = parse_parse_date(row.get("createdAt", ""))
    new_row["updatedAt"]  = parse_parse_date(row.get("updatedAt", ""))

    # pid 映射
    old_pid = row.get("pid", "")
    new_row["pid"] = objectId_to_newId.get(old_pid, r"\N") if old_pid else r"\N"

    # rid 映射
    old_rid = row.get("rid", "")
    new_row["rid"] = objectId_to_newId.get(old_rid, r"\N") if old_rid else r"\N"

    output_rows.append(new_row)

# 写入 CSV
with open(output_file, "w", encoding="utf-8", newline="") as f:
    writer = csv.DictWriter(f, fieldnames=target_fields)
    writer.writeheader()
    writer.writerows(output_rows)

print("完成：所有空值已转换为 \\N，适用于 TiDB Lightning 导入。")