""" 資料庫初始化腳本 使用 SQLAlchemy 建立資料表和插入範例資料 """ from sqlalchemy import create_engine from backend.database.connection import Base, engine from backend.database.models import User, Product, Order, OrderItem, LineMessage, UserSession from backend.config import settings import logging logger = logging.getLogger(__name__) def create_tables(): """建立所有資料表""" try: logger.info("開始建立資料表...") Base.metadata.create_all(bind=engine) logger.info("資料表建立完成!") return True except Exception as e: logger.error(f"建立資料表失敗: {str(e)}") return False def insert_sample_data(): """插入範例資料""" from backend.database.connection import SessionLocal db = SessionLocal() try: logger.info("開始插入範例資料...") # 檢查是否已有資料 existing_products = db.query(Product).count() if existing_products > 0: logger.info("資料表已有資料,跳過範例資料插入") return True # 插入範例商品 sample_products = [ Product( name="iPhone 15 Pro", description="最新款 iPhone,配備 A17 Pro 晶片", price=35900, stock=50, category="手機" ), Product( name="MacBook Air M2", description="輕薄筆記型電腦,搭載 M2 晶片", price=37900, stock=30, category="筆電" ), Product( name="AirPods Pro", description="主動降噪無線耳機", price=7490, stock=100, category="耳機" ), Product( name="iPad Air", description="輕薄平板電腦,適合工作和娛樂", price=18900, stock=25, category="平板" ), Product( name="Apple Watch Series 9", description="智慧手錶,健康監測功能", price=12900, stock=40, category="穿戴裝置" ) ] for product in sample_products: db.add(product) # 插入範例用戶 sample_users = [ User( user_id="U001", name="張小明", email="ming@example.com", display_name="小明" ), User( user_id="U002", name="李小華", email="hua@example.com", display_name="小華" ), User( user_id="U003", name="王小美", email="mei@example.com", display_name="小美" ) ] for user in sample_users: db.add(user) # 提交變更 db.commit() logger.info("範例資料插入完成!") return True except Exception as e: db.rollback() logger.error(f"插入範例資料失敗: {str(e)}") return False finally: db.close() def init_database(): """初始化資料庫""" logger.info("=== 開始初始化資料庫 ===") # 建立資料表 if not create_tables(): logger.error("資料表建立失敗,停止初始化") return False # 插入範例資料 if not insert_sample_data(): logger.error("範例資料插入失敗") return False logger.info("=== 資料庫初始化完成 ===") return True def drop_all_tables(): """刪除所有資料表(謹慎使用)""" try: logger.warning("正在刪除所有資料表...") Base.metadata.drop_all(bind=engine) logger.warning("所有資料表已刪除!") return True except Exception as e: logger.error(f"刪除資料表失敗: {str(e)}") return False if __name__ == "__main__": import sys if len(sys.argv) > 1 and sys.argv[1] == "--drop": # 刪除並重建資料表 drop_all_tables() init_database() else: # 正常初始化 init_database()