Mercurial > libervia-web
view libervia/server/tasks/implicit/task_brython.py @ 1513:ff95501abe74
server (websocket): don't crash when a force-close is failing
author | Goffi <goffi@goffi.org> |
---|---|
date | Mon, 22 May 2023 11:57:49 +0200 |
parents | 106bae41f5c8 |
children | 73c848c2f41e |
line wrap: on
line source
#!/ur/bin/env python3 from ast import literal_eval import json from pathlib import Path import shutil from typing import Any, Dict from sat.core import exceptions from sat.core.i18n import _ from sat.core.log import getLogger from sat.tools.common import utils from libervia.server.classes import Script from libervia.server.constants import Const as C from libervia.server.tasks import task log = getLogger(__name__) class Task(task.Task): def prepare(self): if "brython" not in self.resource.browser_modules: raise exceptions.CancelError("No brython module found") brython_js = self.build_path / "brython.js" if not brython_js.is_file(): installed_ver = None else: with brython_js.open() as f: for line in f: if line.startswith('// implementation ['): installed_ver = literal_eval(line[18:])[:3] log.debug( f"brython v{'.'.join(str(v) for v in installed_ver)} already " f"installed") break else: log.warning( f"brython file at {brython_js} doesn't has implementation " f"version" ) installed_ver = None try: import brython from brython import version as brython_version except ModuleNotFoundError as e: log.error('"brython" module is missing, can\'t use browser code for Brython') raise e ver = [int(v) for v in brython_version.implementation.split('.')[:3]] if ver != installed_ver: log.info(_("Installing Brython v{version}").format( version='.'.join(str(v) for v in ver))) data_path = Path(brython.__file__).parent / 'data' # shutil has blocking method, but the task is run before we start # the web server, so it's not a big deal shutil.copyfile(data_path / "brython.js", brython_js) shutil.copy(data_path / "brython_stdlib.js", self.build_path) else: log.debug("Brython is already installed") self.WATCH_DIRS = [] self.set_common_scripts() def set_common_scripts(self): for dyn_data in self.resource.browser_modules["brython"]: url_hash = dyn_data['url_hash'] import_url = f"/{C.BUILD_DIR}/{C.BUILD_DIR_DYN}/{url_hash}" dyn_data.setdefault('scripts', utils.OrderedSet()).update([ Script(src=f"/{C.BUILD_DIR}/brython.js"), Script(src=f"/{C.BUILD_DIR}/brython_stdlib.js"), ]) dyn_data.setdefault('template', {})['body_onload'] = self.get_body_onload( extra_path=[import_url]) self.WATCH_DIRS.append(dyn_data['path'].resolve()) def get_body_onload(self, debug=True, cache=True, extra_path=None): on_load_opts: Dict[str, Any] = {"pythonpath": [f"/{C.BUILD_DIR}"]} if debug: on_load_opts["debug"] = 1 if cache: on_load_opts["cache"] = True if extra_path is not None: on_load_opts["pythonpath"].extend(extra_path) return f"brython({json.dumps(on_load_opts)})" def copy_files(self, files_paths, dest): for p in files_paths: log.debug(f"copying {p}") if p.is_dir(): if p.name == '__pycache__': continue shutil.copytree(p, dest / p.name) else: shutil.copy(p, dest) async def on_dir_event(self, host, filepath, flags): self.set_common_scripts() await self.manager.run_task_instance(self) def start(self): dyn_path = self.build_path / C.BUILD_DIR_DYN for dyn_data in self.resource.browser_modules["brython"]: url_hash = dyn_data['url_hash'] if url_hash is None: # root modules url_prefix = dyn_data.get('url_prefix') if url_prefix is None: dest = self.build_path init_dest_url = f"/{C.BUILD_DIR}/__init__.py" else: dest = self.build_path / url_prefix dest.mkdir(exist_ok = True) init_dest_url = f"/{C.BUILD_DIR}/{url_prefix}/__init__.py" self.copy_files(dyn_data['path'].glob('*py'), dest) init_file = dyn_data['path'] / '__init__.py' if init_file.is_file(): self.resource.dyn_data_common['scripts'].update([ Script(src=f"/{C.BUILD_DIR}/brython.js"), Script(src=f"/{C.BUILD_DIR}/brython_stdlib.js"), Script(type='text/python', src=init_dest_url) ]) self.resource.dyn_data_common.setdefault( "template", {})['body_onload'] = self.get_body_onload() else: page_dyn_path = dyn_path / url_hash log.debug(f"using dynamic path at {page_dyn_path}") if page_dyn_path.exists(): log.debug("cleaning existing path") shutil.rmtree(page_dyn_path) page_dyn_path.mkdir(parents=True, exist_ok=True) log.debug("copying browser python files") self.copy_files(dyn_data['path'].iterdir(), page_dyn_path) script = Script( type='text/python', src=f"/{C.BUILD_DIR}/{C.BUILD_DIR_DYN}/{url_hash}/__init__.py" ) dyn_data.setdefault('scripts', utils.OrderedSet()).add(script)