1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

┌─────────────────────────────────────────────────────────────────────────────────────┐
│                        需求到上线（Idea to Production）                              │
├─────────────────────────────────────────────────────────────────────────────────────┤
│                                                                                     │
│  ┌──────────┐  WT:3d  ┌──────────┐  WT:5d  ┌──────────┐  WT:2d  ┌──────────┐      │
│  │ 需求评审  │ ──────→ │  开发     │ ──────→ │ 代码审查  │ ──────→ │  测试     │      │
│  │ PT: 2h   │         │ PT: 5d   │         │ PT: 4h   │         │ PT: 2d   │      │
│  │ %CA: 80% │         │ %CA: 70% │         │ %CA: 60% │         │ %CA: 75% │      │
│  └──────────┘         └──────────┘         └──────────┘         └──────────┘      │
│                                                                        │            │
│                                               WT: 4d                   │            │
│                                                                        ▼            │
│  ┌──────────┐  WT:1d  ┌──────────┐         ┌──────────┐  WT: 2d  ┌──────────┐     │
│  │ 生产部署  │ ←────── │ 预发验证  │ ←────── │ 安全审查  │ ←────── │ 回归测试  │     │
│  │ PT: 1h   │         │ PT: 2h   │         │ PT: 3h   │         │ PT: 1d   │     │
│  │ %CA: 95% │         │ %CA: 85% │         │ %CA: 90% │         │ %CA: 80% │     │
│  └──────────┘         └──────────┘         └──────────┘         └──────────┘     │
│                                                                                     │
├─────────────────────────────────────────────────────────────────────────────────────┤
│  总 Lead Time: ~24 天  |  总 Process Time: ~9.5 天  |  增值比: 39.6%              │
│  总 Wait Time: ~17 天  |  WIP（平均）: 12 个需求                                   │
└─────────────────────────────────────────────────────────────────────────────────────┘

图例：PT = Process Time（增值时间）  WT = Wait Time（等待时间）  %CA = %Complete/Accurate

1
2
3
4
5
6
7

DORA 指标 → 发现表现差距
    ↓
VSM → 定位差距的根因（哪个步骤的等待时间/返工率最高）
    ↓
改进动作 → 针对性优化
    ↓
DORA 指标 → 验证改进效果

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134

#!/usr/bin/env python3
"""
VSM 度量数据采集脚本
从 GitLab API 自动采集 Lead Time、Cycle Time、Deployment Frequency
"""

import requests
from datetime import datetime, timedelta
from collections import defaultdict
import json

GITLAB_URL = "https://gitlab.example.com"
TOKEN = "your-private-token"
PROJECT_ID = 12345
HEADERS = {"PRIVATE-TOKEN": TOKEN}

def get_merge_requests(state="merged", since_days=30):
    """获取指定时间范围内的 Merge Requests"""
    since = (datetime.now() - timedelta(days=since_days)).isoformat()
    url = f"{GITLAB_URL}/api/v4/projects/{PROJECT_ID}/merge_requests"
    params = {
        "state": state,
        "updated_after": since,
        "per_page": 100,
        "order_by": "updated_at",
    }
    resp = requests.get(url, headers=HEADERS, params=params)
    resp.raise_for_status()
    return resp.json()

def get_pipeline_jobs(pipeline_id):
    """获取流水线的所有 Job，用于计算各阶段耗时"""
    url = f"{GITLAB_URL}/api/v4/projects/{PROJECT_ID}/pipelines/{pipeline_id}/jobs"
    resp = requests.get(url, headers=HEADERS, params={"per_page": 100})
    resp.raise_for_status()
    return resp.json()

def get_deployments(since_days=30):
    """获取部署事件"""
    since = (datetime.now() - timedelta(days=since_days)).isoformat()
    url = f"{GITLAB_URL}/api/v4/projects/{PROJECT_ID}/deployments"
    params = {"updated_after": since, "per_page": 100}
    resp = requests.get(url, headers=HEADERS, params=params)
    resp.raise_for_status()
    return resp.json()

def calc_time_diff(start_str, end_str):
    """计算两个 ISO 时间戳之间的小时差"""
    start = datetime.fromisoformat(start_str.replace("Z", "+00:00"))
    end = datetime.fromisoformat(end_str.replace("Z", "+00:00"))
    return (end - start).total_seconds() / 3600

