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',
        'query_customs_declaration_files',
    ))
    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

    TABLE_COLUMNS = {
        'query_order_exact': {
            'order_number': ('订单号',),
            'reference_number': ('客户参考号',),
            'status_txt_name': ('状态',),
            'check_status_txt_name': ('是否已查验',),
            'customer_name': ('客户名称',),
            'customer_account_type_name': ('客户属性',),
            'inbound_date': ('入库时间',),
            'wo_num': ('未完成工单',),
            'product_name': ('物流产品',),
            'inbound_pieces': ('件数',),
            'inbound_volume': ('体积(CBM)',),
            'inbound_weight': ('重量(KG)',),
            'pro_cn_name': ('品名',),
            'export_declaration_type': ('报关方式',),
            'merge_declare_number': ('合并报关单号',),
            'delivery_address': ('派送地址',),
            'container_code': ('柜号',),
            'out_status_txt': ('排舱单状态',),
            'hinge_of_destination': ('目的港',),
            'etd': ('ETD',),
            'atd': ('ATD',),
            'eta': ('ETA',),
            'ata': ('ATA',),
            'release_time': ('清关放行时间',),
            'oversea_inbound_date': ('海外入库时间',),
            'appt_time': ('APPT时间',),
            'est_loading_time': ('预计装柜时间',),
            'pickup_time': ('海外提柜时间',),
            'delivery_way_title': ('派送方式',),
            'tracking_number': ('快递单号', '承运商跟踪号码'),
            'shipment_id': ('Shipment ID',),
            'goods_attribute': ('商品属性',),
            'sku': ('SKU',),
            'sales_user': ('商务经理',),
            'service_user': ('客户经理',),
            'department_name': ('事业部',),
            'remark': ('订单备注',),
            'importer_name': ('进口商',),
            'warehouse_name': ('交货仓库',),
            'paid_status_name': ('付款状态',),
        },
        'query_track': {
            'status': ('轨迹节点', '轨迹状态，如：开始集港、离港放行、清关、配送等'),
            'location': ('轨迹地点', '发生地点'),
            'time': ('时间', '轨迹发生时间（已转换为用户时区）'),
            'content': ('轨迹内容', '详细描述'),
            'tracking_number': ('跟踪号', '快递单号'),
            'shipment_id': ('Shipment ID', '包裹ID'),
        },
        'query_customs_declaration_files': {
            'outbound_number': ('排舱单号',),
            'order_number': ('订单号',),
            'file_name': ('文件名',),
            'file_type': ('文件类型',),
            'file_url': ('文件链接',),
        },
    }

    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': '每页数量',
        },
        'query_customs_declaration_files': {
            'outbound_numbers': '排舱单号',
            'order_numbers': '订单号',
            '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': '排舱单号',
            'container_codes': '柜号',
            'bl_numbers': '提单号',
            'so_numbers': 'SO号',
            '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(
                tool_name,
                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, tool_name, 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)

        allowed_columns = self.TABLE_COLUMNS.get(tool_name)
        if not isinstance(allowed_columns, dict):
            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)
            key = key.strip()
            definition = allowed_columns.get(key)
            if not isinstance(definition, tuple) or not definition:
                return self._format_error(meta)
            keys.append(key)
            header = {'label': definition[0]}
            if len(definition) > 1:
                header['description'] = definition[1]
            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,
        )