a38tool

#!/usr/bin/python3
from __future__ import annotations
import argparse
import contextlib
import fnmatch
import logging
import os.path
import re
import shutil
import sys
from pathlib import Path
from typing import TYPE_CHECKING, Optional, IO, Union

from a38 import codec, models

if TYPE_CHECKING:
    import fattura
    from .fattura import Fattura
    from .fattura_semplificata import FatturaElettronicaSemplificata

log = logging.getLogger("a38tool")


class Fail(Exception):
    pass


class App:
    NAME = None

    def __init__(self, args):
        self.args = args

    def load_fattura(self, pathname) -> Union[Fattura, FatturaElettronicaSemplificata]:
        codecs = codec.Codecs()
        codec_cls = codecs.codec_from_filename(pathname)
        return codec_cls().load(pathname)

    @classmethod
    def add_subparser(cls, subparsers):
        name = getattr(cls, "NAME", None)
        if name is None:
            name = cls.__name__.lower()
        parser = subparsers.add_parser(name, help=cls.__doc__.strip())
        parser.set_defaults(app=cls)
        return parser


class Diff(App):
    """
    show the difference between two fatture
    """

    NAME = "diff"

    def __init__(self, args):
        super().__init__(args)
        self.first = args.first
        self.second = args.second

    @classmethod
    def add_subparser(cls, subparsers):
        parser = super().add_subparser(subparsers)
        parser.add_argument("first", help="first input file (.xml or .xml.p7m)")
        parser.add_argument("second", help="second input file (.xml or .xml.p7m)")
        return parser

    def run(self):
        first = self.load_fattura(self.first)
        second = self.load_fattura(self.second)
        from a38.diff import Diff

        res = Diff()
        first.diff(res, second)
        if res.differences:
            for d in res.differences:
                print(d)
            return 1


class Validate(App):
    """
    validate the contents of a fattura
    """

    NAME = "validate"

    def __init__(self, args):
        super().__init__(args)
        self.pathname = args.file

    @classmethod
    def add_subparser(cls, subparsers):
        parser = super().add_subparser(subparsers)
        parser.add_argument("file", help="input file (.xml or .xml.p7m)")
        return parser

    def run(self):
        f = self.load_fattura(self.pathname)
        from a38.validation import Validation

        res = Validation()
        f.validate(res)
        if res.warnings:
            for w in res.warnings:
                print(str(w), file=sys.stderr)
        if res.errors:
            for e in res.errors:
                print(str(e), file=sys.stderr)
            return 1


class Exporter(App):
    def __init__(self, args):
        super().__init__(args)
        self.files = args.files
        self.output = args.output
        self.codec = self.get_codec()

    def get_codec(self) -> codec.Codec:
        """
        Instantiate the output codec to use for this exporter
        """
        raise NotImplementedError(
            f"{self.__class__.__name__}.get_codec is not implemented"
        )

    def write(self, f: models.Model, file: Union[IO[str], IO[bytes]]):
        self.codec.write_file(f, file)

    @contextlib.contextmanager
    def open_output(self):
        if self.output is None:
            if self.codec.binary:
                yield sys.stdout.buffer
            else:
                yield sys.stdout
        else:
            with open(self.output, "wb" if self.codec.binary else "wt") as out:
                yield out

    def run(self):
        with self.open_output() as out:
            for pathname in self.files:
                f = self.load_fattura(pathname)
                self.write(f, out)

    @classmethod
    def add_subparser(cls, subparsers):
        parser = super().add_subparser(subparsers)
        parser.add_argument(
            "-o", "--output", help="output file (default: standard output)"
        )
        parser.add_argument("files", nargs="+", help="input files (.xml or .xml.p7m)")
        return parser


class ExportJSON(Exporter):
    """
    output a fattura in JSON
    """

    NAME = "json"

    def get_codec(self) -> codec.Codec:
        if self.args.indent == "no":
            indent = None
        else:
            try:
                indent = int(self.args.indent)
            except ValueError:
                raise Fail("--indent argument must be an integer on 'no'")

        return codec.JSON(indent=indent)

    @classmethod
    def add_subparser(cls, subparsers):
        parser = super().add_subparser(subparsers)
        parser.add_argument(
            "--indent",
            default="1",
            help="indentation space (default: 1, use 'no' for all in one line)",
        )
        return parser


class ExportYAML(Exporter):
    """
    output a fattura in JSON
    """

    NAME = "yaml"

    def get_codec(self) -> codec.Codec:
        return codec.YAML()


class ExportXML(Exporter):
    """
    output a fattura in XML
    """

    NAME = "xml"

    def get_codec(self) -> codec.Codec:
        return codec.XML()


class ExportPython(Exporter):
    """
    output a fattura as Python code
    """

    NAME = "python"

    def get_codec(self) -> codec.Codec:
        namespace = self.args.namespace
        if namespace == "":
            namespace = False

        return codec.Python(namespace=namespace, unformatted=self.args.unformatted)

    @classmethod
    def add_subparser(cls, subparsers):
        parser = super().add_subparser(subparsers)
        parser.add_argument(
            "--namespace",
            default=None,
            help="namespace to use for the model classes (default: the module fully qualified name)",
        )
        parser.add_argument(
            "--unformatted",
            action="store_true",
            help="disable code formatting, outputting a single-line statement",
        )
        return parser


