177 lines
7.5 KiB
Python
177 lines
7.5 KiB
Python
# pyrefly: ignore [missing-import]
|
|
import logging
|
|
from odoo import api, fields, models
|
|
from odoo.exceptions import ValidationError
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
|
|
class EducationAnnouncement(models.Model):
|
|
_name = 'education.announcement'
|
|
_description = 'Education Announcement'
|
|
_order = 'priority desc, date_published desc, id desc'
|
|
|
|
title = fields.Char(string='Title', required=True, translate=True)
|
|
content = fields.Html(string='Content', required=True)
|
|
date_published = fields.Date(string='Date Published', required=True, default=fields.Date.context_today)
|
|
expiry_date = fields.Date(string='Expiry Date')
|
|
announcement_type = fields.Selection([
|
|
('notice', 'Notice'),
|
|
('circular', 'Circular'),
|
|
('event', 'Event')
|
|
], string='Type', required=True, default='notice')
|
|
priority = fields.Selection([
|
|
('0', 'Normal'),
|
|
('1', 'Important'),
|
|
('2', 'Urgent')
|
|
], string='Priority', default='0', index=True)
|
|
state = fields.Selection([
|
|
('draft', 'Draft'),
|
|
('published', 'Published'),
|
|
('archived', 'Archived')
|
|
], string='Status', default='published', required=True)
|
|
active = fields.Boolean(default=True)
|
|
author_id = fields.Many2one(
|
|
'res.users',
|
|
string='Author',
|
|
default=lambda self: self.env.user,
|
|
readonly=True
|
|
)
|
|
company_id = fields.Many2one(
|
|
'res.company',
|
|
string='Company',
|
|
required=True,
|
|
default=lambda self: self.env.company,
|
|
index=True,
|
|
)
|
|
circular_file = fields.Binary(string='Circular File', attachment=True)
|
|
circular_filename = fields.Char(string='Circular Filename')
|
|
|
|
# Audience & Cohort Targeting
|
|
audience_type = fields.Selection([
|
|
('all', 'All Students'),
|
|
('cohort', 'Specific Academic Cohort')
|
|
], string='Target Audience', default='all', required=True)
|
|
program_id = fields.Many2one('education.program', string='Target Program', index=True)
|
|
academic_year_id = fields.Many2one('education.academic.year', string='Target Academic Year', index=True)
|
|
academic_term_id = fields.Many2one('education.academic.term', string='Target Academic Term', index=True)
|
|
batch_id = fields.Many2one('education.batch', string='Target Batch', index=True)
|
|
semester_id = fields.Many2one('education.semester', string='Target Semester', index=True)
|
|
section_id = fields.Many2one('education.section', string='Target Section', index=True)
|
|
|
|
# Push Notification Audit
|
|
notification_sent = fields.Boolean(string='Notification Dispatched', default=False, copy=False)
|
|
notification_sent_date = fields.Datetime(string='Dispatched At', readonly=True, copy=False)
|
|
notification_recipient_count = fields.Integer(string='Recipients Count', readonly=True, copy=False)
|
|
|
|
@api.constrains('date_published', 'expiry_date')
|
|
def _check_dates(self):
|
|
for rec in self:
|
|
if rec.expiry_date and rec.expiry_date < rec.date_published:
|
|
raise ValidationError("Expiry date must be on or after the publication date.")
|
|
|
|
def action_publish(self):
|
|
for rec in self:
|
|
if rec.expiry_date and rec.expiry_date < rec.date_published:
|
|
raise ValidationError("Expiry date must be on or after the publication date.")
|
|
rec.state = 'published'
|
|
rec._dispatch_notifications()
|
|
|
|
def action_archive(self):
|
|
for rec in self:
|
|
rec.state = 'archived'
|
|
|
|
def action_draft(self):
|
|
for rec in self:
|
|
rec.state = 'draft'
|
|
|
|
def action_resend_notification(self):
|
|
for rec in self:
|
|
rec._dispatch_notifications(force=True)
|
|
|
|
def check_student_audience(self, student):
|
|
"""Verify whether an announcement applies to a given student."""
|
|
self.ensure_one()
|
|
if self.audience_type == 'all':
|
|
return True
|
|
if self.program_id and self.program_id.id != student.program_id.id:
|
|
return False
|
|
if self.academic_year_id and self.academic_year_id.id != student.academic_year_id.id:
|
|
return False
|
|
if self.academic_term_id and self.academic_term_id.id != student.academic_term_id.id:
|
|
return False
|
|
if self.batch_id and self.batch_id.id != student.batch_id.id:
|
|
return False
|
|
if self.semester_id and self.semester_id.id != student.semester_id.id:
|
|
return False
|
|
if self.section_id and self.section_id.id != student.section_id.id:
|
|
return False
|
|
return True
|
|
|
|
@api.model
|
|
def get_announcements_for_student(self, student, limit=None, offset=0, announcement_type=None):
|
|
"""Retrieve active published announcements valid for the student within the effective date window."""
|
|
today = fields.Date.context_today(self)
|
|
domain = [
|
|
('active', '=', True),
|
|
('state', '=', 'published'),
|
|
('date_published', '<=', today),
|
|
'|', ('expiry_date', '=', False), ('expiry_date', '>=', today),
|
|
]
|
|
if student.company_id:
|
|
domain.append(('company_id', '=', student.company_id.id))
|
|
if announcement_type:
|
|
domain.append(('announcement_type', '=', announcement_type))
|
|
|
|
records = self.sudo().search(domain, order="priority desc, date_published desc, id desc")
|
|
filtered = records.filtered(lambda a: a.check_student_audience(student))
|
|
total_count = len(filtered)
|
|
if offset:
|
|
filtered = filtered[offset:]
|
|
if limit:
|
|
filtered = filtered[:limit]
|
|
return filtered, total_count
|
|
|
|
def _dispatch_notifications(self, force=False):
|
|
"""Provider-independent notification dispatcher with in-app logging and duplicate prevention."""
|
|
for rec in self:
|
|
if rec.state != 'published':
|
|
continue
|
|
if rec.notification_sent and not force:
|
|
_logger.info("Notification already dispatched for announcement #%s ('%s'). Skipping duplicate.", rec.id, rec.title)
|
|
continue
|
|
|
|
# Resolve targeted student recipients
|
|
domain = []
|
|
if rec.company_id:
|
|
domain.append(('company_id', '=', rec.company_id.id))
|
|
if rec.audience_type == 'cohort':
|
|
if rec.program_id:
|
|
domain.append(('program_id', '=', rec.program_id.id))
|
|
if rec.academic_year_id:
|
|
domain.append(('academic_year_id', '=', rec.academic_year_id.id))
|
|
if rec.academic_term_id:
|
|
domain.append(('academic_term_id', '=', rec.academic_term_id.id))
|
|
if rec.batch_id:
|
|
domain.append(('batch_id', '=', rec.batch_id.id))
|
|
if rec.semester_id:
|
|
domain.append(('semester_id', '=', rec.semester_id.id))
|
|
if rec.section_id:
|
|
domain.append(('section_id', '=', rec.section_id.id))
|
|
|
|
students = self.env['education.student'].sudo().search(domain)
|
|
recipient_count = len(students)
|
|
|
|
# In-App / Decoupled Notification Dispatch Log
|
|
_logger.info(
|
|
"DISPATCH NOTIFICATION: Announcement #%s ('%s', Priority: %s, Audience: %s) dispatched to %s students.",
|
|
rec.id, rec.title, rec.priority, rec.audience_type, recipient_count
|
|
)
|
|
|
|
rec.write({
|
|
'notification_sent': True,
|
|
'notification_sent_date': fields.Datetime.now(),
|
|
'notification_recipient_count': recipient_count,
|
|
})
|
|
|