import calendar from datetime import datetime, timedelta, timezone from typing import Any from zoneinfo import ZoneInfo from sqlalchemy import select from sqlalchemy.orm import Session from app.db.models import Reminder from app.homelab.context import resolve_timezone from app.reminders.notify import bump_notify_seq, get_notify_seq RECURRENCE_NONE = "none" RECURRENCE_DAILY = "daily" RECURRENCE_WEEKLY = "weekly" RECURRENCE_MONTHLY = "monthly" VALID_RECURRENCE = frozenset({RECURRENCE_NONE, RECURRENCE_DAILY, RECURRENCE_WEEKLY, RECURRENCE_MONTHLY}) def _utcnow() -> datetime: return datetime.now(timezone.utc) def _parse_due_at(raw: str, tz_name: str) -> datetime: clean = raw.strip() if not clean: raise ValueError("due_at не может быть пустым") try: dt = datetime.fromisoformat(clean.replace("Z", "+00:00")) except ValueError as exc: raise ValueError(f"Неверный формат даты: {raw}") from exc if dt.tzinfo is None: try: dt = dt.replace(tzinfo=ZoneInfo(tz_name)) except Exception: dt = dt.replace(tzinfo=ZoneInfo("Europe/Moscow")) return dt.astimezone(timezone.utc) def _advance_due(due_at: datetime, recurrence: str) -> datetime: if recurrence == RECURRENCE_DAILY: return due_at + timedelta(days=1) if recurrence == RECURRENCE_WEEKLY: return due_at + timedelta(weeks=1) if recurrence == RECURRENCE_MONTHLY: month = due_at.month + 1 year = due_at.year if month > 12: month = 1 year += 1 day = min(due_at.day, calendar.monthrange(year, month)[1]) return due_at.replace(year=year, month=month, day=day) return due_at def _format_local(dt: datetime, tz_name: str, *, all_day: bool = False) -> str: try: local = dt.astimezone(ZoneInfo(tz_name)) except Exception: local = dt.astimezone(ZoneInfo("Europe/Moscow")) if all_day: return local.strftime("%Y-%m-%d") return local.strftime("%Y-%m-%d %H:%M") class RemindersService: def __init__(self, db: Session): self.db = db def _tz(self) -> str: return resolve_timezone(self.db) def _to_dict(self, row: Reminder) -> dict[str, Any]: tz = row.timezone or self._tz() return { "id": row.id, "title": row.title, "notes": row.notes, "due_at": row.due_at.isoformat(), "due_at_local": _format_local(row.due_at, tz, all_day=row.all_day), "all_day": row.all_day, "recurrence": row.recurrence, "enabled": row.enabled, "completed_at": row.completed_at.isoformat() if row.completed_at else None, "timezone": tz, "created_at": row.created_at.isoformat() if row.created_at else None, } def snapshot(self) -> dict[str, Any]: upcoming = self.list_upcoming(limit=12) return { "notify_seq": get_notify_seq(self.db), "upcoming": upcoming, "upcoming_count": len(upcoming), "timezone": self._tz(), } def list_upcoming(self, *, limit: int = 30) -> list[dict[str, Any]]: stmt = ( select(Reminder) .where( Reminder.enabled.is_(True), Reminder.completed_at.is_(None), ) .order_by(Reminder.due_at.asc()) .limit(limit) ) return [self._to_dict(row) for row in self.db.scalars(stmt).all()] def list_in_range( self, *, date_from: datetime, date_to: datetime, ) -> list[dict[str, Any]]: stmt = ( select(Reminder) .where( Reminder.enabled.is_(True), Reminder.completed_at.is_(None), Reminder.due_at >= date_from, Reminder.due_at < date_to, ) .order_by(Reminder.due_at.asc()) ) return [self._to_dict(row) for row in self.db.scalars(stmt).all()] def get(self, reminder_id: int) -> dict[str, Any] | None: row = self.db.get(Reminder, reminder_id) return self._to_dict(row) if row else None def create( self, *, title: str, due_at: str, notes: str = "", all_day: bool = False, recurrence: str = RECURRENCE_NONE, ) -> dict[str, Any]: clean_title = title.strip() if not clean_title: raise ValueError("Название напоминания не может быть пустым") rec = (recurrence or RECURRENCE_NONE).strip().lower() if rec not in VALID_RECURRENCE: raise ValueError(f"recurrence должен быть один из: {', '.join(sorted(VALID_RECURRENCE))}") tz = self._tz() due = _parse_due_at(due_at, tz) row = Reminder( title=clean_title, notes=notes.strip(), due_at=due, all_day=all_day, recurrence=rec, timezone=tz, ) self.db.add(row) self.db.commit() self.db.refresh(row) bump_notify_seq(self.db) return {"ok": True, "reminder": self._to_dict(row), "created": True} def update( self, reminder_id: int, *, title: str | None = None, due_at: str | None = None, notes: str | None = None, all_day: bool | None = None, recurrence: str | None = None, enabled: bool | None = None, ) -> dict[str, Any]: row = self.db.get(Reminder, reminder_id) if not row: raise ValueError("Напоминание не найдено") if title is not None: clean = title.strip() if not clean: raise ValueError("Название не может быть пустым") row.title = clean if notes is not None: row.notes = notes.strip() if due_at is not None: row.due_at = _parse_due_at(due_at, row.timezone or self._tz()) row.last_fired_at = None if all_day is not None: row.all_day = all_day if recurrence is not None: rec = recurrence.strip().lower() if rec not in VALID_RECURRENCE: raise ValueError(f"recurrence должен быть один из: {', '.join(sorted(VALID_RECURRENCE))}") row.recurrence = rec if enabled is not None: row.enabled = enabled row.updated_at = _utcnow() self.db.commit() self.db.refresh(row) bump_notify_seq(self.db) return {"ok": True, "reminder": self._to_dict(row)} def delete(self, reminder_id: int) -> dict[str, Any]: row = self.db.get(Reminder, reminder_id) if not row: raise ValueError("Напоминание не найдено") title = row.title self.db.delete(row) self.db.commit() bump_notify_seq(self.db) return {"ok": True, "deleted_id": reminder_id, "title": title} def complete(self, reminder_id: int) -> dict[str, Any]: row = self.db.get(Reminder, reminder_id) if not row: raise ValueError("Напоминание не найдено") now = _utcnow() row.completed_at = now row.enabled = False row.updated_at = now self.db.commit() self.db.refresh(row) bump_notify_seq(self.db) return {"ok": True, "reminder": self._to_dict(row)}