from odoo import _, api, fields, models from odoo.exceptions import ValidationError class XenoHrAnnouncement(models.Model): _name = 'xeno.hr.announcement' _description = 'HR Announcement' _inherit = ['mail.thread'] _order = 'start_date desc, id desc' name = fields.Char(string='Title', required=True, tracking=True) body = fields.Html(string='Message') start_date = fields.Date(default=fields.Date.context_today, required=True, tracking=True) end_date = fields.Date(default=fields.Date.context_today, required=True, tracking=True) is_all_day = fields.Boolean(string='All Day', default=True) start_time = fields.Float(string='Start Time') end_time = fields.Float(string='End Time') active = fields.Boolean(default=True) target_audience_type = fields.Selection( [('all', 'All Employees'), ('department', 'Specific Department'), ('category', 'Employee Group'), ('job', 'Job Position')], string='Target Audience', default='all', required=True, tracking=True, help="Real targeting, not just a label: this decides who actually " "gets notified/emailed, and which employees see this on their " "own landing page.") target_department_id = fields.Many2one( 'hr.department', string='Department', help="Required when Target Audience is 'Specific Department'.") target_category_id = fields.Many2one( 'hr.employee.category', string='Employee Group', help="Required when Target Audience is 'Employee Group' (Odoo's " "own Employee Tags). Every employee carrying this tag is in " "the audience, regardless of department.") target_job_id = fields.Many2one( 'hr.job', string='Job Position', help="Required when Target Audience is 'Job Position'. Every " "employee whose job position matches this is in the " "audience, regardless of department.") attachment = fields.Binary(string='Attachment', attachment=True) attachment_filename = fields.Char(string='Attachment Filename') send_email = fields.Boolean( string='Send Email Notification', help="On top of the always-on in-app notification, also send this " "announcement as a real email to every employee in the target " "audience. Only fires once -- ticking it again after it has " "already sent (see Email Sent At) does not resend.") email_sent_at = fields.Datetime(string='Email Sent At', readonly=True, copy=False) source_policy_id = fields.Many2one( 'xeno.hr.policy', string='Created From Policy', ondelete='set null', copy=False, help="Set when this announcement was spun off from a policy via " "its 'Create Announcement' button -- used to log a link back " "on that policy's own chatter once this is saved.") source_activity_id = fields.Many2one( 'xeno.hr.activity', string='Created From Activity', ondelete='set null', copy=False, help="Set when this announcement was spun off from a company " "activity via its 'Create Announcement' button -- used to log " "a link back on that activity's own chatter once this is saved.") @api.onchange('is_all_day') def _onchange_is_all_day(self): for rec in self: if rec.is_all_day: rec.start_time = 0.0 rec.end_time = 0.0 @api.constrains('start_date', 'end_date') def _check_dates(self): for rec in self: if rec.start_date and rec.end_date and rec.end_date < rec.start_date: raise ValidationError(_("End date cannot be before start date.")) @api.constrains('target_audience_type', 'target_department_id', 'target_category_id', 'target_job_id') def _check_target_selection(self): for rec in self: if rec.target_audience_type == 'department' and not rec.target_department_id: raise ValidationError(_( "Select a department for a department-targeted announcement.")) if rec.target_audience_type == 'category' and not rec.target_category_id: raise ValidationError(_( "Select an employee group for a group-targeted announcement.")) if rec.target_audience_type == 'job' and not rec.target_job_id: raise ValidationError(_( "Select a job position for a job-position-targeted announcement.")) @api.model def xeno_get_my_announcements(self, limit=5): """Active, currently-in-window announcements targeted at the logged-in user's own employee (all-employees, their department, their employee group/tag, or their job position). Hard-scoped to self.env.user.employee_id server-side, sudo()'d -- deliberately NOT implemented as a client-built domain against a raw employee read, since hr.employee.category_ids carries a genuine groups="hr.group_hr_user" restriction on the real field (a regular employee reading their own category_ids at all raises an AccessError, confirmed live) -- this method reads it server-side instead of ever sending it to the client. """ employee = self.env.user.employee_id if not employee: return [] emp = employee.sudo() today = fields.Date.context_today(self) domain = [ ('active', '=', True), ('start_date', '<=', today), ('end_date', '>=', today), '|', '|', '|', ('target_audience_type', '=', 'all'), '&', ('target_audience_type', '=', 'department'), ('target_department_id', '=', emp.department_id.id), '&', ('target_audience_type', '=', 'category'), ('target_category_id', 'in', emp.category_ids.ids), '&', ('target_audience_type', '=', 'job'), ('target_job_id', '=', emp.job_id.id), ] announcements = self.env['xeno.hr.announcement'].sudo().search(domain, limit=limit) return announcements.read(['name', 'start_date', 'body']) def _xeno_target_employees(self): self.ensure_one() domain = [('active', '=', True)] if self.target_audience_type == 'department' and self.target_department_id: domain.append(('department_id', '=', self.target_department_id.id)) elif self.target_audience_type == 'category' and self.target_category_id: domain.append(('category_ids', 'in', self.target_category_id.id)) elif self.target_audience_type == 'job' and self.target_job_id: domain.append(('job_id', '=', self.target_job_id.id)) return self.env['hr.employee'].sudo().search(domain) def _xeno_notify_announcement(self): """Always-on in-app (bell/inbox) notification, regardless of send_email -- "every new announcement notifies in the Odoo HR system" per susu.""" self.ensure_one() partners = self._xeno_target_employees().mapped('user_id.partner_id') if not partners: return self.message_notify( partner_ids=partners.ids, subject=self.name, body=self.body or '', subtype_xmlid='mail.mt_comment', ) def _xeno_send_announcement_email(self): self.ensure_one() employees = self._xeno_target_employees() emails = [] for emp in employees: email = emp.work_email or (emp.user_id.email if emp.user_id else False) if email: emails.append(email) if not emails: return attachment_ids = [] if self.attachment: attachment = self.env['ir.attachment'].sudo().create({ 'name': self.attachment_filename or self.name, 'datas': self.attachment, 'res_model': self._name, 'res_id': self.id, }) attachment_ids = [attachment.id] self.env['mail.mail'].sudo().create({ 'subject': self.name, 'body_html': self.body or '', 'email_to': ','.join(emails), 'attachment_ids': [(6, 0, attachment_ids)], 'auto_delete': True, }).send() self.email_sent_at = fields.Datetime.now() @api.model_create_multi def create(self, vals_list): records = super().create(vals_list) for rec in records: rec._xeno_notify_announcement() if rec.send_email: rec._xeno_send_announcement_email() if rec.source_policy_id: rec.source_policy_id._xeno_log_announcement_created(rec) if rec.source_activity_id: rec.source_activity_id._xeno_log_announcement_created(rec) return records def write(self, vals): res = super().write(vals) if vals.get('send_email'): for rec in self: if rec.send_email and not rec.email_sent_at: rec._xeno_send_announcement_email() return res