Compare commits

...

12 Commits

Author SHA1 Message Date
Mikan
4b56ec4641 Better usage 2026-03-10 05:42:27 +03:00
Mikan
4c260b1bad [AI] Youtrack integration 2026-03-10 05:35:01 +03:00
Mikan
97dbcfd8f0 [AI] Tracking 2026-03-10 02:47:12 +03:00
Mikan
bf51fc454d [AI] Tracking 2026-03-10 02:29:37 +03:00
Mikan
07ed356cf1 [AI] MVP 1 2026-03-10 02:20:51 +03:00
Mikan
6946b8fb50 initial 2 2026-03-10 01:40:10 +03:00
Mikan
346ea77fd2 before refactor 2026-03-10 00:44:46 +03:00
Mikan
320f20b974 [AI] window 2026-03-09 22:33:31 +03:00
Mikan
daf1c2a37b [AI] window is not creating 2026-03-09 22:26:51 +03:00
Mikan
01cf0f87d0 [AI] fix 2026-03-09 22:05:48 +03:00
Mikan
c635b0261b [AI] fix 2026-03-09 21:56:44 +03:00
Mikan
89231b9005 [AI] fix 2026-03-09 21:49:38 +03:00
31 changed files with 1109 additions and 1426 deletions

3
.env.example Normal file
View File

@@ -0,0 +1,3 @@
YOUTRACK_URL=https://your-company.myjetbrains.com/youtrack
YOUTRACK_TOKEN=perm:abc123...
# YOUTRACK_CA_BUNDLE=certs/your-ca.pem

91
.gitignore vendored Normal file
View File

@@ -0,0 +1,91 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
# PyInstaller
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# Pyre type checker
.pyre/
# IDEs
.vscode/
.idea/
*.swp
*.swo
# OS generated files
.DS_Store
.DS_Store?
._*
.Spotlight-V100
.Trashes
ehthumbs.db
Thumbs.db
Icon?
# Configuration and data files generated by the application
config.json
tasks/
*.json
.env

View File

@@ -15,7 +15,7 @@ A personal time tracking tool that monitors your work sessions, tracks time spen
## Installation
```bash
pip install -e .
pip install .
```
## Usage
@@ -23,14 +23,15 @@ pip install -e .
Run the application:
```bash
python -m work_tracker
python -m citrus_time_tracker
```
or just
```bash
citrus_time_tracker
```
Use hotkeys:
- Ctrl+Alt+T: Toggle tracking
- Ctrl+Alt+N: New task
- Ctrl+Alt+S: End current session
- Ctrl+Alt+M: Adjust time

View File

@@ -3,27 +3,33 @@ requires = ["setuptools>=61.0"]
build-backend = "setuptools.build_meta"
[project]
name = "work_tracker"
version = "0.1.0"
description = "Personal time tracking tool with YouTrack integration"
name = "citrus_time_tracker"
version = "1.0.0"
description = "Time tracking tool with YouTrack integration"
readme = "README.md"
authors = [{name = "Mikan"}]
license = {text = "MIT"}
requires-python = ">=3.8"
dependencies = [
"pynput>=1.7.6",
"pywin32>=306; sys_platform == 'win32'"
"pywin32>=306; sys_platform == 'win32'",
"psutil>=5.9.0; sys_platform == 'win32'",
"pydantic>=2.0.0",
"python-dotenv>=1.0.0", # <-- новое
"httpx>=0.25.0", # для асинхронных/синхронных HTTP-запросов
]
[project.optional-dependencies]
dev = [
"pytest>=7.0",
"pytest-mock>=3.10.0",
]
[project.scripts]
citrus_time_tracker = "citrus_time_tracker.__main__:main"
[tool.setuptools.packages.find]
where = ["src"]
include = ["time_tracker*"]
include = ["citrus_time_tracker*"]
[tool.pytest.ini_options]
testpaths = ["tests"]

View File

View File

@@ -0,0 +1,18 @@
import sys
from .app import Application
def main():
try:
app = Application()
from .ui.overlay import OverlayWindow
overlay = OverlayWindow(app)
overlay.run()
except RuntimeError as e:
print(f"Ошибка: {e}", file=sys.stderr)
input("Нажмите Enter для выхода...")
sys.exit(1)
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,313 @@
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)

View File

@@ -0,0 +1,74 @@
from datetime import datetime
from typing import List, Optional
from pydantic import BaseModel, Field
from typing_extensions import Literal
class WorkDetail(BaseModel):
group_name: str
duration_seconds: float
class WorkSession(BaseModel):
start_time: datetime
end_time: datetime
details: List[WorkDetail]
synchronized: bool = False
class TrackedTask(BaseModel):
task_id: str
project_id: Optional[str] = None
estimated_seconds: Optional[float] = None
start_time: datetime
end_time: Optional[datetime] = None
sessions: List[WorkSession] = Field(default_factory=list)
@property
def total_worked_seconds(self) -> float:
return sum(
detail.duration_seconds
for session in self.sessions
for detail in session.details
)
@property
def remaining_seconds(self) -> Optional[float]:
if self.estimated_seconds is None:
return None
return max(0.0, self.estimated_seconds - self.total_worked_seconds)
class WindowPolicy(BaseModel):
group_name: str
window_patterns: List[str] = Field(default_factory=list)
process_names: List[str] = Field(default_factory=list)
policy_type: Literal["work", "blocked"]
class TrackerConfig(BaseModel):
tracking_enabled: bool = False
current_task_id: Optional[str] = None
window_policies: List[WindowPolicy] = Field(default_factory=list)
@staticmethod
def default_policies() -> List[WindowPolicy]:
return [
WindowPolicy(
group_name="IDE",
window_patterns=["PyCharm", "IntelliJ", "Visual Studio Code", "Code", "Sublime", "Vim", "Neovim"],
process_names=["pycharm64.exe", "idea64.exe", "code.exe", "sublime_text.exe", "nvim.exe", "vim.exe"],
policy_type="work"
),
WindowPolicy(
group_name="Gaming & Distractions",
window_patterns=["Steam", "Discord", "YouTube", "Twitch", "Spotify", "Netflix"],
process_names=["steam.exe", "discord.exe", "spotify.exe"],
policy_type="blocked"
),
]
def __init__(self, **data):
super().__init__(**data)
if not self.window_policies:
self.window_policies = self.default_policies()

View File

@@ -0,0 +1,16 @@
from abc import ABC, abstractmethod
from typing import NamedTuple
class WindowInfo(NamedTuple):
title: str
process_name: str
class PlatformBase(ABC):
@abstractmethod
def get_active_window(self) -> WindowInfo:
"""
Получает информацию об активном окне: заголовок и имя процесса.
"""
pass

View File

@@ -0,0 +1,34 @@
import win32gui
import win32process
import psutil
from .base import PlatformBase, WindowInfo
class WindowsPlatform(PlatformBase):
def get_active_window(self) -> WindowInfo:
try:
hwnd = win32gui.GetForegroundWindow()
if not hwnd:
return WindowInfo(title="", process_name="")
# Получаем заголовок окна
title = win32gui.GetWindowText(hwnd).strip()
# Получаем PID процесса
_, pid = win32process.GetWindowThreadProcessId(hwnd)
if pid == 0:
return WindowInfo(title=title, process_name="")
# Получаем имя исполняемого файла процесса
try:
process = psutil.Process(pid)
process_name = process.name()
except (psutil.NoSuchProcess, psutil.AccessDenied):
process_name = ""
return WindowInfo(title=title, process_name=process_name)
except Exception:
# В случае ошибки (например, недоступно окно) — возвращаем пустые данные
return WindowInfo(title="", process_name="")

View File

@@ -0,0 +1 @@
from .overlay import OverlayWindow

View File

