# 将所有日志输出到一个统一的位置
sudo rsyslog -n /etc/rsyslog.conf

# 或者使用 Docker 的 JSON 日志驱动
docker service create \
  --log-driver json-file \
  --log-opt max-size=10m \
  --log-opt max-file=3 \
  myapp:latest

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

class LogPreprocessor:
    """日志预处理：去重、分类、时间窗口聚合"""
    
    def __init__(self):
        self.severity_map = {
            'emerg': 0, 'alert': 1, 'crit': 2, 'err': 3,
            'warning': 4, 'notice': 5, 'info': 6, 'debug': 7
        }
    
    def normalize(self, log_line):
        """标准化日志行，提取关键信息"""
        # 解析常见日志格式
        patterns = [
            r'^(\w+\s+\d+\s+[\d:]+)\s+(\S+)\s+(\S+?):\s+(.*)',  # syslog
            r'(\d{4}-\d{2}-\d{T}\d{2}:\d{2}:\d{2}[.\d+Z-]+)\s+(.*?)\s+(\w+)\s+(.*)',  # ISO
        ]
        
        for pattern in patterns:
            import re
            match = re.search(pattern, log_line)
            if match:
                groups = match.groups()
                return {
                    'timestamp': groups[0],
                    'host': groups[1] if len(groups) > 1 else 'unknown',
                    'service': groups[2] if len(groups) > 2 else 'system',
                    'message': groups[-1],
                    'raw': log_line
                }
        return {'message': log_line, 'raw': log_line}
    
    def deduplicate(self, logs):
        """基于消息内容的去重"""
        seen = {}
        result = []
        for log in logs:
            msg = log.get('message', '')[:200]  # 截断长消息
            if msg not in seen:
                seen[msg] = 0
            seen[msg] += 1
            if seen[msg] == 1:
                result.append({**log, 'count': 1})
            else:
                # 更新最后一条重复日志的计数
                result[-1]['count'] = seen[msg]
        return result
    
    def aggregate_by_window(self, logs, window_minutes=5):
        """按时间窗口聚合日志"""
        windows = defaultdict(list)
        for log in logs:
            ts = log.get('timestamp', '')
            try:
                dt = datetime.fromisoformat(ts.replace('Z', '+00:00'))
                window_key = dt.replace(
                    minute=(dt.minute // window_minutes) * window_minutes,
                    second=0, microsecond=0
                )
                windows[str(window_key)].append(log)
            except (ValueError, TypeError):
                windows['unknown'].append(log)
        return dict(windows)

SYSTEM_PROMPT = """你是一个经验丰富的 Linux 系统运维专家和安全分析师。
你的任务是根据提供的系统日志分析潜在问题，并按以下格式输出：

1. 【问题摘要】用一句话概括发现的核心问题
2. 【严重程度】P0(紧急) / P1(高) / P2(中) / P3(低)
3. 【影响范围】受影响的服务/用户/功能
4. 【根因分析】详细解释问题的可能原因
5. 【证据链】列出支持该判断的关键日志条目
6. 【修复建议】提供具体的、可执行的修复步骤
7. 【预防措施】如何避免类似问题再次发生

注意：只基于提供的日志进行分析，不要臆测不存在的信息。
如果日志中没有明显异常，请明确说明"未检测到已知问题模式"。"""

USER_PROMPT_TEMPLATE = """以下是 VPS 在 {time_range} 内收集的日志摘要：

服务器信息：
- 操作系统：{os_info}
- 内核版本：{kernel_version}
- 运行时间：{uptime}
- 负载情况：{load_average}

--- 日志内容 ---
{log_content}

--- 已知告警 ---
{alerts}

请开始分析。"""

import openai

class LogAnalyzer:
    """基于 LLM 的智能日志分析器"""
    
    def __init__(self, api_key, model="gpt-4o"):
        self.client = openai.OpenAI(api_key=api_key)
        self.model = model
    
    def analyze_logs(self, server_info, log_entries, alerts=None):
        """分析日志并返回结构化诊断报告"""
        
        # 将日志转换为 LLM 友好的格式
        log_text = self._format_logs(log_entries)
        
        user_prompt = USER_PROMPT_TEMPLATE.format(
            time_range=server_info.get('time_range', '最近30分钟'),
            os_info=server_info.get('os', 'Linux'),
            kernel_version=server_info.get('kernel', 'unknown'),
            uptime=server_info.get('uptime', 'unknown'),
            load_average=server_info.get('load_average', 'unknown'),
            log_content=log_text,
            alerts=alerts or "无已知告警"
        )
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": SYSTEM_PROMPT},
                {"role": "user", "content": user_prompt}
            ],
            temperature=0.3,  # 较低的温度以获得更一致的分析结果
            max_tokens=2000
        )
        
        return self._parse_analysis(response.choices[0].message.content)
    
    def _format_logs(self, logs):
        """将日志列表格式化为可读文本"""
        formatted = []
        for log in logs[:200]:  # 限制数量防止超出 token 限制
            severity = log.get('severity', 'INFO').upper()
            timestamp = log.get('timestamp', '?')
            service = log.get('service', 'system')
            message = log.get('message', '')[:300]
            count = log.get('count', 1)
            
            entry = f"[{timestamp}] [{severity}] {service}: {message}"
            if count > 1:
                entry += f" (重复 {count} 次)"
            formatted.append(entry)
        
        return "\n".join(formatted)
    
    def _parse_analysis(self, analysis_text):
        """将 LLM 的分析结果解析为结构化数据"""
        result = {
            'summary': '',
            'severity': 'P3',
            'impact': '',
            'root_cause': '',
            'evidence': [],
            'fix_steps': [],
            'prevention': ''
        }
        
        # 简单解析（实际项目中可以用更复杂的 NLP 方法）
        current_section = None
        section_map = {
            '问题摘要': 'summary',
            '严重程度': 'severity',
            '影响范围': 'impact',
            '根因分析': 'root_cause',
            '证据链': 'evidence',
            '修复建议': 'fix_steps',
            '预防措施': 'prevention'
        }
        
        for line in analysis_text.split('\n'):
            line = line.strip()
            for cn_title, key in section_map.items():
                if cn_title in line:
                    current_section = key
                    result[key] = line.split(cn_title)[-1].strip().lstrip('：:')
                    break
        
        return result

