path: root/fracture/invoice.py

import json
import sqlite3
from itertools import groupby
from datetime import date, datetime, timedelta
from configparser import ConfigParser
from collections import OrderedDict
import io

class Conf:
    # MOVE ALL THE CONFIG HERE
    CURRENCY  = None
    CURRENCY_DECIMAL = 0
    @classmethod
    def round(cls, num):
        return round(num, cls.CURRENCY_DECIMAL)


class Tax:
    def __init__(self, name="", ratio = 0.0):
        self.name  = name.upper()
        self.ratio = ratio
    def to_dict(self):
        return {
            "name": self.name,
            "ratio": self.ratio,
        }

class Product:
    VATS = ()
    def __init__(self, description, units=1.0,
            price_unit=0.0, vat = 0.21):
        if len(description) == 0:
            raise ValueError("Product: No description provided")
        self.description = description
        self.units       = units
        self.price_unit  = price_unit
        if vat in self.VATS:
            self.vat     = vat
        else:
            raise ValueError("Product: No valid VAT. Valid are:"+\
                    str(Product.VATS))
    @property
    def total(self):
        return self.base+self.charged_vat
    @property
    def base(self):
        return self.price_unit * self.units
    @property
    def charged_vat(self):
        return self.base * self.vat
    def to_dict(self):
        return {
            "description": self.description,
            "units":       self.units,
            "price-unit":  Conf.round(self.price_unit),
            "vat":         self.vat,
            "vat-charged": Conf.round(self.charged_vat),
            "base":        Conf.round(self.base),
            "total":       Conf.round(self.total)
        }

class Customer:
    def __init__(self, name, id = "", address = ""):
        if len(name) == 0:
            raise ValueError("Customer: No name provided")
        self.name    = name
        self.id      = id
        self.address = address

