mcp/services/output_presenter.py

import json

from constants import DEVICE_INVALID_MESSAGE


# Builds AI-safe MCP results for the explicitly supported tool set.
class OutputPresenter:
    TABLE_TOOLS = frozenset((
        'query_order_exact',
        'query_track',
    ))
    OPTION_TOOLS = frozenset((
        'list_order_filter_options',
        'list_pending_outbound_export_filter_options',
    ))
    EXPORT_TOOLS = frozenset((
        'export_pending_outbound_orders',
        'export_out_of_province_port_data',
    ))
    SAFE_TOOLS = TABLE_TOOLS | OPTION_TOOLS | EXPORT_TOOLS

    FIELD_LABELS = {
        'query_order_exact': {
            'order_number': '订单号',
            'order_numbers': '订单号',
            'reference_number': '客户参考号',
            'reference_numbers': '客户参考号',
            'tracking_number': '快递单号',
            'tracking_numbers': '快递单号',
            'outbound_number': '排舱单号',
            'outbound_numbers': '排舱单号',
            'container_code': '柜号',
            'container_codes': '柜号',
            'so_number': 'SO号',
            'so_numbers': 'SO号',
            'shipment_id': 'Shipment ID',
            'receiver_country': '收货国家',
            'product_ids': '物流产品',
            'customer_ids': '客户',
            'sales_id': '销售人员',
            'warehouse_ids': '仓库',
            'department_id': '事业部',
            'inbound_date_start': '入库开始日期',
            'inbound_date_end': '入库结束日期',
            'outbound_date_start': '出库开始日期',
            'outbound_date_end': '出库结束日期',
            'page': '页码',
            'limit': '每页数量',
        },
        'query_track': {
            'order_id': '订单ID',
            'order_number': '订单号',
            'tracking_number': '快递单号',
            'page': '页码',
            'limit': '每页数量',
        },
        'list_order_filter_options': {
            'filter_type': '筛选项类型',
            'keyword': '关键词',
            'page': '页码',
            'limit': '每页数量',
        },
        'list_pending_outbound_export_filter_options': {
            'filter_type': '筛选项类型',
            'product_type_id': '产品分类',
            'keyword': '关键词',
            'page': '页码',
            'limit': '每页数量',
        },
        'export_pending_outbound_orders': {
            'number': '单号',
            'product_type_id': '产品分类',
            'product_id': '物流产品',
            'order_warehouse_id': '集货仓库',
            'receiver_country': '目的国',
            'is_remove': '是否装柜剔除',
            'address': '派送地址',
            'inbound_date': '入库时间',
            'status': '订单状态',
            'merge_declare_number': '合并报关单号',
            'importer_id': '进口商',
            'packing_type': '货物类型',
            'is_battery': '带电属性',
            'is_magnetic': '带磁属性',
            'is_wood': '带木属性',
            'is_other': '其他商品属性',
            'is_fda': 'FDA产品属性',
            'is_toy': '玩具属性',
            'is_ultra_limit': '超长超重属性',
            'is_sensitive': '敏感货属性',
            'is_food': '食品属性',
            'no_property': '无属性',
        },
        'export_out_of_province_port_data': {
            'outbound_numbers': '排舱单号',
            'file_type': '资料类型',
        },
    }

    ERROR_MESSAGES = {
        'MCP_1101': '设备配置已失效，请重新生成配置',
        'MCP_1102': '设备会话已过期，请重新连接',
        'MCP_1201': '工具当前不可用',
        'MCP_1202': '工具当前不可用',
        'MCP_1301': '没有权限使用此工具',
        'MCP_9001': '工具调用失败，请稍后重试',
    }

    def handles(self, tool_name):
        return isinstance(tool_name, str) and tool_name in self.SAFE_TOOLS

    # Convert a backend envelope into text and structured display data.
    def present(self, tool_name, tool_result):
        if not self.handles(tool_name) or not isinstance(tool_result, dict):
            return self._format_error()

        code = str(tool_result.get('code') or '').strip()
        meta = self._build_meta(tool_result.get('meta'))
        if code not in ('MCP_0000', '0'):
            return self._business_error(
                tool_name,
                code or 'MCP_9001',
                tool_result.get('msg'),
                meta,
            )

        data = tool_result.get('data')
        if not isinstance(data, dict):
            return self._format_error(meta)

        if tool_name in self.TABLE_TOOLS:
            return self._present_table(data, tool_result.get('meta'), meta)
        if tool_name in self.OPTION_TOOLS:
            return self._present_options(data, tool_result.get('meta'), meta)
        return self._present_export(data, meta)

    # Map local tool exceptions without exposing internal error details.
    def present_exception(self, tool_name, exception):
        if not self.handles(tool_name):
            return self._format_error()

        message = str(exception or '').strip()
        if isinstance(exception, ValueError):
            label = self._find_field_label(tool_name, message)
            return self._error_result(
                'MCP_1401',
                self._parameter_message(label),
                False,
            )
        if message == DEVICE_INVALID_MESSAGE:
            return self._error_result('MCP_1101', DEVICE_INVALID_MESSAGE, False)
        if message.lower().startswith('tool disabled'):
            return self._error_result('MCP_1202', '工具当前不可用', False)
        return self._error_result(
            'MCP_9001',
            '工具调用失败，请稍后重试',
            True,
        )

    def _present_table(self, data, raw_meta, meta):
        columns = data.get('columns')
        records = data.get('records')
        if not isinstance(columns, list) or not columns or not isinstance(records, list):
            return self._format_error(meta)

        headers = []
        keys = []
        for column in columns:
            if not isinstance(column, dict):
                return self._format_error(meta)
            key = column.get('key')
            label = column.get('name')
            if not isinstance(key, str) or not key.strip():
                return self._format_error(meta)
            if not isinstance(label, str) or not label.strip():
                return self._format_error(meta)
            keys.append(key.strip())
            header = {'label': label.strip()}
            description = column.get('description')
            if isinstance(description, str) and description.strip():
                header['description'] = description.strip()
            headers.append(header)

        rows = []
        for record in records:
            if not isinstance(record, dict):
                return self._format_error(meta)
            rows.append([
                '' if record.get(key) is None else record.get(key, '')
                for key in keys
            ])

        content = {
            'summary': self._safe_text(data.get('summary')),
            'headers': headers,
            'rows': rows,
        }
        tips = self._safe_tips(data.get('tips'))
        if tips:
            content['tips'] = tips
        pagination = self._build_pagination(raw_meta)
        if pagination:
            content['pagination'] = pagination
        return self._success_result(content, self._render_table(content), meta)

    def _present_options(self, data, raw_meta, meta):
        records = data.get('records')
        if not isinstance(records, list):
            return self._format_error(meta)

        rows = []
        for record in records:
            if not isinstance(record, dict):
                return self._format_error(meta)
            rows.append([
                record.get('value', ''),
                self._safe_text(record.get('label')),
                self._safe_text(record.get('code')),
            ])

        content = {
            'headers': [
                {'label': '可传值'},
                {'label': '显示名称'},
                {'label': '业务编码'},
            ],
            'rows': rows,
        }
        pagination = self._build_pagination(raw_meta)
        if pagination:
            content['pagination'] = pagination
        return self._success_result(content, self._render_table(content), meta)

    def _present_export(self, data, meta):
        url = data.get('file_url')
        if not isinstance(url, str) or not url.strip():
            return self._format_error(meta)
        content = {
            'message': '文件已生成',
            'files': [{'label': '导出文件', 'url': url.strip()}],
        }
        text = '文件已生成\n- 导出文件: {0}'.format(url.strip())
        return self._success_result(content, text, meta)

    def _business_error(self, tool_name, code, raw_message, meta):
        if code == 'MCP_1401':
            label = self._find_field_label(tool_name, raw_message)
            message = self._parameter_message(label)
            retryable = False
        else:
            message = self.ERROR_MESSAGES.get(
                code,
                '工具调用失败，请稍后重试',
            )
            retryable = code not in ('MCP_1101', 'MCP_1102', 'MCP_1201', 'MCP_1202', 'MCP_1301')
        return self._error_result(code, message, retryable, meta)

    def _find_field_label(self, tool_name, raw_message):
        message = str(raw_message or '')
        fields = self.FIELD_LABELS.get(tool_name, {})
        for field in sorted(fields, key=len, reverse=True):
            if field in message:
                return fields[field]
        return ''

    @staticmethod
    def _parameter_message(label):
        return '{0}参数不正确'.format(label) if label else '工具参数不正确，请检查后重试'

    @staticmethod
    def _build_meta(raw_meta):
        if not isinstance(raw_meta, dict):
            return {}
        request_id = raw_meta.get('request_id')
        if not isinstance(request_id, str) or not request_id.strip():
            return {}
        return {'request_id': request_id.strip()}

    @staticmethod
    def _build_pagination(raw_meta):
        if not isinstance(raw_meta, dict):
            return {}
        return {
            key: raw_meta[key]
            for key in ('page', 'limit', 'has_more', 'total')
            if key in raw_meta
        }

    @staticmethod
    def _safe_text(value):
        if value is None:
            return ''
        return str(value).strip()

    @classmethod
    def _safe_tips(cls, value):
        if not isinstance(value, list):
            return []
        return [cls._safe_text(item) for item in value if cls._safe_text(item)]

    @classmethod
    def _render_table(cls, content):
        headers = content.get('headers') or []
        rows = content.get('rows') or []
        lines = []
        summary = cls._safe_text(content.get('summary'))
        if summary:
            lines.append(summary)
        lines.append('表头共 {0} 列：'.format(len(headers)))
        for index, header in enumerate(headers, start=1):
            lines.append('{0}. {1}'.format(index, header['label']))
        for index, row in enumerate(rows, start=1):
            lines.append('记录 {0}：'.format(index))
            for header, value in zip(headers, row):
                if isinstance(value, (dict, list)):
                    value = json.dumps(value, ensure_ascii=False)
                lines.append('- {0}: {1}'.format(header['label'], value))
        tips = content.get('tips') or []
        if tips:
            lines.append('提示：{0}'.format('；'.join(tips)))
        return '\n'.join(lines)

    @staticmethod
    def _success_result(content, text, meta):
        return {
            'structured_content': content,
            'text': text,
            'is_error': False,
            'meta': meta,
        }

    @staticmethod
    def _error_result(code, message, retryable, meta=None):
        return {
            'structured_content': {
                'code': code,
                'message': message,
                'retryable': retryable,
            },
            'text': '{0}: {1}'.format(code, message),
            'is_error': True,
            'meta': meta or {},
        }

    @classmethod
    def _format_error(cls, meta=None):
        return cls._error_result(
            'MCP_9001',
            '工具返回格式异常',
            True,
            meta,
        )