# 安装 Vector 日志采集器
curl -fsSL https://sh.vector.dev | bash

# 配置采集 syslog、auth.log、nginx access/error log
cat > /etc/vector/vector.toml << 'EOF'
[sources.system_logs]
type = "vector_sources"
read_from = "beginning"

[sources.auth_logs]
type = "vector_sources"
read_from = "beginning"

[transforms.parse_auth]
type = "remap"
inputs = ["auth_logs"]
source = '''
  .parsed = parse_regex(.message, r'^(?P<timestamp>\S+\s+\d+\s+\S+)\s+(?P<host>\S+)\s+(?P<service>\S+):?\s*(?P<message>.*)') ?? {}
'''

[transforms.parse_nginx]
type = "remap"
inputs = ["nginx_logs"]
source = '''
  .parsed = parse_json(.message) ?? {}
'''

[sinks.chroma_db]
type = "http"
inputs = ["parse_auth", "parse_nginx"]
encoding.codec = "json"
uri = "http://localhost:8000/api/v1/ingest"
EOF

# import_tickets.py - 将工单转为知识库条目
import json
from datetime import datetime

def convert_ticket_to_knowledge(ticket: dict) -> dict:
    return {
        "id": ticket["id"],
        "type": "incident",
        "title": ticket["subject"],
        "description": ticket["description"],
        "symptoms": extract_symptoms(ticket),
        "root_cause": ticket.get("resolution", {}).get("root_cause", ""),
        "solution": ticket.get("resolution", {}).get("solution", ""),
        "tags": ticket.get("labels", []),
        "severity": ticket.get("priority", "medium"),
        "created_at": ticket["created_at"],
        "resolved_at": ticket["resolved_at"],
        "duration_minutes": calculate_duration(ticket),
        "confidence": 0.9 if ticket.get("resolution") else 0.5,
    }

def extract_symptoms(ticket: dict) -> list:
    """从工单描述中提取症状关键词"""
    symptoms = []
    text = ticket["description"].lower()
    symptom_patterns = {
        "high_cpu": ["cpu", "负载", "100%", "top"],
        "high_memory": ["内存", "oom", "swap", "memory"],
        "disk_full": ["磁盘", "disk", "no space", "容量"],
        "network_down": ["网络", "timeout", "unreachable", "连接"],
        "service_crash": ["崩溃", "crash", "重启", "restart"],
        "slow_response": ["慢", "slow", "延迟", "timeout"],
        "auth_failure": ["登录失败", "permission denied", "认证"],
    }
    for symptom, keywords in symptom_patterns.items():
        if any(kw in text for kw in keywords):
            symptoms.append(symptom)
    return symptoms

# knowledge_collector.py
import os
import glob
from pathlib import Path

class KnowledgeCollector:
    def __init__(self, sources: list[str]):
        self.sources = sources
        self.documents = []

    def collect(self) -> list[dict]:
        for source in self.sources:
            if source.startswith("/"):
                self._collect_files(source)
            elif source.startswith("http"):
                self._collect_web(source)
            elif source == "journal":
                self._collect_journalctl()
        return self.documents

    def _collect_files(self, path: str):
        for pattern in ["*.md", "*.txt", "*.yaml", "*.yml", "*.conf"]:
            for file in glob.glob(os.path.join(path, "**", pattern), recursive=True):
                content = Path(file).read_text(encoding="utf-8", errors="ignore")
                self.documents.append({
                    "source": file,
                    "content": content,
                    "type": "document",
                    "updated_at": datetime.now().isoformat(),
                })

    def _collect_journalctl(self):
        """采集 systemd journal 中的关键事件"""
        import subprocess
        result = subprocess.run(
            ["journalctl", "-u", "nginx", "--since", "7 days ago", "-p", "err"],
            capture_output=True, text=True
        )
        if result.returncode == 0 and result.stdout.strip():
            self.documents.append({
                "source": "systemd-journal",
                "content": result.stdout,
                "type": "log",
                "updated_at": datetime.now().isoformat(),
            })

    def _collect_web(self, url: str):
        """采集 Web 文档（Wiki、Confluence 等）"""
        # 使用 requests + BeautifulSoup
        pass

