98 lines
3.2 KiB
Python
98 lines
3.2 KiB
Python
import logging
|
|
import os
|
|
|
|
from typing import Optional, Union
|
|
|
|
from exec.file import copy_file, get_temp_dir, makedir
|
|
from utils import download_file, sha256sum
|
|
|
|
|
|
class PackageInfo:
|
|
name: str
|
|
version: str
|
|
|
|
|
|
class BinaryPackage(PackageInfo):
|
|
arch: str
|
|
filename: str
|
|
resolved_url: Optional[str]
|
|
_desc: Optional[dict[str, Union[str, list[str]]]]
|
|
|
|
def __init__(
|
|
self,
|
|
name: str,
|
|
version: str,
|
|
arch: str,
|
|
filename: str,
|
|
resolved_url: Optional[str] = None,
|
|
):
|
|
self.name = name
|
|
self.version = version
|
|
self.arch = arch
|
|
self.filename = filename
|
|
self.resolved_url = resolved_url
|
|
|
|
def __repr__(self):
|
|
return f'{self.name}@{self.version}'
|
|
|
|
@classmethod
|
|
def parse_desc(clss, desc_str: str, resolved_repo_url=None):
|
|
"""Parses a desc file, returning a PackageInfo"""
|
|
desc: dict[str, Union[str, list[str]]] = {}
|
|
for segment in f'\n{desc_str}'.split('\n%'):
|
|
if not segment.strip():
|
|
continue
|
|
key, elements = (e.strip() for e in segment.strip().split('%\n', 1))
|
|
elements_split = elements.split('\n')
|
|
desc[key] = elements if len(elements_split) == 1 else elements_split
|
|
validated: dict[str, str] = {}
|
|
for key in ['NAME', 'VERSION', 'ARCH', 'FILENAME']:
|
|
assert key in desc
|
|
value = desc[key]
|
|
assert isinstance(value, str)
|
|
validated[key] = value
|
|
p = clss(
|
|
name=validated['NAME'],
|
|
version=validated['VERSION'],
|
|
arch=validated['ARCH'],
|
|
filename=validated['FILENAME'],
|
|
resolved_url='/'.join([resolved_repo_url, validated['FILENAME']]),
|
|
)
|
|
p._desc = desc
|
|
return p
|
|
|
|
def acquire(self, dest_dir: Optional[str] = None, filename: Optional[str] = None) -> tuple[str, bool]:
|
|
raise NotImplementedError()
|
|
|
|
|
|
class LocalPackage(BinaryPackage):
|
|
|
|
def acquire(self, dest_dir: Optional[str] = None, filename: Optional[str] = None) -> tuple[str, bool]:
|
|
changed = False
|
|
assert self.resolved_url and self.filename and self.filename in self.resolved_url
|
|
path = f'{self.resolved_url.split("file://")[1]}'
|
|
if dest_dir:
|
|
makedir(dest_dir)
|
|
target = os.path.join(dest_dir, filename or self.filename)
|
|
if os.path.getsize(path) != os.path.getsize(target) or sha256sum(path) != sha256sum(target):
|
|
copy_file(path, target, follow_symlinks=True)
|
|
changed = True
|
|
path = target
|
|
return path, changed
|
|
|
|
|
|
class RemotePackage(BinaryPackage):
|
|
|
|
def acquire(self, dest_dir: Optional[str] = None, filename: Optional[str] = None) -> tuple[str, bool]:
|
|
assert self.resolved_url and '.pkg.tar.' in self.resolved_url
|
|
url = f"{self.resolved_url}"
|
|
assert url
|
|
|
|
dest_dir = dest_dir or get_temp_dir()
|
|
makedir(dest_dir)
|
|
dest_file_path = os.path.join(dest_dir, filename or self.filename)
|
|
|
|
logging.info(f"Trying to download package {url}")
|
|
changed = download_file(dest_file_path, url)
|
|
logging.info(f"{self.filename} {'already ' if not changed else ''}downloaded from repos")
|
|
return dest_file_path, changed
|