高级主题

深入探讨自动化系统的高级应用场景和技术。

📱 跨设备同步

macOS + iPhone/iPad

iCloud Drive同步

# 将自动化配置存储在iCloud
AUTOMATION_DIR=~/Library/Mobile\ Documents/com~apple~CloudDocs/Automation

# 同步脚本
ln -s "$AUTOMATION_DIR/scripts" ~/.automation/scripts

# 同步配置
ln -s "$AUTOMATION_DIR/config" ~/.automation/config

Shortcuts联动

// iPhone快速捕获 → Mac处理

// 1. iPhone Shortcuts
// 操作: 获取剪贴板 → 添加到"快速笔记"文件(iCloud)

// 2. Mac监听文件变化
// fswatch + 自动处理

cat > ~/Library/LaunchAgents/com.user.quicknote.plist << 'EOF'
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN"
"http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
    <key>Label</key>
    <string>com.user.quicknote</string>
    <key>ProgramArguments</key>
    <array>
        <string>/Users/username/.automation/scripts/process-quick-notes.sh</string>
    </array>
    <key>WatchPaths</key>
    <array>
        <string>/Users/username/Library/Mobile Documents/com~apple~CloudDocs/QuickNotes</string>
    </array>
</dict>
</plist>
EOF

# process-quick-notes.sh
#!/bin/bash
NOTES_FILE="$HOME/Library/Mobile Documents/com~apple~CloudDocs/QuickNotes/inbox.txt"

if [ -f "$NOTES_FILE" ] && [ -s "$NOTES_FILE" ]; then
    # 读取并处理
    while IFS= read -r line; do
        # AI分类和保存
        category=$(echo "$line" | claude-cli "Classify this note: idea, task, or info?")

        case $category in
            task)
                things://add?title="$line"
                ;;
            idea)
                echo "- $line" >> ~/Documents/Ideas/inbox.md
                ;;
            info)
                echo "$line" >> ~/Documents/Notes/misc.md
                ;;
        esac
    done < "$NOTES_FILE"

    # 清空文件
    > "$NOTES_FILE"
fi

Universal Clipboard

# 跨设备剪贴板历史

# macOS监听剪贴板
#!/bin/bash
# clipboard-sync.sh

LAST_CLIP=""
CLIP_HISTORY="$HOME/.automation/clipboard-history.txt"

while true; do
    CURRENT=$(pbpaste)

    if [ "$CURRENT" != "$LAST_CLIP" ]; then
        # 保存到iCloud
        echo "[$(date +%s)] $CURRENT" >> "$CLIP_HISTORY"

        # 同步到所有设备
        LAST_CLIP="$CURRENT"
    fi

    sleep 2
done

🌐 远程工作场景

VPN自动连接

#!/bin/bash
# auto-vpn.sh

function check_vpn() {
    # 检查是否在公司网络
    if ping -c 1 internal.company.com &> /dev/null; then
        echo "✅ 在公司网络"
        return 0
    else
        echo "❌ 不在公司网络"
        return 1
    fi
}

function connect_vpn() {
    echo "🔒 连接VPN..."

    # 使用networksetup或第三方VPN
    # 示例: Cisco AnyConnect
    /opt/cisco/anyconnect/bin/vpn -s connect company-vpn

    # 等待连接
    sleep 5

    if check_vpn; then
        echo "✅ VPN已连接"
        osascript -e 'display notification "VPN Connected" with title "🔒 Security"'
    else
        echo "❌ VPN连接失败"
        return 1
    fi
}

# 定期检查
while true; do
    if ! check_vpn; then
        connect_vpn
    fi

    sleep 300  # 5分钟检查一次
done

工作环境自动切换

#!/bin/bash
# location-aware-automation.sh

function detect_location() {
    # 通过WiFi SSID判断位置
    SSID=$(networksetup -getairportnetwork en0 | cut -d: -f2 | xargs)

    case "$SSID" in
        "Company-WiFi")
            echo "office"
            ;;
        "Home-WiFi")
            echo "home"
            ;;
        "Cafe-WiFi")
            echo "cafe"
            ;;
        *)
            echo "unknown"
            ;;
    esac
}

function setup_office_environment() {
    echo "🏢 设置办公室环境"

    # 断开VPN（在内网）
    # 连接办公室打印机
    # 设置工作配置文件
    git config --global user.email "work@company.com"

    # 打开办公工具
    open -a "Slack"
    open -a "Zoom"
}

function setup_home_environment() {
    echo "🏠 设置家庭环境"

    # 连接VPN
    connect_vpn

    # 使用个人配置
    git config --global user.email "personal@gmail.com"

    # 启动个人工具
}

# 监听网络变化
LOCATION=$(detect_location)

case "$LOCATION" in
    office)
        setup_office_environment
        ;;
    home)
        setup_home_environment
        ;;
esac

🤖 高级AI应用

本地LLM部署

使用Ollama

# 安装
brew install ollama

# 启动服务
ollama serve &

# 下载模型
ollama pull llama2           # 7B参数，适合日常
ollama pull codellama        # 代码专用
ollama pull mistral          # 速度快