@@ -0,0 +1,67 @@
import tkinter as tk
from datetime import datetime
from tkinter import Toplevel, Label, Entry, Button
from typing import Callable, TYPE_CHECKING
if TYPE_CHECKING:
from ..app import Application
class AdjustTimeWindow:
def __init__(self, parent, app: "Application", on_close_callback: Callable):
self.parent = parent
self.app = app
self.on_close_callback = on_close_callback
self.window = Toplevel(parent)
self.window.title("Изменить время")
self.window.geometry("250x120")
self.window.transient(parent)
self.window.grab_set()
Label(self.window, text="Изменить время (минуты):").pack(pady=(10, 5))
Label(self.window, text="Отрицательное — убрать время").pack()
self.minutes_entry = Entry(self.window, justify="center")
self.minutes_entry.pack(pady=5)
self.minutes_entry.insert(0, "0")
Button(self.window, text="Применить", command=self.on_apply).pack(pady=5)
def on_apply(self):
if not self.app.current_task:
self.window.destroy()
return
try:
minutes = float(self.minutes_entry.get())
except ValueError:
return
if minutes == 0:
self.window.destroy()
return
seconds = abs(minutes * 60)
# Создаём новую сессию или добавляем в последнюю?
# Для простоты — добавим новую сессию с одним WorkDetail
from ..models import WorkDetail, WorkSession
detail = WorkDetail(
group_name="Manual",
duration_seconds=seconds if minutes > 0 else -seconds # можно хранить отрицательное
)
session = WorkSession(
start_time=datetime.now(),
end_time=datetime.now(),
details=[detail],
synchronized=False
)
self.app.current_task.sessions.append(session)
self.app.save_task(self.app.current_task)
self.on_close_callback()
self.window.destroy()

View File

@@ -0,0 +1,69 @@
import tkinter as tk
from datetime import datetime
from tkinter import Toplevel, Label, Entry, Button
from typing import Callable, TYPE_CHECKING
if TYPE_CHECKING:
from ..app import Application
class NewTaskWindow:
def __init__(self, parent, app: "Application", on_close_callback: Callable):
self.parent = parent
self.app = app
self.on_close_callback = on_close_callback
self.window = Toplevel(parent)
self.window.title("Новая задача")
self.window.geometry("300x200")
self.window.transient(parent)
self.window.grab_set()
Label(self.window, text="ID задачи:").pack(pady=(10, 0))
self.task_id_entry = Entry(self.window)
self.task_id_entry.pack(pady=5)
Label(self.window, text="Оценка времени (мин):").pack()
self.estimate_entry = Entry(self.window)
self.estimate_entry.pack(pady=5)
Label(self.window, text="ID проекта (опц.):").pack()
self.project_id_entry = Entry(self.window)
self.project_id_entry.pack(pady=5)
Button(self.window, text="Создать", command=self.on_create).pack(pady=10)
def on_create(self):
task_id = self.task_id_entry.get().strip()
if not task_id:
return
try:
estimate_str = self.estimate_entry.get().strip()
estimate_seconds = float(estimate_str) * 60 if estimate_str else None
except ValueError:
return
project_id = self.project_id_entry.get().strip() or None
# Создаём новую задачу
from ..models import TrackedTask
new_task = TrackedTask(
task_id=task_id,
project_id=project_id,
estimated_seconds=estimate_seconds,
start_time=datetime.now()
)
# Сохраняем
self.app.save_task(new_task)
# Обновляем конфиг
self.app.config.current_task_id = task_id
self.app._save_config(self.app.config)
# Обновляем текущую задачу в памяти
self.app.current_task = new_task
self.on_close_callback()
self.window.destroy()

View File

