History.get_user_requests() - Code Metrics - Inspection of "Merge remote-tracking branch 'origin/4.5.x' into m..." - conda/conda - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 1eecb7...7e6e06 )

by Kale

created 2018-07-02 19:05 UTC

History.get_user_requests() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	20

Importance

Changes	1
Bugs	1	Features	0

Metric	Value
cc	4
dl	0
loc	24
rs	9.304
c	1
b	1
f	0
ccs	0
cts	15
cp	0
crap	20

# -*- coding: utf-8 -*-
# Copyright (C) 2012 Anaconda, Inc
# SPDX-License-Identifier: BSD-3-Clause
from __future__ import absolute_import, division, print_function, unicode_literals

from ast import literal_eval
import errno
import logging
from operator import itemgetter
import os
from os.path import isdir, isfile, join
import re
import sys
import time
import warnings

from .base.constants import DEFAULTS_CHANNEL_NAME
from .common.compat import ensure_text_type, iteritems, open, text_type
from .core.prefix_data import PrefixData
from .exceptions import CondaFileIOError, CondaHistoryError
from .gateways.disk.update import touch
from .models.dist import dist_str_to_quad
from .resolve import MatchSpec

try:
    from cytoolz.itertoolz import groupby
except ImportError:  # pragma: no cover
    from ._vendor.toolz.itertoolz import groupby  # NOQA


log = logging.getLogger(__name__)


class CondaHistoryWarning(Warning):
    pass


def write_head(fo):
    fo.write("==> %s <==\n" % time.strftime('%Y-%m-%d %H:%M:%S'))
    fo.write("# cmd: %s\n" % (' '.join(ensure_text_type(s) for s in sys.argv)))


def is_diff(content):
    return any(s.startswith(('-', '+')) for s in content)


def pretty_diff(diff):
    added = {}
    removed = {}
    for s in diff:
        fn = s[1:]
        name, version, _, channel = dist_str_to_quad(fn)
        if channel != DEFAULTS_CHANNEL_NAME:
            version += ' (%s)' % channel
        if s.startswith('-'):
            removed[name.lower()] = version
        elif s.startswith('+'):
            added[name.lower()] = version
    changed = set(added) & set(removed)
    for name in sorted(changed):
        yield ' %s  {%s -> %s}' % (name, removed[name], added[name])
    for name in sorted(set(removed) - changed):
        yield '-%s-%s' % (name, removed[name])
    for name in sorted(set(added) - changed):
        yield '+%s-%s' % (name, added[name])


def pretty_content(content):
    if is_diff(content):
        return pretty_diff(content)
    else:
        return iter(sorted(content))


