這 25 天，教練一直都是被動的，只有用戶來問才回答。好的教練不會只等著被問。真正的教練會主動提醒你該做功課了、任務快到期了，或是問你這一週的學習狀況。今天要讓系統在固定時間自動發送通知，不必等用戶來問。

實作方式是用 APScheduler 在後端定時執行三種任務：每天早上生成今日清單、任務快到期時提醒、每週日做週回顧。這些通知不會寄到信箱或手機，只存在系統內。用戶登入時會看到「1 條新通知」，點開就能看到歷史紀錄。

概念：定時工作就像鬧鐘

你的手機裡有很多鬧鐘：早上 6 點起床、下午 3 點喝水、晚上 10 點睡覺。APScheduler 就是在軟體裡裝一個鬧鐘。你告訴它「每天早上 6 點執行 X」，它就會一直等，時間一到就執行。

APScheduler 監視時間
        ↓ 早上 6:00
   執行「生成今日任務」
        ↓ 存進資料庫
   通知表新增一筆記錄
        ↓ 用戶看到
   「您有 1 條新通知」

三種定時任務各有用途：

這些通知要持久化，也就是存進資料庫。用戶關掉網頁後，通知仍然會保留。

實作步驟

步驟 1：新增 Notification 模型

檔案位置: backend/models.py
狀態: 修改檔案（接在檔案最後面）
用途: 定義通知的資料結構

class Notification(Base):
    """應用內通知紀錄"""
    __tablename__ = "notifications"

    id = Column(Integer, primary_key=True, index=True)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
    title = Column(String, nullable=False)  # 通知標題
    message = Column(String, nullable=False)  # 通知內容
    notification_type = Column(String, default="info")  # info / warning / success
    is_read = Column(Boolean, default=False)  # 是否已讀
    created_at = Column(DateTime, default=datetime.datetime.utcnow, nullable=False)
    
    # 反向參照
    user = relationship("User", back_populates="notifications")

同時，在 User 模型裡加上反向參照：

# 在 User 類別裡新增
notifications = relationship("Notification", back_populates="user", cascade="all, delete-orphan")

步驟 2：定義通知的回傳格式

檔案位置: backend/schemas.py
狀態: 修改檔案（接在檔案最後面）
用途: 定義 API 回傳通知時的格式

class NotificationResponse(BaseModel):
    """通知物件"""
    id: int
    title: str
    message: str
    notification_type: str  # info, warning, success
    is_read: bool
    created_at: str  # ISO 格式時間

    class Config:
        from_attributes = True

步驟 3：建立 Notification 表

執行資料庫初始化：

# 在 backend/database.py 或 main.py 的初始化代碼裡執行
Base.metadata.create_all(bind=engine)

這會自動建立 notifications 表。

步驟 4：寫一個通知發送函式

檔案位置: backend/services/notification_service.py
狀態: 新建檔案
用途: 集中管理所有通知的建立邏輯

from sqlalchemy.orm import Session
from datetime import datetime
from models import Notification, User

class NotificationService:
    """管理應用內通知"""

    @staticmethod
    def create_notification(
        session: Session,
        user_id: int,
        title: str,
        message: str,
        notification_type: str = "info"
    ) -> Notification:
        """建立一則通知"""
        notification = Notification(
            user_id=user_id,
            title=title,
            message=message,
            notification_type=notification_type
        )
        session.add(notification)
        session.commit()
        session.refresh(notification)
        return notification

    @staticmethod
    def get_unread_notifications(session: Session, user_id: int):
        """取得該使用者的未讀通知"""
        return session.query(Notification).filter(
            Notification.user_id == user_id,
            Notification.is_read == False
        ).order_by(Notification.created_at.desc()).all()

    @staticmethod
    def mark_as_read(session: Session, notification_id: int):
        """標記通知已讀"""
        notification = session.query(Notification).get(notification_id)
        if notification:
            notification.is_read = True
            session.commit()
        return notification

步驟 5：新增通知查詢 API

檔案位置: backend/main.py
狀態: 修改檔案（接在現有端點後面）
用途: 讓前端能查詢和標記通知