@@ -0,0 +1,188 @@
import tkinter as tk
from tkinter import Menu, Label, Button, Frame
from typing import TYPE_CHECKING
from ..utils import format_seconds
if TYPE_CHECKING:
from ..app import Application
class OverlayWindow:
def __init__(self, app: "Application"):
self.app = app
self.session_timer_label = None
self.root = tk.Tk()
self.root.protocol("WM_DELETE_WINDOW", self.on_close)
self.root.title("Citrus Time Tracker")
self.root.geometry("380x190")
self.root.overrideredirect(True) # убираем заголовок
self.root.attributes("-topmost", True) # всегда поверх
self.root.attributes("-alpha", 0.5) # полупрозрачность по умолчанию
self.root.bind("<Enter>", self.on_mouse_enter)
self.root.bind("<Leave>", self.on_mouse_leave)
# Для перетаскивания
self._drag_data = {"x": 0, "y": 0}
self.root.bind("<Button-1>", self.start_drag)
self.root.bind("<B1-Motion>", self.drag)
# Контекстное меню
self.context_menu = Menu(self.root, tearoff=0)
self.context_menu.add_command(label="Закрыть", command=self.root.destroy)
self.root.bind("<Button-3>", self.show_context_menu)
# Центрирование
self.center_window()
# Содержимое
self._build_ui()
self._auto_update()
def on_close(self):
self.app.shutdown()
self.root.destroy()
def center_window(self):
self.root.update_idletasks()
screen_width = self.root.winfo_screenwidth()
screen_height = self.root.winfo_screenheight()
x = (screen_width // 2) - (380 // 2)
y = (screen_height // 2) - (190 // 2)
self.root.geometry(f"380x190+{x}+{y}")
def _build_ui(self):
# Главный фрейм с отступами
main_frame = Frame(self.root, padx=10, pady=10)
main_frame.pack(fill="both", expand=True)
# Заголовок
self.status_label = Label(main_frame, text="Citrus Time Tracker: Бездействие", anchor="w")
self.status_label.pack(anchor="w")
# Задача
self.task_label = Label(main_frame, text="Задача: <Не выбрано>", anchor="w")
self.task_label.pack(anchor="w")
# Время
self.time_info_label = Label(main_frame, text="Времени нет", anchor="w", font=("TkDefaultFont", 9))
self.time_info_label.pack(anchor="w")
self.session_timer_label = Label(main_frame, text="0:00", font=("Arial", 16, "bold"), fg="green")
self.session_timer_label.pack(pady=(5, 0))
# Текущее окно
self.window_group_label = Label(main_frame, text="Неизвестное приложение", anchor="w", fg="gray")
self.window_group_label.pack(anchor="w")
self.window_detail_label = Label(main_frame, text="", anchor="w", font=("TkDefaultFont", 8), fg="gray")
self.window_detail_label.pack(anchor="w")
# Кнопки
btn_frame = Frame(main_frame)
btn_frame.pack(side="bottom", fill="x", pady=(10, 0))
self.pause_button = Button(btn_frame, text="⏸ Пауза", command=self.on_pause)
self.pause_button.pack(side="left", fill="x", expand=True, padx=(0, 2))
self.new_task_btn = Button(btn_frame, text=" Новая", command=self.on_new_task)
self.new_task_btn.pack(side="left", fill="x", expand=True, padx=2)
self.select_task_btn = Button(btn_frame, text="🔍 Выбрать", command=self.on_select_task)
self.select_task_btn.pack(side="left", fill="x", expand=True, padx=2)
self.adjust_time_btn = Button(btn_frame, text="✏️ Время", command=self.on_adjust_time)
self.adjust_time_btn.pack(side="left", fill="x", expand=True, padx=(2, 0))
Button(btn_frame, text="📊 Отчёт", command=self.on_show_report).pack(side="left", fill="x", expand=True, padx=2)
self._update_display()
def _update_display(self):
status = self.app.get_display_status()
self.status_label.config(text=f"Citrus Time Tracker: {status}")
task_id = self.app.config.current_task_id or "<Не выбрано>"
self.task_label.config(text=f"Задача: {task_id}")
# Время
if self.app.current_task:
worked_str = format_seconds(self.app.current_task.total_worked_seconds)
est = self.app.current_task.estimated_seconds
if est is not None:
est_str = format_seconds(est)
time_info = f"Затрачено: {worked_str} | Оценка: {est_str}"
else:
time_info = f"Затрачено: {worked_str} | Оценка: —"
else:
time_info = "Задача не выбрана"
self.time_info_label.config(text=time_info)
# Окно
win_info = self.app.get_current_window_info()
self.window_group_label.config(text=win_info["group"])
self.window_detail_label.config(text=win_info["detail"])
# Активность кнопок
has_task = self.app.current_task is not None
self.pause_button.config(state="normal" if has_task else "disabled")
for btn in [self.new_task_btn, self.select_task_btn, self.adjust_time_btn]:
btn.config(state="normal")
# Но "Добавить время" — только если есть задача
self.adjust_time_btn.config(state="normal" if has_task else "disabled")
def _auto_update(self):
self._update_display()
# Обновляем таймер сессии каждую секунду
session_sec = self.app.get_current_session_duration()
session_str = format_seconds(session_sec)
self.session_timer_label.config(text=session_str)
self.root.after(500, self._auto_update) # теперь каждую секунду
# === Обработчики событий (пока без логики, только открытие окон) ===
def on_pause(self):
self.app.toggle_tracking()
def on_new_task(self):
from .new_task_window import NewTaskWindow
NewTaskWindow(self.root, self.app, self._update_display)
def on_select_task(self):
from .select_task_window import SelectTaskWindow
SelectTaskWindow(self.root, self.app, self._update_display)
def on_adjust_time(self):
from .adjust_time_window import AdjustTimeWindow
AdjustTimeWindow(self.root, self.app, self._update_display)
def on_show_report(self):
from .report_window import ReportWindow
ReportWindow(self.root, self.app)
# === UI-управление ===
def on_mouse_enter(self, event):
self.root.attributes("-alpha", 1.0)
def on_mouse_leave(self, event):
self.root.attributes("-alpha", 0.5)
def start_drag(self, event):
self._drag_data["x"] = event.x
self._drag_data["y"] = event.y
def drag(self, event):
x = self.root.winfo_x() + event.x - self._drag_data["x"]
y = self.root.winfo_y() + event.y - self._drag_data["y"]
self.root.geometry(f"+{x}+{y}")
def show_context_menu(self, event):
self.context_menu.tk_popup(event.x_root, event.y_root)
def run(self):
self.root.mainloop()

View File

@@ -0,0 +1,53 @@
import tkinter as tk
from tkinter import Toplevel, Text, Scrollbar, Button
from ..utils import format_seconds
class ReportWindow:
def __init__(self, parent, app: "Application"):
self.window = Toplevel(parent)
self.window.title("Отчёт по задаче")
self.window.geometry("500x400")
self.window.transient(parent)
self.window.grab_set()
if not app.current_task:
tk.Label(self.window, text="Нет активной задачи", font=("Arial", 12)).pack(pady=20)
Button(self.window, text="Закрыть", command=self.window.destroy).pack()
return
task = app.current_task
text_widget = Text(self.window, wrap="word", padx=10, pady=10)
scrollbar = Scrollbar(self.window, command=text_widget.yview)
text_widget.config(yscrollcommand=scrollbar.set)
# Собираем отчёт
report = f"Задача: {task.task_id}\n"
if task.project_id:
report += f"Проект: {task.project_id}\n"
report += f"Всего затрачено: {format_seconds(task.total_worked_seconds)}\n\n"
# Группировка по группам
group_totals = {}
for session in task.sessions:
for detail in session.details:
group = detail.group_name
group_totals[group] = group_totals.get(group, 0) + detail.duration_seconds
report += "Распределение по группам:\n"
for group, total in sorted(group_totals.items(), key=lambda x: -x[1]):
report += f"{group}: {format_seconds(total)}\n"
report += "\nПодробные сессии:\n"
for i, session in enumerate(task.sessions, 1):
report += f"\nСессия {i} ({session.start_time.strftime('%H:%M')}{session.end_time.strftime('%H:%M')}):\n"
for detail in session.details:
report += f" - {detail.group_name}: {format_seconds(detail.duration_seconds)}\n"
text_widget.insert("1.0", report)
text_widget.config(state="disabled")
text_widget.pack(side="left", fill="both", expand=True)
scrollbar.pack(side="right", fill="y")
Button(self.window, text="Закрыть", command=self.window.destroy).pack(pady=10)

View File

@@ -0,0 +1,55 @@
import tkinter as tk
from pathlib import Path
from tkinter import Toplevel, Listbox, Scrollbar, Button
from typing import Callable, TYPE_CHECKING
if TYPE_CHECKING:
from ..app import Application
class SelectTaskWindow:
def __init__(self, parent, app: "Application", on_close_callback: Callable):
self.parent = parent
self.app = app
self.on_close_callback = on_close_callback
self.window = Toplevel(parent)
self.window.title("Выбрать задачу")
self.window.geometry("300x250")
self.window.transient(parent)
self.window.grab_set()
# Сканируем папку tasks/
task_files = list(Path("tasks").glob("*.json")) if Path("tasks").exists() else []
self.task_ids = [f.stem for f in task_files]
list_frame = tk.Frame(self.window)
list_frame.pack(fill="both", expand=True, padx=10, pady=10)
scrollbar = Scrollbar(list_frame)
scrollbar.pack(side="right", fill="y")
self.listbox = Listbox(list_frame, yscrollcommand=scrollbar.set)
for tid in self.task_ids:
self.listbox.insert("end", tid)
self.listbox.pack(side="left", fill="both", expand=True)
scrollbar.config(command=self.listbox.yview)
Button(self.window, text="Выбрать", command=self.on_select).pack(pady=5)
def on_select(self):
selection = self.listbox.curselection()
if not selection:
return
task_id = self.task_ids[selection[0]]
# Загружаем задачу
task = self.app._load_task(task_id)
if task:
self.app.current_task = task
self.app.config.current_task_id = task_id
self.app._save_config(self.app.config)
self.on_close_callback()
self.window.destroy()

View File

@@ -0,0 +1,26 @@
import os
import sys
from pathlib import Path
from .platform.base import PlatformBase
from .platform.windows import WindowsPlatform
def get_platform() -> PlatformBase:
if sys.platform == "win32":
return WindowsPlatform()
else:
raise RuntimeError(f"Unsupported OS: {sys.platform}. Only Windows is supported for now.")
def ensure_dir(path: Path) -> None:
path.mkdir(parents=True, exist_ok=True)
def format_seconds(seconds: float) -> str:
seconds = int(abs(seconds))
h = seconds // 3600
m = (seconds % 3600) // 60
s = seconds % 60
if h > 0:
return f"{h}:{m:02d}:{s:02d}"
else:
return f"{m}:{s:02d}"

View File

@@ -0,0 +1,81 @@
import httpx
import os
import logging
from pathlib import Path
logging.basicConfig(level=logging.WARNING)
logger = logging.getLogger("YouTrackClient")
class YouTrackClient:
def __init__(self):
self.base_url = os.getenv("YOUTRACK_URL")
self.token = os.getenv("YOUTRACK_TOKEN")
self.ca_bundle = os.getenv("YOUTRACK_CA_BUNDLE")
if not self.base_url or not self.token:
self.enabled = False
logger.info("YouTrack: .env не настроен — синхронизация отключена")
return
# Определяем verify параметр для httpx
if self.ca_bundle:
ca_path = Path(self.ca_bundle)
if ca_path.exists():
self.verify = str(ca_path.resolve())
logger.info(f"YouTrack: использую CA bundle: {self.verify}")
else:
logger.error(f"YouTrack: CA bundle не найден: {ca_path}")
self.verify = True # fallback на системные
else:
self.verify = True # стандартное поведение
self.enabled = True
self.headers = {
"Authorization": f"Bearer {self.token}",
"Content-Type": "application/json",
"Accept": "application/json"
}
self.base_url = self.base_url.rstrip('/')
def _get_client(self) -> httpx.Client:
"""Создаёт HTTP-клиент с правильной настройкой SSL"""
return httpx.Client(verify=self.verify, timeout=15)
def issue_exists(self, issue_id: str) -> bool:
if not self.enabled:
return False
try:
with self._get_client() as client:
url = f"{self.base_url}/api/issues/{issue_id}"
resp = client.get(url, headers=self.headers)
exists = resp.status_code == 200
if not exists:
logger.warning(f"YouTrack: задача {issue_id} не найдена (HTTP {resp.status_code})")
return exists
except Exception as e:
logger.error(f"YouTrack: ошибка при проверке задачи {issue_id}: {e}")
return False
def add_work_item(self, issue_id: str, minutes: float) -> bool:
if not self.enabled:
return False
if minutes <= 0:
return True
payload = {"duration": {"minutes": round(minutes)}}
try:
with self._get_client() as client:
url = f"{self.base_url}/api/issues/{issue_id}/timeTracking/workItems"
resp = client.post(url, headers=self.headers, json=payload)
if resp.status_code in (200, 201):
logger.info(f"YouTrack: +{round(minutes)} мин к {issue_id}")
return True
else:
logger.error(f"YouTrack: ошибка отправки времени ({issue_id}): {resp.status_code} {resp.text}")
return False
except Exception as e:
logger.error(f"YouTrack: сетевая ошибка при отправке времени ({issue_id}): {e}")
return False

View File

@@ -1 +0,0 @@
"""Time Tracker package."""

View File

@@ -1,33 +0,0 @@
import json
import os
from typing import Dict, Any
CONFIG_FILE = "config.json"
def load_config() -> Dict[str, Any]:
"""
Load configuration from config.json.
Creates default config if file doesn't exist.
"""
if os.path.exists(CONFIG_FILE):
with open(CONFIG_FILE, "r", encoding="utf-8") as f:
return json.load(f)
else:
# Default configuration
default_config = {
"tracking_enabled": True,
"current_task_id": None,
"window_rules": {
"work": ["PyCharm", "VS Code", "Visual Studio", "Sublime Text", "Atom", "IntelliJ IDEA"],
"distraction": ["YouTube", "Twitter", "Facebook", "Instagram", "TikTok", "Discord", "Slack"]
}
}
save_config(default_config)
return default_config
def save_config(config: Dict[str, Any]) -> None:
"""
Save configuration to config.json.
"""
with open(CONFIG_FILE, "w", encoding="utf-8") as f:
json.dump(config, f, indent=2, ensure_ascii=False)

View File

@@ -1,221 +0,0 @@
import json
import os
from datetime import datetime
from typing import Dict, List, Optional
from pathlib import Path
class SessionManager:
def __init__(self, tasks_directory: str):
self.tasks_directory = Path(tasks_directory)
self.tasks_directory.mkdir(exist_ok=True)
self.current_session: Optional[Dict] = None
self.session_dirty = False
def create_session(self, task_id: str, project_id: str) -> Dict:
"""
Create a new work session.
"""
session_id = datetime.now().strftime("%Y%m%dT%H%M%S")
session = {
"session_id": session_id,
"task_id": task_id,
"project_id": project_id,
"started_at": datetime.now().isoformat(),
"duration_seconds": 0,
"window_details": {},
"distractions": [],
"manual_adjustments": [],
"synced_to_youtrack": False
}
self.current_session = session
self.session_dirty = True
return session
def record_window_time(self, window_title: str, seconds: int) -> None:
"""
Record time spent on a specific window/application.
"""
if not self.current_session:
return
current_time = self.current_session["window_details"].get(window_title, 0)
self.current_session["window_details"][window_title] = current_time + seconds
self.session_dirty = True
def record_distraction(self, window_title: str, duration_seconds: int) -> None:
"""
Record a distraction period.
"""
if not self.current_session:
return
distraction = {
"window": window_title,
"start": datetime.now().isoformat(),
"duration_seconds": duration_seconds
}
self.current_session["distractions"].append(distraction)
self.session_dirty = True
def add_manual_adjustment(self, seconds: int, reason: str) -> None:
"""
Add a manual time adjustment (positive to add, negative to subtract).
"""
if not self.current_session:
return
operation = "add" if seconds >= 0 else "subtract"
adjustment = {
"operation": operation,
"seconds": abs(seconds),
"reason": reason,
"applied_at": datetime.now().isoformat()
}
self.current_session["manual_adjustments"].append(adjustment)
self.session_dirty = True
def increment_duration(self, seconds: int) -> None:
"""
Increment the total duration of the current session.
"""
if not self.current_session:
return
self.current_session["duration_seconds"] += seconds
self.session_dirty = True
def end_current_session(self) -> Optional[Dict]:
"""
End the current session and save it to disk.
"""
if not self.current_session:
return None
# Finalize session
self.current_session["ended_at"] = datetime.now().isoformat()
# Calculate final duration with adjustments
total_duration = self.current_session["duration_seconds"]
for adj in self.current_session.get("manual_adjustments", []):
if adj["operation"] == "add":
total_duration += adj["seconds"]
else:
total_duration -= adj["seconds"]
self.current_session["final_duration_seconds"] = max(0, total_duration)
# Save session to disk
self._save_session(self.current_session)
# Store reference to return
ended_session = self.current_session
# Clear current session
self.current_session = None
self.session_dirty = False
return ended_session
def end_session(self, session: Dict) -> None:
"""
End a specific session and save it to disk.
"""
if "ended_at" not in session:
session["ended_at"] = datetime.now().isoformat()
# Calculate final duration with adjustments
total_duration = session["duration_seconds"]
for adj in session.get("manual_adjustments", []):
if adj["operation"] == "add":
total_duration += adj["seconds"]
else:
total_duration -= adj["seconds"]
session["final_duration_seconds"] = max(0, total_duration)
self._save_session(session)
def _save_session(self, session: Dict) -> None:
"""
Save a session to its respective task directory.
"""
task_dir = self.tasks_directory / session["task_id"]
task_dir.mkdir(exist_ok=True)
sessions_dir = task_dir / "sessions"
sessions_dir.mkdir(exist_ok=True)
filename = f"{session['session_id']}_{session['task_id']}.json"
filepath = sessions_dir / filename
with open(filepath, 'w', encoding='utf-8') as f:
json.dump(session, f, indent=2, ensure_ascii=False)
def get_current_session_duration(self) -> int:
"""
Get the current session's total duration including adjustments.
"""
if not self.current_session:
return 0
total_duration = self.current_session["duration_seconds"]
for adj in self.current_session.get("manual_adjustments", []):
if adj["operation"] == "add":
total_duration += adj["seconds"]
else:
total_duration -= adj["seconds"]
return max(0, total_duration)
def get_total_work_today(self) -> int:
"""
Get total work time for today across all tasks.
"""
today = datetime.now().date().isoformat()
total = 0
for task_dir in self.tasks_directory.iterdir():
if not task_dir.is_dir():
continue
sessions_dir = task_dir / "sessions"
if not sessions_dir.exists():
continue
for session_file in sessions_dir.glob("*.json"):
with open(session_file, 'r', encoding='utf-8') as f:
session = json.load(f)
# Check if session started today
start_date = datetime.fromisoformat(session['started_at']).date().isoformat()
if start_date == today:
total += session.get('final_duration_seconds', session.get('duration_seconds', 0))
return total
def load_sessions_for_task(self, task_id: str) -> List[Dict]:
"""
Load all sessions for a specific task.
"""
task_dir = self.tasks_directory / task_id
sessions_dir = task_dir / "sessions"
if not sessions_dir.exists():
return []
sessions = []
for session_file in sessions_dir.glob("*.json"):
with open(session_file, 'r', encoding='utf-8') as f:
session = json.load(f)
sessions.append(session)
# Sort by ended_at (or started_at if ended_at doesn't exist), newest first
sessions.sort(
key=lambda s: s.get('ended_at', s.get('started_at', '')),
reverse=True
)
return sessions

View File

@@ -1,319 +0,0 @@
import tkinter as tk
from tkinter import simpledialog, messagebox
import threading
import time
from datetime import datetime
import os
from typing import Dict, List, Optional
try:
from .core.config import load_config, save_config
from .core.session_manager import SessionManager
from .ui.overlay import OverlayWindow
# Try to import platform-specific modules
import sys
if sys.platform.startswith('win'):
from .platform.windows import get_active_window, register_session_listener, get_idle_time
else:
# Fallback for other platforms
def get_active_window():
return "Unsupported platform"
def register_session_listener(callback):
print("Session monitoring not supported on this platform")
def get_idle_time():
return 0
except ImportError:
# For running as a single file
from core.config import load_config, save_config
from core.session_manager import SessionManager
from ui.overlay import OverlayWindow
import sys
if sys.platform.startswith('win'):
from platform.windows import get_active_window, register_session_listener, get_idle_time
else:
def get_active_window():
return "Unsupported platform"
def register_session_listener(callback):
print("Session monitoring not supported on this platform")
def get_idle_time():
return 0
class WorkTracker:
def __init__(self):
self.config = load_config()
self.session_manager = SessionManager("tasks")
self.active_window = ""
self.window_classification = "neutral"
self.last_activity = time.time()
self.session_locked = False
self.is_tracking = self.config.get("tracking_enabled", True)
self.current_task_id = self.config.get("current_task_id")
# Start monitoring in background
self.start_monitoring()
def start_monitoring(self):
"""Start background monitoring threads"""
# Thread for window tracking
window_thread = threading.Thread(target=self.monitor_windows, daemon=True)
window_thread.start()
# Thread for input activity tracking
input_thread = threading.Thread(target=self.monitor_input_activity, daemon=True)
input_thread.start()
# Register session listener
register_session_listener(self.on_session_change)
def monitor_windows(self):
"""Monitor active windows and update classifications"""
while True:
try:
current_window = get_active_window()
# Only update if window changed
if current_window != self.active_window:
self.active_window = current_window
self.window_classification = self.classify_window(current_window)
# Show notification if it's a distraction or neutral app
if self.window_classification == "distraction":
self.show_distraction_notification(current_window)
elif self.window_classification == "neutral":
self.show_neutral_notification(current_window)
time.sleep(1)
except Exception as e:
print(f"Error in window monitoring: {e}")
time.sleep(5)
def monitor_input_activity(self):
"""Monitor keyboard and mouse activity"""
try:
from pynput import mouse, keyboard
def on_activity(*args):
self.last_activity = time.time()
# Start listeners
mouse.Listener(on_move=on_activity, on_click=on_activity).start()
keyboard.Listener(on_press=on_activity).start()
except ImportError:
print("pynput not available, using fallback idle detection")
# Fallback: just use system idle time
pass
def classify_window(self, title: str) -> str:
"""Classify window as work, distraction, or neutral"""
title_lower = title.lower()
rules = self.config.get("window_rules", {})
# Check distractions first
for keyword in rules.get("distraction", []):
if keyword.lower() in title_lower:
return "distraction"
# Then check work apps
for keyword in rules.get("work", []):
if keyword.lower() in title_lower:
return "work"
# Otherwise neutral
return "neutral"
def on_session_change(self, event_type: str):
"""Handle session lock/unlock events"""
if event_type == "locked":
self.session_locked = True
# End current session if active
if self.session_manager.current_session:
self.session_manager.end_current_session()
elif event_type == "unlocked":
self.session_locked = False
self.last_activity = time.time() # Reset idle time on unlock
def should_count_as_work(self) -> bool:
"""Check if current conditions should count as work time"""
# Check if tracking is enabled
if not self.is_tracking:
return False
# Check if session is locked
if self.session_locked:
return False
# Check idle time (more than 60 seconds of inactivity)
idle_time = time.time() - self.last_activity
if idle_time > 60:
return False
# Check window classification
if self.window_classification != "work":
return False
return True
def run_main_loop(self):
"""Main tracking loop"""
while True:
try:
if self.should_count_as_work():
# Increment current session duration
if self.session_manager.current_session:
self.session_manager.increment_duration(1)
# Record time for current window
self.session_manager.record_window_time(self.active_window, 1)
time.sleep(1)
except Exception as e:
print(f"Error in main loop: {e}")
time.sleep(5)
def toggle_tracking(self):
"""Toggle tracking on/off"""
self.is_tracking = not self.is_tracking
self.config["tracking_enabled"] = self.is_tracking
save_config(self.config)
def create_task(self, task_id: str):
"""Create a new task and start tracking it"""
project_id = task_id.split('-')[0] if '-' in task_id else task_id
self.session_manager.create_session(task_id, project_id)
self.current_task_id = task_id
self.config["current_task_id"] = task_id
save_config(self.config)
def switch_task(self, task_id: str):
"""Switch to an existing task"""
# End current session if exists
if self.session_manager.current_session:
self.session_manager.end_current_session()
# Start new session for the task
project_id = task_id.split('-')[0] if '-' in task_id else task_id
self.session_manager.create_session(task_id, project_id)
self.current_task_id = task_id
self.config["current_task_id"] = task_id
save_config(self.config)
def get_available_tasks(self) -> List[str]:
"""Get list of all available tasks"""
tasks_dir = "tasks"
if not os.path.exists(tasks_dir):
return []
tasks = []
for item in os.listdir(tasks_dir):
item_path = os.path.join(tasks_dir, item)
if os.path.isdir(item_path):
tasks.append(item)
return tasks
def has_active_session(self) -> bool:
"""Check if there's an active session"""
return self.session_manager.current_session is not None
def add_manual_adjustment(self, seconds: int, reason: str):
"""Add a manual time adjustment"""
if self.session_manager.current_session:
self.session_manager.add_manual_adjustment(seconds, reason)
def get_current_session_duration(self) -> int:
"""Get duration of current session"""
return self.session_manager.get_current_session_duration()
def get_total_work_today(self) -> int:
"""Get total work time for today"""
return self.session_manager.get_total_work_today()
def get_planned_duration(self, task_id: str) -> Optional[int]:
"""Get planned duration for a task (placeholder implementation)"""
# In a full implementation, this would look up the planned time for the task
# For now, return None to indicate no planned time
return None
def show_distraction_notification(self, window_title: str):
"""Show notification about distraction window"""
# In a full implementation, this would show a GUI notification
print(f"DISTRACTION: {window_title}")
def show_neutral_notification(self, window_title: str):
"""Show notification about neutral window"""
# In a full implementation, this would show a GUI notification
print(f"NEUTRAL APP: {window_title}")
def main():
# Create main Tkinter root
root = tk.Tk()
root.withdraw() # Hide the root window
# Initialize tracker
tracker = WorkTracker()
# Create overlay window
overlay = OverlayWindow(root, tracker, tracker.config)
# Start main tracking loop in background
main_loop_thread = threading.Thread(target=tracker.run_main_loop, daemon=True)
main_loop_thread.start()
# Set up hotkeys
try:
from pynput import keyboard
def on_hotkey(key):
try:
if key == keyboard.Key.alt_l: # Left Alt pressed
return # Wait for combination
elif hasattr(key, 'char') and key.char == 't':
# Ctrl+Alt+T: Toggle tracking
tracker.toggle_tracking()
overlay.update_display()
elif hasattr(key, 'char') and key.char == 'n':
# Ctrl+Alt+N: New task
overlay.new_task()
elif hasattr(key, 'char') and key.char == 's':
# Ctrl+Alt+S: End session
if tracker.session_manager.current_session:
tracker.session_manager.end_current_session()
overlay.update_display()
elif hasattr(key, 'char') and key.char == 'm':
# Ctrl+Alt+M: Adjust time
overlay.adjust_time()
except AttributeError:
# Special keys like Ctrl, Alt don't have char attribute
pass
# Start keyboard listener
keyboard.Listener(on_press=on_hotkey).start()
except ImportError:
print("pynput not available, hotkeys disabled")
# Run the GUI
try:
root.mainloop()
except KeyboardInterrupt:
print("\nShutting down...")
# End any current session before exiting
if tracker.session_manager.current_session:
tracker.session_manager.end_current_session()
if __name__ == "__main__":
main()

View File

@@ -1,19 +0,0 @@
from abc import ABC, abstractmethod
from typing import Callable
class PlatformInterface(ABC):
@abstractmethod
def get_active_window(self) -> str:
"""Get the currently active window title"""
pass
@abstractmethod
def register_session_listener(self, callback: Callable[[str], None]) -> None:
"""Register a callback for session lock/unlock events"""
pass
@abstractmethod
def get_idle_time(self) -> int:
"""Get the system idle time in seconds"""
pass

View File

@@ -1,75 +0,0 @@
import win32gui
import win32con
import win32api
import threading
from typing import Callable
def get_active_window() -> str:
"""Get the currently active window title on Windows"""
try:
hwnd = win32gui.GetForegroundWindow()
return win32gui.GetWindowText(hwnd) or "Unknown Window"
except:
return "Error getting window title"
def register_session_listener(callback: Callable[[str], None]) -> None:
"""Register a callback for Windows session lock/unlock events"""
def session_event_handler(hwnd, msg, wparam, lparam):
if msg == win32con.WM_WTSSESSION_CHANGE:
if wparam == win32con.WTS_SESSION_LOCK:
callback("locked")
elif wparam == win32con.WTS_SESSION_UNLOCK:
callback("unlocked")
return win32gui.DefWindowProc(hwnd, msg, wparam, lparam)
def run_message_loop():
hinst = win32api.GetModuleHandle(None)
wndclass = win32gui.WNDCLASS()
wndclass.hInstance = hinst
wndclass.lpszClassName = "SessionWatcher"
wndclass.lpfnWndProc = session_event_handler
try:
win32gui.RegisterClass(wndclass)
except:
pass # Already registered
hwnd = win32gui.CreateWindow(
wndclass.lpszClassName,
"Session Watcher",
0, 0, 0, 0, 0,
0, hinst, None
)
# Register for session notifications
try:
from win32ts import WTSRegisterSessionNotification
WTSRegisterSessionNotification(hwnd, 1) # NOTIFY_FOR_THIS_SESSION
except ImportError:
# pywin32 might not have win32ts on all systems
pass
# Start message loop
win32gui.PumpMessages()
# Run in a separate thread
thread = threading.Thread(target=run_message_loop, daemon=True)
thread.start()
def get_idle_time() -> int:
"""Get the system idle time in seconds on Windows"""
from ctypes import Structure, windll, c_uint, sizeof, byref
class LASTINPUTINFO(Structure):
_fields_ = [('cbSize', c_uint), ('dwTime', c_uint)]
lastInputInfo = LASTINPUTINFO()
lastInputInfo.cbSize = sizeof(lastInputInfo)
windll.user32.GetLastInputInfo(byref(lastInputInfo))
millis = windll.kernel32.GetTickCount() - lastInputInfo.dwTime
return millis // 1000 # Convert milliseconds to seconds

View File

@@ -1,261 +0,0 @@
import tkinter as tk
from tkinter import simpledialog, messagebox
import threading
import time
from typing import Any, Callable
class OverlayWindow:
def __init__(self, root: tk.Tk, tracker: Any, config: Any):
self.root = root
self.tracker = tracker
self.config = config
# Configure window to be always on top
self.root.attributes("-topmost", True)
self.root.overrideredirect(True) # Remove window decorations
self.root.geometry("300x150+100+100")
# Make window transparent (optional)
self.root.wm_attributes("-transparentcolor", "white")
# Create UI elements
self.create_widgets()
# Bind drag functionality
self.setup_drag()
# Update display periodically
self.update_display_periodically()
def create_widgets(self):
"""Create all UI widgets"""
# Main frame
main_frame = tk.Frame(self.root, bg='white', bd=2, relief='solid')
main_frame.pack(fill='both', expand=True, padx=2, pady=2)
# Status line (tracking enabled/disabled)
self.status_label = tk.Label(main_frame, text="", bg='white', font=('Arial', 10))
self.status_label.pack(anchor='w', padx=5, pady=2)
# Task info
self.task_label = tk.Label(main_frame, text="", bg='white', font=('Arial', 10, 'bold'))
self.task_label.pack(anchor='w', padx=5, pady=2)
# Window info
self.window_label = tk.Label(main_frame, text="", bg='white', font=('Arial', 9))
self.window_label.pack(anchor='w', padx=5, pady=2)
# Total time info
self.total_label = tk.Label(main_frame, text="", bg='white', font=('Arial', 9))
self.total_label.pack(anchor='w', padx=5, pady=2)
# Buttons frame
buttons_frame = tk.Frame(main_frame, bg='white')
buttons_frame.pack(fill='x', padx=5, pady=5)
# Tracking toggle button
self.track_btn = tk.Button(buttons_frame, text="", command=self.toggle_tracking, width=3)
self.track_btn.pack(side='left', padx=2)
# New task button
new_task_btn = tk.Button(buttons_frame, text="N", command=self.new_task, width=3)
new_task_btn.pack(side='left', padx=2)
# Switch task button
switch_task_btn = tk.Button(buttons_frame, text="S", command=self.switch_task, width=3)
switch_task_btn.pack(side='left', padx=2)
# Adjust time button
adjust_btn = tk.Button(buttons_frame, text="±", command=self.adjust_time, width=3)
adjust_btn.pack(side='left', padx=2)
def setup_drag(self):
"""Setup window dragging functionality"""
def start_move(event):
self.root.x = event.x
self.root.y = event.y
def do_move(event):
x = self.root.winfo_x() + (event.x - self.root.x)
y = self.root.winfo_y() + (event.y - self.root.y)
self.root.geometry(f"+{x}+{y}")
# Bind drag events to the main label (covers entire window)
self.status_label.bind("<Button-1>", start_move)
self.status_label.bind("<B1-Motion>", do_move)
# Also bind to other labels for full coverage
for widget in [self.task_label, self.window_label, self.total_label]:
widget.bind("<Button-1>", start_move)
widget.bind("<B1-Motion>", do_move)
def toggle_tracking(self):
"""Toggle tracking state"""
self.tracker.toggle_tracking()
self.update_display()
def new_task(self):
"""Create a new task"""
task_id = simpledialog.askstring("Новая задача", "Введите ID задачи (например, PRJ-123):")
if task_id:
self.tracker.create_task(task_id)
self.update_display()
def switch_task(self):
"""Switch to an existing task"""
available_tasks = self.tracker.get_available_tasks()
if not available_tasks:
messagebox.showinfo("Переключить задачу", "Нет доступных задач")
return
task_list = "\n".join(available_tasks)
selected_task = simpledialog.askstring(
"Переключить задачу",
f"Доступные задачи:\n{task_list}\n\nВведите ID задачи:"
)
if selected_task and selected_task in available_tasks:
self.tracker.switch_task(selected_task)
self.update_display()
def adjust_time(self):
"""Show dialog to adjust time manually"""
self.show_adjust_time_dialog()
def show_adjust_time_dialog(self):
"""Show dialog for manual time adjustment"""
adjustment_window = tk.Toplevel(self.root)
adjustment_window.title("Ручная корректировка времени")
adjustment_window.geometry("300x150")
adjustment_window.transient(self.root)
adjustment_window.grab_set() # Modal window
# Operation selection
operation_var = tk.StringVar(value="+")
tk.Radiobutton(adjustment_window, text="Добавить время", variable=operation_var, value="+").pack(anchor='w',
padx=10,
pady=5)
tk.Radiobutton(adjustment_window, text="Вычесть время", variable=operation_var, value="-").pack(anchor='w',
padx=10, pady=5)
# Minutes entry
tk.Label(adjustment_window, text="Минут:").pack(anchor='w', padx=10)
minutes_entry = tk.Entry(adjustment_window)
minutes_entry.pack(padx=10, pady=5)
minutes_entry.insert(0, "15") # Default value
# Reason entry
tk.Label(adjustment_window, text="Причина:").pack(anchor='w', padx=10)
reason_entry = tk.Entry(adjustment_window)
reason_entry.pack(padx=10, pady=5)
reason_entry.insert(0, "Ручная корректировка") # Default value
def apply_adjustment():
try:
minutes = int(minutes_entry.get())
operation = operation_var.get()
seconds = minutes * 60
if operation == "-":
seconds = -seconds
reason = reason_entry.get()
if self.tracker.has_active_session():
self.tracker.add_manual_adjustment(seconds, reason)
messagebox.showinfo("Успешно", f"Время скорректировано: {operation}{minutes} мин")
adjustment_window.destroy()
self.update_display()
else:
messagebox.showwarning("Внимание", "Нет активной сессии для корректировки")
except ValueError:
messagebox.showerror("Ошибка", "Введите корректное число минут")
# Apply button
apply_btn = tk.Button(adjustment_window, text="Применить", command=apply_adjustment)
apply_btn.pack(pady=10)
def update_display(self):
"""Update the display with current information"""
# Update tracking status
track_status = "" if self.config.tracking_enabled else ""
status_text = f"[{track_status}] Трекинг: {'ВКЛ' if self.config.tracking_enabled else 'ВЫКЛ'}"
self.status_label.config(text=status_text)
# Update task information
if self.config.current_task_id:
# Get current session duration
current_duration = self.tracker.get_current_session_duration()
current_duration_str = self.format_duration(current_duration)
# Get planned duration if available
planned_duration = self.tracker.get_planned_duration(self.config.current_task_id)
if planned_duration:
planned_str = self.format_duration(planned_duration * 60) # Convert minutes to seconds
task_text = f"Задача: {self.config.current_task_id} | {current_duration_str} / {planned_str}"
else:
task_text = f"Задача: {self.config.current_task_id} | {current_duration_str}"
else:
task_text = "Нет активной задачи"
self.task_label.config(text=task_text)
# Update window information
if hasattr(self.tracker, 'active_window'):
window_title = getattr(self.tracker, 'active_window', 'Неизвестно')
classification = getattr(self.tracker, 'window_classification', 'neutral')
class_symbol = {"work": "", "distraction": "", "neutral": ""}[classification]
window_text = f"Окно: {window_title[:30]}{'...' if len(window_title) > 30 else ''} {class_symbol}"
self.window_label.config(text=window_text)
else:
self.window_label.config(text="Окно: Неизвестно")
# Update total time today
total_today = self.tracker.get_total_work_today()
total_str = self.format_duration(total_today)
self.total_label.config(text=f"Всего сегодня: {total_str}")
# Update button colors based on tracking status
if self.config.tracking_enabled:
self.track_btn.config(bg='lightgreen', text="")
else:
self.track_btn.config(bg='lightcoral', text="")
def format_duration(self, seconds: int) -> str:
"""Format seconds into human-readable string (h:mm)"""
if seconds < 0:
seconds = 0
hours = seconds // 3600
minutes = (seconds % 3600) // 60
if hours > 0:
return f"{hours}ч{minutes:02d}м"
else:
return f"{minutes}м"
def update_display_periodically(self):
"""Update display every second"""
self.update_display()
self.root.after(1000, self.update_display_periodically)
def show_notification(self, title: str, message: str, duration: int = 5000):
"""Show a temporary notification"""
# Create notification window
notif = tk.Toplevel(self.root)
notif.title(title)
notif.geometry("250x80+200+200")
notif.overrideredirect(True)
notif.attributes("-topmost", True)
# Style similar to main window
frame = tk.Frame(notif, bg='lightyellow', bd=2, relief='solid')
frame.pack(fill='both', expand=True)
tk.Label(frame, text=title, bg='lightyellow', font=('Arial', 10, 'bold')).pack(pady=5)
tk.Label(frame, text=message, bg='lightyellow', font=('Arial', 9)).pack(pady=5)
# Close after specified duration
notif.after(duration, notif.destroy)

View File

@@ -1,3 +0,0 @@
"""
Package initialization for tests
"""

View File

@@ -1,82 +0,0 @@
import json
import os
import tempfile
from unittest.mock import patch
import pytest
from time_tracker.core.config import load_config, save_config
def test_load_config_creates_default_if_not_exists():
""" creates default when file doesn't exist"""
with tempfile.NamedTemporaryFile(suffix='.json', delete=False) as tmp:
tmp_path = tmp.name
# Remove the file so it doesn't exist
os.unlink(tmp_path)
# Mock the config file path
with patch('time_tracker.core.config.CONFIG_FILE', tmp_path):
config = load_config()
# Check default values
assert config['tracking_enabled'] is True
assert config['window_rules']['work'] == ['PyCharm', 'VS Code']
assert config['window_rules']['distraction'] == ['YouTube', 'Twitter']
assert config.get('current_task_id') is None
# Verify file was created
assert os.path.exists(tmp_path)
def test_save_and_load_config():
"""Test saving and loading config preserves data"""
with tempfile.NamedTemporaryFile(suffix='.json', delete=False) as tmp:
tmp_path = tmp.name
# Clean up after ourselves
if os.path.exists(tmp_path):
os.unlink(tmp_path)
test_config = {
'tracking_enabled': False,
'current_task_id': 'TEST-123',
'window_rules': {
'work': ['Editor'],
'distraction': ['Social Media']
}
}
# Mock the config file path
with patch('time_tracker.core.config.CONFIG_FILE', tmp_path):
save_config(test_config)
loaded_config = load_config()
assert loaded_config['tracking_enabled'] == False
assert loaded_config['current_task_id'] == 'TEST-123'
assert loaded_config['window_rules']['work'] == ['Editor']
assert loaded_config['window_rules']['distraction'] == ['Social Media']
def test_load_config_preserves_existing_values():
"""Test that loading config doesn't overwrite existing values unnecessarily"""
with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as tmp:
initial_config = {
'tracking_enabled': True,
'current_task_id': 'EXISTING-123',
'window_rules': {
'work': ['Custom Editor'],
'distraction': ['Gaming App']
}
}
json.dump(initial_config, tmp)
tmp_path = tmp.name
# Mock the config file path
with patch('time_tracker.core.config.CONFIG_FILE', tmp_path):
loaded_config = load_config()
assert loaded_config['tracking_enabled'] == True
assert loaded_config['current_task_id'] == 'EXISTING-123'
assert loaded_config['window_rules']['work'] == ['Custom Editor']
assert loaded_config['window_rules']['distraction'] == ['Gaming App']

View File

@@ -1,39 +0,0 @@
import sys
from unittest.mock import Mock, patch, MagicMock
import pytest
# Skip these tests if not on Windows
pytestmark = pytest.mark.skipif(sys.platform != "win32", reason="Windows-specific tests")
from time_tracker.platform.windows import get_active_window, register_session_listener
def test_get_active_window():
"""Test getting active window on Windows"""
# This is a complex test that would require actual Windows APIs
# For now, we'll just test that the function exists and doesn't crash
# In real testing, we'd mock win32gui functions
with patch('win32gui.GetForegroundWindow') as mock_get_fg:
with patch('win32gui.GetWindowText') as mock_get_text:
mock_get_fg.return_value = 12345
mock_get_text.return_value = "Test Window"
result = get_active_window()
assert result == "Test Window"
def test_register_session_listener():
"""Test registering session listener on Windows"""
# This is a complex function that requires Windows message handling
# We'll test that the function can be called without error
mock_callback = Mock()
# Since this function starts background threads, we'll just ensure it doesn't crash
try:
register_session_listener(mock_callback)
# If we reach here, the function at least started without crashing
assert True
except Exception:
# If there are import issues or other problems, that's acceptable in test environment
pass

View File

@@ -1,204 +0,0 @@
import json
import os
import tempfile
from datetime import datetime
import pytest
from unittest.mock import patch, MagicMock
from time_tracker.core.session_manager import SessionManager
def test_create_new_session():
"""Test creating a new session"""
with tempfile.TemporaryDirectory() as temp_dir:
# Create tasks directory
tasks_dir = os.path.join(temp_dir, 'tasks')
os.makedirs(tasks_dir, exist_ok=True)
session_manager = SessionManager(tasks_dir)
# Create a new session
session = session_manager.create_session('PRJ-123', 'PRJ')
# Check session has required fields
assert 'session_id' in session
assert session['task_id'] == 'PRJ-123'
assert session['project_id'] == 'PRJ'
assert 'started_at' in session
assert session['duration_seconds'] == 0
assert session['window_details'] == {}
assert session['distractions'] == []
assert session['synced_to_youtrack'] == False
def test_end_current_session():
"""Test ending the current session"""
with tempfile.TemporaryDirectory() as temp_dir:
tasks_dir = os.path.join(temp_dir, 'tasks')
os.makedirs(tasks_dir, exist_ok=True)
session_manager = SessionManager(tasks_dir)
# Start a session
session = session_manager.create_session('PRJ-123', 'PRJ')
session_manager.current_session = session
# Add some window time
session_manager.record_window_time('PyCharm', 300) # 5 minutes
session_manager.record_distraction('YouTube', 120) # 2 minutes
# End the session
ended_session = session_manager.end_current_session()
# Check that session was properly ended
assert 'ended_at' in ended_session
assert ended_session['duration_seconds'] >= 420 # At least 7 minutes
assert ended_session['window_details']['PyCharm'] == 300
assert len(ended_session['distractions']) == 1
assert ended_session['distractions'][0]['window'] == 'YouTube'
assert ended_session['distractions'][0]['duration_seconds'] == 120
# Check that file was saved
session_file_path = os.path.join(tasks_dir, 'PRJ-123', 'sessions',
f"{ended_session['session_id']}_PRJ-123.json")
assert os.path.exists(session_file_path)
# Verify content of saved file
with open(session_file_path, 'r') as f:
saved_session = json.load(f)
assert saved_session == ended_session
def test_record_window_time():
"""Test recording time for specific windows"""
with tempfile.TemporaryDirectory() as temp_dir:
tasks_dir = os.path.join(temp_dir, 'tasks')
os.makedirs(tasks_dir, exist_ok=True)
session_manager = SessionManager(tasks_dir)
session = session_manager.create_session('PRJ-123', 'PRJ')
session_manager.current_session = session
# Record time for multiple windows
session_manager.record_window_time('PyCharm', 300) # 5 minutes
session_manager.record_window_time('Chrome', 120) # 2 minutes
session_manager.record_window_time('PyCharm', 60) # Additional minute
# Check accumulated times
assert session['window_details']['PyCharm'] == 360 # 6 minutes
assert session['window_details']['Chrome'] == 120 # 2 minutes
def test_record_distraction():
"""Test recording distractions"""
with tempfile.TemporaryDirectory() as temp_dir:
tasks_dir = os.path.join(temp_dir, 'tasks')
os.makedirs(tasks_dir, exist_ok=True)
session_manager = SessionManager(tasks_dir)
session = session_manager.create_session('PRJ-123', 'PRJ')
session_manager.current_session = session
# Record a distraction
session_manager.record_distraction('YouTube', 120)
# Check distraction was recorded
assert len(session['distractions']) == 1
distraction = session['distractions'][0]
assert distraction['window'] == 'YouTube'
assert distraction['duration_seconds'] == 120
assert 'start' in distraction
def test_manual_adjustment():
"""Test adding manual time adjustments"""
with tempfile.TemporaryDirectory() as temp_dir:
tasks_dir = os.path.join(temp_dir, 'tasks')
os.makedirs(tasks_dir, exist_ok=True)
session_manager = SessionManager(tasks_dir)
session = session_manager.create_session('PRJ-123', 'PRJ')
session_manager.current_session = session
# Add manual adjustment
session_manager.add_manual_adjustment(900, 'Zoom meeting')
# Check adjustment was recorded
assert len(session['manual_adjustments']) == 1
adjustment = session['manual_adjustments'][0]
assert adjustment['operation'] == 'add'
assert adjustment['seconds'] == 900
assert adjustment['reason'] == 'Zoom meeting'
assert 'applied_at' in adjustment
# Test subtraction
session_manager.add_manual_adjustment(-300, 'Break')
assert len(session['manual_adjustments']) == 2
sub_adjustment = session['manual_adjustments'][1]
assert sub_adjustment['operation'] == 'subtract'
assert sub_adjustment['seconds'] == 300
assert sub_adjustment['reason'] == 'Break'
def test_get_current_session_duration():
"""Test getting current session duration with adjustments"""
with tempfile.TemporaryDirectory() as temp_dir:
tasks_dir = os.path.join(temp_dir, 'tasks')
os.makedirs(tasks_dir, exist_ok=True)
session_manager = SessionManager(tasks_dir)
session = session_manager.create_session('PRJ-123', 'PRJ')
session_manager.current_session = session
# Add some base duration
session['duration_seconds'] = 1800 # 30 minutes
# Add manual adjustments
session_manager.add_manual_adjustment(600, 'Extra work') # +10 min
session_manager.add_manual_adjustment(-300, 'Break') # -5 min
# Calculate final duration
expected_duration = 1800 + 600 - 300 # 2100 seconds = 35 minutes
assert session_manager.get_current_session_duration() == expected_duration
def test_load_sessions_for_task():
"""Test loading all sessions for a specific task"""
with tempfile.TemporaryDirectory() as temp_dir:
tasks_dir = os.path.join(temp_dir, 'tasks')
os.makedirs(tasks_dir, exist_ok=True)
session_manager = SessionManager(tasks_dir)
# Create multiple sessions for the same task
session1 = session_manager.create_session('PRJ-123', 'PRJ')
session1['duration_seconds'] = 1800
session1['ended_at'] = '2026-03-10T10:00:00'
session_manager.end_session(session1)
session2 = session_manager.create_session('PRJ-123', 'PRJ')
session2['duration_seconds'] = 2400
session2['ended_at'] = '2026-03-10T11:00:00'
session_manager.end_session(session2)
# Load sessions for the task
sessions = session_manager.load_sessions_for_task('PRJ-123')
assert len(sessions) == 2
# Sessions should be sorted by end time (newest first)
assert sessions[0]['duration_seconds'] == 2400
assert sessions[1]['duration_seconds'] == 1800
def test_no_sessions_for_task():
"""Test that loading sessions for non-existent task returns empty list"""
with tempfile.TemporaryDirectory() as temp_dir:
tasks_dir = os.path.join(temp_dir, 'tasks')
os.makedirs(tasks_dir, exist_ok=True)
session_manager = SessionManager(tasks_dir)
# Load sessions for non-existent task
sessions = session_manager.load_sessions_for_task('NONEXISTENT-123')
assert sessions == []

View File

@@ -1,156 +0,0 @@
import tkinter as tk
from unittest.mock import Mock, patch, MagicMock
import pytest
from time_tracker.ui.overlay import OverlayWindow
def test_overlay_window_creation():
"""Test that overlay window is created correctly"""
root = tk.Tk()
root.withdraw() # Hide the root window
# Mock dependencies
mock_tracker = Mock()
mock_config = Mock()
overlay = OverlayWindow(root, mock_tracker, mock_config)
# Check that window properties are set
assert overlay.root.wm_attributes('-topmost') == 1
assert overlay.root.overrideredirect() is True
# Check that labels exist
assert hasattr(overlay, 'status_label')
assert hasattr(overlay, 'task_label')
assert hasattr(overlay, 'window_label')
root.destroy()
def test_update_display_with_active_task():
"""Test updating display with an active task"""
root = tk.Tk()
root.withdraw()
mock_tracker = Mock()
mock_config = Mock()
# Set up mock returns
mock_config.tracking_enabled = True
mock_config.current_task_id = 'PRJ-123'
mock_tracker.get_current_session_duration.return_value = 3600 # 1 hour
mock_tracker.get_total_work_today.return_value = 7200 # 2 hours
overlay = OverlayWindow(root, mock_tracker, mock_config)
overlay.update_display()
# Check that labels were updated (we can check the text property)
assert 'PRJ-123' in overlay.task_label.cget('text')
assert '1ч00м' in overlay.status_label.cget('text')
root.destroy()
def test_update_display_without_active_task():
"""Test updating display when no active task"""
root = tk.Tk()
root.withdraw()
mock_tracker = Mock()
mock_config = Mock()
# Set up mock returns
mock_config.tracking_enabled = True
mock_config.current_task_id = None
mock_tracker.get_current_session_duration.return_value = 0
mock_tracker.get_total_work_today.return_value = 0
overlay = OverlayWindow(root, mock_tracker, mock_config)
overlay.update_display()
# Check that labels show appropriate messages
assert 'Нет активной задачи' in overlay.task_label.cget('text')
root.destroy()
def test_toggle_tracking_callback():
"""Test that toggle tracking button works"""
root = tk.Tk()
root.withdraw()
mock_tracker = Mock()
mock_config = Mock()
overlay = OverlayWindow(root, mock_tracker, mock_config)
# Call the toggle method
overlay.toggle_tracking()
# Verify that the tracker's toggle method was called
mock_tracker.toggle_tracking.assert_called_once()
root.destroy()
def test_new_task_callback():
"""Test that new task button opens dialog"""
root = tk.Tk()
root.withdraw()
mock_tracker = Mock()
mock_config = Mock()
overlay = OverlayWindow(root, mock_tracker, mock_config)
# Mock the simpledialog
with patch('tkinter.simpledialog.askstring') as mock_dialog:
mock_dialog.return_value = 'NEW-456'
overlay.new_task()
# Verify that the tracker's create_task method was called
mock_tracker.create_task.assert_called_once_with('NEW-456')
root.destroy()
def test_switch_task_callback():
"""Test that switch task button works"""
root = tk.Tk()
root.withdraw()
mock_tracker = Mock()
mock_config = Mock()
overlay = OverlayWindow(root, mock_tracker, mock_config)
# Mock the available tasks
mock_tracker.get_available_tasks.return_value = ['TASK-1', 'TASK-2', 'TASK-3']
# Mock the selection dialog
with patch('tkinter.simpledialog.askstring') as mock_dialog:
mock_dialog.return_value = 'TASK-2'
overlay.switch_task()
# Verify that the tracker's switch_task method was called
mock_tracker.switch_task.assert_called_once_with('TASK-2')
root.destroy()
def test_adjust_time_callback():
"""Test that adjust time button opens dialog"""
root = tk.Tk()
root.withdraw()
mock_tracker = Mock()
mock_config = Mock()
overlay = OverlayWindow(root, mock_tracker, mock_config)
# Mock the adjustment dialog
with patch.object(overlay, 'show_adjust_time_dialog') as mock_dialog:
overlay.adjust_time()
mock_dialog.assert_called_once()
root.destroy()