削除スクリプトの生成を改善し、手動チェックポイント削除スクリプトの書き込み処理を修正

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