himport/himports/dolibarrAlchemyHledger.py

# -*- coding: utf-8 -*-
from __future__ import unicode_literals

import settings
import datetime

from himports.dolibarrAlchemy import DolibarrSQLAlchemy


#
# HledgerEntry : Base class for an hledger entry
#
class HledgerEntry(object):
    accounting_years = settings.get('ACCOUNTING_YEARS')
    pc_default_tier = settings.get('PC_REFS')['default_tier']
    pc_default_client = settings.get('PC_REFS')['default_client']
    pc_default_supplier = settings.get('PC_REFS')['default_supplier']
    pc_default_income = settings.get('PC_REFS')['default_income']
    pc_default_expense = settings.get('PC_REFS')['default_expense']
    pc_default_bank = settings.get('PC_REFS')['default_bank']

    tva_type = settings.get('TVA_TYPE')

    # the sql class need to be defined in the derived classes
    def _sql_class(self):
        assert(False)

    # the sql_class corresponding the hledger class
    sql_class = property(_sql_class)

    # Date defining the current accounting year
    k_accounting_date = None

    def __init__(self, dolibarr_alchemy, e):
        super(HledgerEntry, self).__init__()
        self.dolibarr_alchemy = dolibarr_alchemy
        self.e = e
        self.accounting_date = e
        for attr in self.k_accounting_date.split('.'):
            self.accounting_date = getattr(self.accounting_date, attr)

    # Retrieve all entries corresponding to cls.sql_class
    @classmethod
    def get_entries(cls, dolibarr_alchemy):
        return [cls(dolibarr_alchemy, i) for i in dolibarr_alchemy.session.query(cls.sql_class).all()]

    # get_ledger : Print ledger output. Only defined in derived class
    def get_ledger(self):
        print("WARNING: get_ledger not done")
        return u""

    # check_pc : verify the accounting chart corresponding to the entry. Only defined in derived class.
    def check_pc(self):
        return ()

    # get_year: return the year corresponding to the current entry. Only defined in derived class.
    def get_year(self):
        raise Exception("TODO: get_year not implemented for class %s" % (self.__class__))

    # get_accounting_year: return the accouting year corresponding to the current entry.
    def get_accounting_year(self):
        date = self.accounting_date
        if isinstance(date, datetime.datetime):
            date = date.date()
        for (year, dbegin, dend) in HledgerEntry.accounting_years:
            if date >= dbegin and date <= dend:
                return year

        return str(date.year)

    # _value: return the value in float with 4 digits
    @staticmethod
    def _value(value):
        return '{number:.{digits}f}'.format(number=value, digits=4)


#
# HledgerJournal: A complete Hledger journal. This is a base class.
#
class HledgerJournal(object):
    def __init__(self, dolibarr_alchemy, cls_entry):
        self.dolibarr_alchemy = dolibarr_alchemy
        self.entries = cls_entry.get_entries(dolibarr_alchemy)

    # get_entries: return the journal entries
    def get_entries(self):
        return self.entries

    # get_by_year: return the journal entries by accounting year.
    def get_by_year(self):
        by_year = {}
        for entry in self.get_entries():

            entry_year = entry.get_accounting_year()
            if entry_year not in by_year:
                by_year[entry_year] = []
            by_year[entry_year].append(entry)
        return by_year

    # check_pc: verify there is an account for earch entry on the current journal
    def check_pc(self):
        pc_missing = set()
        for entry in self.get_entries():
            pc_missing.update(entry.check_pc())

        return pc_missing


