6.4.1 DORA 四指标 + SPACE 框架

# dora_github_collector.py
# 从 GitHub API 采集 DORA 四指标

import os
import requests
from datetime import datetime, timedelta
from typing import List, Dict

class GitHubDORACollector:
    def __init__(self, token: str, owner: str, repo: str):
        self.token = token
        self.owner = owner
        self.repo = repo
        self.base_url = f"https://api.github.com/repos/{owner}/{repo}"
        self.headers = {
            "Authorization": f"Bearer {token}",
            "Accept": "application/vnd.github+json",
        }

    def collect_deployments(self, since_days: int = 30) -> List[Dict]:
        """采集部署事件 → 部署频率 + 变更前置时间"""
        url = f"{self.base_url}/deployments"
        params = {
            "per_page": 100,
            "environment": "production",
        }
        resp = requests.get(url, headers=self.headers, params=params)
        resp.raise_for_status()
        deployments = resp.json()

        # 过滤近 N 天
        cutoff = datetime.now() - timedelta(days=since_days)
        return [
            {
                "id": d["id"],
                "sha": d["sha"],
                "created_at": d["created_at"],
                "environment": d["environment"],
            }
            for d in deployments
            if datetime.fromisoformat(d["created_at"].rstrip("Z")) > cutoff
        ]

    def collect_failed_workflows(self, since_days: int = 30) -> int:
        """采集失败的 CI Workflow → 变更失败率"""
        url = f"{self.base_url}/actions/runs"
        params = {
            "per_page": 100,
            "status": "failure",
            "created": f">{since_days}d",
        }
        resp = requests.get(url, headers=self.headers, params=params)
        resp.raise_for_status()
        return len(resp.json()["workflow_runs"])

    def collect_lead_time(self, deployment_sha: str) -> float:
        """采集变更前置时间(commit → prod 部署)"""
        # 拿 commit 时间
        commit_url = f"{self.base_url}/commits/{deployment_sha}"
        resp = requests.get(commit_url, headers=self.headers)
        resp.raise_for_status()
        commit_time = datetime.fromisoformat(
            resp.json()["commit"]["author"]["date"].rstrip("Z")
        )
        # 拿 deployment 时间
        deploy_time = datetime.now()  # 实际从 deployment API 取
        return (deploy_time - commit_time).total_seconds() / 3600  # 小时

    def calculate_dora_metrics(self) -> Dict:
        """计算 DORA 四指标"""
        since = 30
        deployments = self.collect_deployments(since)
        failed_count = self.collect_failed_workflows(since)

        deploy_freq = len(deployments) / since  # 次/天
        cfr = failed_count / max(len(deployments), 1) * 100  # %

        lead_times = [self.collect_lead_time(d["sha"]) for d in deployments[:10]]
        avg_lead_time = sum(lead_times) / len(lead_times) if lead_times else 0

        # MTTR 需要 incident 数据,这里用 placeholder
        return {
            "deployment_frequency_per_day": round(deploy_freq, 2),
            "change_failure_rate_pct": round(cfr, 2),
            "lead_time_for_changes_hours": round(avg_lead_time, 2),
            "mttr_minutes": None,  # 需 PagerDuty/OpsGenie 数据
        }

if __name__ == "__main__":
    collector = GitHubDORACollector(
        token=os.environ["GITHUB_TOKEN"],
        owner="myorg",
        repo="myrepo",
    )
    metrics = collector.calculate_dora_metrics()
    print(metrics)

# dora_gitlab_collector.py
# 从 GitLab API 采集 DORA 四指标

import os
import requests
from datetime import datetime, timedelta

