# # Copyright 2024 © AE UTBM # ae@utbm.fr / ae.info@utbm.fr # # This file is part of the website of the UTBM Student Association (AE UTBM), # https://ae.utbm.fr. # # You can find the source code of the website at https://github.com/ae-utbm/sith # # LICENSED UNDER THE GNU GENERAL PUBLIC LICENSE VERSION 3 (GPLv3) # SEE : https://raw.githubusercontent.com/ae-utbm/sith/master/LICENSE # OR WITHIN THE LOCAL FILE "LICENSE" # # import hashlib import multiprocessing import os import platform import shutil import subprocess import sys import tarfile from dataclasses import dataclass from pathlib import Path from typing import Self import tomli import urllib3 from django.core.management.base import BaseCommand, CommandParser, OutputWrapper from urllib3.response import HTTPException @dataclass class XapianSpec: version: str core_sha1: str bindings_sha1: str @classmethod def from_pyproject(cls) -> Self: with open( Path(__file__).parent.parent.parent.parent / "pyproject.toml", "rb" ) as f: pyproject = tomli.load(f) spec = pyproject["tool"]["xapian"] return cls( version=spec["version"], core_sha1=spec["core-sha1"], bindings_sha1=spec["bindings-sha1"], ) class Command(BaseCommand): help = "Install xapian" def add_arguments(self, parser: CommandParser): parser.add_argument( "-f", "--force", action="store_true", help="Force installation even if already installed", ) def _current_version(self) -> str | None: try: import xapian except ImportError: return None return xapian.version_string() def handle(self, *args, force: bool, **options): if not os.environ.get("VIRTUAL_ENV", None): self.stdout.write( "No virtual environment detected, this command can't be used" ) return desired = XapianSpec.from_pyproject() if desired.version == self._current_version(): if not force: self.stdout.write( f"Version {desired.version} is already installed, use --force to re-install" ) return self.stdout.write( f"Version {desired.version} is already installed, re-installing" ) XapianInstaller(desired, self.stdout, self.stderr).run() self.stdout.write("Installation success") class XapianInstaller: def __init__( self, spec: XapianSpec, stdout: OutputWrapper, stderr: OutputWrapper, ): self._version = spec.version self._core_sha1 = spec.core_sha1 self._bindings_sha1 = spec.bindings_sha1 self._stdout = stdout self._stderr = stderr self._virtual_env = os.environ.get("VIRTUAL_ENV", None) if not self._virtual_env: raise RuntimeError("You are not inside a virtual environment") self._virtual_env = Path(self._virtual_env) self._dest_dir = Path(self._virtual_env) / "packages" self._core = f"xapian-core-{self._version}" self._bindings = f"xapian-bindings-{self._version}" def _util_download(self, url: str, dest: Path, sha1_hash: str) -> None: resp = urllib3.request("GET", url) if resp.status != 200: raise HTTPException(f"Could not download {url}") if hashlib.sha1(resp.data).hexdigest() != sha1_hash: raise ValueError(f"File downloaded from {url} is compromised") with open(dest, "wb") as f: f.write(resp.data) def _setup_env(self): os.environ.update( { "CPATH": "", "LIBRARY_PATH": "", "CFLAGS": "", "LDFLAGS": "", "CCFLAGS": "", "CXXFLAGS": "", "CPPFLAGS": "", } ) def _prepare_dest_folder(self): shutil.rmtree(self._dest_dir, ignore_errors=True) self._dest_dir.mkdir(parents=True) def _setup_windows(self): if "64bit" not in platform.architecture(): raise OSError("Only windows 64bit is supported") extractor = self._dest_dir / "" installer = self._dest_dir / "w64devkit-x64-2.0.0.exe" self._util_download( "https://github.com/ip7z/7zip/releases/download/24.08/7zr.exe", extractor, "d99de792fd08db53bb552cd28f0080137274f897", ) self._util_download( "https://github.com/skeeto/w64devkit/releases/download/v2.0.0/w64devkit-x64-2.0.0.exe", installer, "b5190c3ca9b06abe2b5cf329d99255a0be3a61ee", ) subprocess.run( [str(extractor), "x", str(installer), f"-o{self._dest_dir}"], check=False ).check_returncode() sys.path.insert(0, str(self._dest_dir / "w64devkit" / "bin")) def _download(self): self._stdout.write("Downloading source…") core = self._dest_dir / f"{self._core}.tar.xz" bindings = self._dest_dir / f"{self._bindings}.tar.xz" self._util_download( f"https://oligarchy.co.uk/xapian/{self._version}/{self._core}.tar.xz", core, self._core_sha1, ) self._util_download( f"https://oligarchy.co.uk/xapian/{self._version}/{self._bindings}.tar.xz", bindings, self._bindings_sha1, ) self._stdout.write("Extracting source …") with tarfile.open(core) as tar: tar.extractall(self._dest_dir) with tarfile.open(bindings) as tar: tar.extractall(self._dest_dir) os.remove(core) os.remove(bindings) def _install(self): self._stdout.write("Installing Xapian-core…") subprocess.run( ["./configure", "--prefix", str(self._virtual_env)], env=dict(os.environ), cwd=self._dest_dir / self._core, check=False, ).check_returncode() subprocess.run( [ "make", "-j", str(multiprocessing.cpu_count()), ], env=dict(os.environ), cwd=self._dest_dir / self._core, check=False, ).check_returncode() subprocess.run( ["make", "install"], env=dict(os.environ), cwd=self._dest_dir / self._core, check=False, ).check_returncode() self._stdout.write("Installing Xapian-bindings") subprocess.run( [ "./configure", "--prefix", str(self._virtual_env), "--with-python3", f"XAPIAN_CONFIG={self._virtual_env / 'bin'/'xapian-config'}", ], env=dict(os.environ), cwd=self._dest_dir / self._bindings, check=False, ).check_returncode() subprocess.run( [ "make", "-j", str(multiprocessing.cpu_count()), ], env=dict(os.environ), cwd=self._dest_dir / self._bindings, check=False, ).check_returncode() subprocess.run( ["make", "install"], env=dict(os.environ), cwd=self._dest_dir / self._bindings, check=False, ).check_returncode() def _post_clean(self): shutil.rmtree(self._dest_dir, ignore_errors=True) def _test(self): subprocess.run( [sys.executable, "-c", "import xapian"], check=False ).check_returncode() def run(self): self._setup_env() self._prepare_dest_folder() if platform.system() == "Windows": self._setup_windows() self._download() self._install() self._post_clean() self._test()