def analyze_merge_requests(mrs):
    """分析 MR 的 Cycle Time 和 WIP"""
    cycle_times = []
    review_waits = []

    for mr in mrs:
        if mr.get("merged_at") and mr.get("created_at"):
            # Cycle Time: 从 MR 创建到合并
            ct = calc_time_diff(mr["created_at"], mr["merged_at"])
            cycle_times.append(ct)

            # 审查等待时间: 从创建到首次审查评论
            if mr.get("first_reviewed_at"):
                rw = calc_time_diff(mr["created_at"], mr["first_reviewed_at"])
                review_waits.append(rw)

    return {
        "avg_cycle_time_hours": round(sum(cycle_times) / len(cycle_times), 1) if cycle_times else 0,
        "p50_cycle_time_hours": round(sorted(cycle_times)[len(cycle_times) // 2], 1) if cycle_times else 0,
        "avg_review_wait_hours": round(sum(review_waits) / len(review_waits), 1) if review_waits else 0,
        "total_merged": len(mrs),
    }

def analyze_deployments(deployments, since_days=30):
    """分析部署频率和变更失败率"""
    total = len(deployments)
    failed = sum(1 for d in deployments if d.get("status") == "failed")

    return {
        "deployment_frequency": f"{total} / {since_days}d",
        "deployment_frequency_daily": round(total / since_days, 2),
        "change_failure_rate": f"{round(failed / total * 100, 1)}%" if total > 0 else "N/A",
        "total_deployments": total,
        "failed_deployments": failed,
    }

def generate_vsm_report():
    """生成 VSM 度量报告"""
    print("=" * 60)
    print("  VSM 度量报告 - 研发价值流健康度")
    print("=" * 60)

    # 采集数据
    mrs = get_merge_requests(since_days=30)
    deployments = get_deployments(since_days=30)

    # 分析
    mr_stats = analyze_merge_requests(mrs)
    deploy_stats = analyze_deployments(deployments, since_days=30)

    # 输出
    print(f"\n📊 Cycle Time 分析（最近 30 天）:")
    print(f"   平均 Cycle Time: {mr_stats['avg_cycle_time_hours']}h")
    print(f"   P50 Cycle Time:  {mr_stats['p50_cycle_time_hours']}h")
    print(f"   平均审查等待:    {mr_stats['avg_review_wait_hours']}h")
    print(f"   合并 MR 总数:    {mr_stats['total_merged']}")

    print(f"\n🚀 部署分析（最近 30 天）:")
    print(f"   部署频率:        {deploy_stats['deployment_frequency_daily']}/天")
    print(f"   变更失败率:      {deploy_stats['change_failure_rate']}")
    print(f"   总部署次数:      {deploy_stats['total_deployments']}")
    print(f"   失败部署次数:    {deploy_stats['failed_deployments']}")

    # 健康度评估
    print(f"\n🏥 健康度评估:")
    daily_freq = deploy_stats["deployment_frequency_daily"]
    avg_ct = mr_stats["avg_cycle_time_hours"]

    if daily_freq >= 1 and avg_ct <= 24:
        print("   ✅ Elite - 按需部署，Cycle Time < 1天")
    elif daily_freq >= 0.14 and avg_ct <= 168:
        print("   🟡 High - 周级部署，Cycle Time < 1周")
    elif daily_freq >= 0.03:
        print("   🟠 Medium - 月级部署，需要优化")
    else:
        print("   🔴 Low - 部署频率过低，建议从 CI/CD 自动化入手")

    print("\n" + "=" * 60)
    return {"mr_stats": mr_stats, "deploy_stats": deploy_stats}

if __name__ == "__main__":
    report = generate_vsm_report()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89

#!/usr/bin/env python3
"""
从 Jira 采集 Lead Time 和 WIP 数据
适用于需求从 "Ready" 到 "Done" 的价值流度量
"""

import requests
from datetime import datetime
from jira import JIRA

JIRA_URL = "https://jira.example.com"
USERNAME = "your-email@example.com"
API_TOKEN = "your-api-token"
PROJECT_KEY = "PROJ"
BOARD_ID = 42

def get_lead_time_data(jira_client, project_key, days=90):
    """获取已完成工单的 Lead Time 分解"""
    jql = (
        f'project = {project_key} '
        f'AND status = Done '
        f'AND resolved >= -{days}d '
        f'ORDER BY resolved DESC'
    )
    issues = jira_client.search_issues(jql, maxResults=200)
    results = []

    for issue in issues:
        changelog = issue.changelog
        history = {}
        for hist in changelog.histories:
            for item in hist.items:
                if item.field == "status":
                    ts = datetime.strptime(hist.created[:19], "%Y-%m-%dT%H:%M:%S")
                    history.setdefault(item.toString, []).append(ts)

        # 关键时间节点
        ready_time = history.get("Ready", [None])[0]
        in_progress_time = history.get("In Progress", [None])[0]
        in_review_time = history.get("In Review", [None])[0]
        testing_time = history.get("Testing", [None])[0]
        done_time = history.get("Done", [None])[0]

        if ready_time and done_time:
            lead_time = (done_time - ready_time).total_seconds() / 3600
            dev_time = (in_review_time - in_progress_time).total_seconds() / 3600 if in_progress_time and in_review_time else None
            wait_review = (testing_time - in_review_time).total_seconds() / 3600 if in_review_time and testing_time else None

            results.append({
                "key": issue.key,
                "lead_time_hours": round(lead_time, 1),
                "dev_time_hours": round(dev_time, 1) if dev_time else None,
                "wait_review_hours": round(wait_review, 1) if wait_review else None,
            })

    return results

def get_current_wip(jira_client, project_key):
    """获取当前 WIP（进行中的工单数）"""
    jql = (
        f'project = {project_key} '
        f'AND status in ("In Progress", "In Review", "Testing") '
        f'AND assignee is not EMPTY'
    )
    issues = jira_client.search_issues(jql, maxResults=500)
    wip_by_status = {}
    for issue in issues:
        wip_by_status[issue.fields.status.name] = wip_by_status.get(issue.fields.status.name, 0) + 1

    return {
        "total_wip": len(issues),
        "by_status": wip_by_status,
    }

if __name__ == "__main__":
    jira = JIRA(JIRA_URL, basic_auth=(USERNAME, API_TOKEN))

    # Lead Time 分析
    lt_data = get_lead_time_data(jira, PROJECT_KEY)
    lt_values = [d["lead_time_hours"] for d in lt_data]
    print(f"Lead Time 均值: {sum(lt_values)/len(lt_values):.1f}h")
    print(f"Lead Time P50:  {sorted(lt_values)[len(lt_values)//2]:.1f}h")
    print(f"Lead Time P95:  {sorted(lt_values)[int(len(lt_values)*0.95)]:.1f}h")

    # WIP 分析
    wip = get_current_wip(jira, PROJECT_KEY)
    print(f"\n当前 WIP: {wip['total_wip']}")
    for status, count in wip["by_status"].items():
        print(f"  {status}: {count}")