alist&Openlist驱动挂载检测
在很早之前就有想过这个需求，因为挂载的阿里云盘刷新令牌经常会掉，自己的很多需求又是用的阿里云盘的直链，苦于网上没有相关教程，自己又是小白一枚、这个需求就一直搁置中。
最近alist的风评不是很好，也就转到Openlist上，
加上现在的Ai写代码功能真的很好用，就写了这个需求满足自己所用
使用环境
- 飞牛OS
- 青龙面板（需要安装Python3依赖requests）依赖
- Openlist&Alist云盘
主要实现功能
- 定时检查失效驱动
- 邮件发送失效状态
- 没有失效驱动时不发送通知邮件（失效时才会发送通知）
代码内容
需要修改 # 配置信息内容 脚本最下方。
邮件我使用的是QQ邮箱，其他邮箱我没有测试是否可行，如果使用其他邮箱 自行测试（有报错的话，可以用Ai排查报错）
import requests
import json
import time
import smtplib
from email.mime.text import MIMEText
from email.header import Header
from email.utils import formataddr
from typing import Dict, List, Tuple, Optional
class AListDriverChecker:
def __init__(self, alist_url: str, token: str, email_config: Optional[Dict] = None):
"""初始化AList驱动检查器"""
self.alist_url = alist_url.rstrip('/')
self.headers = {
"Authorization": token,
"Content-Type": "application/json"
}
self.api_endpoints = {
"drivers": f"{self.alist_url}/api/admin/storage/list",
"test": f"{self.alist_url}/api/admin/storage/test",
"ls": f"{self.alist_url}/api/fs/list"
}
self.failed_drivers_log = []
self.valid_drivers_log = [] # 存储未失效驱动
self.email_config = email_config or {}
self.debug = True # 调试模式
# 定义有效的状态列表（不区分大小写）
self.valid_statuses = {"work", "disabled"}
# 状态中文映射
self.status_mapping = {
"work": "工作",
"disabled": "禁用"
}
# 驱动类型中文映射
self.driver_type_mapping = {
# 本地及基础存储
"local": "本机存储",
"smb": "SMB",
"ftp": "FTP",
"sftp": "SFTP",
"webdav": "WebDAV",
"strm": "Strm",
"alias": "别名",
# 云存储服务
"aliyundrive": "阿里云盘",
"aliyundrive_open": "阿里云盘 Open",
"115": "115 网盘 / 分享",
"115_open": "115 开放平台",
"onedrive": "OneDrive / 分享",
"onedrive_app": "OneDrive APP",
"ctyun": "电信天翼云盘",
"cucloud": "中国联通云盘",
"cmcloud": "中国移动云盘",
"scctyun": "四川电信魔盘",
"123pan": "123 网盘 / 分享 / 直链",
"123pan_open": "123 开放平台",
"baiduyun": "百度网盘",
"yidun": "一刻相册",
"lanzou": "蓝奏云盘",
"lanzou_vip": "蓝奏云优享版",
"xiaofeiji": "小飞机网盘",
"crypt": "Crypt（加密）",
"uss": "又拍云存储",
"s3": "对象存储（S3）",
"google": "谷歌云盘",
"google_photos": "谷歌相册",
"weiyun": "腾讯微云",
"teambition": "Teambition",
"pikpak": "PikPak / 分享",
"chaoxing": "超星星小组盘",
"xunlei": "迅雷云盘 / X / 浏览器",
"lenovo_share": "联想家庭储存链接分享",
"neteasemusic": "**云音乐云盘",
"**": "夸克网盘 / TV",
"uc": "UC / TV",
"addresstree": "地址树",
"fenmiao": "分秒帧",
"6pan": "6盘",
"mega": "MEGA Disk",
"seafile": "Seafile",
"febbox": "FebBox",
"kodbox": "可道云",
"dropbox": "Dropbox",
"yandex": "Yandex云盘",
"cloudreve_v3": "Cloudreve V3",
"cloudreve_v4": "Cloudreve V4 / 分享",
"misskey": "Misskey Drive",
"terabox": "Terabox（海外百度）",
"github": "GitHub API",
"github_release": "GitHub 发行版"
}
def get_all_drivers(self) -> List[Dict]:
"""获取所有已配置的存储驱动"""
try:
response = requests.get(
self.api_endpoints["drivers"],
headers=self.headers,
timeout=10
)
response.raise_for_status()
data = response.json()
if data.get("code") == 200:
if isinstance(data.get("data"), list):
return data.get("data", [])
return data.get("data", {}).get("content", [])
else:
print(f"获取驱动列表失败: {data.get('message', '未知错误')}")
return []
except Exception as e:
print(f"获取驱动列表时发生错误: {str(e)}")
return []
def get_driver_info(self, driver: Dict) -> Tuple[str, str, str, str]:
"""安全获取驱动信息，全面识别各类驱动类型"""
name = driver.get("name", "未命名") # 仅内部使用
mount_path = driver.get("mount_path") or driver.get("path", "未知路径")
# 多维度提取驱动类型标识
storage_type = "未知类型"
# 优先检查可能的驱动标识字段
if "driver" in driver:
storage_type = driver["driver"]
elif "type" in driver:
storage_type = driver["type"]
elif "storage" in driver and isinstance(driver["storage"], dict):
if "driver" in driver["storage"]:
storage_type = driver["storage"]["driver"]
elif "type" in driver["storage"]:
storage_type = driver["storage"]["type"]
elif "driver_type" in driver["storage"]:
storage_type = driver["storage"]["driver_type"]
elif "driver_type" in driver:
storage_type = driver["driver_type"]
# 标准化处理（去除空格、转为小写）
storage_type = storage_type.strip().lower()
status = str(driver.get("status", "未知状态"))
return name, mount_path, storage_type, status
def _get_status_display(self, status: str) -> str:
"""将状态转换为中文显示"""
status_lower = status.lower()
return self.status_mapping.get(status_lower, status)
def _get_driver_type_display(self, driver_type: str) -> str:
"""将驱动标识转换为易读中文名称"""
# 尝试精确匹配
if driver_type in self.driver_type_mapping:
return self.driver_type_mapping[driver_type]
# 尝试模糊匹配（处理可能的标识变体）
for key, value in self.driver_type_mapping.items():
if key in driver_type or driver_type in key:
return value
# 未匹配到则返回原始标识
return driver_type
def test_driver_connection(self, driver: Dict) -> Tuple[bool, str]:
"""测试驱动连接是否有效"""
_, _, storage_type, status = self.get_driver_info(driver)
status_lower = status.lower()
status_display = self._get_status_display(status)
type_display = self._get_driver_type_display(storage_type)
# 检查状态是否为有效状态
if status_lower in self.valid_statuses:
return (True, f"状态正常: {status_display}（{type_display}）")
# 状态无效时进行连接测试
try:
_, mount_path, _ = self.get_driver_info(driver)[:3]
if "id" in driver and "storage" in driver:
payload = {
"id": driver["id"],
"storage": driver["storage"]
}
response = requests.post(
self.api_endpoints["test"],
headers=self.headers,
data=json.dumps(payload),
timeout=15
)
response.raise_for_status()
data = response.json()
if data.get("code") == 200:
return (True, f"连接正常（{type_display}）")
else:
return (False, f"测试失败: {data.get('message', '未知错误')}（{type_display}）")
else:
raise ValueError("缺少测试所需的驱动信息")
except Exception as e:
try:
_, mount_path, _ = self.get_driver_info(driver)[:3]
path = f"/{mount_path.lstrip('/')}"
response = requests.get(
f"{self.api_endpoints['ls']}?path={path}",
headers=self.headers,
timeout=15
)
response.raise_for_status()
data = response.json()
if data.get("code") == 200:
return (True, f"连接正常 (通过文件列表验证)（{type_display}）")
else:
return (False, f"文件列表测试失败: {data.get('message', '未知错误')}（{type_display}）")
except Exception as e2:
return (False, f"连接失败: {str(e2)}（{type_display}）")
def check_all_drivers(self) -> List[Dict]:
"""检查所有驱动并返回结果"""
drivers = self.get_all_drivers()
results = []
self.failed_drivers_log = []
self.valid_drivers_log = [] # 重置未失效驱动列表
if not drivers:
print("没有找到任何存储驱动")
return results
print(f"找到 {len(drivers)} 个存储驱动，开始检测...\n")
for i, driver in enumerate(drivers, 1):
_, mount_path, storage_type, status = self.get_driver_info(driver)
status_lower = status.lower()
status_display = self._get_status_display(status)
type_display = self._get_driver_type_display(storage_type)
print(f"检测驱动 {i}/{len(drivers)}: {mount_path}（{type_display}） - 状态: {status_display}")
start_time = time.time()
is_valid, message = self.test_driver_connection(driver)
elapsed = round(time.time() - start_time, 2)
# 最终有效性判断
final_valid = is_valid and (status_lower in self.valid_statuses)
if not final_valid and status_lower in self.valid_statuses:
message = f"状态有效: {status_display}（{type_display}）"
result = {
"mount_path": mount_path,
"type": storage_type, # 原始驱动标识
"type_display": type_display, # 显示用的驱动名称
"status": status, # 原始状态
"status_display": status_display, # 显示用的状态
"is_valid": final_valid,
"message": message,
"response_time": elapsed
}
results.append(result)
# 分别记录未失效和失效驱动
if final_valid:
self.valid_drivers_log.append(result)
else:
self.failed_drivers_log.append(result)
status_text = "有效" if final_valid else "失效"
print(f" 状态: {status_text} | 响应时间: {elapsed}s | 消息: {message}\n")
time.sleep(0.5)
if self.debug:
print(f"\n调试信息: 发现{len(self.failed_drivers_log)}个失效驱动，{len(self.valid_drivers_log)}个未失效驱动")
print(f"调试信息: 邮件配置状态: {'已配置' if self.email_config else '未配置'}")
# 仅当存在失效驱动且邮件配置有效时才发送邮件
if self.failed_drivers_log and self.email_config:
self.send_notification_email()
elif self.debug:
print("调试信息: 没有检测到失效驱动，不发送邮件")
return results
def send_notification_email(self):
"""发送邮件通知，仅包含失效驱动信息"""
try:
if self.debug:
print("\n开始尝试发送邮件...")
smtp_server = self.email_config.get('smtp_server')
smtp_port = self.email_config.get('smtp_port', 587)
sender_email = self.email_config.get('sender')
sender_name = self.email_config.get('sender_name', 'AList驱动检测系统') # 发件人名称，默认值
password = self.email_config.get('password')
receiver = self.email_config.get('receiver')
if not all([smtp_server, sender_email, password, receiver]):
raise ValueError("邮件配置不完整")
if self.debug:
print(f"调试信息: 准备通过 {smtp_server}:{smtp_port} 发送邮件")
print(f"调试信息: 发件人: {sender_name} <{sender_email}>, 收件人: {receiver}")
total = len(self.valid_drivers_log) + len(self.failed_drivers_log)
subject = f"AList驱动检测报告: 发现{len(self.failed_drivers_log)}个失效驱动（共{total}个）"
# HTML邮件内容
html_content = f"""
<html>
<body>
<h2>AList驱动检测报告</h2>
<p>检测时间: {time.strftime('%Y-%m-%d %H:%M:%S')}</p>
<p>AList地址: <a href="{self.alist_url}" target="_blank">{self.alist_url}</a></p>
<p>检测总结: 共{total}个驱动（有效: {len(self.valid_drivers_log)}, 失效: {len(self.failed_drivers_log)}）</p>
<!-- 失效驱动部分 -->
<h3 style="color: #dc3545;">失效的驱动</h3>
"""
# 添加失效驱动表格
if self.failed_drivers_log:
html_content += f"""
<table border="1" cellpadding="8" style="margin-bottom: 20px; border-collapse: collapse;">
<tr style="background-color: #f8f9fa;">
<th>序号</th>
<th>路径</th>
<th>驱动类型</th>
<th>状态</th>
<th>原因</th>
<th>响应时间</th>
</tr>
"""
for i, driver in enumerate(self.failed_drivers_log, 1):
html_content += f"""
<tr>
<td>{i}</td>
<td>{driver['mount_path']}</td>
<td>{driver['type_display']}</td>
<td>{driver['status_display']}</td>
<td>{driver['message']}</td>
<td>{driver['response_time']}s</td>
</tr>
"""
html_content += "</table>"
else:
html_content += "<p>没有检测到失效的驱动</p>"
# 未失效驱动部分（可选显示）
html_content += """
<h3 style="color: #28a745; margin-top: 30px;">未失效的驱动</h3>
"""
if self.valid_drivers_log:
html_content += f"""
<table border="1" cellpadding="8" style="margin-bottom: 20px; border-collapse: collapse;">
<tr style="background-color: #f8f9fa;">
<th>序号</th>
<th>路径</th>
<th>驱动类型</th>
<th>状态</th>
<th>响应时间</th>
</tr>
"""
for i, driver in enumerate(self.valid_drivers_log, 1):
html_content += f"""
<tr>
<td>{i}</td>
<td>{driver['mount_path']}</td>
<td>{driver['type_display']}</td>
<td>{driver['status_display']}</td>
<td>{driver['response_time']}s</td>
</tr>
"""
html_content += "</table>"
else:
html_content += "<p>没有检测到未失效的驱动</p>"
html_content += """
</body>
</html>
"""
msg = MIMEText(html_content, 'html', 'utf-8')
# 正确设置发件人格式
msg['From'] = formataddr((sender_name, sender_email))
msg['To'] = receiver
msg['Subject'] = Header(subject, 'utf-8')
if self.debug:
print("调试信息: 正在连接到SMTP服务器...")
if smtp_port == 465:
server = smtplib.SMTP_SSL(smtp_server, smtp_port)
else:
server = smtplib.SMTP(smtp_server, smtp_port)
server.starttls()
if self.debug:
print("调试信息: 已连接到SMTP服务器，准备登录...")
server.login(sender_email, password)
recipients = receiver.split(',')
server.sendmail(sender_email, recipients, msg.as_string())
server.quit()
print(f"已成功发送邮件通知至 {receiver}")
if self.debug:
print("调试信息: 邮件发送成功")
except Exception as e:
print(f"发送邮件失败: {str(e)}")
if self.debug:
import traceback
print("详细错误信息:")
traceback.print_exc()
def print_failed_drivers_log(self):
"""输出失效驱动日志"""
if not self.failed_drivers_log:
print("\n没有发现失效的驱动")
return
print("\n" + "="*50)
print("失效驱动详细日志")
print("="*50)
for i, driver in enumerate(self.failed_drivers_log, 1):
print(f"{i}. 路径: {driver['mount_path']}")
print(f" 驱动类型: {driver['type_display']}")
print(f" 状态: {driver['status_display']}")
print(f" 原因: {driver['message']}")
print(f" 响应时间: {driver['response_time']}s")
print("-"*50)
def print_summary(self, results: List[Dict]):
"""打印检测结果摘要"""
if not results:
return
total = len(results)
valid = sum(1 for r in results if r["is_valid"])
invalid = total - valid
# 按驱动类型统计
type_stats = {}
for r in results:
type_key = r["type_display"]
if type_key not in type_stats:
type_stats[type_key] = {"total": 0, "valid": 0, "invalid": 0}
type_stats[type_key]["total"] += 1
if r["is_valid"]:
type_stats[type_key]["valid"] += 1
else:
type_stats[type_key]["invalid"] += 1
print("="*50)
print("检测结果摘要")
print("="*50)
print(f"总驱动数: {total}")
print(f"有效驱动: {valid}")
print(f"失效驱动: {invalid}\n")
print("按驱动类型统计:")
for type_name, stats in sorted(type_stats.items()):
print(f" {type_name}: 共{stats['total']}个 (有效: {stats['valid']}, 失效: {stats['invalid']})")
def test_email_config(self):
"""测试邮件配置"""
if not self.email_config:
print("未配置邮件信息，无法测试")
return
print("开始测试邮件配置...")
# 添加测试用的未失效驱动
self.valid_drivers_log = [{
"mount_path": "/test-valid",
"type": "local",
"type_display": self._get_driver_type_display("local"),
"status": "work",
"status_display": self._get_status_display("work"),
"is_valid": True,
"message": "状态正常",
"response_time": 0.2
}]
# 添加测试用的失效驱动
self.failed_drivers_log = [{
"mount_path": "/test-failed",
"type": "aliyundrive",
"type_display": self._get_driver_type_display("aliyundrive"),
"status": "error",
"status_display": "错误",
"is_valid": False,
"message": "连接超时",
"response_time": 5.0
}]
self.send_notification_email()
if __name__ == "__main__":
# 配置信息
AList_URL = "http://192.168.2.1:5244" # 替换为你的AList地址（内外网皆可）
AList_TOKEN = "*******************" # 替换为你的API令牌
email_config = {
"smtp_server": "smtp.qq.com",
"smtp_port": 465,
"sender": "******@qq.com",
"sender_name": "OpenList驱动监控系统", # 发件人名称
"password": "*******",
"receiver": "*******@qq.com"
}
checker = AListDriverChecker(AList_URL, AList_TOKEN, email_config)
# 测试邮件（可选）
# checker.test_email_config()
# 执行驱动检查
print("开始检查驱动...")
results = checker.check_all_drivers()
checker.print_summary(results)
checker.print_failed_drivers_log()
以上。
附件：openlist_monitor.pdf