invoice/lib/invoice/cli.py

#!/usr/bin/python3
# encoding: utf-8
from __future__ import print_function

import os, sys, argparse, datetime, subprocess
import locale
import logging
logging.basicConfig(level=logging.DEBUG)
log = logging.getLogger()

import invoice.db

class SanityCheckError(Exception):
    pass

class Application:
    my_company = "my-company"

    editor = os.environ.get("EDITOR") or "vim"
    viewer = os.environ.get("PAGER") or "less"
    tex_program = "pdflatex"
    pdf_program = "xdg-open"

    def __init__(self, template_path):
        self._parse_args()
        self._load_config()
        self.year = self.args.__dict__.pop("year")
        self.user_path = os.path.expanduser(self.args.__dict__.pop("user_data"))
        self.method = self.args.__dict__.pop("method")
        self.data_path = os.path.join(self.user_path, "{year}", "data", "{directory}")
        self.tmp_path = os.path.join(self.user_path, "tmp")
        self.output_path =  os.path.join(self.user_path, "{year}", "output")
        self.template_path = template_path
        self.db = invoice.db.Database(
            year = self.year,
            data_path = self.data_path)

    def _load_config(self):
        config_dir = os.path.expanduser(self.args.user_data)
        config_file = os.path.join(config_dir, "config")

        if not os.path.exists(config_dir):
            os.mkdir(config_dir)
        if not os.path.exists(config_file):
            with open(config_file, "w") as f:
                pass

        exec(open(config_file).read(),
            {"__builtins__": None}, self.__dict__)

    def _parse_args(self):
        parser = argparse.ArgumentParser(
            description = "Pavel Šimerda's invoice CLI application.",
            conflict_handler = "resolve")
        parser.add_argument("--year", "-y", action="store")
        parser.add_argument("--user-data", "-d", action="store")
        parser.add_argument("--debug", "-D", action="store_const", dest="log_level", const=logging.DEBUG)
        #parser.add_argument("--verbose", "-v", action="store_const", dest="log_level", const=logging.INFO)
        #parser.add_argument("--config", "-C", action="store")
        parser.set_defaults(
            year = datetime.date.today().year,
            user_data = "~/.invoice",
            log_level = logging.INFO)

        subparsers = parser.add_subparsers(title="subcommands",
            description="valid subcommands",
            help="additional help")

        for list_ in "invoices", "companies":
            for action in "list", "summary", "new", "edit", "paid", "show", "pdf", "delete":
                if action in ("pdf", "paid", "summary") and list_ != "invoices":
                    continue
                suffix = ''
                if list_ == "companies":
                    suffix = "-companies" if action=="list" else "-company"
                method = getattr(self, "do_"+(action+suffix).replace("-", "_"))
                subparser = subparsers.add_parser(action+suffix, help=method.__doc__)
                if method == self.do_pdf:
                    subparser.add_argument("--generate", "-g", action="store_true")
                if action == "delete":
                    subparser.add_argument("--force", "-f", action="store_true")
                if action == "new":
                    subparser.add_argument("name" if suffix else "company_name")
                if action in ("show", "pdf", "edit", "paid", "delete"):
                    subparser.add_argument("selector", nargs="?")
                if action == "paid":
                    subparser.add_argument("date")
                subparser.set_defaults(method=method)

        self.args = parser.parse_args()
        log.setLevel(self.args.__dict__.pop("log_level"))
        log.debug("Arguments: {0}".format(self.args))

    def run(self):
        try:
            self.method(**vars(self.args))
        except (SanityCheckError) as error:
            print("Error: {0} Use '--force' to suppress this check.".format(error), file=sys.stderr)
            if log.isEnabledFor(logging.DEBUG):
                raise
        except invoice.db.DatabaseError as error:
            print("Error: {0}".format(error), file=sys.stderr)
            if log.isEnabledFor(logging.DEBUG):
                raise

    def do_list(self):
        """List invoices."""
        for item in sorted(self.db.invoices):
            print(item)

    def do_summary(self):
        """Show invoice summary."""
        total = paid = 0
        for invoice in sorted(self.db.invoices):
            data = invoice.data()
            log.debug(data._data)
            print("{number:7} {date!s:10} {due!s:10} {paid!s:10} {sum:>6} {company_name}"
                .format(**data._data))
            total += data.sum
            if data.paid:
                paid += data.sum

        print()
        print("Total: {0:6}".format(total))
        print("Paid:  {0:6}".format(paid))
        print("Unpaid: {0:6}".format(total-paid))

    def do_new(self, company_name):
        """Create and edit a new invoice."""
        item = self.db.invoices.new(company_name)
        self._edit(item._path)

    def do_edit(self, selector):
        """Edit invoice in external editor.

        The external editor is determined by EDITOR environment variable
        using 'vim' as the default. Item is edited in-place.
        """
        self._edit(self.db.invoices[selector]._path)

    def do_paid(self, selector, date):
        path = self.db.invoices[selector]._path
        with open(path, "a") as stream:
            stream.write("Paid: {0}\n".format(date))
        self._show(path)

    def _edit(self, path):
        log.debug("Editing file: {0}".format(path))
        assert os.path.exists(path)
        subprocess.call((self.editor, path))

    def do_show(self, selector):
        """View invoice in external viewer.

        The external viewer is determined by PAGER environment variable
        using 'less' as the default.
        """
        item = self.db.invoices[selector]
        self._show(item._path)

    def do_pdf(self, selector, generate):
        """Generate and view a PDF invoice.

        This requires Tempita 0.5.
        """
        import tempita
        invoice = self.db.invoices[selector]

        tmp_path = self.tmp_path.format(year=self.year)
        output_path = self.output_path.format(year=self.year)
        log.debug("tmp_path={0}".format(tmp_path))

        tex_template = os.path.join(self.template_path, "invoice.tex")
        tex_file = os.path.join(tmp_path, "{0}.tex".format(invoice._name))
        tmp_pdf_file = os.path.join(tmp_path, "{0}.pdf".format(invoice._name))
        pdf_file = os.path.join(output_path, "{0}.pdf".format(invoice._name))

        if generate:
            #if(not os.path.exists(pdf_file) or
            #        os.path.getmtime(invoice._path) > os.path.getmtime(pdf_file)):
            issuer = self.db.companies[self.my_company]
            customer = self.db.companies[invoice.company_name]

            invoice_data = invoice.data()
            issuer_data = issuer.data()
            customer_data = customer.data()

            log.debug("Invoice: {0}".format(invoice_data._data))
            log.debug("Issuer: {0}".format(issuer_data._data))
            log.debug("Customer: {0}".format(customer_data._data))

            log.debug("Creating TeX invoice...")
            self._ensure_directory(self.tmp_path)

            format_decimal=lambda x: '{:20,.2f}'.format(x).replace(',', '\\,').replace('.', ',')
            format_eur=lambda x: format_decimal(x) + ' EUR'

            result = tempita.Template(open(tex_template).read()).substitute(
                invoice=invoice_data,
                issuer=issuer_data,
                customer=customer_data,
                decimal=format_decimal,
                eur=format_eur)
            open(tex_file, "w").write(str(result))
            assert(os.path.exists(tex_file))

            log.debug("Creating PDF invoice...")
            if subprocess.call((self.tex_program, "{0}.tex".format(invoice._name)), cwd=tmp_path) != 0:
                raise GenerationError("PDF generation failed.")
            assert(os.path.exists(tmp_pdf_file))

            log.debug("Moving PDF file to the output directory...")
            self._ensure_directory(output_path)
            os.rename(tmp_pdf_file, pdf_file)

        assert(os.path.exists(pdf_file))
        #log.debug("Running PDF viewer...")
        #subprocess.call((self.pdf_program, pdf_file))

    def _check_path(self, path):
        if not os.path.exists(path):
            raise LookupError("Directory doesn't exist: {0}".format(path))

    def _ensure_directory(self, path):
        try:
            self._check_path(path)
        except LookupError as e:
            os.makedirs(path, exist_ok=True)

    def do_delete(self, selector, force):
        """List invoices."""
        if selector:
            invoice = self.db.invoices[selector]
        else:
            invoice = self.db.invoices.last()
        if not force:
            raise SanityCheckError("It is not recommended to delete invoices.")
        invoice.delete()

    def do_list_companies(self):
        """List companies."""
        for item in sorted(self.db.companies):
            print(item)

    def do_new_company(self, name):
        """Create and edit a new company."""
        item = self.db.companies.new(name)
        self._edit(item._path)

    def do_edit_company(self, selector):
        """Edit company in external editor.

        The external editor is determined by EDITOR environment variable
        using 'vim' as the default. Item is edited in-place.
        """
        item = self.db.companies[selector]
        self._edit(item._path)

    def do_show_company(self, selector):
        """View company in external viewer.

        The external viewer is determined by PAGER environment variable
        using 'less' as the default.
        """
        item = self.db.companies[selector]
        print("# {0}".format(item._name))
        self._show(item._path)

    def _show(self, path):
        log.debug("Viewing file: {0}".format(path))
        assert os.path.exists(path)
        print("# {0}".format(path))
        subprocess.call((self.viewer, path))

    def do_delete_company(self, selector, force):
        """Delete a company."""
        company = self.db.companies[selector]
        if not force:
            invoices = self.db.invoices.select({"company_name": company._name})
            if invoices:
                for invoice in invoices:
                    log.info("Dependent invoice: {0}".format(invoice))
                raise SanityCheckError("This company is used by some invoices. You should not delete it.")
        company.delete()