class History(object):

    com_pat = re.compile(r'#\s*cmd:\s*(.+)')
    spec_pat = re.compile(r'#\s*(\w+)\s*specs:\s*(.+)?')

    def __init__(self, prefix):
        self.prefix = prefix
        self.meta_dir = join(prefix, 'conda-meta')
        self.path = join(self.meta_dir, 'history')

    def __enter__(self):
        self.init_log_file()
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.update()

    def init_log_file(self):
        touch(self.path, True)

    def file_is_empty(self):
        return os.stat(self.path).st_size == 0

    def update(self):
        """
        update the history file (creating a new one if necessary)
        """
        try:
            try:
                last = set(self.get_state())
            except CondaHistoryError as e:
                warnings.warn("Error in %s: %s" % (self.path, e),
                              CondaHistoryWarning)
                return
            pd = PrefixData(self.prefix)
            curr = set(prefix_rec.dist_str() for prefix_rec in pd.iter_records())
            self.write_changes(last, curr)
        except IOError as e:
            if e.errno == errno.EACCES:
                log.debug("Can't write the history file")
            else:
                raise CondaFileIOError(self.path, "Can't write the history file %s" % e)

    def parse(self):
        """
        parse the history file and return a list of
        tuples(datetime strings, set of distributions/diffs, comments)
        """
        res = []
        if not isfile(self.path):
            return res
        sep_pat = re.compile(r'==>\s*(.+?)\s*<==')
        with open(self.path) as f:
            lines = f.read().splitlines()
        for line in lines:
            line = line.strip()
            if not line:
                continue
            m = sep_pat.match(line)
            if m:
                res.append((m.group(1), set(), []))
            elif line.startswith('#'):
                res[-1][2].append(line)
            elif len(res) > 0:
                res[-1][1].add(line)
        return res

    @staticmethod
    def _parse_old_format_specs_string(specs_string):
        """
        Parse specifications string that use conda<4.5 syntax.

        Examples
        --------
          - "param >=1.5.1,<2.0'"
          - "python>=3.5.1,jupyter >=1.0.0,<2.0,matplotlib >=1.5.1,<2.0"
        """
        specs = []
        for spec in specs_string.split(','):
            # See https://github.com/conda/conda/issues/6691
            if spec[0].isalpha():
                # A valid spec starts with a letter since it is a package name
                specs.append(spec)
            else:
                # Otherwise it is a condition and has to be appended to the
                # last valid spec on the specs list
                specs[-1] = ','.join([specs[-1], spec])

        return specs

    @classmethod
    def _parse_comment_line(cls, line):
        """
        Parse comment lines in the history file.

        These lines can be of command type or action type.

        Examples
        --------
          - "# cmd: /scratch/mc3/bin/conda install -c conda-forge param>=1.5.1,<2.0"
          - "# install specs: python>=3.5.1,jupyter >=1.0.0,<2.0,matplotlib >=1.5.1,<2.0"
        """
        item = {}
        m = cls.com_pat.match(line)
        if m:
            argv = m.group(1).split()
            if argv[0].endswith('conda'):
                argv[0] = 'conda'
            item['cmd'] = argv

        m = cls.spec_pat.match(line)
        if m:
            action, specs_string = m.groups()
            specs_string = specs_string or ""
            item['action'] = action

            if specs_string.startswith('['):
                specs = literal_eval(specs_string)
            elif '[' not in specs_string:
                specs = History._parse_old_format_specs_string(specs_string)

            specs = [spec for spec in specs if spec and not spec.endswith('@')]

            if specs and action in ('update', 'install', 'create'):
                item['update_specs'] = item['specs'] = specs
            elif specs and action in ('remove', 'uninstall'):
                item['remove_specs'] = item['specs'] = specs

        return item

    def get_user_requests(self):
        """
        return a list of user requested items.  Each item is a dict with the
        following keys:
        'date': the date and time running the command
        'cmd': a list of argv of the actual command which was run
        'action': install/remove/update
        'specs': the specs being used
        """
        res = []
        for dt, unused_cont, comments in self.parse():
            item = {'date': dt}
            for line in comments:
                comment_items = self._parse_comment_line(line)
                item.update(comment_items)

            if 'cmd' in item:
                res.append(item)

            dists = groupby(itemgetter(0), unused_cont)
            item['unlink_dists'] = dists.get('-', ())
            item['link_dists'] = dists.get('+', ())

        return res

    def get_requested_specs_map(self):
        # keys are package names and values are specs
        spec_map = {}
        for request in self.get_user_requests():
            remove_specs = (MatchSpec(spec) for spec in request.get('remove_specs', ()))
            for spec in remove_specs:
                spec_map.pop(spec.name, None)
            update_specs = (MatchSpec(spec) for spec in request.get('update_specs', ()))
            spec_map.update(((s.name, s) for s in update_specs))

        # Conda hasn't always been good about recording when specs have been removed from
        # environments.  If the package isn't installed in the current environment, then we
        # shouldn't try to force it here.
        prefix_recs = tuple(PrefixData(self.prefix).iter_records())
        return dict((name, spec) for name, spec in iteritems(spec_map)
                    if any(spec.match(dist) for dist in prefix_recs))

    def construct_states(self):
        """
        return a list of tuples(datetime strings, set of distributions)
        """
        res = []
        cur = set([])
        for dt, cont, unused_com in self.parse():
            if not is_diff(cont):
                cur = cont
            else:
                for s in cont:
                    if s.startswith('-'):
                        cur.discard(s[1:])
                    elif s.startswith('+'):
                        cur.add(s[1:])
                    else:
                        raise CondaHistoryError('Did not expect: %s' % s)
            res.append((dt, cur.copy()))
        return res

    def get_state(self, rev=-1):
        """
        return the state, i.e. the set of distributions, for a given revision,
        defaults to latest (which is the same as the current state when
        the log file is up-to-date)

        Returns a list of dist_strs
        """
        states = self.construct_states()
        if not states:
            return set([])
        times, pkgs = zip(*states)
        return pkgs[rev]

    def print_log(self):
        for i, (date, content, unused_com) in enumerate(self.parse()):
            print('%s  (rev %d)' % (date, i))
            for line in pretty_content(content):
                print('    %s' % line)
            print()

    def object_log(self):
        result = []
        for i, (date, content, unused_com) in enumerate(self.parse()):
            # Based on Mateusz's code; provides more details about the
            # history event
            event = {
                'date': date,
                'rev': i,
                'install': [],
                'remove': [],
                'upgrade': [],
                'downgrade': []
            }
            added = {}
            removed = {}
            if is_diff(content):
                for pkg in content:
                    name, version, build, channel = dist_str_to_quad(pkg[1:])
                    if pkg.startswith('+'):
                        added[name.lower()] = (version, build, channel)
                    elif pkg.startswith('-'):
                        removed[name.lower()] = (version, build, channel)

                changed = set(added) & set(removed)
                for name in sorted(changed):
                    old = removed[name]
                    new = added[name]
                    details = {
                        'old': '-'.join((name,) + old),
                        'new': '-'.join((name,) + new)
                    }

                    if new > old:
                        event['upgrade'].append(details)
                    else:
                        event['downgrade'].append(details)

                for name in sorted(set(removed) - changed):
                    event['remove'].append('-'.join((name,) + removed[name]))

                for name in sorted(set(added) - changed):
                    event['install'].append('-'.join((name,) + added[name]))
            else:
                for pkg in sorted(content):
                    event['install'].append(pkg)
            result.append(event)
        return result

    def write_changes(self, last_state, current_state):
        if not isdir(self.meta_dir):
            os.makedirs(self.meta_dir)
        with open(self.path, 'a') as fo:
            write_head(fo)
            for fn in sorted(last_state - current_state):
                fo.write('-%s\n' % fn)
            for fn in sorted(current_state - last_state):
                fo.write('+%s\n' % fn)

    def write_specs(self, remove_specs=(), update_specs=()):
        remove_specs = [text_type(MatchSpec(s)) for s in remove_specs]
        update_specs = [text_type(MatchSpec(s)) for s in update_specs]
        if update_specs or remove_specs:
            with open(self.path, 'a') as fh:
                if remove_specs:
                    fh.write("# remove specs: %s\n" % remove_specs)
                if update_specs:
                    fh.write("# update specs: %s\n" % update_specs)


