12th_fix error

app/__init__.py

@@ -125,7 +125,7 @@ def create_app(config_name=None):
     # 建立資料表
     with app.app_context():
         # 導入模型
-        from app.models import User, TranslationJob, JobFile, TranslationCache, APIUsageStats, SystemLog
+        from app.models import User, TranslationJob, JobFile, TranslationCache, APIUsageStats, SystemLog, Notification
         
         db.create_all()
         
@@ -135,6 +135,10 @@ def create_app(config_name=None):
     # 創建 Celery 實例
     app.celery = make_celery(app)
     
+    # 初始化 WebSocket
+    from app.websocket import init_websocket
+    app.socketio = init_websocket(app)
+    
     app.logger.info("Flask application created successfully")
     return app
 

app/api/__init__.py

@@ -14,11 +14,12 @@ from flask import Blueprint
 api_v1 = Blueprint('api_v1', __name__, url_prefix='/api/v1')
 
 # 匯入各 API 模組
-from . import auth, jobs, files, admin, health
+from . import auth, jobs, files, admin, health, notification
 
 # 註冊路由
 api_v1.register_blueprint(auth.auth_bp)
 api_v1.register_blueprint(jobs.jobs_bp)
 api_v1.register_blueprint(files.files_bp)
 api_v1.register_blueprint(admin.admin_bp)
-api_v1.register_blueprint(health.health_bp)
+api_v1.register_blueprint(health.health_bp)
+api_v1.register_blueprint(notification.notification_bp)

app/api/notification.py

