feat(formatter): unify data pipeline with ReportRow and report_builder

This commit is contained in:
Кокос Артем Николаевич
2026-01-24 16:09:25 +07:00
parent 245ea0a3fa
commit e344715f61
8 changed files with 194 additions and 234 deletions

View File

@@ -5,6 +5,7 @@ from redminelib.resources import Issue
from .config import Config
from .client import fetch_issues_with_spent_time
from .report_builder import build_grouped_report
from .formatter import format_compact, format_table
from .formatter_odt import format_odt
from .formatter_csv import format_csv
@@ -76,6 +77,8 @@ def main(argv: Optional[List[str]] = None) -> int:
print(f"✅ Total issues: {len(issue_hours)} [{args.date}]")
rows = build_grouped_report(issue_hours, fill_time=not args.no_time)
if args.output:
if not (args.output.endswith(".odt") or args.output.endswith(".csv") or args.output.endswith(".md")):
print("❌ Output file must end with .odt, .csv or .md", file=sys.stderr)
@@ -84,21 +87,20 @@ def main(argv: Optional[List[str]] = None) -> int:
try:
if args.output.endswith(".odt"):
doc = format_odt(
issue_hours,
rows,
author=Config.get_author(args.author),
from_date=from_date,
to_date=to_date,
fill_time=not args.no_time
)
doc.save(args.output)
elif args.output.endswith(".csv"):
csv_content = format_csv(issue_hours, fill_time=not args.no_time)
content = format_csv(rows)
with open(args.output, "w", encoding="utf-8", newline="") as f:
f.write(csv_content)
f.write(content)
elif args.output.endswith(".md"):
md_content = format_md(issue_hours, fill_time=not args.no_time)
content = format_md(rows)
with open(args.output, "w", encoding="utf-8") as f:
f.write(md_content)
f.write(content)
print(f"✅ Report saved to {args.output}")
except ImportError as e:
@@ -114,9 +116,9 @@ def main(argv: Optional[List[str]] = None) -> int:
else:
try:
if args.compact:
output = format_compact(issue_hours, fill_time=not args.no_time)
output = format_compact(rows)
else:
output = format_table(issue_hours, fill_time=not args.no_time)
output = format_table(rows)
print(output)
except Exception as e:
print(f"❌ Formatting error: {e}", file=sys.stderr)

View File

