class PythonDriverEnvUtils {

	public static PythonEnvironment preparePythonEnvironment(
			PythonDriverOptions pythonDriverOptions,
			String tmpDir) throws IOException, InterruptedException {
		PythonEnvironment env = new PythonEnvironment();

		tmpDir = new File(tmpDir).getAbsolutePath();

		// 1. setup temporary local directory for the user files
		Path tmpDirPath = new Path(tmpDir);
		FileSystem fs = tmpDirPath.getFileSystem();
		fs.mkdirs(tmpDirPath);

		env.tempDirectory = tmpDir;

		// 2. append the internal lib files to PYTHONPATH.
		List<String> pythonPathList = new ArrayList<>();

		List<File> internalLibs = extractBuiltInDependencies(
			tmpDir,
			UUID.randomUUID().toString(),
			true);

		for (File file: internalLibs) {
			pythonPathList.add(file.getAbsolutePath());
			file.deleteOnExit();
		}

		// 3. copy relevant python files to tmp dir and set them in PYTHONPATH.
		for (Path pythonFile : pythonDriverOptions.getPythonLibFiles()) {
			String sourceFileName = pythonFile.getName();
			// add random UUID parent directory to avoid name conflict.
			Path targetPath = new Path(
				tmpDirPath,
				String.join(File.separator, UUID.randomUUID().toString(), sourceFileName));
			if (!pythonFile.getFileSystem().isDistributedFS()) {
				// if the path is local file, try to create symbolic link.
				new File(targetPath.getParent().toString()).mkdir();
				createSymbolicLinkForPyflinkLib(
					Paths.get(new File(pythonFile.getPath()).getAbsolutePath()),
					Paths.get(targetPath.toString()));
			} else {
				FileUtils.copy(pythonFile, targetPath, true);
			}
			if (Files.isRegularFile(Paths.get(targetPath.toString()).toRealPath()) && sourceFileName.endsWith(".py")) {
				// add the parent directory of .py file itself to PYTHONPATH
				pythonPathList.add(targetPath.getParent().toString());
			} else {
				pythonPathList.add(targetPath.toString());
			}
		}

		env.pythonPath = String.join(File.pathSeparator, pythonPathList);

		if (!pythonDriverOptions.getPyFiles().isEmpty()) {
			env.systemEnv.put(PYFLINK_PY_FILES, String.join("\n", pythonDriverOptions.getPyFiles()));
		}
		if (!pythonDriverOptions.getPyArchives().isEmpty()) {
			env.systemEnv.put(
				PYFLINK_PY_ARCHIVES,
				joinTuples(pythonDriverOptions.getPyArchives()));
		}
		pythonDriverOptions.getPyRequirements().ifPresent(
			pyRequirements -> env.systemEnv.put(
				PYFLINK_PY_REQUIREMENTS,
				joinTuples(Collections.singleton(pyRequirements))));
		pythonDriverOptions.getPyExecutable().ifPresent(
			pyExecutable -> env.systemEnv.put(PYFLINK_PY_EXECUTABLE, pythonDriverOptions.getPyExecutable().get()));
		return env;
	}

}