You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 

695 lines
28 KiB

# © 2016 Julien Coux (Camptocamp)
# Copyright 2020 ForgeFlow S.L. (https://www.forgeflow.com)
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
import calendar
import datetime
from operator import itemgetter
from natsort import natsorted
from odoo import api, models
class GeneralLedgerReport(models.AbstractModel):
_name = "report.account_financial_report.general_ledger"
def _get_accounts_data(self, account_ids):
accounts = self.env["account.account"].browse(account_ids)
accounts_data = {}
for account in accounts:
accounts_data.update(
{
account.id: {
"id": account.id,
"code": account.code,
"name": account.name,
"group_id": account.group_id.id,
"currency_id": account.currency_id or False,
"currency_name": account.currency_id.name,
}
}
)
return accounts_data
def _get_journals_data(self, journals_ids):
journals = self.env["account.journal"].browse(journals_ids)
journals_data = {}
for journal in journals:
journals_data.update({journal.id: {"id": journal.id, "code": journal.code}})
return journals_data
def _get_tags_data(self, tags_ids):
tags = self.env["account.analytic.tag"].browse(tags_ids)
tags_data = {}
for tag in tags:
tags_data.update({tag.id: {"name": tag.name}})
return tags_data
def _get_taxes_data(self, taxes_ids):
taxes = self.env["account.tax"].browse(taxes_ids)
taxes_data = {}
for tax in taxes:
taxes_data.update(
{
tax.id: {
"id": tax.id,
"amount": tax.amount,
"amount_type": tax.amount_type,
}
}
)
if tax.amount_type == "percent" or tax.amount_type == "division":
taxes_data[tax.id]["string"] = "%"
else:
taxes_data[tax.id]["string"] = ""
return taxes_data
def _get_acc_prt_accounts_ids(self, company_id):
accounts_domain = [
("company_id", "=", company_id),
("internal_type", "in", ["receivable", "payable"]),
]
acc_prt_accounts = self.env["account.account"].search(accounts_domain)
return acc_prt_accounts.ids
def _get_initial_balances_bs_ml_domain(
self, account_ids, company_id, date_from, base_domain, acc_prt=False
):
accounts_domain = [
("company_id", "=", company_id),
("user_type_id.include_initial_balance", "=", True),
]
if account_ids:
accounts_domain += [("id", "in", account_ids)]
domain = []
domain += base_domain
domain += [("date", "<", date_from)]
accounts = self.env["account.account"].search(accounts_domain)
domain += [("account_id", "in", accounts.ids)]
if acc_prt:
domain += [("account_id.internal_type", "in", ["receivable", "payable"])]
return domain
def _get_initial_balances_pl_ml_domain(
self, account_ids, company_id, date_from, fy_start_date, base_domain
):
accounts_domain = [
("company_id", "=", company_id),
("user_type_id.include_initial_balance", "=", False),
]
if account_ids:
accounts_domain += [("id", "in", account_ids)]
domain = []
domain += base_domain
domain += [("date", "<", date_from), ("date", ">=", fy_start_date)]
accounts = self.env["account.account"].search(accounts_domain)
domain += [("account_id", "in", accounts.ids)]
return domain
def _get_accounts_initial_balance(self, initial_domain_bs, initial_domain_pl):
gl_initial_acc_bs = self.env["account.move.line"].read_group(
domain=initial_domain_bs,
fields=["account_id", "debit", "credit", "balance", "amount_currency"],
groupby=["account_id"],
)
gl_initial_acc_pl = self.env["account.move.line"].read_group(
domain=initial_domain_pl,
fields=["account_id", "debit", "credit", "balance", "amount_currency"],
groupby=["account_id"],
)
gl_initial_acc = gl_initial_acc_bs + gl_initial_acc_pl
return gl_initial_acc
def _get_initial_balance_fy_pl_ml_domain(
self, account_ids, company_id, fy_start_date, base_domain
):
accounts_domain = [
("company_id", "=", company_id),
("user_type_id.include_initial_balance", "=", False),
]
if account_ids:
accounts_domain += [("id", "in", account_ids)]
domain = []
domain += base_domain
domain += [("date", "<", fy_start_date)]
accounts = self.env["account.account"].search(accounts_domain)
domain += [("account_id", "in", accounts.ids)]
return domain
def _get_pl_initial_balance(
self, account_ids, company_id, fy_start_date, foreign_currency, base_domain
):
domain = self._get_initial_balance_fy_pl_ml_domain(
account_ids, company_id, fy_start_date, base_domain
)
initial_balances = self.env["account.move.line"].read_group(
domain=domain,
fields=["account_id", "debit", "credit", "balance", "amount_currency"],
groupby=["account_id"],
)
pl_initial_balance = {
"debit": 0.0,
"credit": 0.0,
"balance": 0.0,
"bal_curr": 0.0,
}
for initial_balance in initial_balances:
pl_initial_balance["debit"] += initial_balance["debit"]
pl_initial_balance["credit"] += initial_balance["credit"]
pl_initial_balance["balance"] += initial_balance["balance"]
pl_initial_balance["bal_curr"] += initial_balance["amount_currency"]
return pl_initial_balance
def _get_initial_balance_data(
self,
account_ids,
partner_ids,
company_id,
date_from,
foreign_currency,
only_posted_moves,
hide_account_at_0,
unaffected_earnings_account,
fy_start_date,
analytic_tag_ids,
cost_center_ids,
):
base_domain = []
if company_id:
base_domain += [("company_id", "=", company_id)]
if partner_ids:
base_domain += [("partner_id", "in", partner_ids)]
if only_posted_moves:
base_domain += [("move_id.state", "=", "posted")]
if analytic_tag_ids:
base_domain += [("analytic_tag_ids", "in", analytic_tag_ids)]
if cost_center_ids:
base_domain += [("analytic_account_id", "in", cost_center_ids)]
initial_domain_bs = self._get_initial_balances_bs_ml_domain(
account_ids, company_id, date_from, base_domain
)
initial_domain_pl = self._get_initial_balances_pl_ml_domain(
account_ids, company_id, date_from, fy_start_date, base_domain
)
gl_initial_acc = self._get_accounts_initial_balance(
initial_domain_bs, initial_domain_pl
)
initial_domain_acc_prt = self._get_initial_balances_bs_ml_domain(
account_ids, company_id, date_from, base_domain, acc_prt=True
)
gl_initial_acc_prt = self.env["account.move.line"].read_group(
domain=initial_domain_acc_prt,
fields=[
"account_id",
"partner_id",
"debit",
"credit",
"balance",
"amount_currency",
],
groupby=["account_id", "partner_id"],
lazy=False,
)
gen_ld_data = {}
for gl in gl_initial_acc:
acc_id = gl["account_id"][0]
gen_ld_data[acc_id] = {}
gen_ld_data[acc_id]["id"] = acc_id
gen_ld_data[acc_id]["partners"] = False
gen_ld_data[acc_id]["init_bal"] = {}
gen_ld_data[acc_id]["init_bal"]["credit"] = gl["credit"]
gen_ld_data[acc_id]["init_bal"]["debit"] = gl["debit"]
gen_ld_data[acc_id]["init_bal"]["balance"] = gl["balance"]
gen_ld_data[acc_id]["fin_bal"] = {}
gen_ld_data[acc_id]["fin_bal"]["credit"] = gl["credit"]
gen_ld_data[acc_id]["fin_bal"]["debit"] = gl["debit"]
gen_ld_data[acc_id]["fin_bal"]["balance"] = gl["balance"]
gen_ld_data[acc_id]["init_bal"]["bal_curr"] = gl["amount_currency"]
gen_ld_data[acc_id]["fin_bal"]["bal_curr"] = gl["amount_currency"]
partners_data = {}
partners_ids = set()
if gl_initial_acc_prt:
for gl in gl_initial_acc_prt:
if not gl["partner_id"]:
prt_id = 0
prt_name = "Missing Partner"
else:
prt_id = gl["partner_id"][0]
prt_name = gl["partner_id"][1]
if prt_id not in partners_ids:
partners_ids.add(prt_id)
partners_data.update({prt_id: {"id": prt_id, "name": prt_name}})
acc_id = gl["account_id"][0]
gen_ld_data[acc_id][prt_id] = {}
gen_ld_data[acc_id][prt_id]["id"] = prt_id
gen_ld_data[acc_id]["partners"] = True
gen_ld_data[acc_id][prt_id]["init_bal"] = {}
gen_ld_data[acc_id][prt_id]["init_bal"]["credit"] = gl["credit"]
gen_ld_data[acc_id][prt_id]["init_bal"]["debit"] = gl["debit"]
gen_ld_data[acc_id][prt_id]["init_bal"]["balance"] = gl["balance"]
gen_ld_data[acc_id][prt_id]["fin_bal"] = {}
gen_ld_data[acc_id][prt_id]["fin_bal"]["credit"] = gl["credit"]
gen_ld_data[acc_id][prt_id]["fin_bal"]["debit"] = gl["debit"]
gen_ld_data[acc_id][prt_id]["fin_bal"]["balance"] = gl["balance"]
gen_ld_data[acc_id][prt_id]["init_bal"]["bal_curr"] = gl[
"amount_currency"
]
gen_ld_data[acc_id][prt_id]["fin_bal"]["bal_curr"] = gl[
"amount_currency"
]
accounts_ids = list(gen_ld_data.keys())
unaffected_id = unaffected_earnings_account
if unaffected_id not in accounts_ids:
accounts_ids.append(unaffected_id)
self._initialize_account(gen_ld_data, unaffected_id, foreign_currency)
pl_initial_balance = self._get_pl_initial_balance(
account_ids, company_id, fy_start_date, foreign_currency, base_domain
)
gen_ld_data[unaffected_id]["init_bal"]["debit"] += pl_initial_balance["debit"]
gen_ld_data[unaffected_id]["init_bal"]["credit"] += pl_initial_balance["credit"]
gen_ld_data[unaffected_id]["init_bal"]["balance"] += pl_initial_balance[
"balance"
]
gen_ld_data[unaffected_id]["fin_bal"]["debit"] += pl_initial_balance["debit"]
gen_ld_data[unaffected_id]["fin_bal"]["credit"] += pl_initial_balance["credit"]
gen_ld_data[unaffected_id]["fin_bal"]["balance"] += pl_initial_balance[
"balance"
]
if foreign_currency:
gen_ld_data[unaffected_id]["init_bal"]["bal_curr"] += pl_initial_balance[
"bal_curr"
]
gen_ld_data[unaffected_id]["fin_bal"]["bal_curr"] += pl_initial_balance[
"bal_curr"
]
return gen_ld_data, partners_data, partner_ids
@api.model
def _get_move_line_data(self, move_line):
move_line_data = {
"id": move_line["id"],
"date": move_line["date"],
"entry": move_line["move_id"][1],
"entry_id": move_line["move_id"][0],
"journal_id": move_line["journal_id"][0],
"account_id": move_line["account_id"][0],
"partner_id": move_line["partner_id"][0]
if move_line["partner_id"]
else False,
"partner_name": move_line["partner_id"][1]
if move_line["partner_id"]
else "",
"ref": move_line["name"],
"tax_ids": move_line["tax_ids"],
"debit": move_line["debit"],
"credit": move_line["credit"],
"balance": move_line["balance"],
"bal_curr": move_line["amount_currency"],
"rec_id": move_line["full_reconcile_id"][0]
if move_line["full_reconcile_id"]
else False,
"rec_name": move_line["full_reconcile_id"][1]
if move_line["full_reconcile_id"]
else "",
"tag_ids": move_line["analytic_tag_ids"],
"currency_id": move_line["currency_id"],
}
return move_line_data
@api.model
def _get_period_domain(
self,
account_ids,
partner_ids,
company_id,
only_posted_moves,
date_to,
date_from,
analytic_tag_ids,
cost_center_ids,
):
domain = [("date", ">=", date_from), ("date", "<=", date_to)]
if account_ids:
domain += [("account_id", "in", account_ids)]
if company_id:
domain += [("company_id", "=", company_id)]
if partner_ids:
domain += [("partner_id", "in", partner_ids)]
if only_posted_moves:
domain += [("move_id.state", "=", "posted")]
if analytic_tag_ids:
domain += [("analytic_tag_ids", "in", analytic_tag_ids)]
if cost_center_ids:
domain += [("analytic_account_id", "in", cost_center_ids)]
return domain
@api.model
def _initialize_partner(self, gen_ld_data, acc_id, prt_id, foreign_currency):
gen_ld_data[acc_id]["partners"] = True
gen_ld_data[acc_id][prt_id] = {}
gen_ld_data[acc_id][prt_id]["id"] = prt_id
gen_ld_data[acc_id][prt_id]["init_bal"] = {}
gen_ld_data[acc_id][prt_id]["init_bal"]["balance"] = 0.0
gen_ld_data[acc_id][prt_id]["init_bal"]["credit"] = 0.0
gen_ld_data[acc_id][prt_id]["init_bal"]["debit"] = 0.0
gen_ld_data[acc_id][prt_id]["fin_bal"] = {}
gen_ld_data[acc_id][prt_id]["fin_bal"]["credit"] = 0.0
gen_ld_data[acc_id][prt_id]["fin_bal"]["debit"] = 0.0
gen_ld_data[acc_id][prt_id]["fin_bal"]["balance"] = 0.0
if foreign_currency:
gen_ld_data[acc_id][prt_id]["init_bal"]["bal_curr"] = 0.0
gen_ld_data[acc_id][prt_id]["fin_bal"]["bal_curr"] = 0.0
return gen_ld_data
def _initialize_account(self, gen_ld_data, acc_id, foreign_currency):
gen_ld_data[acc_id] = {}
gen_ld_data[acc_id]["id"] = acc_id
gen_ld_data[acc_id]["partners"] = False
gen_ld_data[acc_id]["init_bal"] = {}
gen_ld_data[acc_id]["init_bal"]["balance"] = 0.0
gen_ld_data[acc_id]["init_bal"]["credit"] = 0.0
gen_ld_data[acc_id]["init_bal"]["debit"] = 0.0
gen_ld_data[acc_id]["fin_bal"] = {}
gen_ld_data[acc_id]["fin_bal"]["credit"] = 0.0
gen_ld_data[acc_id]["fin_bal"]["debit"] = 0.0
gen_ld_data[acc_id]["fin_bal"]["balance"] = 0.0
if foreign_currency:
gen_ld_data[acc_id]["init_bal"]["bal_curr"] = 0.0
gen_ld_data[acc_id]["fin_bal"]["bal_curr"] = 0.0
return gen_ld_data
def _get_period_ml_data(
self,
account_ids,
partner_ids,
company_id,
foreign_currency,
only_posted_moves,
hide_account_at_0,
date_from,
date_to,
partners_data,
gen_ld_data,
partners_ids,
centralize,
analytic_tag_ids,
cost_center_ids,
):
domain = self._get_period_domain(
account_ids,
partner_ids,
company_id,
only_posted_moves,
date_to,
date_from,
analytic_tag_ids,
cost_center_ids,
)
ml_fields = [
"id",
"name",
"date",
"move_id",
"journal_id",
"account_id",
"partner_id",
"debit",
"credit",
"balance",
"currency_id",
"full_reconcile_id",
"tax_ids",
"analytic_tag_ids",
"amount_currency",
]
move_lines = self.env["account.move.line"].search_read(
domain=domain, fields=ml_fields
)
journal_ids = set()
full_reconcile_ids = set()
taxes_ids = set()
tags_ids = set()
full_reconcile_data = {}
acc_prt_account_ids = self._get_acc_prt_accounts_ids(company_id)
for move_line in move_lines:
journal_ids.add(move_line["journal_id"][0])
for tax_id in move_line["tax_ids"]:
taxes_ids.add(tax_id)
for analytic_tag_id in move_line["analytic_tag_ids"]:
tags_ids.add(analytic_tag_id)
if move_line["full_reconcile_id"]:
rec_id = move_line["full_reconcile_id"][0]
if rec_id not in full_reconcile_ids:
full_reconcile_data.update(
{
rec_id: {
"id": rec_id,
"name": move_line["full_reconcile_id"][1],
}
}
)
full_reconcile_ids.add(rec_id)
acc_id = move_line["account_id"][0]
ml_id = move_line["id"]
if move_line["partner_id"]:
prt_id = move_line["partner_id"][0]
partner_name = move_line["partner_id"][1]
if acc_id not in gen_ld_data.keys():
gen_ld_data = self._initialize_account(
gen_ld_data, acc_id, foreign_currency
)
if acc_id in acc_prt_account_ids:
if not move_line["partner_id"]:
prt_id = 0
partner_name = "Missing Partner"
if gen_ld_data:
if prt_id not in gen_ld_data[acc_id]:
if prt_id not in partners_ids:
partners_ids.append(prt_id)
partners_data.update(
{prt_id: {"id": prt_id, "name": partner_name}}
)
gen_ld_data = self._initialize_partner(
gen_ld_data, acc_id, prt_id, foreign_currency
)
else:
partners_ids.append(prt_id)
partners_data.update({prt_id: {"id": prt_id, "name": partner_name}})
gen_ld_data = self._initialize_partner(
gen_ld_data, acc_id, prt_id, foreign_currency
)
gen_ld_data[acc_id][prt_id][ml_id] = self._get_move_line_data(move_line)
gen_ld_data[acc_id][prt_id]["fin_bal"]["credit"] += move_line["credit"]
gen_ld_data[acc_id][prt_id]["fin_bal"]["debit"] += move_line["debit"]
gen_ld_data[acc_id][prt_id]["fin_bal"]["balance"] += move_line[
"balance"
]
if foreign_currency:
gen_ld_data[acc_id][prt_id]["fin_bal"]["bal_curr"] += move_line[
"amount_currency"
]
else:
gen_ld_data[acc_id][ml_id] = self._get_move_line_data(move_line)
gen_ld_data[acc_id]["fin_bal"]["credit"] += move_line["credit"]
gen_ld_data[acc_id]["fin_bal"]["debit"] += move_line["debit"]
gen_ld_data[acc_id]["fin_bal"]["balance"] += move_line["balance"]
if foreign_currency:
gen_ld_data[acc_id]["fin_bal"]["bal_curr"] += move_line[
"amount_currency"
]
journals_data = self._get_journals_data(list(journal_ids))
accounts_data = self._get_accounts_data(gen_ld_data.keys())
taxes_data = self._get_taxes_data(list(taxes_ids))
tags_data = self._get_tags_data(list(tags_ids))
return (
gen_ld_data,
accounts_data,
partners_data,
journals_data,
full_reconcile_data,
taxes_data,
tags_data,
)
@api.model
def _create_general_ledger(self, gen_led_data, accounts_data):
general_ledger = []
for acc_id in gen_led_data.keys():
account = {}
account.update(
{
"code": accounts_data[acc_id]["code"],
"name": accounts_data[acc_id]["name"],
"type": "account",
"currency_id": accounts_data[acc_id]["currency_id"],
}
)
if not gen_led_data[acc_id]["partners"]:
move_lines = []
for ml_id in gen_led_data[acc_id].keys():
if not isinstance(ml_id, int):
account.update({ml_id: gen_led_data[acc_id][ml_id]})
else:
move_lines += [gen_led_data[acc_id][ml_id]]
account.update({"move_lines": move_lines})
else:
list_partner = []
for prt_id in gen_led_data[acc_id].keys():
partner = {}
move_lines = []
if not isinstance(prt_id, int):
account.update({prt_id: gen_led_data[acc_id][prt_id]})
else:
for ml_id in gen_led_data[acc_id][prt_id].keys():
if not isinstance(ml_id, int):
partner.update(
{ml_id: gen_led_data[acc_id][prt_id][ml_id]}
)
else:
move_lines += [gen_led_data[acc_id][prt_id][ml_id]]
partner.update({"move_lines": move_lines})
list_partner += [partner]
account.update({"list_partner": list_partner})
general_ledger += [account]
return general_ledger
@api.model
def _get_centralized_ml(self, partners, date_to):
centralized_ml = {}
if isinstance(date_to, str):
date_to = datetime.datetime.strptime(date_to, "%Y-%m-%d").date()
for partner in partners:
for move_line in partner["move_lines"]:
jnl_id = move_line["journal_id"]
month = move_line["date"].month
if jnl_id not in centralized_ml.keys():
centralized_ml[jnl_id] = {}
if month not in centralized_ml[jnl_id].keys():
centralized_ml[jnl_id][month] = {}
last_day_month = calendar.monthrange(move_line["date"].year, month)
date = datetime.date(
move_line["date"].year, month, last_day_month[1]
)
if date > date_to:
date = date_to
centralized_ml[jnl_id][month].update(
{
"journal_id": jnl_id,
"ref": "Centralized entries",
"date": date,
"debit": 0.0,
"credit": 0.0,
"balance": 0.0,
"bal_curr": 0.0,
"partner_id": False,
"rec_id": 0,
"entry_id": False,
"tax_ids": [],
"full_reconcile_id": False,
"id": False,
"tag_ids": False,
"currency_id": False,
}
)
centralized_ml[jnl_id][month]["debit"] += move_line["debit"]
centralized_ml[jnl_id][month]["credit"] += move_line["credit"]
centralized_ml[jnl_id][month]["balance"] += move_line["balance"]
centralized_ml[jnl_id][month]["bal_curr"] += move_line["bal_curr"]
list_centralized_ml = []
for jnl_id in centralized_ml.keys():
list_centralized_ml += list(centralized_ml[jnl_id].values())
return list_centralized_ml
@api.multi
def _get_report_values(self, docids, data):
wizard_id = data["wizard_id"]
company = self.env["res.company"].browse(data["company_id"])
company_id = data["company_id"]
date_to = data["date_to"]
date_from = data["date_from"]
partner_ids = data["partner_ids"]
if not partner_ids:
filter_partner_ids = False
else:
filter_partner_ids = True
account_ids = data["account_ids"]
analytic_tag_ids = data["analytic_tag_ids"]
cost_center_ids = data["cost_center_ids"]
hide_account_at_0 = data["hide_account_at_0"]
foreign_currency = data["foreign_currency"]
only_posted_moves = data["only_posted_moves"]
unaffected_earnings_account = data["unaffected_earnings_account"]
fy_start_date = data["fy_start_date"]
gen_ld_data, partners_data, partners_ids = self._get_initial_balance_data(
account_ids,
partner_ids,
company_id,
date_from,
foreign_currency,
only_posted_moves,
hide_account_at_0,
unaffected_earnings_account,
fy_start_date,
analytic_tag_ids,
cost_center_ids,
)
centralize = data["centralize"]
(
gen_ld_data,
accounts_data,
partners_data,
journals_data,
full_reconcile_data,
taxes_data,
tags_data,
) = self._get_period_ml_data(
account_ids,
partner_ids,
company_id,
foreign_currency,
only_posted_moves,
hide_account_at_0,
date_from,
date_to,
partners_data,
gen_ld_data,
partners_ids,
centralize,
analytic_tag_ids,
cost_center_ids,
)
general_ledger = self._create_general_ledger(gen_ld_data, accounts_data)
if centralize:
for account in general_ledger:
if account["partners"]:
centralized_ml = self._get_centralized_ml(
account["list_partner"], date_to
)
account["move_lines"] = centralized_ml
account["partners"] = False
del account["list_partner"]
general_ledger = natsorted(general_ledger, key=itemgetter("code"))
return {
"doc_ids": [wizard_id],
"doc_model": "general.ledger.report.wizard",
"docs": self.env["general.ledger.report.wizard"].browse(wizard_id),
"foreign_currency": data["foreign_currency"],
"company_name": company.display_name,
"company_currency": company.currency_id,
"currency_name": company.currency_id.name,
"date_from": data["date_from"],
"date_to": data["date_to"],
"only_posted_moves": data["only_posted_moves"],
"hide_account_at_0": data["hide_account_at_0"],
"show_analytic_tags": data["show_analytic_tags"],
"general_ledger": general_ledger,
"accounts_data": accounts_data,
"partners_data": partners_data,
"journals_data": journals_data,
"full_reconcile_data": full_reconcile_data,
"taxes_data": taxes_data,
"centralize": centralize,
"tags_data": tags_data,
"filter_partner_ids": filter_partner_ids,
}