diff --git a/suite_nodes.py b/suite_nodes.py index 810c4f1..f8579c4 100644 --- a/suite_nodes.py +++ b/suite_nodes.py @@ -324,30 +324,32 @@ def _resolve_checkpoint_unique(ckpt_name_str: str): def _write_delete_script(): - roots, targets = _build_delete_targets() - script_path = _delete_script_path() - plan_path = _delete_plan_path() + """Write the manual checkpoint deletion script into ComfyUI's temp directory. + + The generated script intentionally uses plain print() for blank lines. + Do not generate string literals like "\\n..." here, because this function + itself writes Python source code and nested escaping is easy to break. + """ + DELETE_SCRIPT_PATH.parent.mkdir(parents=True, exist_ok=True) roots_json = json.dumps( - [str(Path(p).resolve()) for p in roots], + [str(root.resolve()) for root in _checkpoint_roots()], ensure_ascii=False, indent=2, ) - script = f'''#!/usr/bin/env python3 + script_template = '''#!/usr/bin/env python3 # Generated by ComfyUI-CheckpointHandpickerSuite. -# This script reads {DELETE_QUEUE_FILENAME} from the same directory at execution time. +# This script reads the delete reservation queue at execution time. # Review each prompt carefully. Default answer is No. import json from pathlib import Path -SCRIPT_DIR = Path(__file__).resolve().parent -QUEUE_PATH = SCRIPT_DIR / "{DELETE_QUEUE_FILENAME}" - -ALLOWED_ROOTS = [Path(p).resolve() for p in {roots_json}] -ALLOWED_SUFFIXES = {{".safetensors", ".json"}} +QUEUE_PATH = Path(__QUEUE_PATH_LITERAL__) +ALLOWED_ROOTS = [Path(p).resolve() for p in __ALLOWED_ROOTS_JSON__] +ALLOWED_SUFFIXES = {".safetensors", ".json"} def is_under_root(path, root): @@ -360,10 +362,7 @@ def is_safe_target(path): path = Path(path).resolve() if path.suffix.lower() not in ALLOWED_SUFFIXES: return False - for root in ALLOWED_ROOTS: - if is_under_root(path, root): - return True - return False + return any(is_under_root(path, root) for root in ALLOWED_ROOTS) def read_records(): @@ -380,56 +379,45 @@ def read_records(): records.append(rec) except Exception as exc: print("Skipping invalid queue record:", exc) - return records def active_records(): - active = {{}} - + active = {} for rec in read_records(): - rec_type = rec.get("type") - relpath = rec.get("ckpt_name_str") or rec.get("ckpt_name_relpath") or "" + typ = rec.get("type") + relpath = rec.get("ckpt_name_str", "") rid = rec.get("id") - if rec_type == "reserve" and relpath: + if typ == "reserve" and rid and relpath: active[relpath] = rec - elif rec_type == "cancel": + elif typ == "cancel": if relpath: active.pop(relpath, None) elif rid: for key, value in list(active.items()): if value.get("id") == rid: active.pop(key, None) - return active def active_targets(): targets = [] - for relpath, rec in sorted(active_records().items(), key=lambda x: x[0].lower()): raw_ckpt_path = rec.get("resolved_path") or rec.get("safetensors_path") or "" if not raw_ckpt_path: print("Skipping target without resolved_path:", relpath) continue - safetensors_path = Path(raw_ckpt_path).resolve() - - raw_json_path = rec.get("json_path") or "" - if raw_json_path: - json_path = Path(raw_json_path).resolve() - else: - json_path = safetensors_path.with_suffix(".json") - - targets.append({{ + ckpt = Path(raw_ckpt_path).resolve() + json_path = Path(rec.get("json_path") or ckpt.with_suffix(".json")).resolve() + targets.append({ "ckpt_name_str": relpath, - "safetensors_path": str(safetensors_path), + "safetensors_path": str(ckpt), "json_path": str(json_path), "reserved_at": rec.get("reserved_at", ""), - }}) - + }) return targets @@ -457,14 +445,19 @@ def main(): print("Checkpoint delete script") print("Queue file:", QUEUE_PATH) + print("Allowed roots:") + for root in ALLOWED_ROOTS: + print(" ", root) print("Targets:", len(targets)) for idx, item in enumerate(targets, start=1): print() - print(f"[{{idx}}/{{len(targets)}}] Delete checkpoint?") + print(f"[{idx}/{len(targets)}] Delete checkpoint?") print(" relpath:", item["ckpt_name_str"]) print(" safetensors:", item["safetensors_path"]) print(" json:", item["json_path"]) + if item.get("reserved_at"): + print(" reserved_at:", item["reserved_at"]) answer = input("Delete this checkpoint? (y/N): ").strip().lower() if answer != "y": @@ -484,28 +477,14 @@ if __name__ == "__main__": main() ''' - script_path.write_text(script, encoding="utf-8", newline="\n") + script = ( + script_template + .replace("__QUEUE_PATH_LITERAL__", repr(str(DELETE_QUEUE_PATH))) + .replace("__ALLOWED_ROOTS_JSON__", roots_json) + ) - lines = [ - "Checkpoint delete plan", - f"Generated at: {_now_iso()}", - f"Queue file: {_delete_queue_path()}", - f"Active targets: {len(targets)}", - "", - ] - - for idx, item in enumerate(targets, start=1): - lines.extend([ - f"[{idx}] {item['ckpt_name_str']}", - f" safetensors: {item['safetensors_path']}", - f" json: {item['json_path']}", - f" reserved_at: {item.get('reserved_at', '')}", - "", - ]) - - plan_path.write_text("\n".join(lines), encoding="utf-8", newline="\n") - - return script_path, plan_path, len(targets) + DELETE_SCRIPT_PATH.write_text(script, encoding="utf-8", newline="\n") + return DELETE_SCRIPT_PATH def _send_event(name: str, payload: dict, client_id=None): server = PromptServer.instance