import json import time import threading from pathlib import Path from datetime import datetime from typing import Optional, List from pynput import mouse, keyboard from .models import TrackerConfig, TrackedTask, WorkDetail, WorkSession from .utils import get_platform, ensure_dir class Application: def __init__(self): self._lock = threading.Lock() self.config_path = Path("config.json") self.tasks_dir = Path("tasks") self.config: TrackerConfig = self._load_or_create_config() self.current_task: Optional[TrackedTask] = None self.platform = get_platform() # Состояние трекинга self.tracking_enabled = self.config.tracking_enabled self.last_activity: float = time.time() # timestamp self.last_window_group: Optional[str] = None self.is_idle: bool = False self._tracking_thread: Optional[threading.Thread] = None self._running = True # Загружаем задачу if self.config.current_task_id: self.current_task = self._load_task(self.config.current_task_id) ensure_dir(self.tasks_dir) # Запускаем слушатели активности (только если tracking_enabled) self._start_activity_listeners() # Запускаем фоновый трекер self._start_tracking_loop() def _start_activity_listeners(self): def on_activity(*_): # Активность зафиксирована — но обработаем её только если окно рабочее self._check_and_update_activity() self.mouse_listener = mouse.Listener(on_move=on_activity, on_click=on_activity) self.keyboard_listener = keyboard.Listener(on_press=on_activity) self.mouse_listener.start() self.keyboard_listener.start() def _check_and_update_activity(self): """Обновляет last_activity, только если текущее окно — рабочее""" window_info = self.platform.get_active_window() group_name = self._match_window_to_policy(window_info) is_blocked = self._is_group_blocked(group_name) is_unknown = (group_name == "Неизвестное приложение") # Считаем "рабочим", если не blocked и не unknown if not is_blocked and not is_unknown: with self._lock: self.last_activity = time.time() self.is_idle = False def _start_tracking_loop(self): def loop(): while self._running: time.sleep(5) self._check_and_update_tracking() self._tracking_thread = threading.Thread(target=loop, daemon=True) self._tracking_thread.start() def _check_and_update_tracking(self): with self._lock: if not self.tracking_enabled or not self.current_task: return current_time = time.time() window_info = self.platform.get_active_window() group_name = self._match_window_to_policy(window_info) is_blocked = self._is_group_blocked(group_name) was_idle = self.is_idle if current_time - self.last_activity > 60 or is_blocked: self.is_idle = True self.last_window_group = None else: self.is_idle = False if self.last_window_group != group_name: self.last_window_group = group_name self._add_work_detail(group_name, 5.0) # Сохраняем при переходе в бездействие if not was_idle and self.is_idle: self.maybe_save_task() def _match_window_to_policy(self, window_info) -> str: title = window_info.title.lower() proc = window_info.process_name.lower() for policy in self.config.window_policies: for pattern in policy.window_patterns: if pattern.lower() in title: return policy.group_name for pname in policy.process_names: if pname.lower() == proc: return policy.group_name return "Неизвестное приложение" def _is_group_blocked(self, group_name: str) -> bool: for policy in self.config.window_policies: if policy.group_name == group_name: return policy.policy_type == "blocked" return False # по умолчанию — не блокируем def _add_work_detail(self, group_name: str, seconds: float): if not self.current_task: return # Создаём новую сессию или используем последнюю if not self.current_task.sessions: session = WorkSession( start_time=datetime.now(), end_time=datetime.now(), details=[], synchronized=False ) self.current_task.sessions.append(session) else: session = self.current_task.sessions[-1] # Ищем detail с такой же группой for detail in session.details: if detail.group_name == group_name: detail.duration_seconds += seconds break else: session.details.append(WorkDetail(group_name=group_name, duration_seconds=seconds)) # Обновляем время сессии session.end_time = datetime.now() # === Публичные методы с локом === def maybe_save_task(self): """Сохраняет текущую задачу, если она есть""" if self.current_task: self.save_task(self.current_task) def toggle_tracking(self): with self._lock: was_enabled = self.tracking_enabled if not self.current_task: self.tracking_enabled = False return self.tracking_enabled = not self.tracking_enabled self.config.tracking_enabled = self.tracking_enabled self._save_config(self.config) # Сохраняем при переходе в паузу if was_enabled and not self.tracking_enabled: self.maybe_save_task() def create_task(self, task_id: str, project_id: Optional[str], estimated_minutes: Optional[float]): with self._lock: ensure_dir(self.tasks_dir) new_task = TrackedTask( task_id=task_id, project_id=project_id, estimated_seconds=estimated_minutes * 60 if estimated_minutes else None, start_time=datetime.now() ) self.save_task(new_task) self.current_task = new_task self.config.current_task_id = task_id self.config.tracking_enabled = True self.tracking_enabled = True self._save_config(self.config) def select_task(self, task_id: str): with self._lock: task = self._load_task(task_id) if task: self.current_task = task self.config.current_task_id = task_id self.config.tracking_enabled = True self.tracking_enabled = True self._save_config(self.config) def add_manual_time(self, minutes: float): with self._lock: if not self.current_task: return seconds = abs(minutes) * 60 sign = 1 if minutes >= 0 else -1 self._add_work_detail("Manual", sign * seconds) def get_display_status(self) -> str: with self._lock: if not self.tracking_enabled: return "Пауза" if not self.current_task: return "Пауза" if self.is_idle: return "Бездействие" return "Активная работа" def get_current_window_info(self) -> dict: with self._lock: if not self.tracking_enabled or not self.current_task: return {"group": "Пауза", "detail": "Трекинг остановлен. Нажмите «Пауза» для запуска."} if self.is_idle: return {"group": "Бездействие", "detail": "Нет активности более 60 сек или отвлекающее окно"} window_info = self.platform.get_active_window() group = self._match_window_to_policy(window_info) return { "group": group, "detail": f"{window_info.title} | {window_info.process_name}" } # === Остальные методы (без изменений, но с локом где нужно) === def _load_or_create_config(self) -> TrackerConfig: if self.config_path.exists(): with open(self.config_path, "r", encoding="utf-8") as f: data = json.load(f) return TrackerConfig.model_validate(data) else: config = TrackerConfig() self._save_config(config) return config def _save_config(self, config: TrackerConfig) -> None: with open(self.config_path, "w", encoding="utf-8") as f: f.write(config.model_dump_json(indent=2)) def _load_task(self, task_id: str) -> Optional[TrackedTask]: task_file = self.tasks_dir / f"{task_id}.json" if task_file.exists(): try: with open(task_file, "r", encoding="utf-8") as f: data = json.load(f) return TrackedTask.model_validate(data) except Exception: return None return None def save_task(self, task: TrackedTask) -> None: task_file = self.tasks_dir / f"{task.task_id}.json" with open(task_file, "w", encoding="utf-8") as f: f.write(task.model_dump_json(indent=2)) def shutdown(self): self._running = False self.maybe_save_task() # ← сохраняем перед выходом if hasattr(self, 'mouse_listener'): self.mouse_listener.stop() if hasattr(self, 'keyboard_listener'): self.keyboard_listener.stop()