semios-packages/lib/sources.py

import os
import subprocess
import sys
import tarfile
import urllib.parse
import urllib.request
import zipfile


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 zip_common_root(zip_file: zipfile.ZipFile) -> str | None:
    members = zip_file.infolist()
    if not members:
        return None
    root = None
    for member in members:
        path = member.filename.replace('\\', '/').lstrip('./').lstrip('/')
        if not path:
            continue
        this_root = path.split('/')[0]
        if this_root in ('', '.', '..'):
            continue
        if root is None:
            root = this_root
        elif 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)

    if download_path.endswith(".zip"):
        with zipfile.ZipFile(download_path, "r") as zip:
            common_root = zip_common_root(zip)
    else:
        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"

    if download_path.endswith(".zip"):
        with zipfile.ZipFile(download_path, "r") as zip:
            zip.extractall(extract_dir)
    else:
        with tarfile.open(download_path, "r") as tar:
            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)