feat: add --my-tasks mode for open assigned issues (#71)

List open Redmine issues assigned to the current (or overridden) user,
sorted by priority rank (P1..P5) then version due date with a stable id
tie-break. Renders an OSC 8 hyperlinked console table on stdout and a
counts/hours summary line on stderr; version due dates are fetched once
per unique version, unavailable versions degrade to an empty due date.

Also migrates the codebase to ruff 0.17.3 defaults (PEP 585/604
annotations, targeted noqa for intentional blind excepts and CLI root
logging). Behavior-neutral: 408 tests pass with -W error.
This commit is contained in:
Кокос Артем Николаевич
2026-10-28 15:26:41 +07:00
parent 3ce4f408ed
commit 5d1cf4ab74
27 changed files with 1255 additions and 137 deletions

View File

@@ -5,17 +5,18 @@ import re
import sys
from datetime import datetime
from pathlib import Path
from typing import List, Optional
import yaml
from dotenv import find_dotenv, load_dotenv
from . import __version__
from .client import RedmineAPIError, fetch_issues_with_spent_time
from .client import RedmineAPIError, fetch_issues_with_spent_time, fetch_open_issues
from .config import Config
from .formatters.factory import get_console_formatter, get_formatter_by_extension
from .formatters.task_table import TaskTableFormatter
from .mailer import send_report
from .report_builder import build_grouped_report, calculate_summary
from .task_list import build_task_rows, calculate_task_summary, format_summary_line
from .yaml_config import ensure_config_dir, resolve_output_path, save_period_to_config
# Маскируем API-ключ Redmine в тексте ошибок (#55):
@@ -208,10 +209,9 @@ def _save_and_maybe_send(
file=sys.stderr,
)
else:
known_exts = ", ".join([".odt", ".csv", ".md", ".html", ".json", ".xlsx"])
print(
f"❌ Неизвестный формат файла: {output_ext!r}. "
f"Поддерживаются: {known_exts}",
"Поддерживаются: .odt, .csv, .md, .html, .json, .xlsx",
file=sys.stderr,
)
return 1, False
@@ -219,7 +219,7 @@ def _save_and_maybe_send(
try:
formatter.save(rows, output_arg)
print(f"✅ Report saved to {output_arg}")
except Exception as e:
except Exception as e: # noqa: BLE001 -- форматёры (odfpy/openpyxl/csv) бросают произвольные типы, CLI обязан вернуть код 1 на любой сбой
fmt = output_ext.lstrip(".").upper()
print(f"❌ {fmt} export error: {e}", file=sys.stderr)
return 1, False
@@ -264,7 +264,47 @@ def _save_and_maybe_send(
return 0, False
def main(argv: Optional[List[str]] = None) -> int:
def _run_my_tasks(args: argparse.Namespace) -> int:
"""Режим --my-tasks: таблица открытых задач пользователя в stdout (#71).
Саммари печатается в stderr, а не в stdout: при перенаправлении вывода
в файл остаётся только таблица без служебных строк.
"""
try:
items = fetch_open_issues(
user_id=args.user_id or args.user_login or args.user_name
)
except RedmineAPIError as e:
print(f"❌ {_sanitize_error_text(e.message)}", file=sys.stderr)
if (args.verbose or args.debug) and e.original is not None:
# Точка входа CLI: root-логгер осознанно настроен через basicConfig
# в main(), module-логгер здесь дал бы второй набор хендлеров.
logging.exception("Original Redmine API error") # noqa: LOG015
return 1
except Exception as e:
print(f"❌ Unexpected error: {_sanitize_error_text(str(e))}", file=sys.stderr)
if args.verbose or args.debug:
logging.exception("Unexpected error") # noqa: LOG015
return 1
if not items:
print("ℹ️ Открытых задач нет.", file=sys.stderr)
return 0
rows = build_task_rows(items, Config.get_status_translation())
summary = calculate_task_summary(items)
print(format_summary_line(summary), file=sys.stderr)
print(file=sys.stderr)
try:
table = TaskTableFormatter(base_url=Config.get_redmine_url()).format(rows)
except Exception as e: # noqa: BLE001 -- сбой рендера таблицы не должен ронять CLI трейсбеком: печатаем ошибку и возвращаем код 1
print(f"❌ Formatting error: {e}", file=sys.stderr)
return 1
print(table)
return 0
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(
prog="redmine-reporter",
description="Generate Redmine issue report based on your time entries.",
@@ -349,6 +389,11 @@ def main(argv: Optional[List[str]] = None) -> int:
action="store_true",
help="Send generated report via email after saving (requires email section in config); on success advance period start to its end date",
)
parser.add_argument(
"--my-tasks",
action="store_true",
help="List open issues assigned to you, sorted by urgency, and exit",
)
args = parser.parse_args(argv)
# --init-config: обработка до всего остального
@@ -374,6 +419,25 @@ def main(argv: Optional[List[str]] = None) -> int:
return 1
return _run_init_config(args.config_path, args.force)
# --my-tasks: отдельный режим, отчётное окно и файлы к нему неприменимы
if args.my_tasks:
report_flags = [
args.date is not None,
args.output,
args.compact,
args.summary,
args.no_time,
args.by_activity,
args.commit,
args.send,
]
if any(report_flags):
print(
"❌ --my-tasks cannot be used with report-generating flags.",
file=sys.stderr,
)
return 1
# Валидация взаимоисключающих флагов пользователя
user_args = [args.user_id, args.user_login, args.user_name]
if sum(bool(a) for a in user_args) > 1:
@@ -407,6 +471,10 @@ def main(argv: Optional[List[str]] = None) -> int:
print(f"❌ Configuration error: {e}", file=sys.stderr)
return 1
# Ветка --my-tasks: даты отчётного окна режиму не нужны.
if args.my_tasks:
return _run_my_tasks(args)
# Если --date не указан, используем дефолтный диапазон
date_arg = args.date if args.date is not None else Config.get_default_date_range()
try:
@@ -425,12 +493,14 @@ def main(argv: Optional[List[str]] = None) -> int:
except RedmineAPIError as e:
print(f"❌ {_sanitize_error_text(e.message)}", file=sys.stderr)
if (args.verbose or args.debug) and e.original is not None:
logging.exception("Original Redmine API error")
# Точка входа CLI: root-логгер осознанно настроен через basicConfig
# в main(), module-логгер здесь дал бы второй набор хендлеров.
logging.exception("Original Redmine API error") # noqa: LOG015
return 1
except Exception as e:
print(f"❌ Unexpected error: {_sanitize_error_text(str(e))}", file=sys.stderr)
if args.verbose or args.debug:
logging.exception("Unexpected error")
logging.exception("Unexpected error") # noqa: LOG015
return 1
if not issue_hours:
@@ -554,7 +624,7 @@ def main(argv: Optional[List[str]] = None) -> int:
try:
output = formatter.format(rows)
print(output)
except Exception as e:
except Exception as e: # noqa: BLE001 -- консольный форматтер не должен ронять CLI: печатаем ошибку и возвращаем код 1
print(f"❌ Formatting error: {e}", file=sys.stderr)
return 1
@@ -569,7 +639,7 @@ def main(argv: Optional[List[str]] = None) -> int:
try:
save_period_to_config(args.config_path, from_str=next_from)
except Exception as e:
except Exception as e: # noqa: BLE001 -- сбой записи конфига не должен ронять CLI трейсбеком: сообщаем и возвращаем код 1
print(
f"❌ Не удалось сохранить период в конфиг: {e}",
file=sys.stderr,

View File

@@ -1,5 +1,6 @@
import sys
from typing import Any, Dict, List, Optional, Tuple, Union
from datetime import date
from typing import Any
import requests
import truststore
@@ -20,13 +21,13 @@ ISSUE_ID_CHUNK_SIZE = 100
class RedmineAPIError(Exception):
"""Пользовательское исключение с понятным сообщением об ошибке Redmine API."""
def __init__(self, message: str, original: Optional[Exception] = None):
def __init__(self, message: str, original: Exception | None = None):
super().__init__(message)
self.message = message
self.original = original
def _get_redmine_auth_kwargs() -> Dict[str, Any]:
def _get_redmine_auth_kwargs() -> dict[str, Any]:
"""Return Redmine auth kwargs. API key has priority over legacy password auth."""
api_key = Config.get_redmine_api_key()
if api_key:
@@ -122,7 +123,7 @@ def _format_redmine_error(exc: Exception) -> str:
return str(exc)
def _load_time_entry_activities(redmine: Redmine) -> Dict[int, str]:
def _load_time_entry_activities(redmine: Redmine) -> dict[int, str]:
"""Загружает справочник типов активности time entries.
Возвращает словарь id -> name. Если справочник недоступен,
@@ -131,7 +132,7 @@ def _load_time_entry_activities(redmine: Redmine) -> Dict[int, str]:
try:
activities = redmine.enumeration.filter(resource="time_entry_activities")
return {int(a.id): str(a.name) for a in activities}
except Exception as exc:
except Exception as exc: # noqa: BLE001 -- python-redmine бросает разные типы; справочник не критичен, деградируем к данным из entries
# #61: не глотаем сбой молча — предупреждаем, что разбивка по
# активностям будет построена по сырым данным из самих entries.
print(
@@ -142,7 +143,7 @@ def _load_time_entry_activities(redmine: Redmine) -> Dict[int, str]:
return {}
def _get_activity_name(entry, activities: Dict[int, str]) -> str:
def _get_activity_name(entry, activities: dict[int, str]) -> str:
"""Определяет название активности для time entry."""
activity = getattr(entry, "activity", None)
if activity is None:
@@ -162,9 +163,9 @@ def _get_activity_name(entry, activities: Dict[int, str]) -> str:
return str(activity)
def _fetch_issues_chunked(redmine: Redmine, issue_ids: List[int]) -> List[Issue]:
def _fetch_issues_chunked(redmine: Redmine, issue_ids: list[int]) -> list[Issue]:
"""Загружает задачи чанками, чтобы не превышать лимит длины URL (#21)."""
all_issues: List[Issue] = []
all_issues: list[Issue] = []
for i in range(0, len(issue_ids), ISSUE_ID_CHUNK_SIZE):
chunk = issue_ids[i : i + ISSUE_ID_CHUNK_SIZE]
issue_list_str = ",".join(str(x) for x in chunk)
@@ -175,7 +176,7 @@ def _fetch_issues_chunked(redmine: Redmine, issue_ids: List[int]) -> List[Issue]
return all_issues
def _resolve_user_id(redmine: Redmine, user_arg: Union[int, str]) -> int:
def _resolve_user_id(redmine: Redmine, user_arg: int | str) -> int:
"""Преобразует строковый идентификатор пользователя в числовой ID.
Если аргумент — число, возвращает его как есть.
@@ -248,9 +249,9 @@ def _get_current_user_id(redmine: Redmine) -> int:
def fetch_issues_with_spent_time(
from_date: str,
to_date: str,
user_id: Optional[Union[int, str]] = None,
user_id: int | str | None = None,
by_activity: bool = False,
) -> Optional[List[Tuple[Issue, float, Optional[Dict[str, float]]]]]:
) -> list[tuple[Issue, float, dict[str, float] | None]] | None:
"""
Fetch unique issues linked to time entries of the given user in date range,
along with total spent hours per issue.
@@ -281,8 +282,8 @@ def fetch_issues_with_spent_time(
raise RedmineAPIError(_format_redmine_error(exc), original=exc) from exc
# Агрегируем часы по issue.id (и активности, если требуется)
spent_time: Dict[int, float] = {}
spent_by_activity: Dict[int, Dict[str, float]] = {}
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"):
@@ -333,3 +334,66 @@ def fetch_issues_with_spent_time(
result.append((issue, total_hours, activity_breakdown))
return result
def fetch_open_issues(
user_id: int | str | None = None,
) -> list[tuple[Issue, date | None]]:
"""
Fetch open issues assigned to the given user across all projects,
paired with the due date of each issue's fixed version.
If user_id is None, uses current user.
Returns list of (issue, due_date) tuples in API order; sorting is done
in the builder layer. Raises RedmineAPIError on API/auth/network failures.
"""
try:
redmine = _create_redmine()
target_user_id = (
_resolve_user_id(redmine, user_id)
if user_id is not None
else _get_current_user_id(redmine)
)
# Без limit: python-redmine автопагинацией вычитывает все страницы,
# как в time_entry.filter выше.
issues = list(
redmine.issue.filter(assigned_to_id=target_user_id, status_id="open")
)
except RedmineAPIError:
raise
except Exception as exc:
raise RedmineAPIError(_format_redmine_error(exc), original=exc) from exc
# fixed_version внутри задачи содержит только id/name, поэтому due_date
# приходится добирать отдельным запросом на каждую уникальную версию (#71).
# Отказ загрузки одной версии не валидует выгрузку: её задачи получат None.
due_by_version: dict[int, date | None] = {}
for issue in issues:
version_ref = getattr(issue, "fixed_version", None)
if version_ref is None:
continue
version_id = getattr(version_ref, "id", None)
if version_id is None or version_id in due_by_version:
continue
try:
version = redmine.version.get(version_id)
except Exception as exc: # noqa: BLE001 -- отказ загрузки одной версии не валидует выгрузку, деградируем к due_date=None
print(
f"⚠️ Could not load version {version_id}: {exc}. "
"Due date will be empty for its issues.",
file=sys.stderr,
)
due_by_version[version_id] = None
else:
due_by_version[version_id] = getattr(version, "due_date", None)
# Порядок задач сохраняем как в ответе API.
result: list[tuple[Issue, date | None]] = []
for issue in issues:
version_ref = getattr(issue, "fixed_version", None)
version_id = (
getattr(version_ref, "id", None) if version_ref is not None else None
)
due = due_by_version.get(version_id) if version_id is not None else None
result.append((issue, due))
return result

View File

@@ -4,7 +4,6 @@ import sys
from dataclasses import dataclass, field
from datetime import date
from pathlib import Path
from typing import Dict, Union
import yaml
from dotenv import load_dotenv
@@ -13,7 +12,7 @@ from .yaml_config import check_file_permissions, resolve_env_vars
logger = logging.getLogger(__name__)
DEFAULT_REDMINE_VERIFY: Union[bool, str] = True
DEFAULT_REDMINE_VERIFY: bool | str = True
FALSE_VALUES = {"0", "false", "no", "off"}
TRUE_VALUES = {"1", "true", "yes", "on"}
@@ -45,17 +44,17 @@ class AppConfig:
redmine_url: str = ""
redmine_api_key: str = ""
redmine_author: str = ""
redmine_verify: Union[bool, str] = DEFAULT_REDMINE_VERIFY
redmine_verify: bool | str = DEFAULT_REDMINE_VERIFY
period_from: str = ""
output_dir: str = ""
output_filename: str = "{author}_{from}_{to}.{ext}"
output_default_format: str = "xlsx"
report_no_time: bool = False
report_status_translation: Dict[str, str] = field(default_factory=dict)
report_status_translation: dict[str, str] = field(default_factory=dict)
email: EmailConfig = field(default_factory=EmailConfig)
@classmethod
def from_yaml(cls, path: Union[str, Path]) -> "AppConfig":
def from_yaml(cls, path: str | Path) -> "AppConfig":
"""Загружает настройки из YAML-файла.
Если файл не существует — возвращает конфиг со значениями по умолчанию.
@@ -118,7 +117,7 @@ class AppConfig:
return str(value)
@staticmethod
def _resolve_str_dict(raw: dict, section: str, key: str) -> Dict[str, str]:
def _resolve_str_dict(raw: dict, section: str, key: str) -> dict[str, str]:
"""Читает вложенный mapping строк из YAML-секции.
Не-mapping значение логируется с warning и игнорируется.
@@ -140,7 +139,7 @@ class AppConfig:
type(value).__name__,
)
return {}
result: Dict[str, str] = {}
result: dict[str, str] = {}
for k, v in value.items():
if v is None or isinstance(v, (dict, list)):
logger.warning(
@@ -168,7 +167,7 @@ class AppConfig:
return False
@classmethod
def _resolve_verify(cls, raw: dict) -> Union[bool, str]:
def _resolve_verify(cls, raw: dict) -> bool | str:
"""Семантика verify_ssl, единая с REDMINE_VERIFY из env:
- true (bool или строка) → True — стандартная проверка TLS (requests);
@@ -314,7 +313,7 @@ class Config:
return os.getenv("REDMINE_PASSWORD", "").strip()
@classmethod
def get_redmine_verify(cls) -> Union[bool, str]:
def get_redmine_verify(cls) -> bool | str:
value = os.getenv("REDMINE_VERIFY", "").strip()
if value:
normalized = value.lower()
@@ -364,7 +363,7 @@ class Config:
return False
@classmethod
def get_status_translation(cls) -> Dict[str, str]:
def get_status_translation(cls) -> dict[str, str]:
"""Полный словарь перевода статусов: встроенный + переопределения из YAML.
Возвращает новый словарь — модульный STATUS_TRANSLATION не мутируется.
@@ -413,8 +412,7 @@ class Config:
# Будущая start инвертирует диапазон и ломает фильтр Redmine,
# поэтому сжимаем окно до сегодняшнего дня.
if start > today:
start = today
start = min(start, today)
return f"{start.isoformat()}--{today.isoformat()}"
@classmethod

View File

@@ -1,5 +1,5 @@
from abc import ABC, abstractmethod
from typing import Any, List
from typing import Any
from ..types import ReportRow
@@ -16,19 +16,17 @@ class Formatter(ABC):
"""
@abstractmethod
def format(self, rows: List[ReportRow]) -> Any:
def format(self, rows: list[ReportRow]) -> Any:
"""
Форматирует список строк отчета в нужный формат.
Для текстовых форматтеров возвращает str.
Для ODTFormatter возвращает объект OpenDocument.
"""
pass
@abstractmethod
def save(self, rows: List[ReportRow], output_path: str) -> None:
def save(self, rows: list[ReportRow], output_path: str) -> None:
"""
Сохраняет отформатированный отчет в файл по указанному пути.
Для форматтеров, которые не поддерживают сохранение (например, консольные),
бросает NotImplementedError.
"""
pass

View File

@@ -1,5 +1,3 @@
from typing import List
from tabulate import tabulate
from ..types import ReportRow
@@ -9,7 +7,7 @@ from .base import Formatter
class TableFormatter(Formatter):
"""Форматтер для вывода красивой таблицы в консоль."""
def format(self, rows: List[ReportRow]) -> str:
def format(self, rows: list[ReportRow]) -> str:
table_rows = [["Проект", "Версия", "Задача", "Статус", "Затрачено"]]
for r in rows:
time_text = r["time_text"].replace("\n", " / ")
@@ -24,7 +22,7 @@ class TableFormatter(Formatter):
)
return tabulate(table_rows, headers="firstrow", tablefmt="fancy_grid")
def save(self, rows: List[ReportRow], output_path: str) -> None:
def save(self, rows: list[ReportRow], output_path: str) -> None:
# Консольные форматтеры не умеют сохранять в файл напрямую.
# Это делается в CLI.
raise NotImplementedError("TableFormatter не поддерживает сохранение в файл.")
@@ -33,7 +31,7 @@ class TableFormatter(Formatter):
class CompactFormatter(Formatter):
"""Форматтер для компактного вывода в консоль."""
def format(self, rows: List[ReportRow]) -> str:
def format(self, rows: list[ReportRow]) -> str:
lines = []
for r in rows:
time_text = r["time_text"].replace("\n", " / ")
@@ -43,7 +41,7 @@ class CompactFormatter(Formatter):
)
return "\n".join(lines)
def save(self, rows: List[ReportRow], output_path: str) -> None:
def save(self, rows: list[ReportRow], output_path: str) -> None:
# Консольные форматтеры не умеют сохранять в файл напрямую.
# Это делается в CLI.
raise NotImplementedError("CompactFormatter не поддерживает сохранение в файл.")

View File

@@ -1,6 +1,5 @@
import csv
import io
from typing import List
from ..types import ReportRow
from .base import Formatter
@@ -19,7 +18,7 @@ class CSVFormatter(Formatter):
super().__init__()
self.no_time = no_time
def format(self, rows: List[ReportRow]) -> str:
def format(self, rows: list[ReportRow]) -> str:
output = io.StringIO()
writer = csv.writer(output, dialect="excel")
headers = ["Project", "Version", "Issue ID", "Subject", "Status"]
@@ -40,7 +39,7 @@ class CSVFormatter(Formatter):
writer.writerow(data)
return output.getvalue()
def save(self, rows: List[ReportRow], output_path: str) -> None:
def save(self, rows: list[ReportRow], output_path: str) -> None:
content = self.format(rows)
with open(output_path, "w", encoding="utf-8-sig", newline="") as f:
f.write(content)

View File

@@ -1,5 +1,3 @@
from typing import Dict, Optional, Type
from .base import Formatter
from .console import CompactFormatter, TableFormatter
from .csv import CSVFormatter
@@ -11,7 +9,7 @@ from .xlsx import XLSXFormatter
# Словарь для сопоставления расширений файлов с классами форматтеров.
# ODT и XLSX намеренно отсутствуют — их импорт отложен (ленивый), так как
# odfpy/openpyxl может быть не установлен. См. get_formatter_by_extension.
FORMATTER_MAP: Dict[str, Type[Formatter]] = {
FORMATTER_MAP: dict[str, type[Formatter]] = {
".csv": CSVFormatter,
".json": JSONFormatter,
".md": MarkdownFormatter,
@@ -21,13 +19,13 @@ FORMATTER_MAP: Dict[str, Type[Formatter]] = {
# Словарь для сопоставления типа вывода (консоль) с классами форматтеров
CONSOLE_FORMATTER_MAP: Dict[str, Type[Formatter]] = {
CONSOLE_FORMATTER_MAP: dict[str, type[Formatter]] = {
"table": TableFormatter,
"compact": CompactFormatter,
}
def get_formatter_by_extension(extension: str, **kwargs) -> Optional[Formatter]:
def get_formatter_by_extension(extension: str, **kwargs) -> Formatter | None:
"""
Возвращает экземпляр форматтера по расширению файла.
Ключевые аргументы (**kwargs) передаются в конструктор форматтера.
@@ -51,7 +49,7 @@ def get_formatter_by_extension(extension: str, **kwargs) -> Optional[Formatter]:
return None
def get_console_formatter(formatter_type: str) -> Optional[Formatter]:
def get_console_formatter(formatter_type: str) -> Formatter | None:
"""
Возвращает экземпляр консольного форматтера по его типу.
"""

View File

@@ -1,5 +1,4 @@
from html import escape
from typing import List
from ..report_builder import group_rows_by_project_and_version
from ..types import ReportRow
@@ -12,7 +11,7 @@ class HTMLFormatter(Formatter):
def __init__(self, **_kwargs):
super().__init__()
def format(self, rows: List[ReportRow]) -> str:
def format(self, rows: list[ReportRow]) -> str:
projects = group_rows_by_project_and_version(rows)
lines = [
@@ -79,7 +78,7 @@ class HTMLFormatter(Formatter):
lines.append("</html>")
return "\n".join(lines)
def save(self, rows: List[ReportRow], output_path: str) -> None:
def save(self, rows: list[ReportRow], output_path: str) -> None:
content = self.format(rows)
with open(output_path, "w", encoding="utf-8") as f:
f.write(content)

View File

@@ -1,5 +1,4 @@
import json
from typing import List
from ..types import ReportRow
from .base import Formatter
@@ -11,7 +10,7 @@ class JSONFormatter(Formatter):
def __init__(self, **_kwargs):
super().__init__()
def format(self, rows: List[ReportRow]) -> str:
def format(self, rows: list[ReportRow]) -> str:
data = []
for r in rows:
item = {
@@ -28,7 +27,7 @@ class JSONFormatter(Formatter):
data.append(item)
return json.dumps(data, ensure_ascii=False, indent=2)
def save(self, rows: List[ReportRow], output_path: str) -> None:
def save(self, rows: list[ReportRow], output_path: str) -> None:
content = self.format(rows)
with open(output_path, "w", encoding="utf-8") as f:
f.write(content)

View File

@@ -1,5 +1,3 @@
from typing import List
from ..types import ReportRow
from .base import Formatter
@@ -14,7 +12,7 @@ class MarkdownFormatter(Formatter):
def __init__(self, **_kwargs):
super().__init__()
def format(self, rows: List[ReportRow]) -> str:
def format(self, rows: list[ReportRow]) -> str:
lines = [
"| Проект | Версия | Задача | Статус | Затрачено |",
"|--------|--------|--------|--------|-----------|",
@@ -30,7 +28,7 @@ class MarkdownFormatter(Formatter):
)
return "\n".join(lines)
def save(self, rows: List[ReportRow], output_path: str) -> None:
def save(self, rows: list[ReportRow], output_path: str) -> None:
content = self.format(rows)
with open(output_path, "w", encoding="utf-8") as f:
f.write(content)

View File

@@ -1,5 +1,4 @@
from importlib import resources
from typing import List
from odf.opendocument import OpenDocument, load
from odf.style import Style, TableCellProperties, TableColumnProperties
@@ -25,7 +24,7 @@ class ODTFormatter(Formatter):
self.from_date = from_date
self.to_date = to_date
def format(self, rows: List[ReportRow]) -> OpenDocument:
def format(self, rows: list[ReportRow]) -> OpenDocument:
"""
Форматирует данные в объект OpenDocument.
"""
@@ -172,7 +171,7 @@ class ODTFormatter(Formatter):
return doc
def save(self, rows: List[ReportRow], output_path: str) -> None:
def save(self, rows: list[ReportRow], output_path: str) -> None:
"""
Сохраняет сформированный документ в файл.
"""

View File

@@ -0,0 +1,44 @@
from tabulate import tabulate
from ..types import TaskRow
class TaskTableFormatter:
"""Консольный форматтер таблицы открытых задач пользователя (режим --my-tasks).
Не входит в фабрику файловых форматтеров: таблица всегда печатается
в stdout, поэтому сохранение в файл не поддержано.
"""
def __init__(self, base_url: str) -> None:
self.base_url = base_url
def _hyperlink(self, issue_id: int) -> str:
# OSC 8-гиперссылка: номер задачи остаётся видимым терминалом без
# поддержки ссылок, а поддерживающие делают его кликабельным.
url = f"{self.base_url.rstrip('/')}/issues/{issue_id}"
return f"\x1b]8;;{url}\x1b\\{issue_id}\x1b]8;;\x1b\\"
def format(self, rows: list[TaskRow]) -> str:
table_rows = [["ID", "Тема", "Статус", "Приор.", "Версия", "Срок"]]
for row in rows:
due_text = row["due_text"]
if row["overdue"]:
due_text = f"{due_text} (!)"
table_rows.append(
[
self._hyperlink(row["issue_id"]),
row["subject"],
row["status_ru"],
row["priority"],
row["version_name"] or "-",
due_text or "-",
]
)
return tabulate(table_rows, headers="firstrow", tablefmt="simple")
def save(self, rows: list[TaskRow], output_path: str) -> None:
# Консольные форматтеры не умеют сохранять в файл напрямую.
raise NotImplementedError(
"TaskTableFormatter не поддерживает сохранение в файл."
)

View File

@@ -1,5 +1,3 @@
from typing import Dict, List, Tuple
from openpyxl import Workbook
from openpyxl.styles import Alignment, Border, Font, PatternFill, Side
from openpyxl.utils import get_column_letter
@@ -36,7 +34,7 @@ class XLSXFormatter(Formatter):
super().__init__()
self.no_time = no_time
def format(self, rows: List[ReportRow]) -> Workbook:
def format(self, rows: list[ReportRow]) -> Workbook:
wb = Workbook()
ws = wb.active
ws.title = "Report"
@@ -56,9 +54,9 @@ class XLSXFormatter(Formatter):
grouped = group_rows_by_project_and_version(rows)
current_row = 2
project_ranges: List[Tuple[int, int]] = []
version_ranges: List[Tuple[int, int]] = []
project_totals: Dict[str, float] = {}
project_ranges: list[tuple[int, int]] = []
version_ranges: list[tuple[int, int]] = []
project_totals: dict[str, float] = {}
for project, versions in grouped.items():
project_start_row = current_row
@@ -175,11 +173,11 @@ class XLSXFormatter(Formatter):
return wb
def save(self, rows: List[ReportRow], output_path: str) -> None:
def save(self, rows: list[ReportRow], output_path: str) -> None:
wb = self.format(rows)
wb.save(output_path)
def _style_header_row(self, ws: Worksheet, headers: List[str]) -> None:
def _style_header_row(self, ws: Worksheet, headers: list[str]) -> None:
for col_idx, _ in enumerate(headers, start=1):
cell = ws.cell(row=1, column=col_idx)
cell.font = Font(bold=True)
@@ -193,9 +191,7 @@ class XLSXFormatter(Formatter):
for col_idx in range(1, 8):
cell = ws.cell(row=row, column=col_idx)
cell.border = self._BORDER
if col_idx in (1, 2):
cell.alignment = Alignment(vertical="top", wrap_text=True)
elif col_idx == 4:
if col_idx in (1, 2) or col_idx == 4:
cell.alignment = Alignment(vertical="top", wrap_text=True)
else:
cell.alignment = Alignment(vertical="top")
@@ -210,7 +206,7 @@ class XLSXFormatter(Formatter):
def _apply_column_widths(self, ws: Worksheet) -> None:
# Минимальные ширины по умолчанию
widths: Dict[int, float] = {
widths: dict[int, float] = {
1: 18.0,
2: 16.0,
3: 12.0,

View File

@@ -5,7 +5,6 @@ import smtplib
from email.mime.application import MIMEApplication
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from typing import Dict, List
from .client import RedmineAPIError
from .config import EmailConfig
@@ -13,7 +12,7 @@ from .types import ReportRow
SMTP_TIMEOUT = 30
MIME_TYPES: Dict[str, str] = {
MIME_TYPES: dict[str, str] = {
".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
".odt": "application/vnd.oasis.opendocument.text",
".csv": "text/csv",
@@ -43,7 +42,7 @@ def _utf8_text_part(text: str, subtype: str = "plain") -> MIMEText:
return part
def _build_html_body(rows: List[ReportRow]) -> str:
def _build_html_body(rows: list[ReportRow]) -> str:
"""Генерирует HTML-версию тела письма через HTMLFormatter."""
from .formatters.html import HTMLFormatter
@@ -56,7 +55,7 @@ def build_message(
file_path: str,
author: str,
period: str,
rows: List[ReportRow],
rows: list[ReportRow],
) -> MIMEMultipart:
"""Формирует MIME-письмо с подстановками, телами и вложением."""
subject = email_config.subject.replace("{author}", author).replace(
@@ -108,7 +107,7 @@ def send_report(
file_path: str,
author: str,
period: str,
rows: List[ReportRow],
rows: list[ReportRow],
) -> None:
"""Отправляет сгенерированный отчёт по email через SMTP.

View File

@@ -1,4 +1,4 @@
from typing import Dict, List, Optional, Tuple, cast
from typing import cast
from redminelib.resources import Issue
@@ -22,7 +22,7 @@ STATUS_TRANSLATION = {
}
def _format_activities(activities: Dict[str, float]) -> str:
def _format_activities(activities: dict[str, float]) -> str:
"""Форматирует разбивку по активностям в многострочный текст."""
if not activities:
return ""
@@ -32,11 +32,11 @@ def _format_activities(activities: Dict[str, float]) -> str:
def build_grouped_report(
issue_hours: List[Tuple[Issue, float, Optional[Dict[str, float]]]],
issue_hours: list[tuple[Issue, float, dict[str, float] | None]],
fill_time: bool = True,
by_activity: bool = False,
status_translation: Optional[Dict[str, str]] = None,
) -> List[ReportRow]:
status_translation: dict[str, str] | None = None,
) -> list[ReportRow]:
"""
Преобразует список задач с затраченным временем в плоский список строк отчёта,
с учётом группировки по проекту и версии (пустые ячейки для повторяющихся значений).
@@ -59,12 +59,12 @@ def build_grouped_report(
status_translation if status_translation is not None else STATUS_TRANSLATION
)
rows: List[ReportRow] = []
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
activities: dict[str, float] | None = rest[0] if rest else None
project = str(issue.project)
version = get_version(issue)
status_en = str(issue.status)
@@ -108,14 +108,14 @@ def build_grouped_report(
def calculate_summary(
rows: List[ReportRow],
rows: list[ReportRow],
by_activity: bool = False,
) -> Dict[str, float]:
) -> dict[str, float]:
"""Возвращает сводку: общее время, время по проектам, версиям и активностям."""
total = 0.0
by_project: Dict[str, float] = {}
by_project_version: Dict[str, float] = {}
by_activity_name: Dict[str, float] = {}
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)
@@ -130,7 +130,7 @@ def calculate_summary(
for name, act_hours in activities.items():
by_activity_name[name] = by_activity_name.get(name, 0.0) + act_hours
result: Dict[str, float] = {
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()},
@@ -144,8 +144,8 @@ def calculate_summary(
def group_rows_by_project_and_version(
rows: List[ReportRow],
) -> Dict[str, Dict[str, List[ReportRow]]]:
rows: list[ReportRow],
) -> dict[str, dict[str, list[ReportRow]]]:
"""
Группирует плоский список строк отчёта в иерархию project → version → [rows].
@@ -153,7 +153,7 @@ def group_rows_by_project_and_version(
Возвращает обычный dict, сохраняющий порядок вставки (Python 3.7+).
Используется форматтерами HTML и ODT для объединения ячеек.
"""
projects: Dict[str, Dict[str, List[ReportRow]]] = {}
projects: dict[str, dict[str, list[ReportRow]]] = {}
for r in rows:
project = r["project"]
version = r["version"]

View File

@@ -0,0 +1,124 @@
from datetime import date
from redminelib.resources import Issue
from .types import TaskRow, TaskSummary
from .utils import get_version
# В этом Redmine ровно пять приоритетов P1..P5; меньший номер -- срочнее.
PRIORITY_RANK: dict[str, int] = {"P1": 1, "P2": 2, "P3": 3, "P4": 4, "P5": 5}
# Ранг для неизвестного имени приоритета: такие задачи уходят в самый конец.
LOWEST_PRIORITY_RANK = len(PRIORITY_RANK) + 1
def priority_rank(name: str) -> int:
"""Возвращает ранг приоритета; неизвестное имя получает низший ранг."""
return PRIORITY_RANK.get(name, LOWEST_PRIORITY_RANK)
def _issue_version_name(issue: Issue) -> str:
"""Имя версии задачи или пустая строка, если версия не назначена."""
return get_version(issue, default="")
def _sort_key(item: tuple[Issue, date | None]) -> tuple[int, bool, date, int]:
"""Ключ сортировки: приоритет, затем срок по возрастанию, затем id.
Задачи без срока уходят в конец своей приоритетной группы: флаг
(due is None) сортирует False раньше True, а date.max служит нейтральным
значением, чтобы сравнение дат осталось корректным.
"""
issue, due = item
rank = priority_rank(str(issue.priority.name))
return (rank, due is None, due if due is not None else date.max, issue.id)
def build_task_rows(
items: list[tuple[Issue, date | None]],
status_translation: dict[str, str],
today: date | None = None,
) -> list[TaskRow]:
"""Строит отсортированный список строк режима --my-tasks.
items -- пары (задача, срок версии или None). today параметризован
для тестируемости; по умолчанию берётся текущая дата.
"""
if today is None:
today = date.today()
rows: list[TaskRow] = []
for issue, due in sorted(items, key=_sort_key):
status_name = str(issue.status)
rows.append(
{
"issue_id": issue.id,
"subject": str(issue.subject),
"status_ru": status_translation.get(status_name, status_name),
"priority": str(issue.priority.name),
"version_name": _issue_version_name(issue),
"due_text": due.isoformat() if due is not None else "",
"overdue": due is not None and due < today,
}
)
return rows
def calculate_task_summary(
items: list[tuple[Issue, date | None]],
today: date | None = None,
) -> TaskSummary:
"""Считает саммари по списку пар (задача, срок версии или None)."""
if today is None:
today = date.today()
overdue = 0
no_version = 0
no_due = 0
spent_hours = 0.0
estimated_hours = 0.0
estimated_count = 0
for issue, due in items:
if due is not None and due < today:
overdue += 1
version = getattr(issue, "fixed_version", None)
if version is None:
no_version += 1
elif due is None:
no_due += 1
raw_spent = getattr(issue, "spent_hours", None)
if raw_spent is not None:
spent_hours += float(raw_spent)
raw_estimated = getattr(issue, "estimated_hours", None)
if raw_estimated is not None:
estimated_hours += float(raw_estimated)
estimated_count += 1
return {
"total": len(items),
"overdue": overdue,
"no_version": no_version,
"no_due": no_due,
"spent_hours": spent_hours,
"estimated_hours": estimated_hours,
"estimated_count": estimated_count,
}
def format_summary_line(s: TaskSummary) -> str:
"""Форматирует однострочное саммари; блок оценки выводится при наличии данных."""
line = (
f"Открытых: {s['total']} | просроченных: {s['overdue']}"
f" | без версии: {s['no_version']} | без срока: {s['no_due']}"
f" | потрачено: {s['spent_hours']:g}h"
)
if s["estimated_count"] > 0:
line += (
f" | оценка: {s['estimated_hours']:g}h"
f" (по {s['estimated_count']} из {s['total']})"
)
return line

View File

@@ -1,4 +1,4 @@
from typing import Dict, Optional, TypedDict
from typing import TypedDict
class ReportRowBase(TypedDict):
@@ -18,4 +18,28 @@ class ReportRowBase(TypedDict):
class ReportRow(ReportRowBase, total=False):
"""Строка итогового отчёта с опциональной разбивкой по активностям."""
activities: Optional[Dict[str, float]]
activities: dict[str, float] | None
class TaskRow(TypedDict):
"""Строка списка открытых задач пользователя (режим --my-tasks)."""
issue_id: int
subject: str
status_ru: str
priority: str
version_name: str
due_text: str
overdue: bool
class TaskSummary(TypedDict):
"""Саммари списка открытых задач пользователя."""
total: int
overdue: int
no_version: int
no_due: int
spent_hours: float
estimated_hours: float
estimated_count: int

View File

@@ -28,11 +28,11 @@ def get_month_name_from_range(from_date: str, to_date: str) -> str:
return months[end.month]
def get_version(issue) -> str:
"""Возвращает версию задачи или '<N/A>', если не задана."""
def get_version(issue, default: str = "<N/A>") -> str:
"""Возвращает версию задачи или default, если не задана."""
version = getattr(issue, "fixed_version", None)
if version is None:
return "<N/A>"
return default
name = getattr(version, "name", None)
return str(name) if name else str(version)