645 lines
26 KiB
Python
645 lines
26 KiB
Python
#!/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"""
|
||
<!DOCTYPE html>
|
||
<html>
|
||
<head>
|
||
<meta charset="utf-8">
|
||
<style>
|
||
body {{ font-family: Arial, sans-serif; line-height: 1.6; color: #333; }}
|
||
.container {{ max-width: 600px; margin: 0 auto; padding: 20px; }}
|
||
.header {{ background-color: #2563eb; color: white; padding: 20px; text-align: center; border-radius: 8px 8px 0 0; }}
|
||
.content {{ background-color: #f8fafc; padding: 30px; border: 1px solid #e5e7eb; }}
|
||
.info-box {{ background-color: #dbeafe; border-left: 4px solid #2563eb; padding: 15px; margin: 20px 0; }}
|
||
.footer {{ background-color: #374151; color: #d1d5db; padding: 15px; text-align: center; font-size: 12px; border-radius: 0 0 8px 8px; }}
|
||
.success {{ color: #059669; font-weight: bold; }}
|
||
.download-section {{ margin: 20px 0; }}
|
||
.download-link {{ display: inline-block; background-color: #2563eb; color: white; padding: 10px 20px; text-decoration: none; border-radius: 5px; margin: 5px; }}
|
||
</style>
|
||
</head>
|
||
<body>
|
||
<div class="container">
|
||
<div class="header">
|
||
<h1>🎉 翻譯任務完成</h1>
|
||
</div>
|
||
|
||
<div class="content">
|
||
<p>親愛的 <strong>{job.user.display_name}</strong>，</p>
|
||
|
||
<p class="success">您的文件翻譯任務已成功完成！</p>
|
||
|
||
<div class="info-box">
|
||
<h3>📋 任務詳細資訊</h3>
|
||
<p><strong>檔案名稱:</strong> {job.original_filename}</p>
|
||
<p><strong>任務編號:</strong> {job.job_uuid}</p>
|
||
<p><strong>來源語言:</strong> {job.source_language}</p>
|
||
<p><strong>目標語言:</strong> {', '.join(job.target_languages)}</p>
|
||
<p><strong>處理時間:</strong> {processing_time}</p>
|
||
<p><strong>完成時間:</strong> {job.completed_at.strftime('%Y-%m-%d %H:%M:%S') if job.completed_at else '未知'}</p>
|
||
{f'<p><strong>總成本:</strong> ${job.total_cost:.4f}</p>' if job.total_cost else ''}
|
||
</div>
|
||
|
||
<div class="download-section">
|
||
<h3>📥 下載翻譯檔案</h3>
|
||
<p>請登入系統下載您的翻譯檔案：</p>
|
||
<p>{'<br>'.join(download_links)}</p>
|
||
<p style="margin-top: 15px;">
|
||
<strong>注意:</strong> 翻譯檔案將在系統中保留 7 天，請及時下載。
|
||
</p>
|
||
</div>
|
||
|
||
<div style="margin-top: 30px; padding-top: 20px; border-top: 1px solid #e5e7eb;">
|
||
<p>感謝您使用 {self.app_name}！</p>
|
||
<p>如有任何問題，請聯繫系統管理員。</p>
|
||
</div>
|
||
</div>
|
||
|
||
<div class="footer">
|
||
<p>此郵件由 {self.app_name} 系統自動發送，請勿回覆。</p>
|
||
<p>發送時間: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
|
||
</div>
|
||
</div>
|
||
</body>
|
||
</html>
|
||
"""
|
||
|
||
# 純文字版本
|
||
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"""
|
||
<!DOCTYPE html>
|
||
<html>
|
||
<head>
|
||
<meta charset="utf-8">
|
||
<style>
|
||
body {{ font-family: Arial, sans-serif; line-height: 1.6; color: #333; }}
|
||
.container {{ max-width: 600px; margin: 0 auto; padding: 20px; }}
|
||
.header {{ background-color: #dc2626; color: white; padding: 20px; text-align: center; border-radius: 8px 8px 0 0; }}
|
||
.content {{ background-color: #f8fafc; padding: 30px; border: 1px solid #e5e7eb; }}
|
||
.error-box {{ background-color: #fef2f2; border-left: 4px solid #dc2626; padding: 15px; margin: 20px 0; }}
|
||
.footer {{ background-color: #374151; color: #d1d5db; padding: 15px; text-align: center; font-size: 12px; border-radius: 0 0 8px 8px; }}
|
||
.error {{ color: #dc2626; font-weight: bold; }}
|
||
</style>
|
||
</head>
|
||
<body>
|
||
<div class="container">
|
||
<div class="header">
|
||
<h1>❌ 翻譯任務失敗</h1>
|
||
</div>
|
||
|
||
<div class="content">
|
||
<p>親愛的 <strong>{job.user.display_name}</strong>，</p>
|
||
|
||
<p class="error">很抱歉，您的文件翻譯任務處理失敗。</p>
|
||
|
||
<div class="error-box">
|
||
<h3>📋 任務資訊</h3>
|
||
<p><strong>檔案名稱:</strong> {job.original_filename}</p>
|
||
<p><strong>任務編號:</strong> {job.job_uuid}</p>
|
||
<p><strong>重試次數:</strong> {job.retry_count}</p>
|
||
<p><strong>錯誤訊息:</strong> {job.error_message or '未知錯誤'}</p>
|
||
<p><strong>失敗時間:</strong> {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
|
||
</div>
|
||
|
||
<div style="margin-top: 20px;">
|
||
<p><strong>建議處理方式:</strong></p>
|
||
<ul>
|
||
<li>檢查檔案格式是否正確</li>
|
||
<li>確認檔案沒有損壞</li>
|
||
<li>稍後再次嘗試上傳</li>
|
||
<li>如問題持續，請聯繫系統管理員</li>
|
||
</ul>
|
||
</div>
|
||
|
||
<div style="margin-top: 30px; padding-top: 20px; border-top: 1px solid #e5e7eb;">
|
||
<p>如需協助，請聯繫系統管理員。</p>
|
||
</div>
|
||
</div>
|
||
|
||
<div class="footer">
|
||
<p>此郵件由 {self.app_name} 系統自動發送，請勿回覆。</p>
|
||
<p>發送時間: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
|
||
</div>
|
||
</div>
|
||
</body>
|
||
</html>
|
||
"""
|
||
|
||
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"""
|
||
<!DOCTYPE html>
|
||
<html>
|
||
<head>
|
||
<meta charset="utf-8">
|
||
<style>
|
||
body {{ font-family: Arial, sans-serif; line-height: 1.6; color: #333; }}
|
||
.container {{ max-width: 600px; margin: 0 auto; padding: 20px; }}
|
||
.header {{ background-color: #f59e0b; color: white; padding: 20px; text-align: center; border-radius: 8px 8px 0 0; }}
|
||
.content {{ background-color: #f8fafc; padding: 30px; border: 1px solid #e5e7eb; }}
|
||
.footer {{ background-color: #374151; color: #d1d5db; padding: 15px; text-align: center; font-size: 12px; border-radius: 0 0 8px 8px; }}
|
||
</style>
|
||
</head>
|
||
<body>
|
||
<div class="container">
|
||
<div class="header">
|
||
<h1>🔔 系統管理通知</h1>
|
||
</div>
|
||
|
||
<div class="content">
|
||
<p>系統管理員您好，</p>
|
||
|
||
<div style="background-color: #fef3c7; border-left: 4px solid #f59e0b; padding: 15px; margin: 20px 0;">
|
||
<h3>{subject}</h3>
|
||
<p>{message}</p>
|
||
</div>
|
||
|
||
<p>發送時間: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
|
||
</div>
|
||
|
||
<div class="footer">
|
||
<p>此郵件由 {self.app_name} 系統自動發送，請勿回覆。</p>
|
||
</div>
|
||
</div>
|
||
</body>
|
||
</html>
|
||
"""
|
||
|
||
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 |