Add migration script and work on tests

2026-05-17 19:15:55 +02:00
parent 81120547ae
commit e23f8904ef
12 changed files with 2300 additions and 2658 deletions
@@ -0,0 +1,108 @@
+import sqlite3
+from datetime import datetime, timedelta
+
+SOURCE_DB = 'backend/data/upload.db'
+DEST_DB = 'backend/data/sleep.db'
+
+def migrate():
+    try:
+        src_conn = sqlite3.connect(SOURCE_DB)
+        src_cur = src_conn.cursor()
+
+        dest_conn = sqlite3.connect(DEST_DB)
+        dest_cur = dest_conn.cursor()
+
+        print(f"Connecting to {SOURCE_DB} and {DEST_DB}...")
+
+        # 1. Truncate the destination table
+        print("Truncating sleep_session table...")
+        dest_cur.execute("DELETE FROM sleep_session")
+        dest_conn.commit()
+
+        # Fetch all records from the Sleep table
+        print("Fetching records from source...")
+        src_cur.execute('SELECT "Time", "Duration", "Note" FROM "Sleep"')
+        rows = src_cur.fetchall()
+
+        if not rows:
+            print("No records found in source Sleep table.")
+            return
+
+        migrated_count = 0
+        error_count = 0
+        now = datetime.now()
+
+        for row in rows:
+            timestamp_seconds, duration_minutes, note = row
+            
+            # Convert Unix timestamp to datetime object
+            try:
+                start_time = datetime.fromtimestamp(timestamp_seconds)
+            except Exception as e:
+                print(f"Error converting timestamp {timestamp_seconds}: {e}")
+                error_count += 1
+                continue
+
+            # Calculate end time based on duration (NOW ASSUMED MINUTES)
+            end_time = start_time + timedelta(minutes=duration_minutes)
+
+            # VERIFICATION 1: Check if conversion matches source duration
+            actual_duration_minutes = (end_time - start_time).total_seconds() / 60.0
+            if abs(actual_duration_minutes - duration_minutes) > 0.1:
+                print(f"CRITICAL ERROR: Duration mismatch for timestamp {timestamp_seconds}. Source (min): {duration_minutes}, Calculated (min): {actual_duration_minutes}")
+                error_count += 1
+                continue
+
+            notes = note if note is not None else ""
+
+            # Insert the new record
+            try:
+                dest_cur.execute(
+                    """
+                    INSERT INTO sleep_session (start_time, end_time, notes, created_at, updated_at)
+                    VALUES (?, ?, ?, ?, ?)
+                    """,
+                    (
+                        start_time.strftime('%Y-%m-%d %H:%M:%S.%f'),
+                        end_time.strftime('%Y-%m-%d %H:%M:%S.%f'),
+                        notes,
+                        now.strftime('%Y-%m-%d %H:%M:%S'),
+                        now.strftime('%Y-%m-%d %H:%M:%S')
+                    )
+                )
+                migrated_count += 1
+            except Exception as e:
+                print(f"Failed to insert record for timestamp {timestamp_seconds}: {e}")
+                error_count += 1
+
+        dest_conn.commit()
+        print(f"Migration complete.")
+        print(f"Successfully migrated: {migrated_count}")
+        print(f"Errors encountered: {error_count}")
+
+        # 2. FINAL VERIFICATION
+        print("Running final verification...")
+        dest_cur.execute("SELECT start_time, end_time FROM sleep_session")
+        dest_rows = dest_cur.fetchall()
+
+        if len(dest_rows) != migrated_count:
+            print(f"VERIFICATION FAILED: Count mismatch. Source attempted: {migrated_count}, Destination count: {len(dest_rows)}")
+        else:
+            print("VERIFICATION SUCCESS: Count matches.")
+
+        # Sample duration verification in destination
+        print("Verifying sample durations from destination...")
+        for i, (st_str, et_str) in enumerate(dest_rows[:5]):
+            st = datetime.strptime(st_str, '%Y-%m-%d %H:%M:%S.%f')
+            et = datetime.strptime(et_str, '%Y-%m-%d %H:%M:%S.%f')
+            duration_sec = (et - st).total_seconds()
+            print(f"  Sample {i+1}: Start={st_str}, End={et_str}, Duration={duration_sec/60.0:.1f}m")
+
+    except Exception as e:
+        print(f"An error occurred during migration: {e}")
+    finally:
+        if 'src_conn' in locals(): src_conn.close()
+        if 'dest_conn' in locals(): dest_conn.close()
+
+if __name__ == "__main__":
+    migrate()