class Edit(App):
    """
    Open a fattura for modification in a text editor
    """

    def __init__(self, args):
        super().__init__(args)
        if self.args.style == "yaml":
            self.edit_codec = codec.YAML()
        elif self.args.style == "python":
            self.edit_codec = codec.Python(loadable=True)
        else:
            raise Fail(f"Unsupported edit style {self.args.style!r}")

    def write_out(self, f):
        """
        Write a fattura, as much as possible over the file being edited
        """
        codecs = codec.Codecs()
        codec_cls = codecs.codec_from_filename(self.args.file)
        if codec_cls == codec.P7M:
            with open(self.args.file[:-4], "wb") as fd:
                codec_cls().write_file(f, fd)
        elif codec_cls.binary:
            with open(self.args.file, "wb") as fd:
                codec_cls().write_file(f, fd)
        else:
            with open(self.args.file, "wt") as fd:
                codec_cls().write_file(f, fd)

    def run(self):
        f = self.load_fattura(self.args.file)
        f1 = self.edit_codec.interactive_edit(f)
        if f1 is not None and f != f1:
            self.write_out(f1)

    @classmethod
    def add_subparser(cls, subparsers):
        parser = super().add_subparser(subparsers)
        parser.add_argument(
            "-s",
            "--style",
            default="yaml",
            help="editable representation to use, one of 'yaml' or 'python'. Default: $(default)s",
        )
        parser.add_argument("file", help="file to edit")
        return parser


class Renderer(App):
    """
    Base class for CLI commands that render a Fattura
    """

    def __init__(self, args):
        from a38.render import HAVE_LXML

        if not HAVE_LXML:
            raise Fail("python3-lxml is needed for XSLT based rendering")

        super().__init__(args)
        self.stylesheet = args.stylesheet
        self.files = args.files
        self.output = args.output
        self.force = args.force

        from a38.render import XSLTTransform

        self.transform = XSLTTransform(self.stylesheet)

    def render(self, f, output: str):
        """
        Render the Fattura to the given destination file
        """
        raise NotImplementedError(
            self.__class__.__name__ + ".render is not implemented"
        )

    def run(self):
        for pathname in self.files:
            dirname = os.path.normpath(os.path.dirname(pathname))
            basename = os.path.basename(pathname)
            basename, ext = os.path.splitext(basename)
            output = self.output.format(dirname=dirname, basename=basename, ext=ext)
            if not self.force and os.path.exists(output):
                log.warning(
                    "%s: output file %s already exists: skipped", pathname, output
                )
            else:
                log.info("%s: writing %s", pathname, output)
            f = self.load_fattura(pathname)
            self.render(f, output)

    @classmethod
    def add_subparser(cls, subparsers):
        parser = super().add_subparser(subparsers)
        parser.add_argument(
            "-f", "--force", action="store_true", help="overwrite existing output files"
        )
        default_output = "{dirname}/{basename}{ext}." + cls.NAME
        parser.add_argument(
            "-o",
            "--output",
            default=default_output,
            help="output file; use {dirname} for the source file path,"
            " {basename} for the source file name"
            " (default: '" + default_output + "'",
        )
        parser.add_argument(
            "stylesheet", help=".xsl/.xslt stylesheet file to use for rendering"
        )
        parser.add_argument("files", nargs="+", help="input files (.xml or .xml.p7m)")
        return parser


class RenderHTML(Renderer):
    """
    render a Fattura as HTML using a .xslt stylesheet
    """

    NAME = "html"

    def render(self, f, output):
        html = self.transform(f)
        html.write(output)


class RenderPDF(Renderer):
    """
    render a Fattura as PDF using a .xslt stylesheet
    """

    NAME = "pdf"

    def __init__(self, args):
        super().__init__(args)
        self.wkhtmltopdf = shutil.which("wkhtmltopdf")
        if self.wkhtmltopdf is None:
            raise Fail("wkhtmltopdf is needed for PDF rendering")

    def render(self, f, output: str):
        self.transform.to_pdf(self.wkhtmltopdf, f, output)


class UpdateCAPath(App):
    """
    create/update an openssl CApath with CA certificates that can be used to
    validate digital signatures
    """

    NAME = "update_capath"

    def __init__(self, args):
        super().__init__(args)
        self.destdir = Path(args.destdir)
        self.remove_old = args.remove_old

    @classmethod
    def add_subparser(cls, subparsers):
        parser = super().add_subparser(subparsers)
        parser.add_argument("destdir", help="CA certificate directory to update")
        parser.add_argument(
            "--remove-old", action="store_true", help="remove old certificates"
        )
        return parser

    def run(self):
        from a38 import trustedlist as tl

        tl.update_capath(self.destdir, remove_old=self.remove_old)


