import argparse from msilib.schema import Error import os import re import multiprocessing as mp from . import get_logger, ts, js MAX_CPU = mp.cpu_count() if MAX_CPU <= 4: MAX_CPU = 1 else: MAX_CPU = MAX_CPU - 2 func = { "ts": ts, "js": js } def worker(opt): """ Helper function, which will be called during a multiprocess. Converts the input. Args: opt: packed options """ type = opt[-2] return parse(func[type].get_parser(), *opt) def parse(parser, logger, input_path, output_path, name, path_to_file, dir_path, type, detailed_logs): """ Function to parse the corresponding Args: parser (_type_): _description_ logger (_type_): _description_ input_path (_type_): _description_ output_path (_type_): _description_ name (_type_): _description_ path_to_file (_type_): _description_ dir_path (_type_): _description_ Returns: str: _description_ """ func = { "ts": ts, "js": js } try: python_name = name.replace(".ts", ".py").replace(".js", ".py") rel_path = dir_path[len(input_path) + 1:] logger.debug(f"determined the following rel-path = {rel_path}") pytho_path_to_file = os.path.join(output_path,rel_path,python_name) try: content = parser.parse( parser.parse(open(path_to_file, encoding="utf-8").read()) ) logger.info(f"converted {path_to_file}") return ( (os.path.join(output_path,rel_path), pytho_path_to_file , content), (False ,False) ) except Exception as err: ptr_to_err = path_to_file try: m = re.search('(?<=line )\d+', str(err)) line = int(m.group(0)) m = re.search('(?<=col )\d+', str(err)) col = int(m.group(0)) ptr_to_err = path_to_file+":"+str(line)+":"+str(col) except: pass logger.error(f"Failed to convert {ptr_to_err}") if detailed_logs: logger.error(err) else: logger.error(str(err).split("\n")[0]) return ( (False, False, False), (str(err), ptr_to_err) ) except Exception as err: # An unknown Error has been found logger.error(err) return ( (False, False,False), (str(err), False) ) def main(): """ The main routine. """ parser = argparse.ArgumentParser( description='Tython. A tool to convert the typescript file to the given python files.') parser.add_argument('--input', type=str, default="./", dest='inputFolder', help='Defines the Folder with the Files, being search.') parser.add_argument('--output', type=str, default="./out/", dest='outputFolder', help='Defines the Files, which should be looked for. Possible Values are "ts" | "js"') parser.add_argument('--type', type=str, default="ts", dest='type', help='Defines the Folder where the converted files should be stored.') parser.add_argument('--debug', dest='debug', action='store_true', help='Shows debug related output') parser.add_argument('--cores', type=int, default=MAX_CPU, dest='cores', help='The Amount of cores, which must be use') parser.add_argument('--details', dest='details', action='store_true', help='Shows all erros.') # Create a Logger: logger = get_logger("nope-py-prepare") args = parser.parse_args() files_to_ignore ={ "js": lambda name: name.endswith(".js") and not (name.endswith(".spec.js") or "index" in name or "\\types\\" in name or "/types/" in name), "ts": lambda name: name.endswith(".ts") and not (name.endswith(".spec.ts") or "index" in name), } if not args.type in ("ts","js"): logger.error("Please use the correct type") logger.error(f"Determined type: '{args.type}'") return logger.warn(f"Working wiht '{args.type}' file-ending.") # Define the Input Path input_path = os.path.join(os.getcwd(), args.inputFolder) typescript_files = [] logger.info(f"Checking dir: '{input_path}'") if os.path.isdir(input_path): # Get all relevant Files. for dir_path, directories, files in os.walk(input_path): for file_name in files: # Generate the Path of files. path_to_file = os.path.join(dir_path, file_name) if files_to_ignore[args.type](path_to_file): # Show a log message logger.debug(f"Found file: '{file_name}' at '{dir_path}'") # Add the file-name, path to the file and the dir path typescript_files.append((file_name, path_to_file, dir_path)) elif os.path.isfile(input_path): # Generate the Path of files. path_to_file = input_path if files_to_ignore[args.type](path_to_file): dir_path = os.path.dirname(path_to_file) file_name = path_to_file[len(dir_path) + 1 :] # Show a log message logger.debug(f"Found file: '{file_name}' at '{dir_path}'") # Add the file-name, path to the file and the dir path typescript_files.append((file_name, path_to_file, dir_path)) else: raise Error("Failed to load the file") # Define the Destination output_path = os.path.join(os.getcwd(), args.outputFolder) cores_to_use = max(1, min(MAX_CPU, args.cores)) logger.info(f"Founf {len(typescript_files)} files. Starting multiprocess with {cores_to_use} cores.") # Create Pool. pool = mp.Pool(cores_to_use) results = pool.map(worker, [ ( logger, # The Logger input_path, # The Input Folder output_path, # The Output Path file_name, # Name of the File path_to_file, # Path to the File dir_path, # Path of the Directory. args.type, args.details ) for file_name, path_to_file, dir_path in typescript_files ] ) # Close the Pool pool.close() # Wait to finish all. pool.join() success = [] failed = [] for (path, py_file_name,content),(err, org_file_name) in results: if py_file_name and content: success.append(py_file_name) os.makedirs(path, exist_ok=True) with open(py_file_name, "w") as file: file.write(content) else: failed.append(org_file_name) if len(failed): logger.warn(f"The following files failed ({len(failed)}):") for (idx, file_name) in enumerate(failed): print("\t", idx, ".\t", file_name) if (args.debug): print("\n"*2) print(f"Created the following files ({len(success)}):") for file_name in success: print("\t- ", file_name) print("\n"*2) logger.info(f"Parsed {len(success)} of {len(typescript_files)} files ({(len(success)/len(typescript_files))*100:.2f} %).") if __name__ == "__main__": main()