先在檔案最上面加 import：

from apscheduler.schedulers.background import BackgroundScheduler
from services.notification_service import NotificationService

再加上兩個端點：

@app.get("/notifications")
async def get_notifications(user_id: int, session: Session = Depends(get_session)):
    """取得使用者的未讀通知"""
    notifications = NotificationService.get_unread_notifications(session, user_id)
    return [
        NotificationResponse(
            id=n.id,
            title=n.title,
            message=n.message,
            notification_type=n.notification_type,
            is_read=n.is_read,
            created_at=n.created_at.isoformat()
        )
        for n in notifications
    ]


@app.patch("/notifications/{notification_id}/read")
async def mark_notification_read(notification_id: int, session: Session = Depends(get_session)):
    """標記通知已讀"""
    notification = NotificationService.mark_as_read(session, notification_id)
    if not notification:
        raise HTTPException(status_code=404, detail="通知不存在")
    return {"status": "ok", "notification_id": notification_id}

步驟 6：建立三個定時任務

檔案位置: backend/scheduled_jobs.py
狀態: 新建檔案
用途: 定義所有定時執行的任務邏輯

from sqlalchemy.orm import Session
from datetime import datetime, timedelta
from models import Task, User, ProgressLog
from services.notification_service import NotificationService
from database import SessionLocal

def daily_morning_reminder():
    """早上 6:00 - 生成今日任務清單"""
    session = SessionLocal()
    try:
        today = datetime.utcnow().date()
        
        # 找出今天的所有任務
        tasks_today = session.query(Task).filter(
            Task.deadline >= today,
            Task.deadline < today + timedelta(days=1),
            Task.status != "completed"
        ).all()
        
        # 針對每個使用者建立通知
        for task in tasks_today:
            plan = task.plan
            user_id = plan.user_id
            
            user_tasks = [t for t in tasks_today if t.plan.user_id == user_id]
            message = f"今天有 {len(user_tasks)} 項任務要完成：\n"
            message += "\n".join([f"- {t.title}" for t in user_tasks[:5]])
            if len(user_tasks) > 5:
                message += f"\n... 還有 {len(user_tasks) - 5} 項"
            
            NotificationService.create_notification(
                session,
                user_id,
                title="📅 今日任務清單",
                message=message,
                notification_type="info"
            )
    finally:
        session.close()


def deadline_reminder():
    """上午 10:00 - 逾期提醒"""
    session = SessionLocal()
    try:
        tomorrow = datetime.utcnow().date() + timedelta(days=1)
        
        # 找出明天到期、還沒完成的任務
        overdue_tasks = session.query(Task).filter(
            Task.deadline == tomorrow,
            Task.status != "completed"
        ).all()
        
        for task in overdue_tasks:
            plan = task.plan
            message = f"『{task.title}』明天就要到期了，趕快完成吧！"
            
            NotificationService.create_notification(
                session,
                plan.user_id,
                title="⏰ 任務即將逾期",
                message=message,
                notification_type="warning"
            )
    finally:
        session.close()


def weekly_review():
    """週日晚上 8:00 - 週回顧"""
    session = SessionLocal()
    try:
        today = datetime.utcnow().date()
        week_ago = today - timedelta(days=7)
        
        # 找出所有使用者
        users = session.query(User).all()
        
        for user in users:
            # 計算這週完成的任務
            completed_this_week = session.query(ProgressLog).join(Task).filter(
                ProgressLog.completed_at >= week_ago,
                ProgressLog.completed_at <= today,
                Task.plan.has(plan_id=Task.plan_id),
                Task.plan.has(user_id=user.id)
            ).count()
            
            # 計算這週的總學習時數
            total_hours = session.query(ProgressLog).join(Task).filter(
                ProgressLog.completed_at >= week_ago,
                ProgressLog.completed_at <= today,
                Task.plan.has(plan_id=Task.plan_id),
                Task.plan.has(user_id=user.id)
            ).with_entities(ProgressLog.actual_hours).all()
            
            total_hours = sum([h[0] for h in total_hours if h[0]])
            
            message = f"本週完成了 {completed_this_week} 項任務，累計 {total_hours:.1f} 小時。繼續加油！"
            
            NotificationService.create_notification(
                session,
                user.id,
                title="📊 週回顧",
                message=message,
                notification_type="success"
            )
    finally:
        session.close()