# chunker.py - 智能文档分块
from langchain.text_splitter import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter

class KnowledgeChunker:
    def __init__(self, chunk_size: int = 512, chunk_overlap: int = 64):
        self.header_splitter = MarkdownHeaderTextSplitter(
            headers_to_split_on=[
                ("#", "heading_1"),
                ("##", "heading_2"),
                ("###", "heading_3"),
            ]
        )
        self.sentence_splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            separators=["\n\n", "\n", ". ", " ", ""],
        )

    def chunk(self, document: dict) -> list[dict]:
        """将文档切分为语义完整的片段"""
        chunks = []

        # 先按标题结构拆分
        if document["type"] == "document" and document["content"].startswith("#"):
            header_chunks = self.header_splitter.split_text(document["content"])
        else:
            header_chunks = [document["content"]]

        # 再对每个部分进行递归分块
        for text in header_chunks:
            sub_chunks = self.sentence_splitter.create_documents([text])
            for chunk in sub_chunks:
                chunks.append({
                    "content": chunk.page_content,
                    "metadata": {
                        **document.get("metadata", {}),
                        "source": document.get("source", "unknown"),
                        "chunk_index": len(chunks),
                    },
                })

        return chunks

# embedding_service.py
from langchain_community.embeddings import OllamaEmbeddings
from chromadb import HttpClient

class VectorStoreService:
    def __init__(self, model_name: str = "nomic-embed-text"):
        self.embeddings = OllamaEmbeddings(
            model=model_name,
            base_url="http://localhost:11434"
        )
        self.client = HttpClient(host="localhost", port=8000)
        self.collection = self.client.get_or_create_collection(
            name="vps_operations",
            metadata={"hnsw:space": "cosine"}
        )

    def embed_and_store(self, chunks: list[dict]):
        """批量向量化并存储"""
        texts = [chunk["content"] for chunk in chunks]
        metadatas = [chunk["metadata"] for chunk in chunks]

        embeddings = self.embeddings.embed_documents(texts)

        self.collection.upsert(
            ids=[f"chunk_{i}" for i in range(len(texts))],
            documents=texts,
            embeddings=embeddings,
            metadatas=metadatas,
        )

    def search(self, query: str, top_k: int = 5) -> list[dict]:
        """语义搜索"""
        query_embedding = self.embeddings.embed_query(query)
        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=top_k,
            include=["documents", "metadatas", "distances"],
        )
        return [
            {
                "content": doc,
                "score": 1 - dist,  # cosine distance → similarity
                "metadata": meta,
            }
            for doc, dist, meta in zip(
                results["documents"][0],
                results["distances"][0],
                results["metadatas"][0],
            )
        ]

# knowledge_graph.py
import networkx as nx

class OperationsKnowledgeGraph:
    def __init__(self):
        self.graph = nx.DiGraph()

    def add_entity(self, name: str, entity_type: str, properties: dict = None):
        self.graph.add_node(name, type=entity_type, **(properties or {}))

    def add_relation(self, source: str, target: str, relation: str):
        self.graph.add_edge(source, target, relation=relation)

    def find_root_causes(self, symptom: str) -> list[str]:
        """从症状反推可能的根因"""
        candidates = []
        for node in self.graph.nodes():
            if self.graph.nodes[node].get("type") == "symptom":
                if symptom.lower() in node.lower() or node.lower() in symptom.lower():
                    # 沿反向边查找根因
                    predecessors = list(self.graph.predecessors(node))
                    for pred in predecessors:
                        if self.graph.nodes[pred].get("type") == "root_cause":
                            candidates.append(pred)
        return candidates

    def get_related_incidents(self, component: str, limit: int = 10) -> list[str]:
        """查找与某组件相关的历史故障"""
        related = []
        for successor in self.graph.successors(component):
            if self.graph.nodes[successor].get("type") == "incident":
                related.append(successor)
        return related[:limit]

# rag_engine.py
from langchain.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser

class VPSOperationsRAG:
    def __init__(self, llm_model: str = "qwen2.5:7b"):
        from langchain_community.llms import Ollama
        self.llm = Ollama(model=llm_model, temperature=0.2)
        self.vector_store = VectorStoreService()
        self.kg = OperationsKnowledgeGraph()

    def ask(self, question: str, context: dict = None) -> dict:
        """处理运维问答"""
        # 1. 意图分类
        intent = self._classify_intent(question)

        # 2. 根据意图选择检索策略
        if intent == "diagnosis":
            relevant_docs = self._diagnose_retrieve(question, context)
        elif intent == "sop":
            relevant_docs = self._sop_retrieve(question)
        elif intent == "general":
            relevant_docs = self.vector_store.search(question, top_k=8)
        else:
            relevant_docs = self.vector_store.search(question, top_k=5)

        # 3. 构建上下文
        context_text = self._build_context(relevant_docs)

        # 4. 生成回答
        answer = self._generate_answer(question, context_text, intent)

        return {
            "question": question,
            "intent": intent,
            "answer": answer,
            "sources": [doc["metadata"] for doc in relevant_docs[:3]],
            "confidence": self._calculate_confidence(relevant_docs),
        }

    def _classify_intent(self, question: str) -> str:
        """判断用户意图"""
        diagnosis_keywords = ["故障", "报错", "异常", "挂", "崩", "error", "fail"]
        sop_keywords = ["步骤", "流程", "怎么办", "如何处理", "sop", "runbook"]
        general_keywords = ["是什么", "怎么配置", "如何", "why"]

        q = question.lower()
        if any(kw in q for kw in diagnosis_keywords):
            return "diagnosis"
        elif any(kw in q for kw in sop_keywords):
            return "sop"
        return "general"

    def _diagnose_retrieve(self, question: str, context: dict) -> list[dict]:
        """故障诊断专用检索：结合症状、指标和历史案例"""
        docs = self.vector_store.search(question, top_k=5)

        # 如果有实时指标，补充相关配置文档
        if context and "metrics" in context:
            metric_names = list(context["metrics"].keys())
            config_query = " ".join(metric_names)
            config_docs = self.vector_store.search(
                f"配置文档 {config_query} 调优", top_k=3
            )
            docs.extend(config_docs)

        return docs

    def _build_context(self, docs: list[dict]) -> str:
        """构建供 LLM 使用的上下文"""
        parts = []
        for i, doc in enumerate(docs, 1):
            score_label = "高相关" if doc["score"] > 0.8 else "中相关" if doc["score"] > 0.6 else "低相关"
            parts.append(f"[来源{i} | 相关度:{score_label}] {doc['content']}")
        return "\n\n".join(parts)

    def _generate_answer(self, question: str, context: str, intent: str) -> str:
        """生成最终回答"""
        if intent == "diagnosis":
            prompt = f"""你是一位资深 Linux 运维专家。请根据以下运维知识库内容，诊断用户的问题并给出解决方案。

## 用户问题
{question}

## 知识库参考
{context}

## 要求
1. 先分析可能的原因（按可能性排序）
2. 给出逐步排查命令
3. 提供修复方案
4. 标注信息来源

请用中文回答，格式清晰。"""
        elif intent == "sop":
            prompt = f"""你是一位资深运维专家。请根据知识库内容，为用户生成标准操作程序(SOP)。

## 用户问题
{question}

## 知识库参考
{context}

## 要求
1. 列出前置条件
2. 给出详细步骤（含命令）
3. 注明回滚方案
4. 标注风险等级

请用中文回答。"""
        else:
            prompt = f"""你是一位资深 Linux 运维专家。请根据知识库内容回答用户问题。

## 用户问题
{question}

## 知识库参考
{context}

请用中文回答，引用相关来源。"""

        return self.llm.invoke(prompt)

    def _calculate_confidence(self, docs: list[dict]) -> float:
        """计算回答置信度"""
        if not docs:
            return 0.0
        avg_score = sum(doc["score"] for doc in docs) / len(docs)
        return min(1.0, avg_score * 1.2)

# conversation_manager.py
from collections import defaultdict