#
# HledgerBankEntry: a bank entry
#
class HledgerBankEntry(HledgerEntry):
    k_accounting_date = 'datev'

    def _sql_class(self):
        return self.dolibarr_alchemy.Bank

    # get_entries: return the bank entries ordered by value date
    @classmethod
    def get_entries(cls, dolibarr_alchemy):
        Bank = dolibarr_alchemy.Bank
        entries = dolibarr_alchemy.session.query(Bank).order_by(Bank.datev, Bank.num_releve).all()
        return [cls(dolibarr_alchemy, e) for e in entries]

    # get_third_code: retrieve the third code corresponding to the entry. It could be
    #       a supplier payment
    #       a tax payment
    #       a customer payment
    #       a value-added tax payment
    #       any payment defined in the function PC_REFS['fn_custom_codes']
    @classmethod
    def get_third_code(cls, e):
        third_code = ""
        if e.url_payment_supplier:
            if e.url_company:
                third_code = e.url_company.societe.code_compta_fournisseur

        if e.url_payment_sc:
            code = e.url_payment_sc.payment_sc.cotisation_sociale.type.code
            if code in settings.get('SOCIAL_REFS'):
                third_code = settings.get('SOCIAL_REFS')[code]

        if e.url_payment:
            if e.url_company:
                third_code = e.url_company.societe.code_compta

        if e.payment_tva:
            third_code = settings.get('PC_REFS')['tva_a_decaisser']

        if third_code == "":
            fns = settings.get('PC_REFS')['fn_custom_codes']
            for fn in fns:
                try:
                    third_code = fn(e)
                    if third_code is None or not isinstance(third_code, basestring):
                        third_code = ""
                except:
                    third_code = ""
                if third_code != "":
                    break

        if third_code == "":
            third_code = cls.pc_default_tier

        return third_code

    # get_description: retrieve the description of the payment
    @classmethod
    def get_description(self, e):
        s_nom = ""
        s_description = ""

        if e.url_company:
            s_nom = e.url_company.societe.nom

        if e.url_payment_supplier:
            f_ids = [f.facture.ref_supplier for f in e.url_payment_supplier.payment_supplier.factures]
            s_description = "Règlement facture fournisseur - %s - %s" % (
                s_nom,
                "|".join(f_ids),
            )
        if e.url_payment:
            f_ids = [f.facture.facnumber for f in e.url_payment.payment.factures]
            s_description = "Règlement facture client - %s - %s" % (
                s_nom,
                "|".join(f_ids),
            )

        if s_description == "":
            s_description = s_nom + " - " + e.label

        return s_description

    # get_ledger: see @HledgerEntry.get_ledger
    def get_ledger(self):
        e = self.e
        s = ""

        s_description = self.get_description(self.e)

        s += "%(date)s    %(description)s\n" % {
            'date': e.datev.strftime("%Y/%m/%d"),
            'description': s_description
        }

        third_code = self.get_third_code(self.e)

        ba_code = e.account.account_number
        if ba_code == "":
            ba_code = self.pc_default_bank

        s += "    %(account)s \t %(amount)s\n" % {
            'account': settings.get_ledger_account(ba_code),
            'amount': self._value(-e.amount)
        }
        s += "    %(account)s \t %(amount)s\n" % {
            'account': settings.get_ledger_account(third_code),
            'amount': self._value(e.amount)
        }

        if e.url_payment_supplier:
            for f in e.url_payment_supplier.payment_supplier.factures:
                tvas = HledgerSupplierEntry.get_tva_payment_amounts(self.dolibarr_alchemy, f.facture, journal="bank")
                for k in tvas:
                    s += "    %(account_tva)s \t %(amount_tva)s\n" % {
                        'account_tva': settings.get_ledger_account(k),
                        'amount_tva': self._value(tvas[k] * (f.amount / f.facture.total_ttc))
                    }
        elif e.url_payment:
            for f in e.url_payment.payment.factures:
                tvas = HledgerSellEntry.get_tva_payment_amounts(self.dolibarr_alchemy, f.facture, journal="bank")
                for k in tvas:
                    s += "    %(account_tva)s \t %(amount_tva)s\n" % {
                        'account_tva': settings.get_ledger_account(k),
                        'amount_tva': self._value(tvas[k] * (f.amount / f.facture.total_ttc))
                    }
        else:
            pass

        return s