@@ -1,85 +1,30 @@
from typing import List, Tuple
from redminelib.resources import Issue
from .utils import get_version
from typing import List
from tabulate import tabulate
from .types import ReportRow
STATUS_TRANSLATION = {
'Closed': 'Закрыто',
'Re-opened': 'В работе',
'New': 'В работе',
'Resolved': 'Решена',
'Pending': 'Ожидание',
'Feedback': 'В работе',
'In Progress': 'В работе',
'Rejected': 'Закрыто',
'Confirming': 'Ожидание',
}
def hours_to_human(hours: float) -> str:
if hours <= 0:
return ""
total_minutes = round(hours * 60)
h = total_minutes // 60
m = total_minutes % 60
parts = []
if h:
parts.append(f"{h}ч")
if m:
parts.append(f"{m}м")
return " ".join(parts) if parts else ""
def format_compact(issue_hours: List[Tuple[Issue, float]], fill_time: bool = True) -> str:
def format_compact(rows: List[ReportRow]) -> str:
lines = []
prev_project = None
prev_version = None
for issue, hours in issue_hours:
project = str(issue.project)
version = get_version(issue)
status = str(issue.status)
time_text = hours_to_human(hours) if fill_time else ""
display_project = project if project != prev_project else ""
display_version = version if (project != prev_project or version != prev_version) else ""
lines.append(f"{display_project} | {display_version} | {issue.id}. {issue.subject} | {status} | {time_text}")
prev_project = project
prev_version = version
for r in rows:
lines.append(
f"{r['display_project']} | {r['display_version']} | "
f"{r['issue_id']}. {r['subject']} | {r['status_ru']} | {r['time_text']}"
)
return "\n".join(lines)
def format_table(issue_hours: List[Tuple[Issue, float]], fill_time: bool = True) -> str:
from tabulate import tabulate
def format_table(rows: List[ReportRow]) -> str:
table_rows = [['Проект', 'Версия', 'Задача', 'Статус', 'Затрачено']]
rows = [['Проект', 'Версия', 'Задача', 'Статус', 'Затрачено']]
prev_project = None
prev_version = None
for issue, hours in issue_hours:
project = str(issue.project)
version = get_version(issue)
status_en = str(issue.status)
status_ru = STATUS_TRANSLATION.get(status_en, status_en)
time_text = hours_to_human(hours) if fill_time else ""
display_project = project if project != prev_project else ""
display_version = version if (project != prev_project or version != prev_version) else ""
rows.append([
display_project,
display_version,
f"{issue.id}. {issue.subject}",
status_ru,
time_text
for r in rows:
table_rows.append([
r['display_project'],
r['display_version'],
f"{r['issue_id']}. {r['subject']}",
r['status_ru'],
r['time_text']
])
prev_project = project
prev_version = version
return tabulate(rows, headers="firstrow", tablefmt="fancy_grid")
return tabulate(table_rows, headers="firstrow", tablefmt="fancy_grid")

View File

@@ -1,40 +1,22 @@
import csv
import io
from typing import List, Tuple
from redminelib.resources import Issue
from .formatter import get_version, hours_to_human, STATUS_TRANSLATION
from typing import List
from .types import ReportRow
def format_csv(
issue_hours: List[Tuple[Issue, float]],
fill_time: bool = True,
dialect: str = "excel"
) -> str:
"""
Formats the list of issues with spent time into CSV.
Returns a string containing the CSV content.
"""
def format_csv(rows: List[ReportRow]) -> str:
output = io.StringIO()
writer = csv.writer(output, dialect=dialect)
# Header
writer = csv.writer(output, dialect="excel")
writer.writerow(["Project", "Version", "Issue ID", "Subject", "Status", "Spent Time"])
for issue, hours in issue_hours:
project = str(issue.project)
version = get_version(issue)
status_en = str(issue.status)
status_ru = STATUS_TRANSLATION.get(status_en, status_en)
time_text = hours_to_human(hours) if fill_time else ""
for r in rows:
writer.writerow([
project,
version,
issue.id,
issue.subject,
status_ru,
time_text
r["project"],
r["version"],
r["issue_id"],
r["subject"],
r["status_ru"],
r["time_text"]
])
return output.getvalue()

View File

@@ -1,35 +1,19 @@
from typing import List, Tuple
from redminelib.resources import Issue
from .formatter import get_version, hours_to_human, STATUS_TRANSLATION
from typing import List
from .types import ReportRow
def format_md(issue_hours: List[Tuple[Issue, float]], fill_time: bool = True) -> str:
"""
Formats the list of issues with spent time into a Markdown table.
Returns a string containing the Markdown content.
"""
def format_md(rows: List[ReportRow]) -> str:
lines = [
"| Проект | Версия | Задача | Статус | Затрачено |",
"|--------|--------|--------|--------|-----------|"
]
lines = []
lines.append("| Проект | Версия | Задача | Статус | Затрачено |")
lines.append("|--------|--------|--------|--------|-----------|")
for r in rows:
task_cell = f"{r['issue_id']}. {r['subject']}"
prev_project = None
prev_version = None
for issue, hours in issue_hours:
project = str(issue.project)
version = get_version(issue)
status_en = str(issue.status)
status_ru = STATUS_TRANSLATION.get(status_en, status_en)
time_text = hours_to_human(hours) if fill_time else ""
display_project = project if project != prev_project else ""
display_version = version if (project != prev_project or version != prev_version) else ""
task_cell = f"{issue.id}. {issue.subject}"
lines.append(f"| {display_project} | {display_version} | {task_cell} | {status_ru} | {time_text} |")
prev_project = project
prev_version = version
lines.append(
f"| {r['display_project']} | {r['display_version']} "
f"| {task_cell} | {r['status_ru']} | {r['time_text']} |"
)
return "\n".join(lines)

View File

@@ -1,21 +1,18 @@
import os
from typing import List, Tuple
from redminelib.resources import Issue
from typing import List
from odf.opendocument import load
from odf.text import P
from odf.table import Table, TableColumn, TableRow, TableCell
from odf.style import Style, TableColumnProperties, TableCellProperties
from .formatter import get_version, hours_to_human, STATUS_TRANSLATION
from .types import ReportRow
from .utils import get_month_name_from_range
def format_odt(
issue_hours: List[Tuple[Issue, float]],
rows: List[ReportRow],
author: str = "",
from_date: str = "",
to_date: str = "",
fill_time: bool = True
) -> "OpenDocument":
template_path = "template.odt"
if not os.path.exists(template_path):
@@ -27,85 +24,61 @@ def format_odt(
# Заголовок
month_name = get_month_name_from_range(from_date, to_date)
header_text = f"{author}. Отчет за месяц {month_name}."
header_paragraph = P(stylename=para_style_name, text=header_text)
doc.text.addElement(header_paragraph)
doc.text.addElement(P(stylename=para_style_name, text=header_text))
doc.text.addElement(P(stylename=para_style_name, text=""))
# Добавляем пустую строку (новый параграф без текста)
empty_paragraph = P(stylename=para_style_name, text="")
doc.text.addElement(empty_paragraph)
# Группировка: project - version - [(issue, hours, status_ru)]
projects = {}
for issue, hours in issue_hours:
project = str(issue.project)
version = get_version(issue)
status_en = str(issue.status)
status_ru = STATUS_TRANSLATION.get(status_en, status_en)
if project not in projects:
projects[project] = {}
if version not in projects[project]:
projects[project][version] = []
projects[project][version].append((issue, hours, status_ru))
# Создаем стиль для ячеек таблицы
# Стиль ячеек
cell_style_name = "TableCellStyle"
cell_style = Style(name=cell_style_name, family="table-cell")
# Устанавливаем отступы (Padding)
cell_props = TableCellProperties(
padding="0.04in",
border="0.05pt solid #000000"
)
cell_props = TableCellProperties(padding="0.04in", border="0.05pt solid #000000")
cell_style.addElement(cell_props)
doc.automaticstyles.addElement(cell_style)
# Создаем стиль для всей таблицы (опционально, но может понадобиться)
table_style_name = "ReportTableStyle"
table_style = Style(name=table_style_name, family="table")
# Создаем таблицу и применяем стиль
table = Table(name="Report", stylename=table_style_name)
# Добавляем стили для каждой колонки (ширины)
# Таблица
table = Table(name="Report")
column_widths = ["1.56in", "1.63in", "3.93in", "1.56in", "1.43in"]
for i, width in enumerate(column_widths):
col_style_name = f"col{i+1}"
col_style = Style(name=col_style_name, family="table-column")
col_props = TableColumnProperties(columnwidth=width, breakbefore="auto")
for width in column_widths:
col_style = Style(name=f"col_{width}", family="table-column")
col_props = TableColumnProperties(columnwidth=width)
col_style.addElement(col_props)
doc.automaticstyles.addElement(col_style)
table.addElement(TableColumn(stylename=col_style))
# Заголовки
header_row = TableRow()
headers = ["Наименование Проекта", "Номер версии*", "Задача", "Статус Готовность*", "Затрачено за отчетный период"]
for text in headers:
for text in ["Наименование Проекта", "Номер версии*", "Задача", "Статус Готовность*", "Затрачено за отчетный период"]:
cell = TableCell(stylename=cell_style_name)
p = P(stylename=para_style_name, text=text)
cell.addElement(p)
cell.addElement(P(stylename=para_style_name, text=text))
header_row.addElement(cell)
table.addElement(header_row)
projects = {}
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)
# Данные с двухуровневой группировкой и объединением ячеек
for project, versions in projects.items():
total_project_rows = sum(len(rows) for rows in versions.values())
total_project_rows = sum(len(rows_for_version) for rows_for_version in versions.values())
first_version_in_project = True
for version, rows in versions.items():
row_span_version = len(rows)
for version, rows_for_version in versions.items():
row_span_version = len(rows_for_version)
first_row_in_version = True
for issue, hours, status_ru in rows:
for r in rows_for_version:
row = TableRow()
# Ячейка "Проект" - только в первой строке всего проекта
if first_version_in_project and first_row_in_version:
cell_project = TableCell(stylename=cell_style_name)
cell_project.setAttribute("numberrowsspanned", str(total_project_rows))
p = P(stylename=para_style_name, text=project)
p = P(stylename=para_style_name, text=project) # Полное название проекта
cell_project.addElement(p)
row.addElement(cell_project)
@@ -123,21 +96,18 @@ def format_odt(
# Остальные колонки
task_cell = TableCell(stylename=cell_style_name)
p = P(stylename=para_style_name, text=f"{issue.id}. {issue.subject}")
task_text = f"{r['issue_id']}. {r['subject']}"
p = P(stylename=para_style_name, text=task_text)
task_cell.addElement(p)
row.addElement(task_cell)
status_cell = TableCell(stylename=cell_style_name)
p = P(stylename=para_style_name, text=status_ru)
p = P(stylename=para_style_name, text=r["status_ru"])
status_cell.addElement(p)
row.addElement(status_cell)
time_cell = TableCell(stylename=cell_style_name)
if fill_time:
time_text = hours_to_human(hours)
else:
time_text = ""
p = P(stylename=para_style_name, text=time_text)
p = P(stylename=para_style_name, text=r["time_text"])
time_cell.addElement(p)
row.addElement(time_cell)
@@ -145,34 +115,17 @@ def format_odt(
first_version_in_project = False
doc.text.addElement(table)
# Добавляем пустую строку (новый параграф без текста)
doc.text.addElement(P(stylename=para_style_name, text=""))
# Справка
doc.text.addElement(P(
stylename=para_style_name,
text="“Наименование Проекта” - Имя собственное устройства или программного обеспечения."
))
doc.text.addElement(P(
stylename=para_style_name,
text="“Номер версии” - Версия в проекте. Опциональное поле."
))
doc.text.addElement(P(
stylename=para_style_name,
text="“Задача” - Номер по Redmine и формулировка."
))
doc.text.addElement(P(
stylename=para_style_name,
text="“Статус” - Актуальное состояние задачи на момент отчета. Статусы: закрыто, в работе, ожидание, решена."
))
doc.text.addElement(P(
stylename=para_style_name,
text="“Готовность” Опциональное поле в процентах."
))
doc.text.addElement(P(
stylename=para_style_name,
text="“Затрачено за отчетный период” - в днях или часах."
))
for line in [
"«Наименование Проекта» - Имя собственное устройства или программного обеспечения.",
"«Номер версии» - Версия в проекте. Опциональное поле.",
"«Задача» - Номер по Redmine и формулировка.",
"«Статус» - Актуальное состояние задачи на момент отчета. Статусы: закрыто, в работе, ожидание, решена.",
"«Готовность» Опциональное поле в процентах.",
"«Затрачено за отчетный период» - в днях или часах."
]:
doc.text.addElement(P(stylename=para_style_name, text=line))
return doc

View File

@@ -0,0 +1,62 @@
from typing import List, Tuple, cast
from redminelib.resources import Issue
from .types import ReportRow
from .utils import get_version, hours_to_human
STATUS_TRANSLATION = {
'Closed': 'Закрыто',
'Re-opened': 'В работе',
'New': 'В работе',
'Resolved': 'Решена',
'Pending': 'Ожидание',
'Feedback': 'В работе',
'In Progress': 'В работе',
'Rejected': 'Закрыто',
'Confirming': 'Ожидание',
}
def build_grouped_report(
issue_hours: List[Tuple[Issue, float]],
fill_time: bool = True,
) -> List[ReportRow]:
"""
Преобразует список задач с затраченным временем в плоский список строк отчёта,
с учётом группировки по проекту и версии (пустые ячейки для повторяющихся значений).
"""
rows: List[ReportRow] = []
prev_project: str = ""
prev_version: str = ""
for issue, hours in issue_hours:
project = str(issue.project)
version = get_version(issue)
status_en = str(issue.status)
status_ru = STATUS_TRANSLATION.get(status_en, status_en)
time_text = hours_to_human(hours) if fill_time else ""
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,
},
)
)
prev_project = project
prev_version = version
return rows

14
redmine_reporter/types.py Normal file
View File

@@ -0,0 +1,14 @@
from typing import TypedDict
class ReportRow(TypedDict):
"""Строка итогового отчёта."""
project: str
version: str
display_project: str
display_version: str
issue_id: int
subject: str
status_ru: str
time_text: str

View File

@@ -2,22 +2,40 @@ from datetime import datetime
def get_month_name_from_range(from_date: str, to_date: str) -> str:
"""Определяет название месяца по диапазону дат.
- Если from == to - возвращает месяц этой даты.
- Если диапазон охватывает несколько месяцев - возвращает месяц из to_date.
"""
"""Определяет название месяца по диапазону дат"""
try:
end = datetime.strptime(to_date, "%Y-%m-%d")
except ValueError:
return "Январь" # fallback, хотя лучше бы не срабатывало
return "Январь"
months = [
"", "Январь", "Февраль", "Март", "Апрель", "Май", "Июнь",
"Июль", "Август", "Сентябрь", "Октябрь", "Ноябрь", "Декабрь"
]
return months[end.month]
def get_version(issue) -> str:
"""Возвращает версию задачи или '<N/A>', если не задана."""
return str(getattr(issue, 'fixed_version', '<N/A>'))
def hours_to_human(hours: float) -> str:
"""Преобразует часы в человекочитаемый формат: '2ч 30м'."""
if hours <= 0:
return ""
total_minutes = round(hours * 60)
h = total_minutes // 60
m = total_minutes % 60
parts = []
if h:
parts.append(f"{h}ч")
if m:
parts.append(f"{m}м")
return " ".join(parts) if parts else ""