3b7c00377f
Apply Black formatter to the entire codebase for consistent code style. Co-Authored-By: Claude (glm-5) <noreply@anthropic.com>
132 lines
3.7 KiB
Python
132 lines
3.7 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
GUI 工具模块
|
||
|
||
提供 GUI 相关的工具类和函数。
|
||
"""
|
||
|
||
import functools
|
||
|
||
|
||
class RealtimeOutput:
|
||
"""实时输出流，每次写入立即回调通知"""
|
||
|
||
def __init__(self, callback):
|
||
"""
|
||
初始化实时输出流
|
||
|
||
Args:
|
||
callback: 写入时的回调函数，接收文本内容
|
||
"""
|
||
self.callback = callback
|
||
self.buffer = []
|
||
|
||
def write(self, text):
|
||
"""写入文本"""
|
||
if text:
|
||
# 将文本按行分割，逐行回调
|
||
lines = text.split("\n")
|
||
for line in lines:
|
||
if line: # 忽略空行（由 split 产生）
|
||
self.callback(line)
|
||
return len(text)
|
||
|
||
def flush(self):
|
||
"""刷新（兼容性方法）"""
|
||
pass
|
||
|
||
def isatty(self):
|
||
"""返回 False，表示不是终端"""
|
||
return False
|
||
|
||
|
||
def admin_only(func):
|
||
"""
|
||
管理员权限装饰器
|
||
|
||
用于标记需要管理员权限的方法。如果当前用户不是管理员，
|
||
方法将不执行任何操作并返回 None。
|
||
|
||
Usage:
|
||
@admin_only
|
||
def some_admin_function(self):
|
||
# 只有管理员才能执行的代码
|
||
pass
|
||
|
||
Note:
|
||
- 被装饰的方法必须属于一个有 session_manager 属性的对象
|
||
- session_manager 必须有 is_admin() 方法
|
||
|
||
Returns:
|
||
如果是管理员，返回原函数的结果；否则返回 None
|
||
"""
|
||
|
||
@functools.wraps(func)
|
||
def wrapper(self, *args, **kwargs):
|
||
# 尝试从 self 获取 session_manager
|
||
session_manager = getattr(self, "session_manager", None)
|
||
|
||
# 如果没有 session_manager，尝试从 main_window 获取
|
||
if session_manager is None:
|
||
main_window = getattr(self, "main_window", None)
|
||
if main_window:
|
||
session_manager = getattr(main_window, "session_manager", None)
|
||
|
||
# 检查是否为管理员
|
||
if session_manager and hasattr(session_manager, "is_admin"):
|
||
if session_manager.is_admin():
|
||
return func(self, *args, **kwargs)
|
||
|
||
# 非管理员，记录日志并返回 None
|
||
logger = getattr(self, "logger", None)
|
||
if logger:
|
||
logger.debug(f"权限拒绝: {func.__name__} 需要管理员权限")
|
||
|
||
return None
|
||
|
||
return wrapper
|
||
|
||
|
||
def require_session(func):
|
||
"""
|
||
会话验证装饰器
|
||
|
||
确保方法执行时有有效的会话。如果会话无效，
|
||
方法将不执行任何操作并返回 None。
|
||
|
||
Usage:
|
||
@require_session
|
||
def some_function(self):
|
||
# 需要有效会话才能执行的代码
|
||
pass
|
||
|
||
Returns:
|
||
如果有有效会话，返回原函数的结果；否则返回 None
|
||
"""
|
||
|
||
@functools.wraps(func)
|
||
def wrapper(self, *args, **kwargs):
|
||
# 尝试从 self 获取 session_manager
|
||
session_manager = getattr(self, "session_manager", None)
|
||
|
||
# 如果没有 session_manager，尝试从 main_window 获取
|
||
if session_manager is None:
|
||
main_window = getattr(self, "main_window", None)
|
||
if main_window:
|
||
session_manager = getattr(main_window, "session_manager", None)
|
||
|
||
# 检查会话是否有效
|
||
if session_manager and hasattr(session_manager, "is_authenticated"):
|
||
if session_manager.is_authenticated():
|
||
return func(self, *args, **kwargs)
|
||
|
||
# 会话无效，记录日志并返回 None
|
||
logger = getattr(self, "logger", None)
|
||
if logger:
|
||
logger.warning(f"会话无效: {func.__name__} 需要有效会话")
|
||
|
||
return None
|
||
|
||
return wrapper
|