WeReadBoard/migrate_to_mysql.py

#!/usr/bin/env python3
"""
SQLite → MySQL 数据迁移脚本（完整版 v2.0）
将 weread.db (SQLite) 全部 31 张表数据迁移至 MySQL weread_dashboard 数据库。

处理逻辑：
  1. 自动读取 SQLite 每张表的列名，与 MySQL 对齐
  2. 时间戳文本列 (_dt, created_at, updated_at) 自动转换为 MySQL TIMESTAMP 格式
  3. 每张表先 TRUNCATE 再插入，确保数据一致性
"""

import os
import sqlite3
import pymysql
from datetime import datetime

# ===== 配置 =====
SQLITE_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "weread.db")

MYSQL_CONFIG = {
    "host": "127.0.0.1",
    "port": 3306,
    "user": "root",
    "password": "258963",
    "database": "weread_dashboard",
    "charset": "utf8mb4",
}

# 需要迁移的表名列表（按依赖顺序排列）
TABLES = [
    # 书架数据
    "books",
    "shelf_albums",
    "shelf_meta",
    "archives",
    # 阅读统计
    "reading_stats",
    "read_times",
    "read_longest",
    "read_stat_items",
    "prefer_categories",
    "prefer_authors",
    "prefer_publishers",
    "prefer_cps",
    "prefer_books",
    "prefer_time",
    "read_rank",
    "medals",
    "year_reports",
    # 笔记数据
    "notebooks",
    "book_highlights",
    "book_chapters",
    "book_reviews",
    "book_best_bookmarks",
    "book_underlines",
    "public_reviews",
    # 书籍详情
    "book_details",
    "book_progress",
    # 发现/推荐
    "recommended_books",
    "similar_books",
    "search_sessions",
    "search_results",
    # 系统
    "fetch_meta",
]


def convert_dt(val):
    """将 SQLite 中的时间文本转换为 MySQL TIMESTAMP 兼容格式"""
    if val is None or val == "":
        return None
    if isinstance(val, (int, float)):
        return None
    # SQLite datetime('now','localtime') 格式: "2026-07-17 10:00:00"
    # 已经是 MySQL 兼容格式，直接返回
    return str(val)


def migrate():
    """执行完整迁移"""
    # 连接 SQLite
    sqlite_conn = sqlite3.connect(SQLITE_PATH)
    sqlite_conn.row_factory = sqlite3.Row
    sc = sqlite_conn.cursor()

    # 连接 MySQL
    mysql_conn = pymysql.connect(**MYSQL_CONFIG)
    mc = mysql_conn.cursor()

    total_rows = 0
    migrated_tables = 0

    print("=" * 60)
    print("SQLite → MySQL 数据迁移 (v2.0)")
    print("=" * 60)

    for table in TABLES:
        # 检查 SQLite 表是否有数据
        sc.execute(f'SELECT COUNT(*) FROM "{table}"')
        row_count = sc.fetchone()[0]

        if row_count == 0:
            print(f"  {table:30s} 跳过 (无数据)")
            continue

        # 获取列名
        sc.execute(f'PRAGMA table_info("{table}")')
        columns = [col[1] for col in sc.fetchall()]

        # 清空 MySQL 表
        mc.execute(f"DELETE FROM `{table}`")

        # 读取 SQLite 数据
        sc.execute(f'SELECT {", ".join(f"\"{c}\"" for c in columns)} FROM "{table}"')
        rows = sc.fetchall()

        # 构建批量 INSERT
        placeholders = ", ".join(["%s"] * len(columns))
        col_list = ", ".join(f"`{c}`" for c in columns)
        insert_sql = f"INSERT INTO `{table}` ({col_list}) VALUES ({placeholders})"

        # 转换数据
        batch = []
        for row in rows:
            values = []
            for col in columns:
                val = row[col]
                values.append(val)
            batch.append(values)

        # 批量插入
        mc.executemany(insert_sql, batch)
        mysql_conn.commit()

        total_rows += row_count
        migrated_tables += 1
        print(f"  {table:30s} {row_count:>6} 行 ✓")

    # 更新 fetch_meta
    mc.execute(
        "INSERT INTO fetch_meta (`key`, `value`) VALUES ('mysql_migrated_at', %s) "
        "ON DUPLICATE KEY UPDATE `value` = %s",
        (datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
         datetime.now().strftime("%Y-%m-%d %H:%M:%S")),
    )
    mysql_conn.commit()

    print(f"\n{'=' * 60}")
    print(f"迁移完成! {migrated_tables} 张表, 共 {total_rows} 行数据")
    print(f"MySQL: {MYSQL_CONFIG['host']}:{MYSQL_CONFIG['port']}/{MYSQL_CONFIG['database']}")
    print(f"{'=' * 60}")

    # 数据校验
    print("\n数据校验:")
    mc.execute("SHOW TABLES")
    mysql_tables = [t[0] for t in mc.fetchall()]
    for t in mysql_tables:
        mc.execute(f"SELECT COUNT(*) FROM `{t}`")
        cnt = mc.fetchone()[0]
        if cnt > 0:
            print(f"  MySQL {t:30s} {cnt:>6} 行")

    sqlite_conn.close()
    mysql_conn.close()


if __name__ == "__main__":
    migrate()