步驟 7：初始化 APScheduler

檔案位置: backend/main.py
狀態: 修改檔案（在 FastAPI 初始化後面加上）
用途: 啟動定時任務排程

from scheduled_jobs import daily_morning_reminder, deadline_reminder, weekly_review

# 建立排程器
scheduler = BackgroundScheduler()

# 註冊三個定時任務
scheduler.add_job(
    daily_morning_reminder,
    'cron',
    hour=6,  # 早上 6:00
    minute=0,
    id='daily_morning_reminder'
)

scheduler.add_job(
    deadline_reminder,
    'cron',
    hour=10,  # 上午 10:00
    minute=0,
    id='deadline_reminder'
)

scheduler.add_job(
    weekly_review,
    'cron',
    day_of_week='sun',  # 週日
    hour=20,  # 晚上 8:00
    minute=0,
    id='weekly_review'
)

# 啟動排程器
scheduler.start()

# 應用關閉時停止排程器
@app.on_event("shutdown")
def shutdown_scheduler():
    scheduler.shutdown()

步驟 8：安裝 APScheduler

在終端執行：

pip install apscheduler

測試定時任務

測試 1：手動執行任務檢查邏輯

# 建立一個測試腳本：backend/test_notifications.py

from database import SessionLocal
from scheduled_jobs import daily_morning_reminder, deadline_reminder, weekly_review
from models import User, Notification

session = SessionLocal()

# 執行早上提醒
print("測試每日提醒...")
daily_morning_reminder()

# 查詢通知表，看有沒有新增記錄
notifications = session.query(Notification).order_by(Notification.created_at.desc()).limit(5).all()
print(f"最新 5 則通知：")
for n in notifications:
    print(f"  - [{n.created_at}] {n.title}: {n.message}")

session.close()

執行：

python backend/test_notifications.py

測試 2：修改排程時間，驗證是否執行

為了測試，可以把排程改成 1 分鐘後執行（別在正式環境這樣做）：

scheduler.add_job(
    daily_morning_reminder,
    'interval',
    minutes=1,  # 改成每分鐘執行一次
    id='daily_morning_reminder'
)

等待 1 分鐘，然後檢查通知表。

測試 3：用 API 查詢通知

# 用 curl 查詢
curl "http://localhost:8000/notifications?user_id=1"

# 或者在 Python 裡
import requests
response = requests.get("http://localhost:8000/notifications?user_id=1")
print(response.json())

常見問題

排程時間與伺服器時區不一致

APScheduler 預設使用伺服器的本地時間。假設伺服器跑在 UTC，程式碼寫的是「早上 6:00」，實際執行的會是 UTC 的 6:00。

解決：確認伺服器時區，或在 APScheduler 設定時明確指定：

from pytz import timezone

scheduler.add_job(
    daily_morning_reminder,
    'cron',
    hour=6,
    minute=0,
    timezone='Asia/Taipei',  # 用台灣時間
    id='daily_morning_reminder'
)

資料庫連線洩漏

每個定時任務都會建立 session。如果忘了在 finally 裡呼叫 close()，session 會不斷累積，最後資料庫連線會被用光。

解決：一定要用 try-finally：

session = SessionLocal()
try:
    # 執行邏輯
    ...
finally:
    session.close()  # 一定要關

任務拋出例外後排程停止

預設情況下，如果任務拋出例外，APScheduler 會重試幾次後放棄。

解決：在任務函式裡包上 try-except，讓例外被記錄下來，同時不影響排程：

def daily_morning_reminder():
    try:
        # 邏輯
        ...
    except Exception as e:
        print(f"Error in daily_morning_reminder: {e}")
        # 或者記錄到日誌

明天的要加入認證與授權，保護使用者資料。系統安全是基礎，通知也一樣，必須確保只有使用者本人看得到。