From 83fb19d5be1b58e34497bc14d4ecf65bbf3a4abb Mon Sep 17 00:00:00 2001 From: gabry Date: Tue, 16 Dec 2025 21:49:54 +0100 Subject: [PATCH] fempkg 1.3.3 core update --- build.py | 359 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ db.py | 49 ++++++++ fempkg | 121 +++++++++++++++++++ utils.py | 283 +++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 812 insertions(+) create mode 100644 build.py create mode 100755 db.py create mode 100755 fempkg create mode 100644 utils.py diff --git a/build.py b/build.py new file mode 100644 index 0000000..dd7900e --- /dev/null +++ b/build.py @@ -0,0 +1,359 @@ +#!/usr/bin/env python3 +# fempkg - a simple package manager +# Copyright (C) 2025 Gabriel Di Martino +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +import os +import subprocess +import shutil +import requests +from db import load_db, register_package, is_installed +from utils import download_extract, version_satisfies, PKG_DIR, _ensure_symlink, save_local_manifest_snapshot, read_manifest_paths, delete_file_and_prune_dirs, promote_local_to_versioned, remove_versioned_manifest +from tqdm import tqdm + +# --- basic dirs --- +TMP_RECIPE_DIR = "/tmp/fempkg" +os.makedirs(TMP_RECIPE_DIR, exist_ok=True) + +RECIPE_CACHE_DIR = os.path.expanduser("/var/lib/fempkg/repo") +MANIFEST_CACHE_DIR = "/var/lib/fempkg/manifests" # current manifests (pkgname.txt) +VERSIONED_MANIFEST_DIR = "/var/lib/fempkg/manifest-versions" # versioned manifests pkgname-version.txt +LOCAL_MANIFESTS_DIR = "/var/lib/fempkg/local-manifests" # temporary snapshots used during install +BINPKG_CACHE_DIR = "/var/lib/fempkg/binpkg" + +for d in (RECIPE_CACHE_DIR, MANIFEST_CACHE_DIR, VERSIONED_MANIFEST_DIR, LOCAL_MANIFESTS_DIR, BINPKG_CACHE_DIR): + os.makedirs(d, exist_ok=True) + + +# -------------------------- +# Recipe / manifest helpers +# -------------------------- +def fetch_recipe(pkgname): + path = os.path.join(RECIPE_CACHE_DIR, f"{pkgname}.recipe.py") + if not os.path.exists(path): + raise FileNotFoundError(f"Recipe for {pkgname} not found in cache. Run `fempkg update` first.") + return path + +def fetch_manifest(pkgname, pkgver=None): + """ + Returns a path to a manifest file. If pkgver is provided, creates a resolved + temporary manifest replacing {pkgver} placeholders and returns that path. + """ + path = os.path.join(MANIFEST_CACHE_DIR, f"{pkgname}.txt") + if not os.path.exists(path): + raise FileNotFoundError(f"Manifest for {pkgname} not found. Run `fempkg update` first.") + if not pkgver: + return path + temp_path = os.path.join(MANIFEST_CACHE_DIR, f"{pkgname}-resolved.txt") + with open(path) as f: + content = f.read() + content = content.replace("{pkgver}", pkgver) + with open(temp_path, "w") as f: + f.write(content) + return temp_path + +# -------------------------- +# Rebuild helper +# -------------------------- +def rebuild_package(packages, repo_dir=None): + if isinstance(packages, str): + packages = [packages] + for pkg in packages: + print(f"[fempkg] Rebuilding dependency: {pkg}") + dep_recipe = None + if repo_dir: + dep_recipe = os.path.join(repo_dir, f"{pkg}.recipe.py") + if not os.path.exists(dep_recipe): + print(f"Warning: recipe for {pkg} not found in {repo_dir}.") + dep_recipe = None + if not dep_recipe: + dep_recipe = fetch_recipe(pkg) + build_package(dep_recipe, repo_dir, force_rebuild=True) + +def extract_tar_zst_with_progress(tar_path, dest="/"): + """ + Extract a .tar.zst archive with a progress bar. + Uses 'tar' with zstd support and shows number of files extracted. + """ + # Step 1: get total number of files in the archive + try: + result = subprocess.run( + ["tar", "--use-compress-program=zstd", "-tf", tar_path], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + check=True + ) + files = result.stdout.splitlines() + total_files = len(files) + except subprocess.CalledProcessError as e: + print(f"[fempkg] Failed to list tar.zst archive: {e.stderr}") + raise + + # Step 2: extract with verbose output and tqdm progress + with tqdm(total=total_files, unit="file", desc=f"Extracting {tar_path}",) as pbar: + proc = subprocess.Popen( + ["tar", "--use-compress-program=zstd", "-xvf", tar_path, "-C", dest], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True + ) + for line in proc.stdout: + if line.strip(): + pbar.update(1) + proc.wait() + if proc.returncode != 0: + raise subprocess.CalledProcessError(proc.returncode, proc.args) + +# -------------------------- +# Main build function +# -------------------------- +def build_package(recipe_file, repo_dir=None, force_rebuild=False): + _ensure_symlink("/tmp/fempkg", "/var/tmp/fempkg") + _ensure_symlink("/tmp/fempkgbuild", "/var/tmp/fempkgbuild") + + db = load_db() + recipe = {} + with open(recipe_file, "r") as f: + exec(f.read(), recipe) + + name, version = recipe["pkgname"], recipe["pkgver"] + source_type = recipe.get("source_type") + deps = recipe.get("deps", []) + triggers = recipe.get("triggers", []) + atomic_upgrade = bool(recipe.get("atomic", False)) + + # --- Check for "source-only" mode --- + source_only = os.environ.get("FEMPKG_SOURCE", "").lower() in ("1", "true", "yes") + # nodelete for debug: if set don't delete old manifest files + nodelete_env = os.environ.get("FEMPKG_NODELETE", "").lower() in ("1", "true", "yes") + + if not force_rebuild and is_installed(name, version, db=db): + print(f"[fempkg] {name}-{version} is already installed. Skipping installation.") + return + + # Build dependencies + for dep_name in deps: + dep_recipe = None + if repo_dir: + dep_recipe = os.path.join(repo_dir, f"{dep_name}.recipe.py") + if not os.path.exists(dep_recipe): + print(f"Warning: recipe for {dep_name} not found in {repo_dir}.") + dep_recipe = None + if not dep_recipe: + dep_recipe = fetch_recipe(dep_name) + dep_info = {} + with open(dep_recipe, "r") as f: + exec(f.read(), dep_info) + dep_latest_ver = dep_info["pkgver"] + installed_ver = db["installed"].get(dep_name) + if installed_ver is None or not version_satisfies(installed_ver, dep_latest_ver): + print(f"Installing/updating dependency {dep_name} " + f"(installed: {installed_ver}, latest: {dep_latest_ver})") + build_package(dep_recipe, repo_dir) + else: + print(f"Dependency {dep_name} is up-to-date ({installed_ver}). Skipping.") + + # --- BINPKG logic (skip if building from source) --- + if not source_only: + binpkg_index = os.path.join(BINPKG_CACHE_DIR, "index.txt") + binpkg_success = False + binpkg_list = [] + if os.path.exists(binpkg_index): + with open(binpkg_index, "r") as f: + binpkg_list = [line.strip() for line in f if line.strip()] + + if name in binpkg_list: + print(f"[fempkg] Found prebuilt binary package for {name}. Preparing to install...") + try: + binpkg_url = f"https://rocketleaguechatp.duckdns.org/binpkg/{name}-{version}.tar.zst" + local_path = os.path.join(BINPKG_CACHE_DIR, f"{name}-{version}.tar.zst") + # download + with requests.get(binpkg_url, stream=True) as r: + r.raise_for_status() + total_size = int(r.headers.get('content-length', 0)) + chunk_size = 1024 * 1024 # 1 MB chunks + + with open(local_path, "wb") as f, tqdm( + total=total_size, unit='B', unit_scale=True, desc=f"Downloading {name}-{version}", + ) as pbar: + for chunk in r.iter_content(chunk_size=chunk_size): + if chunk: # filter out keep-alive chunks + f.write(chunk) + pbar.update(len(chunk)) + + # --------------------------- + # Manifest / deletion workflow + # --------------------------- + # resolved manifest (from manifests repo) for this package/version + try: + resolved_manifest = fetch_manifest(name, pkgver=version) + except FileNotFoundError: + # if manifest not present in repo, fallback to existing manifest if present + resolved_manifest = os.path.join(MANIFEST_CACHE_DIR, f"{name}.txt") + if not os.path.exists(resolved_manifest): + resolved_manifest = None + + # If we have a resolved manifest, snapshot it to local-manifests/-.txt + if resolved_manifest: + try: + os.makedirs(LOCAL_MANIFESTS_DIR, exist_ok=True) + local_snap = save_local_manifest_snapshot(name, version, resolved_manifest) + print(f"[fempkg] Saved local manifest snapshot: {local_snap}") + except Exception as e: + print(f"[fempkg] Warning: failed to save local manifest snapshot: {e}") + else: + print(f"[fempkg] No manifest available to snapshot for {name} {version}.") + + # If package already installed and we are NOT in atomic mode and not nodelete -> delete old manifest paths + old_installed_version = db["installed"].get(name) + if old_installed_version and (not atomic_upgrade) and (not nodelete_env): + old_versioned_path = os.path.join(VERSIONED_MANIFEST_DIR, f"{name}-{old_installed_version}.txt") + if os.path.exists(old_versioned_path): + print(f"[fempkg] Removing files from old manifest: {old_versioned_path}") + old_paths = read_manifest_paths(old_versioned_path) + for p in old_paths: + if not p: + continue + # Avoid accidental relative paths + if not os.path.isabs(p): + print(f"[fempkg] Skipping (not absolute) path from manifest: {p}") + continue + if os.path.exists(p): + print(f"[fempkg] Removing old file: {p}") + delete_file_and_prune_dirs(p) + else: + # not present, skip + pass + else: + print(f"[fempkg] No old versioned manifest found at {old_versioned_path}; nothing to remove.") + else: + if not old_installed_version: + print(f"[fempkg] {name} not currently installed; nothing to delete.") + elif atomic_upgrade: + print(f"[fempkg] Atomic upgrade requested in recipe - skipping deletion of old files.") + elif nodelete_env: + print(f"[fempkg] FEMPKG_NODELETE set - skipping deletion of old files.") + + # Verify GPG signature first + asc_path = local_path + ".asc" + if not os.path.exists(asc_path): + # download .asc if missing + binpkg_asc_url = f"https://rocketleaguechatp.duckdns.org/binpkg/{name}-{version}.tar.zst.asc" + with requests.get(binpkg_asc_url) as r: + r.raise_for_status() + with open(asc_path, "wb") as f: + f.write(r.content) + + print(f"[fempkg] Verifying GPG signature for {name}-{version}...") + try: + subprocess.run(["gpg", "--verify", asc_path, local_path], check=True) + print(f"[fempkg] Signature verified successfully.") + except subprocess.CalledProcessError as e: + raise RuntimeError(f"[fempkg] GPG verification failed for {name}-{version}: {e}") + + # Now extract the verified package + print(f"[fempkg] Extracting binary package to / : {local_path}") + extract_tar_zst_with_progress(local_path, dest="/") + + + # After successful extraction, register package + register_package(name, version, db=db) + print(f"[fempkg] Installed {name}-{version} from binary package.") + + # Promote the local snapshot (if saved) to versioned manifests and current manifest + if resolved_manifest: + promoted = promote_local_to_versioned(name, version) + if promoted: + print(f"[fempkg] Promoted manifest to versioned: {promoted}") + else: + print(f"[fempkg] Warning: promotion of manifest failed for {name}-{version}") + + # Remove the old versioned manifest (we only keep the new one) + if old_installed_version: + remove_versioned_manifest(name, old_installed_version) + print(f"[fempkg] Removed old versioned manifest for {name}-{old_installed_version} (if present).") + + binpkg_success = True + + except Exception as e: + print(f"[fempkg] Failed to use binary package for {name}: {e}. Falling back to build from source.") + binpkg_success = False + + if binpkg_success: + # run triggers and return + if triggers: + print(f"[fempkg] Running post triggers for {name}...") + for trig in triggers: + if isinstance(trig, str): + print(f"> {trig}") + subprocess.run(trig, shell=True, check=True) + elif isinstance(trig, dict) and "rebuild_package" in trig: + rebuild_package(trig["rebuild_package"], repo_dir) + else: + print(f"[fempkg] Unknown trigger type: {trig}") + return + + # --- If we get here: source build (or binpkg fallback) --- + print(f"[fempkg] Building {name}-{version} from source...") + manifest_path = fetch_manifest(name, pkgver=version) + with open(manifest_path) as f: + files = sorted(line.strip() for line in f if line.strip()) + print(f"Using manifest for {name} ({len(files)} files)") + + download_extract(recipe["source"], source_type) + for cmd in recipe.get("build", []): + print(f"> {cmd}") + subprocess.run(f". /etc/profile && mkdir -p /tmp/fempkg && {cmd}", shell=True, check=True) + + # After build/install steps (source path), register package & save manifest snapshot/versioned + register_package(name, version, db=db) + try: + # snapshot current manifest (from resolved manifest that we used earlier) + resolved_manifest = fetch_manifest(name, pkgver=version) + if resolved_manifest: + # save snapshot to versioned dir directly (source builds produce final files already) + os.makedirs(VERSIONED_MANIFEST_DIR, exist_ok=True) + versioned_path = os.path.join(VERSIONED_MANIFEST_DIR, f"{name}-{version}.txt") + shutil.copy(resolved_manifest, versioned_path) + # also update the current manifest + shutil.copy(versioned_path, os.path.join(MANIFEST_CACHE_DIR, f"{name}.txt")) + # remove old versioned manifest if existing + old_installed_version = db["installed"].get(name) + if old_installed_version and old_installed_version != version: + remove_versioned_manifest(name, old_installed_version) + except Exception as e: + print(f"[fempkg] Warning: failed to save versioned manifest for {name}: {e}") + + # Cleanup temp dirs and downloaded tarballs from PKG_DIR if present + for cleanup_path in ["/tmp/fempkg", "/tmp/fempkgbuild"]: + target = os.path.realpath(cleanup_path) + if os.path.exists(target): + shutil.rmtree(target, ignore_errors=True) + os.makedirs(target, exist_ok=True) + basename = os.path.basename(recipe["source"]) + tarball_path = os.path.join(PKG_DIR, basename) + if os.path.exists(tarball_path): + os.remove(tarball_path) + + if triggers: + print(f"[fempkg] Running post triggers for {name}...") + for trig in triggers: + if isinstance(trig, str): + print(f"> {trig}") + subprocess.run(trig, shell=True, check=True) + elif isinstance(trig, dict) and "rebuild_package" in trig: + rebuild_package(trig["rebuild_package"], repo_dir) + else: + print(f"[fempkg] Unknown trigger type: {trig}") diff --git a/db.py b/db.py new file mode 100755 index 0000000..246b6a3 --- /dev/null +++ b/db.py @@ -0,0 +1,49 @@ +#!/usr/bin/env python3 +# fempkg - a simple package manager +# Copyright (C) 2025 Gabriel Di Martino +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +import json, os +from pathlib import Path + +DB_PATH = "/var/lib/fempkg/db.json" +Path(os.path.dirname(DB_PATH)).mkdir(parents=True, exist_ok=True) + +def load_db(): + if os.path.exists(DB_PATH): + with open(DB_PATH, "r") as f: + return json.load(f) + return {"installed": {}} + +def save_db(db): + with open(DB_PATH, "w") as f: + json.dump(db, f, indent=2) + +def register_package(name, version, db=None): + if db is None: + db = load_db() + db["installed"][name] = version + save_db(db) + print(f"Registered: {name}-{version}") + +def is_installed(name, version=None, db=None): + if db is None: + db = load_db() + installed_ver = db["installed"].get(name) + if installed_ver is None: + return False + if version: + return installed_ver == version + return True + diff --git a/fempkg b/fempkg new file mode 100755 index 0000000..8286608 --- /dev/null +++ b/fempkg @@ -0,0 +1,121 @@ +#!/usr/bin/env python3 +# fempkg - a simple package manager +# Copyright (C) 2025 Gabriel Di Martino +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +import sys, os +from db import register_package, load_db, save_db +from build import build_package, fetch_recipe, MANIFEST_CACHE_DIR +from utils import fetch_all + +RECIPE_CACHE_DIR = "/var/lib/fempkg/repo" + +def list_installed(): + db = load_db() + for pkg, ver in db["installed"].items(): + print(f"{pkg}-{ver}") + +def update_all(): + fetch_all() + print("Recipe, manifest and the binpkg index cache updated.") + +def upgrade_packages(pkg=None): + db = load_db() + if pkg: + recipe_file = os.path.join(RECIPE_CACHE_DIR, f"{pkg}.recipe.py") + if not os.path.exists(recipe_file): + print(f"Recipe for {pkg} not found. Run `fempkg update` first.") + return + build_package(recipe_file) + else: + for pkgname in db["installed"]: + recipe_file = os.path.join(RECIPE_CACHE_DIR, f"{pkgname}.recipe.py") + if not os.path.exists(recipe_file): + print(f"Recipe for {pkgname} missing, skipping.") + continue + try: build_package(recipe_file) + except Exception as e: + print(f"Failed to upgrade {pkgname}: {e}") + +def uninstall_package(pkgname): + manifest_path = os.path.join(MANIFEST_CACHE_DIR, f"{pkgname}.txt") + if not os.path.exists(manifest_path): + print(f"No manifest for {pkgname}, cannot uninstall safely. Run `fempkg update` first.") + return + + with open(manifest_path) as f: + files = [line.strip() for line in f if line.strip()] + + # Remove files/directories in reverse order (leaves first) + for file in reversed(sorted(files)): + try: + if os.path.islink(file) or os.path.isfile(file): + os.remove(file) + print(f"Removed {file}") + elif os.path.isdir(file): + try: + shutil.rmtree(file) + print(f"Removed directory {file}") + except Exception: + print(f"Directory {file} not empty or protected, skipping") + except Exception as e: + print(f"Failed to remove {file}: {e}") + + # Remove manifest + try: + os.remove(manifest_path) + except Exception as e: + print(f"Failed to remove manifest {manifest_path}: {e}") + + # Remove from DB + db = load_db() + if pkgname in db["installed"]: + db["installed"].pop(pkgname) + save_db(db) + + print(f"{pkgname} uninstalled.") + + +def help_menu(): + print("Usage: fempkg [list|register|install|update|upgrade|uninstall]") + print(" list - list installed packages") + print(" register - manually register package") + print(" install - install from recipe") + print(" update - update recipe+manifest+binpkg cache") + print(" upgrade - update all packages, optionally one") + print(" uninstall - uninstall a package") + +def main(): + if len(sys.argv) < 2: + help_menu() + return + cmd = sys.argv[1] + if cmd == "list": list_installed() + elif cmd == "register" and len(sys.argv) == 4: register_package(sys.argv[2], sys.argv[3]) + elif cmd == "install" and len(sys.argv) >= 3: + for pkgname in sys.argv[2:]: + print(f"Installing {pkgname} …") + recipe_file = fetch_recipe(pkgname) + build_package(recipe_file) + print(f"{pkgname} installed") + elif cmd == "update" and len(sys.argv) == 2: update_all() + elif cmd == "upgrade": + if len(sys.argv) == 3: upgrade_packages(sys.argv[2]) + elif len(sys.argv) == 2: upgrade_packages() + else: help_menu() + elif cmd == "uninstall" and len(sys.argv) == 3: uninstall_package(sys.argv[2]) + else: help_menu() + +if __name__ == "__main__": + main() diff --git a/utils.py b/utils.py new file mode 100644 index 0000000..eb587a3 --- /dev/null +++ b/utils.py @@ -0,0 +1,283 @@ +#!/usr/bin/env python3 +# fempkg - a simple package manager +# Copyright (C) 2025 Gabriel Di Martino +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +import os +import shutil +import ssl +import subprocess +import errno +import urllib +from urllib.parse import urlparse +from tqdm import tqdm +import requests + +# Directories +BUILD_DIR = "/tmp/fempkgbuild" +PKG_DIR = "/var/lib/fempkg/pkgs" +RECIPE_CACHE_DIR = os.path.expanduser("/var/lib/fempkg/repo") +MANIFEST_CACHE_DIR = "/var/lib/fempkg/manifests" # current manifests (pkgname.txt) +VERSIONED_MANIFEST_DIR = "/var/lib/fempkg/manifest-versions" # versioned manifests pkgname-version.txt +LOCAL_MANIFESTS_DIR = "/var/lib/fempkg/local-manifests" # temporary snapshots used during install +BINPKG_CACHE_DIR = "/var/lib/fempkg/binpkg" + +os.makedirs(PKG_DIR, exist_ok=True) +os.makedirs(BUILD_DIR, exist_ok=True) + +# Utility functions +def _ensure_dir(path): + try: + os.makedirs(path, exist_ok=True) + except OSError as e: + if e.errno != errno.EEXIST: + raise + +def version_satisfies(installed, latest): + def parse(v): + parts = v.replace("-", ".").split(".") + parsed = [] + for p in parts: + parsed.append(int(p) if p.isdigit() else p) + return parsed + return parse(installed) >= parse(latest) + +def _install_insecure_opener_if_requested(): + """Sets up insecure SSL handling if FEMPKG_INSECURE=1""" + if os.environ.get("FEMPKG_INSECURE") == "1": + ctx = ssl.create_default_context() + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + opener = urllib.request.build_opener(urllib.request.HTTPSHandler(context=ctx)) + urllib.request.install_opener(opener) + +def _safe_basename_from_url(url): + parsed = urlparse(url) + name = os.path.basename(parsed.path) + return name or "downloaded.file" + +def _strip_archive_ext(name): + for ext in [".tar.gz", ".tar.bz2", ".tar.xz", ".tgz", ".tar"]: + if name.endswith(ext): + return name[:-len(ext)] + return os.path.splitext(name)[0] + +def _clone_git_repo(url): + """Clone a git repository into BUILD_DIR""" + repo_name = os.path.splitext(os.path.basename(url))[0] + dest = os.path.join(BUILD_DIR, repo_name) + if os.path.exists(dest): + shutil.rmtree(dest) + os.makedirs(BUILD_DIR, exist_ok=True) + subprocess.check_call(["git", "clone", url, dest]) + return dest + +# Download function with progress bar (ABI-safe) +def download_to(url, dest): + _install_insecure_opener_if_requested() + os.makedirs(os.path.dirname(dest), exist_ok=True) + + # Handle insecure SSL if requested + verify_ssl = os.environ.get("FEMPKG_INSECURE") != "1" + + headers = { + "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/118.0.0.0 Safari/537.36" + } + + with requests.get(url, headers=headers, stream=True, verify=verify_ssl) as r: + r.raise_for_status() + total_size = int(r.headers.get('content-length', 0)) + chunk_size = 1024 * 1024 # 1 MB + + with open(dest, "wb") as f, tqdm( + total=total_size, + unit='B', + unit_scale=True, + desc=f"Downloading {os.path.basename(dest)}", + ) as pbar: + for chunk in r.iter_content(chunk_size=chunk_size): + if chunk: # filter out keep-alive chunks + f.write(chunk) + pbar.update(len(chunk)) + +# Main download + prepare directory +def download_extract(url, source_type): + _install_insecure_opener_if_requested() + + if source_type == "git": + return _clone_git_repo(url) + + basename = _safe_basename_from_url(url) + tarball_path = os.path.join(PKG_DIR, basename) + + if not os.path.exists(tarball_path): + download_to(url, tarball_path) + + dest_dir = os.path.join(BUILD_DIR, _strip_archive_ext(basename)) + if os.path.exists(dest_dir): + shutil.rmtree(dest_dir) + os.makedirs(dest_dir) + + return dest_dir + +# --- helpers --- +def safe_rmtree(path): + """Remove path recursively or symlink target.""" + if os.path.islink(path): + real_path = os.path.realpath(path) + if os.path.exists(real_path): + shutil.rmtree(real_path, ignore_errors=True) + os.unlink(path) + elif os.path.exists(path): + shutil.rmtree(path, ignore_errors=True) + +def _ensure_symlink(src: str, target: str): + if os.path.islink(src): + if os.readlink(src) == target: + return + else: + os.unlink(src) + elif os.path.exists(src): + shutil.rmtree(src, ignore_errors=True) + os.makedirs(target, exist_ok=True) + os.symlink(target, src) + +_ensure_symlink("/tmp/fempkg", "/var/tmp/fempkg") +_ensure_symlink("/tmp/fempkgbuild", "/var/tmp/fempkgbuild") + +# -------------------------- +# Fetch recipes & manifests +# -------------------------- + +def fetch_all_recipes(repo_url="https://rocketleaguechatp.duckdns.org/recipes"): + index_url = f"{repo_url}/index.txt" + print(f"Fetching recipe index from {index_url}") + try: + with requests.Session() as session: + resp = session.get(index_url) + resp.raise_for_status() + entries = [e for e in resp.text.splitlines() if e.endswith(".recipe.py")] + + for entry in tqdm(entries, desc="Recipes", unit="file"): + url = f"{repo_url}/{entry}" + dest_path = os.path.join(RECIPE_CACHE_DIR, entry) + r = session.get(url) + r.raise_for_status() + with open(dest_path, "wb") as f: + f.write(r.content) + except Exception as e: + print(f"Error fetching index/recipes: {e}") + + +def fetch_all_manifests(repo_url="https://rocketleaguechatp.duckdns.org/manifests"): + index_url = f"{repo_url}/index.txt" + print(f"Fetching manifest index from {index_url}") + try: + with requests.Session() as session: + resp = session.get(index_url) + resp.raise_for_status() + entries = [e for e in resp.text.splitlines() if e.endswith(".txt")] + + for entry in tqdm(entries, desc="Manifests", unit="file"): + url = f"{repo_url}/{entry}" + dest_path = os.path.join(MANIFEST_CACHE_DIR, entry) + r = session.get(url) + r.raise_for_status() + with open(dest_path, "wb") as f: + f.write(r.content) + except Exception as e: + print(f"Error fetching index/manifests: {e}") + + +def fetch_binpkg_index(repo_url="https://rocketleaguechatp.duckdns.org/binpkg"): + index_url = f"{repo_url}/index.txt" + print(f"Fetching binpkg index from {index_url}") + try: + with requests.Session() as session: + resp = session.get(index_url) + resp.raise_for_status() + dest_path = os.path.join(BINPKG_CACHE_DIR, "index.txt") + r = session.get(index_url) + r.raise_for_status() + with open(dest_path, "wb") as f: + f.write(r.content) + print("Binpkg index downloaded successfully.") + except Exception as e: + print(f"Error fetching binpkg index: {e}") + + +def fetch_all(repo_url_recipes=None, repo_url_manifests=None, repo_url_binpkg=None): + os.system(f"rm -rf {RECIPE_CACHE_DIR}/*") + fetch_all_recipes(repo_url_recipes or "https://rocketleaguechatp.duckdns.org/recipes") + fetch_all_manifests(repo_url_manifests or "https://rocketleaguechatp.duckdns.org/manifests") + fetch_binpkg_index(repo_url_binpkg or "https://rocketleaguechatp.duckdns.org/binpkg") + print("All recipes, manifests and the binpkg index have been fetched successfully.") + +# -------------------------- +# Manifest/version utilities +# -------------------------- +def save_local_manifest_snapshot(pkgname, version, src_manifest_path): + """Copy the current manifest to local-manifests/-.txt (temporary snapshot).""" + os.makedirs(LOCAL_MANIFESTS_DIR, exist_ok=True) + dest = os.path.join(LOCAL_MANIFESTS_DIR, f"{pkgname}-{version}.txt") + shutil.copy(src_manifest_path, dest) + return dest + +def promote_local_to_versioned(pkgname, version): + """Move local snapshot to the versioned directory and update the current manifest.""" + local = os.path.join(LOCAL_MANIFESTS_DIR, f"{pkgname}-{version}.txt") + if not os.path.exists(local): + return None + os.makedirs(VERSIONED_MANIFEST_DIR, exist_ok=True) + versioned = os.path.join(VERSIONED_MANIFEST_DIR, f"{pkgname}-{version}.txt") + shutil.move(local, versioned) + # copy the versioned file to be the current manifest too (manifests/pkgname.txt) + current = os.path.join(MANIFEST_CACHE_DIR, f"{pkgname}.txt") + shutil.copy(versioned, current) + return versioned + +def remove_versioned_manifest(pkgname, version): + p = os.path.join(VERSIONED_MANIFEST_DIR, f"{pkgname}-{version}.txt") + if os.path.exists(p): + os.remove(p) + +def read_manifest_paths(manifest_path): + """Return a list of absolute paths listed in manifest_path.""" + if not os.path.exists(manifest_path): + return [] + with open(manifest_path) as f: + return [line.strip() for line in f if line.strip()] + +def delete_file_and_prune_dirs(path): + """Remove a file and try to prune empty parents up to / (but stop at /).""" + try: + if os.path.islink(path) or os.path.isfile(path): + os.remove(path) + elif os.path.isdir(path): + # If entire directory path listed, try rmdir (only if empty) + try: + os.rmdir(path) + except OSError: + # not empty + return + # prune parents + parent = os.path.dirname(path) + while parent and parent != "/" and parent != "": + try: + os.rmdir(parent) + except OSError: + break + parent = os.path.dirname(parent) + except Exception as e: + print(f"[fempkg] Warning: failed to remove {path}: {e}") \ No newline at end of file