import csv from aiohttp import web from .log import mklog from .utils import ( backup_file, import_install, reqs_map, run_command, styles_dir, ) endlog = mklog("mtb endpoint") # - ACTIONS import sys from pathlib import Path import_install("requirements") def ACTIONS_installDependency(dependency_names=None): if dependency_names is None: return {"error": "No dependency name provided"} endlog.debug(f"Received Install Dependency request for {dependency_names}") # reqs = [] resolved_names = [reqs_map.get(name, name) for name in dependency_names] try: run_command( [Path(sys.executable), "-m", "pip", "install"] + resolved_names ) return {"success": True} except Exception as e: return {"error": f"Failed to install dependencies: {e}"} # if platform.system() == "Windows": # reqs = list(requirements.parse((here / "reqs_windows.txt").read_text())) # else: # reqs = list(requirements.parse((here / "reqs.txt").read_text())) # print([x.specs for x in reqs]) # print( # "\n".join([f"{x.line} {''.join(x.specs[0] if x.specs else '')}" for x in reqs]) # ) # for dependency_name in dependency_names: # for req in reqs: # if req.name == dependency_name: # endlog.debug(f"Dependency {dependency_name} installed") # break def ACTIONS_getStyles(style_name=None): from .nodes.conditions import MTB_StylesLoader styles = MTB_StylesLoader.options match_list = ["name"] if styles: filtered_styles = { key: value for key, value in styles.items() if not key.startswith("__") and key not in match_list } if style_name: return filtered_styles.get( style_name, {"error": "Style not found"} ) return filtered_styles return {"error": "No styles found"} def ACTIONS_saveStyle(data): # endlog.debug(f"Received Save Styles for {data.keys()}") # endlog.debug(data) styles = [f.name for f in styles_dir.iterdir() if f.suffix == ".csv"] target = None rows = [] for fp, content in data.items(): if fp in styles: endlog.debug(f"Overwriting {fp}") target = styles_dir / fp rows = content break if not target: endlog.warning( f"Could not determine the target file for {data.keys()}" ) return {"error": "Could not determine the target file for the style"} backup_file(target) with target.open("w", newline="", encoding="utf-8") as file: csv_writer = csv.writer(file, quoting=csv.QUOTE_ALL) for row in rows: csv_writer.writerow(row) async def do_action(request) -> web.Response: endlog.debug("Init action request") request_data = await request.json() name = request_data.get("name") args = request_data.get("args") endlog.debug(f"Received action request: {name} {args}") method_name = f"ACTIONS_{name}" method = globals().get(method_name) if callable(method): result = method(args) if args else method() endlog.debug(f"Action result: {result}") return web.json_response({"result": result}) available_methods = [ attr[len("ACTIONS_") :] for attr in globals() if attr.startswith("ACTIONS_") ] return web.json_response( { "error": "Invalid method name.", "available_methods": available_methods, } ) # - HTML UTILS def dependencies_button(name, dependencies): deps = ",".join([f"'{x}'" for x in dependencies]) return f""" """ def csv_editor(): inputs = [f for f in styles_dir.iterdir() if f.suffix == ".csv"] # rows = {f.stem: list(csv.reader(f.read_text("utf8"))) for f in styles} style_files = {} for file in inputs: with open(file, encoding="utf8") as f: parsed = csv.reader(f) style_files[file.name] = [] for row in parsed: endlog.debug(f"Adding style {row[0]}") style_files[file.name].append((row[0], row[1], row[2])) html_out = """
| Name | Description |
|---|