# Install Vector log collector
curl -fsSL https://sh.vector.dev | bash

# Configure collection of syslog, auth.log, nginx access/error logs
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 - Convert tickets to knowledge entries
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:
    """Extract symptom keywords from ticket description"""
    symptoms = []
    text = ticket["description"].lower()
    symptom_patterns = {
        "high_cpu": ["cpu", "load", "100%", "top"],
        "high_memory": ["memory", "oom", "swap", "mem"],
        "disk_full": ["disk", "no space", "capacity", "storage"],
        "network_down": ["network", "timeout", "unreachable", "connection"],
        "service_crash": ["crash", "restart", "reboot", "down"],
        "slow_response": ["slow", "latency", "timeout", "hang"],
        "auth_failure": ["login failed", "permission denied", "authentication"],
    }
    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):
        """Collect key events from 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):
        """Collect web documentation (Wiki, Confluence, etc.)"""
        # Use requests + BeautifulSoup
        pass

# chunker.py - Intelligent document chunking
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]:
        """Split documents into semantically coherent chunks"""
        chunks = []

        # First split by heading structure
        if document["type"] == "document" and document["content"].startswith("#"):
            header_chunks = self.header_splitter.split_text(document["content"])
        else:
            header_chunks = [document["content"]]

        # Then recursively chunk each part
        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]):
        """Batch embed and store"""
        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]:
        """Semantic search"""
        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,
                "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]:
        """Trace possible root causes from symptoms"""
        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]:
        """Find historical incidents related to a component"""
        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:
        """Handle ops Q&A"""
        # 1. Intent classification
        intent = self._classify_intent(question)

        # 2. Retrieve based on intent
        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. Build context
        context_text = self._build_context(relevant_docs)

        # 4. Generate answer
        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:
        """Classify user intent"""
        diagnosis_keywords = ["fault", "error", "exception", "down", "crash", "fail"]
        sop_keywords = ["steps", "procedure", "how to", "process", "sop", "runbook"]
        general_keywords = ["what is", "configure", "setup", "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]:
        """Diagnosis-specific retrieval: combine symptoms, metrics, and history"""
        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"configuration {config_query} tuning", top_k=3
            )
            docs.extend(config_docs)

        return docs

    def _build_context(self, docs: list[dict]) -> str:
        """Build context for the LLM"""
        parts = []
        for i, doc in enumerate(docs, 1):
            score_label = "High" if doc["score"] > 0.8 else "Medium" if doc["score"] > 0.6 else "Low"
            parts.append(f"[Source {i} | Relevance:{score_label}] {doc['content']}")
        return "\n\n".join(parts)

    def _generate_answer(self, question: str, context: str, intent: str) -> str:
        """Generate final answer"""
        if intent == "diagnosis":
            prompt = f"""You are a senior Linux operations engineer. Based on the following ops knowledge base content, diagnose the user's problem and provide a solution.

## User Question
{question}

## Knowledge Base Reference
{context}

## Requirements
1. Analyze possible causes (sorted by likelihood)
2. Provide step-by-step diagnostic commands
3. Offer fix solutions
4. Cite information sources

Respond in English with clear formatting."""
        elif intent == "sop":
            prompt = f"""You are a senior ops engineer. Based on the knowledge base, generate a Standard Operating Procedure (SOP) for the user.

## User Question
{question}

## Knowledge Base Reference
{context}

## Requirements
1. List prerequisites
2. Provide detailed steps (with commands)
3. Include rollback plan
4. Mark risk level

Respond in English."""
        else:
            prompt = f"""You are a senior Linux operations engineer. Answer the user's question based on the knowledge base.

## User Question
{question}

## Knowledge Base Reference
{context}

