# -*- coding: utf-8 -*- # Copyright 2015-2016 Mir Calculate. http://www.calculate-linux.org # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. from itertools import chain import sys from os import path import os import re from calculate.install.distr import DistributiveError, IsoDistributive from calculate.install.variables.kernel import KernelConfig, VariableInterface from calculate.lib.utils.dracut import Dracut from calculate.lib.utils.grub import GrubCommand from calculate.lib.utils.portage import getSquashList, isPkgInstalled from calculate.lib.variables.system import RootType import calculate.update.variables.update as update from .action import Actions from . import linux as vars_linux import datetime from calculate.install import distr import calculate.lib.utils.device as device from calculate.lib.utils.device import humanreadableSize from calculate.lib.utils.files import (process, typeFile, listDirectory, isEmpty, pathJoin, readFile, check_rw) from calculate.lib.utils.mount import isMount from calculate.lib.utils.kernel import InitrdFile from calculate.lib.utils.text import _u8 from calculate.lib.utils.tools import max_default from calculate.lib.utils.git import Git from ..build_storage import BuildStorage, Build from ..drive_spool import DriveSpool from .action import Actions as BuilderActions from calculate.lib.datavars import (Variable, VariableError, ReadonlyVariable, TableVariable, WRITEABLE, ReadonlyTableVariable, HumanReadable) from functools import wraps _ = lambda x: x from calculate.lib.cl_lang import setLocalTranslate, getLazyLocalTranslate setLocalTranslate('cl_builder3', sys.modules[__name__]) __ = getLazyLocalTranslate(_) def is_action(*available_action, **action_kwargs): def decorator(func): @wraps(func) def _wrapped_func(self, *args, **kw): if self.Get('cl_action') in available_action: return func(self, *args, **kw) else: return action_kwargs.get('default_value', '') return _wrapped_func return decorator class IsoTypeHelper(VariableInterface): """ Вспомогательный объект для переменных контейнера """ def uncompatible(self): if self.GetBool("cl_builder_container_set"): return _("This option is not compatible with container image") return "" class ContainerHelper(VariableInterface): """ Вспомогательный объект для переменных контейнера """ def uncompatible(self): if not self.GetBool("cl_builder_container_set"): return _("This option is not compatible with ISO image") return "" class ImagePathHelper(VariableInterface): """ Создание временного каталога для подготовки образа """ @property def base_dn(self): return self.Get('cl_builder_iso_base_path') @property def build_id(self): return self.Get('cl_builder_id') @property def build_id_path(self): return self.Get('cl_builder_id_path') prefix = "" def get(self): if self.build_id: dn = "%s%s" % (self.prefix, self.build_id_path) directory = path.join(self.base_dn, dn) new_dn = directory for i in range(0, 9999): if not path.exists(new_dn): return new_dn else: new_dn = "%s.%04d" % (directory, i) return new_dn return "" class VariableClBuilderStorage(ReadonlyVariable): type = "object" def get(self): return BuildStorage() class VariableClBuilderAvailableDev(Variable): """ Список дисков, которые можно использовать для сборки """ type = "list" def get(self): if self.Get('os_root_type_ext') in RootType.Live: return [path.join(Dracut.Initramfs, "builder")] return [] class VariableClBuilderDeviceSpool(ReadonlyVariable): """ Стэк дисков, которые можно использовать для сборки """ type = "object" def get(self): ds = DriveSpool(self.Get('cl_builder_available_dev')) return ds class BaseBuildId(Variable): """ Базовый класс для переменной id """ opt = ["--id"] metavalue = "ID" def init(self): self.label = _("Build ID") self.help = _("build ID") class VariableClBuilderVideoDriverPath(Variable): """ Имя файла, содержащего данные об установки драйверов во время загрузки """ def get(self): builder_path = self.Get('cl_builder_path') return path.join( builder_path, 'var/cache/calculate/video_drivers') class VariableClBuilderVideodrvSet(IsoTypeHelper, Variable): """ Нужно ли скачивать пакеты в дистрибутив для установки проприетарных драйверов во время загрузки """ type = "bool" opt = ["--video", "-V"] autodetect = False default_value = "off" def init(self): self.label = _("Include proprietary video drivers") self.help = _("include proprietary video drivers") def get(self): if self.autodetect: fn = self.Get('cl_builder_video_driver_path') return "on" if path.exists(fn) else "off" else: return self.default_value class VariableClBuilderSourceFilename(Variable): """ Названия файла исходного дистрибутива """ type = "file" element = 'file' opt = ["--source"] metavalue = "SOURCE" untrusted = True check_after = ["cl_builder_layered_set"] def init(self): self.label = _("Source image") self.help = _("source image") def check(self, isoimage): """Set image file""" if self.Get('cl_action') in Actions.NewAssemble and not isoimage: raise VariableError(_("You need to select a source image")) if self.GetBool('cl_builder_layered_set'): if not isinstance(self.Get('cl_builder_source'), IsoDistributive): raise VariableError( _("Wrong source distributive for layered build")) shortname = self.Get('os_builder_linux_shortname') build = self.Get('os_builder_linux_build') arch = self.Get('os_builder_arch_machine') local_arch = self.Get('os_arch_machine') if not build or not shortname or not arch: raise VariableError(_("Wrong image file")) if local_arch == "i686" and arch != local_arch: raise VariableError( _("Unable to build the system {what} from {local}").format( what=arch, local=local_arch)) def humanReadable(self): fullname = self.Get('os_builder_linux_name') subname = self.Get('os_builder_linux_subname') ver = self.Get('os_builder_linux_ver') if subname: subname = " %s" % subname if ver: ver = " %s" % ver arch = self.Get('os_builder_arch_machine') build = self.Get('os_builder_linux_build') return "{fullname}{ver} {arch} {build}".format( fullname="%s%s" % (fullname, subname), build=build, ver=ver, arch=arch) def get(self): if self.Get('os_root_type') == 'livecd': return Dracut.IsoLive return "" class VariableClBuilderSource(ReadonlyVariable): """ Объект исходного дистрибутива для разворачивания сборки """ type = "object" def get(self): """Get image file from distributive repository""" try: filename = self.Get('cl_builder_source_filename') if filename: return distr.Distributive.fromFile(filename) except distr.DistributiveError: pass return "" def humanReadable(self): filename = self.Get('cl_builder_source') if filename: return filename.getType() return filename class VariableClBuilderPreparePath(Variable): """ Путь по умолчанию для собираемой системы """ value = "/var/calculate/builder" class VariableClBuilderDiskDev(Variable): """ Диск или директория, куда будет развёрнут образ """ type = "choiceedit" untrusted = True opt = ["-d", "--disk"] metavalue = "DEST" def init(self): self.label = _("Build files") self.help = _("partition or directory intended for build") def get_builder_image_dn(self, ds=None): base_name = self.Get('cl_builder_id_path') if self.Get('os_root_type') != "livecd": default_path = self.Get('cl_builder_prepare_path') for i in chain((0,), range(2, 9999)): if i: dn = "%s.%02d" % (base_name, i) else: dn = base_name full_dn = path.join(default_path, dn) if ((not path.exists(full_dn) or not isMount(full_dn) and isEmpty(full_dn)) and (ds is None or isinstance(ds, DriveSpool) and ds.lock_device(full_dn))): return full_dn else: return "" return "" def get(self): ds = self.Get('cl_builder_device_spool') return ds.get() or self.get_builder_image_dn(ds) def choice(self): devices = self.Select( 'install.os_disk_dev', where='install.os_disk_mount', eq='') devices = [x for x in devices if not isMount(x)] default_dn = self.get_builder_image_dn() if default_dn: return [default_dn] + devices else: return devices def check(self, value): if not value: raise VariableError( _("You need to select a destination for build")) if value.startswith("/dev"): cnDisk = device.udev.get_device_info( name=value).get('DEVNAME', value) if cnDisk not in self.Get('install.os_disk_dev'): raise VariableError(_("Wrong device '%s'") % value) if not value.startswith('/'): raise VariableError( _("Wrong directory '%s'") % value) if isMount(value): raise VariableError( _("Destination '%s' is already in use") % value) class DiskFreeHelper(ReadonlyVariable): """ Переменные содержащие свободное место на дисках """ type = "int" def get_free_for(self, dn): try: dfdn = dn while True and dfdn and dfdn.startswith("/"): dfProcess = process("/bin/df", "--output=avail", dfdn) data = dfProcess.readlines() if len(data) > 1: return int(data[1].strip()) * 1024 if dfdn == '/': return 0 dfdn = path.dirname(dfdn) except ValueError: return 0 return 0 def humanReadable(self): value = self.Get() if not value: value = "0" return humanreadableSize(int(value)) class VariableClBuilderDiskSize(DiskFreeHelper): """ """ def init(self): self.label = _("Free disk space") @is_action(Actions.Prepare) def get(self): device = self.Get('cl_builder_disk_dev') if device: if device.startswith('/dev/'): return self.Select('install.os_disk_size', where='install.os_disk_dev', eq=device, limit=1) or "0" else: return str(self.get_free_for(device)) return "0" class VariableClBuilderPrepareFreeSize(IsoTypeHelper, DiskFreeHelper): """ Свободное место используемое для подготовки образа """ def init(self): self.label = _("Free disk space for ISO building") @is_action(Actions.Image) def get(self): dn = self.Get('cl_builder_iso_base_path') return str(self.get_free_for(dn)) class VariableClBuilderImageFreeSize(IsoTypeHelper, DiskFreeHelper): """ Свободное место на диске, где создается iso образ """ def init(self): self.label = _("Free disk space for ISO image") @is_action(Actions.Image) def get(self): dn = self.Get('cl_builder_image_filename') return str(self.get_free_for(dn)) class VariableClBuilderContainerFreeSize(ContainerHelper, DiskFreeHelper): """ Свободное место на диске, где создается iso образ """ def init(self): self.label = _("Free disk space for container image") @is_action(Actions.Image) def get(self): dn = self.Get('cl_builder_image_filename') return str(self.get_free_for(dn)) class VariableClBuilderLayeredSet(Variable): """ Сборка будет выполняться в слое """ type = "bool" opt = ["--layers"] kernel_opt = "CONFIG_OVERLAY_FS" def init(self): self.help = _("use layers for build") self.label = _("Use layers for build") def check_kernel_option(self): """ Проверить возможность ядра использовать overlay fs """ return ( self.kernel_opt in self.Get('install.os_kernel_config')) def get(self): try: self.check_on() return "on" except VariableError: return "off" def check_on(self): if not self.check_kernel_option(): raise VariableError( _("You need kernel with %s for use layers") % self.kernel_opt) if self.Get('cl_builder_disk_dev').startswith('/dev'): raise VariableError( _("Layers are used for building in a directory only")) if isinstance(self.Get('cl_builder_source'), distr.ArchiveDistributive): raise VariableError( _("Layers are used for building from ISO image")) def check(self, value): if value == "on": self.check_on() class VariableClBuilderPath(ReadonlyVariable): """ Путь, где будет собираться дистрбутив """ def get(self): try: image = self.Get('cl_builder_target') if image: return image.getDirectory() except DistributiveError: return "" return "" class VariableClBuilderTarget(Variable): """ Объект собираемого дистрибутива """ type = "object" def get_create(self): builder_disk_dev = self.Get('cl_builder_disk_dev') if not builder_disk_dev: return "" build_id_path = self.Get('cl_builder_id_path') mount_dir = path.join(distr.DefaultMountPath.BaseMountPath, build_id_path) source = self.Get('cl_builder_source_filename') if self.GetBool('cl_builder_layered_set'): dist_obj = distr.LayeredDistributive(mount_dir, builder_disk_dev, source) return dist_obj else: if builder_disk_dev.startswith('/dev'): dist_obj = distr.PartitionDistributive(builder_disk_dev, mdirectory=mount_dir) return dist_obj else: dist_obj = distr.DirectoryDistributive(builder_disk_dev, mdirectory=mount_dir) return dist_obj def get_worked(self): build = self.Get('cl_builder_build') if build: dist_obj = build.distributive return dist_obj return "" def get(self): action = self.Get('cl_action') if action in Actions.NewAssemble: return self.get_create() elif action in Actions.WorkAssemble: return self.get_worked() elif action in Actions.BrokenAssemble: return self.get_worked() else: return "" class VariableClBuilderClearSet(Variable): """ Очистить дистрибутив при отключении сборки """ type = "bool" value = "on" opt = ["--clear"] def init(self): self.label = _("Clear after unmount") self.help = _("clear data after unmount") class VariableClBuilderClearPkgSet(Variable): """ Удалять бинарные пакеты при отключении сборки """ type = "bool" value = "off" opt = ["--clear-pkg"] def init(self): self.label = _("Clear binary packages after unmount") self.help = _("clear binary packages after unmount") class VariableClBuilderBuild(Variable): """ Объект сборки """ type = "object" def get_create(self): distr = self.Get('cl_builder_target') if not distr: return "" storage = self.Get('cl_builder_storage') buildid = self.Get('cl_builder_id') build = Build(buildid, distr, storage) return build def get_worked(self): storage = self.Get('cl_builder_storage') buildid = self.Get('cl_builder_id') build = storage.get_build(buildid) if build and build.distributive: build.distributive.reserve() for child in build.distributive.childs: child.reserve() return build or "" def get(self): action = self.Get('cl_action') if action in Actions.NewAssemble: return self.get_create() if action in Actions.WorkAssemble: return self.get_worked() elif action in Actions.BrokenAssemble: return self.get_worked() return "" class VariableClBuilderNewId(BaseBuildId): """ Id сборки при развертывании """ value = "" untrusted = True def get(self): return self.Get('cl_builder_profile_name') def check(self, value): if not value and self.Get('cl_builder_source_filename'): raise VariableError(_("Please specify the build ID")) if value and not re.match("^[A-Za-z][A-Za-z0-9/:_+-]+$", value): raise VariableError(_("Wrong symbols in the build ID")) if value in self.Get('cl_builder_storage'): raise VariableError(_("Build %s already exists") % _u8(value)) class VariableClBuilderPreparedId(BaseBuildId): """ Id развёрнутой сборки """ type = "choice" untrusted = True def available(self): bs = self.Get('cl_builder_storage') action = self.Get('cl_action') for x in bs: build = bs.get_build(x) if (build and (action == Actions.Break or build.status == Build.Status.Worked)): yield x def get(self): l = list(self.available()) if l and len(l) == 1: return l[0] return "" def choice(self): bs = self.Get('cl_builder_storage') for x in bs: build = bs.get_build(x) if build: if build.status == Build.Status.Broken: yield (x, "%s (%s)" % (x, _("broken"))) else: yield (x, x) def _is_iso_compatible(self): prefix = self.Get('cl_builder_path') if len(self.Get('cl_builder_kernel_config')): for pn in ('sys-kernel/calculate-sources', 'sys-kernel/dracut', 'media-gfx/gfxboot-themes-calculate', 'sys-boot/grub'): if not isPkgInstalled(pn, prefix=prefix): return False return True return False def check(self, value): if not value: raise VariableError(_("Please select the build ID")) l = list(self.available()) if not l: raise VariableError(_("Build %s is not found") % value) if (self.Get('cl_builder_build').status == Build.Status.Broken and self.Get('cl_action') != Actions.Break): raise VariableError( _("Build %s is broken, try to restore build") % value) image_type = self.Get('cl_builder_image_type') if image_type == "iso": if not self._is_iso_compatible(): prefix = self.Get('cl_builder_path') for pn in ('sys-kernel/calculate-sources', 'sys-kernel/dracut', 'media-gfx/gfxboot-themes-calculate', 'sys-boot/grub'): if not isPkgInstalled(pn, prefix=prefix): raise VariableError( _("Build without %s can be container only") % pn) raise VariableError( _("Build without kernel can be container only")) class VariableClBuilderBrokenId(BaseBuildId): """ Id развёрнутой сборки """ type = "choice" untrusted = True def available(self): bs = self.Get('cl_builder_storage') for x in bs: build = bs.get_build(x) if build and build.status == Build.Status.Broken: yield x @is_action(Actions.Restore) def get(self): l = list(self.available()) if l and len(l) == 1: return l[0] return "" @is_action(Actions.Restore, default_value=[]) def choice(self): bs = self.Get('cl_builder_storage') for x in bs: build = bs.get_build(x) if build.status == Build.Status.Broken: yield x def check(self, value): if not value: raise VariableError(_("Please select the build ID")) l = list(self.available()) if not l: raise VariableError(_("Build %s is not found") % value) class VariableClBuilderIdPath(ReadonlyVariable): """ Преобразование сборки id в имя походящее для путей """ def get(self): build_id = self.Get('cl_builder_id') if build_id: return re.sub("[/:]", "_", self.Get('cl_builder_id')) return "" class VariableClBuilderId(ReadonlyVariable): """ Общий id сборки """ def get(self): action = self.Get('cl_action') if action in Actions.NewAssemble: return self.Get('cl_builder_new_id') elif action in Actions.WorkAssemble: return self.Get('cl_builder_prepared_id') elif action in Actions.BrokenAssemble: return self.Get('cl_builder_broken_id') return "" class VariableOsBuilderMakeopts(Variable): """ Параметры MAKEOPTS """ def get(self): return self.Get('install.os_install_makeopts') class VariableClBuilderBuildpkgSet(Variable): """ Собирать бинарные пакеты в сборке """ type = "bool" value = "off" class VariableClBuilderBasePath(Variable): """ Базовый путь до сборок (директория, куда будут помещены готовые iso образы, бинарные пакеты и т.д.) """ value = "/var/calculate/remote/builder" class VariableClBuilderParentPath(ReadonlyVariable): """ Путь в ".." до родительской системы """ def get(self): builder_path = self.Get('cl_builder_path') return ("../" * len([x for x in builder_path.split('/') if x]))[:-1] class VariableClBuilderStageSet(ReadonlyVariable): """ Разворачиваемый образ является stage (Gentoo системой) """ type = "bool" def get(self): return ("on" if self.Get('os_builder_linux_shortname') == "Gentoo" else "off") class VariableClBuilderPkgdir(Variable): """ Путь собираемых бинарных архивов """ def fallback(self): return path.join(self.Get('cl_builder_base_path'), self.Get('cl_builder_id_path')) def get(self): action = self.Get('cl_action') if action in Actions.NewAssemble: return self.fallback() elif action in Actions.WorkAssemble: build = self.Get('cl_builder_build') return build.pkgdir or self.fallback() else: return "" class VariableClBuilderPkgdirStable(Variable): """ Директория со стабильным набором бинарных пакетов """ value_format = "{cl_builder_pkgdir}.stable" class VariableClBuilderPkgdirFull(ReadonlyVariable): """ Директория с полным набором бинарных пакетов зависит от того использует ли builder base бинхост """ def get(self): if self.GetBool('cl_builder_binhost_base_set'): return self.Get('cl_builder_pkgdir_stable') else: return self.Get('cl_builder_pkgdir') class VariableClBuilderPkgdirTrunk(Variable): """ Директория со набором бинарных пакетов, осавшихся в результате прерывания сборки """ value_format = "{cl_builder_pkgdir}.trunk" class VariableClBuilderAction(ReadonlyVariable): """ Дополнительное действие по созданию образа: iso, squash. """ value = "" class VariableClBuilderImagePath(Variable): """ Путь по умолчанию для образов """ value = '/var/calculate/linux' class VariableClBuilderContainerPath(Variable): """ Путь по умолчанию для образов LXC """ value = '/var/calculate/container' class VariableClBuilderConainterIndexPath(Variable): """ Путь по умолчанию для создания индекса контейнеров """ value = "{cl_builder_container_path}/meta/1.0/index-system" class VariableClBuilderImageFilename(Variable): """ Название iso образа """ opt = ["--image"] metavalue = "IMAGE" untrusted = True check_after = "cl_builder_container_set" def init(self): self.label = _("Image path") self.help = _("set image path") def check(self, value): if self.Get('cl_builder_id'): if not value: raise VariableError(_("You must specify image filename")) if not self.GetBool("cl_builder_container_set"): if not value.endswith(".iso"): raise VariableError(_("You must specify image filename with " "'iso' suffix")) if path.isdir(value): raise VariableError(_("The image shouldn't be a directory")) for line in process("/bin/losetup", "-j", value): lodev, op, line = line.partition(":") if lodev: raise VariableError( _("Image %s is already used") % value) elif path.exists(value) and not path.isdir(value): raise VariableError(_("The image should be a directory")) def _isoname(self, dn, rewrite=False, addon_dn=None, suffix=".iso"): shortname = self.Get('os_builder_linux_shortname').lower() buildnumber = self.Get('os_builder_linux_build') arch = self.Get('os_builder_arch_machine') for i in range(0, 999): if i: isoname = "%s-%s-%d-%s%s" % (shortname, buildnumber, i, arch, suffix) else: isoname = "%s-%s-%s%s" % (shortname, buildnumber, arch, suffix) fullname = "" for i_dn in (x for x in (addon_dn, dn) if x): fullname = path.join(i_dn, isoname) if not rewrite and path.exists(fullname): break else: return fullname def get(self): build_id = self.Get('cl_builder_id') if self.GetBool('cl_builder_container_set'): base_dn = self.Get('cl_builder_container_path') else: base_dn = self.Get('cl_builder_image_path') if build_id: root_type_ext = self.Get('os_root_type_ext') if root_type_ext in RootType.RebuildAvailable: base_dn = self.Get('cl_builder_flash_repository') addon_dn = base_dn.replace("/linux", "/iso") return self._isoname(base_dn, rewrite=False, addon_dn=addon_dn) elif root_type_ext in RootType.IsoScanGrub: base_dn = self.Get('cl_builder_livemenu_path') if not base_dn: raise VariableError( _("Failed to determine the isoscan path")) return self._isoname(base_dn, rewrite=False) elif self.GetBool('cl_builder_container_set'): return self._isoname(base_dn, rewrite=False, suffix='') else: return self._isoname(base_dn, rewrite=False) return "" class VariableClBuilderIsoBasePath(Variable): """ Базовый путь, где будут подготавливаться данные, которые будут запакованы """ default_value = "/var/calculate/tmp" def get(self): root_type_ext = self.Get('os_root_type_ext') flash_path = self.Get('cl_builder_flash_path') isoscan_path = self.Get('cl_isoscan_base_path') if root_type_ext in RootType.RebuildAvailable and flash_path: return path.join(flash_path, "tmp") if root_type_ext in RootType.IsoScanGrub and isoscan_path: return path.join(isoscan_path, "tmp") return self.default_value class VariableClBuilderIsoPath(ImagePathHelper, ReadonlyVariable): """ Путь, где будут подготавливаться данные, которые будут запакованы в iso """ prefix = "iso-" class VariableClBuilderSquashPath(ReadonlyVariable): """ Путь от iso до содержимого squash """ @is_action(Actions.Image) def get(self): return path.relpath(self.Get('cl_builder_path'), self.Get('cl_builder_iso_path')) class VariableClBuilderImage(ReadonlyVariable): """ Создаваемый образ """ @is_action(Actions.Image) def get(self): image_name = self.Get('cl_builder_image_filename') bdn = self.Get('cl_builder_iso_path') exclude_list = self.Get('cl_builder_squash_exclude') include_list = self.Get('cl_builder_video_ebuilds') if self.GetBool('cl_builder_container_set'): image = distr.ContainerDistributive( image_name, exclude=exclude_list, include=include_list) else: image = distr.IsoDistributive( image_name, bdirectory=bdn, vol_id=self.Get('cl_builder_iso_label'), exclude=exclude_list, include=include_list, compress=self.Get('cl_builder_compress')) return image class VariableClBuilderLiveSet(Variable): """ Вызывать только live шаблоны при первой загрузке """ type = "bool" value = "on" opt = ["--live"] def init(self): self.help = _("use only live templates on startup") self.label = _("Use only live templates on startup") class VariableClBuilderCdname(ReadonlyVariable): """ Type of iso (CD/DVD) """ @is_action(Actions.Image) def get(self): squashfile = pathJoin(self.Get('cl_builder_iso_path'), self.Get('cl_builder_current_squash')) kernelfile = pathJoin(self.Get('cl_builder_iso_path'), self.Get('cl_builder_squash_path'), 'boot', self.Get('cl_builder_kernel')) initrdfile = pathJoin(self.Get('cl_builder_iso_path'), self.Get('cl_builder_squash_path'), 'boot', self.Get('cl_builder_initrd_install')) if os.access(squashfile, os.R_OK) and os.access(kernelfile, os.R_OK) and \ os.access(initrdfile, os.R_OK): isosize = path.getsize(squashfile) + path.getsize(kernelfile) + \ path.getsize(initrdfile) + 2 * 1024 * 1024 if isosize > 700 * 1024 * 1024: return "DVD" else: return "CD" return "" class VariableClBuilderIsoLabel(Variable): """ LABEL для iso """ @is_action(Actions.Image) def get(self): return "%s-%s" % (self.Get('os_builder_linux_shortname').upper(), self.Get('os_builder_linux_build')) class VariableClBuilderRootParam(Variable): """ параметр root= для livecd """ @is_action(Actions.Image) def get(self): return "live:LABEL=%s" % self.Get('cl_builder_iso_label') class VariableClBuilderCurrentSquash(ReadonlyVariable): """ Создаваемый livecd.squash """ value = "livecd.squashfs" class VariableClBuilderKernelCmd(ReadonlyVariable): """ Параметры по умолчанию для calcboot """ value = "" class KernelInfo(ReadonlyVariable): def get_current_kernel_src(self, prefix): src_path = "usr/src" current_linux_src = path.join(src_path, "linux") symlink_kernel = path.join(prefix, current_linux_src) if not path.exists(symlink_kernel) or not path.islink(symlink_kernel): raise ValueError("Failed to determine current kernel version") return path.join(src_path, os.readlink(symlink_kernel)) class VariableClBuilderKernelConfig(KernelInfo): """ Конфиг ядра """ def get(self): prefix = self.Get('cl_builder_path') if prefix: try: kernel_src = self.get_current_kernel_src(prefix) except ValueError: return "" config_path = path.join(kernel_src, ".config") return KernelConfig(path.join(prefix, config_path)) return "" class VariableClBuilderKernelVer(KernelInfo): """ Текущая версия ядра """ def init(self): self.label = _("Kernel version") def get(self): prefix = self.Get('cl_builder_path') if prefix: try: current_src = self.get_current_kernel_src(prefix) except ValueError: return "" src = path.join(prefix, current_src) return self.get_src_kernel_version(src) def get_config_version(self, configfile): re_config = re.compile("Automatically generated file;.*\n" ".*?Linux/\S+\s+(\S+)\s", re.M) if path.exists(configfile): with open(configfile) as f: match = re_config.search(f.read(200)) if match: return match.group(1) def get_src_kernel_version(self, kernel_src): """ Get version of kernel from .config """ config_path = path.join(kernel_src, ".config") makefile_path = path.join(kernel_src, "Makefile") # get version from config version = self.get_config_version(config_path) if version: return version # get version from Makefile re_makefile = re.compile("^VERSION = (\S+)\n" "PATCHLEVEL = (\S+)\n" "SUBLEVEL = (\S+)\n" "EXTRAVERSION = (\S*)\n", re.M) if path.exists(makefile_path): with open(makefile_path) as f: match = re_makefile.search(f.read(200)) if match: return "{0}.{1}.{2}{3}".format(*match.groups()) return "" class KernelData(ReadonlyVariable): """ Данные о текущем ядре """ kernel_object = "kernel" file_description = "" def filter(self, x, version=None): return x def list(self, prefix='/', bootdir='boot'): boot_dir = path.join(prefix, bootdir) return self.get_files_by_type(boot_dir, self.file_description) def get(self): prefix = self.Get('cl_builder_path') version = self.Get('cl_builder_kernel_ver') if not prefix: return "" obj_file = max_default( self.filter(self.list(prefix), version=version), key=path.getmtime, default="") if obj_file: obj_file = path.basename(obj_file) return obj_file def get_files_by_type(self, pathname, descr): ftype = typeFile(magic=0x4).getMType # print("DEBUG get files by type") # print("pathname: ", pathname) # print("listDirectory: ", listDirectory(pathname, fullPath=True)) # raise Exception(f"path: {pathname}") for x in listDirectory(pathname, fullPath=True): if descr in ftype(x): yield x class VariableClBuilderInitrdInstall(KernelData): """ Текущий initrd """ file_description = "ASCII cpio archive" def init(self): self.label = _("Init RAM fs") def list(self, prefix='/', bootdir='boot'): boot_dir = path.join(prefix, bootdir) for x in listDirectory(boot_dir, fullPath=True): if InitrdFile.is_cpio(x): yield x def filter(self, iterable, version=None): for fn in iterable: if InitrdFile(fn).get_kernel_version() == version: yield fn class VariableClBuilderKernel(KernelData): """ Текущее ядро """ file_description = "boot executable bzImage" def init(self): self.label = _("Kernel file") def filter(self, iterable, version=None): ftype = typeFile(magic=0x4).getMType re_kver = re.compile("bzImage, version (\S+)\s") for fn in iterable: m = re_kver.search(ftype(fn)) if m.group(1) == version: yield fn class VariableClBuilderTemplateLocation(Variable): """ Устанавливаются только дистрибутивные шаблоны """ type = "list" def get(self): return [x for x in self.Get('main.cl_template_location') if x not in ('remote', 'local')] class VariableClBuilderOutdateSet(ReadonlyVariable): """ Флаг устанавливаемый в ходе обновления репозиториев, сообщающий что хотя бы один из запланированных репозиториев обновлен и следует обновляет различные метаданные Если обновляются прочие оверлеи - данные считаются что устарели """ type = "bool" def get(self): if (self.Get('update.cl_update_other_set') == 'on' and self.Get('cl_builder_other_rep_name')): return "on" return "off" class VariableClBuilderBranchData(TableVariable): """ Выбор веток репозиториев до которых необходимо обновиться """ opt = ["--branch"] metavalue = 'REFS' untrusted = True source = ["cl_builder_branch_rep", "cl_builder_branch_name"] def init(self): self.help = _("set references for repository (REPOSITORY:REF)") self.label = _("Repositories references") def raiseReadonlyIndexError(self, fieldname="", variablename="", value=""): """ Неизвестный оврелей """ raise VariableError(_("Repository %s not found") % value) class VariableClBuilderBriefBranchData(ReadonlyTableVariable): """ Вывод синхронизируемые репозиториев для brief """ source = ["cl_builder_branch_rep", "cl_builder_branch_name"] def init(self): self.help = _("set references for repository (REPOSITORY:REF)") self.label = _("Repositories references") def get(self, hr=HumanReadable.No): return list(reversed(self.Get('cl_builder_branch_data', humanreadable=hr))) class VariableClBuilderBranchRep(ReadonlyVariable): """ Список доступных репозиториев """ type = "list" def init(self): self.label = _("Repository") def get(self): dv = self.Get('cl_builder_linux_datavars') if dv: return dv.Get('cl_update_rep_name') return [] def humanReadable(self): return [x.capitalize() for x in self.Get()] class VariableClBuilderBranchName(Variable): """ Список доступных репозиторием """ type = "choiceedit-list" def init(self): self.label = _("Reference") def choice(self): return [ Git.Reference.Tag, Git.Reference.Master, Git.Reference.Develop, Git.Reference.Update] def get(self): dv = self.Get('cl_builder_linux_datavars') if dv: if dv.GetBool('cl_update_usetag_set'): return [Git.Reference.Tag for x in dv.Get('cl_update_rep_name')] else: branch = dv.Get('cl_update_branch') return [branch for x in dv.Get('cl_update_rep_name')] return [] class VariableClBuilderCompress(IsoTypeHelper, Variable): """ Тип сжатия образа squash """ type = "choice" opt = ["-c", "--compress"] metavalue = "COMPRESS" untrusted = True check_after = ["cl_builder_id", "cl_builder_container_set"] def init(self): self.label = _("Compressor") self.help = _("set the compressor") def choice(self): config_param_prefix = "CONFIG_SQUASHFS_%s" kernel_config = self.Get('cl_builder_kernel_config') params = {"xz": "XZ", "lzma": "XZ", "lzo": "LZO", "gzip": "ZLIB"} def generator(): for compress in getSquashList(): if compress in params: if config_param_prefix % params[compress] in kernel_config: yield compress else: yield compress weight = {'zstd': 4, 'xz': 3, 'lz4': 2, 'gzip': 1} ret = list(sorted(generator(), key=lambda x: (-weight.get(x, 0), x))) if not ret: return [""] return ret def get(self): if self.GetBool("cl_builder_container_set"): return "xz" values = self.choice() if values: return values[0] return "" class VariableClBuilderIsohybridSet(IsoTypeHelper, Variable): """ Преобразовать полученный iso образ в гибридный """ type = "bool" opt = ["--isohybrid"] check_after = ["cl_builder_container_set"] def init(self): self.help = _("create the ISO image with isohybrid") self.label = _("ISO hybrid feature") def check(self, value): if self.GetBool("cl_builder_container_set") and value != "off": raise VariableError(_("Iso hybrid is uncompatible with container")) def get(self): if self.GetBool("cl_builder_container_set"): return "off" return "on" class VariableClBuilderKeepTreeSet(Variable): """ Не удалять ебилды из портежей о оверлеев """ type = "bool" opt = ["--keep-tree"] value = "on" def init(self): self.help = _("keep portage tree in image") self.label = _("Keep portage tree") class VariableClBuilderVideoEbuilds(Variable): """ Переменная заполняемая при выполнении задачи утилиты """ type = "list" value = [] class VariableClBuilderSquashExclude(ReadonlyVariable): """ Список файлов, которые не будут запакованы в livecd.squashfs """ type = "list" def get(self): builder_path = self.Get('cl_builder_path') keep_tree = self.GetBool('cl_builder_keep_tree_set') excludes = ["", "metadata"] important = ["distfiles", "packages", "eclass", "metadata", "profiles", "layout.conf"] exclude_dirs = ["profiles/templates/deprecated", "profiles/updates"] important_files = self.Get('cl_builder_video_ebuilds') def generator(): for rep_dn in self.Get('cl_builder_repository_location'): for exclude_dn in excludes: dn = pathJoin(builder_path, rep_dn, exclude_dn) for exclude_name in (x for x in listDirectory(dn) if x not in important): yield path.join(rep_dn, exclude_dn, exclude_name)[1:] for exclude_dn in exclude_dirs: dn = pathJoin(builder_path, rep_dn, exclude_dn) if path.exists(dn): yield path.join(rep_dn, exclude_dn)[1:] if not keep_tree and builder_path: return list(generator()) return [] class VariableClBuilderPrelinkSet(Variable): """ Выполнять ли prelink """ type = Variable.Types.Boolean opt = ["--prelink"] value = Variable.Off def init(self): self.help = _("perform prelink") self.label = _("Perform prelink") class VariableClBuilderRebuildChangedSet(Variable): """ """ type = Variable.Types.Boolean opt = ["--rebuild-changed-packages"] value = Variable.On def init(self): self.help = _("rebuild changed packages") self.label = _("Rebuild changed packages") class VariableClBuilderRebuiltBinariesSet(Variable): """ Выполнить замещение пакетов на бинарную версию если их BUILDTIME меньше, чем в Packages """ type = Variable.Types.Boolean opt = ["-B", "--rebuilt-binaries"] value = Variable.Off def init(self): self.help = _("replace installed packages with binary packages " "that have been rebuilt") self.label = _("Rebuild installed packages with binary packages") class VariableClBuilderRescratchSet(ReadonlyVariable): """ Сборка iso образа на самой флешке """ type = "bool" def get(self): if self.Get('os_root_type_ext') in RootType.RebuildAvailable: isoimage = self.Get('cl_builder_image_filename') flash_rep = self.Get('cl_builder_flash_repository') isoimage_dir = path.dirname(isoimage) if isoimage_dir == flash_rep: return "on" return "off" class VariableClBuilderFlashRepository(ReadonlyVariable): """ Хранилище по умолчанию для iso образов на flash """ suffix = "linux" def get(self): flash_path = self.Get('cl_builder_flash_path') if flash_path: return path.join(flash_path, self.suffix) return "" class VariableClBuilderLivemenuType(Variable): """ Тип обновляемого live меню """ type = "choice" opt = ["--type"] metavalue = "TYPE" def init(self): self.label = _("Type") self.help = _("set livemenu type") untrusted = True def choice(self): return [("grub", _("Grub")), ("flash", _("Flash"))] def get(self): root_type_ext = self.Get('os_root_type_ext') if root_type_ext in RootType.Grubable: return "grub" elif root_type_ext in RootType.RebuildAvailable: return "flash" return "flash" class VariableClBuilderImageRepository(ReadonlyVariable): """ Хранилище образов """ def get(self): livemenu_type = self.Get('cl_builder_livemenu_type') if livemenu_type == "grub": return self.Get('cl_builder_livemenu_path') else: return self.Get('cl_builder_flash_repository') class VariableClLivemenuPath(ReadonlyVariable): """ Текущий путь до репозитория образов """ grub_config = "/etc/grub.d/35_calculate" def get(self): data = readFile(self.grub_config) m = re.search(r'^livepath=([^\n]+)', data, re.M) if m: return m.group(1).strip("'\"") return "" class VariableClBuilderLivemenuPath(Variable): """ Путь до устройства flash """ opt = ["cl_builder_livemenu_path"] type = "choiceedit" metavalue = "PATH" untrusted = True none_humanreadable = __("Clear the boot from ISO images") def init(self): self.help = _("set path") self.label = _("Path") def get_isoscan_grub_repository(self): isoscan_path = self.Get('cl_isoscan_fullpath') if isoscan_path: return path.dirname(isoscan_path) return "" def get_live_path(self): live_path = self.Get('cl_livemenu_path') if not live_path and self.Get('cl_action') == BuilderActions.Image: live_path = self.Get('cl_live_path') return live_path def get(self): menutype = self.Get('cl_builder_livemenu_type') root_type_ext = self.Get('os_root_type_ext') if menutype == "flash": if root_type_ext in RootType.LiveFlash: return Dracut.IsoLive if root_type_ext in RootType.IsoScanFlash: return Dracut.IsoScanPath else: if root_type_ext in RootType.IsoScanGrub: return self.get_isoscan_grub_repository() elif root_type_ext in RootType.Grubable: return self.get_live_path() return "" def choice(self): root_type_ext = self.Get('os_root_type_ext') if root_type_ext in RootType.LiveFlash: yield (Dracut.IsoLive, _("Boot Flash")) elif root_type_ext in RootType.IsoScanFlash: yield (Dracut.IsoScanPath, _("Boot Flash")) elif root_type_ext in RootType.IsoScanGrub: grub_rep = self.get_isoscan_grub_repository() if grub_rep: yield (grub_rep, _("HDD Grub")) elif root_type_ext in RootType.Grubable: live_path = self.get_live_path() if live_path: yield (live_path, _("Distributions directory (%s)") % live_path) for dev in self.select('install.os_disk_dev', install_os_disk_format="vfat"): yield (dev, self.flash_name(dev)) if root_type_ext in RootType.Grubable and \ root_type_ext not in RootType.IsoScanGrub: yield ("none", str(self.none_humanreadable)) def check_flash(self, value): try: with IsoDistributive(value) as image: d = image.getIsoContentDirectory() if not check_rw(d): raise ValueError("Selected device is read-only") except DistributiveError as e: raise VariableError(str(e).strip()) def flash_name(self, dev): parent_dev = self.select("install.os_disk_parent", install_os_disk_dev=dev, limit=1) if parent_dev: dev_name = self.select("install.os_device_name", install_os_device_dev=parent_dev, limit=1) if dev_name: return "%s (%s)" % (dev_name, dev) return dev def humanReadable(self): """ Преобразовать указанный путь в устройство и получить его имя """ value = self.Get() if not value: return str(self.none_humanreadable) if not value.startswith('/dev/'): m = isMount(value) else: m = value if m: return self.flash_name(m) return value def check(self, value): livemenu_type = self.Get('cl_builder_livemenu_type') if livemenu_type == "flash": if not value: raise VariableError(_("Please specify the Flash drive")) if not path.exists(value): raise VariableError(_("Flash drive %s not found") % value) if not self.Get('cl_builder_flash_path'): self.check_flash(value) else: root_type_ext = self.Get('os_root_type_ext') if not value and root_type_ext in RootType.IsoScanGrub: raise VariableError(_("Please specify the images directory")) if (not path.exists(value) or value.startswith('/dev') or not path.isdir(value)) and value: raise VariableError(_("Please specify the directory")) if root_type_ext in RootType.Grubable: if root_type_ext in RootType.IsoScanGrub: grub_rep = self.get_isoscan_grub_repository() if grub_rep and value != grub_rep: raise VariableError( _("You cannot change the images directory")) else: if value: gc = GrubCommand() if not gc.is_available(value): raise VariableError( _("%s is unavailable for the Grub") % value) else: grub_file = path.join(value, "grub.cfg") if not path.exists(grub_file): raise VariableError( _("Please specify the LiveHDD directory")) def set(self, value): if value == "none": value = "" if value: return path.normpath(value) return "" class VariableClBuilderFlashDev(ReadonlyVariable): """ Устройство flash """ object = "" def get(self): typemenu = self.Get('cl_builder_livemenu_type') if typemenu == "flash": try: iso_path = self.Get('cl_builder_livemenu_path') if iso_path.startswith('/dev/') and isMount(iso_path): iso_path = isMount(iso_path) if iso_path: return IsoDistributive(iso_path) except DistributiveError: return "" return "" class VariableClBuilderFlashPath(ReadonlyVariable): """ Путь, куда подключена flash (загрузка либо с флешки, либо с iso) """ def get(self): try: iso = self.Get('cl_builder_flash_dev') if iso: return iso.getIsoContentDirectory() except DistributiveError: return "" return "" class VariableClBuilderFlashUuid(ReadonlyVariable): """ UUID flash, используется для перестроения меню """ def get(self): flash_path = self.Get('cl_builder_flash_path') if flash_path: dev = isMount(flash_path) if dev: return self.select( 'install.os_disk_uuid', install_os_disk_dev=dev, limit=1) return "" class VariableClBuilderGrubRefreshSet(ReadonlyVariable): """ Обновить меню grub (HDD c включенным LiveHDD или isoscan-grub """ def get(self): root_type_ext = self.Get('os_root_type_ext') action = self.Get('cl_action') if action == Actions.Image: if root_type_ext in RootType.HDDs: rep_dir = self.Get('cl_livemenu_path') if rep_dir: isoimage = self.Get('cl_builder_image_filename') isoimage_dir = path.dirname(isoimage) if isoimage_dir == rep_dir: return "on" elif root_type_ext in RootType.IsoScanGrub: isoimage = self.Get('cl_builder_image_filename') rep_dir = self.Get('cl_builder_livemenu_path') isoimage_dir = path.dirname(isoimage) if isoimage_dir == rep_dir: return "on" return "off" class VariableClBuilderEixRepositories(ReadonlyVariable): """ Отображаемый список обновляемых репозиториев eix """ def get(self): return ", ".join( x.capitalize() for x in reversed(list(chain( self.Get('update.cl_update_rep_name'), self.Get('builder.cl_builder_other_rep_name'))))) class VariableClBuilderBinhostStableSet( vars_linux.BuilderLinux, update.VariableClUpdateBinhostStableSet): """ Использовать стабильное или нестабильное зеркало """ variable = "cl_update_binhost_stable_set" def init(self): self.label = _("Use only stable updates") def get_fallback(self): return Variable.On class VariableClBuilderBinhostStableOptSet( update.VariableClUpdateBinhostStableOptSet): """ Использовать стабильное или нестабильное зеркало """ mode = Variable.Writable def get(self): return self.Get('cl_builder_binhost_stable_set') class VariableClBuilderBinhostScanSet(Variable): """ Сканировать или нет зеркала (переменная переключается во время действия) """ value = "off" class VariableClBuilderCheckAutomagicSet(Variable): """ Выполнить проверку на наличие неописанных автоматических зависимостей """ type = "bool" value = "off" opt = ["--check-autodep"] def init(self): self.label = _("Check for auto depends") self.help = _("check for auto depends") class VariableClBuilderUpdateThemesSet(Variable): """ Перенастроить темы перед запаковкой """ type = "bool" value = "off" opt = ["--update-themes", "-U"] def init(self): self.label = _("Update themes") self.help = _("update themes") class VariableClBuilderImageType(Variable): """ Тип собираемого образа """ type = "choice" def init(self): self.label = _("Image type") self.help = _("set image type") def choice(self): return (("container", _("Container")), ("iso", _("ISO"))) class VariableClBuilderContainerSet(ReadonlyVariable): """ Запаковать образ как LXC контейнер """ type = "bool" def get(self): if self.Get('cl_builder_image_type') == 'container': return 'on' return 'off' class VariableClBuilderContainerDist(Variable): """ Название операционной системы """ value = "calculate" class VariableClBuilderContainerRelease(Variable): """ Версия ОС контейнера """ def get(self): return self.Get('os_builder_linux_ver') class VariableClBuilderContainerArch(Variable): """ Архитектура ОС в контейнере """ def get(self): return self.Get('os_builder_arch_machine_gentoo') class VariableClBuilderContainerVariant(Variable): """ Variant of image """ value = "default" class VariableClBuilderContainerExpiryTimestamp(Variable): """ Expire date """ month = 60 * 60 * 24 * 30 type = "int" def get(self): return str(int(self.Get('cl_builder_container_create_timestamp')) + self.month) class VariableClBuilderContainerCreateTimestamp(Variable): """ Expire date """ type = "int" def get(self): return datetime.datetime.now().strftime("%s") class VariableClBuilderContainerDate(Variable): """ Creation date of image """ def get(self): ts = int(self.Get('cl_builder_container_create_timestamp')) return datetime.datetime.fromtimestamp(ts).strftime("%Y%m%d_%H:%M") class VariableClBuilderContainerBasePath(Variable): """ Базовый путь для подготовки архивов meta.tar.xz lxd.tar.xz """ value = "/var/calculate/tmp" class VariableClBuilderContainerDataPath(ImagePathHelper, Variable): """ Путь где будут подготавливаться данные для контейнера """ @property def base_dn(self): return self.Get('cl_builder_container_base_path') prefix = "lxc-" class VariableClBuilderWithBdepsSet(Variable): """ Добавление --with-bdeps при вызове emerge """ type = Variable.Types.Tristate def get(self): dv = self.Get('cl_builder_linux_datavars') if dv: return dv.Get('cl_update_with_bdeps_set') return Variable.Auto class VariableClBuilderWithBdepsOptSet(Variable): """ Добавление --with-bdeps при вызове emerge (опция) """ type = Variable.Types.Tristate opt = ["--with-bdeps"] metavalue = "ON/OFF/AUTO" def init(self): self.help = _("save packages used during build") self.label = _("Save packages used during build") def get(self): return self.Get('cl_builder_with_bdeps_set') class VariableClBuilderBootParams(Variable): """ Параметры загрузки ядра в собираемом образе и boot live """ value = "" class VariableClCheckUpdateLevel(Variable): """ Проверять уровень обновления при соединении с binhost """ type = "bool" value = "off"