169 lines
5.0 KiB
Python
169 lines
5.0 KiB
Python
import os
|
|
import subprocess
|
|
import sys
|
|
import tarfile
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
|
|
def filename(url):
|
|
parsed = urllib.parse.urlparse(url)
|
|
path = parsed.path
|
|
filename = os.path.basename(path)
|
|
|
|
if not filename:
|
|
raise ValueError("No filename found in URL")
|
|
|
|
return filename
|
|
|
|
|
|
def download(url, output):
|
|
last_percent = -1
|
|
|
|
def report_progress(percent):
|
|
nonlocal last_percent
|
|
if percent != last_percent:
|
|
if last_percent >= 0:
|
|
sys.stdout.write("\b" * len(str(last_percent)))
|
|
sys.stdout.write(str(percent))
|
|
sys.stdout.flush()
|
|
last_percent = percent
|
|
|
|
try:
|
|
if url.startswith("file://"):
|
|
file_path = url[7:]
|
|
if not os.path.exists(file_path):
|
|
raise Exception("File not found: " + file_path)
|
|
|
|
total_size = os.path.getsize(file_path)
|
|
if total_size <= 0:
|
|
print("Copying...")
|
|
with open(file_path, "rb") as src, open(output, "wb") as dst:
|
|
while True:
|
|
chunk = src.read(8192)
|
|
if not chunk:
|
|
break
|
|
dst.write(chunk)
|
|
print("Download completed: " + output)
|
|
return output
|
|
|
|
sys.stdout.write("downloading... 0%")
|
|
sys.stdout.flush()
|
|
|
|
with open(file_path, "rb") as src, open(output, "wb") as dst:
|
|
downloaded = 0
|
|
block_size = 8192
|
|
while True:
|
|
chunk = src.read(block_size)
|
|
if not chunk:
|
|
break
|
|
dst.write(chunk)
|
|
downloaded += len(chunk)
|
|
percent = int(downloaded * 100 / total_size)
|
|
report_progress(percent)
|
|
|
|
if last_percent != 100:
|
|
report_progress(100)
|
|
|
|
sys.stdout.write("\n")
|
|
print("Download completed: " + output)
|
|
return output
|
|
|
|
req = urllib.request.Request(url, method="GET")
|
|
req.add_header(
|
|
"User-Agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
|
|
)
|
|
|
|
with urllib.request.urlopen(req) as response:
|
|
total_size = response.length
|
|
if total_size is None or total_size <= 0:
|
|
print("Downloading...")
|
|
with open(output, "wb") as f:
|
|
while True:
|
|
chunk = response.read(8192)
|
|
if not chunk:
|
|
break
|
|
f.write(chunk)
|
|
print("Download completed: " + output)
|
|
return output
|
|
|
|
sys.stdout.write("downloading... 0%")
|
|
sys.stdout.flush()
|
|
|
|
with open(output, "wb") as f:
|
|
downloaded = 0
|
|
block_size = 8192
|
|
|
|
while True:
|
|
chunk = response.read(block_size)
|
|
if not chunk:
|
|
break
|
|
f.write(chunk)
|
|
downloaded += len(chunk)
|
|
percent = int(downloaded * 100 / total_size)
|
|
report_progress(percent)
|
|
|
|
if last_percent != 100:
|
|
report_progress(100)
|
|
|
|
sys.stdout.write("\n")
|
|
|
|
print("Download completed: " + output)
|
|
return output
|
|
|
|
except Exception as e:
|
|
if os.path.exists(output):
|
|
os.remove(output)
|
|
raise Exception("Download failed: " + str(e))
|
|
|
|
|
|
def tar_common_root(tar: tarfile.TarFile) -> str | None:
|
|
root = None
|
|
for member in tar.getmembers():
|
|
this_root = member.path.split("/")[0]
|
|
if root is None:
|
|
root = this_root
|
|
if this_root != root:
|
|
return None
|
|
return root
|
|
|
|
|
|
def sources_dir_name() -> str:
|
|
with open("./target/sources.tag") as file:
|
|
return file.read()
|
|
|
|
|
|
def source_url(url: str):
|
|
if os.path.exists("./target/sources.tag"):
|
|
return
|
|
download_path = f"./target/{filename(url)}"
|
|
download(url, download_path)
|
|
|
|
with tarfile.open(download_path, "r") as tar:
|
|
common_root = tar_common_root(tar)
|
|
extract_dir = "./target"
|
|
sources_dir = f"./target/{common_root}"
|
|
if common_root in ["", ".", "None"]:
|
|
extract_dir = "./target/sources"
|
|
sources_dir = "./target/sources"
|
|
tar.extractall(extract_dir)
|
|
with open("./target/sources.tag", "wt+") as file:
|
|
file.write(sources_dir)
|
|
|
|
|
|
def _git_repo_name(url: str) -> str:
|
|
return url.split("/")[-1].replace(".git", "")
|
|
|
|
|
|
def source_git(url: str, branch: str | None = None):
|
|
source_path = os.getcwd() + "/target/" + _git_repo_name(url)
|
|
if os.path.exists("./target/sources.tag"):
|
|
return
|
|
cmdline = ["git", "clone", "--depth=1"]
|
|
if branch:
|
|
cmdline += ["-b", branch]
|
|
cmdline += [url, source_path]
|
|
subprocess.run(cmdline, check=True)
|
|
with open("./target/sources.tag", "wt+") as file:
|
|
file.write(source_path)
|