全功能备份系统配置指南

注：此教程开发初衷为备份哪吒面板，但是经测试可以支持所有目录备份至S3

一、环境准备

1. 系统要求

• ✅ Linux系统（Debian/Ubuntu/CentOS等）
• ✅ Python 3.6+
• ✅ 基础工具：tar, curl

2. 依赖安装（自动）

# 首次运行时会自动安装所需依赖

二、关键配置说明

1. 备份配置

BACKUP_DIR = "/opt/nezha"          # 监控目录（必须存在）
BACKUP_ROOT = "/tmp/nezha_backups" # 备份存储路径（自动创建）
MAX_DAYS = 7                       # 备份保留天数

2. S3存储配置（以AList为例）

S3_ENABLED = True
S3_ENDPOINT = "http://your-s3-endpoint:9000"  # S3服务地址
S3_BUCKET = "nezha-backups"        # 存储桶名称
S3_REGION = "us-east-1"            # 区域
S3_ACCESS_KEY = "your-access-key"  # 访问密钥
S3_SECRET_KEY = "your-secret-key"  # 私有密钥
S3_PATH = "auto-backups"           # 存储路径前缀

图2：AList密钥获取位置

3. 通知配置

飞书机器人

NOTIFY_FEISHU = True
FEISHU_WEBHOOK = "https://open.feishu.cn/open-apis/bot/v2/hook/xxx" # Webhook地址

邮件通知

NOTIFY_EMAIL = True
EMAIL_TO = "admin@example.com"     # 接收邮箱
EMAIL_FROM = "backup@example.com"  # 发送邮箱
SMTP_SERVER = "smtp.example.com"   # SMTP服务器
SMTP_PORT = 587                    # SMTP端口
SMTP_USER = "backup@example.com"   # SMTP用户名
SMTP_PASSWORD = "password"         # SMTP密码

Telegram通知

# Telegram配置
TG_BOT_TOKEN = "123456:ABC-DEF123" # Bot Token
TG_CHAT_ID = "123456789"           # 聊天ID

三、部署步骤

1. 复制脚本脚本至nezha_backup.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Nezha 全系统兼容备份脚本
功能：备份→压缩→上传→通知→清理
1. 自动检测系统类型并安装依赖
2. 打包指定目录到压缩文件
3. 支持上传到S3兼容存储
4. 多通道通知（邮件/Telegram/飞书）
5. 自动清理旧备份
"""

import os
import sys
import tarfile
import json
import subprocess
import datetime
from pathlib import Path
import smtplib
from email.mime.text import MIMEText
import requests
import boto3
from botocore.exceptions import ClientError, NoCredentialsError
import socket
import ssl

# ==================== 配置区域 ====================
class Config:
    # 备份配置
    BACKUP_DIR = "/opt/nezha"          # 要备份的目录
    BACKUP_ROOT = "/tmp/nezha_backups" # 备份文件存放目录
    MAX_DAYS = 7                       # 保留多少天的备份
    
    # S3配置 (AList/MinIO等兼容S3的服务)
    S3_ENABLED = True
    S3_ENDPOINT = "http://your-s3-endpoint:9000"  # S3服务地址
    S3_BUCKET = "nezha-backups"        # 存储桶名称
    S3_REGION = "us-east-1"            # 区域
    S3_ACCESS_KEY = "your-access-key"  # 访问密钥
    S3_SECRET_KEY = "your-secret-key"  # 私有密钥
    S3_PATH = "auto-backups"           # 存储路径前缀
    
    # 通知配置
    NOTIFY_TELEGRAM = False            # 启用Telegram通知
    NOTIFY_EMAIL = True                # 启用邮件通知
    NOTIFY_FEISHU = False              # 启用飞书通知
    
    # Telegram配置
    TG_BOT_TOKEN = "123456:ABC-DEF123" # Bot Token
    TG_CHAT_ID = "123456789"           # 聊天ID
    
    # 邮件配置
    EMAIL_TO = "admin@example.com"     # 接收邮箱
    EMAIL_FROM = "backup@example.com"  # 发送邮箱
    SMTP_SERVER = "smtp.example.com"   # SMTP服务器
    SMTP_PORT = 587                    # SMTP端口
    SMTP_USER = "backup@example.com"   # SMTP用户名
    SMTP_PASSWORD = "password"         # SMTP密码
    
    # 飞书配置
    FEISHU_WEBHOOK = "https://open.feishu.cn/..." # Webhook地址

# ==================== 网络诊断工具 ====================
class NetworkTester:
    @staticmethod
    def test_connection(url, port=443, timeout=5):
        """测试网络连通性"""
        try:
            sock = socket.create_connection((url, port), timeout)
            if port == 443:
                context = ssl.create_default_context()
                sock = context.wrap_socket(sock, server_hostname=url)
            sock.close()
            return True
        except Exception as e:
            print(f"连接测试失败 {url}:{port} - {str(e)}")
            return False

    @classmethod
    def check_all(cls):
        """检查所有依赖服务的网络连通性"""
        services = {
            "S3存储": (Config.S3_ENDPOINT.split('//')[1].split(':')[0], 
                     int(Config.S3_ENDPOINT.split(':')[-1])),
            "飞书API": ("open.feishu.cn", 443),
            "Telegram": ("api.telegram.org", 443),
            "SMTP服务": (Config.SMTP_SERVER, Config.SMTP_PORT)
        }
        
        print("=== 网络连通性测试 ===")
        for name, (host, port) in services.items():
            status = "✅" if cls.test_connection(host, port) else "❌"
            print(f"{status} {name.ljust(8)} {host}:{port}")

# ==================== 强化版通知模块 ==================== 
class Notifier:
    @staticmethod
    def _format_content(backup, extra_msg=""):
        """生成标准化通知内容"""
        base = f"""
