class PythonEnvUtils {

	public static PythonEnvironment preparePythonEnvironment(List<Path> pythonLibFiles) throws IOException {
		PythonEnvironment env = new PythonEnvironment();

		// 1. setup temporary local directory for the user files
		String tmpDir = System.getProperty("java.io.tmpdir") +
			File.separator + "pyflink" + File.separator + UUID.randomUUID();

		Path tmpDirPath = new Path(tmpDir);
		FileSystem fs = tmpDirPath.getFileSystem();
		if (fs.exists(tmpDirPath)) {
			fs.delete(tmpDirPath, true);
		}
		fs.mkdirs(tmpDirPath);

		env.workingDirectory = tmpDirPath.toString();

		StringBuilder pythonPathEnv = new StringBuilder();

		pythonPathEnv.append(env.workingDirectory);

		// 2. create symbolLink in the working directory for the pyflink dependency libs.
		List<java.nio.file.Path> pythonLibs = getLibFiles(FLINK_OPT_DIR_PYTHON);
		for (java.nio.file.Path libPath : pythonLibs) {
			java.nio.file.Path symbolicLinkFilePath = FileSystems.getDefault().getPath(env.workingDirectory,
				libPath.getFileName().toString());
			createSymbolicLinkForPyflinkLib(libPath, symbolicLinkFilePath);
			pythonPathEnv.append(File.pathSeparator);
			pythonPathEnv.append(symbolicLinkFilePath.toString());
		}

		// 3. copy relevant python files to tmp dir and set them in PYTHONPATH.
		for (Path pythonFile : pythonLibFiles) {
			String sourceFileName = pythonFile.getName();
			Path targetPath = new Path(tmpDirPath, sourceFileName);
			FileUtils.copy(pythonFile, targetPath, true);
			String targetFileNames = Files.walk(Paths.get(targetPath.toString()))
				.filter(Files::isRegularFile)
				.filter(f -> !f.toString().endsWith(".py"))
				.map(java.nio.file.Path::toString)
				.collect(Collectors.joining(File.pathSeparator));
			pythonPathEnv.append(File.pathSeparator);
			pythonPathEnv.append(targetFileNames);
		}

		// 4. add the parent directory to PYTHONPATH for files suffixed with .py
		String pyFileParents = Files.walk(Paths.get(tmpDirPath.toString()))
			.filter(file -> file.toString().endsWith(".py"))
			.map(java.nio.file.Path::getParent)
			.distinct()
			.map(java.nio.file.Path::toString)
			.collect(Collectors.joining(File.pathSeparator));
		if (!StringUtils.isNullOrWhitespaceOnly(pyFileParents)) {
			pythonPathEnv.append(File.pathSeparator);
			pythonPathEnv.append(pyFileParents);
		}

		env.pythonPath = pythonPathEnv.toString();
		return env;
	}

}