Autoparts-DB/vehicle_database/scripts/csv_importer.py

#!/usr/bin/env python3
"""
CSV Import Script for Vehicle Database
This script allows importing vehicle data from CSV files
"""

import csv
import sqlite3
import os
from typing import Dict, List


class CSVImporter:
    def __init__(self, db_path: str = "vehicle_database.db"):
        self.db_path = db_path
    
    def import_brands_from_csv(self, csv_file: str):
        """Import brands from a CSV file"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        with open(csv_file, 'r', newline='', encoding='utf-8') as file:
            reader = csv.DictReader(file)
            for row in reader:
                cursor.execute(
                    "INSERT OR IGNORE INTO brands (name, country, founded_year) VALUES (?, ?, ?)",
                    (row['name'], row.get('country'), row.get('founded_year'))
                )
        
        conn.commit()
        conn.close()
        print(f"Imported brands from {csv_file}")
    
    def import_engines_from_csv(self, csv_file: str):
        """Import engines from a CSV file"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        with open(csv_file, 'r', newline='', encoding='utf-8') as file:
            reader = csv.DictReader(file)
            for row in reader:
                cursor.execute(
                    """INSERT OR IGNORE INTO engines 
                       (name, displacement_cc, cylinders, fuel_type, power_hp, torque_nm, engine_code) 
                       VALUES (?, ?, ?, ?, ?, ?, ?)""",
                    (
                        row['name'], 
                        row.get('displacement_cc'), 
                        row.get('cylinders'), 
                        row.get('fuel_type'), 
                        row.get('power_hp'), 
                        row.get('torque_nm'), 
                        row.get('engine_code')
                    )
                )
        
        conn.commit()
        conn.close()
        print(f"Imported engines from {csv_file}")
    
    def import_models_from_csv(self, csv_file: str):
        """Import models from a CSV file"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        with open(csv_file, 'r', newline='', encoding='utf-8') as file:
            reader = csv.DictReader(file)
            
            # First, create a mapping of brand names to IDs
            cursor.execute("SELECT id, name FROM brands")
            brand_map = {row[1]: row[0] for row in cursor.fetchall()}
            
            for row in reader:
                brand_id = brand_map.get(row['brand_name'])
                if brand_id is None:
                    print(f"Warning: Brand '{row['brand_name']}' not found in database")
                    continue
                
                cursor.execute(
                    """INSERT OR IGNORE INTO models 
                       (brand_id, name, body_type, generation, production_start_year, production_end_year) 
                       VALUES (?, ?, ?, ?, ?, ?)""",
                    (
                        brand_id,
                        row['name'], 
                        row.get('body_type'), 
                        row.get('generation'), 
                        row.get('production_start_year'), 
                        row.get('production_end_year')
                    )
                )
        
        conn.commit()
        conn.close()
        print(f"Imported models from {csv_file}")
    
    def create_sample_csv_files(self):
        """Create sample CSV files with example data"""
        os.makedirs("data", exist_ok=True)
        
        # Sample brands CSV
        with open("data/brands.csv", "w", newline="", encoding="utf-8") as csvfile:
            fieldnames = ["name", "country", "founded_year"]
            writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
            writer.writeheader()
            writer.writerows([
                {"name": "Mercedes-Benz", "country": "Germany", "founded_year": 1926},
                {"name": "Audi", "country": "Germany", "founded_year": 1909},
                {"name": "Nissan", "country": "Japan", "founded_year": 1933},
                {"name": "Chevrolet", "country": "USA", "founded_year": 1911},
                {"name": "Volkswagen", "country": "Germany", "founded_year": 1937}
            ])
        
        # Sample engines CSV
        with open("data/engines.csv", "w", newline="", encoding="utf-8") as csvfile:
            fieldnames = ["name", "displacement_cc", "cylinders", "fuel_type", "power_hp", "torque_nm", "engine_code"]
            writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
            writer.writeheader()
            writer.writerows([
                {"name": "VQ35DE", "displacement_cc": 3500, "cylinders": 6, "fuel_type": "gasoline", "power_hp": 280, "torque_nm": 260, "engine_code": "VQ35DE"},
                {"name": "LS3", "displacement_cc": 6200, "cylinders": 8, "fuel_type": "gasoline", "power_hp": 430, "torque_nm": 424, "engine_code": "LS3"},
                {"name": "EA888", "displacement_cc": 2000, "cylinders": 4, "fuel_type": "gasoline", "power_hp": 228, "torque_nm": 258, "engine_code": "EA888"},
                {"name": "M274", "displacement_cc": 2000, "cylinders": 4, "fuel_type": "gasoline", "power_hp": 241, "torque_nm": 273, "engine_code": "M274"}
            ])
        
        # Sample models CSV
        with open("data/models.csv", "w", newline="", encoding="utf-8") as csvfile:
            fieldnames = ["brand_name", "name", "body_type", "generation", "production_start_year", "production_end_year"]
            writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
            writer.writeheader()
            writer.writerows([
                {"brand_name": "Toyota", "name": "Corolla", "body_type": "sedan", "generation": "E210", "production_start_year": 2018, "production_end_year": None},
                {"brand_name": "Honda", "name": "Accord", "body_type": "sedan", "generation": "X", "production_start_year": 2018, "production_end_year": None},
                {"brand_name": "Ford", "name": "F-150", "body_type": "truck", "generation": "13th Gen", "production_start_year": 2015, "production_end_year": None},
                {"brand_name": "BMW", "name": "3 Series", "body_type": "sedan", "generation": "G20", "production_start_year": 2018, "production_end_year": None}
            ])
        
        print("Sample CSV files created in data/ directory")


def main():
    importer = CSVImporter()
    
    # Create sample CSV files
    importer.create_sample_csv_files()
    
    # Import sample data (these would be imported by the main script, but showing the functionality)
    print("CSV import functionality created. Sample CSV files are in the data/ directory.")


if __name__ == "__main__":
    main()