scripts/analyze.py - master - birko/analytics_claude-code

scripts/analyze.py
225 строк8 KB
birko
Initial setup: Jira status duration analyzer
29 май 2026, 16:42
29 май 2026, 16:42720ec83
#!/usr/bin/env python3
"""
Jira Status Duration Analyzer
Анализирует время пребывания задач в каждом статусе
"""

import pandas as pd
import numpy as np
from pathlib import Path
from datetime import datetime
import json

DATA_DIR = Path(__file__).parent.parent / "data"
REPORTS_DIR = Path(__file__).parent.parent / "reports"


def load_csv(file_path: Path) -> pd.DataFrame:
    """Загрузка CSV выгрузки из Jira"""
    return pd.read_csv(file_path)


def load_json(file_path: Path) -> pd.DataFrame:
    """Загрузка JSON выгрузки из Jira"""
    with open(file_path, 'r', encoding='utf-8') as f:
        data = json.load(f)

    # Разбираем историю переходов
    rows = []
    for issue in data.get('issues', []):
        key = issue['key']
        changelog = issue.get('changelog', {}).get('histories', [])

        prev_status = None
        for entry in changelog:
            for item in entry.get('items', []):
                if item['field'] == 'status':
                    rows.append({
                        'key': key,
                        'from_status': item.get('fromString', prev_status),
                        'to_status': item['toString'],
                        'transition_date': entry['created'],
                        'author': entry.get('author', {}).get('displayName', '')
                    })
                    prev_status = item['toString']

    return pd.DataFrame(rows)


def calculate_status_durations(df: pd.DataFrame) -> pd.DataFrame:
    """
    Расчёт времени в каждом статусе для каждой задачи
    """
    df = df.copy()
    df['transition_date'] = pd.to_datetime(df['transition_date'])
    df = df.sort_values(['key', 'transition_date'])

    # Считаем время до следующего перехода
    df['next_date'] = df.groupby('key')['transition_date'].shift(-1)
    df['duration_days'] = (df['next_date'] - df['transition_date']).dt.total_seconds() / 86400

    # Для последнего статуса - считаем до текущего момента
    mask = df['next_date'].isna()
    df.loc[mask, 'duration_days'] = (datetime.now() - df.loc[mask, 'transition_date']).dt.total_seconds() / 86400

    return df


def analyze_delays(df: pd.DataFrame) -> dict:
    """
    Анализ задержек по статусам
    """
    results = {
        'summary': {},
        'by_status': {},
        'top_delays': [],
        'anomalies': []
    }

    # Общая статистика по статусам
    status_stats = df.groupby('to_status')['duration_days'].agg([
        ('count', 'count'),
        ('mean', 'mean'),
        ('median', 'median'),
        ('std', 'std'),
        ('p90', lambda x: x.quantile(0.9)),
        ('max', 'max'),
        ('total', 'sum')
    ]).round(2)

    results['by_status'] = status_stats.to_dict('index')

    # Сводка
    results['summary'] = {
        'total_issues': df['key'].nunique(),
        'total_transitions': len(df),
        'date_range': {
            'from': df['transition_date'].min().isoformat(),
            'to': df['transition_date'].max().isoformat()
        },
        'avg_cycle_time': df.groupby('key')['duration_days'].sum().mean()
    }

    # Топ задач по задержкам
    issue_totals = df.groupby(['key', 'to_status'])['duration_days'].sum().reset_index()
    top_delays = issue_totals.nlargest(20, 'duration_days')
    results['top_delays'] = top_delays.to_dict('records')

    # Аномалии (дольше 90-го перцентиля)
    p90 = df['duration_days'].quantile(0.9)
    anomalies = df[df['duration_days'] > p90][['key', 'to_status', 'duration_days']].copy()
    anomalies = anomalies.sort_values('duration_days', ascending=False)
    results['anomalies'] = anomalies.head(50).to_dict('records')
    results['anomaly_threshold'] = round(p90, 2)

    return results


