精益思想的 AI 增强：从消除浪费到自动化改进

# 闻起来像过度设计
- 抽象层数 > 3
- 接口只有 1 个实现
- "为未来扩展预留"的代码
- 注释写着"万一以后..."

# github-wait-time.py
import requests
from datetime import datetime

def analyze_pr_wait_time(repo, days=30):
    """分析 PR 等待时间分布"""
    prs = fetch_merged_prs(repo, days)
    
    wait_times = []
    for pr in prs:
        created = pr['created_at']
        first_review = get_first_review(pr)
        if first_review:
            wait = (first_review - created).total_seconds() / 3600
            wait_times.append(wait)
    
    return {
        'p50': percentile(wait_times, 50),
        'p90': percentile(wait_times, 90),
        'blocking_prs': count_prs_wait_over(wait_times, 24)  # 超过24小时
    }

# analyze_focus_time.py
import pandas as pd
from datetime import timedelta

def identify_focus_blocks(commits):
    """识别专注时间块（>90分钟无中断）"""
    df = pd.DataFrame(commits, columns=['timestamp'])
    df['gap'] = df['timestamp'].diff()
    
    # 超过30分钟的间隔视为上下文切换
    switches = df[df['gap'] > timedelta(minutes=30)]
    
    return {
        'avg_focus_block_minutes': calculate_focus_blocks(df),
        'context_switches_per_day': len(switches) / num_days,
        'focus_score': calculate_focus_score(df)  # 0-100
    }

# churn_analysis.py
def calculate_code_churn(repo_path, weeks=4):
    """计算代码变动率，识别热点文件"""
    result = subprocess.run(
        ['git', 'log', '--since=f'{weeks} weeks ago'', '--pretty=format:', 
         '--name-only', '--diff-filter=ACM'],
        capture_output=True, text=True, cwd=repo_path
    )
    
    files = result.stdout.strip().split('\n')
    file_counts = Counter(f for f in files if f and not f.startswith('test/'))
    
    # 高变动文件 = 可能的稳定性问题
    hotspots = {f: c for f, c in file_counts.most_common(20) if c > 5}
    return hotspots

# risk_score.py - 预测 PR 风险
def calculate_pr_risk(pr_data):
    """基于多个信号计算风险分数"""
    risk_factors = {
        'lines_changed': min(pr_data['additions'] + pr_data['deletions'], 500) / 500 * 25,
        'files_touched': min(len(pr_data['files']), 20) / 20 * 20,
        'test_coverage_drop': pr_data.get('coverage_delta', 0) * -1 * 20,
        'author_experience': (1 - pr_data['author_repo_experience']) * 15,
        'complexity_increase': pr_data.get('cyclomatic_increase', 0) * 10,
        'hotspot_touch': 10 if any(f in hotspots for f in pr_data['files']) else 0
    }
    
    return sum(risk_factors.values()), risk_factors

# weekly_insights.py
import json
from datetime import datetime, timedelta

def generate_weekly_report(metrics_dir):
    # 加载最近7天数据
    week_data = load_last_n_days(metrics_dir, 7)
    
    insights = {
        'trends': calculate_trends(week_data),
        'anomalies': detect_anomalies(week_data),
        'recommendations': generate_recommendations(week_data)
    }
    
    return format_report(insights)

def generate_recommendations(data):
    recs = []
    
    # 基于数据的自动建议
    if data['avg_review_time'] > 8:
        recs.append({
            'type': 'waiting',
            'severity': 'high',
            'message': 'PR 平均等待时间超过8小时，建议设置 review 轮值',
            'action': '实施 "Review Buddy" 轮值制度'
        })
    
    if data['cycle_time_trend'] > 1.2:  # 比上周慢20%
        recs.append({
            'type': 'flow',
            'severity': 'medium', 
            'message': '交付周期比上周延长 20%',
            'action': '检查在制品数量，可能 WIP 过多'
        })
    
    if len(data['stale_branches']) > 3:
        recs.append({
            'type': 'inventory',
            'severity': 'low',
            'message': f'发现 {len(data["stale_branches"])} 个过期分支',
            'action': '周五下午清理分支（可以脚本化）'
        })
    
    return recs

# value_stream_mapper.py
from dataclasses import dataclass
from typing import List
import asyncio

@dataclass
class Stage:
    name: str
    avg_time: timedelta
    wait_time: timedelta
    touch_time: timedelta
    completion_rate: float

async def map_value_stream(repo: str) -> List[Stage]:
    """自动绘制价值流图"""
    
    stages = [
        Stage("需求澄清", await analyze_requirement_stage(repo)),
        Stage("开发", await analyze_development_stage(repo)),
        Stage("Code Review", await analyze_review_stage(repo)),
        Stage("测试", await analyze_test_stage(repo)),
        Stage("部署", await analyze_deploy_stage(repo)),
        Stage("用户激活", await analyze_activation_stage(repo))
    ]
    
    return stages

def calculate_flow_efficiency(stages: List[Stage]) -> float:
    """计算流效率 = 价值增加时间 / 总周期时间"""
    total_time = sum(s.avg_time for s in stages)
    touch_time = sum(s.touch_time for s in stages)
    return touch_time / total_time if total_time else 0

# bottleneck_detector.py
def identify_bottleneck(stages: List[Stage]) -> dict:
    """识别价值流中的瓶颈"""
    
    # 方法1：最长等待时间
    max_wait = max(stages, key=lambda s: s.wait_time)
    
    # 方法2：最低完成率
    lowest_completion = min(stages, key=lambda s: s.completion_rate)
    
    # 方法3：累积排队（Little's Law）
    wip_per_stage = calculate_wip(stages)
    
    return {
        'primary_bottleneck': max_wait.name,
        'wait_time_hours': max_wait.wait_time.total_seconds() / 3600,
        'recommendation': generate_bottleneck_fix(max_wait),
        'secondary_issues': [
            s.name for s in stages 
            if s.completion_rate < 0.9
        ]
    }

def generate_bottleneck_fix(stage: Stage) -> str:
    fixes = {
        'Code Review': [
            '实施小批量 PR（<200行）',
            '设置 review SLAs（4小时）',
            '引入 AI 预审过滤明显问题'
        ],
        '测试': [
            '并行化测试套件',
            '识别并隔离 flaky tests',
            '智能测试选择（只跑相关测试）'
        ],
        '部署': [
            '蓝绿部署减少等待',
            '自动化回滚机制',
            '特性开关替代分支'
        ]
    }
    return fixes.get(stage.name, ['分析具体阻塞原因'])

# diagnosis_runner.py
from lean_ai import DiagnosticEngine

def run_diagnosis():
    engine = DiagnosticEngine()
    
    # 运行所有诊断
    results = {
        'waste_analysis': engine.analyze_7_wastes(),
        'flow_efficiency': engine.calculate_flow_efficiency(),
        'bottleneck_map': engine.map_bottlenecks(),
        'quality_gates': engine.assess_quality_gates()
    }
    
    # 生成改进建议优先级
    priorities = engine.prioritize_improvements(results)
    
    return {
        'current_state': results,
        'top_3_improvements': priorities[:3],
        'expected_impact': engine.simulate_improvements(priorities[:3])
    }