e.vlasov/redmine_tracker
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

добавил объединение повторных активностей Clockify

Browse Source
This commit is contained in:
Власов Эдуард
2026-07-14 15:23:02 +03:00
parent 9352a84723
commit 5dca3ae446
4 changed files with 181 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
classes/__init__.py
View File

@@ -1 +1,7 @@
from .classes import Activity, ActivityType, Tag, VetroProject # noqa: F401
from .classes import ( # noqa: F401
Activity,
ActivityType,
AggregatedActivity,
Tag,
VetroProject,
)

9
classes/classes.py
View File

@@ -6,6 +6,8 @@ from enums import ClockifyProjects, VetroProjects
@dataclass
class Activity:
"""Описывает одну активность, полученную из Clockify."""
id: str
author_id: int
vetro_project_id: VetroProjects
@@ -22,6 +24,13 @@ class Activity:
return f'({self.task_id}) - "{self.description}"'[:60]
@dataclass
class AggregatedActivity(Activity):
"""Представляет несколько активностей Clockify одной записью Redmine."""
source_activities: tuple[Activity, ...] = ()
@dataclass
class Tag:
id: str

68
services/tracking.py
View File

@@ -4,11 +4,48 @@ import logging
from colorama import Fore
from classes import Activity, AggregatedActivity
from utils import format_date, format_hours
logger = logging.getLogger(__name__)
def aggregate_activities(activities: list[Activity]) -> list[AggregatedActivity]:
"""Объединяет одинаковые активности и суммирует их трудозатраты."""
grouped_activities = {}
for activity in activities:
key = (
activity.author_id,
activity.task_id,
activity.tag_id,
activity.project_id,
activity.vetro_project_id,
activity.date_start.date(),
)
grouped_activities.setdefault(key, []).append(activity)
result = []
for grouped in grouped_activities.values():
descriptions = list(
dict.fromkeys(
description.rstrip(".")
for activity in grouped
if (description := activity.description.strip())
)
)
result.append(
AggregatedActivity(
**{
**grouped[0].__dict__,
"description": ". ".join(descriptions),
"time_spent": sum(activity.time_spent for activity in grouped),
},
source_activities=tuple(grouped),
)
)
return result
class TrackingService:
"""Координирует загрузку, сохранение и перенос активностей в Redmine."""
@@ -35,13 +72,20 @@ class TrackingService:
if not activities:
logger.info(Fore.YELLOW + "Нет задач для занесения трудочасов" + Fore.RESET)
return []
activities_to_track = aggregate_activities(
[
activity
for activity in activities
if not activity.is_tracked and activity.date_end
]
)
if self.is_dry_run:
self._log_preview(activities)
self._log_preview(activities, activities_to_track)
return activities
self.db.insert_activities(activities)
self.red.track_activities(
activities,
activities_to_track,
on_success=None if self.is_debug else self._save_tracked_activity,
)
return activities
@@ -66,22 +110,20 @@ class TrackingService:
def _save_tracked_activity(self, activity):
"""Сохраняет активность как успешно перенесенную в Redmine."""
self.db.insert_activities([activity])
self.db.insert_time_entries([activity])
source_activities = list(activity.source_activities or (activity,))
for source_activity in source_activities:
source_activity.is_tracked = True
self.db.insert_activities(source_activities)
self.db.insert_time_entries(source_activities)
def _log_preview(self, activities):
def _log_preview(self, activities, activities_to_track):
"""Выводит план переноса активностей без сохранения и отправки."""
total_hours = sum(activity.time_spent for activity in activities if activity.date_end)
trackable_activities = [
activity
for activity in activities
if not activity.is_tracked and activity.date_end
]
total_hours = sum(activity.time_spent for activity in activities_to_track)
logger.info(f"{Fore.YELLOW}Предпросмотр переноса трудочасов{Fore.RESET}")
logger.info(f"Найдено активностей: {len(activities)}")
logger.info(f"Готово к переносу: {len(trackable_activities)}")
logger.info(f"Готово к переносу: {len(activities_to_track)}")
logger.info(f"Всего часов: {format_hours(total_hours)}")
for activity in trackable_activities:
for activity in activities_to_track:
logger.info(
f"{activity.date_start.date()} | #{activity.task_id} | "
f"{format_hours(activity.time_spent)} | {activity.description}"

110
tests/test_tracking.py Normal file
View File

@@ -0,0 +1,110 @@
"""Тесты агрегации активностей Clockify перед переносом в Redmine."""
from datetime import datetime
from unittest import TestCase
from unittest.mock import Mock
from classes import Activity
from services.tracking import TrackingService, aggregate_activities
class TrackingServiceTestCase(TestCase):
"""Проверяет объединение и сохранение активностей Clockify."""
def create_activity(self, activity_id: str, description: str, **changes) -> Activity:
"""Создаёт активность с типовыми значениями для теста."""
values = {
"id": activity_id,
"author_id": 1,
"vetro_project_id": "vetro-project",
"task_id": 51906,
"tag_id": "tag",
"description": description,
"project_id": "project",
"time_spent": 0.5,
"date_start": datetime(2026, 7, 14, 9),
"date_end": datetime(2026, 7, 14, 9, 30),
}
values.update(changes)
return Activity(**values)
def test_fully_matching_activities_are_merged(self):
"""Полностью одинаковые активности образуют одну запись без повтора описания."""
activities = [
self.create_activity("first", "Доработки"),
self.create_activity(
"second",
"Доработки",
date_start=datetime(2026, 7, 14, 10),
date_end=datetime(2026, 7, 14, 10, 30),
),
]
result = aggregate_activities(activities)
self.assertEqual(len(result), 1)
self.assertEqual(result[0].time_spent, 1)
self.assertEqual(result[0].description, "Доработки")
self.assertEqual(result[0].source_activities, tuple(activities))
def test_descriptions_are_joined_for_otherwise_matching_activities(self):
"""Разные описания одинаковых активностей объединяются через точку."""
result = aggregate_activities(
[
self.create_activity("first", "Доработки"),
self.create_activity("second", "Влил в тест.", time_spent=0.25),
self.create_activity("third", "Доработки", time_spent=0.25),
]
)
self.assertEqual(len(result), 1)
self.assertEqual(result[0].time_spent, 1)
self.assertEqual(result[0].description, "Доработки. Влил в тест")
def test_activities_for_different_dates_are_not_merged(self):
"""Активности за разные даты остаются отдельными записями Redmine."""
result = aggregate_activities(
[
self.create_activity("first", "Доработки"),
self.create_activity(
"second",
"Доработки",
date_start=datetime(2026, 7, 15, 9),
date_end=datetime(2026, 7, 15, 9, 30),
),
]
)
self.assertEqual(len(result), 2)
def test_all_source_activities_are_marked_as_tracked(self):
"""После успешной отправки сохраняются все исходные таймеры Clockify."""
activities = [
self.create_activity("first", "Доработки"),
self.create_activity("second", "Влил в тест"),
]
aggregated_activity = aggregate_activities(activities)[0]
db = Mock()
service = TrackingService(
clock=Mock(),
red=Mock(),
db=db,
is_debug=False,
)
service._save_tracked_activity(aggregated_activity)
self.assertTrue(all(activity.is_tracked for activity in activities))
db.insert_activities.assert_called_once_with(activities)
db.insert_time_entries.assert_called_once_with(activities)
def test_activities_for_different_tasks_are_not_merged(self):
"""Активности разных задач остаются отдельными записями Redmine."""
result = aggregate_activities(
[
self.create_activity("first", "Доработки"),
self.create_activity("second", "Доработки", task_id=60130),
]
)
self.assertEqual(len(result), 2)