class ConversationManager:
    def __init__(self):
        self.sessions = defaultdict(list)

    def add_message(self, session_id: str, role: str, content: str):
        self.sessions[session_id].append({
            "role": role,
            "content": content,
            "timestamp": __import__("datetime").datetime.now().isoformat(),
        })

    def get_context(self, session_id: str, last_n: int = 6) -> list[dict]:
        """获取最近 N 条消息作为上下文"""
        messages = self.sessions[session_id]
        return messages[-last_n:]

    def summarize_session(self, session_id: str) -> str:
        """会话结束后生成摘要"""
        messages = self.sessions[session_id]
        if not messages:
            return ""
        questions = [m["content"] for m in messages if m["role"] == "user"]
        return f"本次会话共提出 {len(questions)} 个问题，主要涉及：{'、'.join(questions[:3])}"

# alert_processor.py
import requests

class AlertProcessor:
    def __init__(self, rag_engine: VPSOperationsRAG, vector_store: VectorStoreService):
        self.rag = rag_engine
        self.vector_store = vector_store

    def process_alert(self, alert: dict) -> dict:
        """处理一条告警，返回诊断结果和推荐 SOP"""
        alert_name = alert.get("name", "")
        alert_message = alert.get("message", "")
        metrics = alert.get("metrics", {})

        # 构建诊断查询
        diagnosis_query = f"""
        VPS 告警: {alert_name}
        告警详情: {alert_message}
        当前指标: CPU={metrics.get('cpu', 'N/A')}%, 
                  Memory={metrics.get('memory', 'N/A')}%, 
                  Disk={metrics.get('disk', 'N/A')}%
        请诊断原因并给出处理步骤。
        """

        # 调用 RAG 引擎
        result = self.rag.ask(diagnosis_query, context={"metrics": metrics})

        # 从知识库中检索相似历史故障
        similar_incidents = self.vector_store.search(
            f"历史故障 {alert_name} {alert_message}", top_k=3
        )

        # 生成 SOP
        sop_query = f"如何处理 {alert_name} 告警？请给出标准操作流程。"
        sop_result = self.rag.ask(sop_query)

        return {
            "alert": alert,
            "diagnosis": result["answer"],
            "similar_incidents": similar_incidents,
            "recommended_sop": sop_result["answer"],
            "confidence": result["confidence"],
            "auto_actions": self._suggest_auto_actions(alert, result),
        }

    def _suggest_auto_actions(self, alert: dict, diagnosis: dict) -> list[str]:
        """根据诊断结果建议自动操作"""
        actions = []
        answer = diagnosis.get("answer", "").lower()

        if "oom" in answer or "内存" in answer:
            actions.append("检查并清理大内存进程")
            actions.append("考虑临时增加 swap")
        if "磁盘" in answer or "disk" in answer:
            actions.append("清理旧日志文件")
            actions.append("检查是否有大文件占用")
        if "nginx" in answer or "端口" in answer:
            actions.append("重启 nginx 服务")
            actions.append("检查端口占用情况")

        return actions

# sop_recommender.py
from dataclasses import dataclass

@dataclass
class SOP:
    id: str
    title: str
    steps: list[dict]
    risk_level: str  # low, medium, high
    rollback_plan: str
    estimated_duration: str
    tags: list[str]

class SOPRecommender:
    def __init__(self, rag_engine: VPSOperationsRAG):
        self.rag = rag_engine

    def recommend(self, scenario: str) -> list[SOP]:
        """根据场景推荐 SOP"""
        query = f"""
        场景描述：{scenario}
        请从知识库中检索相关的标准操作流程(SOP)，
        并按优先级排序返回。
        """
        result = self.rag.ask(query, intent="sop")
        return self._parse_sop_result(result)

    def generate_sop(self, incident_description: str) -> SOP:
        """根据新故障自动生成 SOP"""
        query = f"""
        请根据以下故障描述，生成一份新的标准操作流程(SOP)：

        {incident_description}

        要求包含：
        1. 故障现象描述
        2. 排查步骤（带命令）
        3. 修复方案
        4. 验证方法
        5. 回滚方案
        6. 风险等级评估
        """
        result = self.rag.ask(query, intent="sop")
        return self._parse_generated_sop(result, incident_description)

    def _parse_sop_result(self, result: str) -> list[SOP]:
        """解析 SOP 推荐结果"""
        # 实际实现中会解析 LLM 输出为结构化 SOP
        return []

    def _parse_generated_sop(self, result: str, incident: str) -> SOP:
        """解析生成的 SOP"""
        return SOP(
            id=f"sop_{__import__('uuid').uuid4().hex[:8]}",
            title=f"针对 {incident[:50]} 的应急处理流程",
            steps=[],
            risk_level="medium",
            rollback_plan="执行回滚脚本并通知相关人员",
            estimated_duration="15-30分钟",
            tags=["ai-generated", "new-incident"],
        )

