diff --git a/routes/daily_sales_routes.py b/routes/daily_sales_routes.py index 929f1ed..d30111a 100644 --- a/routes/daily_sales_routes.py +++ b/routes/daily_sales_routes.py @@ -74,34 +74,8 @@ def _is_cache_valid(cache_key): # 輔助函數 # ========================================== -def validate_table_name(table_name): - """驗證表名(防止 SQL Injection)""" - import re - if not re.match(r'^[a-zA-Z_][a-zA-Z0-9_]*$', table_name): - raise ValueError(f"Invalid table name: {table_name}") - return table_name - - -def safe_read_sql(table_name, columns=None, engine=None, where_clause=None, limit=None, params=None): - """安全的 SQL 查詢函數,防止 SQL Injection""" - table_name = validate_table_name(table_name) - - if columns: - col_str = ', '.join([f'"{col}"' for col in columns]) - else: - col_str = '*' - - try: - query = f'SELECT {col_str} FROM "{table_name}"' - if where_clause: - query += f' WHERE {where_clause}' - if limit: - query += f' LIMIT {int(limit)}' - - return pd.read_sql(text(query), engine, params=params) - except Exception as e: - sys_log.error(f"[Security] SQL 查詢失敗: {e}") - raise +# 共用工具改 import 自 utils(去重,原本檔案內定義已移除) +from utils.security import validate_table_name, safe_read_sql # noqa: E402, F401 def preprocess_daily_sales_data(df): diff --git a/routes/sales_routes.py b/routes/sales_routes.py index 7f4997a..aeed001 100644 --- a/routes/sales_routes.py +++ b/routes/sales_routes.py @@ -38,43 +38,9 @@ _SALES_OPTIONS_CACHE = {} # 輔助函數 # ========================================== -def find_col(df_cols, keywords): - """從欄位列表中,根據關鍵字列表找出最匹配的欄位名稱""" - for k in keywords: - for col in df_cols: - if k in str(col): - return col - return None - - -def validate_table_name(table_name): - """驗證表名(防止 SQL Injection)""" - import re - if not re.match(r'^[a-zA-Z_][a-zA-Z0-9_]*$', table_name): - raise ValueError(f"Invalid table name: {table_name}") - return table_name - - -def safe_read_sql(table_name, columns=None, engine=None, where_clause=None, limit=None, params=None): - """安全的 SQL 查詢函數,防止 SQL Injection""" - table_name = validate_table_name(table_name) - - if columns: - col_str = ', '.join([f'"{col}"' for col in columns]) - else: - col_str = '*' - - try: - query = f'SELECT {col_str} FROM "{table_name}"' - if where_clause: - query += f' WHERE {where_clause}' - if limit: - query += f' LIMIT {int(limit)}' - - return pd.read_sql(text(query), engine, params=params) - except Exception as e: - sys_log.error(f"[Security] SQL 查詢失敗: {e}") - raise +# 共用工具改 import 自 utils(去重,原各 routes 各自定義已移除) +from utils.df_helpers import find_col # noqa: E402, F401 +from utils.security import validate_table_name, safe_read_sql # noqa: E402, F401 def _get_filtered_sales_data(cache_key):