feat: output path defaults (#43) + datetime precision & dedup (#47)

#43 — Имя файла и пути по умолчанию:
- resolve_output_path() в yaml_config.py: резолвит --output с учётом
  output.dir, filename_template, default_format из YAML-конфига
- Bare format (xlsx/odt/...) → путь по шаблону
- Без расширения → автодописывается default_format (.xlsx)
- Config.get_output_dir/filename/default_format()

#47 — datetime precision и дедупликация:
- period.last_used.from/to в YAML-конфиге и AppConfig
- period.precision: date|datetime
- _compute_dedup_cutoff() в cli.py: при precision=datetime вычисляет
  cutoff из period.last_used.to
- _parse_datetime() + AND-логика дедупликации в client.py:
  запись исключается если created_on И updated_on < cutoff
- Пропуск issues без часов после дедупликации

Closes #43
Closes #47
This commit is contained in:
Кокос Артем Николаевич
2026-07-07 10:16:46 +07:00
parent 67350bfcd6
commit 47152f8f04
9 changed files with 718 additions and 15 deletions

View File

@@ -1,3 +1,4 @@
from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple, Union
import requests
@@ -144,6 +145,25 @@ def _get_activity_name(entry, activities: Dict[int, str]) -> str:
return str(activity)
def _parse_datetime(value: Any) -> Optional[datetime]:
"""Parse a datetime from Redmine API response.
Accepts datetime objects, ISO strings (with or without timezone),
or None. Returns a timezone-aware datetime or None.
"""
if value is None:
return None
if isinstance(value, datetime):
return value
if isinstance(value, str):
try:
dt = datetime.fromisoformat(value.replace("Z", "+00:00"))
return dt
except (ValueError, TypeError):
return None
return None
def _fetch_issues_chunked(redmine: Redmine, issue_ids: List[int]) -> List[Issue]:
"""Загружает задачи чанками, чтобы не превышать лимит длины URL (#21)."""
all_issues: List[Issue] = []
@@ -220,12 +240,15 @@ def fetch_issues_with_spent_time(
to_date: str,
user_id: Optional[Union[int, str]] = None,
by_activity: bool = False,
dedup_before: Optional[datetime] = None,
) -> 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.
If by_activity is True, returns per-activity breakdown as third tuple element.
If dedup_before is set, filters out time entries whose created_on AND updated_on
are both before dedup_before (AND logic: both must be < cutoff to exclude).
Returns list of (issue, total_hours, activities) tuples.
Raises RedmineAPIError on API/auth/network failures.
"""
@@ -238,14 +261,34 @@ def fetch_issues_with_spent_time(
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
time_entries = list(
redmine.time_entry.filter(user_id=target_user_id, from_date=from_date, to_date=to_date)
)
except RedmineAPIError:
raise
except Exception as exc:
raise RedmineAPIError(_format_redmine_error(exc), original=exc) from exc
# Дедупликация: отсекаем записи, которые были учтены в предыдущем отчёте.
# Запись исключается, если BOTH created_on AND updated_on < dedup_before.
# Записи без метаданных (created_on/updated_on == None) не фильтруются.
if dedup_before is not None:
filtered: list = []
for entry in time_entries:
created = _parse_datetime(getattr(entry, "created_on", None))
updated = _parse_datetime(getattr(entry, "updated_on", None))
if created is None and updated is None:
filtered.append(entry)
elif created is not None and updated is not None:
if created >= dedup_before and updated >= dedup_before:
filtered.append(entry)
elif created is not None and created >= dedup_before:
filtered.append(entry)
elif updated is not None and updated >= dedup_before:
filtered.append(entry)
time_entries = filtered
# Агрегируем часы по issue.id (и активности, если требуется)
spent_time: Dict[int, float] = {}
spent_by_activity: Dict[int, Dict[str, float]] = {}
@@ -278,7 +321,9 @@ def fetch_issues_with_spent_time(
result = []
for issue in issues:
iid = issue.id
total_hours = spent_time.get(iid, 0.0)
if iid not in spent_time:
continue
total_hours = spent_time[iid]
activity_breakdown = spent_by_activity.get(iid) if by_activity else None
result.append((issue, total_hours, activity_breakdown))