debdry 0.2.2-1 / usr / lib / python3 / dist-packages / debdrylib / tree.py

#!/usr/bin/python3
# coding: utf8
import os
import io
import re
import shutil
import apt_pkg
from collections import OrderedDict
import tempfile
import difflib
import logging

log = logging.getLogger(__name__)

class Node:
    pass


class FilesystemNode(Node):
    def __init__(self, relname, absname):
        self.relname = relname
        self.absname = absname

    def dump(self, out, level=0):
        print("{}{}".format(" " * level, self.relname), file=out)


class DirectoryBase(FilesystemNode):
    def __init__(self, relname, absname):
        super().__init__(relname, absname)
        self.files = {}

    def dump(self, out, level=0):
        super().dump(out, level)
        for n in self.files.values():
            n.dump(out, level + 1)


class DebianDirectory(DirectoryBase):
    """
    Represent a debian/ directory
    """
    def __init__(self, srcdir, absname=None):
        if absname is None: absname = os.path.join(srcdir, "debian")
        super().__init__(os.path.basename(absname), absname)
        self.srcdir = srcdir

    @classmethod
    def scan(cls, srcdir, absname=None):
        """
        Init the DebianDirectory by scanning the filesystem
        """
        self = cls(srcdir, absname)
        for relname in os.listdir(self.absname):
            absname = os.path.join(self.absname, relname)
            if relname == "control":
                log.info("%s: control file", absname)
                self.files[relname] = Control.scan(relname, absname)
            elif relname == "rules":
                log.info("%s: rules file", absname)
                self.files[relname] = Rules.scan(relname, absname)
            elif os.path.isdir(absname):
                log.info("%s: subdirectory", absname)
                self.files[relname] = Directory.scan(relname, absname)
            else:
                log.info("%s: plain file", absname)
                self.files[relname] = File.scan(relname, absname)
        return self

    @classmethod
    def combine(cls, srcdir, absname, base, extra):
        """
        Init the DebianDirectory by overlaying extra on top of base
        """
        os.makedirs(absname, exist_ok=True)
        self = cls(srcdir, absname)
        for k in base.files.keys() - extra.files.keys():
            self.files[k] = base.files[k].copy_to(k, os.path.join(self.absname, k))
        for k in extra.files.keys() - base.files.keys():
            self.files[k] = extra.files[k].copy_to(k, os.path.join(self.absname, k))
        for k in base.files.keys() & extra.files.keys():
            self.files[k] = base.files[k].combine(k, os.path.join(self.absname, k), extra.files[k])
        return self

    @classmethod
    def diff(cls, srcdir, absname, base, extra):
        """
        Init the DebianDirectory with the diff to go from base to extra
        """
        os.makedirs(absname, exist_ok=True)
        self = cls(srcdir, absname)
        for k in base.files.keys() - extra.files.keys():
            log.warn("{}: exists in {} but not in {}: ignoring file".format(
                self.relname, base.absname, extra.absname))
        for k in extra.files.keys() - base.files.keys():
            self.files[k] = extra.files[k].copy_to(k, os.path.join(self.absname, k))
        for k in base.files.keys() & extra.files.keys():
            self.files[k] = base.files[k].diff(k, os.path.join(self.absname, k), extra.files[k])
        return self

class Directory(DirectoryBase):
    """
    Represent a subdirectory of debian/, any level down
    """
    @classmethod
    def scan(cls, relname, absname):
        self = cls(relname, absname)
        for relname in os.listdir(self.absname):
            absname = os.path.join(self.absname, relname)
            if os.path.isdir(absname):
                log.info("%s: subdirectory", absname)
                self.files[relname] = Directory(relname, absname)
            else:
                log.info("%s: plain file", absname)
                self.files[relname] = File(relname, absname)
        return self

    def copy_to(self, relname, absname):
        log.debug("%s: generating subdir", absname)
        os.makedirs(absname, exist_ok=True)
        for name, node in self.files.items():
            node.copy_to(name, os.path.join(absname, name))
        return self.__class__(relname, absname)

    def combine(self, relname, absname, other):
        os.makedirs(absname, exist_ok=True)
        res = self.__class__(relname, absname)
        for k in self.files.keys() - other.files.keys():
            self.files[k] = self.files[k].copy_to(k, os.path.join(self.absname, k))
        for k in other.files.keys() - self.files.keys():
            self.files[k] = other.files[k].copy_to(k, os.path.join(self.absname, k))
        for k in other.files.keys() & self.files.keys():
            self.files[k] = self.files[k].combine(k, os.path.join(self.absname, k), other.files[k])
        return self

    def diff(self, relname, absname, other):
        os.makedirs(absname, exist_ok=True)
        res = self.__class__(relname, absname)
        for k in self.files.keys() - other.files.keys():
            log.warn("{}: exists in {} but not in {}: ignoring file".format(
                self.relname, base.absname, extra.absname))
        for k in other.files.keys() - self.files.keys():
            self.files[k] = other.files[k].copy_to(k, os.path.join(self.absname, k))
        for k in other.files.keys() & self.files.keys():
            self.files[k] = self.files[k].diff(k, os.path.join(self.absname, k), other.files[k])
        return self


