Загрузка данных



напиши код парсера на 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()