#
# HledgerBillingEntry: An entry corresponding to a bill (Supplier or Client)
#
class HledgerBillingEntry(HledgerEntry):

    # is_tva_facture: return if the value added tax must be processed on the billing date
    @classmethod
    def is_tva_facture(cls, ed):
        return ed.productcls.tva_type == 'service_sur_debit' and ed.product_type == 1

    # is_tva_paiement: return if the value-added tax must be processed on the payment date
    @classmethod
    def is_tva_paiement(cls, ed):
        return cls.tva_type != 'service_sur_debit' or ed.product_type != 1

    # get_tva_amounts: return the amount of value-added taxes.
    @classmethod
    def get_tva_amounts(cls, dolibarr_alchemy, e, journal):

        tvas = dict()
        for ed in e.details:
            if isinstance(e, dolibarr_alchemy.Facture):
                total_tva = -ed.total_tva
            elif isinstance(e, dolibarr_alchemy.FactureFourn):
                total_tva = ed.tva
            else:
                raise Exception("Should be either Facture or FactureFourn")

            if total_tva == 0:
                continue

            tva_account = cls.get_tva_account(ed)
            tva_regul = cls.get_tva_regul_account(ed)

            if journal == "bank":
                if ed.product_type == 1 and cls.tva_type == 'standard':
                    if tva_regul not in tvas:
                        tvas[tva_regul] = 0
                    if tva_account not in tvas:
                        tvas[tva_account] = 0
                    tvas[tva_account] += -total_tva
                    tvas[tva_regul] += total_tva

            elif journal == "sell" or journal == "supplier":
                if ed.product_type == 1 and cls.tva_type == 'standard':
                    if tva_regul not in tvas:
                        tvas[tva_regul] = 0
                    tvas[tva_regul] += -total_tva
                else:
                    if tva_account not in tvas:
                        tvas[tva_account] = 0
                    tvas[tva_account] += -total_tva

        return tvas

    # get_tva_regul_account: retourn the reference corresponding to the
    #                        value-added tax regulation account
    @classmethod
    def get_tva_regul_account(cls, ed):
        tx = int(float(ed.tva_tx) * 100)

        key = "tva_regul_%s" % (tx,)

        return settings.get('PC_REFS')[key]

    # get_tva_billing_amounts: return the value-added tax amount to collect
    @classmethod
    def get_tva_billing_amounts(cls, dolibarr_alchemy, e, journal):
        return cls.get_tva_amounts(dolibarr_alchemy, e, journal)

    # get_tva_payment_amounts: return value-added tax amount to deduce
    @classmethod
    def get_tva_payment_amounts(cls, dolibarr_alchemy, e, journal):
        return cls.get_tva_amounts(dolibarr_alchemy, e, journal)


#
# HledgerSupplierEntry: Billing entry corresponding to a supplier
#
class HledgerSupplierEntry(HledgerBillingEntry):

    k_accounting_date = 'datef'

    def _sql_class(self):
        return self.dolibarr_alchemy.FactureFourn

    # get_entries: return the bill entries ordered by billing date
    @classmethod
    def get_entries(cls, dolibarr_alchemy):
        FactureFourn = dolibarr_alchemy.FactureFourn
        return [cls(dolibarr_alchemy, e) for e in dolibarr_alchemy.session.query(FactureFourn).order_by(FactureFourn.datef).all()]

    # check: check if the entry is coherent
    def check(self):
        e = self.e

        total_ttc = e.total_ttc
        total_tva = e.total_tva
        total_ht = e.total_ht

        for ed in e.details:
            total_ttc -= ed.total_ttc
            total_tva -= ed.tva
            total_ht -= ed.total_ht

        if total_ttc > 1e-10:
            print("Erreur dans l'écriture %s: total ttc = %s" % (e.ref_supplier, total_ttc))
        if total_ht > 1e-10:
            print("Erreur dans l'écriture %s: total ht = %s" % (e.ref_supplier, total_ht))
        if total_tva > 1e-10:
            print("Erreur dans l'écriture %s: total tva = %s" % (e.ref_supplier, total_tva))

    # getMissingPC: retrieve missing accounts
    def getMissingPC(self):
        e = self.e
        pc_missing = []
        if e.societe.code_compta_fournisseur == "":
            pc_missing.append("tiers:fournisseur: %s %s" % (e.societe.nom, e.societe.ape))

        for ed in e.details:
            if self.get_product_account_code(ed) == self.pc_default_expense:
                pc_missing.append("achat: %s - %s" % (e.ref_supplier, ed.description.splitlines()[0]))

        return pc_missing

    # get_ledger: return the corresponding ledger entries
    def get_ledger(self):
        e = self.e
        self.check()

        s = ""
        s += "%(date)s  %(description)s\n" % {
            'date':  e.datef.strftime("%Y/%m/%d"),
            'description': e.ref_supplier + " - " + e.societe.nom,
        }

        s_code = self.get_supplier_code(self.e)
        s += "    %(compte_tiers)s \t %(amount_ttc)s\n" % {
            'compte_tiers': settings.get_ledger_account(s_code),
            'amount_ttc': self._value(e.total_ttc),
        }

        for ed in e.details:
            p_code = self.get_product_account_code(ed)
            s += "    %(compte_produit)s \t %(amount_ht)s\n" % {
                'compte_produit': settings.get_ledger_account(p_code),
                'amount_ht': self._value(-ed.total_ht)
            }

        tvas = self.get_tva_billing_amounts(self.dolibarr_alchemy, self.e, journal="supplier")
        for k in tvas:
            s += "    %(compte_tva)s \t %(amount_tva)s\n" % {
                'compte_tva': settings.get_ledger_account(k),
                'amount_tva': self._value(tvas[k]),
            }

        return s

    # get_tva_account: return the value-added tax account
    @classmethod
    def get_tva_account(cls, ed):
        p_code = cls.get_product_account_code(ed)
        tx = int(float(ed.tva_tx) * 100)

        if p_code.startswith("2"):
            prefix = 'tva_deductible'
        else:
            prefix = 'tva_deductible_immo'
        key = "%s_%s" % (prefix, tx)
        tva_account = settings.get('PC_REFS')[key]
        return tva_account

    # get_product_account_code: return the account code for the product of the current entry
    @classmethod
    def get_product_account_code(cls, ed):
        p_code = ""
        if ed.accounting_account:
            p_code = ed.accounting_account.account_number
        elif ed.product:
            p_code = ed.product.accountancy_code_buy
        else:
            p_code = cls.pc_default_expense
        return p_code

    # get_supplier_code: return the supplier account code for the current entry
    @classmethod
    def get_supplier_code(cls, e):
        s_code = e.societe.code_compta_fournisseur
        if s_code == "":
            s_code = cls.pc_default_supplier
        return s_code


