-
Notifications
You must be signed in to change notification settings - Fork 134
/
groups_manager.py
316 lines (283 loc) · 13.2 KB
/
groups_manager.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
# groups_manager.py
# DNF plugin for managing comps groups metadata files
#
# Copyright (C) 2020 Red Hat, Inc.
#
# This copyrighted material is made available to anyone wishing to use,
# modify, copy, or redistribute it subject to the terms and conditions of
# the GNU General Public License v.2, or (at your option) any later version.
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY expressed or implied, including the implied warranties of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General
# Public License for more details. You should have received a copy of the
# GNU General Public License along with this program; if not, write to the
# Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
# 02110-1301, USA. Any Red Hat trademarks that are incorporated in the
# source code or documentation are not subject to the GNU General Public
# License and may only be used or replicated with the express permission of
# Red Hat, Inc.
#
from __future__ import absolute_import
from __future__ import unicode_literals
import argparse
import gzip
import libcomps
import os
import re
import shutil
import tempfile
from dnfpluginscore import _, logger
import dnf
import dnf.cli
RE_GROUP_ID_VALID = '-a-z0-9_.:'
RE_GROUP_ID = re.compile(r'^[{}]+$'.format(RE_GROUP_ID_VALID))
RE_LANG = re.compile(r'^[-a-zA-Z0-9_.@]+$')
COMPS_XML_OPTIONS = {
'default_explicit': True,
'uservisible_explicit': True,
'empty_groups': True}
def group_id_type(value):
'''group id validator'''
if not RE_GROUP_ID.match(value):
raise argparse.ArgumentTypeError(_('Invalid group id'))
return value
def translation_type(value):
'''translated texts validator'''
data = value.split(':', 2)
if len(data) != 2:
raise argparse.ArgumentTypeError(
_("Invalid translated data, should be in form 'lang:text'"))
lang, text = data
if not RE_LANG.match(lang):
raise argparse.ArgumentTypeError(_('Invalid/empty language for translated data'))
return lang, text
def text_to_id(text):
'''generate group id based on its name'''
group_id = text.lower()
group_id = re.sub('[^{}]'.format(RE_GROUP_ID_VALID), '', group_id)
if not group_id:
raise dnf.cli.CliError(
_("Can't generate group id from '{}'. Please specify group id using --id.").format(
text))
return group_id
@dnf.plugin.register_command
class GroupsManagerCommand(dnf.cli.Command):
aliases = ('groups-manager',)
summary = _('create and edit groups metadata file')
def __init__(self, cli):
super(GroupsManagerCommand, self).__init__(cli)
self.comps = libcomps.Comps()
@staticmethod
def set_argparser(parser):
# input / output options
parser.add_argument('--load', action='append', default=[],
metavar='COMPS.XML',
help=_('load groups metadata from file'))
parser.add_argument('--save', action='append', default=[],
metavar='COMPS.XML',
help=_('save groups metadata to file'))
parser.add_argument('--merge', metavar='COMPS.XML',
help=_('load and save groups metadata to file'))
parser.add_argument('--print', action='store_true', default=False,
help=_('print the result metadata to stdout'))
# group options
parser.add_argument('--id', type=group_id_type,
help=_('group id'))
parser.add_argument('-n', '--name', help=_('group name'))
parser.add_argument('--description',
help=_('group description'))
parser.add_argument('--display-order', type=int,
help=_('group display order'))
parser.add_argument('--translated-name', action='append', default=[],
metavar='LANG:TEXT', type=translation_type,
help=_('translated name for the group'))
parser.add_argument('--translated-description', action='append', default=[],
metavar='LANG:TEXT', type=translation_type,
help=_('translated description for the group'))
visible = parser.add_mutually_exclusive_group()
visible.add_argument('--user-visible', dest='user_visible', action='store_true',
default=None,
help=_('make the group user visible (default)'))
visible.add_argument('--not-user-visible', dest='user_visible', action='store_false',
default=None,
help=_('make the group user invisible'))
# package list options
section = parser.add_mutually_exclusive_group()
section.add_argument('--mandatory', action='store_true',
help=_('add packages to the mandatory section'))
section.add_argument('--optional', action='store_true',
help=_('add packages to the optional section'))
section.add_argument('--remove', action='store_true', default=False,
help=_('remove packages from the group instead of adding them'))
parser.add_argument('--dependencies', action='store_true',
help=_('include also direct dependencies for packages'))
parser.add_argument("packages", nargs='*', metavar='PACKAGE',
help=_('package specification'))
def configure(self):
demands = self.cli.demands
if self.opts.packages:
demands.sack_activation = True
demands.available_repos = True
demands.load_system_repo = False
# handle --merge option (shortcut to --load and --save the same file)
if self.opts.merge:
self.opts.load.insert(0, self.opts.merge)
self.opts.save.append(self.opts.merge)
# check that group is specified when editing is attempted
if (self.opts.description
or self.opts.display_order
or self.opts.translated_name
or self.opts.translated_description
or self.opts.user_visible is not None
or self.opts.packages):
if not self.opts.id and not self.opts.name:
raise dnf.cli.CliError(
_("Can't edit group without specifying it (use --id or --name)"))
def load_input_files(self):
"""
Loads all input xml files.
Returns True if at least one file was successfuly loaded
"""
for file_name in self.opts.load:
file_comps = libcomps.Comps()
try:
if file_name.endswith('.gz'):
# libcomps does not support gzipped files - decompress to temporary
# location
with gzip.open(file_name) as gz_file:
temp_file = tempfile.NamedTemporaryFile(delete=False)
try:
shutil.copyfileobj(gz_file, temp_file)
# close temp_file to ensure the content is flushed to disk
temp_file.close()
file_comps.fromxml_f(temp_file.name)
finally:
os.unlink(temp_file.name)
else:
file_comps.fromxml_f(file_name)
except (IOError, OSError, libcomps.ParserError) as err:
# gzip module raises OSError on reading from malformed gz file
# get_last_errors() output often contains duplicit lines, remove them
seen = set()
for error in file_comps.get_last_errors():
if error in seen:
continue
logger.error(error.strip())
seen.add(error)
raise dnf.exceptions.Error(
_("Can't load file \"{}\": {}").format(file_name, err))
else:
self.comps += file_comps
def save_output_files(self):
for file_name in self.opts.save:
try:
# xml_f returns a list of errors / log entries
errors = self.comps.xml_f(file_name, xml_options=COMPS_XML_OPTIONS)
except libcomps.XMLGenError as err:
errors = [err]
if errors:
# xml_f() method could return more than one error. In this case
# raise the latest of them and log the others.
for err in errors[:-1]:
logger.error(err.strip())
raise dnf.exceptions.Error(_("Can't save file \"{}\": {}").format(
file_name, errors[-1].strip()))
def find_group(self, group_id, name):
'''
Try to find group according to command line parameters - first by id
then by name.
'''
group = None
if group_id:
for grp in self.comps.groups:
if grp.id == group_id:
group = grp
break
if group is None and name:
for grp in self.comps.groups:
if grp.name == name:
group = grp
break
return group
def edit_group(self, group):
'''
Set attributes and package lists for selected group
'''
def langlist_to_strdict(lst):
str_dict = libcomps.StrDict()
for lang, text in lst:
str_dict[lang] = text
return str_dict
# set group attributes
if self.opts.name:
group.name = self.opts.name
if self.opts.description:
group.desc = self.opts.description
if self.opts.display_order:
group.display_order = self.opts.display_order
if self.opts.user_visible is not None:
group.uservisible = self.opts.user_visible
if self.opts.translated_name:
group.name_by_lang = langlist_to_strdict(self.opts.translated_name)
if self.opts.translated_description:
group.desc_by_lang = langlist_to_strdict(self.opts.translated_description)
# edit packages list
if self.opts.packages:
# find packages according to specifications from command line
packages = set()
for pkg_spec in self.opts.packages:
subj = dnf.subject.Subject(pkg_spec)
q = subj.get_best_query(self.base.sack, with_nevra=True,
with_provides=False, with_filenames=False).latest()
if not q:
logger.warning(_("No match for argument: {}").format(pkg_spec))
continue
packages.update(q)
if self.opts.dependencies:
# add packages that provide requirements
requirements = set()
for pkg in packages:
requirements.update(pkg.requires)
packages.update(self.base.sack.query().filterm(provides=requirements))
pkg_names = {pkg.name for pkg in packages}
if self.opts.remove:
for pkg_name in pkg_names:
for pkg in group.packages_match(name=pkg_name,
type=libcomps.PACKAGE_TYPE_UNKNOWN):
group.packages.remove(pkg)
else:
if self.opts.mandatory:
pkg_type = libcomps.PACKAGE_TYPE_MANDATORY
elif self.opts.optional:
pkg_type = libcomps.PACKAGE_TYPE_OPTIONAL
else:
pkg_type = libcomps.PACKAGE_TYPE_DEFAULT
for pkg_name in sorted(pkg_names):
if not group.packages_match(name=pkg_name, type=pkg_type):
group.packages.append(libcomps.Package(name=pkg_name, type=pkg_type))
def run(self):
self.load_input_files()
if self.opts.id or self.opts.name:
# we are adding / editing a group
group = self.find_group(group_id=self.opts.id, name=self.opts.name)
if group is None:
# create a new group
if self.opts.remove:
raise dnf.exceptions.Error(_("Can't remove packages from non-existent group"))
group = libcomps.Group()
if self.opts.id:
group.id = self.opts.id
group.name = self.opts.id
elif self.opts.name:
group_id = text_to_id(self.opts.name)
if self.find_group(group_id=group_id, name=None):
raise dnf.cli.CliError(
_("Group id '{}' generated from '{}' is duplicit. "
"Please specify group id using --id.").format(
group_id, self.opts.name))
group.id = group_id
self.comps.groups.append(group)
self.edit_group(group)
self.save_output_files()
if self.opts.print or (not self.opts.save):
print(self.comps.xml_str(xml_options=COMPS_XML_OPTIONS))