class GitLabDORACollector:
    def __init__(self, token: str, project_id: str, gitlab_url: str = "https://gitlab.com"):
        self.token = token
        self.project_id = project_id
        self.base_url = f"{gitlab_url}/api/v4/projects/{project_id}"
        self.headers = {"PRIVATE-TOKEN": token}

    def get_merged_merge_requests(self, since_days: int = 30) -> list:
        """采集合并的 MR"""
        url = f"{self.base_url}/merge_requests"
        params = {
            "state": "merged",
            "updated_after": (datetime.now() - timedelta(days=since_days)).isoformat(),
            "per_page": 100,
        }
        resp = requests.get(url, headers=self.headers, params=params)
        resp.raise_for_status()
        return resp.json()

    def get_deployment_frequency(self, since_days: int = 30) -> float:
        """GitLab 内置 DORA API"""
        # GitLab 14.5+ 提供原生 DORA API
        url = f"{self.base_url}/analytics"
        params = {
            "metric": "deployment_frequency",
            "interval": "daily",
            "start_date": (datetime.now() - timedelta(days=since_days)).date().isoformat(),
            "end_date": datetime.now().date().isoformat(),
        }
        resp = requests.get(url, headers=self.headers, params=params)
        resp.raise_for_status()
        data = resp.json()
        return sum(d["value"] for d in data) / since_days  # 次/天

    def get_lead_time(self, since_days: int = 30) -> float:
        """GitLab 原生 Lead Time API"""
        url = f"{self.base_url}/analytics"
        params = {
            "metric": "lead_time",
            "interval": "daily",
            "start_date": (datetime.now() - timedelta(days=since_days)).date().isoformat(),
            "end_date": datetime.now().date().isoformat(),
        }
        resp = requests.get(url, headers=self.headers, params=params)
        resp.raise_for_status()
        data = resp.json()
        return sum(d["value"] for d in data) / len(data) if data else 0  # 天

    def calculate_dora(self) -> dict:
        """完整 DORA 四指标"""
        mrs = self.get_merged_merge_requests()
        total = len(mrs)
        # 失败 MR(被 revert 或有 incident 关联)
        failed = sum(1 for m in mrs if m.get("labels", []).count("incident") > 0)
        return {
            "deployment_frequency_per_day": self.get_deployment_frequency(),
            "lead_time_days": self.get_lead_time(),
            "change_failure_rate_pct": (failed / total * 100) if total > 0 else 0,
            "mttr_minutes": self._get_mttr(),
        }

    def _get_mttr(self) -> float:
        """MTTR:从 incident 创建到解决"""
        url = f"{self.base_url}/incidents"
        params = {"state": "closed", "per_page": 100}
        resp = requests.get(url, headers=self.headers, params=params)
        resp.raise_for_status()
        incidents = resp.json()
        if not incidents:
            return 0
        total_min = 0
        for inc in incidents:
            start = datetime.fromisoformat(inc["created_at"].rstrip("Z"))
            end = datetime.fromisoformat(inc["closed_at"].rstrip("Z"))
            total_min += (end - start).total_seconds() / 60
        return total_min / len(incidents)

# space_survey.py
# SPACE 框架季度调研

from dataclasses import dataclass, asdict
from typing import List
import json

@dataclass
class SPACESurvey:
    satisfaction_score: int  # 1-5
    performance_score: int   # 1-5
    activity_score: int      # 1-5
    communication_score: int # 1-5
    efficiency_score: int    # 1-5
    enps: int                # -100 ~ 100

# 调研问卷题目
QUESTIONS = {
    "satisfaction": [
        "我愿意推荐朋友加入这个团队(1-5)",
        "我对当前使用的工具满意(1-5)",
        "我对工作流程满意(1-5)",
        "我能在合理时间内完成任务(1-5)",
    ],
    "performance": [
        "我们的产出对客户有价值(1-5)",
        "我们的部署频率让我满意(1-5)",
        "我们的故障恢复速度让我满意(1-5)",
    ],
    "activity": [
        "我的 PR 通常有意义(1-5)",
        "我参与代码评审(1-5)",
        "我贡献文档和知识分享(1-5)",
    ],
    "communication": [
        "团队会议占比 < 30%(1-5)",
        "Code Review 反馈及时(1-5)",
        "知识在团队内有效共享(1-5)",
    ],
    "efficiency": [
        "我的专注时间 > 4 小时/天(1-5)",
        "我的 Context Switch < 3 次/天(1-5)",
        "CI 等待 < 10 分钟(1-5)",
    ],
}

