diff --git a/Rayko_System_Prompt.py b/Rayko_System_Prompt.py index 95d4388..8492a6e 100644 --- a/Rayko_System_Prompt.py +++ b/Rayko_System_Prompt.py @@ -21,6 +21,10 @@ from aiohttp import web NODE_DIR = os.path.dirname(os.path.abspath(__file__)) PROMPT_DIR = os.path.join(NODE_DIR, "system_prompts") +ROUTE_GET_LIST = "/rs_system_prompt_dropdown/get_prompts" +ROUTE_GET_CONTENT = "/rs_system_prompt_dropdown/get_prompt_content" +ROUTE_SAVE = "/rs_system_prompt_dropdown/save_prompt" + def ensure_prompt_dir(): try: @@ -47,45 +51,181 @@ def _scan_prompts(): return files -def resolve_prompt_path(prompt_file): +def resolve_prompt_path(prompt_file, must_exist=False): if not prompt_file: return None try: - file_path = os.path.abspath( - os.path.join(PROMPT_DIR, prompt_file.replace("/", os.sep)) - ) + rel = prompt_file.replace("\\", "/").replace("/", os.sep) + file_path = os.path.abspath(os.path.join(PROMPT_DIR, rel)) except (OSError, ValueError): return None prompt_root = os.path.abspath(PROMPT_DIR) if not file_path.startswith(prompt_root + os.sep): return None + if file_path == prompt_root: + return None + if must_exist and not os.path.isfile(file_path): + return None return file_path -@server.PromptServer.instance.routes.get("/rs_system_prompt_dropdown/get_prompts") +@server.PromptServer.instance.routes.get(ROUTE_GET_LIST) async def get_prompts_handler(request): return web.json_response({"files": _scan_prompts()}) +@server.PromptServer.instance.routes.get(ROUTE_GET_CONTENT) +async def get_prompt_content_handler(request): + filename = request.query.get("file", "") + file_path = resolve_prompt_path(filename, must_exist=True) + if file_path is None: + return web.json_response( + {"ok": False, "error": "File not found or invalid path"}, + status=404, + ) + try: + with open(file_path, "r", encoding="utf-8-sig") as file: + content = file.read() + return web.json_response({"ok": True, "content": content}) + except (OSError, UnicodeDecodeError) as error: + return web.json_response( + {"ok": False, "error": f"Error reading file: {error}"}, + status=500, + ) + + +@server.PromptServer.instance.routes.post(ROUTE_SAVE) +async def save_prompt_handler(request): + try: + data = await request.json() + except Exception: + return web.json_response( + {"ok": False, "error": "Invalid JSON body"}, + status=400, + ) + + filename = (data or {}).get("filename", "") + content = (data or {}).get("content", "") + + if not isinstance(filename, str) or not isinstance(content, str): + return web.json_response( + {"ok": False, "error": "filename and content must be strings"}, + status=400, + ) + + filename = filename.strip().replace("\\", "/") + if not filename: + return web.json_response( + {"ok": False, "error": "Empty file name"}, + status=400, + ) + + if not filename.lower().endswith(".txt"): + filename += ".txt" + + if not ensure_prompt_dir(): + return web.json_response( + {"ok": False, "error": "Cannot access system_prompts directory"}, + status=500, + ) + + file_path = resolve_prompt_path(filename) + if file_path is None: + return web.json_response( + {"ok": False, "error": "Invalid path"}, + status=400, + ) + + created = not os.path.exists(file_path) + + try: + parent = os.path.dirname(file_path) + if parent: + os.makedirs(parent, exist_ok=True) + except OSError as error: + return web.json_response( + {"ok": False, "error": f"Cannot create folder: {error}"}, + status=500, + ) + + try: + with open(file_path, "w", encoding="utf-8", newline="\n") as file: + file.write(content) + except OSError as error: + return web.json_response( + {"ok": False, "error": f"Error writing file: {error}"}, + status=500, + ) + + rel_path = os.path.relpath(file_path, PROMPT_DIR).replace(os.sep, "/") + return web.json_response( + {"ok": True, "path": rel_path, "created": created} + ) + + +ROUTE_DELETE = "/rs_system_prompt_dropdown/delete_prompt" + + +@server.PromptServer.instance.routes.post(ROUTE_DELETE) +async def delete_prompt_handler(request): + try: + data = await request.json() + except Exception: + return web.json_response( + {"ok": False, "error": "Invalid JSON body"}, + status=400, + ) + + filename = (data or {}).get("filename", "") + if not isinstance(filename, str) or not filename.strip(): + return web.json_response( + {"ok": False, "error": "Empty file name"}, + status=400, + ) + + file_path = resolve_prompt_path(filename.strip(), must_exist=True) + if file_path is None: + return web.json_response( + {"ok": False, "error": "File not found or invalid path"}, + status=404, + ) + + try: + os.remove(file_path) + except OSError as error: + return web.json_response( + {"ok": False, "error": f"Error deleting file: {error}"}, + status=500, + ) + + return web.json_response({"ok": True, "path": filename.strip()}) + + class RSLoadSystemPrompt: @classmethod def INPUT_TYPES(cls): - files = _scan_prompts() - values = files if files else [""] return { "required": { - "prompt_file": ("STRING", {"default": values[0] if values else ""}), + "system_prompt": ( + "STRING", + { + "default": "", + "multiline": True, + "placeholder": "Pick a file above or type a system prompt here", + }, + ), "prompt": ( "STRING", { "default": "", "multiline": True, - "placeholder": "Additional prompt text (combined with system prompt)", + "placeholder": "Additional user prompt", }, ), }, "hidden": { + "prompt_file": ("STRING", {"default": ""}), "selected_file": ("STRING", {"default": ""}), }, } @@ -95,30 +235,8 @@ class RSLoadSystemPrompt: FUNCTION = "load_system_prompt" CATEGORY = "🦊 RaykoStudio" - @classmethod - def IS_CHANGED(cls, prompt_file, prompt, selected_file=""): - target = selected_file or prompt_file - path = resolve_prompt_path(target) - if path is None: - return "" - try: - return os.path.getmtime(path) - except OSError: - return "" - - def load_system_prompt(self, prompt_file, prompt, selected_file=""): - target = selected_file or prompt_file - file_path = resolve_prompt_path(target) - - if file_path is None: - return ("", prompt or "") - - try: - with open(file_path, "r", encoding="utf-8-sig") as file: - text = file.read() - return (text, prompt or "") - except (OSError, UnicodeDecodeError) as error: - return (f"Error reading file: {error}", prompt or "") + def load_system_prompt(self, system_prompt, prompt, prompt_file="", selected_file=""): + return (system_prompt or "", prompt or "") NODE_CLASS_MAPPINGS = { @@ -127,4 +245,4 @@ NODE_CLASS_MAPPINGS = { NODE_DISPLAY_NAME_MAPPINGS = { "RSLoadSystemPrompt": "🦊 RS Load System Prompt", -} \ No newline at end of file +}