# web_ui.py
import gradio as gr
from rag_engine import VPSOperationsRAG

rag = VPSOperationsRAG()

def chat(message: str, history: list[list[str]]):
    """处理聊天消息"""
    if not message.strip():
        return "", history

    result = rag.ask(message)
    response = f"**置信度**: {result['confidence']:.0%}\n\n{result['answer']}\n\n---\n**信息来源**:\n"
    for i, source in enumerate(result["sources"], 1):
        response += f"{i}. {source.get('source', 'unknown')}\n"

    return response, history + [[message, response]]

demo = gr.ChatInterface(
    fn=chat,
    title="🔧 VPS 智能运维助手",
    description="基于 AI 知识库的运维问答系统，支持故障诊断、SOP 查询、配置指导。",
    examples=[
        "nginx 502 Bad Gateway 怎么处理？",
        "服务器内存使用率超过 90% 怎么办？",
        "如何配置 Prometheus 告警规则？",
        "SSH 被暴力破解了怎么办？",
    ],
    theme="soft",
)

if __name__ == "__main__":
    demo.launch(server_name="0.0.0.0", server_port=7860)

# telegram_bot.py
from telegram import Update
from telegram.ext import ApplicationBuilder, CommandHandler, MessageHandler, filters
from rag_engine import VPSOperationsRAG

rag = VPSOperationsRAG()

async def start(update: Update, context):
    await update.message.reply_text(
        "👋 欢迎使用 VPS 运维 AI 助手！\n\n"
        "你可以问我：\n"
        "• 故障诊断：描述问题，我帮你分析\n"
        "• SOP 查询：获取标准操作流程\n"
        "• 配置指导：任何运维配置问题\n\n"
        "/help 查看帮助"
    )

async def help_command(update: Update, context):
    await update.message.reply_text(
        "💡 使用提示：\n\n"
        "1. 直接发送问题即可\n"
        "2. 提供更多信息可获得更准确的回答\n"
        "3. 可以发送日志片段让我分析\n"
        "4. 输入 /reset 清空对话历史"
    )

async def handle_message(update: Update, context):
    message = update.message.text
    result = rag.ask(message)

    # 截断过长回复
    max_length = 4000
    if len(result["answer"]) > max_length:
        answer = result["answer"][:max_length] + "\n...（回复过长，已截断）"
    else:
        answer = result["answer"]

    await update.message.reply_text(
        f"🔍 **诊断结果** (置信度: {result['confidence']:.0%})\n\n"
        f"{answer}",
        parse_mode="Markdown"
    )

async def reset(update: Update, context):
    await update.message.reply_text("✅ 对话历史已清空")

def main():
    application = ApplicationBuilder().token("YOUR_BOT_TOKEN").build()

    application.add_handler(CommandHandler("start", start))
    application.add_handler(CommandHandler("help", help_command))
    application.add_handler(CommandHandler("reset", reset))
    application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))

    print("🤖 Telegram Bot 已启动")
    application.run_polling()

if __name__ == "__main__":
    main()

# feedback_loop.py
class FeedbackLoop:
    def __init__(self, vector_store: VectorStoreService):
        self.vector_store = vector_store

    def record_feedback(self, query: str, answer: str, rating: int, comment: str = ""):
        """记录用户对回答的反馈"""
        feedback_doc = {
            "type": "feedback",
            "query": query,
            "answer": answer,
            "rating": rating,  # 1-5
            "comment": comment,
            "timestamp": __import__("datetime").datetime.now().isoformat(),
        }

        # 低分回答触发知识更新
        if rating <= 2:
            self._trigger_knowledge_update(query, answer, comment)

    def _trigger_knowledge_update(self, query: str, answer: str, comment: str):
        """触发知识更新流程"""
        update_prompt = f"""
        用户对以下回答给出了低评分反馈。请分析原因并提出改进建议。

        用户问题: {query}
        原回答: {answer}
        用户反馈: {comment}

        请生成改进后的回答，并标注需要更新的知识库条目。
        """
        # 调用 LLM 生成改进建议
        # 并将改进后的内容加入知识库
        pass

    def auto_archive_resolution(self, incident: dict):
        """自动归档故障解决过程"""
        archive_entry = {
            "type": "resolved_incident",
            "title": incident["title"],
            "symptoms": incident["symptoms"],
            "root_cause": incident["root_cause"],
            "solution": incident["solution"],
            "commands_used": incident.get("commands", []),
            "duration": incident.get("duration", ""),
            "confidence": 0.95,
        }
        # 存入向量数据库
        self.vector_store.embed_and_store([archive_entry])