if __name__ == '__main__':
    from pprint import pprint
    # Don't use in context manager mode---it augments the history every time
    h = History(sys.prefix)
    pprint(h.get_user_requests())
    print(h.get_requested_specs_map())


1	7	# -- coding: utf-8 --
2		# Copyright (C) 2012 Anaconda, Inc
3	7	# SPDX-License-Identifier: BSD-3-Clause
4	7	from __future__ import absolute_import, division, print_function, unicode_literals
5	7
6	7	from ast import literal_eval
7	7	import errno
8	7	import logging
9	7	from operator import itemgetter
10	7	import os
11		from os.path import isdir, isfile, join
12	7	import re
13	7	import sys
14		import time
15	7	import warnings
16
17		from .base.constants import DEFAULTS_CHANNEL_NAME
18	7	from .common.compat import ensure_text_type, iteritems, open, text_type
19	7	from .core.prefix_data import PrefixData
20		from .exceptions import CondaFileIOError, CondaHistoryError
21		from .gateways.disk.update import touch
22	7	from .models.dist import dist_str_to_quad
23	7	from .resolve import MatchSpec
24	7
25		try:
26		from cytoolz.itertoolz import groupby
27	7	except ImportError: # pragma: no cover
28	7	from ._vendor.toolz.itertoolz import groupby # NOQA
29
30
31	7	log = logging.getLogger(__name__)
32
33
34		class CondaHistoryWarning(Warning):
35		pass
36
37
38		def write_head(fo):
39		fo.write("==> %s <==\n" % time.strftime('%Y-%m-%d %H:%M:%S'))
40		fo.write("# cmd: %s\n" % (' '.join(ensure_text_type(s) for s in sys.argv)))
41
42
43		def is_diff(content):
44		return any(s.startswith(('-', '+')) for s in content)
45
46
47		def pretty_diff(diff):
48		added = {}
49		removed = {}
50		for s in diff:
51		fn = s[1:]
52	7	name, version, _, channel = dist_str_to_quad(fn)
53	7	if channel != DEFAULTS_CHANNEL_NAME:
54		version += ' (%s)' % channel
55		if s.startswith('-'):
56	7	removed[name.lower()] = version
57		elif s.startswith('+'):
58		added[name.lower()] = version
59	7	changed = set(added) & set(removed)
60		for name in sorted(changed):
61	7	yield ' %s {%s -> %s}' % (name, removed[name], added[name])
62	7	for name in sorted(set(removed) - changed):
63	7	yield '-%s-%s' % (name, removed[name])
64	7	for name in sorted(set(added) - changed):
65		yield '+%s-%s' % (name, added[name])
66	7
67	7
68	7	def pretty_content(content):
69		if is_diff(content):
70	7	return pretty_diff(content)
71	7	else:
72		return iter(sorted(content))
73	7
74	7
75	7	class History(object):
76	7
77		com_pat = re.compile(r'#\scmd:\s(.+)')
78	7	spec_pat = re.compile(r'#\s(\w+)\sspecs:\s*(.+)?')
79	7
80		def __init__(self, prefix):
81	7	self.prefix = prefix
82		self.meta_dir = join(prefix, 'conda-meta')
83		self.path = join(self.meta_dir, 'history')
84
85	7	def __enter__(self):
86	7	self.init_log_file()
87	7	return self
88	7
89		def __exit__(self, exc_type, exc_value, traceback):
90		self.update()
91
92		def init_log_file(self):
93	7	touch(self.path, True)
94	7
95		def file_is_empty(self):
96	7	return os.stat(self.path).st_size == 0
97	7
98	7	def update(self):
99	7	"""
100	7	update the history file (creating a new one if necessary)
101		"""
102		try:
103		try:
104		last = set(self.get_state())
105		except CondaHistoryError as e:
106		warnings.warn("Error in %s: %s" % (self.path, e),
107	7	CondaHistoryWarning)
108		return
109		pd = PrefixData(self.prefix)
110		curr = set(prefix_rec.dist_str() for prefix_rec in pd.iter_records())
111		self.write_changes(last, curr)
112	7	except IOError as e:
113	7	if e.errno == errno.EACCES:
114		log.debug("Can't write the history file")
115	7	else:
116	7	raise CondaFileIOError(self.path, "Can't write the history file %s" % e)
117	7
118	7	def parse(self):
119	7	"""
120	7	parse the history file and return a list of
121		tuples(datetime strings, set of distributions/diffs, comments)
122	7	"""
123	7	res = []
124	7	if not isfile(self.path):
125	7	return res
126	7	sep_pat = re.compile(r'==>\s(.+?)\s<==')
127		with open(self.path) as f:
128	7	lines = f.read().splitlines()
129	7	for line in lines:
130		line = line.strip()
131	7	if not line:
132		continue
133		m = sep_pat.match(line)
134		if m:
135		res.append((m.group(1), set(), []))
136		elif line.startswith('#'):
137		res[-1][2].append(line)
138		elif len(res) > 0:
139		res[-1][1].add(line)
140	7	return res
141	7
142	7	@staticmethod
143	7	def _parse_old_format_specs_string(specs_string):
144	7	"""
145	7	Parse specifications string that use conda<4.5 syntax.
146	7
147	7	Examples
148	7	--------
149	7	- "param >=1.5.1,<2.0'"
150	7	- "python>=3.5.1,jupyter >=1.0.0,<2.0,matplotlib >=1.5.1,<2.0"
151	7	"""
152	7	specs = []
153	7	for spec in specs_string.split(','):
154	7	# See https://github.com/conda/conda/issues/6691
155	7	if spec[0].isalpha():
156	7	# A valid spec starts with a letter since it is a package name
157	7	specs.append(spec)
158	7	else:
159	7	# Otherwise it is a condition and has to be appended to the
160		# last valid spec on the specs list
161	7	specs[-1] = ','.join([specs[-1], spec])
162
163		return specs
164
165	7	@classmethod
166	7	def _parse_comment_line(cls, line):
167	7	"""
168	7	Parse comment lines in the history file.
169
170		These lines can be of command type or action type.
171	7
172	7	Examples
173	7	--------
174	7	- "# cmd: /scratch/mc3/bin/conda install -c conda-forge param>=1.5.1,<2.0"
175	7	- "# install specs: python>=3.5.1,jupyter >=1.0.0,<2.0,matplotlib >=1.5.1,<2.0"
176		"""
177		item = {}
178	7	m = cls.com_pat.match(line)
179	7	if m:
180		argv = m.group(1).split()
181	7	if argv[0].endswith('conda'):
182		argv[0] = 'conda'
183		item['cmd'] = argv
184
185		m = cls.spec_pat.match(line)
186		if m:
187	7	action, specs_string = m.groups()
188	7	specs_string = specs_string or ""
189	7	item['action'] = action
190	7
191	7	if specs_string.startswith('['):
192		specs = literal_eval(specs_string)
193	7	elif '[' not in specs_string:
194	7	specs = History._parse_old_format_specs_string(specs_string)
195	7
196	7	specs = [spec for spec in specs if spec and not spec.endswith('@')]
197
198	7	if specs and action in ('update', 'install', 'create'):
199		item['update_specs'] = item['specs'] = specs
200	7	elif specs and action in ('remove', 'uninstall'):
201		item['remove_specs'] = item['specs'] = specs
202
203		return item
204
205		def get_user_requests(self):
206		"""
207		return a list of user requested items. Each item is a dict with the
208		following keys:
209		'date': the date and time running the command
210		'cmd': a list of argv of the actual command which was run
211		'action': install/remove/update
212		'specs': the specs being used
213		"""
214		res = []
215		for dt, unused_cont, comments in self.parse():
216		item = {'date': dt}
217		for line in comments:
218		comment_items = self._parse_comment_line(line)
219		item.update(comment_items)
220
221		if 'cmd' in item:
222		res.append(item)
223
224		dists = groupby(itemgetter(0), unused_cont)
225		item['unlink_dists'] = dists.get('-', ())
226		item['link_dists'] = dists.get('+', ())
227
228		return res
229
230		def get_requested_specs_map(self):
231		# keys are package names and values are specs
232		spec_map = {}
233		for request in self.get_user_requests():
234		remove_specs = (MatchSpec(spec) for spec in request.get('remove_specs', ()))
235		for spec in remove_specs:
236		spec_map.pop(spec.name, None)
237		update_specs = (MatchSpec(spec) for spec in request.get('update_specs', ()))
238		spec_map.update(((s.name, s) for s in update_specs))
239
240		# Conda hasn't always been good about recording when specs have been removed from
241		# environments. If the package isn't installed in the current environment, then we
242		# shouldn't try to force it here.
243		prefix_recs = tuple(PrefixData(self.prefix).iter_records())
244		return dict((name, spec) for name, spec in iteritems(spec_map)
245		if any(spec.match(dist) for dist in prefix_recs))
246
247		def construct_states(self):
248	7	"""
249	7	return a list of tuples(datetime strings, set of distributions)
250	7	"""
251	7	res = []
252	7	cur = set([])
253		for dt, cont, unused_com in self.parse():
254		if not is_diff(cont):
255		cur = cont
256		else:
257	7	for s in cont:
258	7	if s.startswith('-'):
259	7	cur.discard(s[1:])
260	7	elif s.startswith('+'):
261	7	cur.add(s[1:])
262	7	else:
263	7	raise CondaHistoryError('Did not expect: %s' % s)
264		res.append((dt, cur.copy()))
265	7	return res
266
267		def get_state(self, rev=-1):
268		"""
269		return the state, i.e. the set of distributions, for a given revision,
270		defaults to latest (which is the same as the current state when
271		the log file is up-to-date)
272
273		Returns a list of dist_strs
274		"""
275		states = self.construct_states()
276		if not states:
277		return set([])
278		times, pkgs = zip(*states)
279		return pkgs[rev]
280
281		def print_log(self):
282		for i, (date, content, unused_com) in enumerate(self.parse()):
283		print('%s (rev %d)' % (date, i))
284		for line in pretty_content(content):
285		print(' %s' % line)
286		print()
287
288		def object_log(self):
289		result = []
290		for i, (date, content, unused_com) in enumerate(self.parse()):
291		# Based on Mateusz's code; provides more details about the
292		# history event
293		event = {
294		'date': date,
295		'rev': i,
296		'install': [],
297		'remove': [],
298		'upgrade': [],
299		'downgrade': []
300		}
301		added = {}
302		removed = {}
303		if is_diff(content):
304		for pkg in content:
305		name, version, build, channel = dist_str_to_quad(pkg[1:])
306		if pkg.startswith('+'):
307		added[name.lower()] = (version, build, channel)
308		elif pkg.startswith('-'):
309		removed[name.lower()] = (version, build, channel)
310
311		changed = set(added) & set(removed)
312		for name in sorted(changed):
313		old = removed[name]
314		new = added[name]
315		details = {
316		'old': '-'.join((name,) + old),
317		'new': '-'.join((name,) + new)
318		}
319
320		if new > old:
321		event['upgrade'].append(details)
322		else:
323		event['downgrade'].append(details)
324
325		for name in sorted(set(removed) - changed):
326		event['remove'].append('-'.join((name,) + removed[name]))
327
328		for name in sorted(set(added) - changed):
329		event['install'].append('-'.join((name,) + added[name]))
330		else:
331		for pkg in sorted(content):
332		event['install'].append(pkg)
333		result.append(event)
334		return result
335
336		def write_changes(self, last_state, current_state):
337		if not isdir(self.meta_dir):
338		os.makedirs(self.meta_dir)
339		with open(self.path, 'a') as fo:
340		write_head(fo)
341		for fn in sorted(last_state - current_state):
342		fo.write('-%s\n' % fn)
343		for fn in sorted(current_state - last_state):
344		fo.write('+%s\n' % fn)
345
346		def write_specs(self, remove_specs=(), update_specs=()):
347		remove_specs = [text_type(MatchSpec(s)) for s in remove_specs]
348		update_specs = [text_type(MatchSpec(s)) for s in update_specs]
349		if update_specs or remove_specs:
350		with open(self.path, 'a') as fh:
351		if remove_specs:
352		fh.write("# remove specs: %s\n" % remove_specs)
353		if update_specs:
354		fh.write("# update specs: %s\n" % update_specs)
355
356
357		if __name__ == '__main__':
358		from pprint import pprint
359		# Don't use in context manager mode---it augments the history every time
360		h = History(sys.prefix)
361		pprint(h.get_user_requests())
362		print(h.get_requested_specs_map())
363

conda / conda

Push — master ( 1eecb7...7e6e06 )

History.get_user_requests() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like