mirror of
https://gitlab.com/kelteseth/ScreenPlay.git
synced 2024-09-18 08:22:33 +02:00
3e283aa5d5
Add ContentBuilder tools
63 lines
1.8 KiB
Python
63 lines
1.8 KiB
Python
import hashlib
|
|
from pathlib import Path
|
|
from pathlib import Path
|
|
from os import chdir
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
import os
|
|
import subprocess
|
|
import zipfile
|
|
|
|
def sftp_exists(sftp, path) -> bool:
|
|
try:
|
|
sftp.stat(path)
|
|
return True
|
|
except FileNotFoundError:
|
|
return False
|
|
|
|
|
|
def run(cmd, cwd=Path.cwd()):
|
|
result = subprocess.run(cmd, shell=True, cwd=cwd)
|
|
if result.returncode != 0:
|
|
raise RuntimeError(f"Failed to execute {cmd}")
|
|
|
|
def run_and_capture_output(cmd, cwd=Path.cwd()) -> str:
|
|
result = subprocess.run(cmd, shell=True, cwd=cwd, stdout=subprocess.PIPE)
|
|
if result.returncode != 0:
|
|
print(f"Failed to execute {cmd}")
|
|
if result.stdout is not None:
|
|
return str(result.stdout.decode('utf-8'))
|
|
return ""
|
|
|
|
def cd_repo_root_path() -> str:
|
|
# Make sure the script is always started from the same
|
|
# ScreenPlay root folder
|
|
root_path = Path.cwd()
|
|
if root_path.name == "Tools":
|
|
root_path = root_path.parent
|
|
print(f"Change root directory to: {root_path}")
|
|
chdir(root_path)
|
|
return root_path
|
|
|
|
def sha256(fname) -> str:
|
|
hash_sha256 = hashlib.sha256()
|
|
with open(fname, "rb") as f:
|
|
for chunk in iter(lambda: f.read(4096), b""):
|
|
hash_sha256.update(chunk)
|
|
return hash_sha256.hexdigest()
|
|
|
|
|
|
|
|
def zipdir(path, ziph):
|
|
# ziph is zipfile handle
|
|
for root, dirs, files in os.walk(path):
|
|
for file in files:
|
|
ziph.write(os.path.join(root, file),
|
|
os.path.relpath(os.path.join(root, file),
|
|
os.path.join(path, '..')))
|
|
|
|
|
|
def run_io_tasks_in_parallel(tasks):
|
|
with ThreadPoolExecutor() as executor:
|
|
running_tasks = [executor.submit(task) for task in tasks]
|
|
for running_task in running_tasks:
|
|
running_task.result() |