# 使用
cat > ~/.automation/lib/local-ai.sh << 'EOF'
#!/bin/bash

function local_ai() {
    local prompt=$1
    local model=${2:-llama2}

    curl -s http://localhost:11434/api/generate -d "{
        \"model\": \"$model\",
        \"prompt\": \"$prompt\",
        \"stream\": false
    }" | jq -r '.response'
}

# 使用示例
local_ai "Explain this error: $error_message"
EOF

RAG (检索增强生成)

#!/usr/bin/env python3
# rag-system.py

import chromadb
from sentence_transformers import SentenceTransformer
import anthropic

class RAGSystem:
    def __init__(self):
        # 向量数据库
        self.client = chromadb.Client()
        self.collection = self.client.create_collection("knowledge_base")

        # 嵌入模型
        self.embedder = SentenceTransformer('all-MiniLM-L6-v2')

        # LLM
        self.claude = anthropic.Client()

    def add_document(self, text, metadata={}):
        """添加文档到知识库"""
        embedding = self.embedder.encode(text).tolist()

        self.collection.add(
            documents=[text],
            embeddings=[embedding],
            metadatas=[metadata],
            ids=[metadata.get('id', str(hash(text)))]
        )

    def query(self, question):
        """查询知识库并生成回答"""
        # 1. 检索相关文档
        query_embedding = self.embedder.encode(question).tolist()

        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=3
        )

        context = "\n\n".join(results['documents'][0])

        # 2. 生成回答
        message = self.claude.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            messages=[{
                "role": "user",
                "content": f"""基于以下上下文回答问题:

上下文:
{context}

问题: {question}

请基于上下文回答，如果上下文中没有相关信息，请说明。"""
            }]
        )

        return message.content[0].text

# 使用
rag = RAGSystem()

# 添加项目文档
rag.add_document(
    "项目使用React和TypeScript构建，部署在Vercel上。",
    {"source": "README.md", "type": "doc"}
)

# 查询
answer = rag.query("项目用什么技术栈？")
print(answer)

AI Agent系统

#!/usr/bin/env python3
# ai-agent.py

from anthropic import Anthropic
import subprocess
import json

class AutomationAgent:
    """智能自动化Agent"""

    def __init__(self):
        self.client = Anthropic()
        self.tools = self.define_tools()

    def define_tools(self):
        """定义Agent可用的工具"""
        return [
            {
                "name": "execute_command",
                "description": "执行shell命令",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "command": {"type": "string"}
                    }
                }
            },
            {
                "name": "read_file",
                "description": "读取文件内容",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "path": {"type": "string"}
                    }
                }
            },
            {
                "name": "write_file",
                "description": "写入文件",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "path": {"type": "string"},
                        "content": {"type": "string"}
                    }
                }
            }
        ]

    def execute_tool(self, tool_name, tool_input):
        """执行工具调用"""
        if tool_name == "execute_command":
            result = subprocess.run(
                tool_input["command"],
                shell=True,
                capture_output=True,
                text=True
            )
            return result.stdout

        elif tool_name == "read_file":
            with open(tool_input["path"]) as f:
                return f.read()

        elif tool_name == "write_file":
            with open(tool_input["path"], 'w') as f:
                f.write(tool_input["content"])
            return f"Written to {tool_input['path']}"

    def run(self, task):
        """运行Agent执行任务"""
        messages = [{
            "role": "user",
            "content": task
        }]

        while True:
            response = self.client.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=4096,
                tools=self.tools,
                messages=messages
            )

            # 检查是否需要使用工具
            if response.stop_reason == "tool_use":
                tool_use = next(
                    block for block in response.content
                    if block.type == "tool_use"
                )

                # 执行工具
                result = self.execute_tool(
                    tool_use.name,
                    tool_use.input
                )

                # 添加工具结果到对话
                messages.append({
                    "role": "assistant",
                    "content": response.content
                })
                messages.append({
                    "role": "user",
                    "content": [{
                        "type": "tool_result",
                        "tool_use_id": tool_use.id,
                        "content": result
                    }]
                })

            else:
                # 完成
                return response.content[0].text

# 使用
agent = AutomationAgent()

# 复杂任务
result = agent.run("""
帮我分析项目中的TODO注释，统计数量，并生成一个任务清单。
项目目录: ~/Developer/my-project
""")

print(result)

🔄 高级工作流引擎

n8n自托管

# docker-compose.yml
version: '3'

services:
  n8n:
    image: n8nio/n8n
    ports:
      - "5678:5678"
    environment:
      - N8N_BASIC_AUTH_ACTIVE=true
      - N8N_BASIC_AUTH_USER=admin
      - N8N_BASIC_AUTH_PASSWORD=password
      - N8N_HOST=localhost
      - N8N_PORT=5678
      - N8N_PROTOCOL=http
      - WEBHOOK_URL=http://localhost:5678/
    volumes:
      - ~/.n8n:/home/node/.n8n

使用场景:

• 监听GitHub webhooks自动部署
• 定时爬取数据并整理
• 跨平台数据同步
• 复杂的if-then-else逻辑

Temporal工作流