class Invoice:
    _TYPES  = {"sent", "received"}
    SERIES = {}
    ID_FORMAT = None
    DB_FILE   = None
    DEFAULT_TAXES = ()

    DEFAULT_PRODUCT_DESC  = "Empty Product"
    DEFAULT_CUSTOMER_DESC = "Empty Customer"

    def __init__(self, type = None,
                       series = None,
                       idate = None,
                       notes= None,
                       products = None,
                       customer = None,
                       taxes = None,
                       number = None):
        # Initializes to empty state if not set
        self.set_type(type or tuple(Invoice._TYPES)[0])
        self.number   = number
        self.set_series(series or tuple(Invoice.SERIES.keys())[0])
        self.date     = idate or date.today()
        self.notes    = notes or ""
        self.products = products or (Product(Invoice.DEFAULT_PRODUCT_DESC),
                                     Product(Invoice.DEFAULT_PRODUCT_DESC))
        self.customer = customer or Customer(Invoice.DEFAULT_CUSTOMER_DESC)
        self.taxes    = taxes or Invoice.DEFAULT_TAXES

    def set_series(self, series):
        """ Series is an integer """
        if series not in Invoice.SERIES.keys():
            raise ValueError ("Not valid series for Invoice. Valid are %s"
                    % Invoice.SERIES)
        self.series = series

    def set_type(self, type):
        if type not in Invoice._TYPES:
            raise ValueError ("Not valid type for Invoice. Valid are: %s"
                    % Invoice._TYPES)
        self.type = type

    @property
    def id(self):
        return Invoice.ID_FORMAT(self.series, self.date, self.number)

    def __repr__(self):
        return "<Invoice object>\n\t"+\
                self.to_config().replace("\n", "\n\t") + \
               "\n</Invoice object>"
    def to_config(self):
        strf = io.StringIO()
        cfg = ConfigParser()
        cfg["invoice"] = {
                "series": self.series,
                "date": self.date,
                "type": self.type,
                "notes": self.notes,
                "customer-name": self.customer.name,
                "customer-address": self.customer.address,
                "customer-id": self.customer.id,
                }
        for i,p in enumerate(self.products):
            cfg["product-"+str(i)] = {
                    "description": p.description,
                    "units": p.units,
                    "price_unit": p.price_unit,
                    "vat": p.vat
                    }

        cfg["taxes"] = { x.name: x.ratio for x in self.taxes }

        cfg.write(strf)
        return strf.getvalue()


    @classmethod
    def from_config(cls, config):
        cfg = ConfigParser()
        cfg.read_string(config)

        # PRODUCTS SECTIONS
        products = ()
        for s in cfg.sections():
            if not s.startswith("product"):
                continue
            desc = cfg[s]["description"]
            un   = float(cfg[s]["units"])
            pu   = float(cfg[s]["price_unit"])
            vat  = float(cfg[s]["vat"])
            products += ( Product(desc,un,pu,vat) ,)
            if cfg[s]["description"] == cls.DEFAULT_PRODUCT_DESC:
                raise ValueError("Product name not set")
        if len(products) == 0:
            raise ValueError("No products assigned")

        # TAXES SECTION
        taxes = ()
        if cfg.has_section("taxes"):
            for x in cfg["taxes"]:
                taxes += ( Tax(x, float(cfg["taxes"][x])) ,)

        # INVOICE SECTION
        if not cfg.has_section("invoice"):
            raise ValueError("[invoice] section needed")
        i        = cfg["invoice"]
        series   = int(i["series"])
        date     = datetime.strptime(i["date"], "%Y-%m-%d").date()
        notes    = i["notes"]
        type     = i["type"]
        customer = Customer( i["customer-name"],
                             i["customer-id"],
                             i["customer-address"])
        if i["customer-name"] == cls.DEFAULT_CUSTOMER_DESC:
            raise ValueError("Customer name not set")

        return cls(
                type,
                series,
                date,
                notes,
                products,
                customer,
                taxes)


    @classmethod
    def load(cls, id, type="sent", debug=False):
        with sqlite3.connect(cls.DB_FILE) as conn:
            if debug:
                conn.set_trace_callback(print)
            conn.row_factory = sqlite3.Row

            c = conn.cursor()

            # PRODUCTS
            products = ()
            c.execute("""SELECT * FROM products WHERE invoice_id = ?
                    AND invoice_type = ?""", (id, type))
            res = c.fetchone()
            while res is not None:
                desc = res["description"]
                un   = res["units"]
                pu   = res["price_unit"]
                vat  = res["vat"]
                products += (Product(desc,un,pu,vat),)
                res = c.fetchone()

            # TAXES
            taxes = ()
            c.execute("""SELECT * FROM taxes WHERE invoice_id = ?
                    AND invoice_type = ?""", (id, type))
            res = c.fetchone()
            while res is not None:
                taxes +=  (Tax(res["name"], res["ratio"]) ,)
                res = c.fetchone()

            # INVOICE
            c.execute("""SELECT * FROM invoices WHERE id = ?
                    AND type = ?""", (id, type))
            res = c.fetchone()
            return cls(
                number   = res["number"],
                type     = res["type"],
                series   = res["series"],
                idate    = datetime.strptime(res["date"],"%Y-%m-%d").date(),
                notes    = res["notes"],
                products = products,
                customer = Customer( res["customer_name"],
                                     res["customer_id"],
                                     res["customer_address"] ),
                taxes    = taxes)

    @classmethod
    def load_many(cls, cond="", vals=(), debug=False):
        with sqlite3.connect(cls.DB_FILE) as conn:
            if debug:
                conn.set_trace_callback(print)

            conn.row_factory = sqlite3.Row

            c = conn.cursor()
            c.execute(("SELECT id, type FROM invoices " + cond), vals)
            return tuple( cls.load( r["id"], r["type"] ) \
                            for r in c.fetchall() )

    @classmethod
    def latest_number(cls, series, type, debug=False):
        with sqlite3.connect(cls.DB_FILE) as conn:
            if debug:
                conn.set_trace_callback(print)

            conn.row_factory = sqlite3.Row

            c = conn.cursor()
            c.execute("""SELECT number FROM invoices
                WHERE series = ?  AND type = ?
                ORDER BY(number) DESC LIMIT 1""", (series, type))
            res = c.fetchall()

            latest_number = 0
            if len(res) != 0:
                latest_number = int(res[0]["number"])
            return latest_number

    @classmethod
    def load_by_date(cls, year, quarter=None):
        dbegin = date(year,   1, 1)
        dend   = date(year+1, 1, 1)
        if quarter is not None:
            month    = (quarter - 1) * 3 + 1
            dbegin   = date(year, month, 1)
            endyear  = year
            endmonth = month + 4
            if endmonth > 12:
                endyear  = year+1
                endmonth = month
            dend = date(endyear, endmonth, 1)
        return cls.load_many(
            """ WHERE
                strftime('%s',?) <= strftime('%s',date)
                AND
                strftime('%s',date) < strftime('%s',?)
            """,
            (dbegin, dend))


    def persist(self):
        conn = sqlite3.connect(Invoice.DB_FILE)
        if self.number is None:
            self.number = self.latest_number(self.series, self.type) + 1
        c = conn.cursor()
        c.execute("""INSERT INTO invoices (
                    type,
                    series,
                    number,
                    id,
                    date,
                    notes,
                    customer_id,
                    customer_name,
                    customer_address
                 ) VALUES (?,?,?,?,?,?,?,?,?)""", (
                    self.type,
                    self.series,
                    self.number,
                    self.id,
                    self.date.strftime("%Y-%m-%d"),
                    self.notes,
                    self.customer.id,
                    self.customer.name,
                    self.customer.address
                ))
        for p in self.products:
            c.execute("""INSERT INTO products (
                        invoice_id,
                        invoice_type,
                        description,
                        units,
                        price_unit,
                        vat
                     ) VALUES (?,?,?,?,?,?)""",(
                         self.id,
                         self.type,
                         p.description,
                         p.units,
                         p.price_unit,
                         p.vat
                    ))
        for x in self.taxes:
            c.execute("""INSERT INTO taxes (
                        invoice_id,
                        invoice_type,
                        name,
                        ratio
                     ) VALUES (?,?,?,?)""", (
                         self.id,
                         self.type,
                         x.name,
                         x.ratio))

        conn.commit()
        conn.close()
        return (self.id, self.type)


    @property
    def vat_charged(self):
        return sum(p.charged_vat for p in self.products)
    @property
    def base(self):
        return sum(p.base for p in self.products)
    @property
    def total(self):
        return sum(p.total           for p in self.products) + \
               sum(self.base*t.ratio for t in self.taxes)

    def to_dict(self):
        return {
            "products":          tuple(p.to_dict() for p in self.products),
            "taxes":             tuple(dict(t.to_dict(), applied=Conf.round(self.base*t.ratio)) for t in self.taxes),
            "type":              self.type,
            "id":                self.id,
            "date":              self.date.strftime("%Y-%m-%d"),
            "customer":         {
                "id":       self.customer.id,
                "name":     self.customer.name,
                "address":  self.customer.address,
            },
            "base":              self.base,
            "notes":             self.notes,
            "total":             Conf.round(self.total),
            "vat-charged":       self.vat_charged
        }
    def to_json(self):
        return json.dumps(self.to_dict())

    def to_row(self):

        row =  OrderedDict(
            type =              self.type,
            id =                self.id,
            date =              self.date.strftime("%Y-%m-%d"),
            customer_id =       self.customer.id,
            customer_name =     self.customer.name,
            base_imp_total =    Conf.round(self.base),
            total =             Conf.round(self.total),
            )


        to_base_key =  lambda v: "base("+str(v*100)+"%)"
        to_vat_key  =  lambda v: "VAT("+str(v*100)+"%)"
        pvat  =        lambda p: p.vat

        vs = groupby(sorted(self.products, key=pvat), key=pvat)

        for vat, ps in vs:
            ps = tuple(ps)
            row[to_base_key(vat)] = Conf.round(sum(p.base        for p in ps))
            row[to_vat_key(vat)]  = Conf.round(sum(p.charged_vat for p in ps))


        to_tax_key = lambda tax: tax.name + "("+ str(tax.ratio*100) +"%)"
        for i in self.taxes:
            tax_amount = i.ratio * self.base
            row[to_tax_key(i)] = round(tax_amount, Conf.CURRENCY_DECIMAL)
        return row