WeReadBoard/migrate_to_mysql.py

#!/usr/bin/env python3
"""
SQLite -> MySQL 数据迁移脚本
将 weread.db (SQLite) 中的全部数据迁移到 MySQL weread_dashboard 数据库
"""

import os
import sqlite3
from datetime import datetime

import pymysql

# ===== 连接配置 =====
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",
}

# ===== 表迁移映射 =====
# (sqlite_table, mysql_table, columns, timestamp_cols)
# timestamp_cols: SQLite 中存为 int(秒级时间戳)，MySQL 中为 TIMESTAMP 的列
TABLE_MAPPING = [
    ("books", "books",
     ["book_id", "title", "author", "cover", "category", "finish_reading",
      "read_update_time", "secret", "is_top", "update_time", "deep_link"],
     []),

    ("reading_stats", "reading_stats",
     ["mode", "period", "total_read_time", "read_days", "day_average_read_time",
      "regist_time", "raw_json"],
     []),  # MySQL created_at 用默认值

    ("read_times", "read_times",
     ["mode", "period", "bucket_timestamp", "duration"],
     []),

    ("read_longest", "read_longest",
     ["mode", "period", "book_id", "title", "author", "cover", "read_time", "tags"],
     []),

    ("prefer_categories", "prefer_categories",
     ["mode", "period", "category_id", "category_title", "parent_category_title",
      "val", "reading_time", "reading_count"],
     []),

    ("prefer_authors", "prefer_authors",
     ["mode", "period", "author_id", "name", "count", "read_time"],
     []),

    ("prefer_time", "prefer_time",
     ["mode", "period", "hour_index", "duration"],
     []),

    ("notebooks", "notebooks",
     ["book_id", "title", "author", "cover", "review_count", "note_count",
      "bookmark_count", "reading_progress", "marked_status", "sort"],
     []),

    ("read_stat_items", "read_stat_items",
     ["mode", "period", "stat_name", "counts", "scheme"],
     []),

    ("archives", "archives",
     ["name", "book_ids"],
     []),

    ("fetch_meta", "fetch_meta",
     ["key", "value"],
     []),

    ("book_details", "book_details",
     ["book_id", "title", "author", "translator", "cover", "intro", "category",
      "publisher", "publish_time", "isbn", "word_count", "new_rating", "new_rating_count"],
     ["fetched_at"]),

    ("book_progress", "book_progress",
     ["book_id", "progress", "chapter_uid", "chapter_offset", "update_time",
      "record_reading_time", "finish_time", "is_start_reading"],
     ["fetched_at"]),

    ("book_highlights", "book_highlights",
     ["book_id", "bookmark_id", "chapter_uid", "mark_text", "create_time",
      "type", "range_text", "color_style"],
     ["fetched_at"]),

    ("book_chapters", "book_chapters",
     ["book_id", "chapter_uid", "chapter_idx", "title"],
     []),

    ("book_reviews", "book_reviews",
     ["book_id", "review_id", "content", "create_time", "star",
      "chapter_name", "is_finish"],
     ["fetched_at"]),

    ("recommended_books", "recommended_books",
     ["book_id", "title", "author", "cover", "intro", "category", "reason",
      "reading_count", "new_rating", "new_rating_count", "rating_title"],
     ["fetched_at"]),

    ("search_results", "search_results",
     ["keyword", "book_id", "title", "author", "cover", "intro", "publisher",
      "category", "reading_count", "new_rating", "new_rating_count",
      "rating_title", "search_idx"],
     ["fetched_at"]),
]


def convert_timestamp(value):
    """将 SQLite 中的 int 时间戳转为 MySQL datetime 字符串"""
    if value is None or value == 0:
        return None
    try:
        return datetime.fromtimestamp(int(value)).strftime("%Y-%m-%d %H:%M:%S")
    except (ValueError, OSError, OverflowError):
        return None


def migrate():
    """执行迁移"""
    sqlite_conn = sqlite3.connect(SQLITE_PATH)
    sqlite_conn.row_factory = sqlite3.Row
    sc = sqlite_conn.cursor()

    mysql_conn = pymysql.connect(**MYSQL_CONFIG)
    mc = mysql_conn.cursor()

    total_rows = 0

    for sqlite_table, mysql_table, columns, ts_cols in TABLE_MAPPING:
        # 读取 SQLite 数据
        col_list = ", ".join(columns)
        sc.execute(f"SELECT {col_list} FROM \"{sqlite_table}\"")
        rows = sc.fetchall()

        if not rows:
            print(f"  {sqlite_table:30s}  0 行 (跳过)")
            continue

        # 构建 INSERT 语句
        placeholders = ", ".join(["%s"] * len(columns))
        col_names = ", ".join(f"`{c}`" for c in columns)
        sql = f"INSERT INTO `{mysql_table}` ({col_names}) VALUES ({placeholders})"

        # 转换数据
        batch = []
        for row in rows:
            values = []
            for i, col in enumerate(columns):
                val = row[i]
                if col in ts_cols:
                    val = convert_timestamp(val)
                values.append(val)
            batch.append(tuple(values))

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

        count = len(batch)
        total_rows += count
        print(f"  {sqlite_table:30s}  {count:>6} 行  ✓")

    # 验证
    print(f"\n{'='*50}")
    print(f"迁移完成！共迁移 {total_rows} 行数据")
    print(f"{'='*50}")

    print("\nMySQL 各表数据量验证：")
    mc.execute("SHOW TABLES")
    tables = [t[0] for t in mc.fetchall()]
    for t in tables:
        mc.execute(f"SELECT COUNT(*) FROM `{t}`")
        cnt = mc.fetchone()[0]
        print(f"  {t:30s}  {cnt:>6} 行")

    sqlite_conn.close()
    mysql_conn.close()


if __name__ == "__main__":
    print("开始迁移 SQLite -> MySQL ...\n")
    migrate()