# -*- coding: utf-8 -*- import base64 import email import imaplib import logging from email.utils import parseaddr, parsedate_to_datetime from odoo import _, api, fields, models from odoo.exceptions import UserError from odoo.tools import email_normalize from .helpdesk_mail_utils import ( decode_mime_header, extract_addresses, extract_email_list, extract_ticket_ref, parse_email_body, ) _logger = logging.getLogger(__name__) FETCH_BATCH_SIZE = 50 class SoHelpdeskMailServer(models.Model): _name = 'so.helpdesk.mail.server' _description = 'Helpdesk Support Mailbox' _order = 'name' name = fields.Char(required=True) active = fields.Boolean(default=True) company_id = fields.Many2one( comodel_name='res.company', default=lambda self: self.env.company, required=True, ) server = fields.Char(string='IMAP Server', required=True) port = fields.Integer(default=993, required=True) is_ssl = fields.Boolean(string='Use SSL', default=True) user = fields.Char(string='Username', required=True) password = fields.Char( string='Password', required=True, groups='so_helpdesk.group_helpdesk_manager,base.group_system', ) folder = fields.Char(default='INBOX', required=True) from_email = fields.Char( string='Reply-From Email', help='Address customers see when agents reply (defaults to username).', ) from_name = fields.Char(string='Reply-From Name', default='Silicon Overdrive Support') default_team_id = fields.Many2one( comodel_name='so.helpdesk.team', string='Default Team', required=True, help='Team assigned to new tickets created from this mailbox.', ) last_uid = fields.Integer(default=0, readonly=True) state = fields.Selection( selection=[ ('draft', 'Not Tested'), ('ok', 'Connected'), ('error', 'Error'), ], default='draft', readonly=True, ) last_error = fields.Text(readonly=True) last_fetch_date = fields.Datetime(readonly=True) ticket_count = fields.Integer(compute='_compute_ticket_count') @api.depends('name') def _compute_ticket_count(self): grouped = self.env['so.helpdesk.ticket'].read_group( [('mail_server_id', 'in', self.ids)], ['mail_server_id'], ['mail_server_id'], ) counts = {row['mail_server_id'][0]: row['mail_server_id_count'] for row in grouped} for record in self: record.ticket_count = counts.get(record.id, 0) def _connect_imap(self): self.ensure_one() port = self.port or (993 if self.is_ssl else 143) try: conn = imaplib.IMAP4_SSL(self.server, port) if self.is_ssl else imaplib.IMAP4(self.server, port) conn.login(self.user, self.password) return conn except imaplib.IMAP4.error as exc: raise UserError(_('IMAP login failed: %s') % exc) from exc except OSError as exc: raise UserError(_('Could not connect to IMAP server: %s') % exc) from exc def action_test_connection(self): self.ensure_one() conn = None try: conn = self._connect_imap() status, _data = conn.select(self.folder, readonly=True) if status != 'OK': raise UserError(_('Could not select folder "%s".') % self.folder) self.write({'state': 'ok', 'last_error': False}) return { 'type': 'ir.actions.client', 'tag': 'display_notification', 'params': { 'title': _('Connection Successful'), 'message': _('Mailbox "%s" is reachable.') % self.name, 'type': 'success', }, } except UserError as exc: self.write({'state': 'error', 'last_error': str(exc)}) raise finally: if conn: try: conn.logout() except Exception: pass def action_fetch_now(self): self.fetch_mail() return { 'type': 'ir.actions.client', 'tag': 'display_notification', 'params': { 'title': _('Fetch Complete'), 'message': _('Support mailbox fetch finished.'), 'type': 'success', }, } @api.model def _cron_fetch_all_mailboxes(self): self.search([('active', '=', True)]).fetch_mail() def fetch_mail(self): for server in self: try: server._fetch_mail_single() except Exception as exc: _logger.exception('Helpdesk mail fetch failed for %s', server.name) server.write({'state': 'error', 'last_error': str(exc)}) def _fetch_mail_single(self): self.ensure_one() conn = None try: conn = self._connect_imap() status, _data = conn.select(self.folder, readonly=True) if status != 'OK': raise UserError(_('Could not select folder "%s".') % self.folder) if self.last_uid: status, data = conn.uid('search', None, 'UID', '%d:*' % (self.last_uid + 1)) else: status, data = conn.uid('search', None, 'UNSEEN') if status != 'OK' or not data or not data[0]: self.write({'state': 'ok', 'last_fetch_date': fields.Datetime.now(), 'last_error': False}) return uid_list = sorted(int(uid) for uid in data[0].split() if int(uid) > self.last_uid) max_uid = self.last_uid for uid in uid_list[:FETCH_BATCH_SIZE]: self._import_uid(conn, uid) max_uid = max(max_uid, uid) self.write({ 'last_uid': max_uid, 'state': 'ok', 'last_error': False, 'last_fetch_date': fields.Datetime.now(), }) finally: if conn: try: conn.logout() except Exception: pass def _import_uid(self, conn, uid): self.ensure_one() status, data = conn.uid('fetch', str(uid), '(RFC822)') if status != 'OK' or not data or not data[0] or not data[0][1]: return False msg = email.message_from_bytes(data[0][1]) subject = decode_mime_header(msg.get('Subject', '')) or _('(No Subject)') email_from = extract_addresses(msg.get('From', '')) or parseaddr(msg.get('From', ''))[1] body_text, body_html = parse_email_body(msg) body = body_html or body_text or '' message_id = (msg.get('Message-ID') or '').strip() date_received = fields.Datetime.now() if msg.get('Date'): try: date_received = parsedate_to_datetime(msg.get('Date')).replace(tzinfo=None) except (TypeError, ValueError, OverflowError): pass ticket_ref = extract_ticket_ref(subject) Ticket = self.env['so.helpdesk.ticket'].sudo() ticket = Ticket.search([('ticket_ref', '=', ticket_ref)], limit=1) if ticket_ref else Ticket.browse() if ticket: ticket._message_receive_from_email( subject=subject, body=body, email_from=email_from, message_id=message_id, ) else: partner = self.env['res.partner'].sudo().find_or_create(email_from) if email_from else False normalized = email_normalize(parseaddr(email_from)[1] or '') ticket = Ticket.create({ 'name': subject, 'team_id': self.default_team_id.id, 'partner_id': partner.id if partner else False, 'partner_email': normalized or (partner.email if partner else False), 'description': body, 'mail_server_id': self.id, 'email_message_id': message_id or False, }) ticket.message_post( body=body, subject=subject, message_type='email', subtype_xmlid='mail.mt_comment', email_from=email_from, ) if partner: ticket.message_subscribe(partner_ids=partner.ids) self._import_attachments(ticket, msg) return True def _import_attachments(self, ticket, msg): Attachment = self.env['ir.attachment'].sudo() for part in msg.walk(): if 'attachment' not in str(part.get('Content-Disposition', '')) and not part.get_filename(): continue payload = part.get_payload(decode=True) if not payload: continue filename = decode_mime_header(part.get_filename() or 'attachment') att = Attachment.create({ 'name': filename, 'type': 'binary', 'datas': base64.b64encode(payload), 'res_model': 'so.helpdesk.ticket', 'res_id': ticket.id, 'mimetype': part.get_content_type(), }) ticket.message_post( body=_('Attachment: %s') % filename, attachment_ids=[att.id], message_type='comment', subtype_xmlid='mail.mt_comment', ) def get_from_address(self): self.ensure_one() return self.from_email or self.user