# -*- 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 self.tva_type != 'none': 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), } # lignes compte fournisseur if self.tva_type == 'none': for ed in e.details: p_code = self.get_product_account_code(ed) s += " %(compte_produit)s \t %(amount_ttc)s\n" % { 'compte_produit': settings.get_ledger_account(p_code), 'amount_ttc': self._value(-ed.total_ttc) } else: 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) } # value-added tax if self.tva_type != 'none': 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 if self.tva_type == 'none': for ed in e.details: p_code = self.get_product_account_code(ed) s += " %(compte_produit)s %(amount_ttc)s\n" % { 'compte_produit': settings.get_ledger_account(p_code), 'amount_ttc': self._value(ed.total_ttc) } else: 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 if self.tva_type != 'none': 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)