Respond in English, citing relevant sources."""

        return self.llm.invoke(prompt)

    def _calculate_confidence(self, docs: list[dict]) -> float:
        """Calculate answer confidence"""
        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]:
        """Get recent N messages as context"""
        messages = self.sessions[session_id]
        return messages[-last_n:]

    def summarize_session(self, session_id: str) -> str:
        """Generate summary after session ends"""
        messages = self.sessions[session_id]
        if not messages:
            return ""
        questions = [m["content"] for m in messages if m["role"] == "user"]
        return f"Session handled {len(questions)} questions, primarily covering: {'、'.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:
        """Process an alert and return diagnosis + recommended SOP"""
        alert_name = alert.get("name", "")
        alert_message = alert.get("message", "")
        metrics = alert.get("metrics", {})

        diagnosis_query = f"""
        VPS Alert: {alert_name}
        Alert Details: {alert_message}
        Current Metrics: CPU={metrics.get('cpu', 'N/A')}%, 
                  Memory={metrics.get('memory', 'N/A')}%, 
                  Disk={metrics.get('disk', 'N/A')}%
        Diagnose the cause and provide handling steps.
        """

        result = self.rag.ask(diagnosis_query, context={"metrics": metrics})

        similar_incidents = self.vector_store.search(
            f"historical incident {alert_name} {alert_message}", top_k=3
        )

        sop_query = f"How to handle {alert_name} alert? Provide standard operating procedure."
        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]:
        """Suggest auto actions based on diagnosis"""
        actions = []
        answer = diagnosis.get("answer", "").lower()

        if "oom" in answer or "memory" in answer:
            actions.append("Check and clean high-memory processes")
            actions.append("Consider temporarily increasing swap")
        if "disk" in answer or "storage" in answer:
            actions.append("Clean old log files")
            actions.append("Check for large files consuming space")
        if "nginx" in answer or "port" in answer:
            actions.append("Restart nginx service")
            actions.append("Check port occupancy")

        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]:
        """Recommend SOPs based on scenario"""
        query = f"""
        Scenario description: {scenario}
        Retrieve relevant Standard Operating Procedures (SOPs) from the knowledge base,
        sorted by priority.
        """
        result = self.rag.ask(query, intent="sop")
        return self._parse_sop_result(result)

    def generate_sop(self, incident_description: str) -> SOP:
        """Auto-generate SOP from a new incident"""
        query = f"""
        Generate a new Standard Operating Procedure (SOP) based on the following incident description:

        {incident_description}

        Must include:
        1. Incident description
        2. Diagnostic steps (with commands)
        3. Fix solution
        4. Verification method
        5. Rollback plan
        6. Risk level assessment
        """
        result = self.rag.ask(query, intent="sop")
        return self._parse_generated_sop(result, incident_description)

    def _parse_sop_result(self, result: str) -> list[SOP]:
        """Parse SOP recommendation result"""
        return []

    def _parse_generated_sop(self, result: str, incident: str) -> SOP:
        """Parse generated SOP"""
        return SOP(
            id=f"sop_{__import__('uuid').uuid4().hex[:8]}",
            title=f"Emergency procedure for {incident[:50]}",
            steps=[],
            risk_level="medium",
            rollback_plan="Execute rollback script and notify stakeholders",
            estimated_duration="15-30 minutes",
            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]]):
    """Handle chat messages"""
    if not message.strip():
        return "", history

    result = rag.ask(message)
    response = f"**Confidence**: {result['confidence']:.0%}\n\n{result['answer']}\n\n---\n**Sources**:\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 AI Ops Assistant",
    description="AI knowledge base-powered ops Q&A system supporting fault diagnosis, SOP queries, and configuration guidance.",
    examples=[
        "How to handle nginx 502 Bad Gateway?",
        "What to do when memory usage exceeds 90%?",
        "How to configure Prometheus alerting rules?",
        "What to do when SSH brute force is detected?",
    ],
    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(
        "👋 Welcome to the VPS Ops AI Assistant!\n\n"
        "You can ask me:\n"
        "• Fault diagnosis: Describe the issue, I'll help analyze\n"
        "• SOP queries: Get standard operating procedures\n"
        "• Configuration guidance: Any ops config question\n\n"
        "/help for help"
    )

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... (truncated)"
    else:
        answer = result["answer"]

    await update.message.reply_text(
        f"🔍 **Diagnosis Result** (Confidence: {result['confidence']:.0%})\n\n"
        f"{answer}",
        parse_mode="Markdown"
    )

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

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

    print("🤖 Telegram Bot started")
    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 = ""):
        """Record user feedback on answers"""
        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):
        """Trigger knowledge update flow"""
        update_prompt = f"""
        Users gave a low rating to the following answer. Analyze the reason and suggest improvements.

        User Question: {query}
        Original Answer: {answer}
        User Feedback: {comment}

        Generate an improved answer and mark knowledge base entries that need updating.
        """
        # Call LLM to generate improvement suggestions
        # And add improved content to the knowledge base
        pass

    def auto_archive_resolution(self, incident: dict):
        """Auto-archive incident resolution process"""
        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:
        """Check knowledge base health status"""
        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]:
        """Find stale documents"""
        pass

    def _find_low_confidence_queries(self) -> list[dict]:
        """Find low-confidence queries"""
        pass

    def _identify_coverage_gaps(self) -> list[str]:
        """Identify knowledge coverage gaps"""
        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=***
      - RAG_ENGINE_URL=http://rag-engine:8080
    depends_on:
      - rag-engine

volumes:
  ollama_data:
  chroma_data:

#!/bin/bash
# init_knowledge_base.sh

echo "🚀 Initializing AI Ops Knowledge Base..."

# 1. Pull required models
ollama pull qwen2.5:7b
ollama pull nomic-embed-text

# 2. Start services
docker-compose up -d

# 3. Import base documents
python3 scripts/import_knowledge.py \
  --source /opt/vps-docs/ \
  --source /var/log/syslog \
  --source /var/log/auth.log \
  --source https://wiki.internal.ops

# 4. Verify services
sleep 10
curl -s http://localhost:8080/health | jq .
curl -s http://localhost:7860/api/info | jq .

echo "✅ Knowledge base initialized!"
echo "📊 Web UI: http://your-vps:7860"
echo "🤖 Telegram Bot: @YourOpsBot"

# security_config.py
class SecurityConfig:
    # Sensitive data redaction
    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'),
    ]

    # Access control
    ALLOWED_ROLES = {
        "admin": ["read", "write", "delete", "admin"],
        "operator": ["read", "write"],
        "viewer": ["read"],
    }

    # Audit logging
    AUDIT_LOG_PATH = "/var/log/vps-kg-audit.log"