from __future__ import annotations import base64 import io import logging import re import zipfile from markupsafe import Markup, escape from odoo import models, api _logger = logging.getLogger(__name__) _HTML_TAG = re.compile(r'<[^>]+>') # How many recent messages to scan when looking for a pending file upload _LOOKBACK_MESSAGES = 10 # File type labels shown in the clarification message _EXT_LABELS = { 'jpg': 'image', 'jpeg': 'image', 'png': 'image', 'gif': 'image', 'bmp': 'image', 'tiff': 'image', 'tif': 'image', 'webp': 'image', 'pdf': 'PDF', 'html': 'HTML', 'htm': 'HTML', 'txt': 'text', 'csv': 'spreadsheet', 'xlsx': 'spreadsheet', 'zip': 'ZIP archive', } def _strip_html(html: str) -> str: return _HTML_TAG.sub(' ', html or '').strip() def _ext(filename: str) -> str: return filename.rsplit('.', 1)[-1].lower() if '.' in filename else '' def _describe_zip(datas_b64: str, zip_name: str) -> Markup: """Return a safe HTML summary of a ZIP's contents without extracting data.""" try: raw = base64.b64decode(datas_b64) with zipfile.ZipFile(io.BytesIO(raw)) as zf: members = [m for m in zf.namelist() if not m.endswith('/')] if not members: return Markup(f'{escape(zip_name)} (empty archive)') counts: dict[str, int] = {} for m in members: label = _EXT_LABELS.get(_ext(m), 'file') counts[label] = counts.get(label, 0) + 1 type_summary = escape(', '.join(f'{n} {t}(s)' for t, n in counts.items())) lines: list[Markup] = [ Markup(f'{escape(zip_name)} — {len(members)} item(s): {type_summary}') ] for m in members[:8]: lines.append(Markup(f'  • {escape(m)}')) if len(members) > 8: lines.append(Markup(f'  … and {len(members) - 8} more')) return Markup('
').join(lines) except Exception as exc: _logger.warning('_describe_zip failed for %s: %s', zip_name, exc) return Markup(f'{escape(zip_name)} (could not inspect contents)') class DiscussChannel(models.Model): _inherit = 'discuss.channel' @api.model def _ai_bot_partner(self): return self.env.ref('activeblue_ai.partner_activeblue_ai', raise_if_not_found=False) def message_post(self, *, body='', author_id=None, **kwargs): result = super().message_post(body=body, author_id=author_id, **kwargs) # Only intercept direct-message channels if self.channel_type != 'chat': return result bot_partner = self._ai_bot_partner() if not bot_partner: return result member_partners = self.channel_member_ids.partner_id if bot_partner not in member_partners: return result # Don't react to the bot's own messages if author_id == bot_partner.id: return result text = _strip_html(body) # Odoo 18 Discuss uploads attachments before posting the message and # passes their IDs in kwargs as attachment_ids (list of ints or ORM # commands). result.attachment_ids resolves those after super() runs. # Log both so we can see exactly what arrives. _logger.info( 'AB AI mail hook: body=%r kwargs_keys=%s ' 'attachment_ids_kwarg=%r result.attachment_ids=%s', (body or '')[:80], list(kwargs.keys()), kwargs.get('attachment_ids'), result.attachment_ids.ids, ) attachments = result.attachment_ids # Nothing to work with if not text and not attachments: return result # ── Case 1: file(s) with no instruction ────────────────────────────── # Show the user what we received and ask what to do with it. if attachments and not text: self._post_file_clarification(attachments, bot_partner) return result # ── Case 2: text only — look back for a pending file upload ────────── # If the user just replied to our clarification question, find the # attachment(s) they uploaded earlier in this conversation. pending = self.env['ir.attachment'].browse() if text and not attachments: pending = self._find_pending_attachments(bot_partner) effective_attachments = attachments or pending # ── Case 3: text (+ possibly pending files) → dispatch ─────────────── human_partner = member_partners.filtered(lambda p: p != bot_partner)[:1] user = self.env['res.users'].search([('partner_id', '=', human_partner.id)], limit=1) uid = user.id if user else self.env.uid try: bot = self.env['ab.ai.bot'].sudo().search([('active', '=', True)], limit=1) if not bot: return result if effective_attachments: response = bot.dispatch_message_with_files( user_id=uid, message=text, attachments=effective_attachments, context={'channel_id': self.id, 'source': 'discuss'}, ) else: response = bot.dispatch_message( user_id=uid, message=text, context={'channel_id': self.id, 'source': 'discuss'}, ) reply_text = ((response or {}).get('reply') or (response or {}).get('message') or 'I could not process your request right now.') # Agent replies are plain text; convert newlines to
for Discuss. reply_html = Markup('
').join(Markup(escape(ln)) for ln in reply_text.split('\n')) self.sudo().message_post( body=reply_html, author_id=bot_partner.id, message_type='comment', subtype_xmlid='mail.mt_comment', ) except Exception as exc: _logger.error('AI bot Discuss reply failed: %s', exc) return result def _post_file_clarification(self, attachments, bot_partner): """Describe the uploaded file(s) and ask the user what to do with them.""" lines: list[Markup] = [] for att in attachments: name = att.name or 'file' ext = _ext(name) if ext == 'zip' and att.datas: lines.append(_describe_zip(att.datas, name)) else: label = _EXT_LABELS.get(ext, 'file') lines.append(Markup(f'{escape(name)} ({escape(label)})')) file_summary = Markup('
').join(lines) question = Markup( 'I received the following file(s):
' ) + file_summary + Markup( '

' 'What would you like me to do with them? Some options:
' '• Create an expense report from these receipts
' '• Import products from this data
' '• Something else — just tell me what you need' ) self.sudo().message_post( body=question, author_id=bot_partner.id, message_type='comment', subtype_xmlid='mail.mt_comment', ) def _find_pending_attachments(self, bot_partner): """ Scan the last _LOOKBACK_MESSAGES messages in this channel for the most recent human-sent message that has attachments. Only returns them if the immediately following bot message looks like a clarification question (i.e. the bot hasn't already acted on those files). """ messages = self.message_ids.sorted('date', reverse=True)[:_LOOKBACK_MESSAGES] _bot_question_phrases = ( 'what would you like me to do', 'suspected duplicate', 'skip duplicates', 'keep all', ) prev_was_bot_question = False for msg in messages: is_bot = msg.author_id == bot_partner if is_bot: body_lower = (msg.body or '').lower() if any(p in body_lower for p in _bot_question_phrases): prev_was_bot_question = True continue # Human message if msg.attachment_ids and prev_was_bot_question: return msg.attachment_ids prev_was_bot_question = False return self.env['ir.attachment'].browse()