feat: 添加 SQLite 到 MySQL 数据迁移脚本

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()