class Allegati(App):
    """
    Show the attachments in the fattura
    """

    def __init__(self, args: argparse.Namespace) -> None:
        super().__init__(args)
        self.pathname = args.file
        self.ids: set[int] = set()
        self.globs: list[re.Pattern] = []
        self.has_filter = False
        for pattern in self.args.attachments:
            self.has_filter = True
            if pattern.isdigit():
                self.ids.add(int(pattern))
            elif pattern.startswith("^"):
                self.globs.append(re.compile(pattern))
            else:
                self.globs.append(re.compile(fnmatch.translate(pattern)))

    @classmethod
    def add_subparser(cls, subparsers):
        parser = super().add_subparser(subparsers)
        parser.add_argument(
            "--extract", "-x", action="store_true", help="extract selected attachments"
        )
        parser.add_argument(
            "--json", action="store_true", help="show attachments in json format"
        )
        parser.add_argument(
            "--yaml", action="store_true", help="show attachments in yaml format"
        )
        parser.add_argument(
            "--output",
            "-o",
            action="store",
            help="destination file name (-o file) or directory (-o dir/)",
        )
        parser.add_argument("file", help="input file (.xml or .xml.p7m)")
        parser.add_argument(
            "attachments",
            nargs="*",
            help="IDs or names of attachments to extract. Shell-like wildcards allowed, or regexps if starting with ^",
        )
        return parser

    def match_allegato(self, index: int, allegato: fattura.Allegati) -> bool:
        """
        Check if the given allegato matches the attachments patterns
        """
        if not self.has_filter:
            return True

        for id in self.ids:
            if index == id:
                return True

        for regex in self.globs:
            if regex.match(allegato.nome_attachment):
                return True

        return False

    def print_allegato(self, index: int, allegato: fattura.Allegati) -> None:
        formato = allegato.formato_attachment or "-"
        print(f"{index:02d}: {formato} {allegato.nome_attachment}")
        if allegato.descrizione_attachment:
            print(f"    {allegato.descrizione_attachment}")

    def run(self):
        f = self.load_fattura(self.pathname)
        selected: list[tuple[int, fattura.Allegati]] = []
        index = 1
        for body in f.fattura_elettronica_body:
            for allegato in body.allegati:
                if self.match_allegato(index, allegato):
                    selected.append((index, allegato))
                index += 1

        if self.args.json or self.args.yaml:
            output = []
            for index, allegato in selected:
                jsonable = {"index": index}
                jsonable.update(allegato.to_jsonable())
                jsonable.pop("attachment", None)
                output.append(jsonable)

            if self.args.json:
                import json

                json.dump(output, sys.stdout, indent=2)
                print()
            else:
                import yaml

                yaml.dump(
                    output,
                    stream=sys.stdout,
                    default_flow_style=False,
                    sort_keys=False,
                    allow_unicode=True,
                    explicit_start=True,
                    Dumper=yaml.CDumper,
                )
        elif self.args.extract:
            destname: Optional[str]
            destdir: str

            if self.args.output:
                if os.path.isdir(self.args.output) or self.args.output.endswith(os.sep):
                    destname = None
                    destdir = self.args.output
                else:
                    destname = self.args.output
                    destdir = "."
            else:
                destname = None
                destdir = "."

            if destname is not None and len(selected) > 1:
                raise Fail(
                    "there are multiple attachment to save, and--output points to a single file name"
                )

            os.makedirs(destdir, exist_ok=True)

            for index, allegato in selected:
                if destname is None:
                    destname = os.path.basename(allegato.nome_attachment)
                dest = os.path.join(destdir, destname)
                log.info("Extracting %s to %s", allegato.nome_attachment, dest)
                with open(dest, "wb") as fd:
                    fd.write(allegato.attachment)
        else:
            for index, allegato in selected:
                self.print_allegato(index, allegato)


def main():
    parser = argparse.ArgumentParser(description="Handle fattura elettronica files")
    parser.add_argument("--verbose", "-v", action="store_true", help="verbose output")
    parser.add_argument("--debug", action="store_true", help="debug output")

    subparsers = parser.add_subparsers(help="actions", required=True)
    subparsers.dest = "command"

    ExportJSON.add_subparser(subparsers)
    ExportYAML.add_subparser(subparsers)
    ExportXML.add_subparser(subparsers)
    ExportPython.add_subparser(subparsers)
    Edit.add_subparser(subparsers)
    Diff.add_subparser(subparsers)
    Validate.add_subparser(subparsers)
    RenderHTML.add_subparser(subparsers)
    RenderPDF.add_subparser(subparsers)
    UpdateCAPath.add_subparser(subparsers)
    Allegati.add_subparser(subparsers)

    args = parser.parse_args()

    log_format = "%(levelname)s %(message)s"
    level = logging.WARN
    if args.debug:
        level = logging.DEBUG
    elif args.verbose:
        level = logging.INFO
    logging.basicConfig(level=level, stream=sys.stderr, format=log_format)

    app = args.app(args)
    res = app.run()
    if isinstance(res, int):
        sys.exit(res)


if __name__ == "__main__":
    try:
        main()
    except Fail as e:
        print(e, file=sys.stderr)
        sys.exit(1)
    except Exception:
        log.exception("uncaught exception")