import pandas as pd # type: ignore import os import sys # 設定專案路徑以導入模組 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from database.manager import DatabaseManager def check_monthly_stats(): print("📊 正在分析資料庫中的月度業績數據...") db = DatabaseManager() try: # 1. 取得資料表欄位名稱 df_head = pd.read_sql("SELECT * FROM realtime_sales_monthly LIMIT 1", db.engine) cols = df_head.columns.tolist() # 自動尋找日期欄位 date_col = next((c for c in cols if '日期' in c or 'Date' in c), None) if not date_col: print(f"❌ 找不到日期欄位。現有欄位: {cols}") return print(f"📅 偵測到日期欄位: {date_col}") # 2. 讀取日期資料 (只讀取日期欄位以節省記憶體) df = pd.read_sql(f"SELECT \"{date_col}\" FROM realtime_sales_monthly", db.engine) if df.empty: print("⚠️ 資料表為空。") return # 3. 轉換與統計 df[date_col] = pd.to_datetime(df[date_col], errors='coerce') monthly_counts = df.groupby(df[date_col].dt.to_period('M')).size() print("\n" + "="*30) print("🗓️ 各月份資料筆數統計") print("="*30) if monthly_counts.empty: print("無法解析日期或無資料。") else: for period, count in monthly_counts.items(): print(f"{period}: {count:6d} 筆") print("="*30) print(f"總計: {len(df)} 筆") except Exception as e: print(f"❌ 讀取或分析失敗: {e}") if __name__ == "__main__": check_monthly_stats()