from __future__ import annotations from datetime import datetime from typing import List, Optional from sqlalchemy import ( Boolean, DateTime, ForeignKey, Integer, String, Text, func, ) from sqlalchemy.orm import Mapped, Session, mapped_column, relationship from bot.models.database import Base class Reminder(Base): __tablename__ = "reminders" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) user_id: Mapped[int] = mapped_column(Integer, ForeignKey("users.id"), nullable=False) title: Mapped[str] = mapped_column(String(256), nullable=False) description: Mapped[Optional[str]] = mapped_column(Text) # once / daily / weekly / monthly / interval reminder_type: Mapped[str] = mapped_column(String(32), nullable=False) once_time: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True)) daily_time: Mapped[Optional[str]] = mapped_column(String(5)) # "HH:MM" weekly_days: Mapped[Optional[str]] = mapped_column(String(32)) # "0,1,2" (Mon=0) monthly_day: Mapped[Optional[int]] = mapped_column(Integer) interval_minutes: Mapped[Optional[int]] = mapped_column(Integer) interval_start_time: Mapped[Optional[str]] = mapped_column(String(5)) # "HH:MM" interval_end_time: Mapped[Optional[str]] = mapped_column(String(5)) # "HH:MM" skip_holidays: Mapped[bool] = mapped_column(Boolean, default=False) is_active: Mapped[bool] = mapped_column(Boolean, default=True) last_sent_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True)) next_run_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True)) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now() ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now(), onupdate=func.now() ) logs: Mapped[List["ReminderLog"]] = relationship( "ReminderLog", back_populates="reminder", cascade="all, delete-orphan" ) @classmethod def get_active(cls, session: Session) -> List["Reminder"]: return session.query(cls).filter_by(is_active=True).all() @classmethod def get_user_reminders(cls, session: Session, user_id: int) -> List["Reminder"]: return session.query(cls).filter_by(user_id=user_id).order_by(cls.created_at).all() def type_display(self) -> str: mapping = { "once": "一次性", "daily": "每日", "weekly": "每周", "monthly": "每月", "interval": "间隔", } return mapping.get(self.reminder_type, self.reminder_type) def schedule_summary(self) -> str: if self.reminder_type == "once" and self.once_time: return self.once_time.strftime("%Y-%m-%d %H:%M") if self.reminder_type == "daily" and self.daily_time: return f"每天 {self.daily_time}" if self.reminder_type == "weekly" and self.weekly_days and self.daily_time: day_names = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"] days = ", ".join(day_names[int(d)] for d in self.weekly_days.split(",")) return f"{days} {self.daily_time}" if self.reminder_type == "interval" and self.interval_minutes: return ( f"每 {self.interval_minutes} 分钟" f"({self.interval_start_time}–{self.interval_end_time})" ) return "—" class ReminderLog(Base): __tablename__ = "reminder_logs" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) reminder_id: Mapped[int] = mapped_column( Integer, ForeignKey("reminders.id"), nullable=False ) sent_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now() ) # sent / snoozed / completed / failed status: Mapped[str] = mapped_column(String(32), default="sent") reminder: Mapped["Reminder"] = relationship("Reminder", back_populates="logs")