consultoria-as/Autoparts-DB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit: Sistema Autoparts DB

Browse Source 
- 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
This commit is contained in:
consultoria-as
2026-01-19 08:45:03 +00:00
commit f395d67136
59 changed files with 10881 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

152
vehicle_database/scripts/csv_importer.py Normal file
View File

@@ -0,0 +1,152 @@
#!/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()

323
vehicle_database/scripts/database_manager.py Normal file
View File

@@ -0,0 +1,323 @@
#!/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()

193
vehicle_database/scripts/query_interface.py Normal file
View File

@@ -0,0 +1,193 @@
#!/usr/bin/env python3
"""
Simple Query Interface for Vehicle Database
Provides a command-line interface for searching the vehicle database
"""
import sqlite3
from typing import List, Tuple
class VehicleQueryInterface:
def __init__(self, db_path: str = "vehicle_database.db"):
self.db_path = db_path
def search_vehicles(self, brand: str = None, model: str = None, year: int = None,
engine: str = None, limit: int = 50) -> List[Tuple]:
"""Search for vehicles based on various criteria"""
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row # Enable column access by name
cursor = conn.cursor()
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,
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:
query += " AND b.name LIKE ?"
params.append(f"%{brand}%")
if model:
query += " AND m.name LIKE ?"
params.append(f"%{model}%")
if year:
query += " AND y.year = ?"
params.append(year)
if engine:
query += " AND e.name LIKE ?"
params.append(f"%{engine}%")
query += f" ORDER BY b.name, m.name, y.year LIMIT {limit}"
cursor.execute(query, params)
results = cursor.fetchall()
conn.close()
return results
def get_all_brands(self) -> List[Tuple]:
"""Get a list of all brands in the database"""
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute("SELECT DISTINCT name FROM brands ORDER BY name")
results = cursor.fetchall()
conn.close()
return [row[0] for row in results]
def get_models_by_brand(self, brand_name: str) -> List[Tuple]:
"""Get all models for a specific brand"""
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute("""
SELECT DISTINCT m.name
FROM models m
JOIN brands b ON m.brand_id = b.id
WHERE b.name = ?
ORDER BY m.name
""", (brand_name,))
results = cursor.fetchall()
conn.close()
return [row[0] for row in results]
def get_years_for_model(self, model_name: str) -> List[int]:
"""Get all years for a specific model"""
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute("""
SELECT DISTINCT y.year
FROM model_year_engine mye
JOIN models m ON mye.model_id = m.id
JOIN years y ON mye.year_id = y.id
WHERE m.name = ?
ORDER BY y.year
""", (model_name,))
results = cursor.fetchall()
conn.close()
return [row[0] for row in results]
def main():
interface = VehicleQueryInterface()
print("Vehicle Database Query Interface")
print("=================================")
while True:
print("\nOptions:")
print("1. Search vehicles")
print("2. List all brands")
print("3. List models for a brand")
print("4. List years for a model")
print("5. Exit")
choice = input("\nEnter your choice (1-5): ").strip()
if choice == "1":
print("\nSearch for vehicles:")
brand = input("Brand (optional): ").strip() or None
model = input("Model (optional): ").strip() or None
year_input = input("Year (optional): ").strip()
year = int(year_input) if year_input else None
engine = input("Engine (optional): ").strip() or None
results = interface.search_vehicles(brand=brand, model=model, year=year, engine=engine)
if results:
print(f"\nFound {len(results)} result(s):")
print("-" * 100)
for i, vehicle in enumerate(results, 1):
print(f"{i:2d}. {vehicle['year']} {vehicle['brand']} {vehicle['model']}")
print(f" Engine: {vehicle['engine']} ({vehicle['power_hp']} HP, {vehicle['displacement_cc']} cc)")
print(f" Trim: {vehicle['trim_level']}, Drivetrain: {vehicle['drivetrain']}, Transmission: {vehicle['transmission']}")
print()
else:
print("No vehicles found matching your criteria.")
elif choice == "2":
brands = interface.get_all_brands()
print(f"\nAll brands ({len(brands)}):")
for i, brand in enumerate(brands, 1):
print(f"{i:2d}. {brand}")
elif choice == "3":
brand = input("Enter brand name: ").strip()
if brand:
models = interface.get_models_by_brand(brand)
if models:
print(f"\nModels for {brand} ({len(models)}):")
for i, model in enumerate(models, 1):
print(f"{i:2d}. {model}")
else:
print(f"No models found for brand: {brand}")
else:
print("Please enter a brand name.")
elif choice == "4":
model = input("Enter model name: ").strip()
if model:
years = interface.get_years_for_model(model)
if years:
print(f"\nYears for {model} ({len(years)}):")
for i, year in enumerate(years, 1):
print(f"{i:2d}. {year}")
else:
print(f"No years found for model: {model}")
else:
print("Please enter a model name.")
elif choice == "5":
print("Thank you for using the Vehicle Database Query Interface!")
break
else:
print("Invalid choice. Please enter 1-5.")
if __name__ == "__main__":
main()

0
vehicle_database/scripts/vehicle_database.db Normal file
View File