@@ -0,0 +1,331 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+通知系統 API 路由
+
+Author: PANJIT IT Team
+Created: 2024-01-28
+Modified: 2024-01-28
+"""
+
+from flask import Blueprint, jsonify, request, g
+from flask_jwt_extended import jwt_required, get_jwt_identity
+from sqlalchemy import desc, and_, or_
+from datetime import datetime, timedelta
+from app import db
+from app.models import Notification, NotificationType, User
+from app.utils.response import create_taiwan_response
+# 移除不需要的導入
+
+# 建立藍圖
+notification_bp = Blueprint('notification', __name__, url_prefix='/notifications')
+
+
+@notification_bp.route('', methods=['GET'])
+@jwt_required()
+def get_notifications():
+    """獲取當前用戶的通知列表"""
+    try:
+        # 獲取當前用戶
+        current_user_id = get_jwt_identity()
+        
+        # 獲取查詢參數
+        page = request.args.get('page', 1, type=int)
+        per_page = min(request.args.get('per_page', 20, type=int), 100)
+        status_filter = request.args.get('status', 'all')
+        type_filter = request.args.get('type', None)
+        
+        # 建構查詢
+        query = Notification.query.filter_by(user_id=current_user_id)
+        
+        # 只顯示未過期的通知
+        query = query.filter(or_(
+            Notification.expires_at.is_(None),
+            Notification.expires_at > datetime.now()
+        ))
+        
+        # 過濾狀態
+        if status_filter == 'unread':
+            query = query.filter_by(is_read=False)
+        elif status_filter == 'read':
+            query = query.filter_by(is_read=True)
+        
+        # 過濾類型
+        if type_filter:
+            query = query.filter_by(type=type_filter)
+        
+        # 排序 - 未讀在前，然後按時間排序
+        query = query.order_by(Notification.is_read.asc(), desc(Notification.created_at))
+        
+        # 分頁
+        paginated = query.paginate(
+            page=page, per_page=per_page, error_out=False
+        )
+        
+        # 獲取未讀數量
+        unread_count = Notification.query.filter_by(
+            user_id=current_user_id,
+            is_read=False
+        ).filter(or_(
+            Notification.expires_at.is_(None),
+            Notification.expires_at > datetime.now()
+        )).count()
+        
+        return jsonify(create_taiwan_response(
+            success=True,
+            data={
+                'notifications': [n.to_dict() for n in paginated.items],
+                'pagination': {
+                    'total': paginated.total,
+                    'page': page,
+                    'per_page': per_page,
+                    'pages': paginated.pages
+                },
+                'unread_count': unread_count
+            },
+            message='獲取通知列表成功'
+        ))
+        
+    except Exception as e:
+        return jsonify(create_taiwan_response(
+            success=False,
+            error=f'獲取通知失敗：{str(e)}'
+        )), 500
+
+
+@notification_bp.route('/<notification_id>', methods=['GET'])
+@jwt_required()
+def get_notification(notification_id):
+    """獲取單個通知詳情"""
+    try:
+        current_user_id = get_jwt_identity()
+        
+        # 查找通知
+        notification = Notification.query.filter_by(
+            notification_uuid=notification_id,
+            user_id=current_user_id
+        ).first()
+        
+        if not notification:
+            return jsonify(create_taiwan_response(
+                success=False,
+                error='通知不存在'
+            )), 404
+        
+        # 自動標記為已讀
+        if not notification.is_read:
+            notification.mark_as_read()
+            db.session.commit()
+        
+        return jsonify(create_taiwan_response(
+            success=True,
+            data=notification.to_dict(),
+            message='獲取通知成功'
+        ))
+        
+    except Exception as e:
+        return jsonify(create_taiwan_response(
+            success=False,
+            error=f'獲取通知失敗：{str(e)}'
+        )), 500
+
+
+@notification_bp.route('/<notification_id>/read', methods=['POST'])
+@jwt_required()
+def mark_notification_read(notification_id):
+    """標記通知為已讀"""
+    try:
+        current_user_id = get_jwt_identity()
+        
+        # 查找通知
+        notification = Notification.query.filter_by(
+            notification_uuid=notification_id,
+            user_id=current_user_id
+        ).first()
+        
+        if not notification:
+            return jsonify(create_taiwan_response(
+                success=False,
+                error='通知不存在'
+            )), 404
+        
+        # 標記為已讀
+        notification.mark_as_read()
+        db.session.commit()
+        
+        return jsonify(create_taiwan_response(
+            success=True,
+            message='標記已讀成功'
+        ))
+        
+    except Exception as e:
+        return jsonify(create_taiwan_response(
+            success=False,
+            error=f'標記已讀失敗：{str(e)}'
+        )), 500
+
+
+@notification_bp.route('/read-all', methods=['POST'])
+@jwt_required()
+def mark_all_read():
+    """標記所有通知為已讀"""
+    try:
+        current_user_id = get_jwt_identity()
+        
+        # 取得所有未讀通知
+        unread_notifications = Notification.query.filter_by(
+            user_id=current_user_id,
+            is_read=False
+        ).filter(or_(
+            Notification.expires_at.is_(None),
+            Notification.expires_at > datetime.now()
+        )).all()
+        
+        # 標記為已讀
+        for notification in unread_notifications:
+            notification.mark_as_read()
+        
+        db.session.commit()
+        
+        return jsonify(create_taiwan_response(
+            success=True,
+            data={'marked_count': len(unread_notifications)},
+            message=f'已標記 {len(unread_notifications)} 個通知為已讀'
+        ))
+        
+    except Exception as e:
+        return jsonify(create_taiwan_response(
+            success=False,
+            error=f'標記全部已讀失敗：{str(e)}'
+        )), 500
+
+
+@notification_bp.route('/<notification_id>', methods=['DELETE'])
+@jwt_required()
+def delete_notification(notification_id):
+    """刪除通知"""
+    try:
+        current_user_id = get_jwt_identity()
+        
+        # 查找通知
+        notification = Notification.query.filter_by(
+            notification_uuid=notification_id,
+            user_id=current_user_id
+        ).first()
+        
+        if not notification:
+            return jsonify(create_taiwan_response(
+                success=False,
+                error='通知不存在'
+            )), 404
+        
+        # 刪除通知
+        db.session.delete(notification)
+        db.session.commit()
+        
+        return jsonify(create_taiwan_response(
+            success=True,
+            message='刪除通知成功'
+        ))
+        
+    except Exception as e:
+        db.session.rollback()
+        return jsonify(create_taiwan_response(
+            success=False,
+            error=f'刪除通知失敗：{str(e)}'
+        )), 500
+
+
+@notification_bp.route('/clear', methods=['POST'])
+@jwt_required()
+def clear_read_notifications():
+    """清空所有已讀通知"""
+    try:
+        current_user_id = get_jwt_identity()
+        
+        # 刪除所有已讀通知
+        deleted_count = Notification.query.filter_by(
+            user_id=current_user_id,
+            is_read=True
+        ).delete()
+        
+        db.session.commit()
+        
+        return jsonify(create_taiwan_response(
+            success=True,
+            data={'deleted_count': deleted_count},
+            message=f'已清除 {deleted_count} 個已讀通知'
+        ))
+        
+    except Exception as e:
+        db.session.rollback()
+        return jsonify(create_taiwan_response(
+            success=False,
+            error=f'清除通知失敗：{str(e)}'
+        )), 500
+
+
+@notification_bp.route('/test', methods=['POST'])
+@jwt_required()
+def create_test_notification():
+    """創建測試通知（開發用）"""
+    try:
+        current_user_id = get_jwt_identity()
+        
+        # 創建測試通知
+        test_notification = create_notification(
+            user_id=current_user_id,
+            title="測試通知",
+            message=f"這是一個測試通知，創建於 {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
+            notification_type=NotificationType.INFO
+        )
+        
+        return jsonify(create_taiwan_response(
+            success=True,
+            data=test_notification.to_dict(),
+            message='測試通知已創建'
+        ))
+        
+    except Exception as e:
+        return jsonify(create_taiwan_response(
+            success=False,
+            error=f'創建測試通知失敗：{str(e)}'
+        )), 500
+
+
+# 工具函數：創建通知
+def create_notification(user_id, title, message, notification_type=NotificationType.INFO, 
+                       job_uuid=None, extra_data=None):
+    """
+    創建通知的工具函數
+    
+    Args:
+        user_id: 用戶ID
+        title: 通知標題
+        message: 通知內容
+        notification_type: 通知類型
+        job_uuid: 關聯的任務UUID（可選）
+        extra_data: 額外數據（可選）
+    
+    Returns:
+        Notification: 創建的通知對象
+    """
+    try:
+        notification = Notification(
+            user_id=user_id,
+            type=notification_type.value,
+            title=title,
+            message=message,
+            job_uuid=job_uuid,
+            extra_data=extra_data,
+            link=f"/job/{job_uuid}" if job_uuid else None
+        )
+        
+        db.session.add(notification)
+        db.session.commit()
+        
+        return notification
+        
+    except Exception as e:
+        db.session.rollback()
+        raise e

app/models/__init__.py

@@ -13,6 +13,7 @@ from .job import TranslationJob, JobFile
 from .cache import TranslationCache
 from .stats import APIUsageStats
 from .log import SystemLog
+from .notification import Notification, NotificationType
 
 __all__ = [
     'User',
@@ -20,5 +21,7 @@ __all__ = [
     'JobFile',
     'TranslationCache',
     'APIUsageStats',
-    'SystemLog'
+    'SystemLog',
+    'Notification',
+    'NotificationType'
 ]

app/models/notification.py

@@ -0,0 +1,97 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+通知系統資料模型
+
+Author: PANJIT IT Team
+Created: 2024-01-28
+Modified: 2024-01-28
+"""
+from datetime import datetime
+from enum import Enum
+from sqlalchemy import func
+from sqlalchemy.orm import relationship
+from app import db
+import uuid
+import json
+
+
+class NotificationType(str, Enum):
+    """通知類型枚舉"""
+    SUCCESS = "success"  # 成功
+    ERROR = "error"      # 錯誤
+    WARNING = "warning"  # 警告
+    INFO = "info"        # 資訊
+    SYSTEM = "system"    # 系統
+
+
+class Notification(db.Model):
+    """通知模型"""
+    __tablename__ = 'dt_notifications'
+    
+    # 主鍵
+    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
+    notification_uuid = db.Column(db.String(36), unique=True, nullable=False, index=True, 
+                                 default=lambda: str(uuid.uuid4()), comment='通知唯一識別碼')
+    
+    # 基本資訊
+    user_id = db.Column(db.Integer, db.ForeignKey('dt_users.id'), nullable=False, comment='使用者ID')
+    type = db.Column(db.String(20), nullable=False, default=NotificationType.INFO.value, comment='通知類型')
+    title = db.Column(db.String(255), nullable=False, comment='通知標題')
+    message = db.Column(db.Text, nullable=False, comment='通知內容')
+    
+    # 關聯資訊（可選）
+    job_uuid = db.Column(db.String(36), nullable=True, comment='關聯任務UUID')
+    link = db.Column(db.String(500), nullable=True, comment='相關連結')
+    
+    # 狀態
+    is_read = db.Column(db.Boolean, default=False, nullable=False, comment='是否已讀')
+    read_at = db.Column(db.DateTime, nullable=True, comment='閱讀時間')
+    
+    # 時間戳記
+    created_at = db.Column(db.DateTime, default=func.now(), nullable=False, comment='建立時間')
+    expires_at = db.Column(db.DateTime, nullable=True, comment='過期時間')
+    
+    # 額外數據（JSON 格式儲存）
+    extra_data = db.Column(db.JSON, nullable=True, comment='額外數據')
+    
+    # 關聯
+    user = db.relationship("User", backref="notifications")
+    
+    def __repr__(self):
+        return f"<Notification {self.notification_uuid}: {self.title}>"
+    
+    def to_dict(self):
+        """轉換為字典格式"""
+        return {
+            'id': self.notification_uuid,  # 前端使用 UUID
+            'user_id': self.user_id,
+            'type': self.type,
+            'title': self.title,
+            'message': self.message,
+            'job_uuid': self.job_uuid,
+            'link': self.link,
+            'is_read': self.is_read,
+            'read': self.is_read,  # 為了前端相容
+            'read_at': self.read_at.isoformat() if self.read_at else None,
+            'created_at': self.created_at.isoformat() if self.created_at else None,
+            'expires_at': self.expires_at.isoformat() if self.expires_at else None,
+            'extra_data': self.extra_data
+        }
+    
+    def mark_as_read(self):
+        """標記為已讀"""
+        self.is_read = True
+        self.read_at = datetime.now()
+        
+    @classmethod
+    def create_job_notification(cls, user_id, job_uuid, title, message, notification_type=NotificationType.INFO):
+        """創建任務相關通知"""
+        return cls(
+            user_id=user_id,
+            job_uuid=job_uuid,
+            type=notification_type.value,
+            title=title,
+            message=message,
+            link=f"/job/{job_uuid}"  # 連結到任務詳情頁
+        )

