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(*_): with self._lock: self.last_activity = time.time() self.is_idle = False 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 _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 # Проверка: спит ли система? (упрощённо — если активность > 60 сек назад) current_time = time.time() if current_time - self.last_activity > 60: self.is_idle = True self.last_window_group = None return # Получаем текущее окно window_info = self.platform.get_active_window() if not window_info.title and not window_info.process_name: self.is_idle = True return # Определяем группу по политике group_name = self._match_window_to_policy(window_info) is_blocked = self._is_group_blocked(group_name) if is_blocked: self.is_idle = True self.last_window_group = None return # Если рабочая группа — накапливаем время self.is_idle = False if self.last_window_group != group_name: # Новая группа — завершаем предыдущую сессию (если была) self.last_window_group = group_name # Добавляем 5 секунд к текущей группе self._add_work_detail(group_name, 5.0) 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 toggle_tracking(self): with self._lock: 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) 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 if hasattr(self, 'mouse_listener'): self.mouse_listener.stop() if hasattr(self, 'keyboard_listener'): self.keyboard_listener.stop()