最終修復

Dockerfile

@@ -14,6 +14,7 @@ RUN apt-get update && apt-get install -y \
     libgomp1 \
     libgl1-mesa-glx \
     libglib2.0-dev \
+    ffmpeg \
     && rm -rf /var/lib/apt/lists/*
 
 # 複製依賴檔案

app.py

@@ -2,6 +2,8 @@
 # -*- coding: utf-8 -*-
 
 import os
+import hmac
+import hashlib
 import json
 import requests
 import cv2
@@ -20,6 +22,16 @@ from werkzeug.utils import secure_filename
 from datetime import datetime
 from flask_socketio import SocketIO, emit
 from openai import OpenAI
+from app_config import get_config
+
+# 選擇 SocketIO 執行模式（優先使用 eventlet）
+ASYNC_MODE = 'threading'
+try:
+    import eventlet
+    eventlet.monkey_patch()
+    ASYNC_MODE = 'eventlet'
+except Exception:
+    ASYNC_MODE = 'threading'
 
 # 環境變數設定
 # OpenAI API KEY 應該從環境變數獲取，不要硬編碼
@@ -34,15 +46,19 @@ os.environ['MEDIAPIPE_DISABLE_GPU'] = '1'  # 禁用GPU避免警告
 IS_HUGGINGFACE = os.environ.get('SPACE_ID') is not None
 IS_LOCAL_DEV = not IS_HUGGINGFACE
 
+# 載入集中設定
+CONFIG = get_config()
+
 # Flask 應用初始化
 app = Flask(__name__)
 app.config['SECRET_KEY'] = 'sign_language_secret_key'
-app.config['MAX_CONTENT_LENGTH'] = 100 * 1024 * 1024  # 100MB max file size
-socketio = SocketIO(app, cors_allowed_origins="*", async_mode='threading')
+app.config['MAX_CONTENT_LENGTH'] = CONFIG.get('MAX_FILE_SIZE', 100 * 1024 * 1024)  # 100MB max file size
+socketio = SocketIO(app, cors_allowed_origins="*", async_mode=ASYNC_MODE)
 
 # Messenger Bot 設定
-VERIFY_TOKEN = os.environ.get('VERIFY_TOKEN', 'your_verify_token')
-PAGE_ACCESS_TOKEN = os.environ.get('PAGE_ACCESS_TOKEN', 'your_page_access_token')
+VERIFY_TOKEN = CONFIG.get('VERIFY_TOKEN', 'your_verify_token')
+PAGE_ACCESS_TOKEN = CONFIG.get('PAGE_ACCESS_TOKEN', 'your_page_access_token')
+APP_SECRET = CONFIG.get('APP_SECRET')
 FACEBOOK_API_URL = 'https://graph.facebook.com/v18.0/me/messages'
 
 # 路徑設定 - 適應不同環境
@@ -77,6 +93,22 @@ class FeatureExtractor:
         self.mp_holistic = mp.solutions.holistic
         self.mp_drawing = mp.solutions.drawing_utils
         self.mp_drawing_styles = mp.solutions.drawing_styles
+        # 建立長駐的 Holistic 實例（避免每幀重建導致效能低落）
+        self.holistic = self.mp_holistic.Holistic(
+            static_image_mode=False,
+            model_complexity=1,
+            smooth_landmarks=True,
+            enable_segmentation=False,
+            min_detection_confidence=0.5,
+            min_tracking_confidence=0.5
+        )
+    
+    def close(self):
+        try:
+            if self.holistic:
+                self.holistic.close()
+        except Exception:
+            pass
         
     def extract_pose_keypoints(self, frame, holistic_results):
         """提取骨架關鍵點"""
@@ -226,6 +258,7 @@ class VideoSignLanguageRecognizer:
     def __init__(self, model_path, threshold=0.7):
         self.model_path = model_path
         self.threshold = threshold
+        self.effective_threshold = threshold
         
         # 初始化特徵提取器
         self.feature_extractor = FeatureExtractor()
@@ -239,7 +272,7 @@ class VideoSignLanguageRecognizer:
         
         # GPT整合
         try:
-            self.openai_client = OpenAI()
+            self.openai_client = OpenAI(timeout=10.0, max_retries=2)
         except Exception as e:
             print(f"初始化OpenAI客户端出錯: {e}")
             self.openai_client = None
@@ -247,37 +280,8 @@ class VideoSignLanguageRecognizer:
         print(f"影片辨識器初始化完成！使用設備: {self.device}")
     
     def _load_label_mapping(self):
-        """加載標籤映射"""
-        label_map = {}
-        
-        # 嘗試從 labels.csv 讀取
-        labels_file = LABELS_PATH
-        print(f"🔍 嘗試載入標籤檔案: {labels_file}")
-        print(f"📂 當前工作目錄: {os.getcwd()}")
-        
-        if os.path.exists(labels_file):
-            try:
-                df = pd.read_csv(labels_file)
-                print(f"📄 標籤檔案內容:")
-                print(df)
-                
-                for _, row in df.iterrows():
-                    label_map[int(row['index'])] = row['label']
-                print(f"✅ 從 {labels_file} 載入了 {len(label_map)} 個類別標籤")
-                print(f"📊 標籤映射: {label_map}")
-            except Exception as e:
-                print(f"❌ 讀取 labels.csv 出錯: {e}")
-                # 使用默認映射
-                label_map = {0: "eat", 1: "fish", 2: "like", 3: "want"}
-        else:
-            print(f"❌ 標籤檔案不存在: {labels_file}")
-        
-        if not label_map:
-            # 使用默認映射
-            label_map = {0: "eat", 1: "fish", 2: "like", 3: "want"}
-            print(f"⚠️ 使用預設標籤映射: {label_map}")
-        
-        return label_map
+        """加載標籤映射（統一由 labels.csv 提供）"""
+        return load_label_mapping_from_csv()
     
     def _load_model(self):
         """加載訓練好的模型"""
@@ -327,6 +331,9 @@ class VideoSignLanguageRecognizer:
         # 提取特徵序列
         keypoints_sequence = []
         frame_count = 0
+        hands_present_count = 0
+        motion_history = []
+        prev_gray = None
         
         while True:
             ret, frame = cap.read()
@@ -335,9 +342,22 @@ class VideoSignLanguageRecognizer:
             
             # 跳幀處理
             if frame_count % 5 == 0:  # 每5幀處理一次
-                keypoints, _ = self._extract_features(frame)
+                keypoints, hands_detected = self._extract_features(frame)
                 if keypoints is not None:
                     keypoints_sequence.append(keypoints)
+                if hands_detected:
+                    hands_present_count += 1
+                
+                # 計算光流運動量
+                try:
+                    gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
+                    if prev_gray is not None:
+                        flow = cv2.calcOpticalFlowFarneback(prev_gray, gray, None, 0.5, 3, 15, 3, 5, 1.2, 0)
+                        mag, ang = cv2.cartToPolar(flow[...,0], flow[...,1])
+                        motion_history.append(float(np.mean(mag)))
+                    prev_gray = gray
+                except Exception:
+                    pass
                     
             frame_count += 1
             
@@ -351,7 +371,18 @@ class VideoSignLanguageRecognizer:
             print(f"❌ 有效幀數不足，無法進行辨識")
             return None, 0
         
-        # 進行預測
+        # 動態調整 threshold（手部存在比例 + 運動量）
+        frames_used = max(1, len(keypoints_sequence))
+        hand_ratio = hands_present_count / frames_used
+        avg_motion = float(np.mean(motion_history)) if motion_history else 0.0
+        dynamic_threshold = self.threshold
+        if hand_ratio < 0.3:
+            dynamic_threshold = min(0.9, dynamic_threshold + 0.1)
+        if avg_motion < 0.05:
+            dynamic_threshold = min(0.9, dynamic_threshold + 0.05)
+        self.effective_threshold = dynamic_threshold
+
+        # 進行預測（使用動態 threshold）
         prediction, confidence, word_sequence, probabilities = self._predict_from_sequence(keypoints_sequence)
         
         # 使用GPT生成完整句子
@@ -365,34 +396,29 @@ class VideoSignLanguageRecognizer:
             'word_sequence': word_sequence,
             'confidence': confidence,
             'probabilities': probabilities,
-            'generated_sentence': generated_sentence
+            'generated_sentence': generated_sentence,
+            'hand_presence_ratio': hand_ratio,
+            'avg_motion': avg_motion,
+            'effective_threshold': dynamic_threshold
         }
 
     def _extract_features(self, frame):
         """從單一幀提取手部和姿勢特徵"""
-        with self.feature_extractor.mp_holistic.Holistic(
-            static_image_mode=False,
-            model_complexity=1,
-            smooth_landmarks=True,
-            enable_segmentation=False,
-            min_detection_confidence=0.1,
-            min_tracking_confidence=0.1
-        ) as holistic:
-            # 轉為RGB
-            frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
-            
-            # 處理圖像
-            results = holistic.process(frame_rgb)
-            
-            # 檢查是否有手部被檢測到
-            hands_detected = (results.left_hand_landmarks is not None or 
-                             results.right_hand_landmarks is not None)
-            
-            try:
-                keypoints = self.feature_extractor.extract_pose_keypoints(frame, results)
-                return keypoints, hands_detected
-            except Exception as e:
-                return None, hands_detected
+        # 轉為RGB
+        frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
+        
+        # 使用長駐的 holistic 實例處理圖像
+        results = self.feature_extractor.holistic.process(frame_rgb)
+        
+        # 檢查是否有手部被檢測到
+        hands_detected = (results.left_hand_landmarks is not None or 
+                          results.right_hand_landmarks is not None)
+        
+        try:
+            keypoints = self.feature_extractor.extract_pose_keypoints(frame, results)
+            return keypoints, hands_detected
+        except Exception as e:
+            return None, hands_detected
 
     def _predict_from_sequence(self, keypoints_sequence):
         """從關鍵點序列進行預測"""
@@ -411,7 +437,8 @@ class VideoSignLanguageRecognizer:
             # 提取所有類別的機率
             probs = probabilities[0].cpu().numpy()
             
-        if confidence >= self.threshold:
+        effective_thr = getattr(self, 'effective_threshold', self.threshold)
+        if confidence >= effective_thr:
             predicted_word = self.label_map.get(predicted_class, f"類別{predicted_class}")
             word_sequence = [predicted_word]
         else:
@@ -460,6 +487,7 @@ class SignLanguageRecognizer:
     def __init__(self, model_path, frame_buffer_size=30, prediction_interval=15, threshold=0.7):
         self.model_path = model_path
         self.threshold = threshold
+        self.dynamic_threshold = threshold
         self.max_buffer_size = frame_buffer_size
         self.prediction_interval = prediction_interval
         
@@ -488,6 +516,9 @@ class SignLanguageRecognizer:
         self.word_sequence = []
         self.last_added_word = None
         self.word_cooldown = 0
+        self.recent_top1_queue = collections.deque(maxlen=15)
+        self.ema_confidence = 0.0
+        self.ema_alpha = 0.3
         
         # 生成的句子
         self.generated_sentence = ""
@@ -495,7 +526,7 @@ class SignLanguageRecognizer:
         
         # GPT整合
         try:
-            self.openai_client = OpenAI()
+            self.openai_client = OpenAI(timeout=10.0, max_retries=2)
         except Exception as e:
             print(f"初始化OpenAI客户端出錯: {e}")
             self.openai_client = None
@@ -503,29 +534,8 @@ class SignLanguageRecognizer:
         print(f"即時辨識器初始化完成！使用設備: {self.device}")
 
     def _load_label_mapping(self):
-        """加載標籤映射"""
-        label_map = {}
-        # 嘗試從特徵目錄推斷類別標籤
-        features_dir = os.path.join(DATA_DIR, 'features', 'keypoints')
-        if os.path.exists(features_dir):
-            unique_labels = set()
-            for file_name in os.listdir(features_dir):
-                if file_name.endswith('_keypoints.npy'):
-                    parts = file_name.split('_')
-                    if len(parts) >= 2:
-                        label = parts[0]
-                        if label not in unique_labels and not (label.startswith("aug") or "aug_" in label):
-                            unique_labels.add(label)
-            
-            if unique_labels:
-                label_map = {i: label for i, label in enumerate(sorted(unique_labels))}
-                print(f"從特徵目錄推斷了 {len(label_map)} 個類別標籤")
-            else:
-                label_map = {0: "eat", 1: "fish", 2: "like", 3: "want"}
-        else:
-            label_map = {0: "eat", 1: "fish", 2: "like", 3: "want"}
-        
-        return label_map
+        """加載標籤映射（統一由 labels.csv 提供）"""
+        return load_label_mapping_from_csv()
 
     def _load_model(self):
         """加載訓練好的模型"""
@@ -575,7 +585,7 @@ class SignLanguageRecognizer:
         # 定期進行預測
         if self.hand_present and self.frame_count % self.prediction_interval == 0 and len(self.keypoints_buffer) > 5:
             self._make_prediction()
-            self._update_word_sequence()
+            self._apply_smoothing_and_decide()
         
         # 手部離開時生成句子
         if self.hand_present == False and self.hand_absent_frames == self.hand_absent_threshold and self.word_sequence:
@@ -683,52 +693,103 @@ class SignLanguageRecognizer:
 
     def _extract_features(self, frame):
         """從單一幀提取手部和姿勢特徵"""
-        with self.feature_extractor.mp_holistic.Holistic(
-            static_image_mode=False,
-            model_complexity=1,
-            smooth_landmarks=True,
-            enable_segmentation=False,
-            min_detection_confidence=0.1,
-            min_tracking_confidence=0.1
-        ) as holistic:
-            frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
-            results = holistic.process(frame_rgb)
-            
-            hands_detected = (results.left_hand_landmarks is not None or 
-                             results.right_hand_landmarks is not None)
-            
-            try:
-                keypoints = self.feature_extractor.extract_pose_keypoints(frame, results)
-                return keypoints, hands_detected
-            except Exception as e:
-                return None, hands_detected
+        frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
+        results = self.feature_extractor.holistic.process(frame_rgb)
+        
+        hands_detected = (results.left_hand_landmarks is not None or 
+                          results.right_hand_landmarks is not None)
+        
+        try:
+            keypoints = self.feature_extractor.extract_pose_keypoints(frame, results)
+            return keypoints, hands_detected
+        except Exception as e:
+            return None, hands_detected
 
     def _make_prediction(self):
-        """使用緩衝區中的特徵進行預測"""
+        """使用緩衝區中的特徵進行預測，並更新平滑緩衝"""
         if len(self.keypoints_buffer) < 2:
             return
-            
-        # 優化tensor創建避免效能警告
+        
         keypoints_array = np.array(list(self.keypoints_buffer), dtype=np.float32)
         keypoints_tensor = torch.from_numpy(keypoints_array).unsqueeze(0).to(self.device)
         
         with torch.no_grad():
             outputs = self.model(keypoints_tensor)
             probabilities = torch.nn.functional.softmax(outputs, dim=1)
-            
             max_prob, predicted_class = torch.max(probabilities, 1)
             predicted_class = predicted_class.item()
             max_prob = max_prob.item()
-            
             probs = probabilities[0].cpu().numpy()
         
-        if max_prob >= self.threshold:
+        # 更新 EMA 信心
+        self.ema_confidence = self.ema_alpha * max_prob + (1 - self.ema_alpha) * self.ema_confidence
+        
+        # 記錄近N次 top1，供投票平滑
+        self.recent_top1_queue.append(predicted_class)
+        
+        # 動態 threshold：手不存在或 EMA 偏低時提高門檻
+        dyn_thr = self.threshold
+        if not self.hand_present:
+            dyn_thr = min(0.95, dyn_thr + 0.15)
+        if self.ema_confidence < 0.5:
+            dyn_thr = min(0.9, dyn_thr + 0.1)
+        self.dynamic_threshold = dyn_thr
+        
+        if max_prob >= dyn_thr:
             self.current_prediction = predicted_class
             self.prediction_probabilities = probs
         else:
             self.current_prediction = -1
             self.prediction_probabilities = probs
 
+    def _apply_smoothing_and_decide(self):
+        """多幀投票 + 冷卻控制，縮減抖動後再加入字串"""
+        if self.current_prediction is None:
+            return
+        
+        # 多幀投票：取近N幀最多數的類別
+        if len(self.recent_top1_queue) >= max(5, self.recent_top1_queue.maxlen // 2):
+            counts = collections.Counter(self.recent_top1_queue)
+            voted_class, voted_count = counts.most_common(1)[0]
+            vote_ratio = voted_count / len(self.recent_top1_queue)
+        else:
+            voted_class, vote_ratio = self.current_prediction, 0.0
+        
+        # 分級門檻：投票占比與 EMA 信心共同決策
+        strong = (vote_ratio >= 0.6 and self.ema_confidence >= 0.6)
+        medium = (vote_ratio >= 0.5 and self.ema_confidence >= 0.5)
+        weak = (vote_ratio >= 0.4 and self.ema_confidence >= 0.45)
+        
+        decided_class = -1
+        if strong or medium or weak:
+            decided_class = voted_class
+        
+        # 產生單詞
+        if decided_class >= 0:
+            self.current_prediction = decided_class
+            self._update_word_sequence()
+
+def load_label_mapping_from_csv(labels_file: str = LABELS_PATH):
+    """從 labels.csv 統一載入標籤映射；失敗則回退到預設。"""
+    label_map = {}
+    print(f"🔍 嘗試載入標籤檔案: {labels_file}")
+    if os.path.exists(labels_file):
+        try:
+            df = pd.read_csv(labels_file)
+            for _, row in df.iterrows():
+                label_map[int(row['index'])] = row['label']
+            print(f"✅ 從 {labels_file} 載入了 {len(label_map)} 個類別標籤")
+            print(f"📊 標籤映射: {label_map}")
+        except Exception as e:
+            print(f"❌ 讀取 labels.csv 出錯: {e}")
+    else:
+        print(f"❌ 標籤檔案不存在: {labels_file}")
+    
+    if not label_map:
+        label_map = {0: "eat", 1: "fish", 2: "like", 3: "want"}
+        print(f"⚠️ 使用預設標籤映射: {label_map}")
+    return label_map
+
 def initialize_recognizer():
     global recognizer
     
@@ -762,7 +823,7 @@ def gen_frames():
             
             socketio.emit('update_frame', {'image': frame_data, 'status': status})
             
-            time.sleep(0.03)  # 約30 FPS
+            time.sleep(0.1)  # 約10 FPS，降低頻寬與CPU
 
 #--------------------
 # 路由定義
@@ -805,6 +866,13 @@ def verify_webhook():
 def handle_webhook():
     """處理從 Messenger 來的訊息"""
     try:
+        # 驗證 Facebook 簽章
+        if APP_SECRET:
+            signature = request.headers.get('X-Hub-Signature-256')
+            if not _verify_facebook_signature(signature, request.data, APP_SECRET):
+                print("簽章驗證失敗")
+                return "簽章驗證失敗", 403
+
         data = request.get_json()
         
         if data.get('object') == 'page':
@@ -821,6 +889,18 @@ def handle_webhook():
         print(f"處理 webhook 時發生錯誤: {e}")
         return "錯誤", 500
 
+def _verify_facebook_signature(signature_header: str, payload: bytes, app_secret: str) -> bool:
+    """驗證 X-Hub-Signature-256 簽章（Facebook Webhook 安全）"""
+    try:
+        if not signature_header or not signature_header.startswith('sha256='):
+            return False
+        received_sig = signature_header.split('=')[1]
+        mac = hmac.new(app_secret.encode('utf-8'), msg=payload, digestmod=hashlib.sha256)
+        expected_sig = mac.hexdigest()
+        return hmac.compare_digest(received_sig, expected_sig)
+    except Exception:
+        return False
+
 @app.route('/receive_recognition_result', methods=['POST'])
 def receive_recognition_result():
     """接收手語辨識結果（內部呼叫）"""
@@ -866,6 +946,13 @@ def process_video():
         
         if video_file.filename == '':
             return jsonify({"status": "error", "message": "沒有選擇檔案"}), 400
+
+        # 基本 MIME 與副檔名檢查
+        allowed_exts = {'.mp4', '.mov', '.avi', '.wmv', '.mkv'}
+        _, ext = os.path.splitext(video_file.filename.lower())
+        content_type = (video_file.content_type or '').lower()
+        if ext not in allowed_exts and not content_type.startswith('video/'):
+            return jsonify({"status": "error", "message": "不支援的影片格式"}), 400
         
         # 使用臨時檔案避免權限問題
         import tempfile
@@ -1067,6 +1154,12 @@ def process_messenger_video(video_url, sender_id):
             send_message(sender_id, generated_sentence)
         else:
             send_message(sender_id, "抱歉，無法辨識您的手語內容，請再試一次。")
+        
+        # 釋放 Mediapipe 資源
+        try:
+            video_recognizer.feature_extractor.close()
+        except Exception:
+            pass
             
     except Exception as e:
         print(f"處理 Messenger 影片時發生錯誤：{e}")
@@ -1134,11 +1227,11 @@ def handle_stop_stream(data):
 #--------------------
 if __name__ == '__main__':
     # HuggingFace Spaces 環境檢測
-    port = int(os.environ.get('PORT', 7860))  # HuggingFace 預設端口
+    port = int(CONFIG.get('PORT', 7860))  # HuggingFace 預設端口
     
     print("🚀 手語辨識整合系統啟動中...")
     print(f"📱 Messenger Bot: {'已配置' if PAGE_ACCESS_TOKEN != 'your_page_access_token' else '未配置'}")
-    print(f"🤖 OpenAI API: {'已配置' if os.environ.get('OPENAI_API_KEY') else '未配置'}")
-    print(f"🔧 運行模式: {'HuggingFace Spaces' if port == 7860 else '本地開發'}")
+    print(f"🤖 OpenAI API: {'已配置' if CONFIG.get('OPENAI_API_KEY') else '未配置'}")
+    print(f"🔧 運行模式: {'HuggingFace Spaces' if port == 7860 else '本地開發'} | SocketIO: {ASYNC_MODE}")
     
-    socketio.run(app, host='0.0.0.0', port=port, debug=False, allow_unsafe_werkzeug=True) 
+    socketio.run(app, host='0.0.0.0', port=port, debug=CONFIG.get('DEBUG', False))

app_config.py

@@ -23,6 +23,7 @@ DEFAULT_CONFIG = {
 def get_config():
     return {
         "OPENAI_API_KEY": os.environ.get("OPENAI_API_KEY"),
+        "APP_SECRET": os.environ.get("APP_SECRET"),
         "VERIFY_TOKEN": os.environ.get("VERIFY_TOKEN", "your_verify_token"),
         "PAGE_ACCESS_TOKEN": os.environ.get("PAGE_ACCESS_TOKEN", "your_page_access_token"),
         "PORT": int(os.environ.get("PORT", 7860)),

requirements.txt

@@ -1,5 +1,6 @@
 flask>=2.3.0,<3.1.0
 flask-socketio>=5.3.0,<6.0.0
+eventlet>=0.33.0,<0.34.0
 opencv-python-headless>=4.8.0,<5.0.0
 numpy>=1.21.0,<2.0.0
 pandas>=1.5.0,<3.0.0