mirror of
https://github.com/OCA/account-financial-tools.git
synced 2025-02-02 12:47:26 +02:00
407 lines
17 KiB
Python
407 lines
17 KiB
Python
# -*- coding: utf-8 -*-
|
|
##############################################################################
|
|
#
|
|
# Author: Nicolas Bessi, Guewen Baconnier
|
|
# Copyright 2012-2014 Camptocamp SA
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License as
|
|
# published by the Free Software Foundation, either version 3 of the
|
|
# License, or (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
#
|
|
##############################################################################
|
|
from openerp import models, fields, api, _
|
|
|
|
|
|
class CreditControlPolicy(models.Model):
|
|
""" Define a policy of reminder """
|
|
|
|
_name = "credit.control.policy"
|
|
_description = """Define a reminder policy"""
|
|
|
|
name = fields.Char('Name', required=True)
|
|
level_ids = fields.One2many('credit.control.policy.level',
|
|
'policy_id',
|
|
string='Policy Levels')
|
|
do_nothing = fields.Boolean('Do nothing',
|
|
help='For policies which should not '
|
|
'generate lines or are obsolete')
|
|
company_id = fields.Many2one('res.company', string='Company')
|
|
account_ids = fields.Many2many(
|
|
'account.account',
|
|
string='Accounts',
|
|
required=True,
|
|
domain="[('type', '=', 'receivable')]",
|
|
help="This policy will be active only"
|
|
" for the selected accounts",
|
|
)
|
|
active = fields.Boolean('Active', default=True)
|
|
|
|
@api.multi
|
|
def _move_lines_domain(self, controlling_date):
|
|
""" Build the default domain for searching move lines """
|
|
self.ensure_one()
|
|
return [('account_id', 'in', self.account_ids.ids),
|
|
('date_maturity', '<=', controlling_date),
|
|
('reconcile_id', '=', False),
|
|
('partner_id', '!=', False)]
|
|
|
|
@api.multi
|
|
@api.returns('account.move.line')
|
|
def _due_move_lines(self, controlling_date):
|
|
""" Get the due move lines for the policy of the company.
|
|
|
|
The set of ids will be reduced and extended according
|
|
to the specific policies defined on partners and invoices.
|
|
|
|
Do not use direct SQL in order to respect security rules.
|
|
|
|
Assume that only the receivable lines have a maturity date and that
|
|
accounts used in the policy are reconcilable.
|
|
"""
|
|
self.ensure_one()
|
|
move_l_obj = self.env['account.move.line']
|
|
user = self.env.user
|
|
if user.company_id.credit_policy_id.id != self.id:
|
|
return move_l_obj.browse()
|
|
domain_line = self._move_lines_domain(controlling_date)
|
|
return move_l_obj.search(domain_line)
|
|
|
|
@api.multi
|
|
@api.returns('account.move.line')
|
|
def _move_lines_subset(self, controlling_date, model, move_relation_field):
|
|
""" Get the move lines related to one model for a policy.
|
|
|
|
Do not use direct SQL in order to respect security rules.
|
|
|
|
Assume that only the receivable lines have a maturity date and that
|
|
accounts used in the policy are reconcilable.
|
|
|
|
The policy relation field must be named credit_policy_id.
|
|
|
|
:param str controlling_date: date of credit control
|
|
:param str model: name of the model where is defined a credit_policy_id
|
|
:param str move_relation_field: name of the field in account.move.line
|
|
which is a many2one to `model`
|
|
:return: recordset to add in the process, recordset to remove from
|
|
the process
|
|
"""
|
|
self.ensure_one()
|
|
# MARK possible place for a good optimisation
|
|
my_obj = self.env[model]
|
|
move_l_obj = self.env['account.move.line']
|
|
default_domain = self._move_lines_domain(controlling_date)
|
|
|
|
to_add = move_l_obj.browse()
|
|
to_remove = move_l_obj.browse()
|
|
|
|
# The lines which are linked to this policy have to be included in the
|
|
# run for this policy.
|
|
# If another object override the credit_policy_id (ie. invoice after
|
|
add_objs = my_obj.search([('credit_policy_id', '=', self.id)])
|
|
if add_objs:
|
|
domain = list(default_domain)
|
|
domain.append((move_relation_field, 'in', add_objs.ids))
|
|
to_add = move_l_obj.search(domain)
|
|
|
|
# The lines which are linked to another policy do not have to be
|
|
# included in the run for this policy.
|
|
neg_objs = my_obj.search([('credit_policy_id', '!=', self.id),
|
|
('credit_policy_id', '!=', False)])
|
|
if neg_objs:
|
|
domain = list(default_domain)
|
|
domain.append((move_relation_field, 'in', neg_objs.ids))
|
|
to_remove = move_l_obj.search(domain)
|
|
return to_add, to_remove
|
|
|
|
@api.multi
|
|
@api.returns('account.move.line')
|
|
def _get_partner_related_lines(self, controlling_date):
|
|
""" Get the move lines for a policy related to a partner.
|
|
|
|
:param str controlling_date: date of credit control
|
|
:param str model: name of the model where is defined a credit_policy_id
|
|
:param str move_relation_field: name of the field in account.move.line
|
|
which is a many2one to `model`
|
|
:return: recordset to add in the process, recordset to remove from
|
|
the process
|
|
"""
|
|
return self._move_lines_subset(controlling_date, 'res.partner',
|
|
'partner_id')
|
|
|
|
@api.multi
|
|
@api.returns('account.move.line')
|
|
def _get_invoice_related_lines(self, controlling_date):
|
|
""" Get the move lines for a policy related to an invoice.
|
|
|
|
:param str controlling_date: date of credit control
|
|
:param str model: name of the model where is defined a credit_policy_id
|
|
:param str move_relation_field: name of the field in account.move.line
|
|
which is a many2one to `model`
|
|
:return: recordset to add in the process, recordset to remove from
|
|
the process
|
|
"""
|
|
return self._move_lines_subset(controlling_date, 'account.invoice',
|
|
'invoice')
|
|
|
|
@api.multi
|
|
@api.returns('account.move.line')
|
|
def _get_move_lines_to_process(self, controlling_date):
|
|
""" Build a list of move lines ids to include in a run
|
|
for a policy at a given date.
|
|
|
|
:param str controlling_date: date of credit control
|
|
:return: recordset to include in the run
|
|
"""
|
|
self.ensure_one()
|
|
# there is a priority between the lines, depicted by the calls below
|
|
lines = self._due_move_lines(controlling_date)
|
|
to_add, to_remove = self._get_partner_related_lines(controlling_date)
|
|
lines = (lines | to_add) - to_remove
|
|
to_add, to_remove = self._get_invoice_related_lines(controlling_date)
|
|
lines = (lines | to_add) - to_remove
|
|
return lines
|
|
|
|
@api.multi
|
|
@api.returns('account.move.line')
|
|
def _lines_different_policy(self, lines):
|
|
""" Return a set of move lines ids for which there is an
|
|
existing credit line but with a different policy.
|
|
"""
|
|
self.ensure_one()
|
|
move_line_obj = self.env['account.move.line']
|
|
different_lines = move_line_obj.browse()
|
|
if not lines:
|
|
return different_lines
|
|
cr = self.env.cr
|
|
cr.execute("SELECT move_line_id FROM credit_control_line"
|
|
" WHERE policy_id != %s and move_line_id in %s"
|
|
" AND manually_overridden IS false",
|
|
(self.id, tuple(lines.ids)))
|
|
res = cr.fetchall()
|
|
if res:
|
|
return move_line_obj.browse([row[0] for row in res])
|
|
return different_lines
|
|
|
|
@api.multi
|
|
def check_policy_against_account(self, account):
|
|
""" Ensure that the policy corresponds to account relation """
|
|
policies = self.search([])
|
|
allowed = [x for x in policies
|
|
if account in x.account_ids or x.do_nothing]
|
|
if self not in allowed:
|
|
allowed_names = u"\n".join(x.name for x in allowed)
|
|
raise api.Warning(
|
|
_('You can only use a policy set on '
|
|
'account %s.\n'
|
|
'Please choose one of the following '
|
|
'policies:\n %s') % (account.name, allowed_names)
|
|
)
|
|
return True
|
|
|
|
|
|
class CreditControlPolicyLevel(models.Model):
|
|
"""Define a policy level. A level allows to determine if
|
|
a move line is due and the level of overdue of the line"""
|
|
|
|
_name = "credit.control.policy.level"
|
|
_order = 'level'
|
|
_description = """A credit control policy level"""
|
|
|
|
name = fields.Char(string='Name', required=True, translate=True)
|
|
policy_id = fields.Many2one('credit.control.policy',
|
|
string='Related Policy',
|
|
required=True)
|
|
level = fields.Integer(string='Level', required=True)
|
|
computation_mode = fields.Selection(
|
|
[('net_days', 'Due Date'),
|
|
('end_of_month', 'Due Date, End Of Month'),
|
|
('previous_date', 'Previous Reminder')],
|
|
string='Compute Mode',
|
|
required=True
|
|
)
|
|
delay_days = fields.Integer(string='Delay (in days)', required=True)
|
|
email_template_id = fields.Many2one('email.template',
|
|
string='Email Template',
|
|
required=True)
|
|
channel = fields.Selection([('letter', 'Letter'),
|
|
('email', 'Email')],
|
|
string='Channel',
|
|
required=True)
|
|
custom_text = fields.Text(string='Custom Message',
|
|
required=True,
|
|
translate=True)
|
|
custom_mail_text = fields.Html(string='Custom Mail Message',
|
|
required=True, translate=True)
|
|
|
|
_sql_constraint = [('unique level',
|
|
'UNIQUE (policy_id, level)',
|
|
'Level must be unique per policy')]
|
|
|
|
@api.one
|
|
@api.constrains('level', 'computation_mode')
|
|
def _check_level_mode(self):
|
|
""" The smallest level of a policy cannot be computed on the
|
|
"previous_date".
|
|
"""
|
|
smallest_level = self.search([('policy_id', '=', self.policy_id.id)],
|
|
order='level asc', limit=1)
|
|
if smallest_level.computation_mode == 'previous_date':
|
|
return api.ValidationError(_('The smallest level can not be of '
|
|
'type Previous Reminder'))
|
|
|
|
@api.multi
|
|
def _previous_level(self):
|
|
""" For one policy level, returns the id of the previous level
|
|
|
|
If there is no previous level, it returns None, it means that's the
|
|
first policy level
|
|
|
|
:return: previous level or None if there is no previous level
|
|
"""
|
|
self.ensure_one()
|
|
previous_levels = self.search([('policy_id', '=', self.policy_id.id),
|
|
('level', '<', self.level)],
|
|
order='level desc',
|
|
limit=1)
|
|
if not previous_levels:
|
|
return None
|
|
return previous_levels
|
|
|
|
# ----- sql time related methods ---------
|
|
|
|
@staticmethod
|
|
def _net_days_get_boundary():
|
|
return (" (mv_line.date_maturity + %(delay)s)::date <= "
|
|
"date(%(controlling_date)s)")
|
|
|
|
@staticmethod
|
|
def _end_of_month_get_boundary():
|
|
return ("(date_trunc('MONTH', (mv_line.date_maturity + %(delay)s))+"
|
|
"INTERVAL '1 MONTH - 1 day')::date"
|
|
"<= date(%(controlling_date)s)")
|
|
|
|
@staticmethod
|
|
def _previous_date_get_boundary():
|
|
return "(cr_line.date + %(delay)s)::date <= date(%(controlling_date)s)"
|
|
|
|
@api.multi
|
|
def _get_sql_date_boundary_for_computation_mode(self, controlling_date):
|
|
""" Return a where clauses statement for the given controlling
|
|
date and computation mode of the level
|
|
"""
|
|
self.ensure_one()
|
|
fname = "_%s_get_boundary" % (self.computation_mode, )
|
|
if hasattr(self, fname):
|
|
fnc = getattr(self, fname)
|
|
return fnc()
|
|
else:
|
|
raise NotImplementedError(
|
|
_('Can not get function for computation mode: '
|
|
'%s is not implemented') % (fname, )
|
|
)
|
|
|
|
# -----------------------------------------
|
|
|
|
@api.multi
|
|
@api.returns('account.move.line')
|
|
def _get_first_level_move_lines(self, controlling_date, lines):
|
|
""" Retrieve all the move lines that are linked to a first level.
|
|
We use Raw SQL for performance. Security rule where applied in
|
|
policy object when the first set of lines were retrieved
|
|
"""
|
|
self.ensure_one()
|
|
move_line_obj = self.env['account.move.line']
|
|
if not lines:
|
|
return move_line_obj.browse()
|
|
cr = self.env.cr
|
|
sql = ("SELECT DISTINCT mv_line.id\n"
|
|
" FROM account_move_line mv_line\n"
|
|
" WHERE mv_line.id in %(line_ids)s\n"
|
|
" AND NOT EXISTS (SELECT id\n"
|
|
" FROM credit_control_line\n"
|
|
" WHERE move_line_id = mv_line.id\n"
|
|
# lines from a previous level with a draft or ignored state
|
|
# or manually overridden
|
|
# have to be generated again for the previous level
|
|
" AND NOT manually_overridden\n"
|
|
" AND state NOT IN ('draft', 'ignored'))"
|
|
" AND (mv_line.debit IS NOT NULL AND mv_line.debit != 0.0)\n")
|
|
sql += " AND"
|
|
_get_sql_date_part = self._get_sql_date_boundary_for_computation_mode
|
|
sql += _get_sql_date_part(controlling_date)
|
|
data_dict = {'controlling_date': controlling_date,
|
|
'line_ids': tuple(lines.ids),
|
|
'delay': self.delay_days}
|
|
cr.execute(sql, data_dict)
|
|
res = cr.fetchall()
|
|
if res:
|
|
return move_line_obj.browse([row[0] for row in res])
|
|
return move_line_obj.browse()
|
|
|
|
@api.multi
|
|
@api.returns('account.move.line')
|
|
def _get_other_level_move_lines(self, controlling_date, lines):
|
|
""" Retrieve the move lines for other levels than first level.
|
|
"""
|
|
self.ensure_one()
|
|
move_line_obj = self.env['account.move.line']
|
|
if not lines:
|
|
return move_line_obj.browse()
|
|
cr = self.env.cr
|
|
sql = ("SELECT mv_line.id\n"
|
|
" FROM account_move_line mv_line\n"
|
|
" JOIN credit_control_line cr_line\n"
|
|
" ON (mv_line.id = cr_line.move_line_id)\n"
|
|
" WHERE cr_line.id = (SELECT credit_control_line.id "
|
|
" FROM credit_control_line\n"
|
|
" WHERE credit_control_line.move_line_id = mv_line.id\n"
|
|
" AND state != 'ignored'"
|
|
" AND NOT manually_overridden"
|
|
" ORDER BY credit_control_line.level desc limit 1)\n"
|
|
" AND cr_line.level = %(previous_level)s\n"
|
|
" AND (mv_line.debit IS NOT NULL AND mv_line.debit != 0.0)\n"
|
|
# lines from a previous level with a draft or ignored state
|
|
# or manually overridden
|
|
# have to be generated again for the previous level
|
|
" AND NOT manually_overridden\n"
|
|
" AND cr_line.state NOT IN ('draft', 'ignored')\n"
|
|
" AND mv_line.id in %(line_ids)s\n")
|
|
sql += " AND "
|
|
_get_sql_date_part = self._get_sql_date_boundary_for_computation_mode
|
|
sql += _get_sql_date_part(controlling_date)
|
|
previous_level = self._previous_level()
|
|
data_dict = {'controlling_date': controlling_date,
|
|
'line_ids': tuple(lines.ids),
|
|
'delay': self.delay_days,
|
|
'previous_level': previous_level.level}
|
|
|
|
# print cr.mogrify(sql, data_dict)
|
|
cr.execute(sql, data_dict)
|
|
res = cr.fetchall()
|
|
if res:
|
|
return move_line_obj.browse([row[0] for row in res])
|
|
return move_line_obj.browse()
|
|
|
|
@api.multi
|
|
@api.returns('account.move.line')
|
|
def get_level_lines(self, controlling_date, lines):
|
|
""" get all move lines in entry lines that match the current level """
|
|
self.ensure_one()
|
|
move_line_obj = self.env['account.move.line']
|
|
matching_lines = move_line_obj.browse()
|
|
if self._previous_level() is None:
|
|
method = self._get_first_level_move_lines
|
|
else:
|
|
method = self._get_other_level_move_lines
|
|
matching_lines |= method(controlling_date, lines)
|
|
return matching_lines
|