282 lines
10 KiB
Python
282 lines
10 KiB
Python
import os
|
||
import sys
|
||
import time
|
||
import logging
|
||
import threading
|
||
from watchdog.observers import Observer
|
||
from watchdog.events import FileSystemEventHandler
|
||
|
||
# ==================== 自动配置区 ====================
|
||
# 自动检测ComfyUI根目录（从utils/workflow目录向上查找）
|
||
current_dir = os.path.dirname(os.path.abspath(__file__))
|
||
UTILS_ROOT = os.path.dirname(current_dir) # utils目录
|
||
PLUGIN_ROOT = os.path.dirname(UTILS_ROOT) # ComfyUI-1hewNodes目录
|
||
COMFYUI_ROOT = os.path.dirname(os.path.dirname(PLUGIN_ROOT)) # 从ComfyUI-1hewNodes向上两级到ComfyUI根目录
|
||
|
||
# 自动检测可能的工作流目录
|
||
POSSIBLE_WORKFLOW_DIRS = [
|
||
os.path.join(COMFYUI_ROOT, "user", "default", "workflows"),
|
||
os.path.join(COMFYUI_ROOT, "workflows"),
|
||
os.path.join(COMFYUI_ROOT, "user", "workflows"),
|
||
os.path.join(os.path.expanduser("~"), "ComfyUI", "workflows"),
|
||
]
|
||
|
||
# 临时文件路径（保存在utils/workflow文件夹中）
|
||
TEMP_FILE_PATH = os.path.join(current_dir, "current_workflow.tmp")
|
||
|
||
# 监控日志级别（DEBUG/INFO/WARNING）
|
||
LOG_LEVEL = logging.INFO
|
||
# ================================================
|
||
|
||
# 配置日志
|
||
logging.basicConfig(
|
||
level=LOG_LEVEL,
|
||
format="%(asctime)s - %(levelname)s - %(message)s",
|
||
datefmt="%H:%M:%S"
|
||
)
|
||
|
||
|
||
def _safe_log_text(value):
|
||
"""Keep workflow paths intact, but make console logging encoding-safe."""
|
||
text = str(value)
|
||
encoding = getattr(getattr(sys, "stdout", None), "encoding", None) or "utf-8"
|
||
try:
|
||
text.encode(encoding)
|
||
return text
|
||
except UnicodeEncodeError:
|
||
return text.encode(encoding, errors="backslashreplace").decode(encoding, errors="replace")
|
||
|
||
class WorkflowFileHandler(FileSystemEventHandler):
|
||
def __init__(self):
|
||
self.last_updated = 0 # 用于过滤短时间内的重复触发
|
||
self.min_interval = 0.5 # 减少最小触发间隔（秒）
|
||
self.last_file = "" # 记录上次处理的文件，避免重复处理
|
||
self.workflow_root = None # 缓存工作流根目录
|
||
|
||
def _get_workflow_root(self):
|
||
"""获取当前使用的工作流根目录"""
|
||
if self.workflow_root is None:
|
||
for workflow_dir in POSSIBLE_WORKFLOW_DIRS:
|
||
if os.path.exists(workflow_dir) and os.path.isdir(workflow_dir):
|
||
self.workflow_root = workflow_dir
|
||
break
|
||
return self.workflow_root
|
||
|
||
def _get_relative_path(self, file_path):
|
||
"""获取相对于workflows目录的路径"""
|
||
workflow_root = self._get_workflow_root()
|
||
if workflow_root is None:
|
||
return os.path.basename(file_path)
|
||
|
||
# 标准化路径分隔符
|
||
file_path = os.path.abspath(file_path).replace("\\", "/")
|
||
workflow_root = os.path.abspath(workflow_root).replace("\\", "/")
|
||
|
||
# 检查文件是否在workflows目录下
|
||
if file_path.startswith(workflow_root):
|
||
# 获取相对路径
|
||
relative_path = file_path[len(workflow_root):].lstrip("/")
|
||
return relative_path
|
||
else:
|
||
# 如果不在workflows目录下，只返回文件名
|
||
return os.path.basename(file_path)
|
||
|
||
def _update_temp_file(self, file_path, event_type="unknown"):
|
||
"""更新临时文件的通用方法"""
|
||
current_time = time.time()
|
||
relative_path = self._get_relative_path(file_path)
|
||
|
||
# 避免重复处理同一文件
|
||
if (current_time - self.last_updated > self.min_interval
|
||
and relative_path != self.last_file):
|
||
|
||
self.last_updated = current_time
|
||
self.last_file = relative_path
|
||
|
||
# 写入临时文件
|
||
try:
|
||
with open(TEMP_FILE_PATH, "w", encoding="utf-8") as f:
|
||
f.write(relative_path)
|
||
logging.info(f"[{event_type}] 已更新工作流路径：{_safe_log_text(relative_path)}")
|
||
except Exception as e:
|
||
logging.error(f"写入临时文件失败：{str(e)}")
|
||
|
||
def on_modified(self, event):
|
||
"""处理文件修改事件（保存操作会触发）"""
|
||
if (not event.is_directory
|
||
and event.src_path.endswith(".json")
|
||
and os.path.abspath(event.src_path) != os.path.abspath(TEMP_FILE_PATH)):
|
||
self._update_temp_file(event.src_path, "修改")
|
||
|
||
def on_moved(self, event):
|
||
"""处理文件移动/重命名事件"""
|
||
if (not event.is_directory
|
||
and event.dest_path.endswith(".json")
|
||
and os.path.abspath(event.dest_path) != os.path.abspath(TEMP_FILE_PATH)):
|
||
self._update_temp_file(event.dest_path, "移动")
|
||
|
||
def on_created(self, event):
|
||
"""处理文件创建事件"""
|
||
if (not event.is_directory
|
||
and event.src_path.endswith(".json")
|
||
and os.path.abspath(event.src_path) != os.path.abspath(TEMP_FILE_PATH)):
|
||
self._update_temp_file(event.src_path, "创建")
|
||
|
||
def on_opened(self, event):
|
||
"""处理文件打开事件（如果支持）"""
|
||
if (hasattr(event, 'src_path') and not event.is_directory
|
||
and event.src_path.endswith(".json")
|
||
and os.path.abspath(event.src_path) != os.path.abspath(TEMP_FILE_PATH)):
|
||
self._update_temp_file(event.src_path, "打开")
|
||
|
||
def find_workflow_directory():
|
||
"""自动查找工作流目录"""
|
||
for workflow_dir in POSSIBLE_WORKFLOW_DIRS:
|
||
if os.path.exists(workflow_dir) and os.path.isdir(workflow_dir):
|
||
# logging.info(f"找到工作流目录：{workflow_dir}")
|
||
return workflow_dir
|
||
|
||
# 如果都不存在，创建默认目录
|
||
default_dir = POSSIBLE_WORKFLOW_DIRS[0]
|
||
try:
|
||
os.makedirs(default_dir, exist_ok=True)
|
||
logging.info(f"创建默认工作流目录：{default_dir}")
|
||
return default_dir
|
||
except Exception as e:
|
||
logging.error(f"无法创建工作流目录：{str(e)}")
|
||
return None
|
||
|
||
def validate_workflow_dir():
|
||
"""验证工作流目录是否存在"""
|
||
workflow_root = find_workflow_directory()
|
||
if workflow_root is None:
|
||
logging.error("无法找到或创建工作流目录")
|
||
logging.info("可能的目录位置：")
|
||
for dir_path in POSSIBLE_WORKFLOW_DIRS:
|
||
logging.info(f" - {dir_path}")
|
||
return False, None
|
||
return True, workflow_root
|
||
|
||
# 全局变量管理监控状态
|
||
_observer = None
|
||
_monitor_thread = None
|
||
_is_monitoring = False
|
||
|
||
def _monitor_worker():
|
||
"""监控工作线程"""
|
||
global _observer, _is_monitoring
|
||
|
||
# 验证并获取工作流目录
|
||
is_valid, workflow_root = validate_workflow_dir()
|
||
if not is_valid:
|
||
logging.error("工作流监控启动失败：无法找到工作流目录")
|
||
return
|
||
|
||
event_handler = WorkflowFileHandler()
|
||
_observer = Observer()
|
||
# 监控目标目录及所有子目录（recursive=True）
|
||
_observer.schedule(event_handler, workflow_root, recursive=True)
|
||
|
||
try:
|
||
_observer.start()
|
||
_is_monitoring = True
|
||
|
||
# 保持监控运行
|
||
while _is_monitoring:
|
||
time.sleep(1)
|
||
|
||
except Exception as e:
|
||
logging.error(f"[ComfyUI-1hewNodes] 工作流监控启动失败：{str(e)}")
|
||
finally:
|
||
if _observer:
|
||
_observer.stop()
|
||
_observer.join()
|
||
_is_monitoring = False
|
||
|
||
def start_workflow_monitor():
|
||
"""启动工作流监控（后台线程模式）"""
|
||
global _monitor_thread, _is_monitoring
|
||
|
||
if _is_monitoring:
|
||
return True
|
||
|
||
try:
|
||
_monitor_thread = threading.Thread(target=_monitor_worker, daemon=True)
|
||
_monitor_thread.start()
|
||
return True
|
||
except Exception as e:
|
||
logging.error(f"[ComfyUI-1hewNodes] 启动工作流监控线程失败：{str(e)}")
|
||
return False
|
||
|
||
def stop_workflow_monitor():
|
||
"""停止工作流监控"""
|
||
global _is_monitoring, _observer, _monitor_thread
|
||
|
||
if not _is_monitoring:
|
||
return
|
||
|
||
try:
|
||
# 设置停止标志
|
||
_is_monitoring = False
|
||
|
||
# 停止Observer
|
||
if _observer:
|
||
_observer.stop()
|
||
# 等待Observer线程结束
|
||
try:
|
||
_observer.join(timeout=5.0) # 最多等待5秒
|
||
except Exception as e:
|
||
logging.warning(f"[ComfyUI-1hewNodes] Observer停止时出现异常：{str(e)}")
|
||
_observer = None
|
||
|
||
# 等待监控线程结束
|
||
if _monitor_thread and _monitor_thread.is_alive():
|
||
try:
|
||
_monitor_thread.join(timeout=3.0) # 最多等待3秒
|
||
except Exception as e:
|
||
logging.warning(f"[ComfyUI-1hewNodes] 监控线程停止时出现异常：{str(e)}")
|
||
_monitor_thread = None
|
||
|
||
except Exception as e:
|
||
logging.error(f"[ComfyUI-1hewNodes] 停止工作流监控时出错：{str(e)}")
|
||
# 强制重置状态
|
||
_is_monitoring = False
|
||
_observer = None
|
||
_monitor_thread = None
|
||
|
||
def is_monitoring():
|
||
"""检查监控是否正在运行"""
|
||
return _is_monitoring
|
||
|
||
# 兼容性：保留原有的start_observer函数用于独立运行
|
||
def start_observer():
|
||
"""启动文件监控器（独立运行模式）"""
|
||
# 验证并获取工作流目录
|
||
is_valid, workflow_root = validate_workflow_dir()
|
||
if not is_valid:
|
||
return False
|
||
|
||
event_handler = WorkflowFileHandler()
|
||
observer = Observer()
|
||
# 监控目标目录及所有子目录（recursive=True）
|
||
observer.schedule(event_handler, workflow_root, recursive=True)
|
||
|
||
try:
|
||
observer.start()
|
||
logging.info(f"监控已启动 | 目标目录：{workflow_root}")
|
||
logging.info("按 Ctrl+C 停止监控...")
|
||
while True:
|
||
time.sleep(1)
|
||
except KeyboardInterrupt:
|
||
observer.stop()
|
||
logging.info("监控已停止")
|
||
except Exception as e:
|
||
logging.error(f"监控启动失败：{str(e)}")
|
||
observer.stop()
|
||
observer.join()
|
||
return True
|
||
|
||
if __name__ == "__main__":
|
||
if not start_observer():
|
||
# 目录无效时暂停，让用户看清错误
|
||
input("按 Enter 退出...") |