import logging from sqlalchemy import func from database.manager import DatabaseManager from database.models import Product # 設定日誌 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') def check_duplicate_products(): """ 檢查 products 表中是否有重複的 i_code """ db = DatabaseManager() session = db.get_session() try: logging.info("🔍 開始檢查 products 資料庫中的重複 i_code...") # 查詢重複的 i_code # 使用 group_by 和 having 來找出出現次數大於 1 的 i_code duplicates = session.query(Product.i_code, func.count(Product.i_code))\ .group_by(Product.i_code)\ .having(func.count(Product.i_code) > 1)\ .all() if duplicates: logging.warning(f"⚠️ 發現 {len(duplicates)} 組重複的 i_code:") for i_code, count in duplicates: logging.warning(f" - i_code: {i_code}, 重複次數: {count}") # 列出重複項目的詳細資訊 (ID, 名稱, 分類) products = session.query(Product).filter(Product.i_code == i_code).all() for p in products: logging.info(f" * ID: {p.id}, Name: {p.name}, Category: {p.category}") else: logging.info("✅ products 表中沒有發現重複的 i_code。") except Exception as e: logging.error(f"❌ 檢查過程中發生錯誤: {e}") finally: session.close() if __name__ == "__main__": check_duplicate_products()