- 建立 Gitea Actions CD pipeline (.gitea/workflows/cd.yaml) - 部署模式: rsync Python 檔案至 188 → docker restart (volume mount) - Dockerfile/requirements 變動時自動重建 Docker image - 部署通知: Telegram (開始/成功/失敗) - 健康檢查: https://mo.wooo.work/health (最多 5 次重試) - 同步最新 CLAUDE.md / ADR-008 / memory (2026-04-19) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
ogt
0
utils/__init__.py
Normal file
0
utils/__init__.py
Normal file
194
utils/image_url_builder.py
Normal file
194
utils/image_url_builder.py
Normal file
@@ -0,0 +1,194 @@
|
||||
# cSpell:ignore momo goodsimg webp
|
||||
"""
|
||||
MOMO 商品圖片 URL 構建器
|
||||
根據 i_code 直接構建商品圖片 URL，不需要從頁面抓取
|
||||
"""
|
||||
import re
|
||||
import random
|
||||
import requests
|
||||
from typing import Optional
|
||||
|
||||
def build_image_url(i_code: str) -> Optional[str]:
|
||||
"""
|
||||
根據 i_code 構建 MOMO 商品圖片 URL
|
||||
|
||||
MOMO 圖片 URL 格式：
|
||||
1. TP 開頭: https://i{1-8}.momoshop.com.tw/{timestamp}/goodsimg/TP000/{XXXX}/{YYYY}/{ZZZ}/{i_code}_O.webp
|
||||
2. 純數字: https://i{1-8}.momoshop.com.tw/{timestamp}/goodsimg/{XXXX}/{YYY}/{ZZZ}/{i_code}_OL_m.webp
|
||||
|
||||
Args:
|
||||
i_code: 商品編號
|
||||
|
||||
Returns:
|
||||
圖片 URL，如果無法構建則返回 None
|
||||
"""
|
||||
if not i_code:
|
||||
return None
|
||||
|
||||
# 隨機選擇一個 CDN 服務器 (i1-i8)
|
||||
cdn_num = random.randint(1, 8)
|
||||
|
||||
if i_code.startswith('TP'):
|
||||
# TP 開頭的商品
|
||||
# 格式: TP00008290000375 -> TP000/0829/0000/375/TP00008290000375_O.webp
|
||||
# 去掉 TP 前綴，然後按照 4-4-4-3 的格式分割
|
||||
code_without_tp = i_code[2:] # 去掉 'TP'
|
||||
|
||||
if len(code_without_tp) >= 15:
|
||||
part1 = code_without_tp[0:4] # 0000
|
||||
part2 = code_without_tp[4:8] # 8290
|
||||
part3 = code_without_tp[8:12] # 0000
|
||||
part4 = code_without_tp[12:] # 375
|
||||
|
||||
# 構建路徑
|
||||
path = f"TP000/{part2}/{part3}/{part4}"
|
||||
|
||||
# 嘗試多種可能的格式
|
||||
possible_formats = [
|
||||
f"https://i{cdn_num}.momoshop.com.tw/goodsimg/{path}/{i_code}_O.webp",
|
||||
f"https://i{cdn_num}.momoshop.com.tw/goodsimg/{path}/{i_code}_O.jpg",
|
||||
f"https://i{cdn_num}.momoshop.com.tw/goodsimg/{path}/{i_code}_O_m.jpg",
|
||||
f"https://i{cdn_num}.momoshop.com.tw/goodsimg/{path}/{i_code}_OL_m.webp",
|
||||
]
|
||||
|
||||
# 嘗試每個格式，返回第一個有效的
|
||||
for url in possible_formats:
|
||||
if verify_image_url(url):
|
||||
return url
|
||||
|
||||
else:
|
||||
# 純數字商品
|
||||
# 格式: 14548538 -> 0014/548/538/14548538_OL_m.webp
|
||||
try:
|
||||
# 確保是純數字
|
||||
code_num = int(i_code)
|
||||
code_str = str(code_num).zfill(8) # 補齊到 8 位
|
||||
|
||||
# 按照 4-3-3 的格式分割（從右邊開始）
|
||||
part3 = code_str[-3:] # 538
|
||||
part2 = code_str[-6:-3] # 548
|
||||
part1 = code_str[:-6].zfill(4) # 0014
|
||||
|
||||
# 構建路徑
|
||||
path = f"{part1}/{part2}/{part3}"
|
||||
|
||||
# 嘗試多種可能的格式
|
||||
possible_formats = [
|
||||
f"https://i{cdn_num}.momoshop.com.tw/goodsimg/{path}/{i_code}_OL_m.webp",
|
||||
f"https://i{cdn_num}.momoshop.com.tw/goodsimg/{path}/{i_code}_O_m.webp",
|
||||
f"https://i{cdn_num}.momoshop.com.tw/goodsimg/{path}/{i_code}_L.webp",
|
||||
f"https://i{cdn_num}.momoshop.com.tw/goodsimg/{path}/{i_code}_OL_m.jpg",
|
||||
f"https://i{cdn_num}.momoshop.com.tw/goodsimg/{path}/{i_code}_O_m.jpg",
|
||||
]
|
||||
|
||||
# 嘗試每個格式
|
||||
for url in possible_formats:
|
||||
if verify_image_url(url):
|
||||
return url
|
||||
|
||||
except ValueError:
|
||||
# 不是純數字
|
||||
pass
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def verify_image_url(url: str, timeout: int = 3) -> bool:
|
||||
"""
|
||||
驗證圖片 URL 是否有效
|
||||
|
||||
Args:
|
||||
url: 圖片 URL
|
||||
timeout: 超時時間（秒）
|
||||
|
||||
Returns:
|
||||
True 如果圖片可訪問，False 否則
|
||||
"""
|
||||
try:
|
||||
response = requests.head(url, timeout=timeout, allow_redirects=True)
|
||||
return response.status_code == 200
|
||||
except:
|
||||
return False
|
||||
|
||||
|
||||
def get_product_image_url(i_code: str) -> Optional[str]:
|
||||
"""
|
||||
獲取商品圖片 URL（直接從商品詳情頁 meta 標籤獲取）
|
||||
|
||||
這是最可靠的方法，因為 MOMO 的圖片 URL 格式包含動態 timestamp
|
||||
|
||||
Args:
|
||||
i_code: 商品編號
|
||||
|
||||
Returns:
|
||||
圖片 URL，如果無法獲取則返回 None
|
||||
"""
|
||||
return fetch_image_from_detail_page(i_code)
|
||||
|
||||
|
||||
def fetch_image_from_detail_page(i_code: str) -> Optional[str]:
|
||||
"""
|
||||
從商品詳情頁面抓取圖片 URL
|
||||
|
||||
Args:
|
||||
i_code: 商品編號
|
||||
|
||||
Returns:
|
||||
圖片 URL，如果無法抓取則返回 None
|
||||
"""
|
||||
try:
|
||||
# MOMO 商品詳情頁 URL
|
||||
detail_url = f"https://www.momoshop.com.tw/goods/GoodsDetail.jsp?i_code={i_code}"
|
||||
|
||||
# 添加必要的 headers 避免被封鎖
|
||||
headers = {
|
||||
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36',
|
||||
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
|
||||
'Accept-Language': 'zh-TW,zh;q=0.9,en-US;q=0.8,en;q=0.7',
|
||||
'Accept-Encoding': 'gzip, deflate, br',
|
||||
'Connection': 'keep-alive',
|
||||
}
|
||||
|
||||
response = requests.get(detail_url, headers=headers, timeout=10)
|
||||
if response.status_code != 200:
|
||||
return None
|
||||
|
||||
html = response.text
|
||||
|
||||
# 使用正則表達式查找圖片 URL
|
||||
# MOMO 的主圖在 og:image meta 標籤中
|
||||
pattern = r'<meta property="og:image" content="([^"]+)"'
|
||||
match = re.search(pattern, html, re.IGNORECASE)
|
||||
|
||||
if match:
|
||||
img_url = match.group(1)
|
||||
# 正規化 URL
|
||||
if img_url.startswith('//'):
|
||||
img_url = 'https:' + img_url
|
||||
elif img_url.startswith('/'):
|
||||
img_url = 'https://www.momoshop.com.tw' + img_url
|
||||
return img_url
|
||||
|
||||
except Exception as e:
|
||||
print(f"從詳情頁抓取圖片失敗: {e}")
|
||||
|
||||
return None
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# 測試
|
||||
test_codes = [
|
||||
"14548538", # 純數字
|
||||
"TP00008290000375", # TP 開頭
|
||||
"12092813", # 另一個純數字
|
||||
]
|
||||
|
||||
print("🧪 測試圖片 URL 構建器\n")
|
||||
for i_code in test_codes:
|
||||
print(f"商品編號: {i_code}")
|
||||
url = build_image_url(i_code)
|
||||
if url:
|
||||
print(f"✅ 圖片 URL: {url}")
|
||||
else:
|
||||
print(f"❌ 無法構建圖片 URL")
|
||||
print()
|
||||
253
utils/validators.py
Normal file
253
utils/validators.py
Normal file
@@ -0,0 +1,253 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
驗證函數模組
|
||||
提供安全驗證、路徑防護、檔案上傳驗證等功能
|
||||
"""
|
||||
|
||||
import re
|
||||
import os
|
||||
import unicodedata
|
||||
from pathlib import Path
|
||||
import pandas as pd
|
||||
from services.logger_manager import SystemLogger
|
||||
|
||||
sys_log = SystemLogger("Validators").get_logger()
|
||||
|
||||
# ==========================================
|
||||
# SQL Injection 防護
|
||||
# ==========================================
|
||||
|
||||
# 允許的資料表白名單
|
||||
ALLOWED_TABLES = {
|
||||
'products', 'price_records', 'promo_products',
|
||||
'monthly_summary_analysis', 'realtime_sales_monthly',
|
||||
'realtime_sales_daily', 'daily_sales_snapshot',
|
||||
'festival_products'
|
||||
}
|
||||
|
||||
|
||||
def validate_table_name(table_name):
|
||||
"""
|
||||
驗證資料表名稱，防止 SQL Injection
|
||||
|
||||
Args:
|
||||
table_name: 要驗證的資料表名稱
|
||||
|
||||
Returns:
|
||||
str: 驗證通過的表名
|
||||
|
||||
Raises:
|
||||
ValueError: 表名不在白名單中
|
||||
"""
|
||||
table_name = str(table_name).strip()
|
||||
|
||||
if not table_name:
|
||||
raise ValueError("表名不能為空")
|
||||
|
||||
if not re.match(r'^[a-zA-Z0-9_]+$', table_name):
|
||||
raise ValueError(f"表名包含非法字符: {table_name}")
|
||||
|
||||
if table_name not in ALLOWED_TABLES:
|
||||
sys_log.warning(f"[Security] 表名不在白名單中: {table_name}")
|
||||
sql_keywords = ['SELECT', 'INSERT', 'UPDATE', 'DELETE', 'DROP', 'CREATE', 'ALTER', 'UNION', 'WHERE', 'FROM']
|
||||
if any(keyword in table_name.upper() for keyword in sql_keywords):
|
||||
raise ValueError(f"表名包含 SQL 關鍵字: {table_name}")
|
||||
|
||||
return table_name
|
||||
|
||||
|
||||
def validate_column_names(column_names):
|
||||
"""
|
||||
驗證欄位名稱列表，防止 SQL Injection
|
||||
|
||||
Args:
|
||||
column_names: 欄位名稱列表
|
||||
|
||||
Returns:
|
||||
list: 驗證通過的欄位名稱列表
|
||||
|
||||
Raises:
|
||||
ValueError: 欄位名稱包含非法字符
|
||||
"""
|
||||
if isinstance(column_names, str):
|
||||
column_names = [column_names]
|
||||
|
||||
validated = []
|
||||
for col in column_names:
|
||||
col = str(col).strip()
|
||||
if not re.match(r'^[\w\u4e00-\u9fff]+$', col):
|
||||
raise ValueError(f"欄位名稱包含非法字符: {col}")
|
||||
validated.append(col)
|
||||
|
||||
return validated
|
||||
|
||||
|
||||
def safe_read_sql(table_name, columns=None, engine=None, where_clause=None, limit=None, params=None):
|
||||
"""
|
||||
安全的 SQL 查詢函數，防止 SQL Injection
|
||||
|
||||
Args:
|
||||
table_name: 資料表名稱
|
||||
columns: 欄位列表，None 表示 *
|
||||
engine: SQLAlchemy engine
|
||||
where_clause: WHERE 子句
|
||||
limit: 限制筆數
|
||||
params: 參數化查詢的參數字典
|
||||
|
||||
Returns:
|
||||
DataFrame: 查詢結果
|
||||
"""
|
||||
from sqlalchemy import text
|
||||
|
||||
table_name = validate_table_name(table_name)
|
||||
|
||||
if columns:
|
||||
columns = validate_column_names(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
|
||||
|
||||
|
||||
# ==========================================
|
||||
# 路徑遍歷防護
|
||||
# ==========================================
|
||||
|
||||
def safe_join(base, *paths):
|
||||
"""
|
||||
安全的路徑拼接，防止路徑遍歷攻擊
|
||||
|
||||
Args:
|
||||
base: 基礎目錄（絕對路徑）
|
||||
*paths: 子路徑組件
|
||||
|
||||
Returns:
|
||||
Path: 安全的完整路徑
|
||||
|
||||
Raises:
|
||||
ValueError: 偵測到路徑遍歷嘗試
|
||||
"""
|
||||
base = Path(base).resolve()
|
||||
|
||||
for path_component in paths:
|
||||
path_str = str(path_component)
|
||||
|
||||
if '\\' in path_str:
|
||||
sys_log.warning(f"[Security] 偵測到路徑遍歷嘗試 (Windows 反斜線) | Base: {base} | Requested: {paths}")
|
||||
raise ValueError(f"路徑遍歷偵測: 不允許使用反斜線")
|
||||
|
||||
if '..' in path_str.replace('\\', '/'):
|
||||
sys_log.warning(f"[Security] 偵測到路徑遍歷嘗試 (雙點) | Base: {base} | Requested: {paths}")
|
||||
raise ValueError(f"路徑遍歷偵測: 不允許使用 '..'")
|
||||
|
||||
full_path = (base / Path(*paths)).resolve()
|
||||
|
||||
try:
|
||||
full_path.relative_to(base)
|
||||
except ValueError:
|
||||
sys_log.warning(f"[Security] 偵測到路徑遍歷嘗試 | Base: {base} | Requested: {paths}")
|
||||
raise ValueError(f"路徑遍歷偵測: 不允許存取基礎目錄外的檔案")
|
||||
|
||||
return full_path
|
||||
|
||||
|
||||
# ==========================================
|
||||
# 檔案上傳安全驗證
|
||||
# ==========================================
|
||||
|
||||
ALLOWED_UPLOAD_EXTENSIONS = {'xlsx', 'xls', 'csv'}
|
||||
ALLOWED_MIME_TYPES = {
|
||||
'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
|
||||
'application/vnd.ms-excel',
|
||||
'text/csv',
|
||||
'application/octet-stream'
|
||||
}
|
||||
|
||||
|
||||
def secure_filename_unicode(filename):
|
||||
"""
|
||||
支援中文的安全檔案名稱清理
|
||||
|
||||
Args:
|
||||
filename: 原始檔案名稱
|
||||
|
||||
Returns:
|
||||
str: 清理後的安全檔案名稱
|
||||
"""
|
||||
filename = unicodedata.normalize('NFKC', filename)
|
||||
safe_chars = re.sub(r'[^\u4e00-\u9fa5a-zA-Z0-9\s\(\)_\-\.]', '', filename)
|
||||
safe_chars = re.sub(r'\s+', ' ', safe_chars)
|
||||
safe_chars = safe_chars.strip()
|
||||
return safe_chars
|
||||
|
||||
|
||||
def allowed_file(filename):
|
||||
"""
|
||||
檢查檔案副檔名是否在白名單中
|
||||
|
||||
Args:
|
||||
filename: 檔案名稱
|
||||
|
||||
Returns:
|
||||
bool: 是否允許上傳
|
||||
"""
|
||||
if not filename or '.' not in filename:
|
||||
return False
|
||||
|
||||
parts = filename.rsplit('.', 1)
|
||||
if len(parts) != 2:
|
||||
return False
|
||||
|
||||
basename, ext = parts
|
||||
|
||||
if not basename or basename.strip() == '':
|
||||
return False
|
||||
|
||||
return ext.lower() in ALLOWED_UPLOAD_EXTENSIONS
|
||||
|
||||
|
||||
def validate_upload_file(file):
|
||||
"""
|
||||
完整的檔案上傳驗證（副檔名、檔案名稱清理）
|
||||
|
||||
Args:
|
||||
file: Flask request.files 物件
|
||||
|
||||
Returns:
|
||||
tuple: (is_valid, error_message, safe_filename)
|
||||
"""
|
||||
if not file or file.filename == '':
|
||||
return False, '未選擇檔案', None
|
||||
|
||||
original_filename = file.filename
|
||||
|
||||
if '..' in original_filename:
|
||||
sys_log.warning(f"[Security] 檔案上傳 - 偵測到路徑遍歷嘗試（雙點）: {original_filename}")
|
||||
return False, '檔案名稱包含非法字元', None
|
||||
|
||||
if os.path.sep in original_filename or (os.path.altsep and os.path.altsep in original_filename):
|
||||
if original_filename.startswith(('/', '\\')) or './' in original_filename or '.\\' in original_filename:
|
||||
sys_log.warning(f"[Security] 檔案上傳 - 偵測到路徑遍歷嘗試（路徑分隔符）: {original_filename}")
|
||||
return False, '檔案名稱包含非法字元', None
|
||||
|
||||
safe_name = secure_filename_unicode(original_filename)
|
||||
if not safe_name:
|
||||
return False, '檔案名稱不合法', None
|
||||
|
||||
if not allowed_file(safe_name):
|
||||
return False, f'不支援的檔案格式，僅允許: {", ".join(ALLOWED_UPLOAD_EXTENSIONS)}', None
|
||||
|
||||
return True, None, safe_name
|
||||
Reference in New Issue
Block a user