class DeletedFile(FilesystemNode):
    def dump(self, out, level=0):
        print("{}{} (deleted)".format(" " * level, self.relname), file=out)


class File(FilesystemNode):
    """
    Represent a file anywhere under debian/
    """
    def open(self, mode="rt", encoding="utf8", **kw):
        return io.open(self.absname, mode=mode, encoding=encoding, **kw)

    def read(self, **kw):
        with io.open(self.absname, **kw) as fd:
            return fd.read()

    def readlines(self, **kw):
        with io.open(self.absname, **kw) as fd:
            return fd.readlines()

    def copy_to(self, relname, absname):
        log.info("%s: copying from %s", absname, self.absname)
        shutil.copy2(self.absname, absname)
        return self.__class__(relname, absname)

    def combine(self, relname, absname, other):
        shutil.copy2(other.absname, absname)
        return self.__class__.scan(relname, absname)

    def diff(self, relname, absname, other):
        if self.read() != other.read():
            shutil.copy2(other.absname, absname)
            return self.__class__.scan(relname, absname)
        else:
            return DeletedFile(relname, absname)

    @classmethod
    def scan(cls, relname, absname):
        return cls(relname, absname)


class Rules(File):
    """
    Represent a debian/rules file
    """
    @property
    def contents(self):
        res = getattr(self, "_contents", None)
        if res is not None: return res
        self._contents = self.read()
        return self._contents

    @classmethod
    def scan(cls, relname, absname):
        return cls(relname, absname)

    def combine(self, relname, absname, other):
        if other.contents.startswith("#!"):
            shutil.copy2(other.absname, absname)
        else:
            with io.open(absname, "wt") as fd:
                fd.write(self.contents)
                fd.write("\n")
                fd.write(other.contents)
        shutil.copystat(other.absname, absname)
        return self.__class__.scan(relname, absname)

    def diff(self, relname, absname, other):
        log.warn("%s: diffing debian/rules is not yet supported: generating a diff instead", self.relname)
        my_lines = self.readlines()
        other_lines = other.readlines()
        dstpathname = absname + ".diff"
        with io.open(dstpathname, "wt") as out:
            for line in difflib.unified_diff(my_lines, other_lines, "debian.auto/rules", "debian/rules"):
                out.write(line)
        return File(relname + ".diff", dstpathname)

    def dump(self, out, level=0):
        if self.contents.startswith("#!"):
            print("{}{} (full)".format(" " * level, self.relname), file=out)
        else:
            print("{}{} (partial)".format(" " * level, self.relname), file=out)


