consultoria-as
f395d67136
- Base de datos SQLite con información de vehículos - Dashboard web con Flask y Bootstrap - Scripts de web scraping para RockAuto - Interfaz CLI para consultas - Documentación completa del proyecto Incluye: - 12 marcas de vehículos - 10,923 modelos - 10,919 especificaciones de motores - 12,075 combinaciones modelo-año-motor
323 lines
11 KiB
Python
323 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Vehicle Database Manager
|
|
A script to initialize and manage the vehicle database with brands, years, models, and engines
|
|
"""
|
|
|
|
import sqlite3
|
|
import os
|
|
from datetime import datetime
|
|
from typing import List, Tuple
|
|
|
|
|
|
class VehicleDatabaseManager:
|
|
def __init__(self, db_path: str = "vehicle_database.db"):
|
|
self.db_path = db_path
|
|
self.connection = None
|
|
|
|
def connect(self):
|
|
"""Connect to the SQLite database"""
|
|
self.connection = sqlite3.connect(self.db_path)
|
|
self.connection.row_factory = sqlite3.Row # Enable column access by name
|
|
print(f"Connected to database: {self.db_path}")
|
|
|
|
def disconnect(self):
|
|
"""Close the database connection"""
|
|
if self.connection:
|
|
self.connection.close()
|
|
print("Disconnected from database")
|
|
|
|
def create_tables(self, schema_file: str = "sql/schema.sql"):
|
|
"""Create tables from schema file"""
|
|
if not os.path.exists(schema_file):
|
|
raise FileNotFoundError(f"Schema file not found: {schema_file}")
|
|
|
|
with open(schema_file, 'r') as f:
|
|
schema = f.read()
|
|
|
|
if self.connection:
|
|
cursor = self.connection.cursor()
|
|
cursor.executescript(schema)
|
|
self.connection.commit()
|
|
print("Tables created successfully")
|
|
|
|
def insert_brand(self, name: str, country: str = None, founded_year: int = None) -> int:
|
|
"""Insert a new brand and return its ID"""
|
|
cursor = self.connection.cursor()
|
|
cursor.execute(
|
|
"INSERT OR IGNORE INTO brands (name, country, founded_year) VALUES (?, ?, ?)",
|
|
(name, country, founded_year)
|
|
)
|
|
self.connection.commit()
|
|
# Return the ID of the inserted or existing brand
|
|
cursor.execute("SELECT id FROM brands WHERE name = ?", (name,))
|
|
return cursor.fetchone()[0]
|
|
|
|
def insert_engine(self, name: str, displacement_cc: float = None, cylinders: int = None,
|
|
fuel_type: str = None, power_hp: int = None, torque_nm: int = None,
|
|
engine_code: str = None) -> int:
|
|
"""Insert a new engine and return its ID"""
|
|
cursor = self.connection.cursor()
|
|
cursor.execute(
|
|
"""INSERT OR IGNORE INTO engines
|
|
(name, displacement_cc, cylinders, fuel_type, power_hp, torque_nm, engine_code)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)""",
|
|
(name, displacement_cc, cylinders, fuel_type, power_hp, torque_nm, engine_code)
|
|
)
|
|
self.connection.commit()
|
|
# Return the ID of the inserted or existing engine
|
|
cursor.execute("SELECT id FROM engines WHERE name = ?", (name,))
|
|
return cursor.fetchone()[0]
|
|
|
|
def insert_model(self, brand_id: int, name: str, body_type: str = None,
|
|
generation: str = None, production_start_year: int = None,
|
|
production_end_year: int = None) -> int:
|
|
"""Insert a new model and return its ID"""
|
|
cursor = self.connection.cursor()
|
|
cursor.execute(
|
|
"""INSERT OR IGNORE INTO models
|
|
(brand_id, name, body_type, generation, production_start_year, production_end_year)
|
|
VALUES (?, ?, ?, ?, ?, ?)""",
|
|
(brand_id, name, body_type, generation, production_start_year, production_end_year)
|
|
)
|
|
self.connection.commit()
|
|
# Return the ID of the inserted or existing model
|
|
cursor.execute("SELECT id FROM models WHERE brand_id = ? AND name = ?", (brand_id, name))
|
|
return cursor.fetchone()[0]
|
|
|
|
def insert_year(self, year: int) -> int:
|
|
"""Insert a new year and return its ID"""
|
|
cursor = self.connection.cursor()
|
|
cursor.execute("INSERT OR IGNORE INTO years (year) VALUES (?)", (year,))
|
|
self.connection.commit()
|
|
# Return the ID of the inserted or existing year
|
|
cursor.execute("SELECT id FROM years WHERE year = ?", (year,))
|
|
return cursor.fetchone()[0]
|
|
|
|
def insert_model_year_engine(self, model_id: int, year_id: int, engine_id: int,
|
|
trim_level: str = None, drivetrain: str = None,
|
|
transmission: str = None):
|
|
"""Insert a model-year-engine combination"""
|
|
cursor = self.connection.cursor()
|
|
cursor.execute(
|
|
"""INSERT OR REPLACE INTO model_year_engine
|
|
(model_id, year_id, engine_id, trim_level, drivetrain, transmission)
|
|
VALUES (?, ?, ?, ?, ?, ?)""",
|
|
(model_id, year_id, engine_id, trim_level, drivetrain, transmission)
|
|
)
|
|
self.connection.commit()
|
|
|
|
def get_brands(self) -> List[Tuple]:
|
|
"""Retrieve all brands"""
|
|
cursor = self.connection.cursor()
|
|
cursor.execute("SELECT * FROM brands ORDER BY name")
|
|
return cursor.fetchall()
|
|
|
|
def get_models_by_brand(self, brand_id: int) -> List[Tuple]:
|
|
"""Retrieve all models for a specific brand"""
|
|
cursor = self.connection.cursor()
|
|
cursor.execute("""
|
|
SELECT m.*, b.name as brand_name
|
|
FROM models m
|
|
JOIN brands b ON m.brand_id = b.id
|
|
WHERE m.brand_id = ?
|
|
ORDER BY m.name
|
|
""", (brand_id,))
|
|
return cursor.fetchall()
|
|
|
|
def get_vehicle_info(self, brand_name: str = None, model_name: str = None,
|
|
year: int = None, engine_name: str = None) -> List[Tuple]:
|
|
"""Get comprehensive vehicle information based on filters"""
|
|
query = """
|
|
SELECT
|
|
b.name AS brand,
|
|
m.name AS model,
|
|
y.year,
|
|
e.name AS engine,
|
|
e.displacement_cc,
|
|
e.cylinders,
|
|
e.fuel_type,
|
|
e.power_hp,
|
|
e.torque_nm,
|
|
e.engine_code,
|
|
mye.trim_level,
|
|
mye.drivetrain,
|
|
mye.transmission
|
|
FROM model_year_engine mye
|
|
JOIN models m ON mye.model_id = m.id
|
|
JOIN brands b ON m.brand_id = b.id
|
|
JOIN years y ON mye.year_id = y.id
|
|
JOIN engines e ON mye.engine_id = e.id
|
|
WHERE 1=1
|
|
"""
|
|
|
|
params = []
|
|
if brand_name:
|
|
query += " AND b.name LIKE ?"
|
|
params.append(f"%{brand_name}%")
|
|
if model_name:
|
|
query += " AND m.name LIKE ?"
|
|
params.append(f"%{model_name}%")
|
|
if year:
|
|
query += " AND y.year = ?"
|
|
params.append(year)
|
|
if engine_name:
|
|
query += " AND e.name LIKE ?"
|
|
params.append(f"%{engine_name}%")
|
|
|
|
query += " ORDER BY b.name, m.name, y.year"
|
|
|
|
cursor = self.connection.cursor()
|
|
cursor.execute(query, params)
|
|
return cursor.fetchall()
|
|
|
|
|
|
def populate_sample_data(db_manager: VehicleDatabaseManager):
|
|
"""Populate the database with sample data"""
|
|
print("Populating sample data...")
|
|
|
|
# Insert sample brands
|
|
toyota_id = db_manager.insert_brand("Toyota", "Japan", 1937)
|
|
honda_id = db_manager.insert_brand("Honda", "Japan", 1948)
|
|
ford_id = db_manager.insert_brand("Ford", "USA", 1903)
|
|
bmw_id = db_manager.insert_brand("BMW", "Germany", 1916)
|
|
|
|
# Insert sample engines
|
|
engine_2jz = db_manager.insert_engine(
|
|
"2JZ-GTE",
|
|
displacement_cc=3000,
|
|
cylinders=6,
|
|
fuel_type="gasoline",
|
|
power_hp=320,
|
|
torque_nm=450,
|
|
engine_code="2JZ-GTE"
|
|
)
|
|
|
|
engine_k20 = db_manager.insert_engine(
|
|
"K20C1",
|
|
displacement_cc=2000,
|
|
cylinders=4,
|
|
fuel_type="gasoline",
|
|
power_hp=280,
|
|
torque_nm=260,
|
|
engine_code="K20C1"
|
|
)
|
|
|
|
engine_ecoboost = db_manager.insert_engine(
|
|
"EcoBoost V6",
|
|
displacement_cc=3500,
|
|
cylinders=6,
|
|
fuel_type="gasoline",
|
|
power_hp=375,
|
|
torque_nm=470,
|
|
engine_code="EcoBoost V6"
|
|
)
|
|
|
|
engine_n52 = db_manager.insert_engine(
|
|
"N52B30",
|
|
displacement_cc=3000,
|
|
cylinders=6,
|
|
fuel_type="gasoline",
|
|
power_hp=255,
|
|
torque_nm=310,
|
|
engine_code="N52B30"
|
|
)
|
|
|
|
# Insert sample models
|
|
camry_id = db_manager.insert_model(
|
|
toyota_id,
|
|
"Camry",
|
|
body_type="sedan",
|
|
generation="XV70",
|
|
production_start_year=2017,
|
|
production_end_year=None
|
|
)
|
|
|
|
civic_id = db_manager.insert_model(
|
|
honda_id,
|
|
"Civic",
|
|
body_type="sedan",
|
|
generation="XI",
|
|
production_start_year=2016,
|
|
production_end_year=None
|
|
)
|
|
|
|
mustang_id = db_manager.insert_model(
|
|
ford_id,
|
|
"Mustang",
|
|
body_type="coupe",
|
|
generation="S550",
|
|
production_start_year=2015,
|
|
production_end_year=None
|
|
)
|
|
|
|
x3_id = db_manager.insert_model(
|
|
bmw_id,
|
|
"X3",
|
|
body_type="suv",
|
|
generation="G01",
|
|
production_start_year=2017,
|
|
production_end_year=None
|
|
)
|
|
|
|
# Insert years
|
|
year_2020 = db_manager.insert_year(2020)
|
|
year_2021 = db_manager.insert_year(2021)
|
|
year_2022 = db_manager.insert_year(2022)
|
|
|
|
# Link models, years, and engines
|
|
db_manager.insert_model_year_engine(civic_id, year_2020, engine_k20, "Sport", "FWD", "automatic")
|
|
db_manager.insert_model_year_engine(mustang_id, year_2020, engine_ecoboost, "GT", "RWD", "automatic")
|
|
db_manager.insert_model_year_engine(x3_id, year_2021, engine_n52, "xDrive30i", "AWD", "automatic")
|
|
db_manager.insert_model_year_engine(camry_id, year_2022, engine_k20, "SE", "FWD", "automatic")
|
|
|
|
print("Sample data populated successfully!")
|
|
|
|
|
|
def main():
|
|
# Initialize the database manager
|
|
db_manager = VehicleDatabaseManager()
|
|
|
|
try:
|
|
# Connect to database
|
|
db_manager.connect()
|
|
|
|
# Create tables
|
|
db_manager.create_tables()
|
|
|
|
# Populate with sample data
|
|
populate_sample_data(db_manager)
|
|
|
|
# Demonstrate queries
|
|
print("\n--- All Brands ---")
|
|
brands = db_manager.get_brands()
|
|
for brand in brands:
|
|
print(f"ID: {brand['id']}, Name: {brand['name']}, Country: {brand['country']}")
|
|
|
|
print(f"\n--- Models for Toyota (ID: {brands[0]['id']}) ---")
|
|
toyota_models = db_manager.get_models_by_brand(brands[0]['id'])
|
|
for model in toyota_models:
|
|
print(f"Model: {model['name']}, Body Type: {model['body_type']}")
|
|
|
|
print("\n--- All Vehicle Information ---")
|
|
all_vehicles = db_manager.get_vehicle_info()
|
|
for vehicle in all_vehicles:
|
|
print(f"{vehicle['brand']} {vehicle['model']} {vehicle['year']} - "
|
|
f"{vehicle['engine']} ({vehicle['power_hp']} HP)")
|
|
|
|
print("\n--- Search for Honda Civic ---")
|
|
honda_civic = db_manager.get_vehicle_info(brand_name="Honda", model_name="Civic")
|
|
for vehicle in honda_civic:
|
|
print(f"{vehicle['brand']} {vehicle['model']} {vehicle['year']} - "
|
|
f"{vehicle['engine']} ({vehicle['power_hp']} HP)")
|
|
|
|
except Exception as e:
|
|
print(f"An error occurred: {e}")
|
|
|
|
finally:
|
|
# Close the connection
|
|
db_manager.disconnect()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |