File: account_secure_entries_wizard.py

package info (click to toggle)
odoo 18.0.0%2Bdfsg-2
  • links: PTS, VCS
  • area: main
  • in suites: sid
  • size: 878,716 kB
  • sloc: javascript: 927,937; python: 685,670; xml: 388,524; sh: 1,033; sql: 415; makefile: 26
file content (267 lines) | stat: -rw-r--r-- 12,597 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
from datetime import timedelta

from odoo import Command, api, fields, models, _
from odoo.exceptions import UserError
from odoo.osv import expression
from odoo.tools import format_list


class AccountSecureEntries(models.TransientModel):
    """
    This wizard is used to secure journal entries (with a hash)
    """
    _name = 'account.secure.entries.wizard'
    _description = 'Secure Journal Entries'

    company_id = fields.Many2one(
        comodel_name='res.company',
        required=True,
        readonly=True,
        default=lambda self: self.env.company,
    )
    country_code = fields.Char(
        related="company_id.account_fiscal_country_id.code",
    )
    hash_date = fields.Date(
        string='Hash All Entries',
        required=True,
        compute="_compute_hash_date",
        store=True,
        readonly=False,
        help="The selected Date",
    )
    chains_to_hash_with_gaps = fields.Json(
        compute='_compute_data',
    )
    max_hash_date = fields.Date(
        string='Max Hash Date',
        compute='_compute_max_hash_date',
        help="Highest Date such that all posted journal entries prior to (including) the date are secured. Only journal entries after the hard lock date are considered.",
    )
    unreconciled_bank_statement_line_ids = fields.Many2many(
        compute='_compute_data',
        comodel_name='account.bank.statement.line',
        help="All unreconciled bank statement lines before the selected date.",
    )
    not_hashable_unlocked_move_ids = fields.Many2many(
        compute='_compute_data',
        comodel_name='account.move',
        help="All unhashable moves before the selected date that are not protected by the Hard Lock Date",
    )
    move_to_hash_ids = fields.Many2many(
        compute='_compute_data',
        comodel_name='account.move',
        help="All moves that will be hashed",
    )
    warnings = fields.Json(
        compute='_compute_warnings',
    )

    @api.depends('max_hash_date')
    def _compute_hash_date(self):
        for wizard in self:
            if not wizard.hash_date:
                wizard.hash_date = wizard.max_hash_date or fields.Date.context_today(self)

    @api.depends('company_id', 'company_id.user_hard_lock_date')
    def _compute_max_hash_date(self):
        today = fields.Date.context_today(self)
        for wizard in self:
            chains_to_hash = wizard._get_chains_to_hash(wizard.company_id, today)
            moves = self.env['account.move'].concat(
                *[chain['moves'] for chain in chains_to_hash],
                *[chain['not_hashable_unlocked_moves'] for chain in chains_to_hash],
            )
            if moves:
                wizard.max_hash_date = min(move.date for move in moves) - timedelta(days=1)
            else:
                wizard.max_hash_date = False

    @api.model
    def _get_chains_to_hash(self, company_id, hash_date):
        self.ensure_one()
        res = []
        moves = self.env['account.move'].sudo().search(
            self._get_unhashed_moves_in_hashed_period_domain(company_id, hash_date, [('state', '=', 'posted')])
        )
        for journal, journal_moves in moves.grouped('journal_id').items():
            for chain_moves in journal_moves.grouped('sequence_prefix').values():
                chain_info = chain_moves._get_chain_info(force_hash=True)
                if chain_info is False:
                    continue

                last_move_hashed = chain_info['last_move_hashed']
                # It is possible that some moves cannot be hashed (i.e. after upgrade).
                # We show a warning ('account_not_hashable_unlocked_moves') if that is the case.
                # These moves are ignored for the warning and max_hash_date in case they are protected by the Hard Lock Date
                if last_move_hashed:
                    # remaining_moves either have a hash already or have a higher sequence_number than the last_move_hashed
                    not_hashable_unlocked_moves = chain_info['remaining_moves'].filtered(
                        lambda move: (not move.inalterable_hash
                                      and move.sequence_number < last_move_hashed.sequence_number
                                      and move.date > self.company_id.user_hard_lock_date)
                    )
                else:
                    not_hashable_unlocked_moves = self.env['account.move']
                chain_info['not_hashable_unlocked_moves'] = not_hashable_unlocked_moves
                res.append(chain_info)
        return res

    @api.depends('company_id', 'company_id.user_hard_lock_date', 'hash_date')
    def _compute_data(self):
        for wizard in self:
            unreconciled_bank_statement_line_ids = []
            chains_to_hash = []
            if wizard.hash_date:
                for chain_info in wizard._get_chains_to_hash(wizard.company_id, wizard.hash_date):
                    if 'unreconciled' in chain_info['warnings']:
                        unreconciled_bank_statement_line_ids.extend(
                            chain_info['moves'].statement_line_ids.filtered(lambda l: not l.is_reconciled).ids
                        )
                    else:
                        chains_to_hash.append(chain_info)
            wizard.unreconciled_bank_statement_line_ids = [Command.set(unreconciled_bank_statement_line_ids)]
            wizard.chains_to_hash_with_gaps = [
                {
                    'first_move_id': chain['moves'][0].id,
                    'last_move_id': chain['moves'][-1].id,
                } for chain in chains_to_hash if 'gap' in chain['warnings']
            ]

            not_hashable_unlocked_moves = []
            move_to_hash_ids = []
            for chain in chains_to_hash:
                not_hashable_unlocked_moves.extend(chain['not_hashable_unlocked_moves'].ids)
                move_to_hash_ids.extend(chain['moves'].ids)
            wizard.not_hashable_unlocked_move_ids = [Command.set(not_hashable_unlocked_moves)]
            wizard.move_to_hash_ids = [Command.set(move_to_hash_ids)]

    @api.depends('company_id', 'chains_to_hash_with_gaps', 'hash_date', 'not_hashable_unlocked_move_ids', 'max_hash_date', 'unreconciled_bank_statement_line_ids')
    def _compute_warnings(self):
        for wizard in self:
            warnings = {}

            if not wizard.hash_date:
                wizard.warnings = warnings
                continue

            if wizard.unreconciled_bank_statement_line_ids:
                ignored_sequence_prefixes = list(set(wizard.unreconciled_bank_statement_line_ids.move_id.mapped('sequence_prefix')))
                warnings['account_unreconciled_bank_statement_line_ids'] = {
                    'message': _("There are still unreconciled bank statement lines before the selected date. "
                                 "The entries from journal prefixes containing them will not be secured: %(prefix_info)s",
                                 prefix_info=format_list(self.env, ignored_sequence_prefixes)),
                    'level': 'danger',
                    'action_text': _("Review"),
                    'action': wizard.company_id._get_unreconciled_statement_lines_redirect_action(wizard.unreconciled_bank_statement_line_ids),
                }

            draft_entries = self.env['account.move'].search_count(
                wizard._get_draft_moves_in_hashed_period_domain(),
                limit=1
            )
            if draft_entries:
                warnings['account_unhashed_draft_entries'] = {
                    'message': _("There are still draft entries before the selected date."),
                    'action_text': _("Review"),
                    'action': wizard.action_show_draft_moves_in_hashed_period(),
                }

            not_hashable_unlocked_moves = wizard.not_hashable_unlocked_move_ids
            if not_hashable_unlocked_moves:
                warnings['account_not_hashable_unlocked_moves'] = {
                    'message': _("There are entries that cannot be hashed. They can be protected by the Hard Lock Date."),
                    'action_text': _("Review"),
                    'action': wizard.action_show_moves(not_hashable_unlocked_moves),
                }

            if wizard.chains_to_hash_with_gaps:
                OR_domains = []
                for chain in wizard.chains_to_hash_with_gaps:
                    first_move = self.env['account.move'].browse(chain['first_move_id'])
                    last_move = self.env['account.move'].browse(chain['last_move_id'])
                    OR_domains.append([
                        *self.env['account.move']._check_company_domain(wizard.company_id),
                        ('journal_id', '=', last_move.journal_id.id),
                        ('sequence_prefix', '=', last_move.sequence_prefix),
                        ('sequence_number', '<=', last_move.sequence_number),
                        ('sequence_number', '>=', first_move.sequence_number),
                    ])
                domain = expression.OR(OR_domains)
                warnings['account_sequence_gap'] = {
                    'message': _("Securing these entries will create at least one gap in the sequence."),
                    'action_text': _("Review"),
                    'action': {
                        **self.env['account.journal']._show_sequence_holes(domain),
                        'views': [[self.env.ref('account.view_move_tree_multi_edit').id, 'list'], [self.env.ref('account.view_move_form').id, 'form']],
                    }
                }

            moves_to_hash_after_selected_date = wizard.move_to_hash_ids.filtered(lambda move: move.date > wizard.hash_date)
            if moves_to_hash_after_selected_date:
                warnings['account_move_to_secure_after_selected_date'] = {
                    'message': _("Securing these entries will also secure entries after the selected date."),
                    'action_text': _("Review"),
                    'action': wizard.action_show_moves(moves_to_hash_after_selected_date),
                }

            wizard.warnings = warnings

    @api.model
    def _get_unhashed_moves_in_hashed_period_domain(self, company_id, hash_date, domain=False):
        """
        Return the domain to find all moves before `self.hash_date` that have not been hashed yet.
        We ignore whether hashing is activated for the journal or not.
        :return a search domain
        """
        if not (company_id and hash_date):
            return [(0, '=', 1)]
        return expression.AND([
            [
                ('date', '<=', fields.Date.to_string(hash_date)),
                ('company_id', 'child_of', company_id.id),
                ('inalterable_hash', '=', False),
            ],
            domain or [],
        ])

    def _get_draft_moves_in_hashed_period_domain(self):
        self.ensure_one()
        return self._get_unhashed_moves_in_hashed_period_domain(self.company_id, self.hash_date, [('state', '=', 'draft')])

    def action_show_moves(self, moves):
        self.ensure_one()
        return {
            'view_mode': 'list',
            'name': _('Journal Entries'),
            'res_model': 'account.move',
            'type': 'ir.actions.act_window',
            'domain': [('id', 'in', moves.ids)],
            'search_view_id': [self.env.ref('account.view_account_move_filter').id, 'search'],
            'views': [[self.env.ref('account.view_move_tree_multi_edit').id, 'list'], [self.env.ref('account.view_move_form').id, 'form']],
        }

    def action_show_draft_moves_in_hashed_period(self):
        self.ensure_one()
        return {
            'view_mode': 'list',
            'name': _('Draft Entries'),
            'res_model': 'account.move',
            'type': 'ir.actions.act_window',
            'domain': self._get_draft_moves_in_hashed_period_domain(),
            'search_view_id': [self.env.ref('account.view_account_move_filter').id, 'search'],
            'views': [[self.env.ref('account.view_move_tree_multi_edit').id, 'list'], [self.env.ref('account.view_move_form').id, 'form']],
        }

    def action_secure_entries(self):
        self.ensure_one()

        if not self.hash_date:
            raise UserError(_("Set a date. The moves will be secured up to including this date."))

        self.env['res.groups']._activate_group_account_secured()

        if not self.move_to_hash_ids:
            return

        self.move_to_hash_ids._hash_moves(force_hash=True, raise_if_gap=False)