Files
test/jira_analyze.py
2026-06-16 00:03:34 +03:00

1281 lines
49 KiB
Python
Executable File

#!/usr/bin/env python3
"""Analyze Jira issues for a project.
The Jira token is intentionally read from JIRA_TOKEN instead of being stored in
this file. Example:
JIRA_TOKEN='...' ./jira_analyze.py --project VDEV
"""
from __future__ import annotations
import argparse
import base64
import csv
import datetime as dt
import html
import json
import os
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from collections import Counter, defaultdict
from typing import Any
DEFAULT_BASE_URL = "https://jira.a1.by"
DEFAULT_PROJECT = "VDEV"
DEFAULT_PAGE_SIZE = 100
def jql_string(value: str) -> str:
escaped = value.replace('\\', '\\\\').replace('"', '\\"')
return f'"{escaped}"'
def jql_project(value: str) -> str:
if value.replace("_", "").isalnum() and " " not in value:
return value
return jql_string(value)
def load_dotenv(path: str) -> None:
if not os.path.exists(path):
return
with open(path, encoding="utf-8") as file:
for line in file:
line = line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, value = line.split("=", 1)
key = key.strip()
value = value.strip().strip('"').strip("'")
if key and key not in os.environ:
os.environ[key] = value
def parse_args() -> argparse.Namespace:
script_dir = os.path.dirname(os.path.abspath(__file__))
load_dotenv(os.path.join(script_dir, ".env"))
load_dotenv(os.path.join(os.getcwd(), ".env"))
parser = argparse.ArgumentParser(
description="Analyze Jira tickets for a project via Jira REST API."
)
parser.add_argument(
"--base-url",
default=os.environ.get("JIRA_BASE_URL", DEFAULT_BASE_URL),
help=f"Jira base URL. Default: {DEFAULT_BASE_URL}",
)
parser.add_argument(
"--project",
default=os.environ.get("JIRA_PROJECT", DEFAULT_PROJECT),
help=f"Jira project key. Default: {DEFAULT_PROJECT}",
)
parser.add_argument(
"--token",
default=os.environ.get("JIRA_TOKEN"),
help="Jira token. Prefer JIRA_TOKEN environment variable.",
)
parser.add_argument(
"--user",
default=os.environ.get("JIRA_USER"),
help="Jira user name for Basic Auth. Prefer JIRA_USER environment variable.",
)
parser.add_argument(
"--jql",
help="Custom JQL. If set, --project is ignored for the search query.",
)
parser.add_argument(
"--page-size",
type=int,
default=DEFAULT_PAGE_SIZE,
help=f"Jira page size. Default: {DEFAULT_PAGE_SIZE}",
)
parser.add_argument(
"--stale-days",
type=int,
default=30,
help="Consider unresolved issues stale after this many days without updates.",
)
parser.add_argument(
"--json-out",
help="Write raw fetched issues to this JSON file.",
)
parser.add_argument(
"--csv-out",
help="Write a flat issue list to this CSV file.",
)
parser.add_argument(
"--html-quarter-out",
help="Write Q1/Q2 closed issue analytics to this HTML file.",
)
parser.add_argument(
"--html-sprint-quarter-out",
help="Write sprint-based quarterly statistics to this HTML file.",
)
parser.add_argument(
"--sprints-from",
default=f"{dt.date.today().year}-01-01",
help="Start date for sprint report. Default: Jan 1 of current year.",
)
parser.add_argument(
"--sprints-to",
default=dt.date.today().isoformat(),
help="End date for sprint report. Default: today.",
)
parser.add_argument(
"--board-id",
type=int,
help="Limit sprint report to one Jira Agile board ID.",
)
parser.add_argument(
"--report-year",
type=int,
default=2026,
help="Year for --html-quarter-out. Default: 2026",
)
parser.add_argument(
"--quarters",
default="1,2",
help="Comma-separated quarters for --html-quarter-out. Default: 1,2",
)
parser.add_argument(
"--closed-status",
default="Закрыт",
help='Closed Jira status name for report filtering. Default: "Закрыт"',
)
parser.add_argument(
"--list-projects",
action="store_true",
help="List Jira projects visible to the token and exit.",
)
parser.add_argument(
"--project-search",
help="Filter --list-projects by project key or name substring.",
)
parser.add_argument(
"--top",
type=int,
default=15,
help="Number of rows to show in top lists. Default: 15",
)
return parser.parse_args()
def auth_header(token: str, user: str | None) -> str:
if user:
raw = f"{user}:{token}".encode("utf-8")
return "Basic " + base64.b64encode(raw).decode("ascii")
return f"Bearer {token}"
def jira_get(base_url: str, token: str, path: str, params: dict[str, Any], user: str | None = None) -> Any:
query = urllib.parse.urlencode(params, doseq=True)
url = f"{base_url.rstrip('/')}{path}?{query}"
request = urllib.request.Request(
url,
headers={
"Accept": "application/json",
"Authorization": auth_header(token, user),
},
)
for attempt in range(4):
try:
with urllib.request.urlopen(request, timeout=60) as response:
return json.loads(response.read().decode("utf-8"))
except urllib.error.HTTPError as exc:
body = exc.read().decode("utf-8", errors="replace")
if exc.code == 429 and attempt < 3:
retry_after = exc.headers.get("Retry-After")
delay = int(retry_after) if retry_after and retry_after.isdigit() else 30 * (attempt + 1)
print(f"Rate limit exceeded. Retrying in {delay} seconds...", file=sys.stderr)
time.sleep(delay)
continue
raise RuntimeError(f"Jira API returned HTTP {exc.code}: {body}") from exc
except urllib.error.URLError as exc:
raise RuntimeError(f"Failed to connect to Jira: {exc}") from exc
raise RuntimeError("Jira request failed after retries")
def fetch_issues(
base_url: str,
token: str,
user: str | None,
jql: str,
page_size: int,
) -> list[dict[str, Any]]:
fields = [
"summary",
"issuetype",
"status",
"priority",
"assignee",
"reporter",
"created",
"updated",
"resolutiondate",
"components",
"labels",
"fixVersions",
"versions",
"duedate",
"timespent",
"timeoriginalestimate",
"aggregatetimespent",
"aggregatetimeoriginalestimate",
]
issues: list[dict[str, Any]] = []
start_at = 0
while True:
data = jira_get(
base_url,
token,
"/rest/api/2/search",
{
"jql": jql,
"startAt": start_at,
"maxResults": page_size,
"fields": ",".join(fields),
},
user,
)
batch = data.get("issues", [])
issues.extend(batch)
total = int(data.get("total", len(issues)))
print(f"Fetched {len(issues)}/{total} issues...", file=sys.stderr)
if not batch or len(issues) >= total:
break
start_at += len(batch)
return issues
def field_name(issue: dict[str, Any], field: str, default: str = "Unspecified") -> str:
value = issue.get("fields", {}).get(field)
if isinstance(value, dict):
return value.get("displayName") or value.get("name") or default
if value:
return str(value)
return default
def parse_jira_datetime(value: str | None) -> dt.datetime | None:
if not value:
return None
normalized = value
if normalized.endswith("Z"):
normalized = normalized[:-1] + "+00:00"
if len(value) >= 5 and value[-5] in {"+", "-"} and value[-3] != ":":
normalized = f"{value[:-2]}:{value[-2:]}"
try:
return dt.datetime.fromisoformat(normalized)
except ValueError:
return None
def month_bucket(value: str | None) -> str:
parsed = parse_jira_datetime(value)
if not parsed:
return "Unknown"
return parsed.strftime("%Y-%m")
def is_unresolved(issue: dict[str, Any]) -> bool:
return not issue.get("fields", {}).get("resolutiondate")
def count_list_field(issues: list[dict[str, Any]], field: str) -> Counter[str]:
counter: Counter[str] = Counter()
for issue in issues:
values = issue.get("fields", {}).get(field) or []
if not values:
counter["None"] += 1
continue
for value in values:
if isinstance(value, dict):
counter[value.get("name") or "Unknown"] += 1
else:
counter[str(value)] += 1
return counter
def print_counter(title: str, counter: Counter[str], top: int) -> None:
print(f"\n{title}")
print("-" * len(title))
if not counter:
print("No data")
return
width = max(len(name) for name, _ in counter.most_common(top))
for name, count in counter.most_common(top):
print(f"{name:<{width}} {count}")
def parse_quarters(value: str) -> list[int]:
quarters: list[int] = []
for raw in value.split(","):
raw = raw.strip()
if not raw:
continue
quarter = int(raw)
if quarter < 1 or quarter > 4:
raise ValueError("quarters must be between 1 and 4")
quarters.append(quarter)
return sorted(set(quarters))
def quarter_bounds(year: int, quarter: int) -> tuple[dt.date, dt.date]:
start_month = (quarter - 1) * 3 + 1
start = dt.date(year, start_month, 1)
if quarter == 4:
end = dt.date(year + 1, 1, 1)
else:
end = dt.date(year, start_month + 3, 1)
return start, end
def quarter_for_date(value: dt.date) -> int:
return ((value.month - 1) // 3) + 1
def jira_date(value: str | None) -> dt.date | None:
parsed = parse_jira_datetime(value)
if parsed:
return parsed.date()
if value:
try:
return dt.date.fromisoformat(value)
except ValueError:
return None
return None
def parse_date_arg(value: str, name: str) -> dt.date:
try:
return dt.date.fromisoformat(value)
except ValueError as exc:
raise ValueError(f"{name} must be YYYY-MM-DD") from exc
def seconds_value(issue: dict[str, Any], *fields: str) -> int:
issue_fields = issue.get("fields", {})
for field in fields:
value = issue_fields.get(field)
if isinstance(value, int) and value > 0:
return value
return 0
def hours(seconds: int) -> float:
return round(seconds / 3600, 1)
def issue_components(issue: dict[str, Any]) -> list[str]:
components = issue.get("fields", {}).get("components") or []
names = [component.get("name") for component in components if isinstance(component, dict)]
return [name for name in names if name]
def team_name(issue: dict[str, Any]) -> str:
components = issue_components(issue)
if components:
return ", ".join(components)
return field_name(issue, "assignee", "No team/component")
def issue_cycle_days(issue: dict[str, Any]) -> int | None:
fields = issue.get("fields", {})
created = jira_date(fields.get("created"))
resolved = jira_date(fields.get("resolutiondate"))
if not created or not resolved:
return None
return max((resolved - created).days, 0)
def html_escape(value: Any) -> str:
return html.escape(str(value if value is not None else ""))
def html_table(headers: list[str], rows: list[list[Any]], empty: str = "Нет данных") -> str:
if not rows:
return f"<p class=\"empty\">{html_escape(empty)}</p>"
header_html = "".join(f"<th>{html_escape(header)}</th>" for header in headers)
row_html = []
for row in rows:
row_html.append("<tr>" + "".join(f"<td>{html_escape(cell)}</td>" for cell in row) + "</tr>")
return f"<table><thead><tr>{header_html}</tr></thead><tbody>{''.join(row_html)}</tbody></table>"
def closed_quarter_issues(
issues: list[dict[str, Any]],
year: int,
quarters: list[int],
closed_status: str,
) -> list[dict[str, Any]]:
selected: list[dict[str, Any]] = []
closed_status_lower = closed_status.casefold()
quarter_set = set(quarters)
for issue in issues:
fields = issue.get("fields", {})
resolved = jira_date(fields.get("resolutiondate"))
if not resolved or resolved.year != year or quarter_for_date(resolved) not in quarter_set:
continue
if field_name(issue, "status").casefold() != closed_status_lower:
continue
selected.append(issue)
return selected
def issue_problem_flags(issue: dict[str, Any]) -> list[str]:
fields = issue.get("fields", {})
flags: list[str] = []
spent = seconds_value(issue, "aggregatetimespent", "timespent")
estimate = seconds_value(issue, "aggregatetimeoriginalestimate", "timeoriginalestimate")
cycle = issue_cycle_days(issue)
due = jira_date(fields.get("duedate"))
resolved = jira_date(fields.get("resolutiondate"))
labels = [str(label).casefold() for label in fields.get("labels") or []]
if spent == 0:
flags.append("нет списанных часов")
if estimate == 0:
flags.append("нет оценки")
if due and resolved and resolved > due:
flags.append("закрыто после срока")
if cycle is not None and cycle > 30:
flags.append("длинный цикл > 30 дней")
high_priorities = {"highest", "high", "blocker", "critical", "критический", "высокий"}
if field_name(issue, "priority").casefold() in high_priorities:
flags.append("высокий приоритет")
if any("block" in label or "блок" in label for label in labels):
flags.append("есть блокирующая метка")
if not issue_components(issue):
flags.append("нет компонента/команды")
return flags
def quarter_summary_rows(issues: list[dict[str, Any]], year: int, quarters: list[int]) -> list[list[Any]]:
rows: list[list[Any]] = []
for quarter in quarters:
start, end = quarter_bounds(year, quarter)
quarter_issues = []
for issue in issues:
resolved = jira_date(issue.get("fields", {}).get("resolutiondate"))
if resolved and start <= resolved < end:
quarter_issues.append(issue)
spent = sum(seconds_value(issue, "aggregatetimespent", "timespent") for issue in quarter_issues)
estimate = sum(
seconds_value(issue, "aggregatetimeoriginalestimate", "timeoriginalestimate")
for issue in quarter_issues
)
cycle_values = [value for issue in quarter_issues if (value := issue_cycle_days(issue)) is not None]
problem_count = sum(1 for issue in quarter_issues if issue_problem_flags(issue))
rows.append(
[
f"Q{quarter} {year}",
len(quarter_issues),
hours(spent),
hours(estimate),
hours(spent - estimate),
round(sum(cycle_values) / len(cycle_values), 1) if cycle_values else 0,
problem_count,
]
)
return rows
def team_summary_rows(issues: list[dict[str, Any]]) -> list[list[Any]]:
teams: dict[str, dict[str, Any]] = defaultdict(
lambda: {"issues": 0, "spent": 0, "estimate": 0, "problems": Counter(), "cycle": []}
)
for issue in issues:
name = team_name(issue)
bucket = teams[name]
bucket["issues"] += 1
bucket["spent"] += seconds_value(issue, "aggregatetimespent", "timespent")
bucket["estimate"] += seconds_value(issue, "aggregatetimeoriginalestimate", "timeoriginalestimate")
cycle = issue_cycle_days(issue)
if cycle is not None:
bucket["cycle"].append(cycle)
bucket["problems"].update(issue_problem_flags(issue))
rows: list[list[Any]] = []
for name, data in sorted(teams.items(), key=lambda item: item[1]["issues"], reverse=True):
cycle_values = data["cycle"]
top_problems = ", ".join(problem for problem, _ in data["problems"].most_common(3))
rows.append(
[
name,
data["issues"],
hours(data["spent"]),
hours(data["estimate"]),
hours(data["spent"] - data["estimate"]),
round(sum(cycle_values) / len(cycle_values), 1) if cycle_values else 0,
top_problems or "нет явных сигналов",
]
)
return rows
def recommendations(issues: list[dict[str, Any]]) -> list[str]:
flags = Counter(flag for issue in issues for flag in issue_problem_flags(issue))
result: list[str] = []
if flags["нет списанных часов"]:
result.append("Закрепить обязательное списание времени перед закрытием задач; без этого отчет по трудозатратам неполный.")
if flags["нет оценки"]:
result.append("Ввести контроль оценки перед стартом работы, чтобы сравнение план/факт по командам было корректным.")
if flags["закрыто после срока"]:
result.append("Для задач с due date добавить раннюю эскалацию риска за 3-5 рабочих дней до срока.")
if flags["длинный цикл > 30 дней"]:
result.append("Разбивать крупные задачи на меньшие deliverables и отдельно отслеживать зависшие задачи старше 30 дней.")
if flags["нет компонента/команды"]:
result.append("Заполнить components или единое поле команды, иначе корректировки по командам будут неточными.")
if not result:
result.append("Критичных системных проблем по закрытым задачам не найдено; продолжить мониторинг план/факт часов и цикла выполнения.")
return result
def write_quarter_html_report(
path: str,
issues: list[dict[str, Any]],
project: str,
year: int,
quarters: list[int],
closed_status: str,
) -> None:
report_issues = closed_quarter_issues(issues, year, quarters, closed_status)
total_spent = sum(seconds_value(issue, "aggregatetimespent", "timespent") for issue in report_issues)
total_estimate = sum(
seconds_value(issue, "aggregatetimeoriginalestimate", "timeoriginalestimate")
for issue in report_issues
)
problem_rows = []
for issue in report_issues:
flags = issue_problem_flags(issue)
if not flags:
continue
fields = issue.get("fields", {})
resolved = jira_date(fields.get("resolutiondate"))
problem_rows.append(
[
issue.get("key", ""),
fields.get("summary", ""),
f"Q{quarter_for_date(resolved)}" if resolved else "",
team_name(issue),
hours(seconds_value(issue, "aggregatetimespent", "timespent")),
", ".join(flags),
]
)
quarter_rows = quarter_summary_rows(report_issues, year, quarters)
team_rows = team_summary_rows(report_issues)
type_rows = [[name, count] for name, count in Counter(field_name(issue, "issuetype") for issue in report_issues).most_common()]
priority_rows = [[name, count] for name, count in Counter(field_name(issue, "priority") for issue in report_issues).most_common()]
recommendation_items = "".join(f"<li>{html_escape(item)}</li>" for item in recommendations(report_issues))
generated_at = dt.datetime.now().strftime("%Y-%m-%d %H:%M")
quarter_label = ", ".join(f"Q{quarter}" for quarter in quarters)
document = f"""<!doctype html>
<html lang="ru">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Jira {html_escape(project)} Analytics {quarter_label} {year}</title>
<style>
:root {{ --bg: #f6f7fb; --card: #ffffff; --ink: #182033; --muted: #657089; --line: #e5e8f0; --accent: #315efb; --warn: #bd5b00; }}
body {{ margin: 0; background: var(--bg); color: var(--ink); font: 14px/1.45 -apple-system, BlinkMacSystemFont, "Segoe UI", Arial, sans-serif; }}
header {{ padding: 32px; color: white; background: linear-gradient(135deg, #16213f, #315efb); }}
main {{ max-width: 1180px; margin: 0 auto; padding: 24px; }}
h1 {{ margin: 0 0 8px; font-size: 30px; }}
h2 {{ margin: 0 0 14px; font-size: 20px; }}
.meta {{ color: rgba(255,255,255,.78); }}
.grid {{ display: grid; grid-template-columns: repeat(4, minmax(0, 1fr)); gap: 16px; margin-bottom: 16px; }}
.card {{ background: var(--card); border: 1px solid var(--line); border-radius: 16px; padding: 18px; box-shadow: 0 8px 24px rgba(28, 39, 69, .06); }}
.metric {{ color: var(--muted); font-size: 13px; }}
.value {{ margin-top: 8px; font-size: 28px; font-weight: 750; }}
.delta {{ color: var(--warn); }}
table {{ width: 100%; border-collapse: collapse; overflow: hidden; border-radius: 12px; }}
th, td {{ padding: 10px 12px; border-bottom: 1px solid var(--line); text-align: left; vertical-align: top; }}
th {{ background: #eef2ff; color: #27345c; font-size: 12px; text-transform: uppercase; letter-spacing: .03em; }}
tr:last-child td {{ border-bottom: 0; }}
section {{ margin-bottom: 16px; }}
.two {{ display: grid; grid-template-columns: 1fr 1fr; gap: 16px; }}
.empty {{ color: var(--muted); margin: 0; }}
ul {{ margin: 0; padding-left: 20px; }}
li {{ margin: 8px 0; }}
@media (max-width: 860px) {{ .grid, .two {{ grid-template-columns: 1fr; }} header {{ padding: 24px; }} main {{ padding: 16px; }} }}
</style>
</head>
<body>
<header>
<h1>Аналитика закрытых задач Jira</h1>
<div class="meta">Проект {html_escape(project)} · статус {html_escape(closed_status)} · {html_escape(quarter_label)} {year} · сформировано {html_escape(generated_at)}</div>
</header>
<main>
<section class="grid">
<div class="card"><div class="metric">Выполнено задач</div><div class="value">{len(report_issues)}</div></div>
<div class="card"><div class="metric">Фактические часы</div><div class="value">{hours(total_spent)}</div></div>
<div class="card"><div class="metric">Плановые часы</div><div class="value">{hours(total_estimate)}</div></div>
<div class="card"><div class="metric">Отклонение факт-план</div><div class="value delta">{hours(total_spent - total_estimate)}</div></div>
</section>
<section class="card"><h2>Сводка по кварталам</h2>{html_table(["Квартал", "Задачи", "Факт, ч", "План, ч", "Отклонение, ч", "Средний цикл, дни", "Задачи с сигналами"], quarter_rows)}</section>
<section class="card"><h2>Команды и корректировки</h2>{html_table(["Команда/компонент", "Задачи", "Факт, ч", "План, ч", "Отклонение, ч", "Средний цикл, дни", "Основные проблемы"], team_rows)}</section>
<section class="card"><h2>Рекомендуемые корректировки</h2><ul>{recommendation_items}</ul></section>
<section class="two">
<div class="card"><h2>Типы задач</h2>{html_table(["Тип", "Кол-во"], type_rows)}</div>
<div class="card"><h2>Приоритеты</h2>{html_table(["Приоритет", "Кол-во"], priority_rows)}</div>
</section>
<section class="card"><h2>Проблемные сигналы по задачам</h2>{html_table(["Ключ", "Summary", "Квартал", "Команда", "Факт, ч", "Сигналы"], problem_rows[:100], "Нет проблемных сигналов")}</section>
</main>
</body>
</html>
"""
with open(path, "w", encoding="utf-8") as file:
file.write(document)
def analyze(issues: list[dict[str, Any]], stale_days: int, top: int) -> None:
now = dt.datetime.now(dt.timezone.utc)
unresolved = [issue for issue in issues if is_unresolved(issue)]
resolved = [issue for issue in issues if not is_unresolved(issue)]
stale: list[tuple[str, str, str, int]] = []
overdue: list[tuple[str, str, str]] = []
by_status = Counter(field_name(issue, "status") for issue in issues)
by_type = Counter(field_name(issue, "issuetype") for issue in issues)
by_priority = Counter(field_name(issue, "priority") for issue in issues)
by_assignee = Counter(field_name(issue, "assignee", "Unassigned") for issue in issues)
created_by_month = Counter(month_bucket(issue.get("fields", {}).get("created")) for issue in issues)
resolved_by_month = Counter(
month_bucket(issue.get("fields", {}).get("resolutiondate")) for issue in resolved
)
for issue in unresolved:
fields = issue.get("fields", {})
updated = parse_jira_datetime(fields.get("updated"))
if updated:
age_days = (now - updated.astimezone(dt.timezone.utc)).days
if age_days >= stale_days:
stale.append(
(
issue.get("key", "Unknown"),
field_name(issue, "status"),
fields.get("summary") or "",
age_days,
)
)
due_date = fields.get("duedate")
if due_date:
try:
due = dt.date.fromisoformat(due_date)
except ValueError:
due = None
if due and due < now.date():
overdue.append(
(
issue.get("key", "Unknown"),
due.isoformat(),
fields.get("summary") or "",
)
)
print("Jira Ticket Analysis")
print("====================")
print(f"Total issues: {len(issues)}")
print(f"Resolved issues: {len(resolved)}")
print(f"Unresolved issues: {len(unresolved)}")
if issues:
print(f"Resolution ratio: {len(resolved) / len(issues):.1%}")
print(f"Stale unresolved: {len(stale)} (updated >= {stale_days} days ago)")
print(f"Overdue unresolved: {len(overdue)}")
print_counter("By Status", by_status, top)
print_counter("By Issue Type", by_type, top)
print_counter("By Priority", by_priority, top)
print_counter("By Assignee", by_assignee, top)
print_counter("By Component", count_list_field(issues, "components"), top)
print_counter("By Label", count_list_field(issues, "labels"), top)
print_counter("Created By Month", Counter(dict(sorted(created_by_month.items()))), top)
print_counter("Resolved By Month", Counter(dict(sorted(resolved_by_month.items()))), top)
print("\nStale Unresolved Issues")
print("-----------------------")
if not stale:
print("No stale unresolved issues")
else:
for key, status, summary, age_days in sorted(stale, key=lambda item: item[3], reverse=True)[:top]:
print(f"{key:<12} {age_days:>4} days {status:<20} {summary}")
print("\nOverdue Unresolved Issues")
print("-------------------------")
if not overdue:
print("No overdue unresolved issues")
else:
for key, due_date, summary in sorted(overdue, key=lambda item: item[1])[:top]:
print(f"{key:<12} {due_date} {summary}")
def write_json(path: str, issues: list[dict[str, Any]]) -> None:
with open(path, "w", encoding="utf-8") as file:
json.dump(issues, file, ensure_ascii=False, indent=2)
def write_csv(path: str, issues: list[dict[str, Any]]) -> None:
with open(path, "w", encoding="utf-8", newline="") as file:
writer = csv.DictWriter(
file,
fieldnames=[
"key",
"type",
"status",
"priority",
"assignee",
"reporter",
"created",
"updated",
"resolutiondate",
"duedate",
"spent_hours",
"estimate_hours",
"summary",
],
)
writer.writeheader()
for issue in issues:
fields = issue.get("fields", {})
writer.writerow(
{
"key": issue.get("key"),
"type": field_name(issue, "issuetype"),
"status": field_name(issue, "status"),
"priority": field_name(issue, "priority"),
"assignee": field_name(issue, "assignee", "Unassigned"),
"reporter": field_name(issue, "reporter"),
"created": fields.get("created"),
"updated": fields.get("updated"),
"resolutiondate": fields.get("resolutiondate"),
"duedate": fields.get("duedate"),
"spent_hours": hours(seconds_value(issue, "aggregatetimespent", "timespent")),
"estimate_hours": hours(
seconds_value(issue, "aggregatetimeoriginalestimate", "timeoriginalestimate")
),
"summary": fields.get("summary"),
}
)
def list_projects(base_url: str, token: str, user: str | None, search: str | None) -> None:
projects = jira_get(base_url, token, "/rest/api/2/project", {}, user)
if not isinstance(projects, list):
raise RuntimeError("Unexpected Jira project list response")
search_lower = search.casefold() if search else None
rows = []
for project in projects:
key = str(project.get("key", ""))
name = str(project.get("name", ""))
project_type = str(project.get("projectTypeKey", ""))
if search_lower and search_lower not in key.casefold() and search_lower not in name.casefold():
continue
rows.append((key, name, project_type))
if not rows:
print("No projects found")
return
key_width = max(len(row[0]) for row in rows)
type_width = max(len(row[2]) for row in rows)
print(f"{'KEY':<{key_width}} {'TYPE':<{type_width}} NAME")
print(f"{'-' * key_width} {'-' * type_width} {'-' * 40}")
for key, name, project_type in sorted(rows):
print(f"{key:<{key_width}} {project_type:<{type_width}} {name}")
def fetch_agile_pages(
base_url: str,
token: str,
user: str | None,
path: str,
params: dict[str, Any],
page_size: int,
) -> list[dict[str, Any]]:
values: list[dict[str, Any]] = []
start_at = 0
while True:
page_params = dict(params)
page_params.update({"startAt": start_at, "maxResults": page_size})
data = jira_get(base_url, token, path, page_params, user)
batch = data.get("values") or data.get("issues") or []
values.extend(batch)
total = data.get("total")
is_last = bool(data.get("isLast", False))
if not batch or is_last:
break
if total is not None and len(values) >= int(total):
break
if total is None and len(batch) < page_size:
break
start_at += len(batch)
return values
def fetch_boards(
base_url: str,
token: str,
user: str | None,
project: str,
board_id: int | None,
page_size: int,
) -> list[dict[str, Any]]:
if board_id is not None:
board = jira_get(base_url, token, f"/rest/agile/1.0/board/{board_id}", {}, user)
return [board]
return fetch_agile_pages(
base_url,
token,
user,
"/rest/agile/1.0/board",
{"projectKeyOrId": project},
page_size,
)
def sprint_reference_date(sprint: dict[str, Any]) -> dt.date | None:
for field in ("completeDate", "endDate", "startDate"):
value = jira_date(sprint.get(field))
if value:
return value
return None
def sprint_overlaps_range(sprint: dict[str, Any], start: dt.date, end: dt.date) -> bool:
sprint_start = jira_date(sprint.get("startDate")) or sprint_reference_date(sprint)
sprint_end = jira_date(sprint.get("completeDate")) or jira_date(sprint.get("endDate")) or sprint_start
if not sprint_start or not sprint_end:
return False
return sprint_start <= end and sprint_end >= start
def fetch_sprints_for_boards(
base_url: str,
token: str,
user: str | None,
boards: list[dict[str, Any]],
start: dt.date,
end: dt.date,
page_size: int,
) -> list[dict[str, Any]]:
sprints_by_id: dict[int, dict[str, Any]] = {}
for board in boards:
board_id = board.get("id")
if board_id is None:
continue
if str(board.get("type", "")).casefold() != "scrum":
continue
sprints = fetch_agile_pages(
base_url,
token,
user,
f"/rest/agile/1.0/board/{board_id}/sprint",
{"state": "closed"},
page_size,
)
for sprint in sprints:
sprint_id = sprint.get("id")
if isinstance(sprint_id, int) and sprint_overlaps_range(sprint, start, end):
enriched = dict(sprint)
enriched["boardId"] = board_id
enriched["boardName"] = board.get("name", "")
sprints_by_id[sprint_id] = enriched
return sorted(sprints_by_id.values(), key=lambda item: sprint_reference_date(item) or dt.date.min)
def fetch_sprint_issues(
base_url: str,
token: str,
user: str | None,
sprint_id: int,
page_size: int,
) -> list[dict[str, Any]]:
fields = [
"summary",
"issuetype",
"status",
"priority",
"assignee",
"created",
"resolutiondate",
"components",
"labels",
"timespent",
"timeoriginalestimate",
"aggregatetimespent",
"aggregatetimeoriginalestimate",
]
return fetch_agile_pages(
base_url,
token,
user,
f"/rest/agile/1.0/sprint/{sprint_id}/issue",
{"fields": ",".join(fields)},
page_size,
)
def is_closed_issue(issue: dict[str, Any], closed_status: str) -> bool:
fields = issue.get("fields", {})
status = fields.get("status") or {}
category = status.get("statusCategory") if isinstance(status, dict) else None
category_key = category.get("key") if isinstance(category, dict) else None
if str(category_key).casefold() == "done":
return True
if fields.get("resolutiondate"):
return True
return field_name(issue, "status").casefold() == closed_status.casefold()
def issue_type_bucket(issue: dict[str, Any]) -> str:
raw = field_name(issue, "issuetype").casefold()
if any(word in raw for word in ("bug", "defect", "дефект", "ошиб")):
return "Баги"
if any(word in raw for word in ("feature", "story", "фич", "истор")):
return "Фичи"
if any(word in raw for word in ("task", "задач")):
return "Задачи"
return "Прочее"
def sprint_quarter_label(sprint: dict[str, Any]) -> str:
value = sprint_reference_date(sprint)
if not value:
return "Unknown"
return f"Q{quarter_for_date(value)} {value.year}"
def sprint_dates_label(sprint: dict[str, Any]) -> str:
start = jira_date(sprint.get("startDate"))
end = jira_date(sprint.get("completeDate")) or jira_date(sprint.get("endDate"))
if start and end:
return f"{start.isoformat()} - {end.isoformat()}"
if start:
return start.isoformat()
if end:
return end.isoformat()
return ""
def build_sprint_quarter_data(
base_url: str,
token: str,
user: str | None,
project: str,
board_id: int | None,
start: dt.date,
end: dt.date,
page_size: int,
closed_status: str,
) -> tuple[list[dict[str, Any]], dict[str, dict[str, Any]], dict[str, dict[str, Any]], list[dict[str, Any]]]:
boards = fetch_boards(base_url, token, user, project, board_id, page_size)
sprints = fetch_sprints_for_boards(base_url, token, user, boards, start, end, page_size)
quarter_stats: dict[str, dict[str, Any]] = defaultdict(
lambda: {"sprints": set(), "issues": set(), "buckets": Counter(), "spent": 0, "estimate": 0}
)
sprint_stats: dict[str, dict[str, Any]] = {}
issue_rows: list[dict[str, Any]] = []
seen_issue_quarter: set[tuple[str, str]] = set()
for index, sprint in enumerate(sprints, start=1):
sprint_id = sprint.get("id")
if not isinstance(sprint_id, int):
continue
print(f"Fetching sprint {index}/{len(sprints)}: {sprint.get('name', sprint_id)}", file=sys.stderr)
quarter = sprint_quarter_label(sprint)
issues = fetch_sprint_issues(base_url, token, user, sprint_id, page_size)
stat = {"closed": 0, "buckets": Counter(), "spent": 0, "estimate": 0}
for issue in issues:
if not is_closed_issue(issue, closed_status):
continue
key = str(issue.get("key", ""))
if not key:
continue
dedupe_key = (quarter, key)
if dedupe_key in seen_issue_quarter:
continue
seen_issue_quarter.add(dedupe_key)
bucket = issue_type_bucket(issue)
spent = seconds_value(issue, "aggregatetimespent", "timespent")
estimate = seconds_value(issue, "aggregatetimeoriginalestimate", "timeoriginalestimate")
stat["closed"] += 1
stat["buckets"][bucket] += 1
stat["spent"] += spent
stat["estimate"] += estimate
quarter_stats[quarter]["sprints"].add(sprint_id)
quarter_stats[quarter]["issues"].add(key)
quarter_stats[quarter]["buckets"][bucket] += 1
quarter_stats[quarter]["spent"] += spent
quarter_stats[quarter]["estimate"] += estimate
issue_rows.append(
{
"quarter": quarter,
"sprint": sprint.get("name", ""),
"key": key,
"type": field_name(issue, "issuetype"),
"bucket": bucket,
"status": field_name(issue, "status"),
"assignee": field_name(issue, "assignee", "Unassigned"),
"spent": hours(spent),
"estimate": hours(estimate),
"summary": issue.get("fields", {}).get("summary", ""),
}
)
sprint_stats[str(sprint_id)] = {
"name": sprint.get("name", ""),
"state": sprint.get("state", ""),
"board": sprint.get("boardName", ""),
"quarter": quarter,
"dates": sprint_dates_label(sprint),
"closed": stat["closed"],
"buckets": stat["buckets"],
"spent": stat["spent"],
"estimate": stat["estimate"],
}
return boards, quarter_stats, sprint_stats, issue_rows
def write_sprint_quarter_html_report(
path: str,
project: str,
start: dt.date,
end: dt.date,
boards: list[dict[str, Any]],
quarter_stats: dict[str, dict[str, Any]],
sprint_stats: dict[str, dict[str, Any]],
issue_rows: list[dict[str, Any]],
) -> None:
quarter_rows = []
for quarter in sorted(quarter_stats):
data = quarter_stats[quarter]
buckets = data["buckets"]
quarter_rows.append(
[
quarter,
len(data["sprints"]),
len(data["issues"]),
buckets.get("Фичи", 0),
buckets.get("Задачи", 0),
buckets.get("Баги", 0),
buckets.get("Прочее", 0),
hours(data["spent"]),
hours(data["estimate"]),
hours(data["spent"] - data["estimate"]),
]
)
sprint_rows = []
for data in sorted(sprint_stats.values(), key=lambda item: (item["quarter"], item["dates"], item["name"])):
buckets = data["buckets"]
sprint_rows.append(
[
data["quarter"],
data["name"],
data["state"],
data["dates"],
data["closed"],
buckets.get("Фичи", 0),
buckets.get("Задачи", 0),
buckets.get("Баги", 0),
hours(data["spent"]),
]
)
detail_rows = [
[
row["quarter"],
row["sprint"],
row["key"],
row["bucket"],
row["type"],
row["status"],
row["assignee"],
row["spent"],
row["summary"],
]
for row in sorted(issue_rows, key=lambda item: (item["quarter"], item["sprint"], item["key"]))
]
total_issues = sum(len(data["issues"]) for data in quarter_stats.values())
total_sprints = len(sprint_stats)
total_spent = sum(data["spent"] for data in quarter_stats.values())
board_names = ", ".join(str(board.get("name", board.get("id", ""))) for board in boards) or "нет boards"
generated_at = dt.datetime.now().strftime("%Y-%m-%d %H:%M")
document = f"""<!doctype html>
<html lang="ru">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Jira {html_escape(project)} Sprint Quarterly Statistics</title>
<style>
:root {{ --bg: #f5f7fb; --card: #ffffff; --ink: #182033; --muted: #69758f; --line: #e2e7f0; --accent: #2251ff; }}
body {{ margin: 0; background: var(--bg); color: var(--ink); font: 14px/1.45 -apple-system, BlinkMacSystemFont, "Segoe UI", Arial, sans-serif; }}
header {{ padding: 32px; color: white; background: linear-gradient(135deg, #14213d, #2251ff); }}
main {{ max-width: 1240px; margin: 0 auto; padding: 24px; }}
h1 {{ margin: 0 0 8px; font-size: 30px; }}
h2 {{ margin: 0 0 14px; font-size: 20px; }}
.meta {{ color: rgba(255,255,255,.78); }}
.grid {{ display: grid; grid-template-columns: repeat(4, minmax(0, 1fr)); gap: 16px; margin-bottom: 16px; }}
.card {{ background: var(--card); border: 1px solid var(--line); border-radius: 16px; padding: 18px; box-shadow: 0 8px 24px rgba(28, 39, 69, .06); }}
.metric {{ color: var(--muted); font-size: 13px; }}
.value {{ margin-top: 8px; font-size: 28px; font-weight: 750; }}
table {{ width: 100%; border-collapse: collapse; }}
th, td {{ padding: 10px 12px; border-bottom: 1px solid var(--line); text-align: left; vertical-align: top; }}
th {{ background: #eef2ff; color: #27345c; font-size: 12px; text-transform: uppercase; letter-spacing: .03em; }}
section {{ margin-bottom: 16px; }}
.empty {{ color: var(--muted); margin: 0; }}
@media (max-width: 860px) {{ .grid {{ grid-template-columns: 1fr; }} header {{ padding: 24px; }} main {{ padding: 16px; }} }}
</style>
</head>
<body>
<header>
<h1>Статистика закрытых фич, задач и багов по спринтам</h1>
<div class="meta">Проект {html_escape(project)} · период {start.isoformat()} - {end.isoformat()} · boards: {html_escape(board_names)} · сформировано {html_escape(generated_at)}</div>
</header>
<main>
<section class="grid">
<div class="card"><div class="metric">Спринтов в периоде</div><div class="value">{total_sprints}</div></div>
<div class="card"><div class="metric">Закрытых задач</div><div class="value">{total_issues}</div></div>
<div class="card"><div class="metric">Фактические часы</div><div class="value">{hours(total_spent)}</div></div>
<div class="card"><div class="metric">Кварталов</div><div class="value">{len(quarter_stats)}</div></div>
</section>
<section class="card"><h2>Итоги по кварталам</h2>{html_table(["Квартал", "Спринты", "Закрыто всего", "Фичи", "Задачи", "Баги", "Прочее", "Факт, ч", "План, ч", "Отклонение, ч"], quarter_rows)}</section>
<section class="card"><h2>Разбивка по спринтам</h2>{html_table(["Квартал", "Спринт", "Статус", "Даты", "Закрыто", "Фичи", "Задачи", "Баги", "Факт, ч"], sprint_rows)}</section>
<section class="card"><h2>Закрытые задачи</h2>{html_table(["Квартал", "Спринт", "Ключ", "Категория", "Тип Jira", "Статус", "Исполнитель", "Факт, ч", "Summary"], detail_rows, "Нет закрытых фич/задач/багов")}</section>
</main>
</body>
</html>
"""
with open(path, "w", encoding="utf-8") as file:
file.write(document)
def main() -> int:
args = parse_args()
if not args.token:
print("Error: set JIRA_TOKEN or pass --token.", file=sys.stderr)
return 2
if args.page_size < 1:
print("Error: --page-size must be greater than 0.", file=sys.stderr)
return 2
if args.list_projects:
try:
list_projects(args.base_url, args.token, args.user, args.project_search)
except RuntimeError as exc:
print(f"Error: {exc}", file=sys.stderr)
return 1
return 0
if args.html_sprint_quarter_out:
try:
sprint_start = parse_date_arg(args.sprints_from, "--sprints-from")
sprint_end = parse_date_arg(args.sprints_to, "--sprints-to")
except ValueError as exc:
print(f"Error: {exc}", file=sys.stderr)
return 2
if sprint_start > sprint_end:
print("Error: --sprints-from must be earlier than --sprints-to.", file=sys.stderr)
return 2
try:
boards, quarter_stats, sprint_stats, issue_rows = build_sprint_quarter_data(
args.base_url,
args.token,
args.user,
args.project,
args.board_id,
sprint_start,
sprint_end,
args.page_size,
args.closed_status,
)
write_sprint_quarter_html_report(
args.html_sprint_quarter_out,
args.project,
sprint_start,
sprint_end,
boards,
quarter_stats,
sprint_stats,
issue_rows,
)
print(f"Sprint quarter HTML report written to {args.html_sprint_quarter_out}")
print(f"Boards: {len(boards)}; sprints: {len(sprint_stats)}; closed issues: {len(issue_rows)}")
except RuntimeError as exc:
print(f"Error: {exc}", file=sys.stderr)
return 1
return 0
try:
quarters = parse_quarters(args.quarters)
except ValueError as exc:
print(f"Error: {exc}", file=sys.stderr)
return 2
if not quarters:
print("Error: --quarters must contain at least one quarter.", file=sys.stderr)
return 2
if args.jql:
jql = args.jql
elif args.html_quarter_out:
start = quarter_bounds(args.report_year, min(quarters))[0]
end = quarter_bounds(args.report_year, max(quarters))[1]
jql = (
f'project = {jql_project(args.project)} AND status = {jql_string(args.closed_status)} '
f'AND resolutiondate >= "{start.isoformat()}" '
f'AND resolutiondate < "{end.isoformat()}" ORDER BY resolutiondate ASC'
)
else:
jql = f"project = {jql_project(args.project)} ORDER BY updated DESC"
try:
issues = fetch_issues(args.base_url, args.token, args.user, jql, args.page_size)
analyze(issues, args.stale_days, args.top)
if args.html_quarter_out:
write_quarter_html_report(
args.html_quarter_out,
issues,
args.project,
args.report_year,
quarters,
args.closed_status,
)
print(f"Quarter HTML report written to {args.html_quarter_out}")
if args.json_out:
write_json(args.json_out, issues)
print(f"\nRaw JSON written to {args.json_out}")
if args.csv_out:
write_csv(args.csv_out, issues)
print(f"CSV written to {args.csv_out}")
except RuntimeError as exc:
print(f"Error: {exc}", file=sys.stderr)
message = str(exc)
if "does not exist for the field 'project'" in message:
print(
"Hint: Jira does not recognize this project key for the current token. "
"Check the real project key with --list-projects, or use a token/user with Browse Projects permission.",
file=sys.stderr,
)
return 1
return 0
if __name__ == "__main__":
raise SystemExit(main())