Student data import implementation: added tables, migration, import script, tests, and documentation.

scripts/import_students.py

@@ -0,0 +1,703 @@
+#!/usr/bin/env python3
+"""
+CSV Import Script for Student Data
+
+This script imports student data from a CSV file (ISO-8859-1 encoded) into the SQLite database.
+It handles:
+- Encoding conversion from ISO-8859-1 to UTF-8
+- Splitting "Apellidos, Nombre" into separate fields
+- Date format conversion from DD/MM/YYYY to YYYY-MM-DD
+- Discarding the "Estudio" field as per requirements
+- Validation and error handling
+
+Usage:
+    python import_students.py <csv_file> [--database <db_file>] [--dry-run]
+
+Examples:
+    python import_students.py Datos_programa.csv
+    python import_students.py Datos_programa.csv --database tablets.db
+    python import_students.py Datos_programa.csv --dry-run
+"""
+
+import argparse
+import csv
+import os
+import re
+import sqlite3
+import sys
+from datetime import datetime
+from typing import Dict, List, Optional, Tuple
+
+
+# Default database file
+DEFAULT_DATABASE = 'tablets.db'
+
+# CSV field mapping
+CSV_FIELDS = [
+    'order_number',
+    'full_name',
+    'birth_date',
+    'cial_code',
+    'nif_nie_passport',
+    'registration_number',
+    'file_number',
+    'gender',
+    'study_group',
+    'study'  # This will be discarded
+]
+
+
+def parse_arguments():
+    """Parse command line arguments."""
+    parser = argparse.ArgumentParser(
+        description='Import student data from CSV file to SQLite database'
+    )
+    parser.add_argument(
+        'csv_file',
+        help='Path to the CSV file to import (ISO-8859-1 encoded)'
+    )
+    parser.add_argument(
+        '--database', '-d',
+        default=DEFAULT_DATABASE,
+        help=f'Path to SQLite database file (default: {DEFAULT_DATABASE})'
+    )
+    parser.add_argument(
+        '--dry-run',
+        action='store_true',
+        help='Validate and preview data without importing'
+    )
+    parser.add_argument(
+        '--delimiter',
+        default=',',
+        help='CSV delimiter (default: comma)'
+    )
+    parser.add_argument(
+        '--quotechar',
+        default='"',
+        help='CSV quote character (default: double quote)'
+    )
+    parser.add_argument(
+        '--encoding',
+        default='iso-8859-1',
+        help='CSV file encoding (default: iso-8859-1)'
+    )
+    parser.add_argument(
+        '--skip-errors',
+        action='store_true',
+        help='Skip rows with errors instead of stopping'
+    )
+    parser.add_argument(
+        '--verbose', '-v',
+        action='store_true',
+        help='Show detailed processing information'
+    )
+    return parser.parse_args()
+
+
+def read_csv_file(filepath: str, encoding: str, delimiter: str, quotechar: str) -> List[Dict[str, str]]:
+    """Read CSV file and return list of dictionaries."""
+    """
+    Read a CSV file with specified encoding and return its contents as a list of dictionaries.
+    
+    Args:
+        filepath: Path to the CSV file
+        encoding: File encoding (e.g., 'iso-8859-1')
+        delimiter: CSV field delimiter
+        quotechar: CSV quote character
+        
+    Returns:
+        List of dictionaries where keys are header names and values are field values
+    """
+    if not os.path.exists(filepath):
+        raise FileNotFoundError(f"CSV file not found: {filepath}")
+    
+    if not os.path.isfile(filepath):
+        raise ValueError(f"Path is not a file: {filepath}")
+    
+    rows = []
+    with open(filepath, 'r', encoding=encoding, newline='') as csvfile:
+        reader = csv.DictReader(csvfile, delimiter=delimiter, quotechar=quotechar)
+        
+        # Convert header names to lowercase and strip whitespace
+        if reader.fieldnames:
+            reader.fieldnames = [name.strip().lower() for name in reader.fieldnames]
+        
+        for row in reader:
+            rows.append(row)
+    
+    return rows
+
+
+def split_full_name(full_name: str) -> Tuple[str, str]:
+    """
+    Split a full name in format 'APELLIDOS, NOMBRE' into last_name and first_name.
+    
+    Args:
+        full_name: Full name string in format 'Last Name, First Name'
+        
+    Returns:
+        Tuple of (last_name, first_name)
+    """
+    if not full_name or full_name.strip() == '':
+        return '', ''
+    
+    # Remove surrounding quotes if present
+    full_name = full_name.strip().strip('"').strip("'")
+    
+    # Split on comma followed by optional whitespace
+    parts = re.split(r',\s*', full_name, maxsplit=1)
+    
+    if len(parts) == 2:
+        last_name = parts[0].strip()
+        first_name = parts[1].strip()
+    else:
+        # If no comma, assume it's just a name
+        last_name = ''
+        first_name = full_name.strip()
+    
+    return last_name, first_name
+
+
+def convert_date(date_str: str) -> Optional[str]:
+    """
+    Convert date from DD/MM/YYYY to YYYY-MM-DD format.
+    
+    Args:
+        date_str: Date string in DD/MM/YYYY format
+        
+    Returns:
+        Date string in YYYY-MM-DD format, or None if invalid
+    """
+    if not date_str or date_str.strip() == '':
+        return None
+    
+    date_str = date_str.strip()
+    
+    try:
+        # Parse DD/MM/YYYY
+        day, month, year = re.split(r'[/-]', date_str)
+        day = int(day)
+        month = int(month)
+        year = int(year)
+        
+        # Validate date
+        datetime(year, month, day)
+        
+        # Format as YYYY-MM-DD
+        return f"{year:04d}-{month:02d}-{day:02d}"
+    except (ValueError, IndexError):
+        return None
+
+
+def validate_gender(gender: str) -> Optional[str]:
+    """
+    Validate and normalize gender field.
+    
+    Args:
+        gender: Gender string (M, F, etc.)
+        
+    Returns:
+        Normalized gender ('M', 'F', or 'O') or None if invalid
+    """
+    if not gender or gender.strip() == '':
+        return None
+    
+    gender = gender.strip().upper()
+    
+    if gender in ['M', 'H', 'MALE', 'HOMBRE']:
+        return 'M'
+    elif gender in ['F', 'MUJER', 'FEMALE']:
+        return 'F'
+    elif gender in ['O', 'OTHER', 'OTRO']:
+        return 'O'
+    else:
+        return None
+
+
+def process_row(row: Dict[str, str], verbose: bool = False) -> Optional[Dict[str, Optional[str]]]:
+    """
+    Process a single CSV row and convert it to database format.
+    
+    Args:
+        row: Dictionary with CSV field names as keys
+        verbose: Whether to print processing details
+        
+    Returns:
+        Dictionary with processed data ready for database insertion, or None if invalid
+    """
+    # Normalize the row keys to lowercase and remove special characters
+    normalized_row = {}
+    for key, value in row.items():
+        # Remove special characters and normalize
+        normalized_key = key.lower().replace('º', '').replace('ó', 'o').replace('á', 'a').replace('í', 'i').replace('ú', 'u').replace('ñ', 'n').replace('.', '').replace(',', '').strip()
+        normalized_row[normalized_key] = value
+    
+    # Map CSV fields to our expected field names
+    mapped_row = {}
+    field_mapping = {
+        'n or': 'order_number',
+        'apellidos nombre': 'full_name',
+        'fecha nac': 'birth_date',
+        'cial': 'cial_code',
+        'nif/nie/pas': 'nif_nie_passport',
+        'registro': 'registration_number',
+        'expediente': 'file_number',
+        'sexo': 'gender',
+        'grupo': 'study_group',
+        'estudio': 'study'
+    }
+    
+    for csv_key, our_key in field_mapping.items():
+        if csv_key in normalized_row:
+            mapped_row[our_key] = normalized_row[csv_key]
+        else:
+            mapped_row[our_key] = ''
+    
+    # Extract and process fields
+    try:
+        # Order number
+        order_number = mapped_row.get('order_number', '')
+        if order_number.strip() == '':
+            order_number = None
+        else:
+            try:
+                order_number = int(order_number.strip())
+            except ValueError:
+                order_number = None
+        
+        # Full name and split into last_name, first_name
+        full_name = mapped_row.get('full_name', '')
+        last_name, first_name = split_full_name(full_name)
+        
+        # Birth date conversion
+        birth_date_str = mapped_row.get('birth_date', '')
+        birth_date = convert_date(birth_date_str)
+        
+        # CIAL code
+        cial_code = mapped_row.get('cial_code', '').strip()
+        if cial_code == '':
+            cial_code = None
+        
+        # NIF/NIE/Passport
+        nif_nie_passport = mapped_row.get('nif_nie_passport', '').strip()
+        if nif_nie_passport == '':
+            nif_nie_passport = None
+        
+        # Registration number
+        registration_number = mapped_row.get('registration_number', '')
+        if registration_number.strip() == '':
+            registration_number = None
+        else:
+            try:
+                registration_number = int(registration_number.strip())
+            except ValueError:
+                registration_number = None
+        
+        # File number
+        file_number = mapped_row.get('file_number', '').strip()
+        if file_number == '':
+            file_number = None
+        
+        # Gender
+        gender = mapped_row.get('gender', '')
+        gender = validate_gender(gender)
+        
+        # Study group
+        study_group = mapped_row.get('study_group', '').strip()
+        if study_group == '':
+            study_group = None
+        
+        # Create result dictionary
+        result = {
+            'order_number': order_number,
+            'cial_code': cial_code,
+            'nif_nie_passport': nif_nie_passport,
+            'registration_number': registration_number,
+            'file_number': file_number,
+            'first_name': first_name if first_name else None,
+            'last_name': last_name if last_name else None,
+            'full_name': full_name.strip() if full_name.strip() else None,
+            'birth_date': birth_date,
+            'gender': gender,
+            'study_group': study_group,
+            'created_at': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
+            'updated_at': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+        }
+        
+        if verbose:
+            print(f"  Processed: {result}")
+        
+        return result
+        
+    except Exception as e:
+        print(f"  Error processing row: {e}")
+        return None
+
+
+def validate_row(data: Dict[str, Optional[str]]) -> Tuple[bool, List[str]]:
+    """
+    Validate processed row data.
+    
+    Args:
+        data: Processed row data
+        
+    Returns:
+        Tuple of (is_valid, list_of_errors)
+    """
+    errors = []
+    
+    # Check required fields
+    if not data.get('cial_code'):
+        errors.append("CIAL code is required")
+    
+    if not data.get('first_name'):
+        errors.append("First name is required")
+    
+    if not data.get('last_name'):
+        errors.append("Last name is required")
+    
+    # Check date format
+    birth_date = data.get('birth_date')
+    if birth_date:
+        try:
+            datetime.strptime(birth_date, '%Y-%m-%d')
+        except ValueError:
+            errors.append(f"Invalid birth date format: {birth_date}")
+    
+    # Check gender
+    gender = data.get('gender')
+    if gender and gender not in ['M', 'F', 'O']:
+        errors.append(f"Invalid gender: {gender}")
+    
+    return len(errors) == 0, errors
+
+
+def create_students_table(conn: sqlite3.Connection) -> bool:
+    """
+    Create the students table if it doesn't exist.
+    
+    Args:
+        conn: SQLite database connection
+        
+    Returns:
+        True if table was created or already exists, False on error
+    """
+    try:
+        cursor = conn.cursor()
+        
+        cursor.execute('''
+            CREATE TABLE IF NOT EXISTS students (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                order_number INTEGER,
+                cial_code TEXT UNIQUE NOT NULL,
+                nif_nie_passport TEXT UNIQUE,
+                registration_number INTEGER,
+                file_number TEXT,
+                first_name TEXT NOT NULL,
+                last_name TEXT NOT NULL,
+                full_name TEXT NOT NULL,
+                birth_date TEXT,
+                gender TEXT CHECK(gender IN ('M', 'F', 'O')),
+                study_group TEXT,
+                created_at TEXT DEFAULT CURRENT_TIMESTAMP,
+                updated_at TEXT DEFAULT CURRENT_TIMESTAMP,
+                CONSTRAINT unique_identification UNIQUE (cial_code, nif_nie_passport)
+            )
+        ''')
+        
+        # Create indexes
+        cursor.execute('''
+            CREATE INDEX IF NOT EXISTS idx_students_cial ON students(cial_code)
+        ''')
+        cursor.execute('''
+            CREATE INDEX IF NOT EXISTS idx_students_nif ON students(nif_nie_passport)
+        ''')
+        cursor.execute('''
+            CREATE INDEX IF NOT EXISTS idx_students_name ON students(last_name, first_name)
+        ''')
+        cursor.execute('''
+            CREATE INDEX IF NOT EXISTS idx_students_birth_date ON students(birth_date)
+        ''')
+        cursor.execute('''
+            CREATE INDEX IF NOT EXISTS idx_students_gender ON students(gender)
+        ''')
+        cursor.execute('''
+            CREATE INDEX IF NOT EXISTS idx_students_group ON students(study_group)
+        ''')
+        
+        conn.commit()
+        return True
+        
+    except sqlite3.Error as e:
+        print(f"Error creating students table: {e}")
+        conn.rollback()
+        return False
+
+
+def insert_student(conn: sqlite3.Connection, data: Dict[str, Optional[str]], verbose: bool = False) -> Tuple[bool, Optional[int]]:
+    """
+    Insert a student record into the database.
+    
+    Args:
+        conn: SQLite database connection
+        data: Student data dictionary
+        verbose: Whether to print details
+        
+    Returns:
+        Tuple of (success, student_id or None)
+    """
+    try:
+        cursor = conn.cursor()
+        
+        # Prepare SQL and parameters
+        sql = '''
+            INSERT INTO students (
+                order_number, cial_code, nif_nie_passport, registration_number, 
+                file_number, first_name, last_name, full_name, birth_date, 
+                gender, study_group, created_at, updated_at
+            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+        '''
+        
+        params = (
+            data['order_number'],
+            data['cial_code'],
+            data['nif_nie_passport'],
+            data['registration_number'],
+            data['file_number'],
+            data['first_name'],
+            data['last_name'],
+            data['full_name'],
+            data['birth_date'],
+            data['gender'],
+            data['study_group'],
+            data['created_at'],
+            data['updated_at']
+        )
+        
+        cursor.execute(sql, params)
+        student_id = cursor.lastrowid
+        conn.commit()
+        
+        if verbose:
+            print(f"  Inserted student ID: {student_id}")
+        
+        return True, student_id
+        
+    except sqlite3.IntegrityError as e:
+        if verbose:
+            print(f"  Duplicate entry (skipped): {e}")
+        conn.rollback()
+        return False, None
+    except sqlite3.Error as e:
+        print(f"  Database error: {e}")
+        conn.rollback()
+        return False, None
+
+
+def check_duplicate(conn: sqlite3.Connection, cial_code: str, nif_nie_passport: Optional[str]) -> bool:
+    """
+    Check if a student with the same CIAL code or NIF/NIE already exists.
+    
+    Args:
+        conn: SQLite database connection
+        cial_code: CIAL code to check
+        nif_nie_passport: NIF/NIE/Passport to check
+        
+    Returns:
+        True if duplicate exists, False otherwise
+    """
+    try:
+        cursor = conn.cursor()
+        
+        # Check by CIAL code
+        cursor.execute("SELECT id FROM students WHERE cial_code = ?", (cial_code,))
+        if cursor.fetchone():
+            return True
+        
+        # Check by NIF/NIE if provided
+        if nif_nie_passport and nif_nie_passport.strip():
+            cursor.execute("SELECT id FROM students WHERE nif_nie_passport = ?", (nif_nie_passport,))
+            if cursor.fetchone():
+                return True
+        
+        return False
+        
+    except sqlite3.Error as e:
+        print(f"Error checking for duplicates: {e}")
+        return False
+
+
+def import_csv(csv_file: str, database: str, dry_run: bool = False, 
+              skip_errors: bool = False, verbose: bool = False) -> Dict[str, int]:
+    """
+    Main import function.
+    
+    Args:
+        csv_file: Path to CSV file
+        database: Path to SQLite database
+        dry_run: If True, validate without importing
+        skip_errors: If True, skip invalid rows instead of stopping
+        verbose: If True, print detailed information
+        
+    Returns:
+        Dictionary with import statistics
+    """
+    stats = {
+        'total_rows': 0,
+        'processed': 0,
+        'valid': 0,
+        'invalid': 0,
+        'imported': 0,
+        'duplicates': 0,
+        'errors': 0
+    }
+    
+    # Read CSV file
+    try:
+        if verbose:
+            print(f"Reading CSV file: {csv_file}")
+        
+        rows = read_csv_file(csv_file, encoding='iso-8859-1', delimiter=',', quotechar='"')
+        stats['total_rows'] = len(rows)
+        
+        if verbose:
+            print(f"Found {len(rows)} rows in CSV file")
+        
+    except Exception as e:
+        print(f"Error reading CSV file: {e}")
+        stats['errors'] += 1
+        return stats
+    
+    # Connect to database
+    try:
+        conn = sqlite3.connect(database)
+        conn.execute("PRAGMA foreign_keys = ON")
+        
+        if not dry_run:
+            # Create students table if it doesn't exist
+            if not create_students_table(conn):
+                print("Failed to create students table")
+                stats['errors'] += 1
+                return stats
+        
+    except sqlite3.Error as e:
+        print(f"Error connecting to database: {e}")
+        stats['errors'] += 1
+        return stats
+    
+    # Process each row
+    for i, row in enumerate(rows, 1):
+        if verbose:
+            print(f"\nProcessing row {i}/{len(rows)}:")
+            print(f"  Raw: {row}")
+        
+        # Process row
+        processed = process_row(row, verbose)
+        
+        if processed is None:
+            stats['errors'] += 1
+            if not skip_errors:
+                print(f"Error processing row {i}, stopping.")
+                break
+            else:
+                if verbose:
+                    print(f"  Skipping row {i} due to processing error")
+                continue
+        
+        stats['processed'] += 1
+        
+        # Validate row
+        is_valid, errors = validate_row(processed)
+        
+        if not is_valid:
+            stats['invalid'] += 1
+            if verbose:
+                print(f"  Invalid row {i}: {errors}")
+            if not skip_errors:
+                print(f"Error validating row {i}, stopping.")
+                break
+            else:
+                continue
+        
+        stats['valid'] += 1
+        
+        if dry_run:
+            # Just print what would be imported
+            if verbose:
+                print(f"  [DRY RUN] Would import: {processed['full_name']} ({processed['cial_code']})")
+        else:
+            # Check for duplicates
+            if check_duplicate(conn, processed['cial_code'], processed.get('nif_nie_passport')):
+                stats['duplicates'] += 1
+                if verbose:
+                    print(f"  Duplicate found for CIAL: {processed['cial_code']}, skipping")
+                continue
+            
+            # Insert into database
+            success, student_id = insert_student(conn, processed, verbose)
+            
+            if success:
+                stats['imported'] += 1
+            else:
+                stats['duplicates'] += 1
+    
+    # Close database connection
+    if not dry_run:
+        conn.close()
+    
+    return stats
+
+
+def print_statistics(stats: Dict[str, int], dry_run: bool = False):
+    """Print import statistics."""
+    print("\n" + "=" * 50)
+    print("IMPORT STATISTICS")
+    print("=" * 50)
+    print(f"Total rows in CSV:     {stats['total_rows']}")
+    print(f"Rows processed:        {stats['processed']}")
+    print(f"Valid rows:            {stats['valid']}")
+    print(f"Invalid rows:          {stats['invalid']}")
+    if not dry_run:
+        print(f"Rows imported:         {stats['imported']}")
+    print(f"Duplicates skipped:    {stats['duplicates']}")
+    print(f"Errors:                {stats['errors']}")
+    
+    if dry_run:
+        print("\n[DRY RUN] No data was imported to the database.")
+    
+    print("=" * 50)
+
+
+def main():
+    """Main entry point."""
+    args = parse_arguments()
+    
+    print(f"Importing student data from: {args.csv_file}")
+    print(f"Database: {args.database}")
+    print(f"Dry run: {args.dry_run}")
+    print(f"Skip errors: {args.skip_errors}")
+    print()
+    
+    # Perform import
+    stats = import_csv(
+        csv_file=args.csv_file,
+        database=args.database,
+        dry_run=args.dry_run,
+        skip_errors=args.skip_errors,
+        verbose=args.verbose
+    )
+    
+    # Print statistics
+    print_statistics(stats, args.dry_run)
+    
+    # Exit with appropriate code
+    if stats['errors'] > 0:
+        sys.exit(1)
+    elif stats['invalid'] > 0 and not args.skip_errors:
+        sys.exit(1)
+    else:
+        sys.exit(0)
+
+
+if __name__ == '__main__':
+    main()