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 from dotenv import load_dotenv import os from .youtrack_client import YouTrackClient if not load_dotenv(".env"): load_dotenv(".env.example") 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 self.session_start_time: Optional[float] = None # Загружаем задачу 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: self.session_start_time = None 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 self.session_start_time = 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 self.session_start_time is None: self.session_start_time = current_time # Сохраняем при переходе в бездействие 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: try: self.sync_with_youtrack() except: pass 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 get_current_session_duration(self) -> float: with self._lock: if self.session_start_time is None: return 0.0 return time.time() - self.session_start_time 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() def sync_with_youtrack(self): """Синхронизирует все несинхронизированные сессии текущей задачи с YouTrack""" if not self.current_task: return client = YouTrackClient() if not client.enabled: return # Проверяем, существует ли задача if not client.issue_exists(self.current_task.task_id): return changed = False for session in self.current_task.sessions: if not session.synchronized: total_minutes = sum(detail.duration_seconds for detail in session.details) / 60.0 if client.add_work_item(self.current_task.task_id, total_minutes): session.synchronized = True changed = True else: # Если ошибка — не помечаем как синхронизированную (повторим позже) pass if changed: self.save_task(self.current_task)