#!/usr/bin/env python3 """ FastDesk Import Tool Migrates data from Lovable.app CSV exports to SQLite database Usage: python3 import.py links-export-*.csv items-export-*.csv [--db /path/to/fastdesk.db] """ import csv import json import sqlite3 import argparse import sys from pathlib import Path from datetime import datetime import uuid def parse_keywords(keywords_str): """Parse keywords from JSON array string format""" if not keywords_str or not keywords_str.strip(): return [] try: # Remove outer quotes if present keywords_str = keywords_str.strip().strip('"') # Parse as JSON array kw_list = json.loads(keywords_str) return [k.strip().lower() for k in kw_list if k.strip()] except (json.JSONDecodeError, TypeError): print(f" ⚠️ Could not parse keywords: {keywords_str[:50]}") return [] def import_data(links_csv, items_csv, db_path, skip_duplicates=True): """Import CSV exports to SQLite""" if not Path(links_csv).exists(): print(f"❌ Links CSV not found: {links_csv}") sys.exit(1) if not Path(items_csv).exists(): print(f"❌ Items CSV not found: {items_csv}") sys.exit(1) # Connect to database conn = sqlite3.connect(db_path) cursor = conn.cursor() # Initialize schema print("🔧 Initializing database schema...") schema_path = Path(__file__).parent / 'fastdesk-schema.sql' if not schema_path.exists(): print(f"❌ Schema file not found: {schema_path}") sys.exit(1) with open(schema_path, 'r') as f: cursor.executescript(f.read()) conn.commit() # Create default user print("👤 Creating default user (admin/password)...") default_user_id = str(uuid.uuid4()) try: cursor.execute(""" INSERT INTO users (id, username, password_hash) VALUES (?, ?, ?) """, (default_user_id, 'admin', 'change_me_immediately')) conn.commit() print(" ✅ Default user created (CHANGE PASSWORD IMMEDIATELY)") except sqlite3.IntegrityError: print(" ℹ️ Admin user already exists") cursor.execute("SELECT id FROM users WHERE username = ?", ('admin',)) default_user_id = cursor.fetchone()[0] # Import cards from links CSV print(f"\n📥 Importing from {Path(links_csv).name}...") with open(links_csv, 'r', encoding='utf-8') as f: reader = csv.DictReader(f, delimiter=';') links = list(reader) cards_inserted = 0 cards_skipped = 0 keywords_inserted = 0 for row in links: card_id = row.get('id', '').strip() if not card_id: cards_skipped += 1 continue try: cursor.execute(""" INSERT INTO cards (id, user_id, label, content, emoji, visibility, enabled, position, storage_path, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( card_id, row.get('user_id', default_user_id), row.get('label', '').strip()[:500], row.get('content', '').strip() if row.get('content') else None, row.get('emoji', '').strip() if row.get('emoji') else None, row.get('visibility', 'private'), 1 if row.get('enabled', 'true').lower() == 'true' else 0, row.get('position'), row.get('storage_path'), row.get('created_at'), row.get('updated_at') )) cards_inserted += 1 # Parse and insert keywords keywords = parse_keywords(row.get('keywords', '')) for kw in keywords: try: cursor.execute(""" INSERT INTO keywords (card_id, keyword) VALUES (?, ?) """, (card_id, kw)) keywords_inserted += 1 except sqlite3.IntegrityError: # Keyword already exists for this card pass except sqlite3.IntegrityError as e: if skip_duplicates: cards_skipped += 1 else: print(f" ❌ Duplicate card ID: {card_id}") raise conn.commit() print(f" ✅ Cards imported: {cards_inserted}") print(f" ⏭️ Cards skipped: {cards_skipped}") print(f" ✅ Keywords inserted: {keywords_inserted}") # Import items print(f"\n📥 Importing from {Path(items_csv).name}...") with open(items_csv, 'r', encoding='utf-8') as f: reader = csv.DictReader(f, delimiter=';') items = list(reader) items_inserted = 0 items_skipped = 0 media_refs = 0 for row in items: item_id = row.get('id', '').strip() card_id = row.get('card_id', '').strip() if not item_id or not card_id: items_skipped += 1 continue # Check if card exists cursor.execute("SELECT id FROM cards WHERE id = ?", (card_id,)) if not cursor.fetchone(): items_skipped += 1 continue try: cursor.execute(""" INSERT INTO items ( id, card_id, user_id, type, label, url, content, position, media_url, media_type, media_file_name, storage_path, thumbnail_url, created_at, updated_at, created_device ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( item_id, card_id, row.get('user_id', default_user_id), row.get('type', 'note').strip()[:50], row.get('label', '').strip()[:500], row.get('url', '').strip() if row.get('url') else None, row.get('content', '').strip() if row.get('content') else None, row.get('position'), row.get('media_url', '').strip() if row.get('media_url') else None, row.get('media_type', '').strip() if row.get('media_type') else None, row.get('media_file_name', '').strip() if row.get('media_file_name') else None, row.get('storage_path'), row.get('thumbnail_url', '').strip() if row.get('thumbnail_url') else None, row.get('created_at'), row.get('updated_at'), row.get('created_device') )) items_inserted += 1 # Track media references for Phase 2 if row.get('media_url'): cursor.execute(""" INSERT INTO media_files ( id, item_id, card_id, user_id, original_url, file_name, storage_path, mime_type ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, ( str(uuid.uuid4()), item_id, card_id, row.get('user_id', default_user_id), row.get('media_url'), row.get('media_file_name'), row.get('storage_path'), row.get('media_type') )) media_refs += 1 except sqlite3.IntegrityError as e: if skip_duplicates: items_skipped += 1 else: print(f" ❌ Duplicate item ID: {item_id}") raise conn.commit() print(f" ✅ Items imported: {items_inserted}") print(f" ⏭️ Items skipped: {items_skipped}") print(f" 📸 Media references tracked: {media_refs}") # Final stats cursor.execute("SELECT COUNT(*) FROM cards WHERE enabled = 1") active_cards = cursor.fetchone()[0] cursor.execute("SELECT COUNT(*) FROM items") total_items = cursor.fetchone()[0] print(f"\n📊 Final Stats:") print(f" Active cards: {active_cards}") print(f" Total items: {total_items}") print(f" Database: {db_path}") print(f"\n✅ Import complete!") print(f"\n⚠️ IMPORTANT:") print(f" 1. Change admin password immediately!") print(f" 2. Some media may be stored in Supabase—still accessible via media_url") print(f" 3. Phase 2 migration tool will help migrate media to NAS when ready") conn.close() def main(): parser = argparse.ArgumentParser(description='FastDesk CSV Import Tool') parser.add_argument('links_csv', help='Path to links-export-*.csv') parser.add_argument('items_csv', help='Path to items-export-*.csv') parser.add_argument('--db', default='fastdesk.db', help='SQLite database path (default: fastdesk.db)') parser.add_argument('--skip-duplicates', action='store_true', default=True, help='Skip duplicate records (default: true)') args = parser.parse_args() print("🚀 FastDesk Import Tool") print("=" * 60) import_data(args.links_csv, args.items_csv, args.db, skip_duplicates=args.skip_duplicates) if __name__ == '__main__': main()