def deep_dive_analysis(analyzer, server_id, date_range, keywords=None):
    """
    对指定时间段内的日志进行深度分析
    
    Args:
        analyzer: LogAnalyzer 实例
        server_id: 服务器标识
        date_range: 时间范围，如 ('2026-07-20', '2026-07-22')
        keywords: 关注的关键词列表
    """
    # 1. 获取历史日志
    logs = fetch_history_logs(server_id, date_range, keywords)
    
    # 2. 按错误类型分组
    error_groups = group_by_error_type(logs)
    
    # 3. 构造上下文丰富的分析请求
    context = {
        'server_id': server_id,
        'time_range': f"{date_range[0]} 至 {date_range[1]}",
        'total_events': len(logs),
        'error_count': sum(1 for l in logs if l.get('severity') in ['err', 'crit', 'alert', 'emerg']),
        'top_errors': error_groups[:10],
        'recent_deployments': get_recent_deployments(server_id, date_range),
        'config_changes': get_config_changes(server_id, date_range)
    }
    
    # 4. 发送给 LLM 深度分析
    report = analyzer.analyze_logs(context, logs)
    
    return report

class AlertManager:
    """基于 AI 分析结果的智能告警和修复"""
    
    def __init__(self):
        self.notification_channels = {
            'slack': self._send_to_slack,
            'email': self._send_email,
            'webhook': self._send_webhook
        }
    
    def process_analysis(self, analysis, server_info):
        """处理 AI 分析结果，决定是否需要告警和修复"""
        
        severity = analysis.get('severity', 'P3')
        
        # 根据严重程度决定动作
        if severity in ['P0', 'P1']:
            # 紧急/高严重性：立即通知 + 尝试自动修复
            self.send_alert(analysis, server_info, priority='critical')
            if severity == 'P0':
                self.try_auto_fix(analysis, server_info)
        
        elif severity == 'P2':
            # 中等：通知但不紧急
            self.send_alert(analysis, server_info, priority='normal')
        
        # 记录分析结果到知识库
        self.save_to_knowledge_base(analysis, server_info)
    
    def send_alert(self, analysis, server_info, priority='normal'):
        """发送告警通知"""
        message = f"""🔧 **VPS 智能诊断报告**

**服务器**: {server_info.get('hostname')} ({server_info.get('ip')})
**严重程度**: {analysis.get('severity', '未知')}
**问题摘要**: {analysis.get('summary', '无')}

**根因分析**:
{analysis.get('root_cause', 'N/A')}

**修复建议**:
{chr(10).join('- ' + step for step in analysis.get('fix_steps', []))}

---
由 selfvps.net AI 日志分析引擎自动生成
"""
        
        # 发送到所有配置的渠道
        for channel_name, channel_fn in self.notification_channels.items():
            channel_fn(message, priority=priority)
    
    def try_auto_fix(self, analysis, server_info):
        """尝试执行 AI 推荐的自动修复"""
        fix_steps = analysis.get('fix_steps', [])
        
        for step in fix_steps:
            # 这里应该有一个安全沙箱来验证修复命令的安全性
            if self._is_safe_command(step):
                result = self._execute_command(step, server_info)
                print(f"执行修复: {step} → {result}")
            else:
                print(f"⚠️ 跳过不安全的修复命令: {step}")
    
    def _is_safe_command(self, command_str):
        """检查命令是否安全可执行"""
        dangerous_patterns = ['rm -rf /', 'dd if=', '> /dev/sda', ':(){ :|:& };:']
        return not any(pattern in command_str for pattern in dangerous_patterns)

