core/timer.py - master - broot/flowee

core/timer.py
290 строк14 KB
broot
Finalize v0.2.0 features, bug fixes, and documentation
27 июл 2026, 22:29
27 июл 2026, 22:297b975f1
"""
timer.py - Ядро таймера Flowee.

Управляет состояниями: idle, running, break.
Взаимодействует с модулем database для сохранения сессий.
Не зависит от GUI - может использоваться с любым интерфейсом.
"""

import time
from datetime import datetime
from typing import Optional, Callable

from core import database


class TimerCore:
    """
    Класс, управляющий одной активной сессией.
    """

    # Тип исходной сессии ('work' или 'break'), восстановленной при старте.
    # Объявлен здесь, чтобы линтер знал о его существовании, т.к. присваивается не в __init__ ,
    # а в _recover_orphaned_session.
    orphaned_session_type: str = ""
    # orphaned_session_type: str = "" создаёт атрибут класса, а не экземпляра.
    # Но когда в _recover_orphaned_session мы делаем self.orphaned_session_type = active["type"],
    # мы создаём атрибут экземпляра, который перекрывает классовый. Поэтому дефолтное значение не мешает.

    def __init__(self, db_conn, config: dict):
        """
        db_conn - соединение с БД (sqlite3.Connection).
        config - словарь с настройками.
        """
        self._db = db_conn
        self._config = config
        self._state = "idle"            # idle | running | break
        self._active_session_id: Optional[int] = None
        self._previous_task_id: Optional[int] = None  # запоминаем, с какой задачи ушли на перерыв
        self._active_task_name = ""  # Имя активной задачи для отображения в UI

        # Callback для оповещения UI об изменении состояния (можно задать извне)
        self.on_tick: Optional[Callable[[], None]] = None
        self.on_state_change: Optional[Callable[[str], None]] = None

        # Период автосохранения в секундах
        self._save_interval = self._config.get("save_interval", 60)
        self._last_save_time = 0.0

        # При создании сразу проверяем, нет ли активной сессии после сбоя
        self._recover_orphaned_session()

    # ------------------------------------------------------------------
    # Публичные методы для управления таймером
    # ------------------------------------------------------------------

    def start(self, task_id: int):
        """Начать новую рабочую сессию для задачи с task_id."""
        if self._state != "idle":
            raise RuntimeError(f"Нельзя начать новую задачу в состоянии {self._state}")
        self._active_session_id = database.start_session(self._db, task_id)
        self._previous_task_id = task_id
        self._state = "running"
        # Сохраняем имя задачи (получаем из базы)
        cur = self._db.execute("SELECT name FROM tasks WHERE id = ?", (task_id,))
        row = cur.fetchone()
        self._active_task_name = row[0] if row else ""
        self._last_save_time = time.time()
        self._emit_state_change()

    def stop(self):
        """Остановить текущую рабочую сессию."""
        if self._state != "running":
            raise RuntimeError(f"Нельзя остановить задачу в состоянии {self._state}")
        database.stop_session(self._db, self._active_session_id)
        self._active_session_id = None
        self._state = "idle"
        self._previous_task_id = None
        self._active_task_name = ""
        self._emit_state_change()

    def start_break(self):
        """Начать перерыв (завершает активную работу и создаёт сессию перерыва)."""
        if self._state != "running":
            raise RuntimeError(f"Нельзя начать перерыв в состоянии {self._state}")
        break_id = database.start_break(self._db, self._previous_task_id)
        self._active_session_id = break_id
        self._state = "break"
        self._last_save_time = time.time()
        self._emit_state_change()

    def resume(self, task_id: int):
        """Возобновить работу после перерыва над задачей task_id."""
        if self._state != "break":
            raise RuntimeError(f"Нельзя продолжить работу в состоянии {self._state}")
        new_session_id = database.resume_work(self._db, task_id, self._previous_task_id)
        self._active_session_id = new_session_id
        self._previous_task_id = task_id
        self._state = "running"
        cur = self._db.execute("SELECT name FROM tasks WHERE id = ?", (task_id,))
        row = cur.fetchone()
        self._active_task_name = row[0] if row else ""
        self._last_save_time = time.time()
        self._emit_state_change()

    def get_task_id(self, name: str) -> int:
        """
        Возвращает ID задачи (создаёт, если нужно).
        UI-обёртка над database.get_or_create_task.
        Позволяет UI не зависеть от внутреннего устройства TimerCore (_db).
        """
        return database.get_or_create_task(self._db, name)

    def get_today_tasks(self) -> list:
        """Возвращает список задач за сегодня (словари с id, name и т.д.)."""
        return database.get_tasks_for_date(self._db)

    def has_orphaned_session(self) -> bool:
        """Возвращает True, если при запуске была обнаружена незавершённая сессия."""
        return self._state != "idle" and self._active_session_id is not None

    def tick(self):
        """
        Вызывается периодически (например, раз в секунду).
        Проверяет необходимость автосохранения и полночного перехода.
        """
        now = time.time()
        # Автосохранение раз в save_interval секунд
        if self._state in ("running", "break") and (now - self._last_save_time >= self._save_interval):
            database.periodic_save(self._db)
            self._last_save_time = now

        # Проверка перехода через полночь
        database.handle_midnight_transition(self._db)

        # Оповестить UI (если задан callback)
        if self.on_tick:
            self.on_tick()

    def resolve_orphaned(self, choice: str):
        """
        Применяет решение пользователя после обнаружения зависшей сессии.
        choice: 'continue', 'discard_and_start', 'discard_and_stop'.
        """
        active = database.get_active_session(self._db)
        if not active:
            return

        # Применяем выбор к базе данных
        if choice in ("discard_and_start", "discard_and_stop"):
            database.resolve_orphaned_session(self._db, active, "discard")
        # choice == "continue" не трогает БД

        if choice == "continue":
            now = datetime.now().isoformat(sep="T", timespec="seconds")
            conn = self._db
            # Завершаем старую сессию
            conn.execute(
                "UPDATE sessions SET end_time = ?, active = 0 WHERE id = ?",
                (now, active["id"]),
            )
            if active["type"] == "work":
                # Продолжаем работу: создаём новую рабочую сессию
                task_id = active["task_id"]
                new_id = database.start_session(conn, task_id)
                self._active_session_id = new_id
                self._previous_task_id = task_id
                self._state = "running"
                cur = conn.execute("SELECT name FROM tasks WHERE id = ?", (task_id,))
                row = cur.fetchone()
                if row:
                    self._active_task_name = row[0]
            else:  # break
                # Продолжаем перерыв: создаём новую сессию перерыва
                related_task_id = active.get("related_task_id")
                cur = conn.execute(
                    "INSERT INTO sessions (task_id, start_time, type, active, related_task_id) "
                    "VALUES (1, ?, 'break', 1, ?)",
                    (now, related_task_id),
                )
                self._active_session_id = cur.lastrowid
                self._state = "break"
                self._previous_task_id = related_task_id
            self._last_save_time = time.time()
            conn.commit()
            self._emit_state_change()
        elif choice == "discard_and_start":
            if active["type"] == "break":
                # Завершаем перерыв и сразу начинаем новую рабочую сессию
                task_id = active.get("related_task_id") or active["task_id"]
                new_session_id = database.start_session(self._db, task_id)
                self._active_session_id = new_session_id
                self._state = "running"
                self._previous_task_id = task_id
                self._last_save_time = time.time()
                self._emit_state_change()
                return
            # Если это была работа, просто переходим в idle (этот случай не должен происходить)
            self._state = "idle"
            self._active_session_id = None
            self._previous_task_id = None
            self._emit_state_change()
        elif choice == "discard_and_stop":
            self._state = "idle"
            self._active_session_id = None
            self._previous_task_id = None
            self._emit_state_change()

    def get_total_work_seconds_today(self) -> int:
        """Возвращает общее количество секунд работы за сегодня."""
        today_str = datetime.now().strftime("%Y-%m-%d")
        cur = self._db.execute(
            """SELECT COALESCE(SUM(strftime('%s', end_time) - strftime('%s', start_time)), 0)
               FROM sessions
               WHERE type = 'work' AND date(start_time) = ? AND active = 0""",
            (today_str,)
        )
        total = int(cur.fetchone()[0])

        cur = self._db.execute(
            """SELECT strftime('%s', 'now', 'localtime') - strftime('%s', start_time)
               FROM sessions
               WHERE type = 'work' AND date(start_time) = ? AND active = 1
               LIMIT 1""",
            (today_str,)
        )
        row = cur.fetchone()
        if row and row[0]:
            total += int(row[0])

        return total

    # ------------------------------------------------------------------
    # Свойства для чтения состояния
    # ------------------------------------------------------------------

    @property
    def state(self) -> str:
        return self._state

    @property
    def active_session_id(self) -> Optional[int]:
        return self._active_session_id

    @property
    def previous_task_id(self) -> Optional[int]:
        return self._previous_task_id

    @property
    def active_task_name(self) -> str:
        return self._active_task_name

    # ------------------------------------------------------------------
    # Внутренние методы
    # ------------------------------------------------------------------

    def _recover_orphaned_session(self):
        """
        Проверяет, осталась ли активная сессия с прошлого запуска.
        Если да - восстанавливает состояние таймера в соответствии с ней.
        """
        active = database.get_active_session(self._db)
        if active:
            self._active_session_id = active["id"]
            self.orphaned_session_type = active["type"]

            if active["type"] == "work":
                self._state = "running"
                self._previous_task_id = active["task_id"]
            else:  # break
                self._state = "break"
                self._previous_task_id = active.get("related_task_id")

            # Получаем имя активной задачи для отображения в UI
            task_for_name = self._previous_task_id if self._state == "running" else active.get("related_task_id")
            if task_for_name:
                cur = self._db.execute("SELECT name FROM tasks WHERE id = ?", (task_for_name,))
                row = cur.fetchone()
                self._active_task_name = row[0] if row else ""
            else:
                self._active_task_name = ""

            # Устанавливаем время последнего сохранения так, чтобы следующий tick сразу сохранил
            self._last_save_time = 0.0
            self._emit_state_change()
        # Если активной сессии нет — остаёмся в idle

    def _emit_state_change(self):
        """Вызывает оповещение UI о смене состояния."""
        if self.on_state_change:
            self.on_state_change(self._state)