feat(activity): add --by-activity flag to break down spent time by activity type

- Load time entry activity names from Redmine enumeration.
- Aggregate hours per issue and per activity in fetch_issues_with_spent_time.
- Add --by-activity CLI flag and propagate it through report builder,
  summary, and all formatters (console, CSV, HTML, JSON, ODT).
- Keep backward-compatible 2-tuple input in build_grouped_report.
- Bump version to 1.8.0.

Closes #41
This commit is contained in:
Кокос Артем Николаевич
2026-06-29 16:44:52 +07:00
parent f6861382e6
commit 59af7ce464
12 changed files with 142 additions and 32 deletions

View File

@@ -111,6 +111,39 @@ def _format_redmine_error(exc: Exception) -> str:
return str(exc)
def _load_time_entry_activities(redmine: Redmine) -> Dict[int, str]:
"""Загружает справочник типов активности time entries.
Возвращает словарь id -> name. Если справочник недоступен,
возвращает пустой словарь — тогда будем использовать данные из самих entries.
"""
try:
activities = redmine.enumeration.filter(resource="time_entry_activities")
return {int(a.id): str(a.name) for a in activities}
except Exception:
return {}
def _get_activity_name(entry, activities: Dict[int, str]) -> str:
"""Определяет название активности для time entry."""
activity = getattr(entry, "activity", None)
if activity is None:
return "<N/A>"
# activity может быть объектом с id/name или просто значением
activity_id = getattr(activity, "id", None)
if activity_id is not None:
name = activities.get(int(activity_id))
if name:
return name
activity_name = getattr(activity, "name", None)
if activity_name:
return str(activity_name)
return str(activity_id)
return str(activity)
def _fetch_issues_chunked(redmine: Redmine, issue_ids: List[int]) -> List[Issue]:
"""Загружает задачи чанками, чтобы не превышать лимит длины URL (#21)."""
all_issues: List[Issue] = []
@@ -186,12 +219,14 @@ def fetch_issues_with_spent_time(
from_date: str,
to_date: str,
user_id: Optional[Union[int, str]] = None,
) -> Optional[List[Tuple[Issue, float]]]:
by_activity: bool = False,
) -> Optional[List[Tuple[Issue, float, Optional[Dict[str, float]]]]]:
"""
Fetch unique issues linked to time entries of the given user in date range,
along with total spent hours per issue.
If user_id is None, uses current user.
Returns list of (issue, total_hours) tuples.
If by_activity is True, returns per-activity breakdown as third tuple element.
Returns list of (issue, total_hours, activities) tuples.
Raises RedmineAPIError on API/auth/network failures.
"""
@@ -202,6 +237,7 @@ def fetch_issues_with_spent_time(
if user_id is not None
else _get_current_user_id(redmine)
)
activities_lookup = _load_time_entry_activities(redmine) if by_activity else {}
time_entries = redmine.time_entry.filter(
user_id=target_user_id, from_date=from_date, to_date=to_date
)
@@ -210,14 +246,21 @@ def fetch_issues_with_spent_time(
except Exception as exc:
raise RedmineAPIError(_format_redmine_error(exc), original=exc) from exc
# Агрегируем часы по issue.id
# Агрегируем часы по issue.id (и активности, если требуется)
spent_time: Dict[int, float] = {}
spent_by_activity: Dict[int, Dict[str, float]] = {}
issue_ids = set()
for entry in time_entries:
if hasattr(entry, "issue") and entry.issue and hasattr(entry, "hours"):
iid = entry.issue.id
hours = float(entry.hours)
issue_ids.add(iid)
spent_time[iid] = spent_time.get(iid, 0.0) + float(entry.hours)
spent_time[iid] = spent_time.get(iid, 0.0) + hours
if by_activity:
activity_name = _get_activity_name(entry, activities_lookup)
by_act = spent_by_activity.setdefault(iid, {})
by_act[activity_name] = by_act.get(activity_name, 0.0) + hours
if not issue_ids:
return None
@@ -234,7 +277,9 @@ def fetch_issues_with_spent_time(
# здесь оставляем порядок API как есть.
result = []
for issue in issues:
total_hours = spent_time.get(issue.id, 0.0)
result.append((issue, total_hours))
iid = issue.id
total_hours = spent_time.get(iid, 0.0)
activity_breakdown = spent_by_activity.get(iid) if by_activity else None
result.append((issue, total_hours, activity_breakdown))
return result