Compare commits
16 Commits
9ad75f11e5
...
master
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
332bc0a150 | ||
|
|
534aa9a7d1 | ||
|
|
0f8a86e536 | ||
|
|
1afa6f71ca | ||
|
|
4b56ec4641 | ||
|
|
4c260b1bad | ||
|
|
97dbcfd8f0 | ||
|
|
bf51fc454d | ||
|
|
07ed356cf1 | ||
|
|
6946b8fb50 | ||
|
|
346ea77fd2 | ||
|
|
320f20b974 | ||
|
|
daf1c2a37b | ||
|
|
01cf0f87d0 | ||
|
|
c635b0261b | ||
|
|
89231b9005 |
3
.env.example
Normal file
3
.env.example
Normal 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
91
.gitignore
vendored
Normal 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
|
||||
15
README.md
15
README.md
@@ -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
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -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.1.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"]
|
||||
|
||||
0
src/citrus_time_tracker/__init__.py
Normal file
0
src/citrus_time_tracker/__init__.py
Normal file
18
src/citrus_time_tracker/__main__.py
Normal file
18
src/citrus_time_tracker/__main__.py
Normal 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()
|
||||
523
src/citrus_time_tracker/app.py
Normal file
523
src/citrus_time_tracker/app.py
Normal file
@@ -0,0 +1,523 @@
|
||||
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)
|
||||
current_idle = (current_time - self.last_activity > self.config.idle_threshold_seconds) or is_blocked
|
||||
|
||||
was_idle = self.is_idle
|
||||
self.is_idle = current_idle
|
||||
|
||||
# === СЛУЧАЙ 1: Переход в бездействие ===
|
||||
if not was_idle and current_idle:
|
||||
self.idle_start_time = current_time
|
||||
self.session_start_time = None
|
||||
self.last_window_group = None
|
||||
self.maybe_save_task()
|
||||
logger.debug(f"Переход в бездействие после {current_time - self.last_activity:.0f} сек")
|
||||
return
|
||||
|
||||
# === СЛУЧАЙ 2: Выход из бездействия ===
|
||||
if was_idle and not current_idle:
|
||||
idle_duration = current_time - self.idle_start_time if self.idle_start_time else 0
|
||||
self.idle_start_time = None
|
||||
self.session_start_time = current_time
|
||||
self.last_window_group = group_name
|
||||
|
||||
# Создаём НОВУЮ сессию при длительном простое или смене дня
|
||||
should_create_new = False
|
||||
reasons = []
|
||||
|
||||
if idle_duration > self.config.new_session_after_idle_seconds:
|
||||
should_create_new = True
|
||||
reasons.append(f"простой {idle_duration / 60:.1f} мин")
|
||||
|
||||
if self.current_task.sessions:
|
||||
last_date = self.current_task.sessions[-1].end_time.date()
|
||||
current_date = datetime.now().date()
|
||||
if last_date < current_date:
|
||||
should_create_new = True
|
||||
reasons.append(f"новый день ({last_date} → {current_date})")
|
||||
|
||||
if not self.current_task.sessions:
|
||||
should_create_new = True
|
||||
reasons.append("первая сессия")
|
||||
|
||||
if should_create_new:
|
||||
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}: {', '.join(reasons)}"
|
||||
)
|
||||
|
||||
# Добавляем первую запись в новую/текущую сессию
|
||||
if self.current_task.sessions:
|
||||
self._add_work_detail_to_session(self.current_task.sessions[-1], group_name, 5.0)
|
||||
return
|
||||
|
||||
# === СЛУЧАЙ 3: Продолжение активной работы ===
|
||||
if not current_idle:
|
||||
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(self.current_task.sessions[-1], group_name, 5.0)
|
||||
if self.session_start_time is None:
|
||||
self.session_start_time = current_time
|
||||
return
|
||||
|
||||
def _add_work_detail_to_session(self, session: WorkSession, group_name: str, seconds: float):
|
||||
"""Добавляет запись в существующую сессию и помечает её как изменённую"""
|
||||
# Объединяем записи одной группы
|
||||
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()
|
||||
|
||||
# КРИТИЧЕСКИ ВАЖНО: сбрасываем флаг при любом изменении!
|
||||
if session.synchronized:
|
||||
session.synchronized = False
|
||||
logger.debug(
|
||||
f"Сессия {session.id or 'без ID'} помечена как изменённая "
|
||||
f"(было: {session.youtrack_duration_minutes or 0:.1f} мин, "
|
||||
f"стало: {session.total_minutes:.1f} мин)"
|
||||
)
|
||||
|
||||
|
||||
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:
|
||||
logger.warning("Невозможно добавить время: задача не выбрана")
|
||||
return
|
||||
|
||||
if minutes == 0:
|
||||
logger.info("Пропущено добавление 0 минут")
|
||||
return
|
||||
|
||||
# Создаём ОТДЕЛЬНУЮ сессию для ручного времени
|
||||
sign = "+" if minutes > 0 else "-"
|
||||
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: {sign}{abs(minutes):.1f} min"
|
||||
)
|
||||
self.current_task.sessions.append(new_session)
|
||||
logger.info(
|
||||
f"Создана ручная сессия: {minutes:+.1f} мин для {self.current_task.task_id} "
|
||||
f"(всего сессий: {len(self.current_task.sessions)})"
|
||||
)
|
||||
|
||||
# Сохраняем и синхронизируем немедленно
|
||||
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):
|
||||
if not self.current_task:
|
||||
return
|
||||
|
||||
client = YouTrackClient()
|
||||
if not client.enabled:
|
||||
logger.info("YouTrack: синхронизация отключена (нет настроек в .env)")
|
||||
return
|
||||
|
||||
if not client.issue_exists(self.current_task.task_id):
|
||||
logger.warning(f"YouTrack: задача {self.current_task.task_id} не существует — синхронизация пропущена")
|
||||
return
|
||||
|
||||
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 = session.total_minutes
|
||||
|
||||
if total_minutes <= 0:
|
||||
session.synchronized = True
|
||||
session.youtrack_duration_minutes = 0.0
|
||||
changed = True
|
||||
continue
|
||||
|
||||
# === СЛУЧАЙ A: Сессия с ID — проверяем необходимость обновления ===
|
||||
if session.id:
|
||||
# Проверяем изменение ДАЖЕ если помечена как synchronized
|
||||
needs_update = not session.synchronized or session.is_modified()
|
||||
|
||||
if needs_update:
|
||||
if client.update_work_item(self.current_task.task_id, session.id, total_minutes):
|
||||
session.synchronized = True
|
||||
session.youtrack_duration_minutes = total_minutes
|
||||
changed = True
|
||||
logger.info(
|
||||
f"Обновлена сессия {session.id} для {self.current_task.task_id}: "
|
||||
f"{total_minutes:.1f} мин"
|
||||
)
|
||||
else:
|
||||
logger.warning(f"Не удалось обновить сессию {session.id}")
|
||||
# else: сессия не изменилась — пропускаем
|
||||
|
||||
# === СЛУЧАЙ B: Сессия без ID — создаём новую запись ===
|
||||
else:
|
||||
work_item_id = client.add_work_item(
|
||||
self.current_task.task_id,
|
||||
total_minutes,
|
||||
description=session.description or f"Auto: {', '.join(set(d.group_name for d in session.details))}"
|
||||
)
|
||||
if work_item_id:
|
||||
session.id = work_item_id
|
||||
session.synchronized = True
|
||||
session.youtrack_duration_minutes = total_minutes
|
||||
# Пытаемся найти точное совпадение для youtrack_created_at
|
||||
remote_match = next((r for r in remote_items if r['id'] == work_item_id), None)
|
||||
if remote_match and remote_match.get('created'):
|
||||
session.youtrack_created_at = datetime.fromtimestamp(remote_match['created'] / 1000)
|
||||
changed = True
|
||||
logger.info(
|
||||
f"Создана запись {work_item_id} для {self.current_task.task_id}: "
|
||||
f"{total_minutes:.1f} мин"
|
||||
)
|
||||
else:
|
||||
logger.warning(f"Не удалось создать запись для сессии без ID")
|
||||
|
||||
# Шаг 2: Добавляем удалённые записи, которых нет локально
|
||||
local_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_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_duration_minutes=remote_item['minutes'],
|
||||
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']} "
|
||||
f"({remote_item['minutes']} мин) в локальную задачу"
|
||||
)
|
||||
|
||||
if changed:
|
||||
self._save_task_no_sync(self.current_task)
|
||||
logger.info(
|
||||
f"Синхронизация завершена для {self.current_task.task_id}: "
|
||||
f"{len(self.current_task.sessions)} сессий"
|
||||
)
|
||||
|
||||
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))
|
||||
106
src/citrus_time_tracker/models.py
Normal file
106
src/citrus_time_tracker/models.py
Normal file
@@ -0,0 +1,106 @@
|
||||
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
|
||||
id: Optional[str] = None
|
||||
youtrack_created_at: Optional[datetime] = None
|
||||
description: Optional[str] = None
|
||||
youtrack_duration_minutes: Optional[float] = None
|
||||
|
||||
class Config:
|
||||
json_encoders = {
|
||||
datetime: lambda v: v.isoformat()
|
||||
}
|
||||
|
||||
@property
|
||||
def total_minutes(self) -> float:
|
||||
return sum(d.duration_seconds for d in self.details) / 60.0
|
||||
|
||||
def is_modified(self) -> bool:
|
||||
"""Проверяет, изменилась ли длительность сессии после последней синхронизации"""
|
||||
if self.youtrack_duration_minutes is None:
|
||||
return not self.synchronized # Несинхронизированная сессия = изменённая
|
||||
return abs(self.total_minutes - self.youtrack_duration_minutes) > 0.5 # погрешность 30 сек
|
||||
|
||||
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)
|
||||
idle_threshold_seconds: int = 60 # Порог перехода в idle
|
||||
new_session_after_idle_seconds: int = 600 # 10 минут для новой сессии
|
||||
|
||||
@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()
|
||||
|
||||
|
||||
class YouTrackTimeTrackerEntryDuration(BaseModel):
|
||||
minutes: int
|
||||
|
||||
class YouTrackTimeTrackerEntry(BaseModel):
|
||||
id: str
|
||||
duration: YouTrackTimeTrackerEntryDuration
|
||||
|
||||
@property
|
||||
def minutes(self) -> int:
|
||||
return self.duration.minutes
|
||||
0
src/citrus_time_tracker/platform/__init__.py
Normal file
0
src/citrus_time_tracker/platform/__init__.py
Normal file
16
src/citrus_time_tracker/platform/base.py
Normal file
16
src/citrus_time_tracker/platform/base.py
Normal 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
|
||||
34
src/citrus_time_tracker/platform/windows.py
Normal file
34
src/citrus_time_tracker/platform/windows.py
Normal 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="")
|
||||
1
src/citrus_time_tracker/ui/__init__.py
Normal file
1
src/citrus_time_tracker/ui/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
from .overlay import OverlayWindow
|
||||
67
src/citrus_time_tracker/ui/adjust_time_window.py
Normal file
67
src/citrus_time_tracker/ui/adjust_time_window.py
Normal 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()
|
||||
69
src/citrus_time_tracker/ui/new_task_window.py
Normal file
69
src/citrus_time_tracker/ui/new_task_window.py
Normal 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()
|
||||
188
src/citrus_time_tracker/ui/overlay.py
Normal file
188
src/citrus_time_tracker/ui/overlay.py
Normal 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="Unknown app", 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()
|
||||
53
src/citrus_time_tracker/ui/report_window.py
Normal file
53
src/citrus_time_tracker/ui/report_window.py
Normal 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)
|
||||
55
src/citrus_time_tracker/ui/select_task_window.py
Normal file
55
src/citrus_time_tracker/ui/select_task_window.py
Normal 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()
|
||||
26
src/citrus_time_tracker/utils.py
Normal file
26
src/citrus_time_tracker/utils.py
Normal 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}"
|
||||
157
src/citrus_time_tracker/youtrack_client.py
Normal file
157
src/citrus_time_tracker/youtrack_client.py
Normal file
@@ -0,0 +1,157 @@
|
||||
import json
|
||||
from typing import List, Optional, Tuple
|
||||
from datetime import datetime
|
||||
|
||||
import httpx
|
||||
import os
|
||||
import logging
|
||||
from pathlib import Path
|
||||
|
||||
from citrus_time_tracker.models import YouTrackTimeTrackerEntry
|
||||
|
||||
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, description: Optional[str] = None) -> Optional[str]:
|
||||
"""
|
||||
Создаёт новый work item и возвращает его ID
|
||||
"""
|
||||
if not self.enabled:
|
||||
return None
|
||||
|
||||
if minutes <= 0:
|
||||
return None
|
||||
|
||||
payload: dict = {
|
||||
"duration": {"minutes": round(minutes)},
|
||||
"date": int(datetime.now().timestamp() * 1000) # milliseconds
|
||||
}
|
||||
if description:
|
||||
payload["text"] = description
|
||||
|
||||
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):
|
||||
data = resp.json()
|
||||
work_item_id = data.get('id')
|
||||
logger.info(f"YouTrack: +{round(minutes)} мин к {issue_id} (ID: {work_item_id})")
|
||||
return work_item_id
|
||||
else:
|
||||
logger.error(f"YouTrack: ошибка создания времени ({issue_id}): {resp.status_code} – {resp.text}")
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.error(f"YouTrack: сетевая ошибка при создании времени ({issue_id}): {e}")
|
||||
return None
|
||||
|
||||
def update_work_item(self, issue_id: str, work_item_id: str, minutes: float) -> bool:
|
||||
"""
|
||||
Обновляет существующий work item
|
||||
"""
|
||||
if not self.enabled:
|
||||
return False
|
||||
|
||||
if minutes <= 0:
|
||||
return False
|
||||
|
||||
payload = {
|
||||
"duration": {"minutes": round(minutes)},
|
||||
"id": work_item_id
|
||||
}
|
||||
|
||||
try:
|
||||
with self._get_client() as client:
|
||||
url = f"{self.base_url}/api/issues/{issue_id}/timeTracking/workItems/{work_item_id}"
|
||||
resp = client.post(url, headers=self.headers, json=payload) # YouTrack использует POST для обновления
|
||||
if resp.status_code in (200, 201):
|
||||
logger.info(f"YouTrack: обновлено время {work_item_id} на {round(minutes)} мин для {issue_id}")
|
||||
return True
|
||||
else:
|
||||
logger.error(
|
||||
f"YouTrack: ошибка обновления времени ({work_item_id}): {resp.status_code} – {resp.text}")
|
||||
return False
|
||||
except Exception as e:
|
||||
logger.error(f"YouTrack: сетевая ошибка при обновлении времени ({work_item_id}): {e}")
|
||||
return False
|
||||
|
||||
def get_issue_work_items(self, issue_id: str) -> List[dict]:
|
||||
"""
|
||||
Возвращает полные данные work items с метаданными для синхронизации
|
||||
"""
|
||||
if not self.enabled:
|
||||
return []
|
||||
try:
|
||||
with self._get_client() as client:
|
||||
# Запрашиваем больше полей для точного сопоставления
|
||||
url = f"{self.base_url}/api/issues/{issue_id}/timeTracking/workItems"
|
||||
url += "?fields=id,duration(minutes),created,text"
|
||||
resp = client.get(url, headers=self.headers)
|
||||
if resp.status_code != 200:
|
||||
logger.error(f"YouTrack: ошибка получения work items ({issue_id}): {resp.status_code}")
|
||||
return []
|
||||
|
||||
data = resp.json()
|
||||
work_items = []
|
||||
for item in data:
|
||||
work_items.append({
|
||||
'id': item.get('id'),
|
||||
'minutes': item.get('duration', {}).get('minutes', 0),
|
||||
'created': item.get('created'), # timestamp в миллисекундах
|
||||
'text': item.get('text', '')
|
||||
})
|
||||
return work_items
|
||||
except Exception as e:
|
||||
logger.error(f"YouTrack: сетевая ошибка при получении work items для {issue_id}: {e}")
|
||||
return []
|
||||
@@ -1 +0,0 @@
|
||||
"""Time Tracker package."""
|
||||
@@ -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)
|
||||
@@ -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
|
||||
@@ -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()
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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)
|
||||
@@ -1,3 +0,0 @@
|
||||
"""
|
||||
Package initialization for tests
|
||||
"""
|
||||
@@ -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']
|
||||
@@ -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
|
||||
@@ -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 == []
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user