🛡️ Nezha备份报告
----------------------------
🏷️ 服务器: {os.uname().nodename}
📅 时间: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
📦 备份文件: {os.path.basename(backup.backup_file)}
📏 大小: {backup.backup_size}
"""
        if hasattr(backup, 's3_url'):
            base += f"🌐 S3位置: {backup.s3_url}\n"
        return base + f"📝 详情: {extra_msg}\n"

    @classmethod
    def send_feishu(cls, backup, status, extra_msg=""):
        if not Config.NOTIFY_FEISHU:
            return False

        try:
            # 强化卡片消息结构（兼容飞书新版API）
            card = {
                "config": {"wide_screen_mode": True},
                "header": {
                    "title": {"tag": "plain_text", "content": f"备份{status}"},
                    "template": "green" if "成功" in status else "red"
                },
                "elements": [
                    {"tag": "markdown", "content": cls._format_content(backup, extra_msg)},
                    {"tag": "hr"},
                    {"tag": "note", "elements": [
                        {"tag": "plain_text", "content": "来自Nezha备份系统"}
                    ]}
                ]
            }

            # 双重发送机制（应对网络抖动）
            for retry in range(2):
                try:
                    resp = requests.post(
                        Config.FEISHU_WEBHOOK,
                        json={"msg_type": "interactive", "card": card},
                        timeout=10,
                        verify='/etc/ssl/certs/ca-certificates.crt'
                    )
                    if resp.status_code == 200 and resp.json().get("StatusCode") == 0:
                        return True
                except requests.exceptions.SSLError:
                    resp = requests.post(Config.FEISHU_WEBHOOK, json={"msg_type": "text", "content": {"text": cls._format_content(backup, extra_msg)}}, verify=False)
                    return resp.status_code == 200

            print(f"飞书通知失败 最终响应: {resp.text}")
            return False
        except Exception as e:
            print(f"飞书通知异常: {str(e)}")
            return False

    @classmethod
    def send_telegram(cls, backup, status, extra_msg=""):
        if not Config.NOTIFY_TELEGRAM:
            return False

        try:
            text = f"*Nezha备份{status}*" + \
                  cls._format_content(backup, extra_msg).replace('\n', '\n• ')
            
            # 使用Session保持连接
            with requests.Session() as s:
                resp = s.post(
                    f"https://api.telegram.org/bot{Config.TG_BOT_TOKEN}/sendMessage",
                    json={
                        "chat_id": Config.TG_CHAT_ID,
                        "text": text,
                        "parse_mode": "Markdown",
                        "disable_web_page_preview": True
                    },
                    timeout=15
                )
                return resp.json().get('ok', False)
        except Exception as e:
            print(f"Telegram通知异常: {str(e)}")
            return False

    @classmethod
    def send_email(cls, backup, status, extra_msg=""):
        if not Config.NOTIFY_EMAIL:
            return False

        try:
            msg = MIMEText(cls._format_content(backup, extra_msg))
            msg['Subject'] = f"Nezha备份{status}"
            msg['From'] = Config.EMAIL_FROM
            msg['To'] = Config.EMAIL_TO
            
            # 自动识别SSL/TLS
            if Config.SMTP_PORT == 465:
                with smtplib.SMTP_SSL(Config.SMTP_SERVER, Config.SMTP_PORT) as server:
                    server.login(Config.SMTP_USER, Config.SMTP_PASSWORD)
                    server.send_message(msg)
            else:
                with smtplib.SMTP(Config.SMTP_SERVER, Config.SMTP_PORT) as server:
                    server.starttls()
                    server.login(Config.SMTP_USER, Config.SMTP_PASSWORD)
                    server.send_message(msg)
            return True
        except Exception as e:
            print(f"邮件发送异常: {str(e)}")
            return False

    @classmethod
    def notify_all(cls, backup, status, extra_msg=""):
        """执行所有通知发送（带错误隔离）"""
        print("\n=== 开始发送通知 ===")
        NetworkTester.check_all()
        
        results = {
            '飞书': cls.send_feishu(backup, status, extra_msg),
            'Telegram': cls.send_telegram(backup, status, extra_msg),
            '邮件': cls.send_email(backup, status, extra_msg)
        }
        
        print("通知发送结果:")
        for name, success in results.items():
            print(f"  {name}: {'✅成功' if success else '❌失败'}")
        
        return any(results.values())

# ==================== 核心备份模块 ====================
class NezhaBackup:
    def __init__(self):
        self.backup_file = ""
        self.backup_size = "0"
        self._setup()

    def _setup(self):
        """初始化环境"""
        Path(Config.BACKUP_ROOT).mkdir(parents=True, exist_ok=True)
        self._install_deps()

    def _install_deps(self):
        """安全安装依赖"""
        try:
            import boto3, requests
        except ImportError:
            print("正在安全安装依赖...")
            subprocess.run([
                sys.executable, "-m", "pip", "install", 
                "--user", "boto3", "requests", "urllib3"
            ], check=True)

    def create_backup(self):
        """创建压缩备份"""
        try:
            timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
            self.backup_file = os.path.join(
                Config.BACKUP_ROOT, 
                f"nezha_{timestamp}.tar.gz"
            )
            
            with tarfile.open(self.backup_file, "w:gz") as tar:
                tar.add(Config.BACKUP_DIR, arcname="nezha_data")
            
            self.backup_size = self._human_size(os.path.getsize(self.backup_file))
            return True
        except Exception as e:
            Notifier.notify_all(self, "失败", f"备份创建错误: {str(e)}")
            return False

    def upload_to_s3(self):
        """上传到S3存储"""
        if not Config.S3_ENABLED:
            return True

        try:
            s3 = boto3.client(
                's3',
                endpoint_url=Config.S3_ENDPOINT,
                aws_access_key_id=Config.S3_ACCESS_KEY,
                aws_secret_access_key=Config.S3_SECRET_KEY,
                region_name=Config.S3_REGION,
                config=boto3.session.Config(
                    signature_version='s3v4',
                    connect_timeout=30,
                    read_timeout=60,
                    retries={'max_attempts': 3}
                )
            )

            object_name = f"{Config.S3_PATH}/{os.path.basename(self.backup_file)}"
            print(f"正在上传到S3: {Config.S3_BUCKET}/{object_name}")
            
            s3.upload_file(
                Filename=self.backup_file,
                Bucket=Config.S3_BUCKET,
                Key=object_name,
                ExtraArgs={'ACL': 'bucket-owner-full-control'}
            )
            
            self.s3_url = f"{Config.S3_ENDPOINT}/{Config.S3_BUCKET}/{object_name}"
            return True
        except Exception as e:
            Notifier.notify_all(self, "失败", f"S3上传错误: {str(e)}")
            return False

    def clean_old(self):
        """清理旧备份"""
        try:
            cutoff = datetime.datetime.now() - datetime.timedelta(days=Config.MAX_DAYS)
            for f in Path(Config.BACKUP_ROOT).glob("nezha_*.tar.gz"):
                if datetime.datetime.fromtimestamp(f.stat().st_mtime) < cutoff:
                    try:
                        f.unlink()
                        print(f"已清理: {f.name}")
                    except Exception as e:
                        print(f"清理失败 {f.name}: {str(e)}")
        except Exception as e:
            print(f"清理过程出错: {str(e)}")

    def _human_size(self, size):
        """转换文件大小"""
        for unit in ['B', 'KB', 'MB', 'GB']:
            if size < 1024.0:
                return f"{size:.1f}{unit}"
            size /= 1024.0
        return f"{size:.1f}TB"

# ==================== 主程序 ====================
def main():
    print("===== Nezha超级备份开始 =====")
    backup = NezhaBackup()
    
    # 执行备份流程
    if not backup.create_backup():
        sys.exit(1)
    
    upload_status = "成功" if backup.upload_to_s3() else "失败"
    backup.clean_old()
    
    # 发送最终通知
    Notifier.notify_all(
        backup,
        upload_status,
        "✅ 所有操作已完成" if upload_status == "成功" else "❌ 请检查错误日志"
    )
    
    print(f"\n===== 备份任务完成 [{upload_status}] =====")

if __name__ == "__main__":
    main()

2.给文件运行权限

chmod +x ./nezha_backup.py

2. 测试运行

# 普通测试模式
./nezha_backup.py

# 调试模式（显示详细输出）
DEBUG=1 ./nezha_backup.py

图3：成功运行输出示例

3. 设置定时任务

# 每天北京时间6点执行
(crontab -l ; echo "0 22 * * * /usr/bin/python3 ./nezha_backup.py >> /var/log/nezha_backup.log 2>&1") | crontab -

四、强制测试通知方法

1. 独立测试飞书通知

python3 -c "
from nezha_backup import Notifier, NezhaBackup
b = NezhaBackup()
b.backup_file = '/tmp/test.tar.gz'
b.backup_size = '1.2GB'
Notifier.send_feishu(b, '测试', '这是手动触发的测试消息')
"

2. 测试所有通知渠道

python3 -c "
from nezha_backup import Notifier, NezhaBackup, Config
Config.NOTIFY_FEISHU = True
Config.NOTIFY_EMAIL = True
Config.NOTIFY_TELEGRAM = True

b = NezhaBackup()
b.backup_file = '/tmp/test.tar.gz'
b.backup_size = '1.5GB'
b.s3_url = 's3://nezha-backups/test.tar.gz'

Notifier.notify_all(b, '测试', '全渠道通知测试')
"

图4：强制测试的输出示例

五、网络诊断技巧

1. 手动检查连通性

# 测试飞书API
curl -I https://open.feishu.cn/open-apis/bot/v2/hook/your-key

# 测试S3连接
aws --endpoint-url http://your-s3-endpoint:5246 s3 ls s3://your-bucket

2. 使用内置诊断工具

python3 -c "from nezha_backup import NetworkTester; NetworkTester.check_all()"

六、常见问题解决

七、通知效果预览

飞书通知卡片

邮件通知

最终效果

附录：

• 飞书机器人文档
• AList S3配置指南
• 脚本GitHub仓库

提示：所有敏感信息（密码/密钥）应通过环境变量传入，不要直接写在脚本中

文档配套图片清单

1. 架构图：使用Draw.io绘制备份流程（压缩→上传→通知）
2. AList截图：红框标注密钥获取位置
3. 终端输出：展示首次成功运行的彩色输出
4. 通知示例：
   • 飞书卡片消息（含成功/失败状态）
   • 邮件纯文本格式
5. 网络测试：NetworkTester.check_all()的输出示例

建议图片规格：

• 格式：PNG（无损）或WebP（压缩）
• 尺寸：宽度≥800px
• 标注：关键配置项用红色方框/箭头标注