# -*- coding: utf-8 -*- import re from odoo import api, fields, models, _ from odoo.exceptions import ValidationError, UserError from odoo.tools import email_normalize class SoHelpdeskTicket(models.Model): _name = 'so.helpdesk.ticket' _description = 'Helpdesk Ticket' _inherit = ['mail.thread', 'mail.activity.mixin', 'utm.mixin', 'portal.mixin'] _order = 'priority desc, id desc' _mail_post_access = 'read' name = fields.Char(string='Subject', required=True, tracking=True) ticket_ref = fields.Char( string='Ticket Reference', copy=False, readonly=True, index=True, ) active = fields.Boolean(default=True) team_id = fields.Many2one( comodel_name='so.helpdesk.team', string='Team', required=True, tracking=True, index=True, ondelete='restrict', ) stage_id = fields.Many2one( comodel_name='so.helpdesk.stage', string='Stage', tracking=True, index=True, group_expand='_read_group_stage_ids', copy=False, ondelete='restrict', ) user_id = fields.Many2one( comodel_name='res.users', string='Assigned to', tracking=True, index=True, domain="[('share', '=', False)]", ) partner_id = fields.Many2one( comodel_name='res.partner', string='Customer', tracking=True, index=True, ) partner_email = fields.Char(string='Email', tracking=True) partner_phone = fields.Char(string='Phone') company_id = fields.Many2one(related='team_id.company_id', store=True, index=True) ticket_type_id = fields.Many2one( comodel_name='so.helpdesk.ticket.type', string='Ticket Type', tracking=True, index=True, ) category = fields.Selection(related='ticket_type_id.category', store=True, readonly=True) product_id = fields.Many2one(comodel_name='product.product', string='Product') service_detail = fields.Char(string='Service Detail') description = fields.Html(sanitize=True) priority = fields.Selection( selection=[ ('0', 'Low'), ('1', 'Normal'), ('2', 'High'), ('3', 'Urgent'), ], string='Priority', default='1', tracking=True, index=True, ) tag_ids = fields.Many2many( comodel_name='so.helpdesk.tag', relation='so_helpdesk_ticket_tag_rel', column1='ticket_id', column2='tag_id', string='Tags', ) color = fields.Integer(string='Color Index') kanban_state = fields.Selection( selection=[ ('normal', 'In Progress'), ('done', 'Ready'), ('blocked', 'Blocked'), ], string='Kanban State', default='normal', ) is_closed = fields.Boolean(compute='_compute_is_closed', store=True, index=True) assign_date = fields.Datetime(string='Assigned On', readonly=True, copy=False) close_date = fields.Datetime(string='Closed On', readonly=True, copy=False) sla_status_ids = fields.One2many( comodel_name='so.helpdesk.sla.status', inverse_name='ticket_id', string='SLA Status', ) sla_deadline = fields.Datetime(compute='_compute_sla_deadline', store=True) sla_fail = fields.Boolean(compute='_compute_sla_fail', store=True) duration = fields.Float(compute='_compute_duration', store=True) # Escalation is_escalated = fields.Boolean(string='Escalated', default=False, tracking=True, index=True) escalated_date = fields.Datetime(string='Escalated On', readonly=True, copy=False) escalated_by = fields.Many2one('res.users', string='Escalated By', readonly=True, copy=False) escalated_to = fields.Many2one('res.users', string='Escalated To', tracking=True) # Email source mail_server_id = fields.Many2one( comodel_name='so.helpdesk.mail.server', string='Support Mailbox', readonly=True, copy=False, ) email_message_id = fields.Char(string='Email Message-ID', index=True, copy=False) # Customer satisfaction (1–3 stars) rating = fields.Selection( selection=[('1', '1 Star'), ('2', '2 Stars'), ('3', '3 Stars')], string='Customer Rating', copy=False, tracking=True, ) rating_feedback = fields.Text(string='Rating Feedback', copy=False) rating_requested = fields.Boolean(string='Rating Requested', copy=False) rating_date = fields.Datetime(string='Rated On', readonly=True, copy=False) @api.model def _read_group_stage_ids(self, stages, domain, order): team_id = self.env.context.get('default_team_id') or self.env.context.get('search_default_team_id') if team_id: if isinstance(team_id, list): team_id = team_id[0] if team_id else False if team_id: return self.env['so.helpdesk.stage'].search([('team_id', '=', team_id)], order=order) search_domain = [('id', 'in', stages.ids)] if stages else [] return self.env['so.helpdesk.stage'].search(search_domain, order=order) @api.depends('stage_id.is_close') def _compute_is_closed(self): for ticket in self: ticket.is_closed = bool(ticket.stage_id.is_close) @api.depends('sla_status_ids.deadline', 'sla_status_ids.status') def _compute_sla_deadline(self): for ticket in self: ongoing = ticket.sla_status_ids.filtered(lambda s: s.status == 'ongoing') ticket.sla_deadline = min(ongoing.mapped('deadline')) if ongoing else False @api.depends('sla_status_ids.status') def _compute_sla_fail(self): for ticket in self: ticket.sla_fail = any(status.status == 'failed' for status in ticket.sla_status_ids) @api.depends('create_date', 'close_date', 'is_closed') def _compute_duration(self): for ticket in self: if not ticket.create_date: ticket.duration = 0.0 continue end = ticket.close_date or fields.Datetime.now() ticket.duration = (end - ticket.create_date).total_seconds() / 3600.0 @api.onchange('partner_id') def _onchange_partner_id(self): if self.partner_id: self.partner_email = self.partner_id.email self.partner_phone = self.partner_id.phone or self.partner_id.mobile @api.onchange('team_id') def _onchange_team_id(self): if self.team_id and self.team_id.stage_ids: self.stage_id = self.team_id.stage_ids.sorted('sequence')[:1] if self.team_id and self.team_id.member_ids: self.user_id = self.team_id.leader_id or self.team_id.member_ids[:1] @api.onchange('ticket_type_id') def _onchange_ticket_type_id(self): if self.ticket_type_id: self.priority = self.ticket_type_id.default_priority @api.model_create_multi def create(self, vals_list): sequence = self.env['ir.sequence'] for vals in vals_list: if not vals.get('ticket_ref'): vals['ticket_ref'] = sequence.next_by_code('so.helpdesk.ticket') or _('New') if vals.get('user_id') and not vals.get('assign_date'): vals['assign_date'] = fields.Datetime.now() tickets = super().create(vals_list) sla_status = self.env['so.helpdesk.sla.status'] for ticket in tickets: if not ticket.stage_id and ticket.team_id.stage_ids: ticket.stage_id = ticket.team_id.stage_ids.sorted('sequence')[:1] sla_status._generate_for_ticket(ticket) ticket._ensure_customer_follower() return tickets def write(self, vals): if vals.get('user_id'): vals.setdefault('assign_date', fields.Datetime.now()) closing = False if 'stage_id' in vals: stage = self.env['so.helpdesk.stage'].browse(vals['stage_id']) if stage.is_close: vals.setdefault('close_date', fields.Datetime.now()) closing = True else: vals.setdefault('close_date', False) result = super().write(vals) if 'stage_id' in vals: for ticket in self: self.env['so.helpdesk.sla.status']._mark_reached_if_applicable(ticket) ticket._send_stage_email() if closing and not ticket.rating and not ticket.rating_requested: ticket._send_rating_request() if any(key in vals for key in ('team_id', 'priority', 'ticket_type_id')): for ticket in self: self.env['so.helpdesk.sla.status']._generate_for_ticket(ticket) return result def _ensure_customer_follower(self): for ticket in self: if ticket.partner_id: ticket.message_subscribe(partner_ids=ticket.partner_id.ids) def _send_stage_email(self): template = self.stage_id.mail_template_id if template and self.partner_email and not self.is_closed: template.send_mail(self.id, force_send=False) def _send_rating_request(self): """Email customer a 1–3 star satisfaction survey when ticket closes.""" self.ensure_one() if not self.partner_email or self.rating_requested: return template = self.env.ref('so_helpdesk.mail_template_ticket_rating', raise_if_not_found=False) if template: template.send_mail(self.id, force_send=True) self.rating_requested = True def submit_rating(self, score, feedback=''): self.ensure_one() if self.rating: raise UserError(_('This ticket has already been rated.')) if str(score) not in ('1', '2', '3'): raise UserError(_('Rating must be 1, 2, or 3 stars.')) self.write({ 'rating': str(score), 'rating_feedback': feedback, 'rating_date': fields.Datetime.now(), }) self.message_post( body=_('Customer rated this ticket: %s stars') % score, message_type='notification', subtype_xmlid='mail.mt_note', ) def _message_receive_from_email(self, subject, body, email_from, message_id=False): """Append an inbound customer email to an existing ticket thread.""" self.ensure_one() if self.is_closed: self.message_post( body=_('Customer replied after ticket was closed: %s') % subject, message_type='notification', subtype_xmlid='mail.mt_note', ) self.message_post( body=body, subject=subject, message_type='email', subtype_xmlid='mail.mt_comment', email_from=email_from, ) if message_id: self.email_message_id = message_id def _message_get_suggested_recipients(self): recipients = super()._message_get_suggested_recipients() for ticket in self: if ticket.partner_id: ticket._message_add_suggested_recipient( recipients, partner=ticket.partner_id, reason=_('Customer'), ) elif ticket.partner_email: ticket._message_add_suggested_recipient( recipients, email=ticket.partner_email, reason=_('Customer'), ) return recipients def _notify_get_reply_to(self, default=None): """Reply via the configured support mailbox when available.""" reply = super()._notify_get_reply_to(default=default) for ticket in self: mailbox = ticket.mail_server_id if mailbox: formatted = '%s <%s>' % (mailbox.from_name, mailbox.get_from_address()) reply[ticket.id] = formatted return reply def _notify_thread_by_email(self, message, msg_vals, **kwargs): """Include ticket reference in outbound email subjects for threading.""" if message.model == self._name and not msg_vals.get('subject'): ticket = self.browse(message.res_id) if ticket.ticket_ref and ticket.ticket_ref not in (message.subject or ''): msg_vals['subject'] = '[%s] %s' % (ticket.ticket_ref, ticket.name) return super()._notify_thread_by_email(message, msg_vals, **kwargs) def message_post(self, **kwargs): """Subscribe customer and email them on public comments until ticket is closed.""" self.ensure_one() subtype_xmlid = kwargs.get('subtype_xmlid') is_internal = subtype_xmlid == 'mail.mt_note' or kwargs.get('subtype_id') == self.env.ref('mail.mt_note').id if not is_internal and not self.is_closed: self._ensure_customer_follower() return super().message_post(**kwargs) @api.constrains('stage_id', 'team_id') def _check_stage_team(self): for ticket in self: if ticket.stage_id and ticket.team_id and ticket.stage_id.team_id != ticket.team_id: raise ValidationError(_( 'Stage "%(stage)s" does not belong to team "%(team)s".', stage=ticket.stage_id.name, team=ticket.team_id.name, )) def action_assign_to_me(self): self.write({'user_id': self.env.user.id}) def action_escalate(self): for ticket in self: leader = ticket.team_id.leader_id or ticket.team_id.member_ids[:1] ticket.write({ 'is_escalated': True, 'escalated_date': fields.Datetime.now(), 'escalated_by': self.env.user.id, 'escalated_to': leader.id if leader else False, 'priority': '3' if ticket.priority != '3' else ticket.priority, }) ticket.message_post( body=_('Ticket escalated to %s') % (leader.name if leader else _('team lead')), message_type='notification', subtype_xmlid='mail.mt_note', ) def action_archive(self): self.write({'active': False}) def action_unarchive(self): self.write({'active': True}) def action_open_canned_responses(self): self.ensure_one() return { 'type': 'ir.actions.act_window', 'name': _('Canned Responses'), 'res_model': 'so.helpdesk.canned.response', 'view_mode': 'tree,form', 'domain': ['|', ('team_ids', '=', False), ('team_ids', 'in', self.team_id.id)], 'context': {'active_id': self.id}, } def _compute_access_url(self): super()._compute_access_url() for ticket in self: ticket.access_url = '/my/helpdesk/ticket/%s' % ticket.id def _get_portal_return_action(self): self.ensure_one() return self.env.ref('so_helpdesk.action_helpdesk_ticket') @api.model def message_new(self, msg, custom_values=None): custom_values = dict(custom_values or {}) body = msg.get('body') subject = msg.get('subject') or _('No Subject') email_from = msg.get('from') or '' match = re.search(r'([^ ]+@[^ ]+)', email_from) partner_email = email_normalize(match.group(1)) if match else False partner = self.env['res.partner'].find_or_create(email_from) if email_from else False defaults = { 'name': subject, 'description': body, 'partner_id': partner.id if partner else False, 'partner_email': partner_email or (partner.email if partner else False), } defaults.update(custom_values) return super().message_new(msg, custom_values=defaults)