class Control(File):
    SOURCE_FIELDS = frozenset(apt_pkg.REWRITE_SOURCE_ORDER)
    BINARY_FIELDS = frozenset(apt_pkg.REWRITE_PACKAGE_ORDER)
    PKGLIST_FIELDS = frozenset(("Depends", "Pre-Depends", "Recommends",
                                "Suggests", "Breaks", "Conflicts", "Provides",
                                "Replaces", "Enhances", "Build-Depends"))

    def __init__(self, relname, absname):
        super().__init__(relname, absname)
        # Source: section
        self.source = ControlSectionSource()
        # Package: sections
        self.binaries = OrderedDict()

    def _scan_lines(self, data):
        for section in apt_pkg.TagFile(data):
            if 'Source' in section:
                for tag in section.keys():
                    self.add_source_field(tag, section[tag])
            elif 'Package' in section:
                name = section["Package"]
                for tag in section.keys():
                    self.add_binary_field(name, tag, section[tag])
            else:
                # If there is no Source or Package in this stanza, dispatch
                # according to field names
                for tag in section.keys():
                    self.add_mixed_field(tag, section[tag])

    @classmethod
    def scan(cls, relname, absname):
        self = cls(relname, absname)
        with self.open() as fd:
            self._scan_lines(fd)
        return self

    @classmethod
    def scan_string(cls, relname, absname, buf):
        self = cls(relname, absname)
        with tempfile.TemporaryFile() as fd:
            fd.write(buf.encode("utf8"))
            fd.seek(0)
            self._scan_lines(fd)
        return self

    def to_string(self):
        sections = []

        # Add the source section
        sections.append(self.source.to_string())

        # Add the binary sections
        for section in self.binaries.values():
            sections.append(section.to_string())

        return "\n".join(sections)

    def write(self):
        # Write out the concatenation of all the stanzas
        with io.open(self.absname, "wt") as outfd:
            outfd.write(self.to_string())

    def combine(self, relname, absname, other):
        res = self.__class__(relname, absname)
        res.source = self.source.combine(other.source)

        # Add the binary sections that are in self but not in other
        for name in self.binaries.keys() - other.binaries.keys():
            res.binaries[name] = self.binaries[name].copy()
            #log.warn("%s: Package: %s exists only in auto: ignored", absname, name)

        # Add the binary sections that are in other but not in self
        for name in other.binaries.keys() - self.binaries.keys():
            res.binaries[name] = other.binaries[name].copy()

        # Merge the binary sections that are in both, taking auto as the
        # baseline
        for name in self.binaries.keys() & other.binaries.keys():
            res.binaries[name] = self.binaries[name].combine(other.binaries[name])

        # Write out the file
        res.write()
        return res

    def diff(self, relname, absname, other):
        res = self.__class__(relname, absname)
        res.source = self.source.diff(other.source)

        # Add the binary sections that are in self but not in other
        for name in self.binaries.keys() - other.binaries.keys():
            log.warn("%s: Package: %s exists only in auto: ignored", absname, name)

        # Add the binary sections that are in other but not in self
        for name in other.binaries.keys() - self.binaries.keys():
            res.binaries[name] = other.binaries[name].copy()

        # Merge the binary sections that are in both, taking auto as the
        # baseline
        for name in self.binaries.keys() & other.binaries.keys():
            res.binaries[name] = self.binaries[name].diff(other.binaries[name])

        # Write out the file
        res.write()
        return res

    def add_source_field(self, tag, value):
        if tag.startswith("X-Debdry-"):
            tag = tag[9:]
            smart = True
        else:
            smart = False

        if tag == "Uploaders":
            self.source.add(tag, UploadersControlField(tag, value, smart))
        elif tag in self.PKGLIST_FIELDS:
            self.source.add(tag, PkglistControlField(tag, value, smart))
        else:
            if smart:
                log.warn("%s: unsupported field X-Debdry-%s found: treating it as %s",
                         self.absname, tag, tag)
            self.source.add(tag, PlainControlField(tag, value))

    def add_binary_field(self, pkgname, tag, value):
        if tag.startswith("X-Debdry-"):
            tag = tag[9:]
            smart = True
        else:
            smart = False

        rec = self.binaries.get(pkgname, None)
        if rec is None:
            self.binaries[pkgname] = rec = ControlSectionBinary()

        if tag in self.PKGLIST_FIELDS:
            rec.add(tag, PkglistControlField(tag, value, smart))
        else:
            if smart:
                log.warn("%s: unsupported field X-Debdry-%s found: treating it as %s",
                         self.absname, tag, tag)
            rec.add(tag, PlainControlField(tag, value))

    def add_mixed_field(self, tag, value):
        orig_tag = tag
        if tag.startswith("X-Debdry-"):
            tag = tag[9:]

        if tag in self.SOURCE_FIELDS:
            self.add_source_field(orig_tag, value)
        elif tag in self.BINARY_FIELDS:
            self.add_binary_field(None, orig_tag, value)
        else:
            log.warn("%s: ignoring unrecognised field '%s'", self.absname, orig_tag)

    def dump(self, out, level=0):
        super().dump(out, level)

        print("{}src:".format(" " * (level + 1)))
        self.source.dump(out, level + 1)

        for pkgname, section in self.binaries.items():
            print("{}bin {}:".format(" " * (level + 1), pkgname))
            section.dump(out, level + 1)


class ControlSection:
    def __init__(self):
        self.fields = OrderedDict()

    def add(self, tag, field):
        if field is None: return
        self.fields[tag] = field

    def dump(self, out, level=0):
        for tag, field in self.fields.items():
            field.dump(out, level + 1)

    def copy(self):
        res = self.__class__()
        for k, v in self.fields.items():
            res.fields[k] = v.copy()
        return res

    def to_string(self):
        section = self.get_empty_tagsection()
        return apt_pkg.rewrite_section(section, self.REWRITE_ORDER,
                                       [f.to_apt() for f in self.fields.values()])

    def combine(self, other):
        res = self.__class__()

        # All fields in self are copied
        for tag in self.fields.keys() - other.fields.keys():
            res.add(tag, self.fields[tag].copy())

        # Then all fields in other are copied
        for tag in other.fields.keys() - self.fields.keys():
            res.add(tag, other.fields[tag].copy())

        # Combine the fields that are in both
        for tag in self.fields.keys() & other.fields.keys():
            res.add(tag, self.fields[tag].combine(other.fields[tag]))

        return res

    def diff(self, other):
        res = self.__class__()

        # FIXME: we still have no way to delete a field
        for tag in self.fields.keys() - other.fields.keys():
            log.warn("control field %s exists only in auto: ignored", tag)

        # Then all fields in other are copied
        for tag in other.fields.keys() - self.fields.keys():
            res.add(tag, other.fields[tag].copy())

        # Combine the fields that are in both
        for tag in self.fields.keys() & other.fields.keys():
            if tag in ("Source", "Package"):
                res.add(tag, other.fields[tag].copy())
            else:
                res.add(tag, self.fields[tag].diff(other.fields[tag]))

        return res

