Added realtime updates to excel file using shadowcopy because of locked files.
This commit is contained in:
SencneS
2026-06-14 15:22:43 -04:00
committed by GitHub
parent b7bfb6b891
commit eb293c53f3
3 changed files with 462 additions and 241 deletions
+12 -76
View File
@@ -1,77 +1,13 @@
import os
import sys
current_dir = os.path.dirname(os.path.abspath(__file__))
if current_dir not in sys.path:
sys.path.insert(0, current_dir)
from .excel_styles_loader import NODE_CLASS_MAPPINGS, NODE_DISPLAY_NAME_MAPPINGS
JS_DIR = os.path.join(current_dir, "js")
os.makedirs(JS_DIR, exist_ok=True)
JAVASCRIPT_CODE = """
import { app } from "../../../scripts/app.js";
app.registerExtension({
name: "ExcelStylesLoader.DynamicDropdown",
async beforeRegisterNodeDef(nodeType, nodeData, appInstance) {
if (nodeData.name === "ExcelStylesLoader") {
const onNodeCreated = nodeType.prototype.onNodeCreated;
nodeType.prototype.onNodeCreated = function () {
if (onNodeCreated) onNodeCreated.apply(this, arguments);
const tabWidget = this.widgets.find(w => w.name === "tab_name");
const styleWidget = this.widgets.find(w => w.name === "style_name");
// Force the widget property configuration to render explicitly as a combo dropdown menu
if (styleWidget) {
styleWidget.type = "combo";
}
const updateStyles = async (tabValue) => {
if (!tabValue || !styleWidget) return;
try {
const response = await fetch(`/custom_nodes/excel_styles_loader/get_styles?tab_name=${encodeURIComponent(tabValue)}`);
const data = await response.json();
if (data && data.styles) {
styleWidget.options.values = data.styles;
if (!data.styles.includes(styleWidget.value)) {
styleWidget.value = data.styles.length > 0 ? data.styles[0] : "";
}
app.canvas.setDirty(true, true);
}
} catch (error) {
console.error("Error updating dynamic excel style widgets:", error);
}
};
if (tabWidget) {
tabWidget.callback = async (value) => {
await updateStyles(value);
};
// Delay slightly to ensure layout initialization is ready before first read
setTimeout(() => {
updateStyles(tabWidget.value);
}, 50);
}
};
}
}
});
"""
try:
js_file_path = os.path.join(JS_DIR, "excel_styles_loader.js")
with open(js_file_path, "w", encoding="utf-8") as f:
f.write(JAVASCRIPT_CODE.strip())
except Exception as e:
print(f"[ExcelStylesLoader] Failed to write UI web script asset: {e}")
WEB_DIRECTORY = "./js"
import os
import sys
current_dir = os.path.dirname(os.path.abspath(__file__))
if current_dir not in sys.path:
sys.path.insert(0, current_dir)
from .excel_styles_loader import NODE_CLASS_MAPPINGS, NODE_DISPLAY_NAME_MAPPINGS
WEB_DIRECTORY = os.path.join(current_dir, "js")
os.makedirs(WEB_DIRECTORY, exist_ok=True)
__all__ = ["NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS", "WEB_DIRECTORY"]
+253 -114
View File
@@ -1,114 +1,253 @@
import os
import openpyxl
from server import PromptServer
from aiohttp import web
_EXCEL_CACHE = {}
def get_filepath():
"""Returns the absolute path to styles.xlsx inside this node's folder."""
return os.path.join(os.path.dirname(os.path.abspath(__file__)), "styles.xlsx")
def load_excel_data():
"""Parses styles.xlsx mapping sheet names to their rows."""
filepath = get_filepath()
if not os.path.exists(filepath):
return {"Error: styles.xlsx missing": {"Place file in node folder": ["", ""]}}
data = {}
try:
wb = openpyxl.load_workbook(filepath, data_only=True)
for sheet_name in wb.sheetnames:
sheet = wb[sheet_name]
data[sheet_name] = {}
rows = list(sheet.iter_rows(values_only=True))
if not rows or len(rows) <= 1:
continue
headers = [str(c).strip().lower() if c is not None else "" for c in rows[0]]
name_idx = headers.index("name") if "name" in headers else 0
pos_idx = headers.index("prompt") if "prompt" in headers else 1
neg_idx = headers.index("negative_prompt") if "negative_prompt" in headers else 2
for row in rows[1:]:
if not row or row[name_idx] is None:
continue
style_name = str(row[name_idx]).strip()
if not style_name:
continue
pos_val = str(row[pos_idx]).strip() if len(row) > pos_idx and row[pos_idx] is not None else ""
neg_val = str(row[neg_idx]).strip() if len(row) > neg_idx and row[neg_idx] is not None else ""
data[sheet_name][style_name] = [pos_val, neg_val]
return data
except Exception as e:
print(f"[ExcelStylesLoader] Error reading workbook: {e}")
return {f"Error: {str(e)}": {"Check file formatting": ["", ""]}}
@PromptServer.instance.routes.get("/custom_nodes/excel_styles_loader/get_styles")
async def get_styles_endpoint(request):
tab_name = request.rel_url.query.get("tab_name", "")
global _EXCEL_CACHE
_EXCEL_CACHE = load_excel_data()
styles_list = list(_EXCEL_CACHE[tab_name].keys()) if tab_name in _EXCEL_CACHE else []
return web.json_response({"styles": styles_list})
class ExcelStylesLoader:
@classmethod
def INPUT_TYPES(cls):
global _EXCEL_CACHE
_EXCEL_CACHE = load_excel_data()
tab_choices = list(_EXCEL_CACHE.keys())
if not tab_choices:
tab_choices = ["No Tabs Found"]
# Give it a tiny default array to initialize as a genuine dropdown element
return {
"required": {
"tab_name": (tab_choices,),
"style_name": (["Select a Tab First"],),
}
}
RETURN_TYPES = ("STRING", "STRING")
RETURN_NAMES = ("positive prompt", "negative prompt")
FUNCTION = "execute"
CATEGORY = "loaders"
def execute(self, tab_name, style_name):
global _EXCEL_CACHE
if not _EXCEL_CACHE:
_EXCEL_CACHE = load_excel_data()
if tab_name in _EXCEL_CACHE and style_name in _EXCEL_CACHE[tab_name]:
return (_EXCEL_CACHE[tab_name][style_name][0], _EXCEL_CACHE[tab_name][style_name][1])
if tab_name in _EXCEL_CACHE and _EXCEL_CACHE[tab_name]:
first_key = list(_EXCEL_CACHE[tab_name].keys())[0]
return (_EXCEL_CACHE[tab_name][first_key][0], _EXCEL_CACHE[tab_name][first_key][1])
return ("", "")
@classmethod
def VALIDATE_INPUTS(cls, **kwargs):
"""
Bypasses strict backend verification list checking completely,
preventing the 'Value not in list' crash during execution.
"""
return True
NODE_CLASS_MAPPINGS = {
"ExcelStylesLoader": ExcelStylesLoader
}
NODE_DISPLAY_NAME_MAPPINGS = {
"ExcelStylesLoader": "Load Styles from Excel (.xlsx)"
}
import os
import time
import shutil
import hashlib
import openpyxl
from server import PromptServer
from aiohttp import web
# ── Paths ────────────────────────────────────────────────────────────────────
def get_source_path():
"""Returns the absolute path to Styles.xlsx inside this node's folder."""
return os.path.join(os.path.dirname(os.path.abspath(__file__)), "Styles.xlsx")
def get_shadow_path():
"""Returns the absolute path to the shadow copy."""
return os.path.join(os.path.dirname(os.path.abspath(__file__)), "._styles_cache.xlsx")
# ── Cache State ──────────────────────────────────────────────────────────────
_EXCEL_CACHE = None # {tab_name: {style_name: [pos, neg], ...}, ...}
_CACHE_MTIME = 0 # mtime of the file we last parsed
_LAST_HASH = None # MD5 hash of source file content
_LAST_COPY_ATTEMPT = 0 # timestamp of last failed copy attempt
_COPY_COOLDOWN = 3.0 # seconds between copy attempts when locked
# ── Shadow Copy Management ───────────────────────────────────────────────────
def _hash_file(filepath, blocksize=65536):
"""MD5 hash of file content. Returns None if file unreadable."""
hasher = hashlib.md5()
try:
with open(filepath, 'rb') as f:
while chunk := f.read(blocksize):
hasher.update(chunk)
return hasher.hexdigest()
except (OSError, PermissionError):
return None
def _try_copy_shadow():
"""
Copy source to shadow if content actually changed (hash-based).
Handles Excel-locked files gracefully.
Returns (success: bool, used_shadow: bool, message: str)
success — operation completed (shadow is usable)
used_shadow — True if we served from existing shadow (no copy made)
message — human-readable status for logging
"""
global _LAST_HASH, _LAST_COPY_ATTEMPT
source = get_source_path()
shadow = get_shadow_path()
# Source missing entirely
if not os.path.exists(source):
if os.path.exists(shadow):
return True, True, "Source missing, using shadow"
return False, False, "Source and shadow both missing"
# Get current content hash
current_hash = _hash_file(source)
if current_hash is None:
# Can't read source (locked) — use shadow if available
if os.path.exists(shadow):
return True, True, "Source locked, using shadow"
return False, False, "Source locked, no shadow available"
# Hash unchanged — no need to copy regardless of mtime
if current_hash == _LAST_HASH and os.path.exists(shadow):
return True, True, "Hash unchanged, using shadow"
# Hash changed — need to update shadow
now = time.time()
_LAST_COPY_ATTEMPT = now
try:
shutil.copy2(source, shadow)
_LAST_HASH = current_hash
print(f"[ExcelStylesLoader] Shadow updated (hash: {current_hash[:8]}...)")
return True, False, "Shadow updated from source"
except (PermissionError, OSError) as e:
# Locked — use existing shadow if we have one
if os.path.exists(shadow):
# Don't update _LAST_HASH — we'll try again next time
return True, True, f"Source locked, using stale shadow ({e})"
return False, False, f"Source locked, no shadow ({e})"
# ── Excel Parsing ────────────────────────────────────────────────────────────
def load_excel_data(force=False):
"""
Parse Styles.xlsx via shadow copy.
Call with force=True to bypass cache (user-triggered refresh).
"""
global _EXCEL_CACHE, _CACHE_MTIME
# Try to refresh shadow first
success, used_shadow, msg = _try_copy_shadow()
if not success:
_EXCEL_CACHE = {"Error: " + msg: {"Check file": ["", ""]}}
return _EXCEL_CACHE
# Determine which file to read
shadow = get_shadow_path()
filepath = shadow if os.path.exists(shadow) else get_source_path()
if not os.path.exists(filepath):
_EXCEL_CACHE = {"Error: No Excel file available": {"": ["", ""]}}
return _EXCEL_CACHE
# Check if we need to reparse
current_mtime = os.path.getmtime(filepath)
if not force and _EXCEL_CACHE is not None and current_mtime == _CACHE_MTIME:
return _EXCEL_CACHE
data = {}
try:
wb = openpyxl.load_workbook(filepath, data_only=True, read_only=True)
for sheet_name in wb.sheetnames:
sheet = wb[sheet_name]
data[sheet_name] = {}
rows = list(sheet.iter_rows(values_only=True))
if not rows or len(rows) <= 1:
continue
# Robust header detection
raw_headers = [str(c).strip() if c is not None else "" for c in rows[0]]
headers = [h.lower() for h in raw_headers]
def find_col(*candidates, default=0):
for cand in candidates:
cand = cand.lower()
if cand in headers:
return headers.index(cand)
for i, h in enumerate(headers):
if cand.replace("_", " ") == h.replace("_", " "):
return i
return default
name_idx = find_col("name", "style", "title", default=0)
pos_idx = find_col("prompt", "positive", "positive prompt", default=1)
neg_idx = find_col("negative_prompt", "negative prompt", "neg", default=2)
for row in rows[1:]:
if not row or row[name_idx] is None:
continue
style_name = str(row[name_idx]).strip()
if not style_name:
continue
pos_val = str(row[pos_idx]).strip() if len(row) > pos_idx and row[pos_idx] is not None else ""
neg_val = str(row[neg_idx]).strip() if len(row) > neg_idx and row[neg_idx] is not None else ""
data[sheet_name][style_name] = [pos_val, neg_val]
_EXCEL_CACHE = data
_CACHE_MTIME = current_mtime
return data
except Exception as e:
print(f"[ExcelStylesLoader] Error reading workbook: {e}")
_EXCEL_CACHE = {f"Error: {str(e)}": {"Check file formatting": ["", ""]}}
return _EXCEL_CACHE
# ── HTTP Endpoints ───────────────────────────────────────────────────────────
@PromptServer.instance.routes.get("/custom_nodes/excel_styles_loader/get_tabs")
async def get_tabs_endpoint(request):
"""Return all sheet names (tabs). ?force=1 to bypass cache."""
force = request.rel_url.query.get("force", "0") == "1"
data = load_excel_data(force=force)
tabs = [t for t in data.keys() if not t.startswith("Error:")]
return web.json_response({
"tabs": tabs,
"error": next((k for k in data.keys() if k.startswith("Error:")), None)
})
@PromptServer.instance.routes.get("/custom_nodes/excel_styles_loader/get_styles")
async def get_styles_endpoint(request):
"""Return style names for a given tab. ?force=1 to bypass cache."""
tab_name = request.rel_url.query.get("tab_name", "")
force = request.rel_url.query.get("force", "0") == "1"
data = load_excel_data(force=force)
styles = list(data.get(tab_name, {}).keys()) if tab_name in data else []
return web.json_response({
"styles": styles,
"tab_exists": tab_name in data
})
# ── ComfyUI Node ─────────────────────────────────────────────────────────────
class ExcelStylesLoader:
@classmethod
def INPUT_TYPES(cls):
# Prime cache so we have real tab names on first node creation
data = load_excel_data()
tab_choices = [t for t in data.keys() if not t.startswith("Error:")]
if not tab_choices:
tab_choices = ["No Tabs Found"]
return {
"required": {
"tab_name": (tab_choices,),
"style_name": (["Select a Tab"],),
}
}
RETURN_TYPES = ("STRING", "STRING")
RETURN_NAMES = ("positive prompt", "negative prompt")
FUNCTION = "execute"
CATEGORY = "loaders"
def execute(self, tab_name, style_name):
data = load_excel_data()
# Exact match
if tab_name in data and style_name in data[tab_name]:
return tuple(data[tab_name][style_name])
# Tab exists but style missing — fallback to first style in tab
if tab_name in data and data[tab_name]:
first_key = next(iter(data[tab_name]))
return tuple(data[tab_name][first_key])
# Complete fallback
return ("", "")
@classmethod
def VALIDATE_INPUTS(cls, tab_name, style_name, **kwargs):
"""
Allow any values — the frontend sends dynamically populated strings
that may not match the original INPUT_TYPES list. Actual validation
happens in execute() with graceful fallbacks.
"""
return True
NODE_CLASS_MAPPINGS = {"ExcelStylesLoader": ExcelStylesLoader}
NODE_DISPLAY_NAME_MAPPINGS = {"ExcelStylesLoader": "Load Styles from Excel (.xlsx)"}
+197 -51
View File
@@ -1,52 +1,198 @@
import { app } from "../../../scripts/app.js";
app.registerExtension({
name: "ExcelStylesLoader.DynamicDropdown",
async beforeRegisterNodeDef(nodeType, nodeData, appInstance) {
if (nodeData.name === "ExcelStylesLoader") {
const onNodeCreated = nodeType.prototype.onNodeCreated;
nodeType.prototype.onNodeCreated = function () {
if (onNodeCreated) onNodeCreated.apply(this, arguments);
const tabWidget = this.widgets.find(w => w.name === "tab_name");
const styleWidget = this.widgets.find(w => w.name === "style_name");
// Force the widget property configuration to render explicitly as a combo dropdown menu
if (styleWidget) {
styleWidget.type = "combo";
}
const updateStyles = async (tabValue) => {
if (!tabValue || !styleWidget) return;
try {
const response = await fetch(`/custom_nodes/excel_styles_loader/get_styles?tab_name=${encodeURIComponent(tabValue)}`);
const data = await response.json();
if (data && data.styles) {
styleWidget.options.values = data.styles;
if (!data.styles.includes(styleWidget.value)) {
styleWidget.value = data.styles.length > 0 ? data.styles[0] : "";
}
app.canvas.setDirty(true, true);
}
} catch (error) {
console.error("Error updating dynamic excel style widgets:", error);
}
};
if (tabWidget) {
tabWidget.callback = async (value) => {
await updateStyles(value);
};
// Delay slightly to ensure layout initialization is ready before first read
setTimeout(() => {
updateStyles(tabWidget.value);
}, 50);
}
};
}
}
import { app } from "../../../scripts/app.js";
app.registerExtension({
name: "ExcelStylesLoader.DynamicDropdown",
async beforeRegisterNodeDef(nodeType, nodeData, appInstance) {
if (nodeData.name !== "ExcelStylesLoader") return;
const origOnNodeCreated = nodeType.prototype.onNodeCreated;
nodeType.prototype.onNodeCreated = function () {
origOnNodeCreated?.apply(this, arguments);
const tabWidget = this.widgets.find(w => w.name === "tab_name");
const styleWidget = this.widgets.find(w => w.name === "style_name");
if (!tabWidget || !styleWidget) return;
// ── State ──────────────────────────────────────────────────────
const _tabCache = new Map(); // tabName -> [styleNames]
let _currentTab = null;
let _isRefreshing = false; // lock to prevent concurrent fetches
// ── Helpers ────────────────────────────────────────────────────
const applyStyleOptions = (styles, preserveValue = false) => {
styleWidget.options.values = styles;
if (!preserveValue || !styles.includes(styleWidget.value)) {
styleWidget.value = styles.length > 0 ? styles[0] : "";
}
app.canvas.setDirty(true, true);
};
const fetchJson = async (url) => {
const res = await fetch(url);
if (!res.ok) throw new Error(`HTTP ${res.status}`);
return res.json();
};
// ── Core: Load styles for a tab ──────────────────────────────
const loadStyles = async (tabName, {preserveValue = false, force = false} = {}) => {
if (!tabName || tabName === "No Tabs Found") {
applyStyleOptions([]);
return;
}
const cacheKey = `${tabName}|${force ? Date.now() : 'cached'}`;
const useCache = !force && _tabCache.has(tabName);
if (useCache) {
applyStyleOptions(_tabCache.get(tabName), preserveValue);
return;
}
try {
const forceParam = force ? "&force=1" : "";
const data = await fetchJson(
`/custom_nodes/excel_styles_loader/get_styles?tab_name=${encodeURIComponent(tabName)}${forceParam}`
);
const styles = data.styles || [];
_tabCache.set(tabName, styles);
applyStyleOptions(styles, preserveValue);
} catch (err) {
console.error("[ExcelStylesLoader] Failed to load styles:", err);
applyStyleOptions([]);
}
};
// ── Core: Full refresh (tabs + styles) ────────────────────────
const fullRefresh = async ({fromUser = false} = {}) => {
if (_isRefreshing) return;
_isRefreshing = true;
try {
const forceParam = fromUser ? "?force=1" : "";
const data = await fetchJson(`/custom_nodes/excel_styles_loader/get_tabs${forceParam}`);
const tabs = data.tabs || [];
const oldTab = tabWidget.value;
const oldTabs = [...tabWidget.options.values];
// Update tab widget
tabWidget.options.values = tabs.length > 0 ? tabs : ["No Tabs Found"];
// Restore or pick best tab
if (tabs.includes(oldTab)) {
tabWidget.value = oldTab;
} else {
tabWidget.value = tabs.length > 0 ? tabs[0] : "No Tabs Found";
}
// Clear style cache if tabs changed
const tabsChanged = oldTabs.length !== tabs.length ||
oldTabs.some((t, i) => t !== tabs[i]);
if (tabsChanged || fromUser) {
_tabCache.clear();
}
// Load styles for current tab
_currentTab = tabWidget.value;
await loadStyles(_currentTab, {
preserveValue: !tabsChanged,
force: fromUser
});
app.canvas.setDirty(true, true);
} catch (err) {
console.error("[ExcelStylesLoader] Refresh failed:", err);
} finally {
_isRefreshing = false;
}
};
// ── Hook: Tab dropdown opened ────────────────────────────────
// Detect when user clicks the tab dropdown
const origTabMouseDown = tabWidget.mouse ? tabWidget.mouse.bind(tabWidget) : null;
// LiteGraph combo widgets expose onMenuShown or we can hook the element
// More reliable: override the widget's onDraw to detect interaction,
// but simplest is to hook the canvas mousedown near the widget
const hookWidgetOpen = (widget, callback) => {
// Store original mousedown handler if any
const origMouseDown = widget.onMouseDown;
widget.onMouseDown = function(e, localPos, node) {
// e is canvas event; check if clicking the dropdown arrow area
const result = origMouseDown?.apply(this, arguments);
// LiteGraph: clicking anywhere on combo opens it
// We trigger refresh when user interacts with widget
callback();
return result;
};
};
// Hook tab widget: refresh tabs when opened
hookWidgetOpen(tabWidget, () => {
fullRefresh({fromUser: true});
});
// Hook style widget: refresh styles when opened
hookWidgetOpen(styleWidget, () => {
loadStyles(tabWidget.value, {preserveValue: true, force: true});
});
// ── Hook: Tab selection change ───────────────────────────────
const origTabCallback = tabWidget.callback;
tabWidget.callback = async (value) => {
origTabCallback?.call(tabWidget, value);
_currentTab = value;
await loadStyles(value, {preserveValue: false, force: true});
};
// ── Hook: Double-click anywhere on node ──────────────────────
const origDblClick = this.onDblClick;
this.onDblClick = function(e) {
console.log("[ExcelStylesLoader] Manual refresh triggered");
fullRefresh({fromUser: true});
return origDblClick?.apply(this, arguments);
};
// ── Hook: Right-click context menu ───────────────────────────
const origGetMenuOptions = this.getExtraMenuOptions;
this.getExtraMenuOptions = function(_, options) {
options.unshift({
content: "🔄 Refresh Excel Styles",
callback: () => fullRefresh({fromUser: true})
});
return origGetMenuOptions?.apply(this, arguments) || options;
};
// ── Initialization ─────────────────────────────────────────
const init = async () => {
_currentTab = tabWidget.value;
// Initial load without force (use cache if valid)
await fullRefresh({fromUser: false});
};
if (appInstance.graph) {
init();
} else {
setTimeout(init, 100);
}
};
}
});