您最多能選擇 25 個主題 主題必須以字母或數字為開頭,可包含連接號「-」且最長為 35 個字元。
calculate-utils-3-update/pym/update/variables/update.py

2137 行
64 KiB

此檔案含有易混淆的 Unicode 字元!

此檔案含有易混淆的 Unicode 字元,這些字元的處理方式可能和下面呈現的不同。若您是有意且合理的使用,您可以放心地忽略此警告。使用 Escape 按鈕標記這些字元。

# -*- coding: utf-8 -*-
# Copyright 2013-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.
import os
import sys
import re
from os import path
import datetime
from calculate.lib.cl_template import SystemIni
from calculate.lib.cl_log import log
from calculate.lib.datavars import (Variable, VariableError,
ReadonlyVariable, ReadonlyTableVariable,
TableVariable, FieldValue,
HumanReadable, CriticalError,
SimpleDataVars, DataVarsError)
from calculate.lib.utils.binhosts import (Binhosts, PackagesIndex, HOURS,
BinhostSignError)
from calculate.lib.utils.files import readFile, listDirectory, process, pathJoin, readLinesFile
from calculate.lib.configparser import ConfigParser
from calculate.lib.cl_lang import setLocalTranslate
from calculate.lib.utils.text import simplify_profiles, _u8
from calculate.lib.utils.git import Git, GitError
from calculate.lib.utils.portage import ReposConf, PortageState
from ..profile import (RepositoryStorageSet, DEFAULT_BRANCH,
LocalStorage, ProfileRepository, CacheStorage, Profile)
from calculate.lib.variables import linux as lib_linux
from calculate.lib.variables import env
from calculate.lib.variables import system as lib_system
from ..update_info import UpdateInfo
from itertools import chain
from urllib.parse import urlparse
import io
import time
_ = lambda x: x
setLocalTranslate('cl_update3', sys.modules[__name__])
class VariableAcUpdateSync(ReadonlyVariable):
"""
Action variable which has value "up" for package install and
install this package
"""
def get(self):
action = self.Get("cl_action")
if action in ("sync",):
if self.Get('cl_update_world'):
return self.Get('cl_update_world')
return ""
class VariableClUpdateWorldDefault(Variable):
"""
Ad-hoc
"""
value = "update"
class VariableClUpdateWorld(Variable):
type = "choice"
opt = ["--world"]
syntax = "--{choice}-world"
metavalue = "MODE"
def init(self):
self.help = ("'rebuild' - " + _("rebuild the system packages list") +
",\n'merge' - " + _("append the profile packages list") +
",\n'update' - " + _("update the system packages list"))
self.label = _("System packages list")
def get(self):
return self.Get('cl_update_world_default')
def choice(self):
return [("rebuild", _("Rebuild")),
("merge", _("Merge")),
("update", _("Update"))]
class VariableClRebuildWorldSet(Variable):
"""
List of action update world, rebuild world,
"""
type = "bool"
opt = ["--rebuild-world"]
untrusted = True
value = "off"
def init(self):
self.help = _("Rebuild world")
self.label = _("Rebuild world")
class VariableClUpdateRevSet(Variable):
"""
List of action update world, rebuild world,
"""
type = "bool"
opt = ["--update-rev"]
untrusted = True
value = "on"
check_after = ["cl_update_sync_rep",
"cl_update_metadata_force",
"cl_update_other_set",
"cl_update_eixupdate_force"]
def init(self):
self.help = _("make a revision update")
self.label = _("Make a revision update")
class VariableClUpdateRep(Variable):
"""
Обновлять репозитории до конкретной ревизии или до последней
"""
type = "choice"
value = "rev"
def choice(self):
return ["last", "rev"]
class VariableClUpdateRepData(ReadonlyTableVariable):
"""
Информация о репозиториях
"""
source = ['cl_update_rep_name',
'cl_update_rep_url',
'cl_update_rep_path',
'cl_update_rep_rev']
class VariableClUpdateRepName(Variable):
"""
Список имен используемых репозиториев
"""
type = "list"
value = []
def humanReadable(self):
return [x.capitalize() for x in self.Get()]
class VariableClUpdateRepUrl(Variable):
"""
Список путей до репозиториев
"""
type = "list"
value = []
class VariableClUpdateRepList(Variable):
"""
Список путей до репозиториев
"""
type = "list"
value = []
class VariableClUpdateInnerRepList(Variable):
value = []
check_after = ['cl_update_rep_list', 'cl_update_rep_hosting']
def get(self):
def gen():
val = self.Get('cl_update_rep_list') or ''
prev_host = self.Get('cl_update_rep_hosting') or ''
if prev_host:
for dct in val:
for key, value in dct.items():
if key == prev_host:
yield value, key
for dct in val:
for key, value in dct.items():
if key != prev_host:
yield value, key
return [x for x in gen()]
class VariableClUpdateRepHosting(Variable):
value = ''
class VariableClUpdateRepHostingChoice(Variable):
type = "choice"
opt = ["--repo-update"]
metavalue = "REPOSITORY URL"
check_after = ['cl_update_inner_rep_list']
def init(self):
self.label = (_("Update repository"))
self.help = (_("Hosting to download repository from"))
def get(self):
if not self.value:
if self.Get('cl_update_rep_hosting'):
self.value = self.Get('cl_update_rep_hosting')
else:
self.value = [x[1] for x in self.Get('cl_update_inner_rep_list')][0] or ['Default']
return self.value
def choice(self):
return [x[1] for x in self.Get('cl_update_inner_rep_list')] or ['Default']
class VariableClUpdateLaymanStorage(Variable):
"""
Путь к репозиториям
"""
param_name = "storage"
fallback_value = "var/lib/layman"
def get(self):
return pathJoin(self.Get('cl_chroot_path'), self.fallback_value)
class VariableClUpdateReposStorage(Variable):
"""
Путь к репозиториям
"""
param_name = "storage"
fallback_value = "var/db/repos"
def get(self):
return pathJoin(self.Get('cl_chroot_path'), self.fallback_value)
class VariableClUpdateRepPath(ReadonlyVariable):
"""
Пути до репозиториев
"""
type = "list"
def get(self):
repPath = self.Get('cl_update_repos_storage')
chroot_path = self.Get('cl_chroot_path')
if os.path.isdir(path.join(chroot_path, 'var/db/repos/gentoo')):
mapPath = {'portage': 'var/db/repos/gentoo',
'gentoo': 'var/db/repos/gentoo'}
else:
mapPath = {'portage': 'usr/portage',
'gentoo': 'usr/portage'}
def generatePaths(names):
for name in names:
if name in mapPath:
yield path.join(chroot_path, mapPath[name])
else:
yield path.join(repPath, name)
return list(generatePaths(self.Get('cl_update_rep_name')))
class VariableClUpdateRepRev(Variable):
"""
Ревизии до которых необходимо обновить репозитории
"""
type = "list"
def get(self):
cp = ConfigParser()
revisions = self.Get('update.cl_update_binhost_revisions')
if revisions:
cp.read_string(revisions[0])
branch = self.Get('cl_update_branch')
def generateBranch():
for repname, branchname in zip(
self.Get('cl_update_rep_name'),
self.Get('cl_update_branch_name')):
if branchname == Git.Reference.Tag:
yield cp.get("vcs", repname, fallback=branch)
else:
yield branchname
return list(generateBranch())
class VariableClUpdateBranch(Variable):
"""
Ветка на которую будет обновляться репозиторий если
не используется синхронизация по тэгам
"""
value = Git.Reference.Master
arch_var = "main.os_arch_machine"
def get(self):
arch = self.Get(self.arch_var)
if arch == "x86_64":
return "master"
else:
return "master"
class VariableClUpdateBranchData(TableVariable):
"""
Выбор веток репозиториев до которых необходимо обновиться
"""
opt = ["--branch"]
metavalue = 'REFS'
untrusted = True
source = ["cl_update_branch_rep",
"cl_update_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 VariableClUpdateBranchRep(ReadonlyVariable):
"""
Список доступных репозиторием
"""
type = "list"
def init(self):
self.label = _("Repository")
def get(self):
return self.Get('cl_update_rep_name')
class VariableClUpdateBinhostSet(Variable):
type = "bool"
value = 'off'
class VariableClUpdateBinhostsInfo(ReadonlyVariable):
"""
Объект для получения информации о серверах обновлений
"""
type = Variable.Types.Object
def get(self):
last_ts = self.Get('cl_update_last_timestamp')
if self.GetBool('cl_update_binhost_stable_opt_set'):
binhost_list = self.Get('cl_update_binhost_list')
else:
binhost_list = self.Get('cl_update_binhost_unstable_list')
return Binhosts(
self.GetInteger('cl_update_binhost_timeout'),
self.Get('cl_update_binhost_revision_path'),
self.Get('cl_update_binhost_timestamp_path'),
last_ts, binhost_list,
self.Get('os_arch_machine'),
gpg=self.Get('cl_update_gpg'))
class VariableClUpdateBinhostData(ReadonlyTableVariable):
"""
Таблица содержащая
binhost/содержимое файла ревизий/время доступа
"""
source = ["cl_update_binhost_host",
"cl_update_binhost_revisions",
"cl_update_binhost_timestamp",
"cl_update_binhost_time",
"cl_update_binhost_level"]
def get(self, hr=HumanReadable.No):
return Variable.EmptyTable
class VariableClUpdateBinhostRecheckSet(Variable):
"""
Принудительно обновить binhost
"""
type = Variable.Types.Boolean
value = Variable.Off
opt = ["--scan"]
def init(self):
self.help = _("search for the most appropriate update server")
self.label = _("Search for the most appropriate update server")
class VariableClUpdateBinhostHost(FieldValue, Variable):
"""
Список имен прочих репозиториев
"""
type = "list"
source_variable = "cl_update_binhost_data"
column = 0
class VariableClUpdateBinhostRevisions(FieldValue, Variable):
"""
Список имен прочих репозиториев
"""
type = "list"
source_variable = "cl_update_binhost_data"
column = 1
class VariableClUpdateBinhostTimestamp(FieldValue, Variable):
"""
Список имен прочих репозиториев
"""
type = "list"
source_variable = "cl_update_binhost_data"
column = 2
class VariableClUpdateBinhostTime(FieldValue, Variable):
"""
Список имен прочих репозиториев
"""
type = "list"
source_variable = "cl_update_binhost_data"
column = 3
class VariableClUpdateBinhostLevel(FieldValue, Variable):
"""
Список имен прочих репозиториев
"""
type = "list"
source_variable = "cl_update_binhost_data"
column = 4
class VariableClUpdateLastTimestamp(ReadonlyVariable):
"""
Текущий timestamp
"""
def get(self):
ini = SystemIni(self.parent)
return ini.getVar('system', 'last_update') or "0"
class VariableClUpdateSettingsChangesSet(Variable):
"""
Определить были ли изменения в /etc/portage, /var/log/emerge.log
"""
type = Variable.Types.Boolean
value = ''
def get(self):
if self.value:
return self.value
ini = SystemIni(self.parent)
old_portage_state_hash = ini.getVar('system', 'portage_hash') or ""
ps = PortageState()
new_portage_state_hash = ps.get_state()
return "on" if new_portage_state_hash != old_portage_state_hash else "off"
class VariableClUpdateSettingsPortageSet(Variable):
"""
Определить были ли изменения в /etc/portage
"""
type = Variable.Types.Boolean
value = ''
def get(self):
if self.value:
return self.value
ini = SystemIni(self.parent)
old_portage_state_hash = ini.getVar('system', 'portage_hash') or ""
ps = PortageState()
new_portage_state_hash = ps.get_state()
a = new_portage_state_hash != old_portage_state_hash
return "on" if new_portage_state_hash != old_portage_state_hash else "off"
class VariableClUpdateBranchName(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):
if self.GetBool('cl_update_usetag_set'):
return [Git.Reference.Tag for x in self.Get('cl_update_rep_name')]
else:
branch = self.Get('cl_update_branch')
return [branch for x in self.Get('cl_update_rep_name')]
class VariableClUpdateSyncRep(Variable):
"""
Обновляемый репозиторий
"""
type = "choice-list"
element = "selecttable"
opt = ["-r", "--repositories"]
metavalue = "REPOSITORIES"
untrusted = True
@property
def rep_name(self):
return self.Get('update.cl_update_rep_name')
def init(self):
self.help = _("synchronized repositories (all by default)")
self.label = _("Synchronized repositories")
def set(self, value):
orderList = self.rep_name
return sorted(value, key=lambda x:
(orderList.index(x) if x in orderList else -1), reverse=True)
def get(self):
return list(reversed(self.rep_name))
def choice(self):
return [ (x, x.capitalize()) for x in self.rep_name]
class VariableClUpdateSyncOverlayRep(ReadonlyVariable):
"""
Обновляемые репозитории (исключая portage)
"""
type = "list"
def get(self):
return [x for x in self.Get('cl_update_sync_rep') if x not in ("portage", "gentoo")]
class VariableClUpdateOutdateSet(ReadonlyVariable):
"""
Флаг устанавливаемый в ходе обновления репозиториев,
сообщающий что хотя бы один из запланированных репозиториев
обновлен и следует обновляет различные метаданные
Если обновляются прочие оверлеи - данные считаются что устарели
"""
type = "bool"
def get(self):
# if (self.Get('cl_update_other_set') == 'on' and
# self.Get('cl_update_other_rep_name')):
# return "on"
return "off"
class VariableClUpdateMetadataForce(Variable):
"""
Принудительное действие с метаданными
"""
type = "choice"
value = "auto"
opt = ["--update-metadata"]
syntax = "--{choice}-update-metadata"
metavalue = "MODE"
# untrusted = True
def init(self):
self.help = ("'force' - " + _("force the update ebuilds metadata") +
",\n'skip' - " + _("skip the ebuild metadata update") +
",\n'auto' - " + _("update metadata if it is outdated"))
self.label = _("Update metadata")
def choice(self):
return [("force", _("Force")),
("skip", _("Skip")),
("auto", _("If needed"))]
class VariableClUpdateEgencacheForce(Variable):
"""
Принудительное выполнение egencache
"""
type = "choice"
value = "auto"
opt = ["--egencache"]
syntax = "--{choice}-egencache"
metavalue = "MODE"
# untrusted = True
def init(self):
self.help = (
"'force' - " + _("force the update of the overlays cache") +
",\n'skip' - " + _("skip the update of the overlays cache") +
",\n'auto' - " + _("update the overlays cache if outdated"))
self.label = _("Update the overlays cache")
def choice(self):
return [("force", _("Force")),
("skip", _("Skip")),
("auto", _("If needed"))]
class VariableClUpdateEixupdateForce(Variable):
"""
Принудительное действие с eix-update
"""
type = "choice"
value = "auto"
opt = ["--eix-update"]
syntax = "--{choice}-eix-update"
metavalue = "MODE"
# untrusted = True
def init(self):
self.help = ("'force' - " + _("force the eix cache update") +
",\n'skip' - " + _("skip the eix cache update") +
",\n'auto' - " + _("update the eix cache if it "
"is outdated"))
self.label = _("Update the eix cache")
def choice(self):
return [("force", _("Force")),
("skip", _("Skip")),
("auto", _("If needed"))]
class VariableClUpdateGpgForce(Variable):
"""
Принудительное действие с eix-update
"""
type = "choice"
value = "force"
opt = ["--check-sign"]
syntax = "--{choice}-check-sign"
metavalue = "MODE"
# untrusted = True
def init(self):
self.help = ("'force' - " + _("force check Package signature") +
",\n'skip' - " + _("skip check Package signature") +
",\n'auto' - " + _("check signature if system has public keys"))
self.label = _("Check Package signature")
def choice(self):
return [("force", _("Force")),
("skip", _("Skip")),
("auto", _("If has passibility"))]
class VariableClUpdateOtherSet(Variable):
"""
Обновить остальные оверлеи
"""
type = "bool3"
opt = ["-o", "--update-other"]
metavalue = _("ON/OFF/AUTO")
value = Variable.Auto
def init(self):
self.label = _("Update other overlays")
self.help = _("Update other overlays. Updates other overlays while updating default overlays if value is not specified")
def get(self):
return self.value
class VariableClUpdateOtherGitExists(Variable):
"""
Проверка, существует ли папка .git в профиле контейнера
"""
type = 'bool'
repos_path = 'cl_update_repos_storage'
def get(self):
repos = self.Get(self.repos_path)
for rep in listDirectory(repos):
if not os.path.exists(os.path.join(repos, os.path.join(rep, '.git'))):
return False
return True
class VariableClUpdateOtherRepData(ReadonlyTableVariable):
"""
Информация о прочих репозиториях
"""
source = ['cl_update_other_rep_name',
'cl_update_other_rep_path']
def generator(self):
repNames = self.Get('update.cl_update_rep_name')
chroot_path = self.Get('cl_chroot_path')
reposconf = ReposConf(
self.Get('update.cl_update_reposconf'),
self.Get('update.cl_update_reposconf_dir'),
prefix=chroot_path)
for rname, rpath in reposconf.get_other_repositories():
yield (_u8(rname), pathJoin(chroot_path, _u8(rpath)))
def get(self, hr=HumanReadable.No):
return list(self.generator())
class VariableClUpdateOtherRepName(FieldValue, ReadonlyVariable):
"""
Список имен прочих репозиториев
"""
type = "list"
source_variable = "cl_update_other_rep_data"
column = 0
class VariableClUpdateOtherRepPath(FieldValue, ReadonlyVariable):
"""
Список путей до прочих репозиториев
"""
type = "list"
source_variable = "cl_update_other_rep_data"
column = 1
class VariableClUpdateReposconfDir(ReadonlyVariable):
"""
Директорий с файлами, описывающими репозитории
"""
def get(self):
chroot = self.Get('cl_chroot_path')
return path.join(chroot, '/etc/portage/repos.conf')
class VariableClUpdateReposconf(ReadonlyVariable):
"""
Путь конфигурационного файла repos.conf, содержащего Calculate репозитории
"""
value_format = "{update.cl_update_reposconf_dir}/zz-calculate.conf"
class VariableClUpdatePretendSet(Variable):
"""
Запустить предварительную проверку на обновления
"""
type = "bool"
value = "off"
opt = ["-p", "--pretend"]
def init(self):
self.label = _("Pretend a package update")
self.help = _("instead of actually performing the update, "
"simply display the list of packages that "
"will be installed")
class VariableClUpdateSyncOnlySet(Variable):
"""
Не выполнять установку/обновление пакетов при обновлении
"""
type = "bool"
value = "off"
opt = ["-s", "--sync-only"]
def init(self):
self.label = _("Only synchronize repositories")
self.help = _("do not update packages")
def check(self, value):
if (value == "on" and self.Get('cl_rebuild_world_set') != 'on' and
not self.Get('cl_update_sync_rep') and
self.Get('cl_update_other_set') == 'off' and
self.Get('cl_update_rev_set') == 'off' and
self.Get('cl_update_metadata_force') != "force" and
self.Get('cl_update_eixupdate_force') != "force"):
raise VariableError(_("Select at least one sync repository"))
class VariableClUpdateWaitAnotherSet(Variable):
"""
Ждать завершения другого процесса обновления
"""
type = "bool"
value = "on"
opt = ["--wait-another-update"]
def init(self):
self.label = _("Wait for another update to be complete")
self.help = _("wait until the running update is finished")
class VariableClUpdateProfileStorage(ReadonlyVariable):
type = "object"
def get(self):
git = self.Get('cl_update_git')
return RepositoryStorageSet(
LocalStorage(git, '/var/db/repos'),
CacheStorage(git, '/var/calculate/tmp/update'))
class VariableClUpdateGit(ReadonlyVariable):
type = "object"
def get(self):
sshkey = self.Get('update.cl_update_sshkey_path')
if path.exists(sshkey):
return Git(sshkey)
else:
return Git()
class VariableClUpdateRepHost(Variable):
type = "list"
value = ['calculate']
class VariableClUpdateRepHostUrl(Variable):
type = "list"
value = ['https://github.com/calculatelinux/%s.git']
class VariableClUpdateProfileDatavars(ReadonlyVariable):
type = "object"
profile = "cl_update_profile_system"
profiles_path = "cl_update_profile_path"
profiles_shortname = "cl_update_profile_shortname"
def get(self):
profile = self.Get(self.profile)
path_profile = self.Select(self.profiles_path,
where=self.profiles_shortname,
eq=profile, limit=1)
if path_profile:
return DataVarsUpdateProfile(path_profile)
return ""
class VariableClUpdateProfileLinuxFullname(ReadonlyVariable):
"""
Имя системы в профиле
"""
def init(self):
self.label = _("Distribution name")
datavars = "cl_update_profile_datavars"
def get(self):
dv = self.Get(self.datavars)
if dv:
try:
subname = dv.Get('os_linux_subname')
linuxname = dv.Get('os_linux_name')
if subname:
return "%s %s" % (linuxname, subname)
return linuxname
except DataVarsError:
raise VariableError("Wrong Calculate Linux profile")
return ""
class VariableClUpdateProfileDependData(ReadonlyTableVariable):
"""
Зависимые репозитории
"""
source = ['cl_update_profile_depend_name',
'cl_update_profile_depend_url']
datavars = "cl_update_profile_datavars"
def init(self):
self.label = _("Used repositories")
@staticmethod
def url_like(url1, url2):
match1 = VariableClUpdateProfileUrl.re_url.search(url1)
match2 = VariableClUpdateProfileUrl.re_url.search(url2)
if match1 and match2:
return match1.group(2).lower() == match2.group(2).lower()
return False
def get(self, hr=HumanReadable.No):
dv = self.Get(self.datavars)
if dv:
if hr == HumanReadable.Yes:
return reversed(list(zip([x.capitalize() for x in dv.Get('cl_update_rep_name')],
dv.Get('cl_update_rep_url'))))
else:
return reversed(list(zip(dv.Get('cl_update_rep_name'),
dv.Get('cl_update_rep_url'))))
return ""
setValue = Variable.setValue
class VariableClUpdateTemplatesLocate(Variable):
"""
Выбранные типы хранилищ шаблонов
"""
type = "choice-list"
element = "selecttable"
opt = ["-T", "--templates"]
metavalue = "TEMPLATES"
untrusted = True
check_after = ['cl_update_profile_system']
profile_datevars = "update.cl_update_profile_datavars"
descriptionMap = {'overlay': _('Overlay templates'),
'local': _('Local templates'),
'calculate': _("Calculate overlay templates"),
'distros': _('Distribution templates'),
'distro': _('Distribution templates'),
'remote': _('Remote templates'),
'clt': _('clt templates')}
def init(self):
self.label = _("Templates location")
self.help = _("select the location for templates %s") \
% ",".join(self.get())
def get(self):
dv = self.Get(self.profile_datevars)
if dv:
return dv.Get('cl_template_location') + ['clt']
else:
return self.Get('cl_templates_locate')
def choice(self):
descr = lambda x: self.descriptionMap.get(x,
_("%s overlay templates" % x))
return [(x, descr(x)) for x in self.get()]
class VariableClUpdateProfileDependName(FieldValue, ReadonlyVariable):
type = "list"
source_variable = "cl_update_profile_depend_data"
column = 0
def init(self):
self.label = _("Name")
class VariableClUpdateProfileDependUrl(FieldValue, ReadonlyVariable):
type = "list"
source_variable = "cl_update_profile_depend_data"
column = 1
def init(self):
self.label = _("URL")
class VariableClUpdateProfileCheckSyncAllowed(Variable):
path = 'cl_update_repos_storage'
def get(self):
repo_path = path.join(self.Get(self.path), self.Get('cl_update_profile_repo_name'))
if path.exists(repo_path) and not os.access(repo_path, os.W_OK):
return False
return True
class VariableClUpdateProfileRepName(ReadonlyVariable):
type = "list"
def get(self):
dv = self.Get('cl_update_profile_datavars')
if dv:
return dv.Get('cl_update_rep_name')
return []
class VariableClUpdateProfileSyncRep(ReadonlyVariable):
type = "list"
def get(self):
return list(reversed(self.Get('cl_update_profile_rep_name')))
class VariableClUpdateProfileRepUrl(ReadonlyVariable):
type = "list"
def get(self):
dv = self.Get('cl_update_profile_datavars')
if dv:
return dv.Get('cl_update_rep_url')
return []
class VariableClUpdateProfileLinuxVer(ReadonlyVariable):
"""
Имя системы в профиле
"""
def init(self):
self.label = _("System profile version")
def get(self):
dv = self.Get('cl_update_profile_datavars')
if dv:
return dv.Get('os_linux_ver')
return ""
class VariableClUpdateProfileLinuxName(ReadonlyVariable):
"""
Имя системы в профиле
"""
def get(self):
dv = self.Get('cl_update_profile_datavars')
if dv:
dv.Get('os_linux_name')
return ""
class VariableClUpdateProfileUrl(Variable):
"""
URL текущего репозитория
"""
#untrusted = True
check_after = ["cl_update_profile_branch"]
check_action = "update_profile"
value = ''
opt = ["--url"]
metavalue = "URL"
profile = "cl_profile_system"
branch = "cl_update_profile_branch"
storage = "cl_update_profile_storage"
profiles_shortname = 'cl_update_profile_shortname'
@property
def current_root(self):
return '/'
@property
def rep_names(self):
return self.Get("update.cl_update_rep_name")
def init(self):
self.label = _("Profile repository")
self.help = _("set the profile repository")
re_url = re.compile(
r"^(?:(%s)://)?((?:git@)?\w[\w\./:-]+?\w)(\.git)?$" % "|".join(
["http", "https", "git", "ssh"]))
re_shortname = re.compile('^(?:([\w\.-]+):)?([\w\.-]+)$')
@classmethod
def normalize_url(cls, url):
url = Git.normalize_url(url)
if not url:
return ''
#raise VariableError(_("Wrong repository URL"))
return url
def url_by_shortname(self, value):
match = self.re_shortname.match(value)
if not match.group(1):
template = self.Get('update.cl_update_rep_host_url')
if template:
template = template[0]
else:
template = self.Select('update.cl_update_rep_host_url',
where='update.cl_update_rep_host',
eq=match.group(1), limit=1)
if not template:
raise VariableError(_("Failed to determine the repository host"))
try:
return template % match.group(2)
except TypeError:
raise VariableError(_("Failed to determine the repository host"))
def set(self, value):
if self.re_shortname.match(value):
value = self.url_by_shortname(value)
return self.normalize_url(value)
def check(self, value):
if not value:
return True
try:
branch = self.Get(self.branch)
self.Get(self.storage).get_profiles(value, branch)
except GitError as e:
raise VariableError(str(e))
if not self.Get(self.profiles_shortname):
raise VariableError(_("Repository %s has no profiles") %
value)
# def get(self):
# try:
# if not self.value:
# return self.value
# profile = self.Get(self.profile)
# if profile:
# while (profile != self.current_root and
# ".git" not in listDirectory(profile) and
# "repo_name" not in listDirectory(
# path.join(profile,"profiles"))):
# profile = path.dirname(profile)
# if profile == self.current_root:
# return self.value
# if ".git" not in listDirectory(profile):
# repo_name = readFile(path.join(
# profile,"profiles/repo_name")).strip()
# if (repo_name in self.rep_names and
# self.Get('cl_action') == self.check_action):
# raise CriticalError(
# _("You need to update the repositories before "
# "you change the profile"))
# return Git.get_url(profile, "origin")
# except CriticalError as e:
# raise VariableError(str(e))
# except Exception as e:
# pass
# return self.value
class VariableClUpdateProfileRepoName(ReadonlyVariable):
def init(self):
self.label = _("Repository name")
storage = "cl_update_profile_storage"
url = "cl_update_profile_url"
def get(self):
rep_set = self.Get(self.storage)
url = self.Get(self.url)
if url:
rep = rep_set.get_repository(url, branch=None)
if rep:
return rep.repo_name
return ""
class VariableClUpdateProfileBranch(Variable):
"""
Текущий репозиторий
"""
untrusted = True
type = "edit"
opt = ["--branch"]
metavalue = "BRANCH"
storage = "cl_update_profile_storage"
url = "cl_update_profile_url"
def init(self):
self.label = _("Repository branch")
self.help = _("set the repository branch")
def get(self):
arch = self.Get("main.os_arch_machine")
if arch == "x86_64":
return "master"
else:
return "master"
class VariableClProfileRepository(ReadonlyVariable):
"""
Репозиторий из которого будут получены профили
"""
type = "object"
def get(self):
try:
git = self.Get('cl_update_git')
profile_dn = self.Get('cl_profile_system')
while profile_dn != '/' and ".git" not in listDirectory(profile_dn):
profile_dn = path.dirname(profile_dn)
if profile_dn == '/':
return ""
ls = LocalStorage(git, path.dirname(profile_dn))
return ProfileRepository(git, path.basename(profile_dn), ls)
except Exception:
return ""
class VariableClProfileData(ReadonlyTableVariable):
"""
Таблица данных о текущих профилях системы
"""
source = []
repository = "cl_profile_repository"
def profile_filter(self, profiles):
arch = self.Get('os_arch_machine_gentoo')
return [x for x in profiles if x.arch == arch]
def get(self, hr=HumanReadable.No):
try:
self.Get(f'cl_{self.section}_profile_url')
except:
return [[]]
if self.Get(f'cl_{self.section}_profile_url'):
rep = self.Get(self.repository)
if not rep:
return [[]]
profiles = rep.get_profiles()
if not profiles:
return [[]]
repo_name = profiles[0].repository.repo_name
else:
full_path = self.Get('cl_profile_system')
storage_path = self.Get('update.cl_update_repos_storage')
if self.section == 'builder':
full_path = pathJoin('/', self.Get('cl_builder_path'), full_path)
if storage_path in full_path:
repo_name = full_path.split(storage_path)[1].replace('/', ' ').split()[0]
profiles_desc = path.join(storage_path, repo_name, "profiles/profiles.desc")
profiles = ProfileRepository.get_local_profiles(path.join(storage_path, repo_name))
for profile in profiles:
profile.path = path.join(storage_path, repo_name, 'profiles', profile.profile)
filtered_profiles = self.profile_filter(profiles)
full_name = [x.profile for x in filtered_profiles]
profile_path = [x.path for x in filtered_profiles]
profile_arch = [x.arch for x in filtered_profiles]
short_name = simplify_profiles(full_name)
return list(zip(full_name,
short_name,
profile_path,
profile_arch))
setValue = Variable.setValue
class VariableClUpdateProfileRepository(ReadonlyVariable):
"""
Репозиторий из которого будет извлечён список профилей
"""
type = "object"
url = "cl_update_profile_url"
storage = "cl_update_profile_storage"
branch = "cl_update_profile_branch"
sync_set = "update.cl_update_profile_sync_set"
def get(self):
url = self.Get(self.url)
if not url:
return ""
try:
rep_set = self.Get(self.storage)
branch = self.Get(self.branch)
rep = rep_set.get_repository(url, branch=None)
if rep and self.Get(self.sync_set) == 'on':
rep.sync()
return rep_set.get_repository(url, branch=None) or ""
except GitError:
return ""
class VariableClUpdateProfileRepositoryName(ReadonlyVariable):
"""
Название репозитория, из которого будут извлечены профили
"""
def get(self):
rep = self.Get('cl_update_profile_repository')
if rep:
return rep.repo_name
return ""
class VariableClUpdateProfileData(VariableClProfileData):
source = ["cl_update_profile_fullname",
"cl_update_profile_shortname",
"cl_update_profile_path",
"cl_update_profile_arch"]
repository = "cl_update_profile_repository"
class VariableClUpdateProfileFullname(FieldValue, ReadonlyVariable):
"""
Полное название профиля
"""
type = "list"
source_variable = "cl_update_profile_data"
column = 0
class VariableClUpdateProfileShortname(FieldValue, ReadonlyVariable):
"""
Упрощенное название профиля
"""
type = "list"
source_variable = "cl_update_profile_data"
column = 1
class VariableClUpdateProfilePath(FieldValue, ReadonlyVariable):
"""
Путь от корня до профиля
"""
type = "list"
source_variable = "cl_update_profile_data"
column = 2
class VariableClUpdateProfileArch(FieldValue, ReadonlyVariable):
"""
Архитектура профиля
"""
type = "list"
source_variable = "cl_update_profile_data"
column = 3
class VariableClUpdateProfileSystem(Variable):
"""
Профиль системы (симлинк /etc/make.profile')
"""
type = "choice"
opt = ["cl_update_profile_system"]
untrusted = True
metavalue = "PROFILE"
profiles_path = "cl_update_profile_path"
profiles_shortname = "cl_update_profile_shortname"
profiles_fullname = "cl_update_profile_fullname"
profiles_arch = "cl_update_profile_arch"
profile = "cl_profile_system"
url = "cl_update_profile_url"
gentoo_arch = 'os_arch_machine_gentoo'
def init(self):
self.label = _("System profile")
self.help = _("set the system profile")
def check(self, profile):
if not profile:
raise VariableError(_("You must specify the profile"))
path_profile = self.Select(self.profiles_path,
where=self.profiles_shortname,
eq=profile, limit=1)
profile_fullname = self.Select(self.profiles_fullname,
where=self.profiles_shortname,
eq=profile, limit=1)
if path_profile:
dv = DataVarsUpdateProfile(path_profile)
if ":" in profile_fullname:
repo_name = profile_fullname.partition(":")[0]
else:
repo_name = ""
try:
if (not dv.Get('cl_update_rep_name') or
not dv.Get('cl_update_rep_url')):
return True
if not dv.Get('os_linux_name'):
raise VariableError("")
except (DataVarsError, VariableError) as e:
if str(e):
message = ". " + str(e)
else:
message = ""
raise VariableError(_("The selected profile is not Calculate")
+ message)
else:
raise VariableError(_("Wrong Calculate profile"))
def get(self):
try:
profile_system = self.Get(self.profile)
profile = self.Select(self.profiles_shortname,
where=self.profiles_path,
eq=profile_system, limit=1)
if profile:
return profile
except VariableError:
pass
shortname = self.Get(self.profiles_shortname)
if len(shortname) == 1:
return shortname[0]
return ""
def choice(self):
arch = self.Get(self.gentoo_arch)
profiles = list(zip(*self.Select([self.profiles_shortname,
self.profiles_fullname],
where=self.profiles_arch, eq=arch)))
if profiles:
short_name, full_name = profiles
return list(zip(short_name, full_name))
return []
class DataVarsUpdateProfile(SimpleDataVars):
"""
Упрощенная модель переменных для получения данных с удаленного профиля
"""
source = ['cl_update_rep_name',
'cl_update_rep_url',
'cl_update_rep_path',
'cl_update_rep_rev',
'cl_update_branch_name']
def __init__(self, profile, chroot_path='/', recheck=None, stable=None):
SimpleDataVars.__init__(
self,
lib_linux.VariableOsLinuxName(),
lib_linux.VariableOsLinuxShortname(),
lib_linux.VariableOsLinuxSubname(),
lib_linux.VariableOsLinuxVer(),
lib_linux.VariableClProfileSystem(),
lib_system.VariableOsArchMachine(),
env.VariableClRepositoryData(),
env.VariableClRepositoryName(),
env.VariableClRepositoryLocation(),
env.VariableClChrootPath(),
env.VariableClTemplateLocation(),
env.VariableClTemplatePath(),
env.VariableClEmergeConfig(systemRoot=chroot_path),
env.VariableClEmergeInfo(systemRoot=chroot_path),
env.VariableClFeatures(),
env.VariableClMakeProfile(),
env.VariableClEbuildPhase(),
VariableClUpdateReposStorage(section="update"),
VariableClUpdateLaymanStorage(section="update"),
VariableClUpdateReposconf(section="update"),
VariableClUpdateReposconfDir(section="update"),
VariableClUpdateRepData(section="update"),
VariableClUpdateRepPath(section="update"),
VariableClUpdateRepRev(section="update"),
VariableClUpdateBranch(section="update"),
VariableClUpdateBranchName(section="update"),
VariableClUpdateUsetagSet(section="update"),
VariableClUpdateRepName(section="update"),
VariableClUpdateRep(section="update"),
VariableClUpdateRepUrl(section="update"),
VariableClUpdateRepList(section="update"),
VariableClUpdateBinhostSet(section="update"),
VariableClUpdateBinhost(section="update"),
VariableClUpdateBinhostData(section="update"),
VariableClUpdateBinhostHost(section="update"),
VariableClUpdateBinhostRecheckSet(section="update"),
VariableClUpdateBinhostRevisions(section="update"),
VariableClUpdateBinhostTime(section="update"),
VariableClUpdateBinhostTimestamp(section="update"),
VariableClUpdateLastTimestamp(section="update"),
VariableClUpdateBinhostTimeout(section="update"),
VariableClUpdateBinhostTimestampPath(section="update"),
VariableClUpdateBinhostList(section="update"),
VariableClUpdateBinhostUnstableList(section="update"),
VariableClUpdateBinhostStableSet(section="update"),
VariableClUpdateBinhostStableOptSet(section="update"),
VariableClUpdateBinhostRevisionPath(section="update"),
VariableClUpdateRepHosting(section="update"),
VariableClUpdateBinhostSet(section="update"),
)
self['cl_profile_system'] = profile
self['cl_chroot_path'] = chroot_path
if recheck is not None:
self['cl_update_binhost_recheck_set'] = recheck
if stable is not None:
self['cl_update_binhost_stable_opt_set'] = stable
self.flIniFileFrom(profile)
def __repr__(self):
return "Profile variables"
class VariableClUpdateProfileSyncSet(Variable):
"""
Синхронизировать репозиторий перед сменой профиля
"""
type = "bool"
value = "off"
opt = ["-u", "--update-cache"]
def init(self):
self.label = _("Update the cache")
self.help = _("Update the cache")
class VariableClUpdateAutocheckSet(Variable):
"""
Выполнять/невыполнять автопроверку обновлений
"""
type = "bool"
value = "on"
opt = ["-a", "--autocheck"]
def init(self):
self.label = _("Automatically check updates")
self.help = _("automatically check updates")
class VariableClUpdateAutocheckInterval(Variable):
"""
Интервал выполнения проверки
"""
type = "choiceedit"
opt = ["-I", "--autocheck-interval"]
metavalue = "INTERVAL"
def init(self):
self.label = _("Interval for the updates checking")
self.help = _("set interval for the updates checking")
def choice(self):
return [["6h", _("every six hours")],
["12h", _("every twelve hours")],
["1d", _("daily")]]
class VariableClUpdateAutocheckScheduleSet(Variable):
"""
Запустить проверку только если со времени последнего запуска прошло
необходимое время
"""
type = "bool"
value = "off"
opt = ["--schedule"]
def init(self):
self.label = _("Consider the autocheck schedule")
self.help = _("consider the autocheck schedule")
class VariableClUpdateEmergelistSet(Variable):
"""
Вывести список пакетов в формате emerge
"""
type = "bool"
value = "off"
opt = ["-e", "--emergelist"]
def init(self):
self.label = _("Emerge-like packages list")
self.help = _("display the packages list in emerge format")
class VariableClUpdateKernelVersion(ReadonlyVariable):
"""
Текущая версия ядра
"""
def get(self):
return process('/bin/uname', '-r').read().strip()
class VariableClUpdateKernelSrcPath(ReadonlyVariable):
"""
Каталог содержащий исходный код текущего ядра
"""
def get(self):
kernel_ver = self.Get('cl_update_kernel_version')
for template_path in ("/lib/modules/%s/build",
"/usr/src/linux-%s"):
src_path = template_path % kernel_ver
if path.exists(src_path):
if path.islink(src_path):
return os.readlink(src_path)
else:
return src_path
else:
return ""
class VariableClUpdateKernelPkg(ReadonlyVariable):
"""
Пакет текущего ядра
"""
def get(self):
src_path = self.Get('cl_update_kernel_src_path')
if src_path:
qfile = process('/usr/bin/qfile', '-vqC', src_path)
if qfile.success():
return qfile.read().strip()
return ""
class VariableClUpdateLinesLimit(Variable):
"""
Количество выводимых строк при ошибке
"""
type = "int"
value = "30"
class VariableClUpdateSkipSetupSet(Variable):
"""
Пропустить выполнение cl-setup-system в cl-update-profile
"""
type = "bool"
value = "off"
opt = ["--skip-setup-system"]
def init(self):
self.label = _("Skip the system setup")
self.help = _("skip the system setup")
class VariableClUpdateCleanpkgSet(Variable):
"""
Пропустить выполнение cl-setup-system в cl-update-profile
"""
type = "bool"
value = "off"
opt = ["--clean-pkg"]
def init(self):
self.label = _("Clean obsolete programs archives")
self.help = _("clean obsolete programs archives")
class VariableClUpdateOutdatedKernelPath(Variable):
"""
Файл-флаг наличия устаревшего, неудаленного ядра
"""
value = "/var/lib/calculate/calculate-update/outdated_kernel"
class VariableClUpdateSkipRbSet(Variable):
"""
Пропусить revdep-rebuild
"""
type = "bool"
value = "off"
opt = ["-R", "--skip-revdep-rebuild"]
def init(self):
self.label = _("Skip reverse dependencies check")
self.help = _("skip reverse dependencies check")
class VariableClUpdateRevdepRebuildSet(Variable):
"""
Запустить revdep-rebuild
"""
type = "bool"
value = "off"
opt = ["-R", "--revdep-rebuild"]
def init(self):
self.label = _("Run reverse dependencies check")
self.help = _("run reverse dependencies check")
class VariableClUpdateOutdatedKernelSet(ReadonlyVariable):
"""
Есть наличие устаревшего ядра
"""
type = "bool"
def get(self):
ui = UpdateInfo(self.parent)
return "on" if ui.outdated_kernel else "off"
class VariableClUpdateAvailableSet(ReadonlyVariable):
"""
Есть обновления
"""
type = "bool"
def get(self):
ui = UpdateInfo(self.parent)
return "on" if ui.need_update() else "off"
class VariableClUpdateBinhostList(Variable):
"""
Список хостов с бинарными обновлениями
"""
type = "list"
value = ["https://mirror.calculate-linux.org/"]
# value = ["ftp://ftp.calculate-linux.ru/calculate"]
class VariableClUpdateBinhostUnstableList(Variable):
"""
Список хостов с бинарными обновлениями
"""
type = "list"
value = ["https://testing.calculate-linux.org/"]
# value = ["ftp://ftp.calculate-linux.ru/testing"]
class VariableClUpdateBinhostMigrationList(Variable):
"""
Список хостов с бинарными обновлениями для стабильной миграции
"""
type = "list"
value = ["ftp://ftp.spb.calculate.ru/pub/calculate/migration/"]
class VariableClUpdateBinhostBase(Variable):
"""
Базовый бинарный хост, используемый для установки базовых пакетов
в небинарных сборочных дистрибутивах
"""
value = ""
class VariableClUpdateBinhostBaseSet(Variable):
"""
Использовать или нет базовый бинарный хост
"""
type = Variable.Types.Boolean
value = Variable.Off
class VariableClUpdateBinhostStableSet(Variable):
"""
Удлять лишние файлы из репозиториев (например созданные пользователем)
"""
type = "bool"
value = "on"
class VariableClUpdateBinhostStableOptSet(Variable):
"""
Удлять лишние файлы из репозиториев (например созданные пользователем)
"""
type = "bool"
value = "on"
opt = ["--stable"]
def init(self):
self.label = _("Use only stable updates")
self.help = _("use only stable updates")
def get(self):
return self.Get('cl_update_binhost_stable_set')
class VariableClUpdateBinhostChoice(Variable):
type = "choiceedit"
value = ''
opt = ["-m", "--mirror-update"]
metavalue = "MIRROR_URL"
def init(self):
self.label = _("Packages mirror")
self.help = _("Url to download binary packages from")
def get(self):
if not self.value:
if self.GetBool("cl_update_binhost_set") and self.Get("cl_update_binhost"):
return self.Get("cl_update_binhost")
return self.value
def choice(self):
if self.GetBool("cl_update_binhost_set") and self.Get("cl_update_binhost"):
if self.Get("cl_update_binhost") not in self.Get("cl_update_binhost_list"):
return ['auto'] + [self.Get("cl_update_binhost")] + self.Get("cl_update_binhost_list")
return ['auto'] + self.Get("cl_update_binhost_list")
class VariableClUpdateBinhost(Variable):
"""
Хост с бинарными обновлениями
"""
value = ''
class VariableClUpdateBinhostRevisionPath(Variable):
"""
Путь до revisions файлов
"""
type = "list"
value = [
"grp/ini.env"
]
class VariableClUpdateBinhostTimestampPath(Variable):
"""
Путь до файла timestamp
"""
value = "timestamp"
class VariableClUpdateBinhostTimeout(Variable):
"""
Таймаут на проверку одного binhost
"""
type = "int"
value = "5"
class VariableClUpdateCheckRepSet(Variable):
"""
Удлять лишние файлы из репозиториев (например созданные пользователем)
"""
type = "bool"
value = "off"
opt = ["--check-repos", "-C"]
def init(self):
self.label = _("Check the repositories integrity")
self.help = _("check and fix the repositories integrity")
class VariableClUpdateForceFixSet(Variable):
"""
Удлять лишние файлы из репозиториев (например созданные пользователем)
"""
type = "bool"
value = "off"
opt = ["--force-fix"]
def init(self):
self.label = _("Force fix the settings")
self.help = _("force fix the settings")
class VariableClUpdateOnedepthSet(Variable):
"""
Удлять лишние файлы из репозиториев (например созданные пользователем)
"""
type = "bool3"
value = 'auto'
max_interval = 'cl_update_onedepth_interval'
check_after = 'cl_update_onedepth_interval'
opt = ["--one-depth", "-1"]
def init(self):
self.label = _("Clear the history of repositories")
self.help = _("clear the history of repositories")
def get_oldest_history(self, rep):
try:
with open(f'{rep}/.git/logs/HEAD') as inf:
oldest_timestamp = int(inf.readline().split()[4])
date = datetime.datetime.fromtimestamp(oldest_timestamp)
return datetime.datetime.now() - date
except OSError:
return 0
def set(self, value):
try:
if value == 'auto' or value is None:
rep_path = self.Get('cl_update_rep_path')
for rep in rep_path:
if self.get_oldest_history(rep).days > int(self.Get(self.max_interval)):
self.value = 'on'
return 'on'
self.value = 'off'
return 'off'
except:
return 'off'
self.value = value
return self.value
def get(self):
if self.value == 'auto' or self.value is None:
return self.set(self.value)
return self.value
class VariableClUpdateOnedepthInterval(Variable):
default_value = 0
def get(self):
if not self.value:
return self.default_value
return self.value
class VariableClUpdateEixRepositories(ReadonlyVariable):
"""
Отображаемый список обновляемых репозиториев eix
"""
def get(self):
return ", ".join(
x.capitalize() for x in reversed(list(chain(
self.Get('update.cl_update_rep_name'),
self.Get('update.cl_update_other_rep_name')))))
class VariableClUpdatePortageBinhost(ReadonlyVariable):
"""
Прописываемый в /etc/portage/make.conf/binhost репозиторий
"""
value_format = "{update.cl_update_binhost}/grp/{os_arch_machine}"
class VariableClUpdatePackageCache(ReadonlyVariable):
"""
Путь кэшированного Packages
"""
def get(self):
cache_path = "var/cache/edb"
chroot_path = self.Get('cl_chroot_path')
base_url = self.Get('cl_update_portage_binhost')
if not base_url:
return ""
parsed_url = urlparse(base_url)
host = parsed_url.netloc
port = parsed_url.port
user = None
passwd = None
user_passwd = ""
if "@" in host:
user, host = host.split("@", 1)
user_passwd = user + "@"
if ":" in user:
user, passwd = user.split(":", 1)
if port is not None:
port_str = ":%s" % (port,)
if host.endswith(port_str):
host = host[:-len(port_str)]
return os.path.join(chroot_path, cache_path, "binhost",
host, parsed_url.path.lstrip("/"),
"Packages")
class VariableClUpdatePackageCacheSign(ReadonlyVariable):
"""
Файл с подписью Packages находящийся в кэше
"""
value_format = "{update.cl_update_package_cache}.asc"
class VariableClUpdatePackageCacheSet(Variable):
"""
Необходимость обновить Packages
"""
type = "bool"
value = ''
def get(self):
if self.value:
return self.value
packages_fn = self.Get('cl_update_package_cache')
gpg = self.Get('cl_update_gpg')
packages_asc_fn = self.Get('cl_update_package_cache_sign')
if (not path.exists(packages_fn) or
gpg and not path.exists(packages_asc_fn)):
return "on"
pi = PackagesIndex(readFile(packages_fn))
try:
ttl = int(pi.get("TTL", "0"))
if ttl > HOURS:
return "off"
except ValueError:
return "on"
return "on"
class VariableClUpdateWithBdepsSet(Variable):
"""
Добавление --with-bdeps при вызове emerge
"""
type = Variable.Types.Tristate
value = Variable.Auto
class VariableClUpdateWithBdepsOptSet(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_update_with_bdeps_set')
class VariableClUpdateForceDepcleanSet(Variable):
"""
Принудительный вызов depclean
"""
type = Variable.Types.Boolean
value = Variable.Off
class VariableClUpdateUsetagSet(Variable):
"""
Использовать тэги при синхронизации репозиториев
"""
type = Variable.Types.Boolean
value = Variable.On
class VariableClUpdateSshkeyPath(Variable):
"""
Путь до ssh ключа
"""
value = "/var/lib/calculate/id_rsa"
class VariableClUpdateBadSignSet(Variable):
"""
Ошибка проверки подписи Packages
"""
type = Variable.Types.Boolean
value = Variable.Off
class VariableClUpdateGpgKeys(Variable):
"""
Список ключей для проверки подписи Packages
"""
type = Variable.Types.List
value = ["/usr/share/openpgp-keys/calculate-release.asc"]
class VariableClUpdateGpg(ReadonlyVariable):
"""
Объект проверки подписи
"""
type = Variable.Types.Object
value = ""
def close(self):
val = self.Get()
if val:
val.close()
class VariableClUpdateUseDowngradeSet(ReadonlyVariable):
"""
Можно использовать downgrade binhost
"""
type = Variable.Types.Boolean
def get(self):
"""
Выполняется переход с unstable ветки на stable
"""
return (self.GetBool("cl_update_binhost_stable_opt_set") and
not self.GetBool("cl_update_binhost_stable_opt_set"))
class VariableClUpdatePortageBinhostPath(Variable):
"""
Файл в котором хранится параметр portage PORTAGE_BINHOST
"""
value = "/etc/portage/make.conf/binhost"
class VariableClUpdateLevel(Variable):
"""
Текущий уровень обновления
Используется для выбора зеркала
"""
type = "int"
value = ""
class VariableClUpdateUseMigrationHost(Variable):
"""
Использовать хост миграции
"""
type = "bool"
value = "off"
class VariableClUpdateIgnoreLevel(Variable):
"""
Проверять уровень обновления
при соединении с binhost
"""
type = "bool"
opt = ["--ignore-level"]
value = "off"
def init(self):
self.label = _("Forcefully ignore update level and migration system")
self.help = _("forcefully ignore update level and migration system")
class VariableClUpdateForceLevel(Variable):
"""
Принудительное обновление до определенного уровня
"""
type = "int"
opt = ["--force-level"]
metavalue = "LEVEL"
def init(self):
self.label = _("Force update to particular level")
self.help = _("force update to particular level")
class VariableClUpdateForceLevelMode(Variable):
"""
Режим принудительного обновления до определенного уровня
"""
type = "bool"
value = "off"
class VariableClUpdateSavedTag(Variable):
"""
Сохраняемый тэг репозитория calculate
"""
type = "int"
class VariableClUpdateSaveCurBinhost(Variable):
value = " "
class VariableClUpdateWorldHashSet(Variable):
"""
Хэш world для проверки изменений
"""
type = Variable.Types.Boolean
value = ""
def get(self):
if self.value:
return self.value
ini = SystemIni(self.parent)
old_world_state_hash = ini.getVar('system', 'world_hash') or ""
#меняем путь сборки хэша на world файл
ps = PortageState()
ps.paths = ['/var/lib/portage/world']
new_world_state_hash = ps.get_state()
return "on" if new_world_state_hash != old_world_state_hash else "off"