#!/usr/bin/env python3
"""
Database initialization script for Stevia Store MCP Server
This script creates the database tables and populates them with sample data
"""
import sqlite3
import json
import os
from pathlib import Path
# Database file path
DATABASE_PATH = os.getenv("DATABASE_PATH", "stevia_store.db")
def create_tables(cursor):
"""Create all necessary database tables"""
# Products table
cursor.execute("""
CREATE TABLE IF NOT EXISTS products (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
category TEXT NOT NULL,
price REAL NOT NULL,
cost REAL NOT NULL,
description TEXT,
sku TEXT UNIQUE,
stock_quantity INTEGER DEFAULT 0,
benefits TEXT,
usage_instructions TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
# Customers table
cursor.execute("""
CREATE TABLE IF NOT EXISTS customers (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
email TEXT UNIQUE,
phone TEXT,
address TEXT,
city TEXT,
preferences TEXT,
registration_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
total_orders INTEGER DEFAULT 0,
total_spent REAL DEFAULT 0.0
)
""")
# Orders table
cursor.execute("""
CREATE TABLE IF NOT EXISTS orders (
id INTEGER PRIMARY KEY AUTOINCREMENT,
customer_id INTEGER,
order_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
status TEXT DEFAULT 'pending',
total_amount REAL NOT NULL,
shipping_address TEXT,
notes TEXT,
FOREIGN KEY (customer_id) REFERENCES customers (id)
)
""")
# Order items table
cursor.execute("""
CREATE TABLE IF NOT EXISTS order_items (
id INTEGER PRIMARY KEY AUTOINCREMENT,
order_id INTEGER,
product_id INTEGER,
quantity INTEGER NOT NULL,
unit_price REAL NOT NULL,
total_price REAL NOT NULL,
FOREIGN KEY (order_id) REFERENCES orders (id),
FOREIGN KEY (product_id) REFERENCES products (id)
)
""")
# Inventory table
cursor.execute("""
CREATE TABLE IF NOT EXISTS inventory (
id INTEGER PRIMARY KEY AUTOINCREMENT,
product_id INTEGER,
quantity INTEGER NOT NULL DEFAULT 0,
low_stock_threshold INTEGER DEFAULT 10,
location TEXT DEFAULT 'Main Warehouse',
last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (product_id) REFERENCES products (id)
)
""")
def load_sample_data(cursor):
"""Load sample data from JSON file"""
# Get the path to the sample data file
script_dir = Path(__file__).parent
sample_data_path = script_dir.parent / "examples" / "sample_data.json"
if not sample_data_path.exists():
print(f"Sample data file not found: {sample_data_path}")
return
with open(sample_data_path, 'r', encoding='utf-8') as f:
data = json.load(f)
# Insert products
for product in data.get('products', []):
benefits_str = ', '.join(product.get('benefits', []))
cursor.execute("""
INSERT OR IGNORE INTO products
(name, category, price, cost, description, sku, stock_quantity, benefits, usage_instructions)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
product['name'],
product['category'],
product['price'],
product['cost'],
product['description'],
product['sku'],
product.get('stock', 0),
benefits_str,
product.get('usage', '')
))
# Create inventory record for this product
product_id = cursor.lastrowid
if product_id:
cursor.execute("""
INSERT INTO inventory (product_id, quantity, low_stock_threshold)
VALUES (?, ?, ?)
""", (product_id, product.get('stock', 0), 10))
# Insert customers
for customer in data.get('customers', []):
preferences_str = ', '.join(customer.get('preferences', []))
cursor.execute("""
INSERT OR IGNORE INTO customers
(name, email, phone, city, preferences)
VALUES (?, ?, ?, ?, ?)
""", (
customer['name'],
customer['email'],
customer.get('phone', ''),
customer.get('city', ''),
preferences_str
))
def main():
"""Main function to initialize the database"""
print(f"Initializing database: {DATABASE_PATH}")
# Create database connection
conn = sqlite3.connect(DATABASE_PATH)
cursor = conn.cursor()
try:
# Create tables
print("Creating tables...")
create_tables(cursor)
# Load sample data
print("Loading sample data...")
load_sample_data(cursor)
# Commit changes
conn.commit()
print("Database initialized successfully!")
# Show summary
cursor.execute("SELECT COUNT(*) FROM products")
product_count = cursor.fetchone()[0]
cursor.execute("SELECT COUNT(*) FROM customers")
customer_count = cursor.fetchone()[0]
print(f" - Products: {product_count}")
print(f" - Customers: {customer_count}")
except Exception as e:
print(f"Error initializing database: {e}")
conn.rollback()
finally:
conn.close()
if __name__ == "__main__":
main()