#
# HledgerSellEntry: The billing entry corresponding to a client
#
class HledgerSellEntry(HledgerBillingEntry):
    k_accounting_date = 'datef'

    def _sql_class(self):
        return self.dolibarr_alchemy.Facture

    # get_entries: return the bill entries ordered by billing date
    @classmethod
    def get_entries(cls, dolibarr_alchemy):
        Facture = dolibarr_alchemy.Facture
        return [cls(dolibarr_alchemy, e) for e in dolibarr_alchemy.session.query(Facture).order_by(Facture.datef).all()]

    # get_ledger: return the ledger corresping to this entry
    def get_ledger(self):
        e = self.e
        self.check()
        s = ""

        # Date et description
        s += "%(date)s  %(description)s\n" % {
            'date':  e.datef.strftime("%Y/%m/%d"),
            'description': e.facnumber + " - " + e.societe.nom,
        }

        # ligne pour compte client
        s_code = self.get_client_code(self.e)
        s += "    %(compte_tiers)s    %(amount_ttc)s\n" % {
            'compte_tiers': settings.get_ledger_account(s_code),
            'amount_ttc': self._value(-e.total_ttc),

        }

        # lignes pour compte de produits
        for ed in e.details:
            p_code = self.get_product_account_code(ed)
            s += "    %(compte_produit)s   %(amount_ht)s\n" % {
                'compte_produit': settings.get_ledger_account(p_code),
                'amount_ht': self._value(ed.total_ht)
            }

        # lignes pour la tva
        tvas = self.get_tva_billing_amounts(self.dolibarr_alchemy, self.e, journal="sell")
        for k in tvas:
            s += "    %(compte_tva)s  %(amount_tva)s\n" % {
                'compte_tva': settings.get_ledger_account(k),
                'amount_tva': self._value(tvas[k]),
            }

        return s

    # get_tva_account: return the value-added tax account for the given product
    @classmethod
    def get_tva_account(cls, ed):
        tx = int(float(ed.tva_tx) * 100)
        key = "tva_collecte_%s" % (tx,)
        return settings.get('PC_REFS')[key]

    # getMissingPC: return the missing account for this entry
    def getMissingPC(self):
        e = self.e
        pc_missing = []
        if e.societe.code_compta == "":
            pc_missing.append("tiers: %s %s" % (e.societe.nom, e.societe.ape))

        for ed in e.details:
            if self.get_product_account_code(ed) == self.pc_default_income:
                if ed.description != "":
                    description = ed.description.splitlines()[0]
                else:
                    description = ed.description
                pc_missing.append("produit: %s - %s - %s" % (e.societe.nom, e.facnumber, description))

        return pc_missing

    # get_product_account_code: return the account code for the product of the current entry
    @classmethod
    def get_product_account_code(cls, ed):
        p_code = ""
        if ed.accounting_account:
            p_code = ed.accounting_account.account_number
        elif ed.product:
            p_code = ed.product.accountancy_code_sell
        else:
            p_code = cls.pc_default_income
        return p_code

    # get_client_code: return the account code for the client entry
    @classmethod
    def get_client_code(cls, e):
        s_code = e.societe.code_compta
        if s_code == "":
            s_code = cls.pc_default_client
        return s_code

    # check: check if the entry is coherent
    def check(self):
        e = self.e
        total_ttc = e.total_ttc
        total_tva = e.tva
        total_ht = e.total

        for ed in e.details:
            total_ttc -= ed.total_ttc
            total_tva -= ed.total_tva
            total_ht -= ed.total_ht

        if total_ttc > 1e-10:
            print("Erreur dans l'écriture %s: total ttc = %s" % (e.facnumber, total_ttc))
        if total_ht > 1e-10:
            print("Erreur dans l'écriture %s: total ht = %s" % (e.facnumber, total_ht))
        if total_tva > 1e-10:
            print("Erreur dans l'écriture %s: total tva = %s" % (e.facnumber, total_tva))