# knowledge_health.py
class KnowledgeHealthChecker:
    def check(self) -> dict:
        """检查知识库健康状况"""
        return {
            "total_documents": self._count_documents(),
            "stale_documents": self._find_stale_documents(),
            "low_confidence_queries": self._find_low_confidence_queries(),
            "missing_coverage": self._identify_coverage_gaps(),
            "recommendations": [],
        }

    def _find_stale_documents(self) -> list[str]:
        """找出过期的文档"""
        # 检查最后更新时间超过 90 天的文档
        pass

    def _find_low_confidence_queries(self) -> list[dict]:
        """找出低置信度的查询"""
        # 统计用户反馈低的查询
        pass

    def _identify_coverage_gaps(self) -> list[str]:
        """识别知识覆盖盲区"""
        # 分析高频但未匹配到知识的查询
        pass

# docker-compose.yml
version: '3.8'

services:
  ollama:
    image: ollama/ollama:latest
    ports:
      - "11434:11434"
    volumes:
      - ollama_data:/root/.ollama
    command: serve

  chromadb:
    image: chromadb/chroma:latest
    ports:
      - "8000:8000"
    volumes:
      - chroma_data:/chroma/chroma

  rag-engine:
    build: ./rag-engine
    ports:
      - "8080:8080"
    environment:
      - OLLAMA_BASE_URL=http://ollama:11434
      - CHROMADB_HOST=chromadb
      - CHROMADB_PORT=8000
    depends_on:
      - ollama
      - chromadb

  web-ui:
    build: ./web-ui
    ports:
      - "7860:7860"
    environment:
      - RAG_ENGINE_URL=http://rag-engine:8080
    depends_on:
      - rag-engine

  telegram-bot:
    build: ./telegram-bot
    environment:
      - BOT_TOKEN=${BOT_TOKEN}
      - RAG_ENGINE_URL=http://rag-engine:8080
    depends_on:
      - rag-engine

volumes:
  ollama_data:
  chroma_data:

#!/bin/bash
# init_knowledge_base.sh

echo "🚀 初始化 AI 运维知识库..."

# 1. 拉取所需模型
ollama pull qwen2.5:7b
ollama pull nomic-embed-text

# 2. 启动服务
docker-compose up -d

# 3. 导入基础文档
python3 scripts/import_knowledge.py \
  --source /opt/vps-docs/ \
  --source /var/log/syslog \
  --source /var/log/auth.log \
  --source https://wiki.internal.ops

# 4. 验证服务
sleep 10
curl -s http://localhost:8080/health | jq .
curl -s http://localhost:7860/api/info | jq .

echo "✅ 知识库初始化完成！"
echo "📊 Web UI: http://your-vps:7860"
echo "🤖 Telegram Bot: @YourOpsBot"

# security_config.py
class SecurityConfig:
    # 敏感信息脱敏
    SENSITIVE_PATTERNS = [
        (r'password\s*=\s*\S+', 'password=***'),
        (r'api[_-]?key\s*=\s*\S+', 'api_key=***'),
        (r'\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b', '[IP_REDACTED]'),
        (r'AKIA[0-9A-Z]{16}', 'AWS_KEY_REDACTED'),
    ]

    # 访问控制
    ALLOWED_ROLES = {
        "admin": ["read", "write", "delete", "admin"],
        "operator": ["read", "write"],
        "viewer": ["read"],
    }

    # 审计日志
    AUDIT_LOG_PATH = "/var/log/vps-kg-audit.log"