| from aiohttp import web |
| from typing import Optional |
| from folder_paths import models_dir, user_directory, output_directory, folder_names_and_paths |
| from api_server.services.file_service import FileService |
| from api_server.services.terminal_service import TerminalService |
| import app.logger |
|
|
| class InternalRoutes: |
| ''' |
| The top level web router for internal routes: /internal/* |
| The endpoints here should NOT be depended upon. It is for ComfyUI frontend use only. |
| Check README.md for more information. |
| ''' |
|
|
| def __init__(self, prompt_server): |
| self.routes: web.RouteTableDef = web.RouteTableDef() |
| self._app: Optional[web.Application] = None |
| self.file_service = FileService({ |
| "models": models_dir, |
| "user": user_directory, |
| "output": output_directory |
| }) |
| self.prompt_server = prompt_server |
| self.terminal_service = TerminalService(prompt_server) |
|
|
| def setup_routes(self): |
| @self.routes.get('/files') |
| async def list_files(request): |
| directory_key = request.query.get('directory', '') |
| try: |
| file_list = self.file_service.list_files(directory_key) |
| return web.json_response({"files": file_list}) |
| except ValueError as e: |
| return web.json_response({"error": str(e)}, status=400) |
| except Exception as e: |
| return web.json_response({"error": str(e)}, status=500) |
|
|
| @self.routes.get('/logs') |
| async def get_logs(request): |
| return web.json_response("".join([(l["t"] + " - " + l["m"]) for l in app.logger.get_logs()])) |
|
|
| @self.routes.get('/logs/raw') |
| async def get_logs(request): |
| self.terminal_service.update_size() |
| return web.json_response({ |
| "entries": list(app.logger.get_logs()), |
| "size": {"cols": self.terminal_service.cols, "rows": self.terminal_service.rows} |
| }) |
|
|
| @self.routes.patch('/logs/subscribe') |
| async def subscribe_logs(request): |
| json_data = await request.json() |
| client_id = json_data["clientId"] |
| enabled = json_data["enabled"] |
| if enabled: |
| self.terminal_service.subscribe(client_id) |
| else: |
| self.terminal_service.unsubscribe(client_id) |
|
|
| return web.Response(status=200) |
|
|
|
|
| @self.routes.get('/folder_paths') |
| async def get_folder_paths(request): |
| response = {} |
| for key in folder_names_and_paths: |
| response[key] = folder_names_and_paths[key][0] |
| return web.json_response(response) |
|
|
| def get_app(self): |
| if self._app is None: |
| self._app = web.Application() |
| self.setup_routes() |
| self._app.add_routes(self.routes) |
| return self._app |
|
|
