# changelog.py -- Python module for Debian changelogs # Copyright (C) 2006-7 James Westby # Copyright (C) 2008 Canonical Ltd. # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty 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 # The parsing code is based on that from dpkg which is: # Copyright 1996 Ian Jackson # Copyright 2005 Frank Lichtenheld # and licensed under the same license as above. """This module implements facilities to deal with Debian changelogs.""" from __future__ import absolute_import import os import pwd import re import socket import warnings import sys import six from debian import debian_support # Python 3 doesn't have StandardError, but let's avoid changing our # exception inheritance hierarchy for Python 2. try: _base_exception_class = StandardError except NameError: _base_exception_class = Exception class ChangelogParseError(_base_exception_class): """Indicates that the changelog could not be parsed""" is_user_error = True def __init__(self, line): self._line=line def __str__(self): return "Could not parse changelog: "+self._line class ChangelogCreateError(_base_exception_class): """Indicates that changelog could not be created, as all the information required was not given""" class VersionError(_base_exception_class): """Indicates that the version does not conform to the required format""" is_user_error = True def __init__(self, version): self._version=version def __str__(self): return "Could not parse version: "+self._version # TODO(jsw): Remove this in favor of using debian_support.Version directly. I # don't think we gain anything by using this empty subclass. class Version(debian_support.Version): """Represents a version of a Debian package.""" # debian_support.Version now has all the functionality we need class ChangeBlock(object): """Holds all the information about one block from the changelog.""" def __init__(self, package=None, version=None, distributions=None, urgency=None, urgency_comment=None, changes=None, author=None, date=None, other_pairs=None, encoding='utf-8'): self._raw_version = None self._set_version(version) self.package = package self.distributions = distributions self.urgency = urgency or "unknown" self.urgency_comment = urgency_comment or '' self._changes = changes self.author = author self.date = date self._trailing = [] self.other_pairs = other_pairs or {} self._encoding = encoding self._no_trailer = False self._trailer_separator = " " def _set_version(self, version): if version is not None: self._raw_version = str(version) def _get_version(self): return Version(self._raw_version) version = property(_get_version, _set_version) def other_keys_normalised(self): norm_dict = {} for (key, value) in other_pairs.items(): key = key[0].upper() + key[1:].lower() m = xbcs_re.match(key) if m is None: key = "XS-%s" % key norm_dict[key] = value return norm_dict def changes(self): return self._changes def add_trailing_line(self, line): self._trailing.append(line) def add_change(self, change): if self._changes is None: self._changes = [change] else: #Bit of trickery to keep the formatting nicer with a blank #line at the end if there is one changes = self._changes changes.reverse() added = False for i in range(len(changes)): m = blankline.match(changes[i]) if m is None: changes.insert(i, change) added = True break changes.reverse() if not added: changes.append(change) self._changes = changes def _format(self): # TODO(jsw): Switch to StringIO or a list to join at the end. block = "" if self.package is None: raise ChangelogCreateError("Package not specified") block += self.package + " " if self._raw_version is None: raise ChangelogCreateError("Version not specified") block += "(" + self._raw_version + ") " if self.distributions is None: raise ChangelogCreateError("Distribution not specified") block += self.distributions + "; " if self.urgency is None: raise ChangelogCreateError("Urgency not specified") block += "urgency=" + self.urgency + self.urgency_comment for (key, value) in self.other_pairs.items(): block += ", %s=%s" % (key, value) block += '\n' if self.changes() is None: raise ChangelogCreateError("Changes not specified") for change in self.changes(): block += change + "\n" if not self._no_trailer: if self.author is None: raise ChangelogCreateError("Author not specified") if self.date is None: raise ChangelogCreateError("Date not specified") block += " -- " + self.author + self._trailer_separator \ + self.date + "\n" for line in self._trailing: block += line + "\n" return block if sys.version >= '3': __str__ = _format def __bytes__(self): return str(self).encode(self._encoding) else: __unicode__ = _format def __str__(self): return unicode(self).encode(self._encoding) topline = re.compile(r'^(\w%(name_chars)s*) \(([^\(\) \t]+)\)' '((\s+%(name_chars)s+)+)\;' % {'name_chars': '[-+0-9a-z.]'}, re.IGNORECASE) blankline = re.compile('^\s*$') change = re.compile('^\s\s+.*$') endline = re.compile('^ -- (.*) <(.*)>( ?)((\w+\,\s*)?\d{1,2}\s+\w+\s+' '\d{4}\s+\d{1,2}:\d\d:\d\d\s+[-+]\d{4}(\s+\([^\\\(\)]\))?\s*)$') endline_nodetails = re.compile('^ --(?: (.*) <(.*)>( ?)((\w+\,\s*)?\d{1,2}' '\s+\w+\s+\d{4}\s+\d{1,2}:\d\d:\d\d\s+[-+]\d{4}' '(\s+\([^\\\(\)]\))?))?\s*$') keyvalue= re.compile('^([-0-9a-z]+)=\s*(.*\S)$', re.IGNORECASE) value_re = re.compile('^([-0-9a-z]+)((\s+.*)?)$', re.IGNORECASE) xbcs_re = re.compile('^X[BCS]+-', re.IGNORECASE) emacs_variables = re.compile('^(;;\s*)?Local variables:', re.IGNORECASE) vim_variables = re.compile('^vim:', re.IGNORECASE) cvs_keyword = re.compile('^\$\w+:.*\$') comments = re.compile('^\# ') more_comments = re.compile('^/\*.*\*/') old_format_re1 = re.compile('^(\w+\s+\w+\s+\d{1,2} \d{1,2}:\d{1,2}:\d{1,2}' '\s+[\w\s]*\d{4})\s+(.*)\s+(<|\()(.*)(\)|>)') old_format_re2 = re.compile('^(\w+\s+\w+\s+\d{1,2},?\s*\d{4})\s+(.*)' '\s+(<|\()(.*)(\)|>)') old_format_re3 = re.compile('^(\w[-+0-9a-z.]*) \(([^\(\) \t]+)\)\;?', re.IGNORECASE) old_format_re4 = re.compile('^([\w.+-]+)(-| )(\S+) Debian (\S+)', re.IGNORECASE) old_format_re5 = re.compile('^Changes from version (.*) to (.*):', re.IGNORECASE) old_format_re6 = re.compile('^Changes for [\w.+-]+-[\w.+-]+:?\s*$', re.IGNORECASE) old_format_re7 = re.compile('^Old Changelog:\s*$', re.IGNORECASE) old_format_re8 = re.compile('^(?:\d+:)?\w[\w.+~-]*:?\s*$') class Changelog(object): """Represents a debian/changelog file.""" # TODO(jsw): Avoid masking the 'file' built-in. def __init__(self, file=None, max_blocks=None, allow_empty_author=False, strict=True, encoding='utf-8'): """Initializer. Args: file: The contents of the changelog, either as a str, unicode object, or an iterator of lines (each of which is either a str or unicode) max_blocks: The maximum number of blocks to parse from the input. (Default: no limit) allow_empty_author: Whether to allow an empty author in the trailer line of a change block. (Default: False) strict: Whether to raise an exception if there are errors. (Default: use a warning) encoding: If the input is a str or iterator of str, the encoding to use when interpreting the input. """ self._encoding = encoding self._blocks = [] self.initial_blank_lines = [] if file is not None: try: self.parse_changelog(file, max_blocks=max_blocks, allow_empty_author=allow_empty_author, strict=strict) except ChangelogParseError: pass def _parse_error(self, message, strict): if strict: raise ChangelogParseError(message) else: warnings.warn(message) def parse_changelog(self, file, max_blocks=None, allow_empty_author=False, strict=True, encoding=None): first_heading = "first heading" next_heading_or_eof = "next heading of EOF" start_of_change_data = "start of change data" more_changes_or_trailer = "more change data or trailer" slurp_to_end = "slurp to end" encoding = encoding or self._encoding if file is None: self._parse_error('Empty changelog file.', strict) return self._blocks = [] self.initial_blank_lines = [] current_block = ChangeBlock(encoding=encoding) changes = [] state = first_heading old_state = None if isinstance(file, bytes): file = file.decode(encoding) if isinstance(file, six.string_types): # Make sure the changelog file is not empty. if len(file.strip()) == 0: self._parse_error('Empty changelog file.', strict) return file = file.splitlines() for line in file: if not isinstance(line, six.text_type): line = line.decode(encoding) # Support both lists of lines without the trailing newline and # those with trailing newlines (e.g. when given a file object # directly) line = line.rstrip('\n') if state == first_heading or state == next_heading_or_eof: top_match = topline.match(line) blank_match = blankline.match(line) if top_match is not None: if (max_blocks is not None and len(self._blocks) >= max_blocks): return current_block.package = top_match.group(1) current_block._raw_version = top_match.group(2) current_block.distributions = top_match.group(3).lstrip() pairs = line.split(";", 1)[1] all_keys = {} other_pairs = {} for pair in pairs.split(','): pair = pair.strip() kv_match = keyvalue.match(pair) if kv_match is None: self._parse_error("Invalid key-value " "pair after ';': %s" % pair, strict) continue key = kv_match.group(1) value = kv_match.group(2) if key.lower() in all_keys: self._parse_error("Repeated key-value: " "%s" % key.lower(), strict) all_keys[key.lower()] = value if key.lower() == "urgency": val_match = value_re.match(value) if val_match is None: self._parse_error("Badly formatted " "urgency value: %s" % value, strict) else: current_block.urgency = val_match.group(1) comment = val_match.group(2) if comment is not None: current_block.urgency_comment = comment else: other_pairs[key] = value current_block.other_pairs = other_pairs state = start_of_change_data elif blank_match is not None: if state == first_heading: self.initial_blank_lines.append(line) else: self._blocks[-1].add_trailing_line(line) else: emacs_match = emacs_variables.match(line) vim_match = vim_variables.match(line) cvs_match = cvs_keyword.match(line) comments_match = comments.match(line) more_comments_match = more_comments.match(line) if ((emacs_match is not None or vim_match is not None) and state != first_heading): self._blocks[-1].add_trailing_line(line) old_state = state state = slurp_to_end continue if (cvs_match is not None or comments_match is not None or more_comments_match is not None): if state == first_heading: self.initial_blank_lines.append(line) else: self._blocks[-1].add_trailing_line(line) continue if ((old_format_re1.match(line) is not None or old_format_re2.match(line) is not None or old_format_re3.match(line) is not None or old_format_re4.match(line) is not None or old_format_re5.match(line) is not None or old_format_re6.match(line) is not None or old_format_re7.match(line) is not None or old_format_re8.match(line) is not None) and state != first_heading): self._blocks[-1].add_trailing_line(line) old_state = state state = slurp_to_end continue self._parse_error("Unexpected line while looking " "for %s: %s" % (state, line), strict) if state == first_heading: self.initial_blank_lines.append(line) else: self._blocks[-1].add_trailing_line(line) elif (state == start_of_change_data or state == more_changes_or_trailer): change_match = change.match(line) end_match = endline.match(line) end_no_details_match = endline_nodetails.match(line) blank_match = blankline.match(line) if change_match is not None: changes.append(line) state = more_changes_or_trailer elif end_match is not None: if end_match.group(3) != ' ': self._parse_error("Badly formatted trailer " "line: %s" % line, strict) current_block._trailer_separator = end_match.group(3) current_block.author = "%s <%s>" \ % (end_match.group(1), end_match.group(2)) current_block.date = end_match.group(4) current_block._changes = changes self._blocks.append(current_block) changes = [] current_block = ChangeBlock(encoding=encoding) state = next_heading_or_eof elif end_no_details_match is not None: if not allow_empty_author: self._parse_error("Badly formatted trailer " "line: %s" % line, strict) continue current_block._changes = changes self._blocks.append(current_block) changes = [] current_block = ChangeBlock(encoding=encoding) state = next_heading_or_eof elif blank_match is not None: changes.append(line) else: cvs_match = cvs_keyword.match(line) comments_match = comments.match(line) more_comments_match = more_comments.match(line) if (cvs_match is not None or comments_match is not None or more_comments_match is not None): changes.append(line) continue self._parse_error("Unexpected line while looking " "for %s: %s" % (state, line), strict) changes.append(line) elif state == slurp_to_end: if old_state == next_heading_or_eof: self._blocks[-1].add_trailing_line(line) else: changes.append(line) else: assert False, "Unknown state: %s" % state if ((state != next_heading_or_eof and state != slurp_to_end) or (state == slurp_to_end and old_state != next_heading_or_eof)): self._parse_error("Found eof where expected %s" % state, strict) current_block._changes = changes current_block._no_trailer = True self._blocks.append(current_block) def get_version(self): """Return a Version object for the last version""" return self._blocks[0].version def set_version(self, version): """Set the version of the last changelog block version can be a full version string, or a Version object """ self._blocks[0].version = Version(version) version = property(get_version, set_version, doc="Version object for last changelog block""") ### For convenience, let's expose some of the version properties full_version = property(lambda self: self.version.full_version) epoch = property(lambda self: self.version.epoch) debian_version = property(lambda self: self.version.debian_revision) debian_revision = property(lambda self: self.version.debian_revision) upstream_version = property(lambda self: self.version.upstream_version) def get_package(self): """Returns the name of the package in the last version.""" return self._blocks[0].package def set_package(self, package): self._blocks[0].package = package package = property(get_package, set_package, doc="Name of the package in the last version") def get_versions(self): """Returns a list of version objects that the package went through.""" return [block.version for block in self._blocks] versions = property(get_versions, doc="List of version objects the package went through") def _raw_versions(self): return [block._raw_version for block in self._blocks] def _format(self): pieces = [] pieces.append(six.u('\n').join(self.initial_blank_lines)) for block in self._blocks: pieces.append(six.text_type(block)) return six.u('').join(pieces) if sys.version >= '3': __str__ = _format def __bytes__(self): return str(self).encode(self._encoding) else: __unicode__ = _format def __str__(self): return unicode(self).encode(self._encoding) def __iter__(self): return iter(self._blocks) def __len__(self): return len(self._blocks) def set_distributions(self, distributions): self._blocks[0].distributions = distributions distributions = property(lambda self: self._blocks[0].distributions, set_distributions) def set_urgency(self, urgency): self._blocks[0].urgency = urgency urgency = property(lambda self: self._blocks[0].urgency, set_urgency) def add_change(self, change): self._blocks[0].add_change(change) def set_author(self, author): self._blocks[0].author = author author = property(lambda self: self._blocks[0].author, set_author) def set_date(self, date): self._blocks[0].date = date date = property(lambda self: self._blocks[0].date, set_date) def new_block(self, **kwargs): kwargs.setdefault('encoding', self._encoding) block = ChangeBlock(**kwargs) block.add_trailing_line('') self._blocks.insert(0, block) def write_to_open_file(self, file): file.write(self.__str__()) def get_maintainer(): """Get the maintainer information in the same manner as dch. This function gets the information about the current user for the maintainer field using environment variables of gecos informations as approriate. It uses the same methods as dch to get the information, namely DEBEMAIL, DEBFULLNAME, EMAIL, NAME, /etc/mailname and gecos. :returns: a tuple of the full name, email pair as strings. Either of the pair may be None if that value couldn't be determined. """ env = os.environ regex = re.compile(r"^(.*)\s+<(.*)>$") # Split email and name if 'DEBEMAIL' in env: match_obj = regex.match(env['DEBEMAIL']) if match_obj: if not 'DEBFULLNAME' in env: env['DEBFULLNAME'] = match_obj.group(1) env['DEBEMAIL'] = match_obj.group(2) if 'DEBEMAIL' not in env or 'DEBFULLNAME' not in env: if 'EMAIL' in env: match_obj = regex.match(env['EMAIL']) if match_obj: if not 'DEBFULLNAME' in env: env['DEBFULLNAME'] = match_obj.group(1) env['EMAIL'] = match_obj.group(2) # Get maintainer's name if 'DEBFULLNAME' in env: maintainer = env['DEBFULLNAME'] elif 'NAME' in env: maintainer = env['NAME'] else: # Use password database if no data in environment variables try: maintainer = re.sub(r',.*', '', pwd.getpwuid(os.getuid()).pw_gecos) except (KeyError, AttributeError): maintainer = None # Get maintainer's mail address if 'DEBEMAIL' in env: email = env['DEBEMAIL'] elif 'EMAIL' in env: email = env['EMAIL'] else: addr = None if os.path.exists('/etc/mailname'): f = open('/etc/mailname') try: addr = f.readline().strip() finally: f.close() if not addr: addr = socket.getfqdn() if addr: user = pwd.getpwuid(os.getuid()).pw_name if not user: addr = None else: addr = "%s@%s" % (user, addr) if addr: email = addr else: email = None return (maintainer, email)