def calculate_space_score(responses: dict) -> SPACESurvey:
    """计算 SPACE 五维度平均分"""
    result = {}
    for dim, qs in QUESTIONS.items():
        scores = [responses[f"q{i}"] for i, q in enumerate(qs)]
        result[f"{dim}_score"] = int(sum(scores) / len(scores))
    result["enps"] = responses.get("enps", 0)
    return SPACESurvey(**result)

# 输出 JSON
if __name__ == "__main__":
    sample = {f"q{i}": 4 for i in range(15)}
    sample["enps"] = 35
    score = calculate_space_score(sample)
    print(json.dumps(asdict(score), indent=2, ensure_ascii=False))

# anti_pattern_personal_loc.py
# 反面教材:度量个人代码行数

import subprocess
from collections import Counter

def count_lines_per_author(repo_path: str, since: str = "1 month ago") -> dict:
    """统计每个作者的代码行数"""
    cmd = [
        "git", "-C", repo_path,
        "log", "--since", since,
        "--pretty=format:%ae",
        "--numstat",
    ]
    result = subprocess.run(cmd, capture_output=True, text=True)
    lines = result.stdout.strip().split("\n")

    stats = Counter()
    current_author = None
    for line in lines:
        if not line.strip():
            continue
        if "\t" not in line:
            current_author = line.strip()
        else:
            parts = line.split("\t")
            added = int(parts[0]) if parts[0] != "-" else 0
            deleted = int(parts[1]) if parts[1] != "-" else 0
            stats[current_author] += added + deleted

    return dict(stats.most_common())

# 输出:
# {
#   "alice@x.com": 15420,  # 行数冠军
#   "bob@x.com": 8230,
#   "carol@x.com": 4120,
# }
#
# 问题:
#  1. 鼓励把代码写长(拆分、无意义注释)
#  2. 惩罚代码精简的人(优秀重构 = 行数下降)
#  3. 鼓励删别人的代码而不是共建
#  4. 与业务价值完全脱节
#
# 结论:❌ 永远不要把 "个人行数" 作为 KPI

# engineer_experience_etl.py
# 推荐:度量工程师体验(团队聚合)

import pandas as pd
from datetime import datetime, timedelta

class EngineerExperienceETL:
    def __init__(self, github_token: str, pagerduty_token: str):
        self.gh_token = github_token
        self.pd_token = pagerduty_token

    def calculate_focus_time(self, calendar_data: list) -> float:
        """工程师每日专注时间(基于日历数据)"""
        total_focus_minutes = 0
        for day in calendar_data:
            meeting_minutes = day["meeting_minutes"]
            total_work_minutes = 8 * 60  # 8 小时工作
            focus = max(0, total_work_minutes - meeting_minutes - day["slack_minutes"])
            total_focus_minutes += focus
        return total_focus_minutes / len(calendar_data) / 60  # 小时

    def calculate_context_switches(self, pr_data: list) -> int:
        """每日任务切换次数"""
        # 统计每日 PR 评论 + Issue 切换数
        switches_per_day = {}
        for activity in pr_data:
            day = activity["date"]
            switches_per_day[day] = switches_per_day.get(day, 0) + 1
        return int(sum(switches_per_day.values()) / len(switches_per_day))

    def calculate_oncall_burden(self, incident_data: list, team_size: int) -> dict:
        """On-Call 负担"""
        total_hours = sum(i["duration_hours"] for i in incident_data)
        after_hours = sum(
            i["duration_hours"] for i in incident_data
            if i["is_after_hours"]
        )
        return {
            "total_oncall_hours_per_engineer_month": total_hours / team_size / 30 * 24,
            "after_hours_pct": after_hours / total_hours * 100 if total_hours > 0 else 0,
        }

    def export_team_kpi(self, output_path: str):
        """输出团队级 KPI 看板"""
        kpis = {
            "deploy_freq_per_day": 5.2,
            "lead_time_hours": 2.3,
            "change_failure_rate_pct": 4.5,
            "mttr_minutes": 18,
            "team_nps": 42,
            "avg_focus_hours_per_day": 4.5,
            "avg_context_switches_per_day": 2.1,
            "after_hours_oncall_pct": 8,
        }
        df = pd.DataFrame([kpis])
        df.to_csv(output_path, index=False)
        return kpis