def generate_report(results: dict, output_path: Path):
    """Генерация отчёта в Markdown"""

    md = ["# Отчёт по задержкам в статусах Jira\n"]
    md.append(f"*Сгенерировано: {datetime.now().strftime('%Y-%m-%d %H:%M')}*\n")

    # Сводка
    s = results['summary']
    md.append("## Общая информация\n")
    md.append(f"- Всего задач: **{s['total_issues']}**")
    md.append(f"- Всего переходов: **{s['total_transitions']}**")
    md.append(f"- Период: {s['date_range']['from'][:10]} — {s['date_range']['to'][:10]}")
    md.append(f"- Среднее время цикла: **{s['avg_cycle_time']:.1f} дней**\n")

    # Статистика по статусам
    md.append("## Время в статусах (сортировка по среднему)\n")
    md.append("| Статус | Кол-во | Среднее (дн) | Медиана | P90 | Макс | Всего |")
    md.append("|--------|--------|--------------|---------|-----|------|-------|")

    sorted_status = sorted(
        results['by_status'].items(),
        key=lambda x: x[1]['mean'],
        reverse=True
    )

    for status, stats in sorted_status:
        md.append(
            f"| {status} | {stats['count']:.0f} | {stats['mean']:.1f} | "
            f"{stats['median']:.1f} | {stats['p90']:.1f} | {stats['max']:.1f} | {stats['total']:.1f} |"
        )
    md.append("")

    # Топ задержек
    md.append("## Топ-20 задач по времени в статусе\n")
    md.append("| Задача | Статус | Дней |")
    md.append("|--------|--------|------|")
    for row in results['top_delays'][:20]:
        md.append(f"| {row['key']} | {row['to_status']} | {row['duration_days']:.1f} |")
    md.append("")

    # Аномалии
    md.append(f"## Аномалии (дольше {results['anomaly_threshold']} дней)\n")
    md.append(f"Всего аномальных задержек: {len(results['anomalies'])}\n")

    if results['anomalies']:
        md.append("| Задача | Статус | Дней |")
        md.append("|--------|--------|------|")
        for row in results['anomalies'][:20]:
            md.append(f"| {row['key']} | {row['to_status']} | {row['duration_days']:.1f} |")

    # Рекомендации
    md.append("\n## Рекомендации\n")
    if sorted_status:
        worst_status = sorted_status[0][0]
        md.append(f"1. **{worst_status}** — статус с наибольшим средним временем. Рекомендуется:")
        md.append("   - Проверить, есть ли блокировки на этом этапе")
        md.append("   - Добавить SLA для этого статуса")
        md.append("   - Настроить автоматические уведомления при задержках")

    md.append("\n2. Настроить дашборд с метриками:")
    md.append("   - Control chart для времени цикла")
    md.append("   - Cumulative flow diagram")
    md.append("   - Aging chart для текущих задач")

    with open(output_path, 'w', encoding='utf-8') as f:
        f.write('\n'.join(md))

    print(f"Отчёт сохранён: {output_path}")


def main():
    """Основной скрипт"""
    # Поиск последней выгрузки
    raw_files = list(DATA_DIR.glob("*.csv")) + list(DATA_DIR.glob("*.json"))
    if not raw_files:
        print(f"Нет файлов в {DATA_DIR}")
        return

    latest_file = max(raw_files, key=lambda p: p.stat().st_mtime)
    print(f"Обрабатываю: {latest_file}")

    # Загрузка
    if latest_file.suffix == '.csv':
        df = load_csv(latest_file)
    else:
        df = load_json(latest_file)

    print(f"Загружено строк: {len(df)}")

    # Анализ
    df = calculate_status_durations(df)
    results = analyze_delays(df)

    # Отчёт
    REPORTS_DIR.mkdir(parents=True, exist_ok=True)
    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
    report_path = REPORTS_DIR / f"report_{timestamp}.md"
    generate_report(results, report_path)

    # JSON с данными для дашборда
    json_path = REPORTS_DIR / f"data_{timestamp}.json"
    with open(json_path, 'w', encoding='utf-8') as f:
        json.dump(results, f, indent=2, default=str)
    print(f"JSON данные: {json_path}")


if __name__ == '__main__':
    main()