diff --git a/hr_payroll_document/__manifest__.py b/hr_payroll_document/__manifest__.py index 539b26a7..2ad2717c 100644 --- a/hr_payroll_document/__manifest__.py +++ b/hr_payroll_document/__manifest__.py @@ -8,7 +8,11 @@ "version": "18.0.1.0.0", "depends": ["hr", "base_vat"], "maintainers": ["peluko00"], - "external_dependencies": {"python": ["pypdf"]}, + "external_dependencies": { + "python": [ + "pypdf", + ], + }, "data": [ "wizard/payroll_management_wizard.xml", "security/ir.model.access.csv", diff --git a/hr_payroll_document/models/hr_employee.py b/hr_payroll_document/models/hr_employee.py index 6789d199..4fa6210b 100644 --- a/hr_payroll_document/models/hr_employee.py +++ b/hr_payroll_document/models/hr_employee.py @@ -1,4 +1,7 @@ -from odoo import fields, models +# Copyright 2025 Simone Rubino - PyTech +# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). + +from odoo import api, fields, models from odoo.exceptions import ValidationError @@ -44,12 +47,21 @@ def action_get_payroll_tree_view(self): ) return action - def write(self, vals): - res = super().write(vals) - if "identification_id" in vals and not self.env["res.partner"].simple_vat_check( - self.env.company.country_id.code, vals["identification_id"] - ): - raise ValidationError( - self.env._("The field identification ID is not valid") - ) - return res + def _validate_payroll_identification(self, code=None): + # Override if the identification should be validated in another way + if code is None and len(self) == 1: + code = self.identification_id + if country_code := self.env.company.country_id.code: + is_valid = self.env["res.partner"].simple_vat_check(country_code, code) + else: + is_valid = True + return is_valid + + @api.constrains("identification_id") + def _constrain_payroll_identification(self): + # Only check the employees that have an `identification_id` + for employee in self.filtered("identification_id"): + if not employee._validate_payroll_identification(): + raise ValidationError( + employee.env._("The field identification ID is not valid") + ) diff --git a/hr_payroll_document/tests/common.py b/hr_payroll_document/tests/common.py new file mode 100644 index 00000000..a670190c --- /dev/null +++ b/hr_payroll_document/tests/common.py @@ -0,0 +1,75 @@ +import base64 +import contextlib +from unittest import mock + +from odoo.tests import common +from odoo.tools.misc import file_path as open_file_path + +from odoo.addons.mail.tests.common import mail_new_test_user + + +class TestHrPayrollDocument(common.TransactionCase): + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.env.user.tz = "Europe/Brussels" + cls.user_admin = cls.env.ref("base.user_admin") + + # Fix Company without country + cls.env.company.country_id = False + + # Test users to use through the various tests + cls.user_employee = mail_new_test_user( + cls.env, login="david", groups="base.group_user" + ) + cls.user_employee_id = cls.user_employee.id + + # Hr Data + cls.employee_emp = cls.env["hr.employee"].create( + { + "name": "David Employee", + "user_id": cls.user_employee_id, + "company_id": 1, + "identification_id": "30831011V", + } + ) + + cls.wizard = cls._create_wizard("January", "hr_payroll_document/tests/test.pdf") + + @classmethod + def _create_wizard(cls, subject, file_path): + with open(open_file_path(file_path), "rb") as pdf_file: + encoded_string = base64.b64encode(pdf_file.read()) + ir_values = { + "name": "test", + "type": "binary", + "datas": encoded_string, + "store_fname": encoded_string, + "res_model": "payroll.management.wizard", + "res_id": 1, + } + cls.attachment = cls.env["ir.attachment"].create(ir_values) + cls.subject = subject + return cls.env["payroll.management.wizard"].create( + {"payrolls": [cls.attachment.id], "subject": cls.subject} + ) + + @contextlib.contextmanager + def _mock_valid_identification(self, employee, identification_code): + def _mocked_validate_payroll_identification(self, code=None): + if code is None: + code = employee.identification_id + return code == identification_code + + with mock.patch.object( + type(employee), + "_validate_payroll_identification", + _mocked_validate_payroll_identification, + ) as patch: + patch.side_effect = _mocked_validate_payroll_identification + yield + + def fill_company_id(self): + self.env.company.country_id = self.env["res.country"].search( + [("name", "=", "Spain")] + ) diff --git a/hr_payroll_document/tests/test_hr_payroll_document.py b/hr_payroll_document/tests/test_hr_payroll_document.py index 6ac56a87..1c141022 100644 --- a/hr_payroll_document/tests/test_hr_payroll_document.py +++ b/hr_payroll_document/tests/test_hr_payroll_document.py @@ -4,74 +4,17 @@ import pypdf from odoo.exceptions import UserError, ValidationError -from odoo.tests import common -from odoo.tools.misc import file_path -from odoo.addons.mail.tests.common import mail_new_test_user +from odoo.addons.hr_payroll_document.tests.common import TestHrPayrollDocument -class TestHRPayrollDocument(common.TransactionCase): - @classmethod - def setUpClass(cls): - super().setUpClass() - cls.env.user.tz = "Europe/Brussels" - cls.user_admin = cls.env.ref("base.user_admin") - - # Fix Company without country - cls.env.company.country_id = False - - # Test users to use through the various tests - cls.user_employee = mail_new_test_user( - cls.env, login="david", groups="base.group_user" - ) - cls.user_employee_id = cls.user_employee.id - - # Hr Data - cls.employee_emp = cls.env["hr.employee"].create( - { - "name": "David Employee", - "user_id": cls.user_employee_id, - "company_id": 1, - "identification_id": "30831011V", - } - ) - - with open(file_path("hr_payroll_document/tests/test.pdf"), "rb") as pdf_file: - encoded_string = base64.b64encode(pdf_file.read()) - ir_values = { - "name": "test", - "type": "binary", - "datas": encoded_string, - "store_fname": encoded_string, - "res_model": "payroll.management.wizard", - "res_id": 1, - } - cls.attachment = cls.env["ir.attachment"].create(ir_values) - cls.subject = "January" - cls.wizard = cls.env["payroll.management.wizard"].create( - {"payrolls": [cls.attachment.id], "subject": cls.subject} - ) - - def fill_company_id(self): - self.env.company.country_id = self.env["res.country"].search( - [("name", "=", "Spain")] - ) +class TestHRPayrollDocument(TestHrPayrollDocument): + def setUp(self, *args, **kwargs): + super().setUp(*args, **kwargs) def test_extension_error(self): - with open(file_path("hr_payroll_document/tests/test.docx"), "rb") as pdf_file: - encoded_string = base64.b64encode(pdf_file.read()) - ir_values = { - "name": "test", - "type": "binary", - "datas": encoded_string, - "store_fname": encoded_string, - "res_model": "payroll.management.wizard", - "res_id": 1, - } - self.attachment = self.env["ir.attachment"].create(ir_values) - self.subject = "January" - self.wizard = self.env["payroll.management.wizard"].create( - {"payrolls": [self.attachment.id], "subject": self.subject} + self.wizard = self._create_wizard( + "January", "hr_payroll_document/tests/test.docx" ) with self.assertRaises(ValidationError): self.wizard.send_payrolls() diff --git a/hr_payroll_document/wizard/payroll_management_wizard.py b/hr_payroll_document/wizard/payroll_management_wizard.py index bb3c96fc..c312552d 100644 --- a/hr_payroll_document/wizard/payroll_management_wizard.py +++ b/hr_payroll_document/wizard/payroll_management_wizard.py @@ -1,7 +1,7 @@ import base64 from base64 import b64decode -from pypdf import PdfReader, PdfWriter +from pypdf import PdfReader, PdfWriter, errors from odoo import fields, models from odoo.exceptions import UserError, ValidationError @@ -19,55 +19,110 @@ class PayrollManagamentWizard(models.TransientModel): "ir.attachment", "payrol_rel", "doc_id", "attach_id3", copy=False, required=True ) - def send_payrolls(self): - not_found = set() - self.merge_pdfs() - reader = PdfReader("/tmp/merged-pdf.pdf") - employees = set() - - # Validate if company have country - if not self.env.company.country_id: - raise UserError(self.env._("You must to filled country field of company")) + def _get_fallback_reader(self, pdf_reader): + # Override to use another reader + pass + + def _read_page_content(self, pdf_reader, page, fallback_reader=None): + try: + page_content = page.extract_text().split() + except errors.PdfReadError: + if fallback_reader: + # The original page cannot be read: + # read the simplified page in the fallback_reader + page_number = pdf_reader.get_page_number(page) + fallback_page = fallback_reader.get_page(page_number) + page_content = fallback_page.extract_text().split() + else: + raise + return page_content + + def _extract_employees(self, pdf_reader, fallback_reader=None): + employee_to_pages = dict() + not_found_ids = set() # Find all IDs of the employees - for page in reader.pages: - for value in page.extract_text().split(): + for page in pdf_reader.pages: + page_content = self._read_page_content( + pdf_reader, page, fallback_reader=fallback_reader + ) + for value in page_content: if self.validate_id(value) and value != self.env.company.vat: employee = self.env["hr.employee"].search( [("identification_id", "=", value)] ) if employee: - employees.add(employee) + employee_to_pages.setdefault(employee, []).append(page) else: - not_found.add(value) + not_found_ids.add(value) + break - for employee in list(employees): - pdfWriter = PdfWriter() - for page in reader.pages: - if employee.identification_id in page.extract_text(): - # Save pdf with payrolls of employee - pdfWriter.add_page(page) + return employee_to_pages, not_found_ids - path = "/tmp/" + self.env._("Payroll ") + employee.name + ".pdf" + def _build_employee_payroll(self, file_name, pdf_pages, encryption_key=None): + """Return the path to the created payroll. - if not employee.no_payroll_encryption: - # Encrypt the payroll file - # with the identification identifier of the employee - pdfWriter.encrypt(employee.identification_id, algorithm="AES-256") + Optionally encrypt the payroll file with `encryption_key`. + """ + pdfWriter = PdfWriter() + for page in pdf_pages: + pdfWriter.add_page(page) - f = open(path, "wb") - pdfWriter.write(f) - f.close() + path = "/tmp/" + file_name - # Send payroll to the employee - self.send_mail(employee, path) + if encryption_key: + pdfWriter.encrypt(encryption_key, algorithm="AES-256") + + with open(path, "wb") as f: + pdfWriter.write(f) + return path + def _show_employees_action(self): action = self.env["ir.actions.actions"]._for_xml_id( "hr_payroll_document.payrolls_view_action" ) action["views"] = [ [self.env.ref("hr_payroll_document.view_payroll_tree").id, "list"] ] + return action + + def send_payrolls(self): + self.merge_pdfs() + # Validate if company have country + if not self.env.company.country_id: + raise UserError(self.env._("You must to filled country field of company")) + + reader = PdfReader("/tmp/merged-pdf.pdf") + + try: + employee_to_pages, not_found = self._extract_employees(reader) + except errors.PdfReadError: + # Couldn't read the file, try again with another reader + fallback_reader = self._get_fallback_reader(reader) + if fallback_reader: + employee_to_pages, not_found = self._extract_employees( + reader, fallback_reader=fallback_reader + ) + else: + raise + + for employee, pages in employee_to_pages.items(): + encryption_key = ( + None if employee.no_payroll_encryption else employee.identification_id + ) + path = self._build_employee_payroll( + self.env._( + "Payroll %(subject)s %(employee)s.pdf", + employee=employee.name, + subject=self.subject, + ), + pages, + encryption_key=encryption_key, + ) + # Send payroll to the employee + self.send_mail(employee, path) + + action = self._show_employees_action() if not_found: return { "type": "ir.actions.client", @@ -156,7 +211,5 @@ def send_mail(self, employee, path): employee.id, force_send=True ) - def validate_id(self, number): - return self.env["res.partner"].simple_vat_check( - self.env.company.country_id.code, number - ) + def validate_id(self, code): + return self.env["hr.employee"]._validate_payroll_identification(code=code)