- 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
153 lines
5.8 KiB
Python
153 lines
5.8 KiB
Python
from typing import Dict, List, Optional, Tuple, cast
|
||
|
||
from redminelib.resources import Issue
|
||
|
||
from .types import ReportRow
|
||
from .utils import get_version, hours_to_human
|
||
|
||
STATUS_TRANSLATION = {
|
||
"New": "В работе",
|
||
"In Progress": "В работе",
|
||
"Feedback": "В работе",
|
||
"Re-opened": "В работе",
|
||
"Code Review": "Решена",
|
||
"Wait Release": "Закрыто",
|
||
"Pending": "Ожидание",
|
||
"Resolved": "Решена",
|
||
"Testing": "Решена",
|
||
"Confirming": "Ожидание",
|
||
"Closed": "Закрыто",
|
||
"Rejected": "Закрыто",
|
||
"Frozen": "Ожидание",
|
||
}
|
||
|
||
|
||
def _format_activities(activities: Dict[str, float]) -> str:
|
||
"""Форматирует разбивку по активностям в многострочный текст."""
|
||
if not activities:
|
||
return ""
|
||
# Сортируем по убыванию часов, затем по алфавиту для стабильности
|
||
items = sorted(activities.items(), key=lambda x: (-x[1], x[0]))
|
||
return "\n".join(f"{hours_to_human(hours)} {name}" for name, hours in items)
|
||
|
||
|
||
def build_grouped_report(
|
||
issue_hours: List[Tuple[Issue, float, Optional[Dict[str, float]]]],
|
||
fill_time: bool = True,
|
||
by_activity: bool = False,
|
||
) -> List[ReportRow]:
|
||
"""
|
||
Преобразует список задач с затраченным временем в плоский список строк отчёта,
|
||
с учётом группировки по проекту и версии (пустые ячейки для повторяющихся значений).
|
||
|
||
Предусловие: issue_hours должен быть отсортирован по (project, version).
|
||
Функция выполняет сортировку самостоятельно для защиты от несортированного ввода.
|
||
"""
|
||
|
||
# Защитная сортировка -- гарантирует корректную группировку независимо от порядка на входе
|
||
issue_hours = sorted(issue_hours, key=lambda x: (str(x[0].project), get_version(x[0]), x[0].id))
|
||
|
||
rows: List[ReportRow] = []
|
||
prev_project: str = ""
|
||
prev_version: str = ""
|
||
|
||
for issue, hours, *rest in issue_hours:
|
||
activities: Optional[Dict[str, float]] = rest[0] if rest else None
|
||
project = str(issue.project)
|
||
version = get_version(issue)
|
||
status_en = str(issue.status)
|
||
status_ru = STATUS_TRANSLATION.get(status_en, status_en)
|
||
|
||
if fill_time:
|
||
if by_activity and activities:
|
||
time_text = _format_activities(activities)
|
||
else:
|
||
time_text = hours_to_human(hours)
|
||
else:
|
||
time_text = ""
|
||
|
||
display_project = project if project != prev_project else ""
|
||
display_version = version if (project != prev_project or version != prev_version) else ""
|
||
|
||
rows.append(
|
||
cast(
|
||
ReportRow,
|
||
{
|
||
"project": project,
|
||
"version": version,
|
||
"display_project": display_project,
|
||
"display_version": display_version,
|
||
"issue_id": issue.id,
|
||
"subject": issue.subject,
|
||
"status_ru": status_ru,
|
||
"time_text": time_text,
|
||
"hours": round(hours, 2),
|
||
"activities": activities,
|
||
},
|
||
)
|
||
)
|
||
|
||
prev_project = project
|
||
prev_version = version
|
||
|
||
return rows
|
||
|
||
|
||
def calculate_summary(
|
||
rows: List[ReportRow],
|
||
by_activity: bool = False,
|
||
) -> Dict[str, float]:
|
||
"""Возвращает сводку: общее время, время по проектам, версиям и активностям."""
|
||
total = 0.0
|
||
by_project: Dict[str, float] = {}
|
||
by_project_version: Dict[str, float] = {}
|
||
by_activity_name: Dict[str, float] = {}
|
||
|
||
for r in rows:
|
||
hours = r.get("hours", 0.0)
|
||
total += hours
|
||
by_project[r["project"]] = by_project.get(r["project"], 0.0) + hours
|
||
key = f"{r['project']}::{r['version']}"
|
||
by_project_version[key] = by_project_version.get(key, 0.0) + hours
|
||
|
||
if by_activity:
|
||
activities = r.get("activities")
|
||
if activities:
|
||
for name, act_hours in activities.items():
|
||
by_activity_name[name] = by_activity_name.get(name, 0.0) + act_hours
|
||
|
||
result: Dict[str, float] = {
|
||
"total": round(total, 2),
|
||
**{f"project:{k}": round(v, 2) for k, v in by_project.items()},
|
||
**{f"version:{k}": round(v, 2) for k, v in by_project_version.items()},
|
||
}
|
||
if by_activity:
|
||
result.update({f"activity:{k}": round(v, 2) for k, v in by_activity_name.items()})
|
||
|
||
return result
|
||
|
||
|
||
def group_rows_by_project_and_version(
|
||
rows: List[ReportRow],
|
||
) -> Dict[str, Dict[str, List[ReportRow]]]:
|
||
"""
|
||
Группирует плоский список строк отчёта в иерархию project → version → [rows].
|
||
|
||
Предполагается, что rows уже отсортирован (build_grouped_report это гарантирует).
|
||
Возвращает обычный dict, сохраняющий порядок вставки (Python 3.7+).
|
||
Используется форматтерами HTML и ODT для объединения ячеек.
|
||
"""
|
||
projects: Dict[str, Dict[str, List[ReportRow]]] = {}
|
||
for r in rows:
|
||
project = r["project"]
|
||
version = r["version"]
|
||
|
||
if project not in projects:
|
||
projects[project] = {}
|
||
if version not in projects[project]:
|
||
projects[project][version] = []
|
||
|
||
projects[project][version].append(r)
|
||
|
||
return projects
|