# 使用 Ollama 本地运行
ollama pull llama3.2

# 通过 API 调用
curl http://localhost:11434/api/generate -d '{
  "model": "llama3.2",
  "prompt": "分析以下系统日志...",
  "stream": false
}'

class KnowledgeBase:
    """运维知识库——积累历史诊断经验"""
    
    def store_case(self, log_pattern, diagnosis, fix, effectiveness):
        """存储一个诊断案例"""
        case = {
            'pattern_hash': hash(log_pattern),
            'diagnosis': diagnosis,
            'fix_applied': fix,
            'effectiveness': effectiveness,  # 修复效果评分 0-1
            'timestamp': datetime.now().isoformat(),
            'server_type': self._classify_server()
        }
        self.db.collection('cases').insert(case)
    
    def find_similar(self, new_logs):
        """查找相似的历史案例"""
        pattern_hash = hash(new_logs)
        similar = self.db.collection('cases').find({
            'pattern_hash': {'$near': pattern_hash},
            'effectiveness': {'$gt': 0.8}
        }).limit(5)
        return similar

def correlated_analysis(server_logs_dict):
    """
    跨服务器日志关联分析
    
    Args:
        server_logs_dict: {server_id: [log_entries]}
    """
    # 将所有服务器的日志按时间对齐
    unified_timeline = merge_timelines(server_logs_dict)
    
    # 寻找时间窗口内的共现模式
    correlations = find_temporal_correlations(unified_timeline)
    
    # 将关联结果发给 LLM 综合判断
    prompt = f"""
    以下事件在多台服务器上同时发生，请分析是否存在关联：
    
    {correlations}
    
    可能的关联场景：
    1. 同一网络变更影响了所有服务器
    2. 上游服务故障导致下游连锁反应
    3. 定时任务在同一时间触发资源竞争
    4. 外部攻击针对了特定服务
    """
    
    return llm_client.generate(prompt)

# prometheus 自定义 exporter
# 当 Prometheus 告警触发时，自动调用 AI 分析

rules:
  - alert: HighErrorRate
    expr: rate(http_requests_total{status=~"5.."}[5m]) > 0.05
    for: 2m
    annotations:
      summary: "HTTP 5xx 错误率超过 5%"
      ai_analysis: "触发 AI 日志分析"

def incremental_analyze(current_logs, previous_state):
    """增量分析：只分析新增的异常"""
    
    # 1. 找出上次分析后新增的日志
    new_logs = filter_since(current_logs, previous_state['last_analysis_time'])
    
    # 2. 只对有异常的窗口进行深入分析
    anomaly_windows = detect_anomaly_windows(new_logs)
    
    # 3. 对每个异常窗口调用 LLM
    for window in anomaly_windows:
        report = llm_analyze(window.logs)
        if report['severity'] in ['P0', 'P1']:
            send_alert(report)
    
    # 4. 更新状态
    previous_state['last_analysis_time'] = datetime.now()
    return previous_state

from celery import Celery

celery_app = Celery('log_analyzer', broker='redis://localhost:6379/0')

@celery_app.task
def async_log_analysis(server_id, log_batch):
    """异步执行日志分析"""
    analyzer = LogAnalyzer(api_key=os.environ['LLM_API_KEY'])
    result = analyzer.analyze_logs(load_server_info(server_id), log_batch)
    
    # 保存结果并触发后续动作
    save_analysis_result(server_id, result)
    if result['severity'] in ['P0', 'P1']:
        trigger_alert.s(result, server_id).delay()
    
    return result['severity']