File size: 5,589 Bytes
3b382fd | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 | # erp_db_init.py
import sqlite3
import os
import pathlib
def init_sqlite_db(db_path='./data/erp_db.sqlite'):
"""Initialize SQLite database with ERP tables."""
try:
# Create the database directory if it doesn't exist
db_dir = pathlib.Path(os.path.dirname(db_path))
db_dir.mkdir(exist_ok=True)
# Connect to SQLite database (will be created if it doesn't exist)
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
# Enable foreign keys
conn.execute("PRAGMA foreign_keys = ON")
# Create customers table
cursor.execute('''
CREATE TABLE IF NOT EXISTS erp_customers (
customer_id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
email TEXT NOT NULL UNIQUE,
phone TEXT,
address TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# Create products table
cursor.execute('''
CREATE TABLE IF NOT EXISTS erp_products (
product_id INTEGER PRIMARY KEY AUTOINCREMENT,
product_name TEXT NOT NULL,
description TEXT,
category TEXT,
price REAL NOT NULL,
stock_quantity INTEGER NOT NULL DEFAULT 0,
sku TEXT UNIQUE
)
''')
# Create orders table
cursor.execute('''
CREATE TABLE IF NOT EXISTS erp_orders (
order_id INTEGER PRIMARY KEY AUTOINCREMENT,
customer_id INTEGER,
order_date DATE DEFAULT CURRENT_DATE,
total_amount REAL NOT NULL,
status TEXT NOT NULL DEFAULT 'Processing',
previous_order_id INTEGER,
estimated_delivery DATE,
actual_delivery DATE,
payment_status TEXT DEFAULT 'Pending',
shipping_address TEXT NOT NULL,
shipping_country TEXT,
destination_country TEXT,
FOREIGN KEY (customer_id) REFERENCES erp_customers (customer_id),
FOREIGN KEY (previous_order_id) REFERENCES erp_orders (order_id)
)
''')
# Create order items table
cursor.execute('''
CREATE TABLE IF NOT EXISTS erp_order_items (
order_item_id INTEGER PRIMARY KEY AUTOINCREMENT,
order_id INTEGER,
product_id INTEGER,
quantity INTEGER NOT NULL,
unit_price REAL NOT NULL,
subtotal REAL NOT NULL,
FOREIGN KEY (order_id) REFERENCES erp_orders (order_id),
FOREIGN KEY (product_id) REFERENCES erp_products (product_id)
)
''')
# Create order history table
cursor.execute('''
CREATE TABLE IF NOT EXISTS erp_order_history (
history_id INTEGER PRIMARY KEY AUTOINCREMENT,
order_id INTEGER,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
status_change TEXT,
notes TEXT,
updated_by TEXT,
FOREIGN KEY (order_id) REFERENCES erp_orders (order_id)
)
''')
# Create invoices table
cursor.execute('''
CREATE TABLE IF NOT EXISTS erp_invoices (
invoice_id INTEGER PRIMARY KEY AUTOINCREMENT,
order_id INTEGER,
invoice_date DATE DEFAULT CURRENT_DATE,
amount REAL NOT NULL,
payment_terms TEXT,
due_date DATE,
is_paid BOOLEAN DEFAULT 0,
invoice_number TEXT UNIQUE,
FOREIGN KEY (order_id) REFERENCES erp_orders (order_id)
)
''')
# Create global disruptions table
cursor.execute('''
CREATE TABLE IF NOT EXISTS live_global_disruptions (
disruption_id INTEGER PRIMARY KEY AUTOINCREMENT,
source_country TEXT NOT NULL,
destination_country TEXT NOT NULL,
disruption_type TEXT NOT NULL,
severity INTEGER NOT NULL CHECK (severity >= 1 AND severity <= 5),
start_date DATE NOT NULL,
expected_end_date DATE,
actual_end_date DATE,
is_active BOOLEAN DEFAULT 1,
description TEXT,
impact_hours INTEGER,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE (source_country, destination_country, disruption_type, start_date)
)
''')
# Create indexes
cursor.execute('CREATE INDEX IF NOT EXISTS idx_orders_customer ON erp_orders (customer_id)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_order_items_order ON erp_order_items (order_id)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_order_items_product ON erp_order_items (product_id)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_order_history_order ON erp_order_history (order_id)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_invoices_order ON erp_invoices (order_id)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_global_disruptions_active ON live_global_disruptions (is_active)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_global_disruptions_countries ON live_global_disruptions (source_country, destination_country)')
conn.commit()
conn.close()
return True
except Exception as e:
print(f"Error initializing SQLite database: {str(e)}")
return False
if __name__ == "__main__":
init_sqlite_db()
|