65 lines
2.1 KiB
Python
65 lines
2.1 KiB
Python
from typing import Any, Dict, List, Optional, Tuple
|
||
from redminelib import Redmine
|
||
from redminelib.resources import Issue
|
||
from .config import Config
|
||
from .utils import get_version
|
||
|
||
REQUESTS_OPTIONS = {"verify": "/etc/ssl/certs/ca-certificates.crt"}
|
||
|
||
|
||
def _get_redmine_auth_kwargs() -> Dict[str, Any]:
|
||
"""Return Redmine auth kwargs. API key has priority over legacy password auth."""
|
||
if Config.REDMINE_API_KEY:
|
||
return {"key": Config.REDMINE_API_KEY}
|
||
return {"username": Config.REDMINE_USER, "password": Config.REDMINE_PASSWORD}
|
||
|
||
|
||
def fetch_issues_with_spent_time(
|
||
from_date: str, to_date: str
|
||
) -> Optional[List[Tuple[Issue, float]]]:
|
||
"""
|
||
Fetch unique issues linked to time entries of the current user in given date range,
|
||
along with total spent hours per issue.
|
||
Returns list of (issue, total_hours) tuples.
|
||
"""
|
||
|
||
redmine = Redmine(
|
||
Config.REDMINE_URL,
|
||
**_get_redmine_auth_kwargs(),
|
||
requests=REQUESTS_OPTIONS,
|
||
)
|
||
|
||
current_user = redmine.user.get("current")
|
||
time_entries = redmine.time_entry.filter(
|
||
user_id=current_user.id, from_date=from_date, to_date=to_date
|
||
)
|
||
|
||
# Агрегируем часы по issue.id
|
||
spent_time: Dict[int, float] = {}
|
||
issue_ids = set()
|
||
for entry in time_entries:
|
||
if hasattr(entry, "issue") and entry.issue and hasattr(entry, "hours"):
|
||
iid = entry.issue.id
|
||
issue_ids.add(iid)
|
||
spent_time[iid] = spent_time.get(iid, 0.0) + float(entry.hours)
|
||
|
||
if not issue_ids:
|
||
return None
|
||
|
||
# Загружаем полные объекты задач
|
||
issue_list_str = ",".join(str(i) for i in issue_ids)
|
||
issues = redmine.issue.filter(
|
||
issue_id=issue_list_str, status_id="*", sort="project:asc"
|
||
)
|
||
|
||
# Сопоставляем задачи с суммарным временем
|
||
result = []
|
||
for issue in issues:
|
||
total_hours = spent_time.get(issue.id, 0.0)
|
||
result.append((issue, total_hours))
|
||
|
||
# Сортируем по (проект, версия)
|
||
result.sort(key=lambda x: (str(x[0].project), get_version(x[0])))
|
||
|
||
return result
|