app/services/notification_service.py

@@ -12,12 +12,14 @@ import os
 import smtplib
 from email.mime.text import MIMEText
 from email.mime.multipart import MIMEMultipart
-from datetime import datetime
-from typing import Optional, List
+from datetime import datetime, timedelta
+from typing import Optional, List, Dict, Any
 from flask import current_app, url_for
+from app import db
 from app.utils.logger import get_logger
 from app.models.job import TranslationJob
 from app.models.user import User
+from app.models.notification import Notification, NotificationType
 
 logger = get_logger(__name__)
 
@@ -385,4 +387,224 @@ class NotificationService:
             return False
         except Exception as e:
             logger.error(f"SMTP connection test failed: {str(e)}")
-            return False
+            return False
+    
+    # ========== 資料庫通知方法 ==========
+    
+    def create_db_notification(
+        self,
+        user_id: int,
+        title: str,
+        message: str,
+        notification_type: NotificationType = NotificationType.INFO,
+        job_uuid: Optional[str] = None,
+        extra_data: Optional[Dict[str, Any]] = None,
+        expires_at: Optional[datetime] = None,
+        link: Optional[str] = None
+    ) -> Optional[Notification]:
+        """
+        創建資料庫通知
+        
+        Args:
+            user_id: 用戶ID
+            title: 通知標題
+            message: 通知內容
+            notification_type: 通知類型
+            job_uuid: 關聯任務UUID
+            extra_data: 額外數據
+            expires_at: 過期時間
+            link: 相關連結
+        
+        Returns:
+            Notification: 創建的通知對象
+        """
+        try:
+            # 如果沒有指定連結但有任務UUID，自動生成任務詳情連結
+            if not link and job_uuid:
+                link = f"/job/{job_uuid}"
+            
+            notification = Notification(
+                user_id=user_id,
+                type=notification_type.value,
+                title=title,
+                message=message,
+                job_uuid=job_uuid,
+                link=link,
+                extra_data=extra_data,
+                expires_at=expires_at
+            )
+            
+            db.session.add(notification)
+            db.session.commit()
+            
+            logger.info(f"資料庫通知已創建: {notification.notification_uuid} for user {user_id}")
+            
+            # 觸發 WebSocket 推送
+            self._send_websocket_notification(notification)
+            
+            return notification
+            
+        except Exception as e:
+            db.session.rollback()
+            logger.error(f"創建資料庫通知失敗: {e}")
+            return None
+    
+    def send_job_started_db_notification(self, job: TranslationJob) -> Optional[Notification]:
+        """
+        發送任務開始處理的資料庫通知
+        
+        Args:
+            job: 翻譯任務對象
+        
+        Returns:
+            Notification: 創建的通知對象
+        """
+        try:
+            title = "翻譯任務開始處理"
+            message = f'您的文件「{job.original_filename}」已開始翻譯處理。'
+            
+            if job.target_languages:
+                languages = ', '.join(job.target_languages)
+                message += f" 目標語言: {languages}"
+            
+            return self.create_db_notification(
+                user_id=job.user_id,
+                title=title,
+                message=message,
+                notification_type=NotificationType.INFO,
+                job_uuid=job.job_uuid,
+                extra_data={
+                    'filename': job.original_filename,
+                    'target_languages': job.target_languages,
+                    'started_at': job.processing_started_at.isoformat() if job.processing_started_at else None
+                }
+            )
+            
+        except Exception as e:
+            logger.error(f"發送任務開始資料庫通知失敗: {e}")
+            return None
+    
+    def send_job_completion_db_notification(self, job: TranslationJob) -> Optional[Notification]:
+        """
+        發送任務完成的資料庫通知
+        
+        Args:
+            job: 翻譯任務對象
+        
+        Returns:
+            Notification: 創建的通知對象
+        """
+        try:
+            if job.status != 'COMPLETED':
+                logger.warning(f"任務 {job.job_uuid} 狀態不是已完成，跳過完成通知")
+                return None
+            
+            # 構建通知內容
+            title = "翻譯任務完成"
+            message = f'您的文件「{job.original_filename}」已成功翻譯完成。'
+            
+            # 添加目標語言信息
+            if job.target_languages:
+                languages = ', '.join(job.target_languages)
+                message += f" 目標語言: {languages}"
+            
+            # 添加處理時間信息
+            if job.processing_started_at and job.completed_at:
+                duration = job.completed_at - job.processing_started_at
+                minutes = int(duration.total_seconds() / 60)
+                if minutes > 0:
+                    message += f" 處理時間: {minutes} 分鐘"
+                else:
+                    message += f" 處理時間: {int(duration.total_seconds())} 秒"
+            
+            return self.create_db_notification(
+                user_id=job.user_id,
+                title=title,
+                message=message,
+                notification_type=NotificationType.SUCCESS,
+                job_uuid=job.job_uuid,
+                extra_data={
+                    'filename': job.original_filename,
+                    'target_languages': job.target_languages,
+                    'total_cost': float(job.total_cost) if job.total_cost else 0,
+                    'completed_at': job.completed_at.isoformat() if job.completed_at else None
+                }
+            )
+            
+        except Exception as e:
+            logger.error(f"發送任務完成資料庫通知失敗: {e}")
+            return None
+    
+    def send_job_failure_db_notification(self, job: TranslationJob, error_message: str = None) -> Optional[Notification]:
+        """
+        發送任務失敗的資料庫通知
+        
+        Args:
+            job: 翻譯任務對象
+            error_message: 錯誤訊息
+        
+        Returns:
+            Notification: 創建的通知對象
+        """
+        try:
+            title = "翻譯任務失敗"
+            message = f'您的文件「{job.original_filename}」翻譯失敗。'
+            
+            if error_message:
+                message += f" 錯誤訊息: {error_message}"
+            
+            if job.retry_count > 0:
+                message += f" 已重試 {job.retry_count} 次。"
+            
+            return self.create_db_notification(
+                user_id=job.user_id,
+                title=title,
+                message=message,
+                notification_type=NotificationType.ERROR,
+                job_uuid=job.job_uuid,
+                extra_data={
+                    'filename': job.original_filename,
+                    'error_message': error_message,
+                    'retry_count': job.retry_count,
+                    'failed_at': datetime.now().isoformat()
+                }
+            )
+            
+        except Exception as e:
+            logger.error(f"發送任務失敗資料庫通知失敗: {e}")
+            return None
+    
+    def _send_websocket_notification(self, notification: Notification):
+        """
+        通過 WebSocket 發送通知
+        
+        Args:
+            notification: 通知對象
+        """
+        try:
+            from app.websocket import send_notification_to_user
+            send_notification_to_user(notification.user_id, notification.to_dict())
+        except Exception as e:
+            logger.error(f"WebSocket 推送通知失敗: {e}")
+    
+    def get_unread_count(self, user_id: int) -> int:
+        """
+        獲取用戶未讀通知數量
+        
+        Args:
+            user_id: 用戶ID
+        
+        Returns:
+            int: 未讀通知數量
+        """
+        try:
+            return Notification.query.filter_by(
+                user_id=user_id,
+                is_read=False
+            ).filter(
+                (Notification.expires_at.is_(None)) | 
+                (Notification.expires_at > datetime.now())
+            ).count()
+        except Exception as e:
+            logger.error(f"獲取未讀通知數量失敗: {e}")
+            return 0

