#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 通知服務 Author: PANJIT IT Team Created: 2024-01-28 Modified: 2024-01-28 """ import os import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart 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__) class NotificationService: """通知服務""" def __init__(self): self.smtp_server = current_app.config.get('SMTP_SERVER') self.smtp_port = current_app.config.get('SMTP_PORT', 587) self.use_tls = current_app.config.get('SMTP_USE_TLS', False) self.use_ssl = current_app.config.get('SMTP_USE_SSL', False) self.auth_required = current_app.config.get('SMTP_AUTH_REQUIRED', False) self.sender_email = current_app.config.get('SMTP_SENDER_EMAIL') self.sender_password = current_app.config.get('SMTP_SENDER_PASSWORD', '') self.app_name = current_app.config.get('APP_NAME', 'PANJIT Document Translator') def _create_smtp_connection(self): """建立 SMTP 連線""" try: if self.use_ssl: server = smtplib.SMTP_SSL(self.smtp_server, self.smtp_port) else: server = smtplib.SMTP(self.smtp_server, self.smtp_port) if self.use_tls: server.starttls() if self.auth_required and self.sender_password: server.login(self.sender_email, self.sender_password) return server except Exception as e: logger.error(f"SMTP connection failed: {str(e)}") return None def _send_email(self, to_email: str, subject: str, html_content: str, text_content: str = None) -> bool: """發送郵件的基礎方法""" try: if not self.smtp_server or not self.sender_email: logger.error("SMTP configuration incomplete") return False # 建立郵件 msg = MIMEMultipart('alternative') msg['From'] = f"{self.app_name} <{self.sender_email}>" msg['To'] = to_email msg['Subject'] = subject # 添加文本內容 if text_content: text_part = MIMEText(text_content, 'plain', 'utf-8') msg.attach(text_part) # 添加 HTML 內容 html_part = MIMEText(html_content, 'html', 'utf-8') msg.attach(html_part) # 發送郵件 server = self._create_smtp_connection() if not server: return False server.send_message(msg) server.quit() logger.info(f"Email sent successfully to {to_email}") return True except Exception as e: logger.error(f"Failed to send email to {to_email}: {str(e)}") return False def send_job_completion_notification(self, job: TranslationJob) -> bool: """發送任務完成通知""" try: if not job.user or not job.user.email: logger.warning(f"No email address for job {job.job_uuid}") return False # 準備郵件內容 subject = f"📄 翻譯完成通知 - {job.original_filename}" # 計算處理時間 processing_time = "" if job.processing_started_at and job.completed_at: duration = job.completed_at - job.processing_started_at total_seconds = int(duration.total_seconds()) if total_seconds < 60: processing_time = f"{total_seconds}秒" elif total_seconds < 3600: minutes = total_seconds // 60 seconds = total_seconds % 60 processing_time = f"{minutes}分{seconds}秒" else: hours = total_seconds // 3600 minutes = (total_seconds % 3600) // 60 processing_time = f"{hours}小時{minutes}分" # 生成下載連結（簡化版本） download_links = [] for lang in job.target_languages: download_links.append(f"• {lang}: [下載翻譯檔案]") html_content = f"""

🎉 翻譯任務完成

親愛的 {job.user.display_name}，

您的文件翻譯任務已成功完成！

📋 任務詳細資訊

檔案名稱: {job.original_filename}

任務編號: {job.job_uuid}

來源語言: {job.source_language}

目標語言: {', '.join(job.target_languages)}

處理時間: {processing_time}

完成時間: {job.completed_at.strftime('%Y-%m-%d %H:%M:%S') if job.completed_at else '未知'}