// 长时间运行的可靠工作流
package main

import (
    "time"
    "go.temporal.io/sdk/workflow"
)

func DeploymentWorkflow(ctx workflow.Context, env string) error {
    // 1. 运行测试 (可重试)
    err := workflow.ExecuteActivity(ctx,
        RunTests,
        workflow.ActivityOptions{
            StartToCloseTimeout: 10 * time.Minute,
            RetryPolicy: &RetryPolicy{
                MaximumAttempts: 3,
            },
        },
    ).Get(ctx, nil)

    if err != nil {
        return err
    }

    // 2. 构建 (幂等)
    err = workflow.ExecuteActivity(ctx, Build).Get(ctx, nil)
    if err != nil {
        return err
    }

    // 3. 部署 (可回滚)
    err = workflow.ExecuteActivity(ctx, Deploy, env).Get(ctx, nil)
    if err != nil {
        // 回滚
        workflow.ExecuteActivity(ctx, Rollback, env)
        return err
    }

    // 4. 健康检查 (等待30秒)
    workflow.Sleep(ctx, 30*time.Second)

    var healthy bool
    err = workflow.ExecuteActivity(ctx, HealthCheck, env).Get(ctx, &healthy)

    if !healthy {
        // 回滚
        workflow.ExecuteActivity(ctx, Rollback, env)
        return errors.New("health check failed")
    }

    return nil
}

🧪 高级测试策略

混沌工程

#!/bin/bash
# chaos-test.sh
# 测试自动化系统的韧性

echo "🔥 混沌测试开始"

# 1. 网络延迟
function test_network_delay() {
    echo "测试: 网络延迟500ms"

    # macOS: 使用pfctl
    sudo dnctl pipe 1 config delay 500

    # 运行自动化
    ./deploy.sh staging

    # 恢复
    sudo dnctl -q flush
}

# 2. API失败
function test_api_failure() {
    echo "测试: API随机失败"

    # 模拟API失败
    export CHAOS_API_FAILURE_RATE=0.5

    ./scripts/with-ai.sh
}

# 3. 磁盘空间不足
function test_disk_full() {
    echo "测试: 磁盘空间不足"

    # 创建大文件占用空间
    dd if=/dev/zero of=/tmp/bigfile bs=1G count=10

    # 运行依赖磁盘的操作
    ./scripts/backup.sh

    # 清理
    rm /tmp/bigfile
}

# 4. 并发冲突
function test_concurrent_execution() {
    echo "测试: 并发执行"

    # 同时运行10个实例
    for i in {1..10}; do
        ./scripts/concurrent-safe.sh &
    done

    wait
}

# 运行所有测试
test_network_delay
test_api_failure
test_disk_full
test_concurrent_execution

echo "✅ 混沌测试完成"

🌍 多语言环境

国际化支持

#!/bin/bash
# i18n-automation.sh

# 语言检测
LANG=${LANG:-en_US}
LANGUAGE=${LANGUAGE:-en}

# 翻译文件
TRANSLATIONS_DIR="$HOME/.automation/i18n"

function t() {
    local key=$1
    local lang_file="$TRANSLATIONS_DIR/$LANGUAGE.sh"

    if [ -f "$lang_file" ]; then
        source "$lang_file"
        echo "${!key}"
    else
        echo "$key"
    fi
}

# en_US.sh
MSG_STARTING="Starting deployment..."
MSG_SUCCESS="Deployment successful!"
MSG_FAILED="Deployment failed!"

# zh_CN.sh
MSG_STARTING="开始部署..."
MSG_SUCCESS="部署成功！"
MSG_FAILED="部署失败！"

# 使用
echo "$(t MSG_STARTING)"

📊 高级监控

Prometheus + Grafana

# 暴露指标
from prometheus_client import Counter, Histogram, start_http_server
import time

# 定义指标
automation_runs = Counter(
    'automation_runs_total',
    'Total automation runs',
    ['script', 'status']
)

automation_duration = Histogram(
    'automation_duration_seconds',
    'Automation duration',
    ['script']
)

# 启动HTTP服务器暴露指标
start_http_server(8000)

# 在脚本中使用
def run_automation(script_name):
    start = time.time()

    try:
        # 执行自动化
        execute(script_name)

        automation_runs.labels(script=script_name, status='success').inc()
    except Exception as e:
        automation_runs.labels(script=script_name, status='failed').inc()
        raise
    finally:
        duration = time.time() - start
        automation_duration.labels(script=script_name).observe(duration)

🔐 零信任架构

# 每次执行都验证
function secure_execute() {
    local script=$1

    # 1. 验证脚本签名
    if ! gpg --verify "$script.sig" "$script"; then
        echo "❌ 签名验证失败"
        return 1
    fi

    # 2. 检查权限
    if ! check_permissions "$script"; then
        echo "❌ 权限不足"
        return 1
    fi

    # 3. 审计日志
    audit_log "execute" "$script"

    # 4. 沙箱执行
    firejail --noprofile "$script"
}

---

提示: 高级主题需要更多的学习和实践。建议先掌握基础，再逐步探索这些高级应用。