""" 通知服務模組處理 Email 發送 """ import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from typing import Optional from html import escape from sqlalchemy.orm import Session import logging from app.core.config import settings from app.models import Report, Subscription, User, NotificationLog, NotificationStatus logger = logging.getLogger(__name__) def send_email(to_email: str, subject: str, html_content: str) -> bool: """ 發送 Email Returns: 是否發送成功 """ if not settings.smtp_host: logger.warning("SMTP 未設定，跳過發送") return False try: msg = MIMEMultipart("alternative") msg["Subject"] = subject msg["From"] = f"{settings.smtp_from_name} <{settings.smtp_from_email}>" msg["To"] = to_email html_part = MIMEText(html_content, "html", "utf-8") msg.attach(html_part) with smtplib.SMTP(settings.smtp_host, settings.smtp_port) as server: server.starttls() if settings.smtp_username and settings.smtp_password: server.login(settings.smtp_username, settings.smtp_password) server.sendmail(settings.smtp_from_email, to_email, msg.as_string()) return True except Exception as e: logger.error("Email 發送失敗", exc_info=True) return False def create_report_email_content(report: Report, base_url: str = "") -> str: """建立報告通知 Email 內容""" summary = report.edited_summary or report.ai_summary or "無摘要內容" # 截取摘要前 500 字 if len(summary) > 500: summary = summary[:500] + "..." # 轉義 HTML 特殊字元，防止 XSS safe_title = escape(report.title) safe_group_name = escape(report.group.name) safe_summary = escape(summary) safe_base_url = escape(base_url) html = f"""

{safe_title}

群組：{safe_group_name}
日期：{report.report_date}

摘要

{safe_summary}

閱讀完整報告

""" return html def send_report_notifications(db: Session, report: Report) -> int: """ 發送報告通知給訂閱者 Returns: 發送成功數量 """ # 取得訂閱此群組的用戶 subscriptions = db.query(Subscription).filter( Subscription.group_id == report.group_id, Subscription.email_notify == True ).all() sent_count = 0 for sub in subscriptions: user = db.query(User).filter(User.id == sub.user_id).first() if not user or not user.email or not user.is_active: continue # 建立通知記錄 notification = NotificationLog( user_id=user.id, report_id=report.id, notification_type="email", subject=f"【每日報導】{report.title}", content=report.edited_summary or report.ai_summary ) db.add(notification) # 發送 Email html_content = create_report_email_content(report) success = send_email( user.email, f"【每日報導】{report.title}", html_content ) if success: notification.status = NotificationStatus.SENT from datetime import datetime notification.sent_at = datetime.utcnow() sent_count += 1 else: notification.status = NotificationStatus.FAILED notification.error_message = "發送失敗" db.commit() return sent_count def send_delay_notification(db: Session, report: Report) -> int: """ 發送延遲發布通知 Returns: 發送成功數量 """ subscriptions = db.query(Subscription).filter( Subscription.group_id == report.group_id, Subscription.email_notify == True ).all() sent_count = 0 for sub in subscriptions: user = db.query(User).filter(User.id == sub.user_id).first() if not user or not user.email or not user.is_active: continue # 轉義 HTML 特殊字元，防止 XSS safe_group_name = escape(report.group.name) html_content = f"""

報告延遲通知

您訂閱的「{safe_group_name}」今日報告延遲發布，敬請稍後。

造成不便，敬請見諒。

""" success = send_email( user.email, f"【每日報導】{report.group.name} 報告延遲通知", html_content ) if success: sent_count += 1 return sent_count