app/tasks/translation.py

@@ -76,7 +76,10 @@ def process_translation_job(self, job_id: int):
                 # 發送完成通知
                 try:
                     notification_service = NotificationService()
+                    # 發送郵件通知
                     notification_service.send_job_completion_notification(job)
+                    # 發送資料庫通知
+                    notification_service.send_job_completion_db_notification(job)
                 except Exception as e:
                     logger.warning(f"Failed to send completion notification: {str(e)}")
                 
@@ -131,6 +134,16 @@ def process_translation_job(self, job_id: int):
                         # 重試次數用盡，標記失敗
                         job.update_status('FAILED')
                         
+                        # 發送失敗通知
+                        try:
+                            notification_service = NotificationService()
+                            # 發送郵件通知
+                            notification_service.send_job_failure_notification(job)
+                            # 發送資料庫通知
+                            notification_service.send_job_failure_db_notification(job, str(exc))
+                        except Exception as e:
+                            logger.warning(f"Failed to send failure notification: {str(e)}")
+                        
                         SystemLog.error(
                             'tasks.translation',
                             f'Translation job failed permanently: {job.job_uuid}',

app/websocket.py

@@ -0,0 +1,230 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+WebSocket 服務模組
+
+Author: PANJIT IT Team
+Created: 2024-01-28
+Modified: 2024-01-28
+"""
+
+from flask_socketio import SocketIO, emit, join_room, leave_room, disconnect
+from flask_jwt_extended import decode_token, get_jwt
+from flask import request
+from functools import wraps
+import logging
+
+# 初始化 SocketIO
+socketio = SocketIO(
+    cors_allowed_origins="*",
+    async_mode='threading',
+    logger=True,
+    engineio_logger=False
+)
+
+# 存儲用戶連接
+connected_users = {}
+
+logger = logging.getLogger(__name__)
+
+
+def jwt_required_ws(f):
+    """WebSocket JWT 驗證裝飾器"""
+    @wraps(f)
+    def decorated_function(*args, **kwargs):
+        try:
+            # 從查詢參數獲取 token
+            token = request.args.get('token')
+            if not token:
+                disconnect()
+                return False
+            
+            # 解碼 token
+            decoded = decode_token(token)
+            user_id = decoded.get('sub')
+            
+            # 儲存用戶信息
+            request.user_id = user_id
+            
+            return f(*args, **kwargs)
+            
+        except Exception as e:
+            logger.error(f"WebSocket authentication failed: {e}")
+            disconnect()
+            return False
+            
+    return decorated_function
+
+
+@socketio.on('connect')
+def handle_connect(auth):
+    """處理客戶端連接"""
+    try:
+        # 從認證數據獲取 token
+        if auth and 'token' in auth:
+            token = auth['token']
+            decoded = decode_token(token)
+            user_id = decoded.get('sub')
+            
+            # 記錄連接
+            connected_users[request.sid] = {
+                'user_id': user_id,
+                'sid': request.sid
+            }
+            
+            # 加入用戶專屬房間
+            join_room(f"user_{user_id}")
+            
+            logger.info(f"User {user_id} connected with session {request.sid}")
+            
+            # 發送連接成功消息
+            emit('connected', {
+                'message': '連接成功',
+                'user_id': user_id
+            })
+            
+            return True
+        else:
+            logger.warning("Connection attempt without authentication")
+            disconnect()
+            return False
+            
+    except Exception as e:
+        logger.error(f"Connection error: {e}")
+        disconnect()
+        return False
+
+
+@socketio.on('disconnect')
+def handle_disconnect():
+    """處理客戶端斷開連接"""
+    try:
+        if request.sid in connected_users:
+            user_info = connected_users[request.sid]
+            user_id = user_info['user_id']
+            
+            # 離開房間
+            leave_room(f"user_{user_id}")
+            
+            # 移除連接記錄
+            del connected_users[request.sid]
+            
+            logger.info(f"User {user_id} disconnected")
+            
+    except Exception as e:
+        logger.error(f"Disconnect error: {e}")
+
+
+@socketio.on('ping')
+def handle_ping():
+    """處理心跳包"""
+    emit('pong', {'timestamp': request.args.get('timestamp')})
+
+
+@socketio.on('subscribe_job')
+def handle_subscribe_job(data):
+    """訂閱任務更新"""
+    try:
+        job_uuid = data.get('job_uuid')
+        if job_uuid:
+            join_room(f"job_{job_uuid}")
+            logger.info(f"Client {request.sid} subscribed to job {job_uuid}")
+            emit('subscribed', {'job_uuid': job_uuid})
+    except Exception as e:
+        logger.error(f"Subscribe job error: {e}")
+
+
+@socketio.on('unsubscribe_job')
+def handle_unsubscribe_job(data):
+    """取消訂閱任務更新"""
+    try:
+        job_uuid = data.get('job_uuid')
+        if job_uuid:
+            leave_room(f"job_{job_uuid}")
+            logger.info(f"Client {request.sid} unsubscribed from job {job_uuid}")
+            emit('unsubscribed', {'job_uuid': job_uuid})
+    except Exception as e:
+        logger.error(f"Unsubscribe job error: {e}")
+
+
+# 工具函數：發送通知
+def send_notification_to_user(user_id, notification_data):
+    """
+    向特定用戶發送通知
+    
+    Args:
+        user_id: 用戶ID
+        notification_data: 通知數據
+    """
+    try:
+        socketio.emit(
+            'new_notification',
+            notification_data,
+            room=f"user_{user_id}",
+            namespace='/'
+        )
+        logger.info(f"Notification sent to user {user_id}")
+        
+    except Exception as e:
+        logger.error(f"Failed to send notification: {e}")
+
+
+def send_job_update(job_uuid, update_data):
+    """
+    發送任務更新
+    
+    Args:
+        job_uuid: 任務UUID
+        update_data: 更新數據
+    """
+    try:
+        socketio.emit(
+            'job_update',
+            {
+                'job_uuid': job_uuid,
+                **update_data
+            },
+            room=f"job_{job_uuid}",
+            namespace='/'
+        )
+        logger.info(f"Job update sent for {job_uuid}")
+        
+    except Exception as e:
+        logger.error(f"Failed to send job update: {e}")
+
+
+def broadcast_system_message(message, message_type='info'):
+    """
+    廣播系統消息給所有連接的用戶
+    
+    Args:
+        message: 消息內容
+        message_type: 消息類型
+    """
+    try:
+        socketio.emit(
+            'system_message',
+            {
+                'message': message,
+                'type': message_type
+            },
+            namespace='/',
+            broadcast=True
+        )
+        logger.info(f"System message broadcasted: {message}")
+        
+    except Exception as e:
+        logger.error(f"Failed to broadcast system message: {e}")
+
+
+# 初始化函數
+def init_websocket(app):
+    """
+    初始化 WebSocket
+    
+    Args:
+        app: Flask 應用實例
+    """
+    socketio.init_app(app)
+    logger.info("WebSocket initialized")
+    return socketio