import json import logging 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") logging.basicConfig(level=logging.WARNING) logger = logging.getLogger("TimeTracker") 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 self.idle_start_time: Optional[float] = None # Загружаем задачу if self.config.current_task_id: self.current_task = self._load_task(self.config.current_task_id) try: self.sync_with_youtrack() except Exception as e: logger.error(f"Ошибка при начальной синхронизации: {e}") 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 == "Unknown app") # Считаем "рабочим", если не 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 self.idle_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 current_idle = (current_time - self.last_activity > self.config.idle_threshold_seconds) or is_blocked # === СЛУЧАЙ 1: Переход в бездействие === if not was_idle and current_idle: self.is_idle = True self.idle_start_time = current_time self.last_window_group = None self.session_start_time = None self.maybe_save_task() return # === СЛУЧАЙ 2: Выход из бездействия === if was_idle and not current_idle: self.is_idle = False idle_duration = current_time - self.idle_start_time if self.idle_start_time else 0 self.idle_start_time = None # Решаем, создавать ли НОВУЮ сессию create_new_session = False reasons = [] # Причина 1: Длительное бездействие (>10 минут) if idle_duration > self.config.new_session_after_idle_seconds: create_new_session = True reasons.append(f"бездействие {idle_duration / 60:.1f} мин") # Причина 2: Смена календарного дня if self.current_task.sessions: last_session_date = self.current_task.sessions[-1].end_time.date() current_date = datetime.now().date() if last_session_date < current_date: create_new_session = True reasons.append(f"смена дня ({last_session_date} → {current_date})") # Причина 3: Первая сессия для задачи if not self.current_task.sessions: create_new_session = True reasons.append("первая сессия задачи") # Создаём новую сессию при необходимости if create_new_session: new_session = WorkSession( start_time=datetime.now(), end_time=datetime.now(), details=[], synchronized=False ) self.current_task.sessions.append(new_session) logger.info( f"Новая сессия для {self.current_task.task_id}: " f"{', '.join(reasons)}" ) # Начинаем отслеживание self.session_start_time = current_time self.last_window_group = group_name self._add_work_detail_to_session(group_name, 5.0) return # === СЛУЧАЙ 3: Продолжение активной работы === if not current_idle: self.is_idle = False if self.last_window_group != group_name: self.last_window_group = group_name # Гарантируем наличие сессии (на случай редких состояний) if not self.current_task.sessions: self.current_task.sessions.append(WorkSession( start_time=datetime.now(), end_time=datetime.now(), details=[], synchronized=False )) self._add_work_detail_to_session(group_name, 5.0) if self.session_start_time is None: self.session_start_time = current_time return # === СЛУЧАЙ 4: Продолжение бездействия === self.is_idle = True def _add_work_detail_to_session(self, group_name: str, seconds: float): """Добавляет запись ТОЛЬКО в последнюю сессию (не создаёт новые)""" if not self.current_task or not self.current_task.sessions: logger.warning("Нет активной сессии для добавления времени") return session = self.current_task.sessions[-1] # Объединяем записи одной группы 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 _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 "Unknown app" 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() self.is_idle = True self.idle_start_time = time.time() self.session_start_time = None elif not was_enabled and self.tracking_enabled: # Возобновление → сбрасываем состояние бездействия self.is_idle = False self.idle_start_time = None self.session_start_time = time.time() 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) try: self.sync_with_youtrack() except Exception as e: logger.error(f"Ошибка при синхронизации после создания задачи: {e}") 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) try: self.sync_with_youtrack() except Exception as e: logger.error(f"Ошибка при синхронизации после выбора задачи: {e}") def add_manual_time(self, minutes: float): with self._lock: if not self.current_task: return if minutes == 0: return # Создаём ОТДЕЛЬНУЮ сессию для ручного времени new_session = WorkSession( start_time=datetime.now(), end_time=datetime.now(), details=[WorkDetail( group_name="Manual Adjustment", duration_seconds=abs(minutes) * 60 )], synchronized=False, description=f"Manual: {minutes:+.1f} min" ) self.current_task.sessions.append(new_session) logger.info(f"Добавлена ручная сессия: {minutes:+.1f} мин для {self.current_task.task_id}") # Сохраняем и синхронизируем self.save_task(self.current_task) try: self.sync_with_youtrack() except Exception as e: logger.error(f"Ошибка синхронизации после ручного добавления: {e}") 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: self._save_task_no_sync(task) 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: 1. Проверяет существование задачи в YouTrack 2. Загружает все work items из YouTrack 3. Сопоставляет локальные сессии с удаленными 4. Отправляет/обновляет несинхронизированные локальные сессии 5. Добавляет новые сессии из YouTrack, которых нет локально """ if not self.current_task: return client = YouTrackClient() if not client.enabled: logger.info("YouTrack: синхронизация отключена (нет настроек в .env)") return # Проверяем существование задачи в YouTrack if not client.issue_exists(self.current_task.task_id): logger.warning(f"YouTrack: задача {self.current_task.task_id} не существует — синхронизация пропущена") return # Загружаем все work items из YouTrack remote_items = client.get_issue_work_items(self.current_task.task_id) if not remote_items: logger.info(f"YouTrack: нет записей времени для задачи {self.current_task.task_id}") with self._lock: changed = False # Шаг 1: Обрабатываем локальные сессии for session in self.current_task.sessions: total_minutes = sum(detail.duration_seconds for detail in session.details) / 60.0 # Пропускаем нулевые или отрицательные сессии if total_minutes <= 0: session.synchronized = True changed = True continue # Случай A: Сессия уже имеет ID — обновляем существующий work item if session.id: if not session.synchronized: if client.update_work_item(self.current_task.task_id, session.id, total_minutes): session.synchronized = True changed = True logger.info(f"Обновлена сессия {session.id} для задачи {self.current_task.task_id}") else: logger.warning(f"Не удалось обновить сессию {session.id}") # Если synchronized=True — ничего не делаем # Случай B: Сессия без ID, но помечена как синхронизированная # (возможно, синхронизирована в предыдущих версиях без сохранения ID) elif session.synchronized: # Пропускаем — считаем, что время уже учтено в YouTrack # Но для надежности можно попробовать найти совпадение по длительности matching_remote = next( (item for item in remote_items if abs(item['minutes'] - total_minutes) < 1), # погрешность 1 минута None ) if matching_remote: session.id = matching_remote['id'] # Удаляем из remote_items, чтобы не дублировать при добавлении remote_items.remove(matching_remote) logger.info(f"Сопоставлена сессия без ID с удаленной записью {session.id}") changed = True # Случай C: Сессия без ID и не синхронизирована — создаем новый work item else: work_item_id = client.add_work_item( self.current_task.task_id, total_minutes, description=f"Auto-tracked: {', '.join(set(d.group_name for d in session.details))}" ) if work_item_id: session.id = work_item_id session.synchronized = True # Сохраняем время создания для будущего сопоставления session.youtrack_created_at = datetime.fromtimestamp( next((item['created'] / 1000 for item in remote_items if item['id'] == work_item_id), time.time()) ) changed = True logger.info( f"Создана новая запись времени {work_item_id} для задачи {self.current_task.task_id}") else: logger.warning(f"Не удалось создать запись времени для сессии") # Шаг 2: Добавляем удаленные work items, которых нет локально # (например, добавленные вручную через веб-интерфейс YouTrack) local_session_ids = {s.id for s in self.current_task.sessions if s.id} for remote_item in remote_items: if remote_item['id'] not in local_session_ids and remote_item['minutes'] > 0: # Создаем новую сессию на основе удаленной записи now = datetime.now() new_session = WorkSession( start_time=now, end_time=now, details=[WorkDetail( group_name="YouTrack Manual", duration_seconds=remote_item['minutes'] * 60 )], synchronized=True, id=remote_item['id'], youtrack_created_at=datetime.fromtimestamp(remote_item['created'] / 1000) if remote_item.get( 'created') else now, description=remote_item.get('text') ) self.current_task.sessions.append(new_session) changed = True logger.info( f"Добавлена удаленная запись времени {remote_item['id']} ({remote_item['minutes']} мин) в локальную задачу") # Сохраняем задачу, если были изменения if changed: self._save_task_no_sync(self.current_task) logger.info(f"Синхронизация завершена для задачи {self.current_task.task_id}") def _save_task_no_sync(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))