#
# HledgerSocialEntry: A ledger entry corresponding to a tax
#
class HledgerSocialEntry(HledgerEntry):
    k_accounting_date = 'periode'

    def _sql_class(self):
        return self.dolibarr_alchemy.CotisationsSociales

    # get_entries: retrieve all the entries for this entry type
    @classmethod
    def get_entries(cls, dolibarr_alchemy):
        CotisationsSociales = dolibarr_alchemy.CotisationsSociales
        entries = dolibarr_alchemy.session.query(CotisationsSociales).order_by(CotisationsSociales.date_ech).all()
        return [cls(dolibarr_alchemy, e) for e in entries]

    # get_third_code: return the third accounting code for this entry
    @classmethod
    def get_third_code(cls, e):
        third_code = ""
        if e.type.code in settings.get('SOCIAL_REFS'):
            third_code = settings.get('SOCIAL_REFS')[e.type.code]
        if third_code == "":
            third_code = cls.pc_default_supplier
        return third_code

    # get_social_code: return the social accounting code for this entry
    @classmethod
    def get_social_code(cls, e):
        s_code = ""
        if e.type:
            s_code = e.type.accountancy_code
        if s_code == "":
            s_code = cls.pc_default_expense
        return s_code

    # getMissingPC: return the missing accounting code for this entry
    def getMissingPC(self):
        e = self.e
        pc_missing = []
        if self.get_social_code(self.e) == self.pc_default_expense:
            pc_missing.append("expenses: %s" % (e.libelle))

        if self.get_third_code(self.e) == self.pc_default_supplier:
            pc_missing.append("tiers: %s (%s)" % (e.libelle, e.type.code))

        return pc_missing

    # get_ledger: return the ledger for this entry
    def get_ledger(self):
        e = self.e

        s = ""

        s += "%(date)s    %(description)s\n" % {
            'date': e.periode.strftime("%Y/%m/%d"),
            'description': e.libelle + " - " + e.type.libelle
        }

        third_code = self.get_third_code(self.e)
        s_code = self.get_social_code(self.e)

        s += "    %(account)s \t %(amount)s\n" % {
            'account': settings.get_ledger_account(third_code),
            'amount': self._value(e.amount)
        }

        s += "    %(account)s \t %(amount)s\n" % {
            'account': settings.get_ledger_account(s_code),
            'amount': self._value(-e.amount)
        }

        return s

    # check: -
    def check(self):
        pass


#
# HledgerDolibarrSQLAlchemy: Top class for retrieving all the journals
#
class HledgerDolibarrSQLAlchemy(DolibarrSQLAlchemy):
    def get_bank_journal(self):
        return HledgerJournal(self, HledgerBankEntry)

    def get_supplier_journal(self):
        return HledgerJournal(self, HledgerSupplierEntry)

    def get_sell_journal(self):
        return HledgerJournal(self, HledgerSellEntry)

    def get_social_journal(self):
        return HledgerJournal(self, HledgerSocialEntry)