initial 2
This commit is contained in:
Mikan
@@ -12,9 +12,10 @@ 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", # <-- добавлено для работы с моделями и JSON
|
||||
]
|
||||
|
||||
[project.optional-dependencies]
|
||||
dev = [
|
||||
"pytest>=7.0",
|
||||
|
||||
@@ -1 +0,0 @@
|
||||
"""Time Tracker package."""
|
||||
@@ -1,6 +1,16 @@
|
||||
"""Entry point for running the time tracker as a module."""
|
||||
import sys
|
||||
from .app import Application
|
||||
|
||||
|
||||
def main():
|
||||
try:
|
||||
app = Application()
|
||||
app.run()
|
||||
except RuntimeError as e:
|
||||
print(f"Ошибка: {e}", file=sys.stderr)
|
||||
input("Нажмите Enter для выхода...")
|
||||
sys.exit(1)
|
||||
|
||||
from .main import main
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
58
src/time_tracker/app.py
Normal file
58
src/time_tracker/app.py
Normal file
@@ -0,0 +1,58 @@
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
from .models import TrackerConfig, TrackedTask
|
||||
from .utils import get_platform, ensure_dir
|
||||
from .ui.overlay import OverlayWindow
|
||||
|
||||
|
||||
class Application:
|
||||
def __init__(self):
|
||||
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()
|
||||
|
||||
# Загружаем текущую задачу, если указан ID
|
||||
if self.config.current_task_id:
|
||||
self.current_task = self._load_task(self.config.current_task_id)
|
||||
|
||||
# Создаём директорию для задач, если нужно
|
||||
ensure_dir(self.tasks_dir)
|
||||
|
||||
def _load_or_create_config(self) -> TrackerConfig:
|
||||
if self.config_path.exists():
|
||||
with open(self.config_path, "r", encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
return TrackerConfig.model_validate(data)
|
||||
else:
|
||||
config = TrackerConfig()
|
||||
self._save_config(config)
|
||||
return config
|
||||
|
||||
def _save_config(self, config: TrackerConfig) -> None:
|
||||
with open(self.config_path, "w", encoding="utf-8") as f:
|
||||
f.write(config.model_dump_json(indent=2))
|
||||
|
||||
def _load_task(self, task_id: str) -> Optional[TrackedTask]:
|
||||
task_file = self.tasks_dir / f"{task_id}.json"
|
||||
if task_file.exists():
|
||||
try:
|
||||
with open(task_file, "r", encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
return TrackedTask.model_validate(data)
|
||||
except Exception:
|
||||
return None
|
||||
return None
|
||||
|
||||
def save_task(self, task: TrackedTask) -> None:
|
||||
task_file = self.tasks_dir / f"{task.task_id}.json"
|
||||
with open(task_file, "w", encoding="utf-8") as f:
|
||||
f.write(task.model_dump_json(indent=2))
|
||||
|
||||
def run(self):
|
||||
# Пока просто запускаем UI
|
||||
# Позже сюда добавим трекинг в фоне
|
||||
overlay = OverlayWindow()
|
||||
overlay.run()
|
||||
@@ -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,231 +0,0 @@
|
||||
import json
|
||||
import os
|
||||
from datetime import datetime
|
||||
from typing import Dict, List, Optional
|
||||
from pathlib import Path
|
||||
import time
|
||||
|
||||
|
||||
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
|
||||
# Keep track of last session creation time to prevent duplicates
|
||||
self._last_session_time = 0
|
||||
|
||||
def create_session(self, task_id: str, project_id: str) -> Dict:
|
||||
"""
|
||||
Create a new work session.
|
||||
"""
|
||||
# Ensure unique session ID even if called rapidly
|
||||
current_time = time.time()
|
||||
if current_time <= self._last_session_time:
|
||||
current_time = self._last_session_time + 0.001 # Add small increment
|
||||
|
||||
self._last_session_time = current_time
|
||||
session_id = datetime.fromtimestamp(current_time).strftime("%Y%m%dT%H%M%S%f")[:-3] # Include milliseconds
|
||||
|
||||
session = {
|
||||
"session_id": session_id,
|
||||
"task_id": task_id,
|
||||
"project_id": project_id,
|
||||
"started_at": datetime.fromtimestamp(current_time).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,272 +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
|
||||
|
||||
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")
|
||||
def get_idle_time():
|
||||
return 0
|
||||
except ImportError:
|
||||
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")
|
||||
def get_idle_time():
|
||||
return 0
|
||||
|
||||
|
||||
class ConfigWrapper:
|
||||
def __init__(self, config_dict: Dict):
|
||||
self.__dict__.update(config_dict)
|
||||
|
||||
def update(self, new_config: Dict):
|
||||
self.__dict__.update(new_config)
|
||||
|
||||
|
||||
class WorkTracker:
|
||||
def __init__(self):
|
||||
raw_config = load_config()
|
||||
self.config = ConfigWrapper(raw_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.tracking_enabled
|
||||
self.current_task_id = self.config.current_task_id
|
||||
|
||||
# Thread safety
|
||||
self._state_lock = threading.Lock()
|
||||
|
||||
self.start_monitoring()
|
||||
|
||||
def start_monitoring(self):
|
||||
window_thread = threading.Thread(target=self.monitor_windows, daemon=True)
|
||||
window_thread.start()
|
||||
|
||||
input_thread = threading.Thread(target=self.monitor_input_activity, daemon=True)
|
||||
input_thread.start()
|
||||
|
||||
register_session_listener(self.on_session_change)
|
||||
|
||||
def monitor_windows(self):
|
||||
while True:
|
||||
try:
|
||||
current_window = get_active_window()
|
||||
if current_window != self.active_window:
|
||||
with self._state_lock:
|
||||
self.active_window = current_window
|
||||
self.window_classification = self.classify_window(current_window)
|
||||
|
||||
# Notifications will be shown via callbacks set in main()
|
||||
time.sleep(1)
|
||||
except Exception as e:
|
||||
print(f"Error in window monitoring: {e}")
|
||||
time.sleep(5)
|
||||
|
||||
def monitor_input_activity(self):
|
||||
try:
|
||||
from pynput import mouse, keyboard
|
||||
|
||||
def on_activity(*args):
|
||||
self.last_activity = time.time()
|
||||
|
||||
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: periodically check system idle time
|
||||
def idle_fallback():
|
||||
while True:
|
||||
try:
|
||||
idle_sec = get_idle_time()
|
||||
if idle_sec < 5: # Consider active if idle < 5 sec
|
||||
self.last_activity = time.time()
|
||||
except:
|
||||
pass
|
||||
time.sleep(2)
|
||||
threading.Thread(target=idle_fallback, daemon=True).start()
|
||||
|
||||
def classify_window(self, title: str) -> str:
|
||||
title_lower = title.lower()
|
||||
rules = self.config.window_rules
|
||||
|
||||
for keyword in rules.get("distraction", []):
|
||||
if keyword.lower() in title_lower:
|
||||
return "distraction"
|
||||
|
||||
for keyword in rules.get("work", []):
|
||||
if keyword.lower() in title_lower:
|
||||
return "work"
|
||||
|
||||
return "neutral"
|
||||
|
||||
def on_session_change(self, event_type: str):
|
||||
if event_type == "locked":
|
||||
self.session_locked = True
|
||||
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()
|
||||
|
||||
def should_count_as_work(self) -> bool:
|
||||
with self._state_lock:
|
||||
if not self.is_tracking:
|
||||
return False
|
||||
if self.session_locked:
|
||||
return False
|
||||
idle_time = time.time() - self.last_activity
|
||||
if idle_time > 60:
|
||||
return False
|
||||
return self.window_classification == "work"
|
||||
|
||||
def run_main_loop(self):
|
||||
while True:
|
||||
try:
|
||||
if self.should_count_as_work():
|
||||
if self.session_manager.current_session:
|
||||
self.session_manager.increment_duration(1)
|
||||
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):
|
||||
with self._state_lock:
|
||||
self.is_tracking = not self.is_tracking
|
||||
self.config.tracking_enabled = self.is_tracking
|
||||
raw_config = {k: v for k, v in vars(self.config).items() if not k.startswith('_')}
|
||||
save_config(raw_config)
|
||||
|
||||
def create_task(self, task_id: str):
|
||||
project_id = task_id.split('-')[0] if '-' in task_id else task_id
|
||||
self.session_manager.create_session(task_id, project_id)
|
||||
with self._state_lock:
|
||||
self.current_task_id = task_id
|
||||
self.config.current_task_id = task_id
|
||||
raw_config = {k: v for k, v in vars(self.config).items() if not k.startswith('_')}
|
||||
save_config(raw_config)
|
||||
|
||||
def switch_task(self, task_id: str):
|
||||
if self.session_manager.current_session:
|
||||
self.session_manager.end_current_session()
|
||||
|
||||
project_id = task_id.split('-')[0] if '-' in task_id else task_id
|
||||
self.session_manager.create_session(task_id, project_id)
|
||||
with self._state_lock:
|
||||
self.current_task_id = task_id
|
||||
self.config.current_task_id = task_id
|
||||
raw_config = {k: v for k, v in vars(self.config).items() if not k.startswith('_')}
|
||||
save_config(raw_config)
|
||||
|
||||
def get_available_tasks(self) -> List[str]:
|
||||
tasks_dir = "tasks"
|
||||
if not os.path.exists(tasks_dir):
|
||||
return []
|
||||
return [item for item in os.listdir(tasks_dir) if os.path.isdir(os.path.join(tasks_dir, item))]
|
||||
|
||||
def has_active_session(self) -> bool:
|
||||
return self.session_manager.current_session is not None
|
||||
|
||||
def add_manual_adjustment(self, seconds: int, reason: str):
|
||||
if self.session_manager.current_session:
|
||||
self.session_manager.add_manual_adjustment(seconds, reason)
|
||||
|
||||
def get_current_session_duration(self) -> int:
|
||||
return self.session_manager.get_current_session_duration()
|
||||
|
||||
def get_total_work_today(self) -> int:
|
||||
return self.session_manager.get_total_work_today()
|
||||
|
||||
def get_planned_duration(self, task_id: str) -> Optional[int]:
|
||||
return None
|
||||
|
||||
# Notification methods will be replaced by callbacks in main()
|
||||
|
||||
|
||||
def setup_hotkeys(tracker, overlay):
|
||||
try:
|
||||
from pynput import keyboard
|
||||
|
||||
def on_toggle():
|
||||
tracker.toggle_tracking()
|
||||
overlay.update_display()
|
||||
|
||||
def on_new_task():
|
||||
overlay.new_task()
|
||||
|
||||
def on_end_session():
|
||||
if tracker.session_manager.current_session:
|
||||
tracker.session_manager.end_current_session()
|
||||
overlay.update_display()
|
||||
|
||||
def on_adjust():
|
||||
overlay.adjust_time()
|
||||
|
||||
hotkey_map = {
|
||||
'<ctrl>+<alt>+t': on_toggle,
|
||||
'<ctrl>+<alt>+n': on_new_task,
|
||||
'<ctrl>+<alt>+s': on_end_session,
|
||||
'<ctrl>+<alt>+m': on_adjust,
|
||||
}
|
||||
|
||||
listener = keyboard.GlobalHotKeys(hotkey_map)
|
||||
listener.start()
|
||||
return listener
|
||||
except ImportError:
|
||||
print("pynput not available, hotkeys disabled")
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
root = tk.Tk()
|
||||
root.title("Work Tracker")
|
||||
root.geometry("1x1+2000+2000")
|
||||
|
||||
tracker = WorkTracker()
|
||||
overlay = OverlayWindow(root, tracker, tracker.config)
|
||||
|
||||
# Inject notification callbacks
|
||||
tracker.show_distraction_notification = lambda title: overlay.show_notification("Отвлечение", f"Вы используете: {title}")
|
||||
tracker.show_neutral_notification = lambda title: overlay.show_notification("Нейтральное приложение", f"Окно: {title}")
|
||||
|
||||
main_loop_thread = threading.Thread(target=tracker.run_main_loop, daemon=True)
|
||||
main_loop_thread.start()
|
||||
|
||||
hotkey_listener = setup_hotkeys(tracker, overlay)
|
||||
|
||||
try:
|
||||
root.mainloop()
|
||||
except KeyboardInterrupt:
|
||||
print("\nShutting down...")
|
||||
if tracker.session_manager.current_session:
|
||||
tracker.session_manager.end_current_session()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
74
src/time_tracker/models.py
Normal file
74
src/time_tracker/models.py
Normal file
@@ -0,0 +1,74 @@
|
||||
from datetime import datetime
|
||||
from typing import List, Optional
|
||||
from pydantic import BaseModel, Field
|
||||
from typing_extensions import Literal
|
||||
|
||||
|
||||
class WorkDetail(BaseModel):
|
||||
group_name: str
|
||||
duration_seconds: float
|
||||
|
||||
|
||||
class WorkSession(BaseModel):
|
||||
start_time: datetime
|
||||
end_time: datetime
|
||||
details: List[WorkDetail]
|
||||
synchronized: bool = False
|
||||
|
||||
|
||||
class TrackedTask(BaseModel):
|
||||
task_id: str
|
||||
project_id: Optional[str] = None
|
||||
estimated_seconds: Optional[float] = None
|
||||
start_time: datetime
|
||||
end_time: Optional[datetime] = None
|
||||
sessions: List[WorkSession] = Field(default_factory=list)
|
||||
|
||||
@property
|
||||
def total_worked_seconds(self) -> float:
|
||||
return sum(
|
||||
detail.duration_seconds
|
||||
for session in self.sessions
|
||||
for detail in session.details
|
||||
)
|
||||
|
||||
@property
|
||||
def remaining_seconds(self) -> Optional[float]:
|
||||
if self.estimated_seconds is None:
|
||||
return None
|
||||
return max(0.0, self.estimated_seconds - self.total_worked_seconds)
|
||||
|
||||
|
||||
class WindowPolicy(BaseModel):
|
||||
group_name: str
|
||||
window_patterns: List[str] = Field(default_factory=list)
|
||||
process_names: List[str] = Field(default_factory=list)
|
||||
policy_type: Literal["work", "blocked"]
|
||||
|
||||
|
||||
class TrackerConfig(BaseModel):
|
||||
tracking_enabled: bool = False
|
||||
current_task_id: Optional[str] = None
|
||||
window_policies: List[WindowPolicy] = Field(default_factory=list)
|
||||
|
||||
@staticmethod
|
||||
def default_policies() -> List[WindowPolicy]:
|
||||
return [
|
||||
WindowPolicy(
|
||||
group_name="IDE",
|
||||
window_patterns=["PyCharm", "IntelliJ", "Visual Studio Code", "Code", "Sublime", "Vim", "Neovim"],
|
||||
process_names=["pycharm64.exe", "idea64.exe", "code.exe", "sublime_text.exe", "nvim.exe", "vim.exe"],
|
||||
policy_type="work"
|
||||
),
|
||||
WindowPolicy(
|
||||
group_name="Gaming & Distractions",
|
||||
window_patterns=["Steam", "Discord", "YouTube", "Twitch", "Spotify", "Netflix"],
|
||||
process_names=["steam.exe", "discord.exe", "spotify.exe"],
|
||||
policy_type="blocked"
|
||||
),
|
||||
]
|
||||
|
||||
def __init__(self, **data):
|
||||
super().__init__(**data)
|
||||
if not self.window_policies:
|
||||
self.window_policies = self.default_policies()
|
||||
0
src/time_tracker/platform/__init__.py
Normal file
0
src/time_tracker/platform/__init__.py
Normal file
@@ -1,19 +1,16 @@
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import Callable
|
||||
from typing import NamedTuple
|
||||
|
||||
|
||||
class PlatformInterface(ABC):
|
||||
class WindowInfo(NamedTuple):
|
||||
title: str
|
||||
process_name: str
|
||||
|
||||
|
||||
class PlatformBase(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"""
|
||||
def get_active_window(self) -> WindowInfo:
|
||||
"""
|
||||
Получает информацию об активном окне: заголовок и имя процесса.
|
||||
"""
|
||||
pass
|
||||
@@ -1,90 +1,34 @@
|
||||
import win32gui
|
||||
import win32process
|
||||
import win32con
|
||||
import win32api
|
||||
import threading
|
||||
from typing import Callable
|
||||
import psutil
|
||||
from .base import PlatformBase, WindowInfo
|
||||
|
||||
|
||||
def get_active_window() -> str:
|
||||
"""Returns a string like 'YouTube - Chrome (chrome.exe)'"""
|
||||
try:
|
||||
hwnd = win32gui.GetForegroundWindow()
|
||||
title = win32gui.GetWindowText(hwnd) or "Unknown Window"
|
||||
|
||||
# Get process name
|
||||
_, pid = win32process.GetWindowThreadProcessId(hwnd)
|
||||
class WindowsPlatform(PlatformBase):
|
||||
def get_active_window(self) -> WindowInfo:
|
||||
try:
|
||||
proc = psutil.Process(pid)
|
||||
exe_name = proc.name()
|
||||
except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
|
||||
exe_name = "unknown.exe"
|
||||
hwnd = win32gui.GetForegroundWindow()
|
||||
if not hwnd:
|
||||
return WindowInfo(title="", process_name="")
|
||||
|
||||
return f"{title} ({exe_name})"
|
||||
except Exception:
|
||||
return "Error getting window info"
|
||||
# Получаем заголовок окна
|
||||
title = win32gui.GetWindowText(hwnd).strip()
|
||||
|
||||
# Получаем PID процесса
|
||||
_, pid = win32process.GetWindowThreadProcessId(hwnd)
|
||||
|
||||
def register_session_listener(callback: Callable[[str], None]) -> None:
|
||||
"""Register a callback for Windows session lock/unlock events"""
|
||||
if pid == 0:
|
||||
return WindowInfo(title=title, process_name="")
|
||||
|
||||
def session_event_handler(hwnd, msg, wparam, lparam):
|
||||
# Fallback numeric constants
|
||||
WM_WTSSESSION_CHANGE = 0x02B1
|
||||
WTS_SESSION_LOCK = 0x1
|
||||
WTS_SESSION_UNLOCK = 0x2
|
||||
# Получаем имя исполняемого файла процесса
|
||||
try:
|
||||
process = psutil.Process(pid)
|
||||
process_name = process.name()
|
||||
except (psutil.NoSuchProcess, psutil.AccessDenied):
|
||||
process_name = ""
|
||||
|
||||
if msg == WM_WTSSESSION_CHANGE:
|
||||
if wparam == WTS_SESSION_LOCK:
|
||||
callback("locked")
|
||||
elif wparam == WTS_SESSION_UNLOCK:
|
||||
callback("unlocked")
|
||||
return WindowInfo(title=title, process_name=process_name)
|
||||
|
||||
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, 0, hinst, None
|
||||
)
|
||||
|
||||
# Register for session notifications
|
||||
try:
|
||||
from win32ts import WTSRegisterSessionNotification
|
||||
WTSRegisterSessionNotification(hwnd, 1)
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
win32gui.PumpMessages()
|
||||
|
||||
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
|
||||
except Exception:
|
||||
# В случае ошибки (например, недоступно окно) — возвращаем пустые данные
|
||||
return WindowInfo(title="", process_name="")
|
||||
1
src/time_tracker/ui/__init__.py
Normal file
1
src/time_tracker/ui/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
from .overlay import OverlayWindow
|
||||
24
src/time_tracker/ui/adjust_time_window.py
Normal file
24
src/time_tracker/ui/adjust_time_window.py
Normal file
@@ -0,0 +1,24 @@
|
||||
import tkinter as tk
|
||||
from tkinter import Toplevel, Label, Entry, Button
|
||||
|
||||
|
||||
class AdjustTimeWindow:
|
||||
def __init__(self, parent):
|
||||
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):
|
||||
# Позже: применить изменение
|
||||
self.window.destroy()
|
||||
29
src/time_tracker/ui/new_task_window.py
Normal file
29
src/time_tracker/ui/new_task_window.py
Normal file
@@ -0,0 +1,29 @@
|
||||
import tkinter as tk
|
||||
from tkinter import Toplevel, Label, Entry, Button, Frame
|
||||
|
||||
|
||||
class NewTaskWindow:
|
||||
def __init__(self, parent):
|
||||
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):
|
||||
# Позже: сохранить задачу
|
||||
self.window.destroy()
|
||||
@@ -1,219 +1,112 @@
|
||||
import tkinter as tk
|
||||
from tkinter import simpledialog, messagebox
|
||||
import threading
|
||||
import time
|
||||
from typing import Any
|
||||
from tkinter import Menu, Label, Button, Frame
|
||||
from typing import Optional
|
||||
|
||||
|
||||
class OverlayWindow:
|
||||
def __init__(self, root: tk.Tk, tracker: Any, config: Any):
|
||||
self.root = root
|
||||
self.tracker = tracker
|
||||
self.config = config
|
||||
def __init__(self):
|
||||
self.root = tk.Tk()
|
||||
self.root.title("Citrus Time Tracker")
|
||||
self.root.geometry("320x180")
|
||||
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)
|
||||
|
||||
# Always on top, no decorations
|
||||
self.root.attributes("-topmost", True)
|
||||
self.root.overrideredirect(True)
|
||||
self.root.geometry("320x180+100+100")
|
||||
# Для перетаскивания
|
||||
self._drag_data = {"x": 0, "y": 0}
|
||||
self.root.bind("<Button-1>", self.start_drag)
|
||||
self.root.bind("<B1-Motion>", self.drag)
|
||||
|
||||
# Semi-transparent by default
|
||||
self.root.wm_attributes("-alpha", 0.7)
|
||||
|
||||
# Hover effects
|
||||
self.root.bind("<Enter>", self.on_hover)
|
||||
self.root.bind("<Leave>", self.on_leave)
|
||||
|
||||
self.create_widgets()
|
||||
self.setup_drag_full()
|
||||
# Контекстное меню
|
||||
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)
|
||||
|
||||
# Start periodic updates
|
||||
self.update_display_periodically()
|
||||
# Центрирование
|
||||
self.center_window()
|
||||
|
||||
def on_hover(self, event=None):
|
||||
self.root.wm_attributes("-alpha", 1.0)
|
||||
# Содержимое
|
||||
self._build_ui()
|
||||
|
||||
def on_leave(self, event=None):
|
||||
self.root.wm_attributes("-alpha", 0.7)
|
||||
def center_window(self):
|
||||
self.root.update_idletasks()
|
||||
screen_width = self.root.winfo_screenwidth()
|
||||
screen_height = self.root.winfo_screenheight()
|
||||
x = (screen_width // 2) - (320 // 2)
|
||||
y = (screen_height // 2) - (180 // 2)
|
||||
self.root.geometry(f"320x180+{x}+{y}")
|
||||
|
||||
def setup_drag_full(self):
|
||||
"""Allow dragging the entire window from any point."""
|
||||
def start_move(event):
|
||||
self._drag_x = event.x
|
||||
self._drag_y = event.y
|
||||
def _build_ui(self):
|
||||
# Главный фрейм с отступами
|
||||
main_frame = Frame(self.root, padx=10, pady=10)
|
||||
main_frame.pack(fill="both", expand=True)
|
||||
|
||||
def do_move(event):
|
||||
x = self.root.winfo_x() + (event.x - self._drag_x)
|
||||
y = self.root.winfo_y() + (event.y - self._drag_y)
|
||||
self.root.geometry(f"+{x}+{y}")
|
||||
# Заголовок
|
||||
self.status_label = Label(main_frame, text="Citrus Time Tracker: Бездействие", anchor="w")
|
||||
self.status_label.pack(anchor="w")
|
||||
|
||||
self.root.bind("<Button-1>", start_move)
|
||||
self.root.bind("<B1-Motion>", do_move)
|
||||
# Задача
|
||||
self.task_label = Label(main_frame, text="Задача: <Не выбрано>", anchor="w")
|
||||
self.task_label.pack(anchor="w")
|
||||
|
||||
def create_widgets(self):
|
||||
main_frame = tk.Frame(self.root, bg='#f0f0f0', bd=2, relief='solid')
|
||||
main_frame.pack(fill='both', expand=True, padx=2, pady=2)
|
||||
# Время
|
||||
self.time_info_label = Label(main_frame, text="Времени нет", anchor="w", font=("TkDefaultFont", 9))
|
||||
self.time_info_label.pack(anchor="w")
|
||||
|
||||
self.status_label = tk.Label(main_frame, text="", bg='#f0f0f0', font=('Arial', 9))
|
||||
self.status_label.pack(anchor='w', padx=5, pady=2)
|
||||
# Текущее окно
|
||||
self.window_group_label = Label(main_frame, text="Неизвестное приложение", anchor="w", fg="gray")
|
||||
self.window_group_label.pack(anchor="w")
|
||||
|
||||
self.task_label = tk.Label(main_frame, text="", bg='#f0f0f0', font=('Arial', 10, 'bold'))
|
||||
self.task_label.pack(anchor='w', padx=5, pady=2)
|
||||
self.window_detail_label = Label(main_frame, text="", anchor="w", font=("TkDefaultFont", 8), fg="gray")
|
||||
self.window_detail_label.pack(anchor="w")
|
||||
|
||||
self.window_label = tk.Label(main_frame, text="", bg='#f0f0f0', font=('Arial', 9))
|
||||
self.window_label.pack(anchor='w', padx=5, pady=2)
|
||||
# Кнопки
|
||||
btn_frame = Frame(main_frame)
|
||||
btn_frame.pack(side="bottom", fill="x", pady=(10, 0))
|
||||
|
||||
self.total_label = tk.Label(main_frame, text="", bg='#f0f0f0', font=('Arial', 9))
|
||||
self.total_label.pack(anchor='w', padx=5, pady=2)
|
||||
Button(btn_frame, text="⏸ Пауза", command=self.on_pause).pack(side="left", fill="x", expand=True, padx=(0, 2))
|
||||
Button(btn_frame, text="➕ Новая", command=self.on_new_task).pack(side="left", fill="x", expand=True, padx=2)
|
||||
Button(btn_frame, text="🔍 Выбрать", command=self.on_select_task).pack(side="left", fill="x", expand=True, padx=2)
|
||||
Button(btn_frame, text="✏️ Время", command=self.on_adjust_time).pack(side="left", fill="x", expand=True, padx=(2, 0))
|
||||
|
||||
buttons_frame = tk.Frame(main_frame, bg='#f0f0f0')
|
||||
buttons_frame.pack(fill='x', padx=5, pady=5)
|
||||
# === Обработчики событий (пока без логики, только открытие окон) ===
|
||||
|
||||
self.track_btn = tk.Button(buttons_frame, text="▶", command=self.toggle_tracking, width=8)
|
||||
self.track_btn.pack(side='left', padx=2)
|
||||
def on_pause(self):
|
||||
pass # будет реализовано позже
|
||||
|
||||
tk.Button(buttons_frame, text="+Задача", command=self.new_task, width=8).pack(side='left', padx=2)
|
||||
tk.Button(buttons_frame, text="↔Задача", command=self.switch_task, width=8).pack(side='left', padx=2)
|
||||
tk.Button(buttons_frame, text="±Время", command=self.adjust_time, width=8).pack(side='left', padx=2)
|
||||
def on_new_task(self):
|
||||
from .new_task_window import NewTaskWindow
|
||||
NewTaskWindow(self.root)
|
||||
|
||||
def on_select_task(self):
|
||||
from .select_task_window import SelectTaskWindow
|
||||
SelectTaskWindow(self.root)
|
||||
|
||||
def on_adjust_time(self):
|
||||
from .adjust_time_window import AdjustTimeWindow
|
||||
AdjustTimeWindow(self.root)
|
||||
|
||||
# === 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):
|
||||
context_menu = tk.Menu(self.root, tearoff=0)
|
||||
context_menu.add_command(label="Закрыть", command=self.root.quit)
|
||||
context_menu.post(event.x_root, event.y_root)
|
||||
self.context_menu.tk_popup(event.x_root, event.y_root)
|
||||
|
||||
def toggle_tracking(self):
|
||||
self.tracker.toggle_tracking()
|
||||
self.update_display()
|
||||
|
||||
def new_task(self):
|
||||
task_id = simpledialog.askstring("Новая задача", "Введите ID задачи (например, PRJ-123):")
|
||||
if task_id and task_id.strip():
|
||||
self.tracker.create_task(task_id.strip())
|
||||
self.update_display()
|
||||
|
||||
def switch_task(self):
|
||||
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.strip() in available_tasks:
|
||||
self.tracker.switch_task(selected_task.strip())
|
||||
self.update_display()
|
||||
|
||||
def adjust_time(self):
|
||||
self.show_adjust_time_dialog()
|
||||
|
||||
def show_adjust_time_dialog(self):
|
||||
adjustment_window = tk.Toplevel(self.root)
|
||||
adjustment_window.title("Ручная корректировка времени")
|
||||
adjustment_window.geometry("350x200")
|
||||
adjustment_window.transient(self.root)
|
||||
adjustment_window.grab_set()
|
||||
|
||||
operation_var = tk.StringVar(value="+")
|
||||
tk.Radiobutton(adjustment_window, text="Добавить время", variable=operation_var, value="+").pack(anchor='w', padx=10, pady=2)
|
||||
tk.Radiobutton(adjustment_window, text="Вычесть время", variable=operation_var, value="-").pack(anchor='w', padx=10, pady=2)
|
||||
|
||||
tk.Label(adjustment_window, text="Минуты:").pack(anchor='w', padx=10, pady=(10,0))
|
||||
minutes_entry = tk.Entry(adjustment_window)
|
||||
minutes_entry.pack(padx=10, pady=2)
|
||||
minutes_entry.insert(0, "15")
|
||||
|
||||
tk.Label(adjustment_window, text="Причина:").pack(anchor='w', padx=10, pady=(10,0))
|
||||
reason_entry = tk.Entry(adjustment_window)
|
||||
reason_entry.pack(padx=10, pady=2)
|
||||
reason_entry.insert(0, "Перерыв / встреча")
|
||||
|
||||
def apply_adjustment():
|
||||
try:
|
||||
minutes = int(minutes_entry.get())
|
||||
if minutes <= 0:
|
||||
raise ValueError("Minutes must be positive")
|
||||
operation = operation_var.get()
|
||||
seconds = minutes * 60 * (1 if operation == "+" else -1)
|
||||
reason = reason_entry.get().strip()
|
||||
if not reason:
|
||||
reason = "Без причины"
|
||||
|
||||
if self.tracker.has_active_session():
|
||||
self.tracker.add_manual_adjustment(seconds, reason)
|
||||
messagebox.showinfo("Успех", f"Изменено: {operation}{minutes} мин\nПричина: {reason}")
|
||||
adjustment_window.destroy()
|
||||
self.update_display()
|
||||
else:
|
||||
messagebox.showwarning("Ошибка", "Нет активной сессии!")
|
||||
except ValueError:
|
||||
messagebox.showerror("Ошибка", "Введите положительное число минут")
|
||||
|
||||
tk.Button(adjustment_window, text="Применить", command=apply_adjustment, bg='lightblue').pack(pady=10)
|
||||
|
||||
def update_display(self):
|
||||
# Use lock to avoid race conditions
|
||||
with self.tracker._state_lock:
|
||||
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)
|
||||
|
||||
if self.config.current_task_id:
|
||||
current_duration = self.tracker.get_current_session_duration()
|
||||
current_str = self.format_duration(current_duration)
|
||||
task_text = f"Задача: {self.config.current_task_id} | {current_str}"
|
||||
else:
|
||||
task_text = "Нет активной задачи"
|
||||
self.task_label.config(text=task_text)
|
||||
|
||||
window_title = getattr(self.tracker, 'active_window', 'Неизвестно')
|
||||
classification = getattr(self.tracker, 'window_classification', 'neutral')
|
||||
class_symbol = {"work": "✅", "distraction": "⚠", "neutral": "⚪"}[classification]
|
||||
display_title = window_title[:40] + '...' if len(window_title) > 40 else window_title
|
||||
window_text = f"Окно: {display_title} {class_symbol}"
|
||||
self.window_label.config(text=window_text)
|
||||
|
||||
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 appearance
|
||||
self.track_btn.config(
|
||||
text="⏸" if self.config.tracking_enabled else "▶",
|
||||
bg='lightgreen' if self.config.tracking_enabled else 'lightcoral'
|
||||
)
|
||||
|
||||
def format_duration(self, seconds: int) -> str:
|
||||
"""Format as H:MM:SS or M:SS"""
|
||||
if seconds < 0:
|
||||
seconds = 0
|
||||
hours, remainder = divmod(seconds, 3600)
|
||||
minutes, secs = divmod(remainder, 60)
|
||||
if hours > 0:
|
||||
return f"{hours}:{minutes:02d}:{secs:02d}"
|
||||
else:
|
||||
return f"{minutes}:{secs:02d}"
|
||||
|
||||
def update_display_periodically(self):
|
||||
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 window"""
|
||||
notif = tk.Toplevel(self.root)
|
||||
notif.title(title)
|
||||
notif.geometry("280x90+200+200")
|
||||
notif.overrideredirect(True)
|
||||
notif.attributes("-topmost", True)
|
||||
notif.wm_attributes("-alpha", 0.9)
|
||||
|
||||
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[:80], bg='lightyellow', font=('Arial', 9), wraplength=260).pack(pady=5)
|
||||
|
||||
notif.after(duration, notif.destroy)
|
||||
def run(self):
|
||||
self.root.mainloop()
|
||||
32
src/time_tracker/ui/select_task_window.py
Normal file
32
src/time_tracker/ui/select_task_window.py
Normal file
@@ -0,0 +1,32 @@
|
||||
import tkinter as tk
|
||||
from tkinter import Toplevel, Listbox, Scrollbar, Button, Frame
|
||||
|
||||
|
||||
class SelectTaskWindow:
|
||||
def __init__(self, parent):
|
||||
self.window = Toplevel(parent)
|
||||
self.window.title("Выбрать задачу")
|
||||
self.window.geometry("300x250")
|
||||
self.window.transient(parent)
|
||||
self.window.grab_set()
|
||||
|
||||
list_frame = 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)
|
||||
self.listbox.pack(side="left", fill="both", expand=True)
|
||||
|
||||
# Заглушка: список задач
|
||||
for i in range(5):
|
||||
self.listbox.insert("end", f"TASK-{i+1} (1 ч 20 мин)")
|
||||
|
||||
scrollbar.config(command=self.listbox.yview)
|
||||
|
||||
Button(self.window, text="Выбрать", command=self.on_select).pack(pady=5)
|
||||
|
||||
def on_select(self):
|
||||
# Позже: выбрать задачу
|
||||
self.window.destroy()
|
||||
16
src/time_tracker/utils.py
Normal file
16
src/time_tracker/utils.py
Normal file
@@ -0,0 +1,16 @@
|
||||
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)
|
||||
@@ -1,3 +0,0 @@
|
||||
"""
|
||||
Package initialization for tests
|
||||
"""
|
||||
@@ -1,85 +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
|
||||
# Updated to match actual defaults
|
||||
expected_work_apps = ['PyCharm', 'VS Code', 'Visual Studio', 'Sublime Text', 'Atom', 'IntelliJ IDEA']
|
||||
assert config['window_rules']['work'] == expected_work_apps
|
||||
expected_distraction_apps = ['YouTube', 'Twitter', 'Facebook', 'Instagram', 'TikTok', 'Discord', 'Slack']
|
||||
assert config['window_rules']['distraction'] == expected_distraction_apps
|
||||
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,133 +0,0 @@
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
from unittest.mock import patch, MagicMock
|
||||
import tkinter as tk
|
||||
from time_tracker.main import WorkTracker, ConfigWrapper
|
||||
|
||||
|
||||
def test_tracker_initialization():
|
||||
"""Test that WorkTracker initializes without errors"""
|
||||
# Create a temporary config file to avoid creating one in the current directory
|
||||
with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as tmp:
|
||||
json.dump({
|
||||
"tracking_enabled": True,
|
||||
"current_task_id": None,
|
||||
"window_rules": {
|
||||
"work": ["PyCharm", "VS Code"],
|
||||
"distraction": ["YouTube", "Twitter"]
|
||||
}
|
||||
}, tmp)
|
||||
tmp_path = tmp.name
|
||||
|
||||
original_config_path = "config.json"
|
||||
|
||||
# Temporarily replace the config file
|
||||
if os.path.exists(original_config_path):
|
||||
os.rename(original_config_path, f"{original_config_path}.backup")
|
||||
|
||||
os.rename(tmp_path, original_config_path)
|
||||
|
||||
try:
|
||||
# Initialize the tracker
|
||||
tracker = WorkTracker()
|
||||
|
||||
# Verify that config is properly loaded as a ConfigWrapper object
|
||||
assert hasattr(tracker, 'config')
|
||||
assert isinstance(tracker.config, ConfigWrapper)
|
||||
assert hasattr(tracker.config, 'tracking_enabled')
|
||||
assert hasattr(tracker.config, 'current_task_id')
|
||||
assert hasattr(tracker.config, 'window_rules')
|
||||
assert tracker.config.tracking_enabled is True
|
||||
assert tracker.config.current_task_id is None
|
||||
|
||||
finally:
|
||||
# Restore original config file
|
||||
os.remove(original_config_path)
|
||||
if os.path.exists(f"{original_config_path}.backup"):
|
||||
os.rename(f"{original_config_path}.backup", original_config_path)
|
||||
|
||||
|
||||
def test_config_wrapper_functionality():
|
||||
"""Test that ConfigWrapper provides attribute access to config values"""
|
||||
config_dict = {
|
||||
"tracking_enabled": True,
|
||||
"current_task_id": "TEST-123",
|
||||
"window_rules": {
|
||||
"work": ["PyCharm", "VS Code"],
|
||||
"distraction": ["YouTube", "Twitter"]
|
||||
}
|
||||
}
|
||||
|
||||
wrapper = ConfigWrapper(config_dict)
|
||||
|
||||
# Test attribute access
|
||||
assert wrapper.tracking_enabled is True
|
||||
assert wrapper.current_task_id == "TEST-123"
|
||||
assert "PyCharm" in wrapper.window_rules["work"]
|
||||
|
||||
# Test updating config
|
||||
wrapper.tracking_enabled = False
|
||||
assert wrapper.tracking_enabled is False
|
||||
# Note: Direct assignment doesn't update the original dict, but the wrapper still works
|
||||
|
||||
|
||||
def test_tracker_config_attribute_access():
|
||||
"""Test that config can be accessed as an object with attributes"""
|
||||
|
||||
# Create a mock config object that behaves like the real config
|
||||
class MockConfig:
|
||||
def __init__(self):
|
||||
self.tracking_enabled = True
|
||||
self.current_task_id = "TEST-123"
|
||||
|
||||
config = MockConfig()
|
||||
assert hasattr(config, 'tracking_enabled')
|
||||
assert config.tracking_enabled is True
|
||||
assert config.current_task_id == "TEST-123"
|
||||
|
||||
|
||||
def test_overlay_compatible_config():
|
||||
"""Test that the config structure is compatible with overlay expectations"""
|
||||
from time_tracker.core.config import load_config
|
||||
|
||||
# Create a temporary config file
|
||||
with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as tmp:
|
||||
json.dump({
|
||||
"tracking_enabled": True,
|
||||
"current_task_id": "TEST-123",
|
||||
"window_rules": {
|
||||
"work": ["PyCharm", "VS Code"],
|
||||
"distraction": ["YouTube", "Twitter"]
|
||||
}
|
||||
}, tmp)
|
||||
tmp_path = tmp.name
|
||||
|
||||
original_config_path = "config.json"
|
||||
|
||||
# Temporarily replace the config file
|
||||
if os.path.exists(original_config_path):
|
||||
os.rename(original_config_path, f"{original_config_path}.backup")
|
||||
|
||||
os.rename(tmp_path, original_config_path)
|
||||
|
||||
try:
|
||||
# Load config using the actual function
|
||||
config = load_config()
|
||||
|
||||
# Verify structure
|
||||
assert isinstance(config, dict)
|
||||
assert config.get("tracking_enabled") is True
|
||||
assert config.get("current_task_id") == "TEST-123"
|
||||
|
||||
# Create a compatible object that supports attribute access
|
||||
wrapped_config = ConfigWrapper(config)
|
||||
assert hasattr(wrapped_config, 'tracking_enabled')
|
||||
assert wrapped_config.tracking_enabled is True
|
||||
assert wrapped_config.current_task_id == "TEST-123"
|
||||
|
||||
finally:
|
||||
# Restore original config file
|
||||
os.remove(original_config_path)
|
||||
if os.path.exists(f"{original_config_path}.backup"):
|
||||
os.rename(f"{original_config_path}.backup", original_config_path)
|
||||
@@ -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,211 +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 duration directly to the session
|
||||
session['duration_seconds'] = 420 # 7 minutes
|
||||
# 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 # Exactly 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 two separate session managers to avoid conflicts
|
||||
# First session
|
||||
sm1 = SessionManager(tasks_dir)
|
||||
session1 = sm1.create_session('PRJ-123', 'PRJ')
|
||||
session1['duration_seconds'] = 1800
|
||||
session1['ended_at'] = '2026-03-10T10:00:00'
|
||||
sm1._save_session(session1) # Directly save to file
|
||||
|
||||
# Second session
|
||||
sm2 = SessionManager(tasks_dir)
|
||||
session2 = sm2.create_session('PRJ-123', 'PRJ')
|
||||
session2['duration_seconds'] = 2400
|
||||
session2['ended_at'] = '2026-03-10T11:00:00'
|
||||
sm2._save_session(session2) # Directly save to file
|
||||
|
||||
# Load sessions for the task using a fresh manager
|
||||
sm3 = SessionManager(tasks_dir)
|
||||
sessions = sm3.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 # Newest session first
|
||||
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,97 +0,0 @@
|
||||
import tkinter as tk
|
||||
from unittest.mock import Mock
|
||||
import pytest
|
||||
from time_tracker.ui.overlay import OverlayWindow
|
||||
|
||||
|
||||
def test_overlay_window_appears():
|
||||
"""Test that overlay window actually appears and has expected widgets"""
|
||||
root = tk.Tk()
|
||||
root.withdraw() # Hide the root window initially
|
||||
|
||||
# Mock dependencies with proper return values
|
||||
mock_tracker = Mock()
|
||||
mock_tracker.get_current_session_duration.return_value = 0
|
||||
mock_tracker.get_total_work_today.return_value = 0
|
||||
mock_tracker.get_planned_duration.return_value = None
|
||||
mock_tracker.active_window = 'Desktop'
|
||||
mock_tracker.window_classification = 'neutral'
|
||||
mock_tracker.has_active_session.return_value = False
|
||||
mock_tracker.get_available_tasks.return_value = []
|
||||
|
||||
mock_config = Mock()
|
||||
mock_config.tracking_enabled = True
|
||||
mock_config.current_task_id = None
|
||||
|
||||
# Create overlay window
|
||||
overlay = OverlayWindow(root, mock_tracker, mock_config)
|
||||
|
||||
# Check that the window is actually created and visible
|
||||
assert overlay.root is not None
|
||||
assert isinstance(overlay.root, tk.Tk)
|
||||
|
||||
# Check that window properties are set
|
||||
assert overlay.root.wm_attributes('-topmost') == 1
|
||||
|
||||
# Check that required widgets exist
|
||||
assert hasattr(overlay, 'status_label')
|
||||
assert hasattr(overlay, 'task_label')
|
||||
assert hasattr(overlay, 'window_label')
|
||||
assert hasattr(overlay, 'total_label')
|
||||
|
||||
# Check that buttons exist
|
||||
assert hasattr(overlay, 'track_btn')
|
||||
|
||||
# Check that labels have been created and are not None
|
||||
assert overlay.status_label is not None
|
||||
assert overlay.task_label is not None
|
||||
assert overlay.window_label is not None
|
||||
assert overlay.total_label is not None
|
||||
|
||||
# Check that the widgets are packed/grid
|
||||
assert overlay.status_label.winfo_exists() == 1
|
||||
assert overlay.task_label.winfo_exists() == 1
|
||||
assert overlay.window_label.winfo_exists() == 1
|
||||
assert overlay.total_label.winfo_exists() == 1
|
||||
|
||||
root.destroy()
|
||||
|
||||
|
||||
def test_overlay_window_widgets_are_visible():
|
||||
"""Test that overlay window widgets are properly configured and visible"""
|
||||
root = tk.Tk()
|
||||
root.withdraw()
|
||||
|
||||
# Mock dependencies
|
||||
mock_tracker = Mock()
|
||||
mock_tracker.get_current_session_duration.return_value = 3600 # 1 hour
|
||||
mock_tracker.get_total_work_today.return_value = 7200 # 2 hours
|
||||
mock_tracker.get_planned_duration.return_value = None
|
||||
mock_tracker.active_window = 'PyCharm'
|
||||
mock_tracker.window_classification = 'work'
|
||||
mock_tracker.has_active_session.return_value = True
|
||||
mock_tracker.get_available_tasks.return_value = ['TASK-1', 'TASK-2']
|
||||
|
||||
mock_config = Mock()
|
||||
mock_config.tracking_enabled = True
|
||||
mock_config.current_task_id = 'PRJ-123'
|
||||
|
||||
overlay = OverlayWindow(root, mock_tracker, mock_config)
|
||||
|
||||
# Force an update to populate the labels
|
||||
overlay.update_display()
|
||||
|
||||
# Check that labels have text content
|
||||
status_text = overlay.status_label.cget('text')
|
||||
task_text = overlay.task_label.cget('text')
|
||||
window_text = overlay.window_label.cget('text')
|
||||
total_text = overlay.total_label.cget('text')
|
||||
|
||||
# Verify that the labels contain expected information
|
||||
assert 'Трекинг:' in status_text
|
||||
assert 'PRJ-123' in task_text
|
||||
assert 'Окно:' in window_text
|
||||
assert 'Всего сегодня:' in total_text
|
||||
|
||||
root.destroy()
|
||||
|
||||
@@ -1,206 +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 with proper return values
|
||||
mock_tracker = Mock()
|
||||
mock_tracker.get_current_session_duration.return_value = 0
|
||||
mock_tracker.get_total_work_today.return_value = 0
|
||||
mock_tracker.get_planned_duration.return_value = None
|
||||
mock_tracker.active_window = 'Desktop'
|
||||
mock_tracker.window_classification = 'neutral'
|
||||
|
||||
mock_config = Mock()
|
||||
mock_config.tracking_enabled = True
|
||||
mock_config.current_task_id = None
|
||||
|
||||
# Create overlay - this should not raise an exception
|
||||
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 dependencies with proper return values
|
||||
mock_tracker = Mock()
|
||||
mock_tracker.get_current_session_duration.return_value = 3600 # 1 hour
|
||||
mock_tracker.get_total_work_today.return_value = 7200 # 2 hours
|
||||
mock_tracker.get_planned_duration.return_value = None
|
||||
mock_tracker.active_window = 'PyCharm'
|
||||
mock_tracker.window_classification = 'work'
|
||||
|
||||
mock_config = Mock()
|
||||
mock_config.tracking_enabled = True
|
||||
mock_config.current_task_id = 'PRJ-123'
|
||||
|
||||
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')
|
||||
# The task label should contain the duration, not the status label
|
||||
assert '1ч00м' in overlay.task_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 dependencies with proper return values
|
||||
mock_tracker = Mock()
|
||||
mock_tracker.get_current_session_duration.return_value = 0
|
||||
mock_tracker.get_total_work_today.return_value = 0
|
||||
mock_tracker.get_planned_duration.return_value = None
|
||||
mock_tracker.active_window = 'Desktop'
|
||||
mock_tracker.window_classification = 'neutral'
|
||||
|
||||
mock_config = Mock()
|
||||
mock_config.tracking_enabled = True
|
||||
mock_config.current_task_id = None
|
||||
|
||||
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 dependencies with proper return values
|
||||
mock_tracker = Mock()
|
||||
mock_tracker.get_current_session_duration.return_value = 0
|
||||
mock_tracker.get_total_work_today.return_value = 0
|
||||
mock_tracker.get_planned_duration.return_value = None
|
||||
mock_tracker.active_window = 'Desktop'
|
||||
mock_tracker.window_classification = 'neutral'
|
||||
|
||||
mock_config = Mock()
|
||||
mock_config.tracking_enabled = True
|
||||
mock_config.current_task_id = None
|
||||
|
||||
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 dependencies with proper return values
|
||||
mock_tracker = Mock()
|
||||
mock_tracker.get_current_session_duration.return_value = 0
|
||||
mock_tracker.get_total_work_today.return_value = 0
|
||||
mock_tracker.get_planned_duration.return_value = None
|
||||
mock_tracker.active_window = 'Desktop'
|
||||
mock_tracker.window_classification = 'neutral'
|
||||
|
||||
mock_config = Mock()
|
||||
mock_config.tracking_enabled = True
|
||||
mock_config.current_task_id = None
|
||||
|
||||
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 dependencies with proper return values
|
||||
mock_tracker = Mock()
|
||||
mock_tracker.get_current_session_duration.return_value = 0
|
||||
mock_tracker.get_total_work_today.return_value = 0
|
||||
mock_tracker.get_planned_duration.return_value = None
|
||||
mock_tracker.active_window = 'Desktop'
|
||||
mock_tracker.window_classification = 'neutral'
|
||||
mock_tracker.get_available_tasks.return_value = ['TASK-1', 'TASK-2', 'TASK-3']
|
||||
|
||||
mock_config = Mock()
|
||||
mock_config.tracking_enabled = True
|
||||
mock_config.current_task_id = None
|
||||
|
||||
overlay = OverlayWindow(root, mock_tracker, mock_config)
|
||||
|
||||
# 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 dependencies with proper return values
|
||||
mock_tracker = Mock()
|
||||
mock_tracker.get_current_session_duration.return_value = 0
|
||||
mock_tracker.get_total_work_today.return_value = 0
|
||||
mock_tracker.get_planned_duration.return_value = None
|
||||
mock_tracker.active_window = 'Desktop'
|
||||
mock_tracker.window_classification = 'neutral'
|
||||
|
||||
mock_config = Mock()
|
||||
mock_config.tracking_enabled = True
|
||||
mock_config.current_task_id = None
|
||||
|
||||
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