diff --git a/scripts/migrate_sqlite_to_mysql.py b/scripts/migrate_sqlite_to_mysql.py new file mode 100755 index 0000000..7044261 --- /dev/null +++ b/scripts/migrate_sqlite_to_mysql.py @@ -0,0 +1,77 @@ +#!/usr/bin/env python3 +"""SQLite 到 MySQL 数据迁移脚本""" + +import sqlite3 +import os +from pathlib import Path + +import mysql.connector + + +def get_sqlite_path() -> Path: + """获取 SQLite 数据库路径""" + env_path = os.environ.get("VITALS_DB_PATH") + if env_path: + return Path(env_path) + return Path.home() / ".vitals" / "vitals.db" + + +def migrate(): + # SQLite 连接 + sqlite_path = get_sqlite_path() + if not sqlite_path.exists(): + print(f"SQLite 数据库不存在: {sqlite_path}") + return + + sqlite_conn = sqlite3.connect(sqlite_path) + sqlite_conn.row_factory = sqlite3.Row + sqlite_cursor = sqlite_conn.cursor() + + # MySQL 连接 + mysql_conn = mysql.connector.connect( + host=os.environ.get("MYSQL_HOST", "localhost"), + port=int(os.environ.get("MYSQL_PORT", "3306")), + user=os.environ.get("MYSQL_USER", "vitals"), + password=os.environ.get("MYSQL_PASSWORD", ""), + database=os.environ.get("MYSQL_DATABASE", "vitals"), + ) + mysql_cursor = mysql_conn.cursor() + + tables = ["users", "exercise", "meal", "sleep", "weight", "reading", "invites", "config"] + + for table in tables: + print(f"迁移表: {table}") + try: + sqlite_cursor.execute(f"SELECT * FROM {table}") + rows = sqlite_cursor.fetchall() + + if not rows: + print(f" - 无数据") + continue + + columns = [desc[0] for desc in sqlite_cursor.description] + placeholders = ", ".join(["%s"] * len(columns)) + column_names = ", ".join([f"`{c}`" for c in columns]) + + insert_sql = f"INSERT INTO {table} ({column_names}) VALUES ({placeholders})" + + for row in rows: + values = tuple(row) + try: + mysql_cursor.execute(insert_sql, values) + except mysql.connector.IntegrityError as e: + print(f" - 跳过重复记录: {e}") + + mysql_conn.commit() + print(f" - 迁移 {len(rows)} 条记录") + + except sqlite3.OperationalError as e: + print(f" - 表不存在或错误: {e}") + + sqlite_conn.close() + mysql_conn.close() + print("迁移完成!") + + +if __name__ == "__main__": + migrate()