112 lines
4.2 KiB
Python
112 lines
4.2 KiB
Python
from __future__ import annotations
|
||
|
||
from datetime import datetime
|
||
from typing import List, Optional
|
||
|
||
import pytz
|
||
|
||
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:
|
||
tz = pytz.timezone("Asia/Shanghai")
|
||
dt = self.once_time
|
||
if dt.tzinfo is None:
|
||
dt = pytz.utc.localize(dt)
|
||
return dt.astimezone(tz).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")
|