306 lines
9.8 KiB
Python
306 lines
9.8 KiB
Python
"""
|
|
通知工具 - 支持桌面通知、声音提示等
|
|
"""
|
|
|
|
import platform
|
|
import subprocess
|
|
import logging
|
|
from typing import Optional
|
|
|
|
|
|
def send_notification(title: str, message: str, urgency: str = "normal") -> bool:
|
|
"""
|
|
发送桌面通知
|
|
|
|
Args:
|
|
title: 通知标题
|
|
message: 通知内容
|
|
urgency: 通知紧急程度 (low, normal, high)
|
|
|
|
Returns:
|
|
是否成功发送
|
|
"""
|
|
system = platform.system()
|
|
|
|
try:
|
|
if system == "Windows":
|
|
# Windows 使用 PowerShell 发送通知
|
|
ps_script = f"""
|
|
[Windows.UI.Notifications.ToastNotificationManager, Windows.UI.Notifications, ContentType=WindowsRuntime] > $null
|
|
$template = [Windows.UI.Notifications.ToastNotificationManager]::GetTemplateContent([Windows.UI.Notifications.ToastTemplateType]::ToastText02)
|
|
$template.SelectSingleNode("//text[@id='1']").AppendChild($template.CreateTextNode('{title}')) > $null
|
|
$template.SelectSingleNode("//text[@id='2']").AppendChild($template.CreateTextNode('{message}')) > $null
|
|
$toast = [Windows.UI.Notifications.ToastNotification]::new($template)
|
|
[Windows.UI.Notifications.ToastNotificationManager]::CreateToastNotifier("Screen2Feishu").Show($toast)
|
|
"""
|
|
|
|
# 使用 PowerShell 执行
|
|
subprocess.run(
|
|
["powershell", "-Command", ps_script],
|
|
capture_output=True,
|
|
timeout=5
|
|
)
|
|
return True
|
|
|
|
elif system == "Darwin": # macOS
|
|
# macOS 使用 osascript 发送通知
|
|
script = f'display notification "{message}" with title "{title}"'
|
|
subprocess.run(
|
|
["osascript", "-e", script],
|
|
capture_output=True,
|
|
timeout=5
|
|
)
|
|
return True
|
|
|
|
elif system == "Linux":
|
|
# Linux 使用 notify-send
|
|
try:
|
|
subprocess.run(
|
|
["notify-send", "-u", urgency, title, message],
|
|
capture_output=True,
|
|
timeout=5
|
|
)
|
|
return True
|
|
except FileNotFoundError:
|
|
# 如果 notify-send 不可用，尝试使用 zenity
|
|
try:
|
|
subprocess.run(
|
|
["zenity", "--info", "--title", title, "--text", message],
|
|
capture_output=True,
|
|
timeout=5
|
|
)
|
|
return True
|
|
except FileNotFoundError:
|
|
return False
|
|
else:
|
|
return False
|
|
|
|
except Exception as e:
|
|
logging.error(f"发送通知失败: {str(e)}")
|
|
return False
|
|
|
|
|
|
def send_notification_with_icon(title: str, message: str, icon_path: Optional[str] = None) -> bool:
|
|
"""
|
|
发送带图标的桌面通知
|
|
|
|
Args:
|
|
title: 通知标题
|
|
message: 通知内容
|
|
icon_path: 图标文件路径
|
|
|
|
Returns:
|
|
是否成功发送
|
|
"""
|
|
system = platform.system()
|
|
|
|
try:
|
|
if system == "Linux" and icon_path:
|
|
# Linux 支持图标
|
|
subprocess.run(
|
|
["notify-send", "-u", "normal", "-i", icon_path, title, message],
|
|
capture_output=True,
|
|
timeout=5
|
|
)
|
|
return True
|
|
else:
|
|
# 其他系统使用普通通知
|
|
return send_notification(title, message)
|
|
|
|
except Exception as e:
|
|
logging.error(f"发送带图标通知失败: {str(e)}")
|
|
return False
|
|
|
|
|
|
def play_sound(sound_type: str = "success") -> bool:
|
|
"""
|
|
播放系统声音
|
|
|
|
Args:
|
|
sound_type: 声音类型 (success, error, warning, info)
|
|
|
|
Returns:
|
|
是否成功播放
|
|
"""
|
|
system = platform.system()
|
|
|
|
try:
|
|
if system == "Windows":
|
|
# Windows 使用 PowerShell 播放系统声音
|
|
sound_map = {
|
|
"success": "SystemAsterisk",
|
|
"error": "SystemHand",
|
|
"warning": "SystemExclamation",
|
|
"info": "SystemDefault"
|
|
}
|
|
|
|
sound = sound_map.get(sound_type, "SystemDefault")
|
|
ps_script = f'[System.Media.SystemSounds]::{sound}.Play()'
|
|
|
|
subprocess.run(
|
|
["powershell", "-Command", ps_script],
|
|
capture_output=True,
|
|
timeout=5
|
|
)
|
|
return True
|
|
|
|
elif system == "Darwin": # macOS
|
|
# macOS 使用 afplay 播放系统声音
|
|
sound_map = {
|
|
"success": "/System/Library/Sounds/Purr.aiff",
|
|
"error": "/System/Library/Sounds/Basso.aiff",
|
|
"warning": "/System/Library/Sounds/Funk.aiff",
|
|
"info": "/System/Library/Sounds/Glass.aiff"
|
|
}
|
|
|
|
sound_file = sound_map.get(sound_type, "/System/Library/Sounds/Glass.aiff")
|
|
|
|
# 异步播放声音
|
|
subprocess.Popen(
|
|
["afplay", sound_file],
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL
|
|
)
|
|
return True
|
|
|
|
elif system == "Linux":
|
|
# Linux 使用 paplay 或 aplay 播放声音
|
|
# 这里使用简单的 beep 声音
|
|
try:
|
|
# 尝试使用 beep
|
|
subprocess.run(
|
|
["beep"],
|
|
capture_output=True,
|
|
timeout=5
|
|
)
|
|
return True
|
|
except FileNotFoundError:
|
|
# 如果 beep 不可用，尝试使用 paplay
|
|
try:
|
|
# 创建临时的声音文件
|
|
import tempfile
|
|
import wave
|
|
import struct
|
|
|
|
# 创建简单的正弦波声音
|
|
with tempfile.NamedTemporaryFile(suffix='.wav', delete=False) as f:
|
|
# 写入 WAV 文件头
|
|
sample_rate = 44100
|
|
duration = 0.2
|
|
num_samples = int(sample_rate * duration)
|
|
|
|
wav_file = wave.open(f.name, 'w')
|
|
wav_file.setnchannels(1)
|
|
wav_file.setsampwidth(2)
|
|
wav_file.setframerate(sample_rate)
|
|
|
|
# 生成正弦波
|
|
for i in range(num_samples):
|
|
value = int(32767 * 0.3 * (1 if i < num_samples // 2 else 0))
|
|
wav_file.writeframes(struct.pack('<h', value))
|
|
|
|
wav_file.close()
|
|
|
|
# 播放声音
|
|
subprocess.run(
|
|
["paplay", f.name],
|
|
capture_output=True,
|
|
timeout=5
|
|
)
|
|
|
|
# 删除临时文件
|
|
import os
|
|
os.unlink(f.name)
|
|
|
|
return True
|
|
except Exception:
|
|
return False
|
|
else:
|
|
return False
|
|
|
|
except Exception as e:
|
|
logging.error(f"播放声音失败: {str(e)}")
|
|
return False
|
|
|
|
|
|
def notify_task_processed(success: bool, file_name: str, count: int = 1) -> bool:
|
|
"""
|
|
通知任务处理结果
|
|
|
|
Args:
|
|
success: 是否成功
|
|
file_name: 文件名
|
|
count: 处理数量
|
|
|
|
Returns:
|
|
是否成功发送通知
|
|
"""
|
|
if success:
|
|
title = " 任务处理成功"
|
|
message = f"已成功处理 {count} 个文件: {file_name}"
|
|
sound_type = "success"
|
|
else:
|
|
title = " 任务处理失败"
|
|
message = f"处理文件失败: {file_name}"
|
|
sound_type = "error"
|
|
|
|
# 发送桌面通知
|
|
notification_sent = send_notification(title, message)
|
|
|
|
# 播放声音提示
|
|
sound_played = play_sound(sound_type)
|
|
|
|
return notification_sent or sound_played
|
|
|
|
|
|
def notify_batch_result(results: dict) -> bool:
|
|
"""
|
|
通知批量处理结果
|
|
|
|
Args:
|
|
results: 处理结果字典 {文件名: 是否成功}
|
|
|
|
Returns:
|
|
是否成功发送通知
|
|
"""
|
|
success_count = sum(1 for v in results.values() if v)
|
|
total_count = len(results)
|
|
|
|
if success_count == total_count:
|
|
title = " 批量处理完成"
|
|
message = f"所有 {total_count} 个文件处理成功"
|
|
sound_type = "success"
|
|
elif success_count == 0:
|
|
title = " 批量处理失败"
|
|
message = f"所有 {total_count} 个文件处理失败"
|
|
sound_type = "error"
|
|
else:
|
|
title = "⚠️ 批量处理部分成功"
|
|
message = f"{success_count}/{total_count} 个文件处理成功"
|
|
sound_type = "warning"
|
|
|
|
# 发送桌面通知
|
|
notification_sent = send_notification(title, message)
|
|
|
|
# 播放声音提示
|
|
sound_played = play_sound(sound_type)
|
|
|
|
return notification_sent or sound_played
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# 测试通知功能
|
|
import logging
|
|
logging.basicConfig(level=logging.INFO)
|
|
|
|
print("测试桌面通知...")
|
|
success = send_notification("测试通知", "这是一个测试通知消息")
|
|
print(f"通知发送: {'成功' if success else '失败'}")
|
|
|
|
print("测试声音播放...")
|
|
success = play_sound("success")
|
|
print(f"声音播放: {'成功' if success else '失败'}")
|
|
|
|
print("测试任务处理通知...")
|
|
success = notify_task_processed(True, "test_image.png")
|
|
print(f"任务通知: {'成功' if success else '失败'}") |