Загрузка данных
напиши код парсера на python, который подключается к Jira и перебирает все задачи из конкретной доски, вытаскивает информацию из всех полей задачи и сохраняет данные в txt файл. код подробно прокомментировать
Как пользоваться:
Self-hosted Jira (Server/DC): задай JIRA_URL, JIRA_BOARD_ID и JIRA_TOKEN (это Personal Access Token). JIRA_EMAIL не задавай.
Jira Cloud: дополнительно задай JIRA_EMAIL, а в JIRA_TOKEN положи API-токен.
Самоподписанный сертификат во внутренней сети: выставь JIRA_VERIFY_SSL=0.
Отдельно про три момента:
Какие задачи попадут: выгружается всё, что попадает под фильтр доски, включая бэклог.
Changelog: история изменений в ответе Agile API может быть урезана до последних ~100 записей. Если нужна полная, её надо догружать через /rest/api/2/issue/{key}/changelog по аналогии с комментариями.
Пустые поля пропускаются. Если нужны все поля без исключения, убери continue после проверки if not text.
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Парсер задач Jira с конкретной доски (Scrum/Kanban).
Что делает:
1. Подключается к Jira (Cloud или Server/Data Center).
2. Получает справочник всех полей, чтобы customfield_10001 и т.п.
превратить в нормальные названия ("Story Points", "Epic Link"...).
3. Постранично перебирает ВСЕ задачи указанной доски.
4. Для каждой задачи вытаскивает все поля, приводит значения к
читаемому виду и пишет в txt-файл.
Зависимости:
pip install requests
Настройка через переменные окружения:
JIRA_URL - адрес Jira, например https://jira.company.local
JIRA_BOARD_ID - числовой ID доски (виден в URL доски: ...rapidView=42 или /boards/42)
JIRA_TOKEN - API-токен (Cloud) или Personal Access Token (Server/DC)
JIRA_EMAIL - email пользователя; ЗАДАВАТЬ ТОЛЬКО для Jira Cloud
(если задан -> Basic auth email:token, иначе -> Bearer PAT)
JIRA_OUTPUT - имя выходного файла (по умолчанию jira_board_<id>.txt)
JIRA_VERIFY_SSL - "0", чтобы отключить проверку сертификата (самоподписанный)
Пример запуска:
export JIRA_URL=https://jira.company.local
export JIRA_BOARD_ID=42
export JIRA_TOKEN=xxxxxxxx
python jira_board_parser.py
"""
import os
import sys
import time
import json
from datetime import datetime
import requests
# ---------------------------------------------------------------------------
# 1. КОНФИГУРАЦИЯ
# ---------------------------------------------------------------------------
# Базовый адрес Jira без завершающего слэша, чтобы потом корректно склеивать URL.
JIRA_URL = os.environ.get("JIRA_URL", "").rstrip("/")
# ID доски. Приводим к строке, в URL он всё равно подставляется как текст.
BOARD_ID = os.environ.get("JIRA_BOARD_ID", "")
# Учётные данные.
JIRA_TOKEN = os.environ.get("JIRA_TOKEN", "")
JIRA_EMAIL = os.environ.get("JIRA_EMAIL", "") # пусто -> считаем, что Server/DC с PAT
# Файл, куда складываем результат.
OUTPUT_FILE = os.environ.get("JIRA_OUTPUT", f"jira_board_{BOARD_ID}.txt")
# Проверка SSL. Во внутренних сетях часто самоподписанные сертификаты.
VERIFY_SSL = os.environ.get("JIRA_VERIFY_SSL", "1") != "0"
# Сколько задач запрашивать за один запрос. Jira обычно ограничивает
# максимум 50-100, больше просить смысла нет - всё равно урежет.
PAGE_SIZE = 50
# Таймаут одного HTTP-запроса в секундах.
REQUEST_TIMEOUT = 60
# Сколько раз повторять запрос при временных ошибках (429, 5xx).
MAX_RETRIES = 5
# ---------------------------------------------------------------------------
# 2. HTTP-СЕССИЯ И АВТОРИЗАЦИЯ
# ---------------------------------------------------------------------------
def build_session() -> requests.Session:
"""
Создаёт requests.Session с нужной авторизацией и заголовками.
Session переиспользует TCP-соединение между запросами - это заметно
быстрее, когда запросов сотни.
"""
session = requests.Session()
session.headers.update({"Accept": "application/json"})
session.verify = VERIFY_SSL
if JIRA_EMAIL:
# Jira Cloud: Basic auth, где логин = email, пароль = API-токен.
session.auth = (JIRA_EMAIL, JIRA_TOKEN)
else:
# Jira Server / Data Center 8.14+: Personal Access Token через Bearer.
session.headers["Authorization"] = f"Bearer {JIRA_TOKEN}"
return session
def api_get(session: requests.Session, path: str, params: dict | None = None) -> dict | list:
"""
Выполняет GET к Jira API с повторами при временных ошибках.
path - путь относительно JIRA_URL, например "/rest/api/2/field"
params - query-параметры запроса
Возвращает распарсенный JSON (dict или list).
"""
url = f"{JIRA_URL}{path}"
for attempt in range(1, MAX_RETRIES + 1):
try:
resp = session.get(url, params=params, timeout=REQUEST_TIMEOUT)
except requests.RequestException as exc:
# Сетевая ошибка (обрыв, таймаут) - ждём и пробуем снова.
wait = 2 ** attempt
print(f" [!] Сетевая ошибка: {exc}. Повтор через {wait} c...", file=sys.stderr)
time.sleep(wait)
continue
# 429 = превышен лимит запросов. Jira часто присылает Retry-After.
# 5xx = проблема на стороне сервера, обычно временная.
if resp.status_code == 429 or resp.status_code >= 500:
wait = int(resp.headers.get("Retry-After", 2 ** attempt))
print(f" [!] HTTP {resp.status_code}. Повтор через {wait} c...", file=sys.stderr)
time.sleep(wait)
continue
# Любая другая ошибка (401, 403, 404...) - повторять бессмысленно.
if not resp.ok:
raise RuntimeError(
f"Ошибка {resp.status_code} при запросе {url}: {resp.text[:500]}"
)
return resp.json()
raise RuntimeError(f"Не удалось выполнить запрос {url} после {MAX_RETRIES} попыток")
# ---------------------------------------------------------------------------
# 3. СПРАВОЧНИК ПОЛЕЙ
# ---------------------------------------------------------------------------
def load_field_names(session: requests.Session) -> dict[str, str]:
"""
Загружает список всех полей Jira и строит словарь id -> название.
Без этого в выгрузке были бы ключи вида customfield_10016,
по которым непонятно, что это за поле.
"""
fields = api_get(session, "/rest/api/2/field")
return {f["id"]: f.get("name", f["id"]) for f in fields}
# ---------------------------------------------------------------------------
# 4. ПЕРЕБОР ЗАДАЧ ДОСКИ
# ---------------------------------------------------------------------------
def iter_board_issues(session: requests.Session, board_id: str):
"""
Генератор: по одной отдаёт все задачи доски.
Используется Agile API: /rest/agile/1.0/board/{id}/issue
Он возвращает задачи, попадающие под фильтр доски (включая бэклог).
Пагинация: передаём startAt и maxResults, пока не соберём total.
"""
start_at = 0
while True:
data = api_get(
session,
f"/rest/agile/1.0/board/{board_id}/issue",
params={
"startAt": start_at,
"maxResults": PAGE_SIZE,
# "*all" - запросить абсолютно все поля, включая кастомные.
"fields": "*all",
# renderedFields - поля в виде HTML (описание, комментарии),
# changelog - история изменений задачи.
"expand": "renderedFields,changelog",
},
)
issues = data.get("issues", [])
total = data.get("total", 0)
for issue in issues:
yield issue
# Сдвигаем окно на количество реально полученных задач
# (сервер может вернуть меньше, чем PAGE_SIZE).
start_at += len(issues)
print(f" Получено {start_at} из {total}")
# Выходим, если всё забрали или сервер вернул пустую страницу
# (защита от бесконечного цикла).
if start_at >= total or not issues:
break
def load_all_comments(session: requests.Session, issue_key: str) -> list[dict]:
"""
Догружает ВСЕ комментарии задачи.
В ответе поиска поле comment может быть обрезано (обычно отдаются
первые N штук), поэтому при необходимости забираем их отдельно
постранично.
"""
comments = []
start_at = 0
while True:
data = api_get(
session,
f"/rest/api/2/issue/{issue_key}/comment",
params={"startAt": start_at, "maxResults": 100},
)
batch = data.get("comments", [])
comments.extend(batch)
start_at += len(batch)
if start_at >= data.get("total", 0) or not batch:
break
return comments
# ---------------------------------------------------------------------------
# 5. ПРИВЕДЕНИЕ ЗНАЧЕНИЙ К ЧИТАЕМОМУ ВИДУ
# ---------------------------------------------------------------------------
def format_value(value) -> str:
"""
Превращает значение поля Jira в строку.
Поля бывают очень разные:
- простые: строка, число, None
- объекты: пользователь, статус, приоритет, версия...
- списки: метки, компоненты, fixVersions, вложения...
У объектов Jira почти всегда есть "человеческое" поле:
displayName / name / value / key. Берём первое найденное.
"""
if value is None or value == "" or value == []:
return ""
if isinstance(value, (str, int, float, bool)):
return str(value)
if isinstance(value, list):
# Список: форматируем каждый элемент и склеиваем через запятую.
parts = [format_value(v) for v in value]
return ", ".join(p for p in parts if p)
if isinstance(value, dict):
# Пользователь (assignee, reporter...)
if "displayName" in value:
email = value.get("emailAddress")
return f"{value['displayName']} <{email}>" if email else value["displayName"]
# Вложение: имя файла + ссылка на скачивание
if "filename" in value and "content" in value:
return f"{value['filename']} ({value['content']})"
# Связь задач (issuelinks): тип связи + связанная задача
if "type" in value and ("inwardIssue" in value or "outwardIssue" in value):
link_type = value["type"]
if "outwardIssue" in value:
return f"{link_type.get('outward', '')} {value['outwardIssue'].get('key', '')}"
return f"{link_type.get('inward', '')} {value['inwardIssue'].get('key', '')}"
# Кастомное поле-список с вложенным значением (cascading select)
if "value" in value:
child = value.get("child")
return f"{value['value']} -> {child['value']}" if child else str(value["value"])
# Статус, приоритет, тип задачи, версия, компонент, резолюция...
if "name" in value:
return str(value["name"])
# Родительская задача / эпик и т.п.
if "key" in value:
summary = value.get("fields", {}).get("summary", "")
return f"{value['key']} {summary}".strip()
# Трекинг времени
if "originalEstimate" in value or "timeSpent" in value:
return ", ".join(f"{k}={v}" for k, v in value.items() if "Seconds" not in k)
# Описание в формате Atlassian Document Format (Jira Cloud, API v3)
if value.get("type") == "doc":
return adf_to_text(value)
# Ничего не подошло - выводим как JSON, чтобы данные не потерялись.
return json.dumps(value, ensure_ascii=False)
return str(value)
def adf_to_text(node: dict) -> str:
"""
Рекурсивно вытаскивает текст из Atlassian Document Format.
Это древовидный JSON, где текст лежит в узлах type="text".
"""
if node.get("type") == "text":
return node.get("text", "")
parts = [adf_to_text(child) for child in node.get("content", [])]
# Параграфы и заголовки разделяем переносами строк.
sep = "\n" if node.get("type") in ("doc", "bulletList", "orderedList") else ""
return sep.join(parts)
# ---------------------------------------------------------------------------
# 6. ЗАПИСЬ ОДНОЙ ЗАДАЧИ В ФАЙЛ
# ---------------------------------------------------------------------------
def write_issue(fh, issue: dict, field_names: dict[str, str], session: requests.Session) -> None:
"""
Пишет одну задачу в открытый текстовый файл.
Формат:
===== KEY-123 =====
Название поля (id): значение
...
--- Комментарии ---
--- История изменений ---
"""
key = issue["key"]
fields = issue.get("fields", {})
fh.write("=" * 80 + "\n")
fh.write(f"ЗАДАЧА: {key}\n")
fh.write(f"Ссылка: {JIRA_URL}/browse/{key}\n")
fh.write("=" * 80 + "\n")
# Сортируем поля по названию, чтобы в каждой задаче порядок был одинаковый
# и файл было удобно читать и сравнивать.
for field_id in sorted(fields, key=lambda f: field_names.get(f, f).lower()):
# Комментарии выводим отдельным блоком ниже, в полном объёме.
if field_id == "comment":
continue
text = format_value(fields[field_id])
if not text:
# Пустые поля пропускаем, иначе файл раздуется в разы.
continue
name = field_names.get(field_id, field_id)
# Многострочные значения (описание) выносим на новую строку с отступом.
if "\n" in text:
indented = "\n".join(" " + line for line in text.splitlines())
fh.write(f"{name} ({field_id}):\n{indented}\n")
else:
fh.write(f"{name} ({field_id}): {text}\n")
# --- Комментарии ---
comment_field = fields.get("comment") or {}
comments = comment_field.get("comments", [])
# Если сервер отдал не все комментарии - догружаем отдельно.
if comment_field.get("total", 0) > len(comments):
comments = load_all_comments(session, key)
if comments:
fh.write("\n--- Комментарии ---\n")
for c in comments:
author = format_value(c.get("author"))
body = format_value(c.get("body"))
fh.write(f"[{c.get('created', '')}] {author}:\n")
fh.write("\n".join(" " + line for line in body.splitlines()) + "\n")
# --- История изменений (changelog) ---
histories = issue.get("changelog", {}).get("histories", [])
if histories:
fh.write("\n--- История изменений ---\n")
for h in histories:
author = format_value(h.get("author"))
for item in h.get("items", []):
fh.write(
f"[{h.get('created', '')}] {author}: {item.get('field')}: "
f"'{item.get('fromString') or ''}' -> '{item.get('toString') or ''}'\n"
)
fh.write("\n\n")
# ---------------------------------------------------------------------------
# 7. ТОЧКА ВХОДА
# ---------------------------------------------------------------------------
def main() -> None:
# Проверяем, что обязательные параметры заданы.
missing = [n for n, v in (("JIRA_URL", JIRA_URL), ("JIRA_BOARD_ID", BOARD_ID),
("JIRA_TOKEN", JIRA_TOKEN)) if not v]
if missing:
sys.exit(f"Не заданы переменные окружения: {', '.join(missing)}")
if not VERIFY_SSL:
# Глушим предупреждения urllib3 о непроверенном сертификате.
requests.packages.urllib3.disable_warnings()
session = build_session()
print("Загружаю справочник полей...")
field_names = load_field_names(session)
print(f" Полей в Jira: {len(field_names)}")
print(f"Выгружаю задачи доски {BOARD_ID}...")
count = 0
# encoding="utf-8" обязателен, иначе кириллица сломается на Windows.
with open(OUTPUT_FILE, "w", encoding="utf-8") as fh:
fh.write(f"Выгрузка доски {BOARD_ID} из {JIRA_URL}\n")
fh.write(f"Дата: {datetime.now():%Y-%m-%d %H:%M:%S}\n\n")
for issue in iter_board_issues(session, BOARD_ID):
write_issue(fh, issue, field_names, session)
count += 1
print(f"Готово: {count} задач сохранено в {OUTPUT_FILE}")
if __name__ == "__main__":
main()