{f'

總成本: ${job.total_cost:.4f}

' if job.total_cost else ''}

📥 下載翻譯檔案

請登入系統下載您的翻譯檔案：

{'
'.join(download_links)}

注意: 翻譯檔案將在系統中保留 7 天，請及時下載。

感謝您使用 {self.app_name}！

如有任何問題，請聯繫系統管理員。

此郵件由 {self.app_name} 系統自動發送，請勿回覆。

發送時間: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}

""" # 純文字版本 text_content = f""" 翻譯任務完成通知 親愛的 {job.user.display_name}， 您的文件翻譯任務已成功完成！ 任務詳細資訊: - 檔案名稱: {job.original_filename} - 任務編號: {job.job_uuid} - 來源語言: {job.source_language} - 目標語言: {', '.join(job.target_languages)} - 處理時間: {processing_time} - 完成時間: {job.completed_at.strftime('%Y-%m-%d %H:%M:%S') if job.completed_at else '未知'} 請登入系統下載您的翻譯檔案。翻譯檔案將在系統中保留 7 天。 感謝您使用 {self.app_name}！ ---- 此郵件由系統自動發送，請勿回覆。 """ return self._send_email(job.user.email, subject, html_content, text_content) except Exception as e: logger.error(f"Failed to send completion notification for job {job.job_uuid}: {str(e)}") return False def send_job_failure_notification(self, job: TranslationJob) -> bool: """發送任務失敗通知""" try: if not job.user or not job.user.email: logger.warning(f"No email address for job {job.job_uuid}") return False subject = f"⚠️ 翻譯失敗通知 - {job.original_filename}" html_content = f"""

❌ 翻譯任務失敗

親愛的 {job.user.display_name}，

很抱歉，您的文件翻譯任務處理失敗。

📋 任務資訊

檔案名稱: {job.original_filename}

任務編號: {job.job_uuid}

重試次數: {job.retry_count}

錯誤訊息: {job.error_message or '未知錯誤'}

失敗時間: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}

建議處理方式:

  • 檢查檔案格式是否正確
  • 確認檔案沒有損壞
  • 稍後再次嘗試上傳
  • 如問題持續，請聯繫系統管理員

如需協助，請聯繫系統管理員。

此郵件由 {self.app_name} 系統自動發送，請勿回覆。

發送時間: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}

""" text_content = f""" 翻譯任務失敗通知 親愛的 {job.user.display_name}， 很抱歉，您的文件翻譯任務處理失敗。 任務資訊: - 檔案名稱: {job.original_filename} - 任務編號: {job.job_uuid} - 重試次數: {job.retry_count} - 錯誤訊息: {job.error_message or '未知錯誤'} 建議處理方式: 1. 檢查檔案格式是否正確 2. 確認檔案沒有損壞 3. 稍後再次嘗試上傳 4. 如問題持續，請聯繫系統管理員 如需協助，請聯繫系統管理員。 ---- 此郵件由 {self.app_name} 系統自動發送，請勿回覆。 """ return self._send_email(job.user.email, subject, html_content, text_content) except Exception as e: logger.error(f"Failed to send failure notification for job {job.job_uuid}: {str(e)}") return False def send_admin_notification(self, subject: str, message: str, admin_emails: List[str] = None) -> bool: """發送管理員通知""" try: if not admin_emails: # 取得所有管理員郵件地址 admin_users = User.get_admin_users() admin_emails = [user.email for user in admin_users if user.email] if not admin_emails: logger.warning("No admin email addresses found") return False html_content = f"""

🔔 系統管理通知

系統管理員您好，

{subject}

{message}

發送時間: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}

此郵件由 {self.app_name} 系統自動發送，請勿回覆。

""" success_count = 0 for email in admin_emails: if self._send_email(email, f"[管理通知] {subject}", html_content): success_count += 1 return success_count > 0 except Exception as e: logger.error(f"Failed to send admin notification: {str(e)}") return False def test_smtp_connection(self) -> bool: """測試 SMTP 連線""" try: server = self._create_smtp_connection() if server: server.quit() return True return False except Exception as e: logger.error(f"SMTP connection test failed: {str(e)}") 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_completion_db_notification_direct(self, job: TranslationJob) -> Optional[Notification]: """ 直接發送任務完成的資料庫通知（不檢查狀態） """ try: # 構建通知內容 title = "翻譯任務完成" message = f'您的文件「{job.original_filename}」已成功翻譯完成。' # 添加目標語言信息 if job.target_languages: languages = ', '.join(job.target_languages) message += f" 目標語言: {languages}" message += " 您可以在任務列表中下載翻譯結果。" # 創建資料庫通知 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