288 lines
9.6 KiB
Python
288 lines
9.6 KiB
Python
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.db.models import PomodoroSession
|
|
from app.pomodoro.cycle import (
|
|
PHASE_LONG_BREAK,
|
|
PHASE_SHORT_BREAK,
|
|
PHASE_WORK,
|
|
CycleManager,
|
|
)
|
|
|
|
|
|
def _utcnow() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
class PomodoroService:
|
|
def __init__(self, db: Session):
|
|
self.db = db
|
|
self.cycle = CycleManager(db)
|
|
|
|
def _get_active(self) -> PomodoroSession | None:
|
|
stmt = (
|
|
select(PomodoroSession)
|
|
.where(PomodoroSession.status.in_(("running", "paused")))
|
|
.order_by(PomodoroSession.id.desc())
|
|
.limit(1)
|
|
)
|
|
return self.db.scalar(stmt)
|
|
|
|
def _elapsed(self, session: PomodoroSession) -> int:
|
|
elapsed = session.elapsed_seconds
|
|
if session.status == "running" and session.started_at:
|
|
started = session.started_at
|
|
if started.tzinfo is None:
|
|
started = started.replace(tzinfo=timezone.utc)
|
|
delta = _utcnow() - started
|
|
elapsed += int(delta.total_seconds())
|
|
return elapsed
|
|
|
|
def _remaining(self, session: PomodoroSession) -> int:
|
|
total = session.duration_min * 60
|
|
return max(0, total - self._elapsed(session))
|
|
|
|
def _try_auto_complete(self, session: PomodoroSession) -> bool:
|
|
if session.status != "running":
|
|
return False
|
|
if self._remaining(session) > 0:
|
|
return False
|
|
self._finalize_session(session, auto=True)
|
|
return True
|
|
|
|
def _finalize_session(
|
|
self,
|
|
session: PomodoroSession,
|
|
*,
|
|
auto: bool,
|
|
result: str = "",
|
|
completed: bool | None = None,
|
|
cancelled: bool = False,
|
|
) -> None:
|
|
session.elapsed_seconds = self._elapsed(session)
|
|
session.started_at = None
|
|
session.finished_at = _utcnow()
|
|
session.completion_notified = False
|
|
session.result = result or None
|
|
|
|
if cancelled:
|
|
session.status = "cancelled"
|
|
session.completed = False
|
|
elif completed is not None:
|
|
session.status = "completed"
|
|
session.completed = completed
|
|
else:
|
|
session.status = "completed"
|
|
session.completed = True
|
|
|
|
self.db.commit()
|
|
self.db.refresh(session)
|
|
|
|
def _start_phase(
|
|
self,
|
|
phase: str,
|
|
*,
|
|
duration_min: int | None = None,
|
|
task_note: str | None = None,
|
|
) -> PomodoroSession:
|
|
active = self._get_active()
|
|
if active:
|
|
raise ValueError("Таймер уже запущен. Сначала остановите текущую сессию.")
|
|
|
|
cycle = self.cycle.get()
|
|
if task_note is not None:
|
|
cycle.task_note = task_note
|
|
elif phase == PHASE_WORK and not cycle.task_note:
|
|
cycle.task_note = ""
|
|
|
|
duration = duration_min or self.cycle.duration_for_phase(phase, cycle)
|
|
note = task_note if task_note is not None else cycle.task_note
|
|
|
|
session = PomodoroSession(
|
|
status="running",
|
|
phase=phase,
|
|
duration_min=duration,
|
|
task_note=note,
|
|
started_at=_utcnow(),
|
|
)
|
|
self.db.add(session)
|
|
self.db.commit()
|
|
self.db.refresh(session)
|
|
return session
|
|
|
|
def _to_status_dict(self, session: PomodoroSession | None) -> dict:
|
|
cycle_dict = self.cycle.to_dict()
|
|
if not session:
|
|
return {
|
|
"status": "idle",
|
|
"phase": PHASE_WORK,
|
|
"duration_min": cycle_dict["work_duration_min"],
|
|
"task_note": cycle_dict["task_note"],
|
|
"elapsed_seconds": 0,
|
|
"remaining_seconds": 0,
|
|
"session_id": None,
|
|
"cycle": cycle_dict,
|
|
}
|
|
|
|
elapsed = self._elapsed(session)
|
|
total = session.duration_min * 60
|
|
remaining = max(0, total - elapsed)
|
|
|
|
return {
|
|
"status": session.status,
|
|
"phase": session.phase,
|
|
"duration_min": session.duration_min,
|
|
"task_note": session.task_note,
|
|
"elapsed_seconds": elapsed,
|
|
"remaining_seconds": remaining,
|
|
"session_id": session.id,
|
|
"started_at": session.started_at.isoformat() if session.started_at else None,
|
|
"finished_at": session.finished_at.isoformat() if session.finished_at else None,
|
|
"cycle": cycle_dict,
|
|
}
|
|
|
|
def get_status(self) -> dict:
|
|
active = self._get_active()
|
|
if active:
|
|
self._try_auto_complete(active)
|
|
active = self._get_active()
|
|
return self._to_status_dict(active)
|
|
|
|
def start_work(self, duration_min: int | None = None, task_note: str = "") -> dict:
|
|
session = self._start_phase(
|
|
PHASE_WORK,
|
|
duration_min=duration_min,
|
|
task_note=task_note,
|
|
)
|
|
return self._to_status_dict(session)
|
|
|
|
def start_short_break(self, duration_min: int | None = None) -> dict:
|
|
session = self._start_phase(PHASE_SHORT_BREAK, duration_min=duration_min)
|
|
return self._to_status_dict(session)
|
|
|
|
def start_long_break(self, duration_min: int | None = None) -> dict:
|
|
session = self._start_phase(PHASE_LONG_BREAK, duration_min=duration_min)
|
|
return self._to_status_dict(session)
|
|
|
|
def start(self, duration_min: int = 25, task_note: str = "") -> dict:
|
|
return self.start_work(duration_min=duration_min, task_note=task_note)
|
|
|
|
def pause(self) -> dict:
|
|
session = self._get_active()
|
|
if not session or session.status != "running":
|
|
raise ValueError("Нет активного запущенного таймера.")
|
|
|
|
session.elapsed_seconds = self._elapsed(session)
|
|
session.status = "paused"
|
|
session.paused_at = _utcnow()
|
|
session.started_at = None
|
|
self.db.commit()
|
|
self.db.refresh(session)
|
|
return self._to_status_dict(session)
|
|
|
|
def resume(self) -> dict:
|
|
session = self._get_active()
|
|
if not session or session.status != "paused":
|
|
raise ValueError("Нет таймера на паузе.")
|
|
|
|
session.status = "running"
|
|
session.started_at = _utcnow()
|
|
session.paused_at = None
|
|
self.db.commit()
|
|
self.db.refresh(session)
|
|
return self._to_status_dict(session)
|
|
|
|
def stop(self, result: str = "", completed: bool = False) -> dict:
|
|
session = self._get_active()
|
|
if not session:
|
|
raise ValueError("Нет активного таймера.")
|
|
|
|
if completed:
|
|
self._finalize_session(session, auto=False, result=result, completed=True)
|
|
else:
|
|
self._finalize_session(session, auto=False, result=result, cancelled=True)
|
|
session.completion_notified = True
|
|
self.db.commit()
|
|
return self._to_status_dict(None)
|
|
|
|
def reset_cycle(self, clear_task: bool = False) -> dict:
|
|
active = self._get_active()
|
|
if active:
|
|
self._finalize_session(active, auto=False, cancelled=True)
|
|
active.completion_notified = True
|
|
self.db.commit()
|
|
cycle = self.cycle.reset(clear_task=clear_task)
|
|
status = self._to_status_dict(None)
|
|
status["cycle"] = cycle
|
|
return status
|
|
|
|
def skip_phase(self) -> dict:
|
|
session = self._get_active()
|
|
if not session:
|
|
raise ValueError("Нет активного таймера.")
|
|
|
|
self._finalize_session(session, auto=True)
|
|
return self._to_status_dict(None)
|
|
|
|
def get_pending_completions(self) -> list[PomodoroSession]:
|
|
stmt = (
|
|
select(PomodoroSession)
|
|
.where(
|
|
PomodoroSession.status == "completed",
|
|
PomodoroSession.completed.is_(True),
|
|
PomodoroSession.completion_notified.is_(False),
|
|
)
|
|
.order_by(PomodoroSession.id.asc())
|
|
)
|
|
return list(self.db.scalars(stmt))
|
|
|
|
def mark_notified(self, session: PomodoroSession) -> None:
|
|
session.completion_notified = True
|
|
self.db.commit()
|
|
|
|
def advance_after_completion(self, session: PomodoroSession) -> dict | None:
|
|
"""Update cycle counters and auto-start next phase. Returns new status or None."""
|
|
phase = session.phase
|
|
cycle = self.cycle.get()
|
|
|
|
if phase == PHASE_WORK:
|
|
next_phase = self.cycle.on_work_completed()
|
|
elif phase == PHASE_SHORT_BREAK:
|
|
next_phase = PHASE_WORK
|
|
elif phase == PHASE_LONG_BREAK:
|
|
self.cycle.on_long_break_completed()
|
|
next_phase = None
|
|
else:
|
|
next_phase = None
|
|
|
|
if not cycle.auto_advance or next_phase is None:
|
|
return None
|
|
|
|
new_session = self._start_phase(next_phase)
|
|
return self._to_status_dict(new_session)
|
|
|
|
def history(self, limit: int = 20) -> list[dict]:
|
|
stmt = (
|
|
select(PomodoroSession)
|
|
.where(PomodoroSession.status.in_(("completed", "cancelled")))
|
|
.order_by(PomodoroSession.finished_at.desc())
|
|
.limit(limit)
|
|
)
|
|
sessions = self.db.scalars(stmt).all()
|
|
return [
|
|
{
|
|
"id": s.id,
|
|
"status": s.status,
|
|
"phase": s.phase,
|
|
"duration_min": s.duration_min,
|
|
"task_note": s.task_note,
|
|
"result": s.result,
|
|
"completed": s.completed,
|
|
"elapsed_seconds": s.elapsed_seconds,
|
|
"finished_at": s.finished_at.isoformat() if s.finished_at else None,
|
|
}
|
|
for s in sessions
|
|
]
|