class ControlSectionSource(ControlSection):
    REWRITE_ORDER = apt_pkg.REWRITE_SOURCE_ORDER

    def get_empty_tagsection(self):
        name = self.fields.get("Source", None)
        if name is None:
            raise RuntimeError("TODO: package is name is missing for source section")
        return apt_pkg.TagSection("Source: {}\n".format(name))

class ControlSectionBinary(ControlSection):
    REWRITE_ORDER = apt_pkg.REWRITE_PACKAGE_ORDER

    def get_empty_tagsection(self):
        name = self.fields.get("Package", None)
        if name is None:
            raise RuntimeError("TODO: package is name is missing for binary section")
        return apt_pkg.TagSection("Package: {}\n".format(name))


class PlainControlField(Node):
    def __init__(self, tag, value):
        self.tag = tag
        self.value = value

    def dump(self, out, level=0):
        print("{}{}: {}".format(" " * level, self.tag, self.value[:40]))

    def copy(self):
        return self.__class__(self.tag, self.value)

    def to_string(self):
        return self.value

    def to_apt(self):
        return (self.tag, self.value)

    def combine(self, other):
        return self.__class__(self.tag, other.value)

    def diff(self, other):
        if self.value == other.value:
            return None
        return self.__class__(self.tag, other.value)

class IndexedControlField(Node):
    def __init__(self, tag, smart):
        self.tag = tag
        self.smart = smart
        self.values = OrderedDict()

    def dump(self, out, level=0):
        if self.smart:
            tag = "X-Debdry-" + self.tag
        else:
            tag = self.tag

        if len(self.values) > 1:
            print("{}{}: {} and {} more".format(" " * level,
                                                tag,
                                                ", ".join(self.values.values())[:40],
                                                len(self.values) - 1))
        else:
            print("{}{}: {}".format(" " * level,
                                    tag,
                                    ", ".join(self.values.values())[:40]))

    def copy(self):
        return self.__class__(self.tag, self.to_string(), self.smart)

    def to_string(self):
        return ", ".join(self.values.values())

    def to_apt(self):
        if self.smart:
            return ("X-Debdry-" + self.tag, self.to_string())
        else:
            return (self.tag, self.to_string())

    def combine(self, other):
        if not other.smart:
            value = other.to_string()
        else:
            res = OrderedDict()
            for k, v in self.values.items():
                res[k] = v
            for k, v in other.values.items():
                res[k] = v
            value = ", ".join(res.values())
        return self.__class__(self.tag, value, self.smart)

    def diff(self, other):
        smart = False
        if self.to_string() == other.to_string():
            value = None
        elif self.values.keys() - other.values.keys():
            # If there are values in self and not in others, do a full field
            # override
            value = other.to_string()
        else:
            res = OrderedDict()

            # Common fields first
            for k in self.values.keys() & other.values.keys():
                theirs = other.values[k]
                # Skip common fields
                if self.values[k] == theirs: continue
                res[k] = theirs

            # Then other-only fields
            for k in other.values.keys() - self.values.keys():
                res[k] = other.values[k]

            if res:
                smart = True
                value = ", ".join(res.values())
            else:
                value = None

        if value is None: return None
        return self.__class__(self.tag, value, smart)



class UploadersControlField(IndexedControlField):
    re_split_uploaders = re.compile(r'(?<=>)\s*,\s*')
    re_email = re.compile(r'<([^>]+)>')

    def __init__(self, tag, value, smart):
        super().__init__(tag, smart)
        # Store uploaders indexed by email address
        for v in self.re_split_uploaders.split(value):
            self.values[self._get_email(v)] = v

    def _get_email(self, maint):
        mo = self.re_email.search(maint)
        if not mo:
            raise ValueError("Unparsable email: {}".format(repr(maint)))
        return mo.group(1)


class PkglistControlField(IndexedControlField):
    re_split_deps = re.compile(r"\s*,\s*")

    def __init__(self, tag, value, smart):
        super().__init__(tag, smart)
        # Store package entries indexed by package name
        for v in self.re_split_deps.split(value):
            if not v: continue
            self.values[v.split()[0]] = v