8.5 継続的改善 ​

#!/usr/bin/env python3
"""
Zabbix監視効果測定システム
"""
import json
import sqlite3
import datetime
from dataclasses import dataclass
from typing import List, Dict, Any
import requests
import pandas as pd
import matplotlib.pyplot as plt

@dataclass
class MonitoringMetrics:
    """監視メトリクスデータクラス"""
    date: datetime.datetime
    mttr: float  # Mean Time To Resolution
    mtbf: float  # Mean Time Between Failures
    availability: float
    alert_accuracy: float
    false_positive_rate: float

class MonitoringEffectivenessAnalyzer:
    """監視効果分析器"""
    
    def __init__(self, zabbix_url: str, api_token: str, db_path: str = "monitoring_metrics.db"):
        self.zabbix_url = zabbix_url
        self.api_token = api_token
        self.db_path = db_path
        self.init_database()
    
    def init_database(self):
        """データベース初期化"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS monitoring_metrics (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                date TEXT NOT NULL,
                mttr REAL,
                mtbf REAL,
                availability REAL,
                alert_accuracy REAL,
                false_positive_rate REAL,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS incidents (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                incident_id TEXT UNIQUE,
                start_time TIMESTAMP,
                end_time TIMESTAMP,
                severity INTEGER,
                resolved BOOLEAN DEFAULT FALSE,
                false_positive BOOLEAN DEFAULT FALSE,
                resolution_time REAL,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        
        conn.commit()
        conn.close()
    
    def get_zabbix_data(self, method: str, params: Dict[str, Any]) -> List[Dict[str, Any]]:
        """Zabbix APIからデータを取得"""
        headers = {
            'Content-Type': 'application/json',
            'Authorization': f'Bearer {self.api_token}'
        }
        
        payload = {
            'jsonrpc': '2.0',
            'method': method,
            'params': params,
            'id': 1
        }
        
        response = requests.post(
            f'{self.zabbix_url}/api_jsonrpc.php',
            headers=headers,
            data=json.dumps(payload)
        )
        
        if response.status_code == 200:
            result = response.json()
            return result.get('result', [])
        return []
    
    def calculate_mttr(self, start_date: datetime.datetime, end_date: datetime.datetime) -> float:
        """平均修復時間（MTTR）を計算"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            SELECT AVG(resolution_time) FROM incidents
            WHERE start_time BETWEEN ? AND ?
            AND resolved = TRUE
            AND false_positive = FALSE
        ''', (start_date.isoformat(), end_date.isoformat()))
        
        result = cursor.fetchone()[0]
        conn.close()
        
        return result if result else 0.0
    
    def calculate_mtbf(self, start_date: datetime.datetime, end_date: datetime.datetime) -> float:
        """平均故障間隔（MTBF）を計算"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            SELECT COUNT(*) FROM incidents
            WHERE start_time BETWEEN ? AND ?
            AND resolved = TRUE
            AND false_positive = FALSE
        ''', (start_date.isoformat(), end_date.isoformat()))
        
        incident_count = cursor.fetchone()[0]
        conn.close()
        
        if incident_count > 1:
            period_hours = (end_date - start_date).total_seconds() / 3600
            return period_hours / (incident_count - 1)
        
        return 0.0
    
    def calculate_availability(self, host_ids: List[str], start_date: datetime.datetime, end_date: datetime.datetime) -> float:
        """システム可用性を計算"""
        total_uptime = 0
        total_time = (end_date - start_date).total_seconds()
        
        for host_id in host_ids:
            # Zabbix APIからホストの稼働状況を取得
            events = self.get_zabbix_data('event.get', {
                'hostids': [host_id],
                'time_from': int(start_date.timestamp()),
                'time_till': int(end_date.timestamp()),
                'selectHosts': 'extend'
            })
            
            # 稼働時間を計算（簡略化された例）
            downtime = sum(
                int(event.get('r_clock', 0)) - int(event.get('clock', 0))
                for event in events
                if event.get('r_clock')
            )
            
            uptime = total_time - downtime
            total_uptime += uptime
        
        if host_ids:
            average_uptime = total_uptime / len(host_ids)
            return (average_uptime / total_time) * 100
        
        return 0.0
    
    def calculate_alert_accuracy(self, start_date: datetime.datetime, end_date: datetime.datetime) -> float:
        """アラート精度を計算"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            SELECT
                COUNT(*) as total_alerts,
                SUM(CASE WHEN false_positive = FALSE THEN 1 ELSE 0 END) as true_alerts
            FROM incidents
            WHERE start_time BETWEEN ? AND ?
        ''', (start_date.isoformat(), end_date.isoformat()))
        
        result = cursor.fetchone()
        total_alerts, true_alerts = result
        conn.close()
        
        if total_alerts > 0:
            return (true_alerts / total_alerts) * 100
        
        return 0.0
    
    def generate_effectiveness_report(self, start_date: datetime.datetime, end_date: datetime.datetime, host_ids: List[str]) -> MonitoringMetrics:
        """監視効果レポート生成"""
        mttr = self.calculate_mttr(start_date, end_date)
        mtbf = self.calculate_mtbf(start_date, end_date)
        availability = self.calculate_availability(host_ids, start_date, end_date)
        alert_accuracy = self.calculate_alert_accuracy(start_date, end_date)
        false_positive_rate = 100 - alert_accuracy
        
        metrics = MonitoringMetrics(
            date=datetime.datetime.now(),
            mttr=mttr,
            mtbf=mtbf,
            availability=availability,
            alert_accuracy=alert_accuracy,
            false_positive_rate=false_positive_rate
        )
        
        # データベースに保存
        self.save_metrics(metrics)
        
        return metrics
    
    def save_metrics(self, metrics: MonitoringMetrics):
        """メトリクスをデータベースに保存"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            INSERT INTO monitoring_metrics 
            (date, mttr, mtbf, availability, alert_accuracy, false_positive_rate)
            VALUES (?, ?, ?, ?, ?, ?)
        ''', (
            metrics.date.isoformat(),
            metrics.mttr,
            metrics.mtbf,
            metrics.availability,
            metrics.alert_accuracy,
            metrics.false_positive_rate
        ))
        
        conn.commit()
        conn.close()
    
    def generate_trend_analysis(self, months: int = 6) -> Dict[str, Any]:
        """トレンド分析レポート生成"""
        conn = sqlite3.connect(self.db_path)
        
        # 過去のメトリクス取得
        df = pd.read_sql_query('''
            SELECT * FROM monitoring_metrics
            WHERE date >= date('now', '-{} month')
            ORDER BY date
        '''.format(months), conn)
        
        conn.close()
        
        if df.empty:
            return {"error": "No data available for trend analysis"}
        
        # トレンド計算
        trends = {
            'mttr_trend': self._calculate_trend(df['mttr'].tolist()),
            'availability_trend': self._calculate_trend(df['availability'].tolist()),
            'alert_accuracy_trend': self._calculate_trend(df['alert_accuracy'].tolist()),
            'improvement_recommendations': self._generate_recommendations(df)
        }
        
        # グラフ生成
        self._generate_trend_charts(df)
        
        return trends
    
    def _calculate_trend(self, values: List[float]) -> str:
        """トレンドを計算"""
        if len(values) < 2:
            return "insufficient_data"
        
        recent_avg = sum(values[-3:]) / len(values[-3:])
        historical_avg = sum(values[:-3]) / len(values[:-3]) if len(values) > 3 else values[0]
        
        change_rate = ((recent_avg - historical_avg) / historical_avg) * 100 if historical_avg != 0 else 0
        
        if change_rate > 5:
            return "improving"
        elif change_rate < -5:
            return "declining"
        else:
            return "stable"
    
    def _generate_recommendations(self, df: pd.DataFrame) -> List[str]:
        """改善推奨事項生成"""
        recommendations = []
        
        # MTTR改善
        if df['mttr'].iloc[-1] > df['mttr'].mean():
            recommendations.append("MTTRが高いため、自動修復機能の実装を検討してください")
        
        # 可用性改善
        if df['availability'].iloc[-1] < 99.0:
            recommendations.append("可用性向上のため、冗長化の実装を検討してください")
        
        # アラート精度改善
        if df['alert_accuracy'].iloc[-1] < 85.0:
            recommendations.append("アラート精度向上のため、閾値の見直しを行ってください")
        
        return recommendations
    
    def _generate_trend_charts(self, df: pd.DataFrame):
        """トレンドチャート生成"""
        fig, axes = plt.subplots(2, 2, figsize=(15, 10))
        
        # MTTR トレンド
        axes[0, 0].plot(df['date'], df['mttr'])
        axes[0, 0].set_title('MTTR Trend')
        axes[0, 0].set_ylabel('Hours')
        
        # 可用性トレンド
        axes[0, 1].plot(df['date'], df['availability'])
        axes[0, 1].set_title('Availability Trend')
        axes[0, 1].set_ylabel('Percentage')
        
        # アラート精度トレンド
        axes[1, 0].plot(df['date'], df['alert_accuracy'])
        axes[1, 0].set_title('Alert Accuracy Trend')
        axes[1, 0].set_ylabel('Percentage')
        
        # 誤検知率トレンド
        axes[1, 1].plot(df['date'], df['false_positive_rate'])
        axes[1, 1].set_title('False Positive Rate Trend')
        axes[1, 1].set_ylabel('Percentage')
        
        plt.tight_layout()
        plt.savefig('monitoring_trends.png')
        plt.close()

# 使用例
if __name__ == "__main__":
    analyzer = MonitoringEffectivenessAnalyzer(
        zabbix_url="https://zabbix.example.com",
        api_token="your_api_token"
    )
    
    # 月次レポート生成
    end_date = datetime.datetime.now()
    start_date = end_date - datetime.timedelta(days=30)
    host_ids = ["10001", "10002", "10003"]
    
    metrics = analyzer.generate_effectiveness_report(start_date, end_date, host_ids)
    print(f"Monthly Metrics: MTTR={metrics.mttr:.2f}h, Availability={metrics.availability:.2f}%")
    
    # トレンド分析
    trends = analyzer.generate_trend_analysis(6)
    print("Trend Analysis:", json.dumps(trends, indent=2))

#!/usr/bin/env python3
"""
KPI計算とレポート生成システム
"""
import json
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase
from email import encoders
import yaml
from datetime import datetime, timedelta
from typing import Dict, List, Any
import jinja2

class KPICalculator:
    """KPI計算器"""
    
    def __init__(self, config_path: str = "kpi_config.yaml"):
        self.config = self.load_config(config_path)
        self.metrics_analyzer = MonitoringEffectivenessAnalyzer(
            self.config['zabbix']['url'],
            self.config['zabbix']['api_token']
        )
    
    def load_config(self, config_path: str) -> Dict[str, Any]:
        """設定ファイル読み込み"""
        with open(config_path, 'r', encoding='utf-8') as f:
            return yaml.safe_load(f)
    
    def calculate_all_kpis(self, start_date: datetime, end_date: datetime) -> Dict[str, Any]:
        """全KPI計算"""
        kpis = {}
        
        # 基本メトリクス取得
        host_ids = self.config.get('monitored_hosts', [])
        base_metrics = self.metrics_analyzer.generate_effectiveness_report(
            start_date, end_date, host_ids
        )
        
        # 可用性メトリクス
        kpis['availability'] = {
            'system_uptime': {
                'value': base_metrics.availability,
                'target': self.config['kpi_targets']['availability']['system_uptime'],
                'status': self._evaluate_kpi(
                    base_metrics.availability,
                    self.config['kpi_targets']['availability']['system_uptime']
                )
            }
        }
        
        # パフォーマンスメトリクス
        kpis['performance'] = {
            'mttr': {
                'value': base_metrics.mttr,
                'target': self.config['kpi_targets']['performance']['mttr'],
                'status': self._evaluate_kpi(
                    base_metrics.mttr,
                    self.config['kpi_targets']['performance']['mttr'],
                    lower_is_better=True
                )
            },
            'mtbf': {
                'value': base_metrics.mtbf,
                'target': self.config['kpi_targets']['performance']['mtbf'],
                'status': self._evaluate_kpi(
                    base_metrics.mtbf,
                    self.config['kpi_targets']['performance']['mtbf']
                )
            }
        }
        
        # 品質メトリクス
        kpis['quality'] = {
            'alert_accuracy': {
                'value': base_metrics.alert_accuracy,
                'target': self.config['kpi_targets']['quality']['alert_accuracy'],
                'status': self._evaluate_kpi(
                    base_metrics.alert_accuracy,
                    self.config['kpi_targets']['quality']['alert_accuracy']
                )
            },
            'false_positive_rate': {
                'value': base_metrics.false_positive_rate,
                'target': self.config['kpi_targets']['quality']['false_positive_rate'],
                'status': self._evaluate_kpi(
                    base_metrics.false_positive_rate,
                    self.config['kpi_targets']['quality']['false_positive_rate'],
                    lower_is_better=True
                )
            }
        }
        
        # 運用メトリクス
        kpis['operational'] = {
            'automation_rate': {
                'value': self._calculate_automation_rate(),
                'target': self.config['kpi_targets']['operational']['automation_rate'],
                'status': 'meeting'  # 簡略化
            }
        }
        
        return kpis
    
    def _evaluate_kpi(self, actual: float, target: Dict[str, Any], lower_is_better: bool = False) -> str:
        """KPI評価"""
        target_value = float(target['target'].strip('%<> '))
        warning_threshold = float(target['threshold_warning'].strip('%<> '))
        critical_threshold = float(target['threshold_critical'].strip('%<> '))
        
        if lower_is_better:
            if actual <= target_value:
                return 'exceeding'
            elif actual <= warning_threshold:
                return 'meeting'
            elif actual <= critical_threshold:
                return 'warning'
            else:
                return 'critical'
        else:
            if actual >= target_value:
                return 'exceeding'
            elif actual >= warning_threshold:
                return 'meeting'
            elif actual >= critical_threshold:
                return 'warning'
            else:
                return 'critical'
    
    def _calculate_automation_rate(self) -> float:
        """自動化率計算（簡略化された例）"""
        # 実際の実装では、自動化されたタスクの割合を計算
        automated_tasks = 15
        total_tasks = 20
        return (automated_tasks / total_tasks) * 100
    
    def generate_kpi_dashboard(self, kpis: Dict[str, Any]) -> str:
        """KPIダッシュボードHTML生成"""
        template_str = """
        <!DOCTYPE html>
        <html>
        <head>
            <title>Zabbix KPI Dashboard</title>
            <style>
                body { font-family: Arial, sans-serif; margin: 20px; }
                .kpi-card { 
                    border: 1px solid #ddd; 
                    border-radius: 8px; 
                    padding: 15px; 
                    margin: 10px; 
                    display: inline-block; 
                    width: 250px; 
                    vertical-align: top;
                }
                .exceeding { border-left: 5px solid #4CAF50; }
                .meeting { border-left: 5px solid #2196F3; }
                .warning { border-left: 5px solid #FF9800; }
                .critical { border-left: 5px solid #F44336; }
                .metric-value { font-size: 24px; font-weight: bold; }
                .metric-target { color: #666; font-size: 14px; }
            </style>
        </head>
        <body>
            <h1>Zabbix監視システム KPIダッシュボード</h1>
            <p>生成日時: {{ generation_time }}</p>
            
            <h2>可用性メトリクス</h2>
            {% for metric_name, metric_data in kpis.availability.items() %}
            <div class="kpi-card {{ metric_data.status }}">
                <h3>{{ metric_name }}</h3>
                <div class="metric-value">{{ "%.2f"|format(metric_data.value) }}%</div>
                <div class="metric-target">目標: {{ metric_data.target.target }}</div>
                <div>ステータス: {{ metric_data.status }}</div>
            </div>
            {% endfor %}
            
            <h2>パフォーマンスメトリクス</h2>
            {% for metric_name, metric_data in kpis.performance.items() %}
            <div class="kpi-card {{ metric_data.status }}">
                <h3>{{ metric_name }}</h3>
                <div class="metric-value">{{ "%.2f"|format(metric_data.value) }}</div>
                <div class="metric-target">目標: {{ metric_data.target.target }}</div>
                <div>ステータス: {{ metric_data.status }}</div>
            </div>
            {% endfor %}
            
            <h2>品質メトリクス</h2>
            {% for metric_name, metric_data in kpis.quality.items() %}
            <div class="kpi-card {{ metric_data.status }}">
                <h3>{{ metric_name }}</h3>
                <div class="metric-value">{{ "%.2f"|format(metric_data.value) }}%</div>
                <div class="metric-target">目標: {{ metric_data.target.target }}</div>
                <div>ステータス: {{ metric_data.status }}</div>
            </div>
            {% endfor %}
        </body>
        </html>
        """
        
        template = jinja2.Template(template_str)
        return template.render(
            kpis=kpis,
            generation_time=datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        )
    
    def send_kpi_report(self, kpis: Dict[str, Any], recipients: List[str]):
        """KPIレポートをメール送信"""
        # HTML レポート生成
        html_report = self.generate_kpi_dashboard(kpis)
        
        # メール作成
        msg = MIMEMultipart('alternative')
        msg['Subject'] = f"Zabbix KPIレポート - {datetime.now().strftime('%Y-%m-%d')}"
        msg['From'] = self.config['email']['from_address']
        msg['To'] = ', '.join(recipients)
        
        # HTML添付
        html_part = MIMEText(html_report, 'html', 'utf-8')
        msg.attach(html_part)
        
        # メール送信
        try:
            server = smtplib.SMTP(self.config['email']['smtp_server'], self.config['email']['smtp_port'])
            if self.config['email'].get('use_tls', True):
                server.starttls()
            server.login(self.config['email']['username'], self.config['email']['password'])
            server.send_message(msg)
            server.quit()
            print("KPIレポートが正常に送信されました")
        except Exception as e:
            print(f"メール送信エラー: {e}")

# 使用例
if __name__ == "__main__":
    calculator = KPICalculator()
    
    # 月次KPI計算
    end_date = datetime.now()
    start_date = end_date - timedelta(days=30)
    
    kpis = calculator.calculate_all_kpis(start_date, end_date)
    
    # ダッシュボード生成
    dashboard_html = calculator.generate_kpi_dashboard(kpis)
    with open('kpi_dashboard.html', 'w', encoding='utf-8') as f:
        f.write(dashboard_html)
    
    # レポート送信
    recipients = ['[email protected]', '[email protected]']
    calculator.send_kpi_report(kpis, recipients)

#!/usr/bin/env python3
"""
自動監視レビューシステム
"""
import json
import yaml
from datetime import datetime, timedelta
from typing import Dict, List, Any, Tuple
from dataclasses import dataclass
import pandas as pd

@dataclass
class ReviewResult:
    """レビュー結果データクラス"""
    category: str
    metric: str
    current_value: float
    target_value: float
    status: str
    recommendations: List[str]
    risk_level: str

class MonitoringReviewSystem:
    """監視レビューシステム"""
    
    def __init__(self, config_path: str = "review_config.yaml"):
        self.config = self.load_config(config_path)
        self.metrics_analyzer = MonitoringEffectivenessAnalyzer(
            self.config['zabbix']['url'],
            self.config['zabbix']['api_token']
        )
    
    def load_config(self, config_path: str) -> Dict[str, Any]:
        """設定ファイル読み込み"""
        with open(config_path, 'r', encoding='utf-8') as f:
            return yaml.safe_load(f)
    
    def conduct_comprehensive_review(self, review_type: str = "monthly") -> Dict[str, Any]:
        """包括的レビュー実施"""
        review_config = self.config['review_types'][review_type]
        
        results = {
            'review_info': {
                'type': review_type,
                'date': datetime.now().isoformat(),
                'scope': review_config['scope'],
                'duration': review_config['duration']
            },
            'technical_assessment': self.assess_technical_aspects(),
            'operational_assessment': self.assess_operational_aspects(),
            'business_assessment': self.assess_business_aspects(),
            'overall_score': 0,
            'recommendations': [],
            'action_items': []
        }
        
        # 総合スコア計算
        results['overall_score'] = self.calculate_overall_score(results)
        
        # 統合推奨事項生成
        results['recommendations'] = self.generate_integrated_recommendations(results)
        
        # アクションアイテム生成
        results['action_items'] = self.generate_action_items(results)
        
        return results
    
    def assess_technical_aspects(self) -> Dict[str, Any]:
        """技術的側面の評価"""
        assessment = {
            'monitoring_coverage': self.assess_monitoring_coverage(),
            'alert_effectiveness': self.assess_alert_effectiveness(),
            'performance_optimization': self.assess_performance_optimization(),
            'security_compliance': self.assess_security_compliance(),
            'scalability': self.assess_scalability()
        }
        
        return assessment
    
    def assess_monitoring_coverage(self) -> ReviewResult:
        """監視カバレッジ評価"""
        # 実際の実装では、Zabbix APIから監視対象の詳細を取得
        total_hosts = 100
        monitored_hosts = 95
        coverage_rate = (monitored_hosts / total_hosts) * 100
        
        recommendations = []
        if coverage_rate < 95:
            recommendations.append("未監視ホストの監視設定を追加")
        if coverage_rate < 90:
            recommendations.append("監視対象の棚卸しと優先度設定")
        
        return ReviewResult(
            category="technical",
            metric="monitoring_coverage",
            current_value=coverage_rate,
            target_value=95.0,
            status="meeting" if coverage_rate >= 95 else "warning",
            recommendations=recommendations,
            risk_level="medium" if coverage_rate < 90 else "low"
        )
    
    def assess_alert_effectiveness(self) -> ReviewResult:
        """アラート効果評価"""
        end_date = datetime.now()
        start_date = end_date - timedelta(days=30)
        
        metrics = self.metrics_analyzer.generate_effectiveness_report(
            start_date, end_date, self.config.get('monitored_hosts', [])
        )
        
        recommendations = []
        if metrics.alert_accuracy < 85:
            recommendations.append("アラート閾値の見直しと調整")
        if metrics.false_positive_rate > 15:
            recommendations.append("誤検知パターンの分析と対策")
        
        return ReviewResult(
            category="technical",
            metric="alert_effectiveness",
            current_value=metrics.alert_accuracy,
            target_value=90.0,
            status="meeting" if metrics.alert_accuracy >= 85 else "warning",
            recommendations=recommendations,
            risk_level="high" if metrics.alert_accuracy < 75 else "medium"
        )
    
    def assess_performance_optimization(self) -> ReviewResult:
        """パフォーマンス最適化評価"""
        # Zabbix サーバーのパフォーマンス評価
        db_performance = self.evaluate_database_performance()
        server_performance = self.evaluate_server_performance()
        
        overall_performance = (db_performance + server_performance) / 2
        
        recommendations = []
        if overall_performance < 80:
            recommendations.append("データベース最適化の実施")
            recommendations.append("サーバーリソースの増強検討")
        
        return ReviewResult(
            category="technical",
            metric="performance_optimization",
            current_value=overall_performance,
            target_value=85.0,
            status="meeting" if overall_performance >= 80 else "warning",
            recommendations=recommendations,
            risk_level="medium" if overall_performance < 70 else "low"
        )
    
    def assess_operational_aspects(self) -> Dict[str, Any]:
        """運用的側面の評価"""
        assessment = {
            'process_efficiency': self.assess_process_efficiency(),
            'documentation_quality': self.assess_documentation_quality(),
            'team_skills': self.assess_team_skills(),
            'tool_utilization': self.assess_tool_utilization()
        }
        
        return assessment
    
    def assess_business_aspects(self) -> Dict[str, Any]:
        """ビジネス的側面の評価"""
        assessment = {
            'business_value': self.assess_business_value(),
            'cost_efficiency': self.assess_cost_efficiency(),
            'sla_achievement': self.assess_sla_achievement(),
            'risk_mitigation': self.assess_risk_mitigation()
        }
        
        return assessment
    
    def evaluate_database_performance(self) -> float:
        """データベースパフォーマンス評価（簡略化）"""
        # 実際の実装では、データベースメトリクスを取得
        return 85.0
    
    def evaluate_server_performance(self) -> float:
        """サーバーパフォーマンス評価（簡略化）"""
        # 実際の実装では、サーバーメトリクスを取得
        return 82.0
    
    def assess_process_efficiency(self) -> ReviewResult:
        """プロセス効率性評価"""
        automation_rate = 78.0  # 実際の実装では自動化率を計算
        
        return ReviewResult(
            category="operational",
            metric="process_efficiency",
            current_value=automation_rate,
            target_value=80.0,
            status="meeting" if automation_rate >= 75 else "warning",
            recommendations=["手動プロセスの自動化推進"] if automation_rate < 80 else [],
            risk_level="medium" if automation_rate < 70 else "low"
        )
    
    def assess_documentation_quality(self) -> ReviewResult:
        """ドキュメント品質評価"""
        # 実装省略
        return ReviewResult(
            category="operational",
            metric="documentation_quality",
            current_value=85.0,
            target_value=90.0,
            status="meeting",
            recommendations=[],
            risk_level="low"
        )
    
    def assess_team_skills(self) -> ReviewResult:
        """チームスキル評価"""
        # 実装省略
        return ReviewResult(
            category="operational",
            metric="team_skills",
            current_value=80.0,
            target_value=85.0,
            status="meeting",
            recommendations=["高度な監視技術の研修実施"],
            risk_level="medium"
        )
    
    def assess_tool_utilization(self) -> ReviewResult:
        """ツール活用度評価"""
        # 実装省略
        return ReviewResult(
            category="operational",
            metric="tool_utilization",
            current_value=75.0,
            target_value=80.0,
            status="warning",
            recommendations=["未活用機能の調査と導入"],
            risk_level="medium"
        )
    
    def assess_business_value(self) -> ReviewResult:
        """ビジネス価値評価"""
        # 実装省略
        return ReviewResult(
            category="business",
            metric="business_value",
            current_value=88.0,
            target_value=85.0,
            status="exceeding",
            recommendations=[],
            risk_level="low"
        )
    
    def assess_cost_efficiency(self) -> ReviewResult:
        """コスト効率性評価"""
        # 実装省略
        return ReviewResult(
            category="business",
            metric="cost_efficiency",
            current_value=82.0,
            target_value=80.0,
            status="meeting",
            recommendations=[],
            risk_level="low"
        )
    
    def assess_sla_achievement(self) -> ReviewResult:
        """SLA達成状況評価"""
        # 実装省略
        return ReviewResult(
            category="business",
            metric="sla_achievement",
            current_value=99.2,
            target_value=99.0,
            status="exceeding",
            recommendations=[],
            risk_level="low"
        )
    
    def assess_risk_mitigation(self) -> ReviewResult:
        """リスク軽減効果評価"""
        # 実装省略
        return ReviewResult(
            category="business",
            metric="risk_mitigation",
            current_value=86.0,
            target_value=85.0,
            status="meeting",
            recommendations=[],
            risk_level="low"
        )
    
    def calculate_overall_score(self, results: Dict[str, Any]) -> float:
        """総合スコア計算"""
        scores = []
        
        for category in ['technical_assessment', 'operational_assessment', 'business_assessment']:
            for metric_result in results[category].values():
                if hasattr(metric_result, 'current_value'):
                    scores.append(metric_result.current_value)
        
        return sum(scores) / len(scores) if scores else 0.0
    
    def generate_integrated_recommendations(self, results: Dict[str, Any]) -> List[str]:
        """統合推奨事項生成"""
        all_recommendations = []
        
        for category in ['technical_assessment', 'operational_assessment', 'business_assessment']:
            for metric_result in results[category].values():
                if hasattr(metric_result, 'recommendations'):
                    all_recommendations.extend(metric_result.recommendations)
        
        # 重複排除と優先度付け
        unique_recommendations = list(set(all_recommendations))
        return sorted(unique_recommendations)
    
    def generate_action_items(self, results: Dict[str, Any]) -> List[Dict[str, Any]]:
        """アクションアイテム生成"""
        action_items = []
        
        # 高リスクアイテムの処理
        for category in ['technical_assessment', 'operational_assessment', 'business_assessment']:
            for metric_name, metric_result in results[category].items():
                if hasattr(metric_result, 'risk_level') and metric_result.risk_level in ['high', 'medium']:
                    action_items.append({
                        'title': f"{metric_name}の改善",
                        'priority': 'high' if metric_result.risk_level == 'high' else 'medium',
                        'due_date': (datetime.now() + timedelta(days=30)).isoformat(),
                        'recommendations': metric_result.recommendations,
                        'owner': 'monitoring_team'
                    })
        
        return action_items
    
    def generate_review_report(self, results: Dict[str, Any]) -> str:
        """レビューレポート生成"""
        template_str = """
        # 監視システムレビューレポート
        
        ## レビュー情報
        - タイプ: {{ results.review_info.type }}
        - 実施日: {{ results.review_info.date }}
        - スコープ: {{ results.review_info.scope }}
        
        ## 総合評価
        - **総合スコア**: {{ "%.1f"|format(results.overall_score) }}/100
        
        ## 技術的評価
        {% for metric_name, metric_result in results.technical_assessment.items() %}
        ### {{ metric_name }}
        - 現在値: {{ metric_result.current_value }}
        - 目標値: {{ metric_result.target_value }}
        - ステータス: {{ metric_result.status }}
        - リスクレベル: {{ metric_result.risk_level }}
        {% if metric_result.recommendations %}
        - 推奨事項:
        {% for rec in metric_result.recommendations %}
          - {{ rec }}
        {% endfor %}
        {% endif %}
        {% endfor %}
        
        ## 運用的評価
        {% for metric_name, metric_result in results.operational_assessment.items() %}
        ### {{ metric_name }}
        - 現在値: {{ metric_result.current_value }}
        - ステータス: {{ metric_result.status }}
        {% endfor %}
        
        ## ビジネス的評価
        {% for metric_name, metric_result in results.business_assessment.items() %}
        ### {{ metric_name }}
        - 現在値: {{ metric_result.current_value }}
        - ステータス: {{ metric_result.status }}
        {% endfor %}
        
        ## 推奨事項
        {% for rec in results.recommendations %}
        - {{ rec }}
        {% endfor %}
        
        ## アクションアイテム
        {% for item in results.action_items %}
        ### {{ item.title }}
        - 優先度: {{ item.priority }}
        - 期限: {{ item.due_date }}
        - 担当: {{ item.owner }}
        {% endfor %}
        """
        
        template = jinja2.Template(template_str)
        return template.render(results=results)

# 使用例
if __name__ == "__main__":
    review_system = MonitoringReviewSystem()
    
    # 月次レビュー実施
    results = review_system.conduct_comprehensive_review("monthly")
    
    # レポート生成
    report = review_system.generate_review_report(results)
    
    # ファイル出力
    with open(f"monitoring_review_{datetime.now().strftime('%Y%m%d')}.md", 'w', encoding='utf-8') as f:
        f.write(report)
    
    print(f"総合スコア: {results['overall_score']:.1f}/100")
    print(f"アクションアイテム数: {len(results['action_items'])}")

#!/usr/bin/env python3
"""
Zabbixチーム教育管理システム
"""
import json
import sqlite3
from datetime import datetime, timedelta
from typing import Dict, List, Any, Optional
from dataclasses import dataclass
from enum import Enum
import yaml

class SkillLevel(Enum):
    BEGINNER = "beginner"
    INTERMEDIATE = "intermediate"
    ADVANCED = "advanced"
    EXPERT = "expert"

class LearningStatus(Enum):
    NOT_STARTED = "not_started"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    CERTIFIED = "certified"

@dataclass
class LearningModule:
    """学習モジュール"""
    id: str
    name: str
    level: SkillLevel
    duration_hours: int
    prerequisites: List[str]
    topics: List[str]
    certification_required: bool

@dataclass
class UserProgress:
    """ユーザー進捗"""
    user_id: str
    module_id: str
    status: LearningStatus
    start_date: Optional[datetime]
    completion_date: Optional[datetime]
    score: Optional[float]
    attempts: int

class LearningManagementSystem:
    """学習管理システム"""
    
    def __init__(self, db_path: str = "learning_system.db"):
        self.db_path = db_path
        self.init_database()
        self.modules = self.load_learning_modules()
    
    def init_database(self):
        """データベース初期化"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id TEXT UNIQUE NOT NULL,
                name TEXT NOT NULL,
                email TEXT NOT NULL,
                role TEXT,
                skill_level TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS learning_modules (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                module_id TEXT UNIQUE NOT NULL,
                name TEXT NOT NULL,
                level TEXT NOT NULL,
                duration_hours INTEGER,
                prerequisites TEXT,
                topics TEXT,
                certification_required BOOLEAN,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS user_progress (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id TEXT NOT NULL,
                module_id TEXT NOT NULL,
                status TEXT NOT NULL,
                start_date TIMESTAMP,
                completion_date TIMESTAMP,
                score REAL,
                attempts INTEGER DEFAULT 0,
                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                UNIQUE(user_id, module_id)
            )
        ''')
        
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS certifications (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id TEXT NOT NULL,
                certification_type TEXT NOT NULL,
                issue_date TIMESTAMP,
                expiry_date TIMESTAMP,
                certificate_id TEXT UNIQUE,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS skill_assessments (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id TEXT NOT NULL,
                assessment_date TIMESTAMP,
                technical_score REAL,
                operational_score REAL,
                leadership_score REAL,
                overall_score REAL,
                assessor TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        
        conn.commit()
        conn.close()
    
    def load_learning_modules(self) -> List[LearningModule]:
        """学習モジュール読み込み"""
        modules = [
            LearningModule(
                id="zabbix_basics",
                name="Zabbix基礎概念",
                level=SkillLevel.BEGINNER,
                duration_hours=8,
                prerequisites=[],
                topics=["監視システム基本原理", "Zabbixアーキテクチャ", "基本用語"],
                certification_required=False
            ),
            LearningModule(
                id="basic_operations",
                name="基本操作習得",
                level=SkillLevel.BEGINNER,
                duration_hours=24,
                prerequisites=["zabbix_basics"],
                topics=["Webインターフェース", "ホスト設定", "トリガー作成"],
                certification_required=True
            ),
            LearningModule(
                id="advanced_configuration",
                name="高度な設定技術",
                level=SkillLevel.INTERMEDIATE,
                duration_hours=40,
                prerequisites=["basic_operations"],
                topics=["複雑なトリガー", "テンプレート設計", "自動ディスカバリ"],
                certification_required=True
            ),
            LearningModule(
                id="operations_management",
                name="運用管理技術",
                level=SkillLevel.INTERMEDIATE,
                duration_hours=40,
                prerequisites=["advanced_configuration"],
                topics=["パフォーマンス最適化", "バックアップ", "セキュリティ"],
                certification_required=True
            ),
            LearningModule(
                id="architecture_design",
                name="アーキテクチャ設計",
                level=SkillLevel.EXPERT,
                duration_hours=40,
                prerequisites=["operations_management"],
                topics=["分散監視設計", "高可用性", "スケーリング"],
                certification_required=True
            )
        ]
        
        # データベースに保存
        self.save_modules_to_db(modules)
        return modules
    
    def save_modules_to_db(self, modules: List[LearningModule]):
        """モジュールをデータベースに保存"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        for module in modules:
            cursor.execute('''
                INSERT OR REPLACE INTO learning_modules 
                (module_id, name, level, duration_hours, prerequisites, topics, certification_required)
                VALUES (?, ?, ?, ?, ?, ?, ?)
            ''', (
                module.id,
                module.name,
                module.level.value,
                module.duration_hours,
                json.dumps(module.prerequisites),
                json.dumps(module.topics),
                module.certification_required
            ))
        
        conn.commit()
        conn.close()
    
    def register_user(self, user_id: str, name: str, email: str, role: str, skill_level: SkillLevel):
        """ユーザー登録"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            INSERT OR REPLACE INTO users (user_id, name, email, role, skill_level)
            VALUES (?, ?, ?, ?, ?)
        ''', (user_id, name, email, role, skill_level.value))
        
        conn.commit()
        conn.close()
    
    def get_recommended_learning_path(self, user_id: str) -> List[LearningModule]:
        """推奨学習パス取得"""
        user_skill_level = self.get_user_skill_level(user_id)
        completed_modules = self.get_completed_modules(user_id)
        
        recommended = []
        
        for module in self.modules:
            # スキルレベルに適合し、前提条件を満たす
            if self.is_module_appropriate(module, user_skill_level, completed_modules):
                recommended.append(module)
        
        return sorted(recommended, key=lambda m: m.duration_hours)
    
    def get_user_skill_level(self, user_id: str) -> SkillLevel:
        """ユーザースキルレベル取得"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('SELECT skill_level FROM users WHERE user_id = ?', (user_id,))
        result = cursor.fetchone()
        conn.close()
        
        if result:
            return SkillLevel(result[0])
        return SkillLevel.BEGINNER
    
    def get_completed_modules(self, user_id: str) -> List[str]:
        """完了済みモジュール取得"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            SELECT module_id FROM user_progress 
            WHERE user_id = ? AND status IN (?, ?)
        ''', (user_id, LearningStatus.COMPLETED.value, LearningStatus.CERTIFIED.value))
        
        results = cursor.fetchall()
        conn.close()
        
        return [row[0] for row in results]
    
    def is_module_appropriate(self, module: LearningModule, user_level: SkillLevel, completed: List[str]) -> bool:
        """モジュールの適切性判定"""
        # スキルレベル判定
        level_order = [SkillLevel.BEGINNER, SkillLevel.INTERMEDIATE, SkillLevel.ADVANCED, SkillLevel.EXPERT]
        module_level_idx = level_order.index(module.level)
        user_level_idx = level_order.index(user_level)
        
        # ユーザーレベル以下または1レベル上まで
        if module_level_idx > user_level_idx + 1:
            return False
        
        # 前提条件確認
        for prereq in module.prerequisites:
            if prereq not in completed:
                return False
        
        # 未完了であること
        return module.id not in completed
    
    def start_module(self, user_id: str, module_id: str):
        """モジュール開始"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            INSERT OR REPLACE INTO user_progress 
            (user_id, module_id, status, start_date, attempts)
            VALUES (?, ?, ?, ?, COALESCE((SELECT attempts FROM user_progress WHERE user_id = ? AND module_id = ?), 0) + 1)
        ''', (user_id, module_id, LearningStatus.IN_PROGRESS.value, datetime.now().isoformat(), user_id, module_id))
        
        conn.commit()
        conn.close()
    
    def complete_module(self, user_id: str, module_id: str, score: float = None):
        """モジュール完了"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            UPDATE user_progress 
            SET status = ?, completion_date = ?, score = ?, updated_at = ?
            WHERE user_id = ? AND module_id = ?
        ''', (
            LearningStatus.COMPLETED.value,
            datetime.now().isoformat(),
            score,
            datetime.now().isoformat(),
            user_id,
            module_id
        ))
        
        conn.commit()
        conn.close()
        
        # 認定が必要な場合の処理
        module = next((m for m in self.modules if m.id == module_id), None)
        if module and module.certification_required and score and score >= 80.0:
            self.issue_certification(user_id, module_id)
    
    def issue_certification(self, user_id: str, module_id: str):
        """認定証発行"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        certificate_id = f"ZABBIX_{module_id.upper()}_{user_id}_{datetime.now().strftime('%Y%m%d')}"
        
        cursor.execute('''
            INSERT INTO certifications 
            (user_id, certification_type, issue_date, expiry_date, certificate_id)
            VALUES (?, ?, ?, ?, ?)
        ''', (
            user_id,
            module_id,
            datetime.now().isoformat(),
            (datetime.now() + timedelta(days=730)).isoformat(),  # 2年有効
            certificate_id
        ))
        
        # ユーザー進捗を認定済みに更新
        cursor.execute('''
            UPDATE user_progress 
            SET status = ?, updated_at = ?
            WHERE user_id = ? AND module_id = ?
        ''', (LearningStatus.CERTIFIED.value, datetime.now().isoformat(), user_id, module_id))
        
        conn.commit()
        conn.close()
    
    def conduct_skill_assessment(self, user_id: str, technical_score: float, operational_score: float, leadership_score: float, assessor: str):
        """スキル評価実施"""
        overall_score = (technical_score + operational_score + leadership_score) / 3
        
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            INSERT INTO skill_assessments 
            (user_id, assessment_date, technical_score, operational_score, leadership_score, overall_score, assessor)
            VALUES (?, ?, ?, ?, ?, ?, ?)
        ''', (
            user_id,
            datetime.now().isoformat(),
            technical_score,
            operational_score,
            leadership_score,
            overall_score,
            assessor
        ))
        
        conn.commit()
        conn.close()
        
        # スキルレベル更新
        self.update_user_skill_level(user_id, overall_score)
    
    def update_user_skill_level(self, user_id: str, overall_score: float):
        """ユーザースキルレベル更新"""
        if overall_score >= 90:
            new_level = SkillLevel.EXPERT
        elif overall_score >= 80:
            new_level = SkillLevel.ADVANCED
        elif overall_score >= 70:
            new_level = SkillLevel.INTERMEDIATE
        else:
            new_level = SkillLevel.BEGINNER
        
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            UPDATE users SET skill_level = ? WHERE user_id = ?
        ''', (new_level.value, user_id))
        
        conn.commit()
        conn.close()
    
    def generate_team_skills_report(self) -> Dict[str, Any]:
        """チームスキルレポート生成"""
        conn = sqlite3.connect(self.db_path)
        
        # スキルレベル分布
        skill_distribution = pd.read_sql_query('''
            SELECT skill_level, COUNT(*) as count
            FROM users 
            GROUP BY skill_level
        ''', conn)
        
        # 認定状況
        certification_status = pd.read_sql_query('''
            SELECT certification_type, COUNT(*) as count
            FROM certifications 
            WHERE expiry_date > ?
            GROUP BY certification_type
        ''', conn, params=[datetime.now().isoformat()])
        
        # 最新スキル評価
        latest_assessments = pd.read_sql_query('''
            SELECT AVG(technical_score) as avg_technical,
                   AVG(operational_score) as avg_operational,
                   AVG(leadership_score) as avg_leadership,
                   AVG(overall_score) as avg_overall
            FROM skill_assessments
            WHERE assessment_date >= ?
        ''', conn, params=[(datetime.now() - timedelta(days=180)).isoformat()])
        
        conn.close()
        
        return {
            'skill_distribution': skill_distribution.to_dict('records'),
            'certification_status': certification_status.to_dict('records'),
            'average_scores': latest_assessments.to_dict('records')[0] if not latest_assessments.empty else {},
            'recommendations': self.generate_team_recommendations(skill_distribution, certification_status)
        }
    
    def generate_team_recommendations(self, skill_dist: pd.DataFrame, cert_status: pd.DataFrame) -> List[str]:
        """チーム推奨事項生成"""
        recommendations = []
        
        # スキルレベル分析
        if not skill_dist.empty:
            expert_count = skill_dist[skill_dist['skill_level'] == 'expert']['count'].sum()
            total_count = skill_dist['count'].sum()
            
            if expert_count / total_count < 0.2:
                recommendations.append("エキスパートレベル人材の育成強化が必要")
        
        # 認定状況分析
        if cert_status.empty or len(cert_status) < 3:
            recommendations.append("基礎レベル認定の取得推進が必要")
        
        return recommendations

# 使用例
if __name__ == "__main__":
    lms = LearningManagementSystem()
    
    # ユーザー登録
    lms.register_user("user001", "田中太郎", "[email protected]", "監視エンジニア", SkillLevel.BEGINNER)
    
    # 推奨学習パス取得
    recommended_path = lms.get_recommended_learning_path("user001")
    print("推奨学習パス:")
    for module in recommended_path:
        print(f"- {module.name} ({module.duration_hours}時間)")
    
    # モジュール開始・完了
    lms.start_module("user001", "zabbix_basics")
    lms.complete_module("user001", "zabbix_basics", 85.0)
    
    # スキル評価
    lms.conduct_skill_assessment("user001", 75.0, 70.0, 65.0, "supervisor001")
    
    # チームレポート
    team_report = lms.generate_team_skills_report()
    print("チームスキルレポート:", json.dumps(team_report, indent=2, ensure_ascii=False))