import json import logging from constants import DEVICE_INVALID_MESSAGE logger = logging.getLogger(__name__) # 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', 'query_outbound_list', )) DETAIL_TOOLS = frozenset(('query_outbound_detail',)) OPTION_TOOLS = frozenset(( 'list_order_filter_options', 'list_outbound_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 | DETAIL_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': ('文件链接',), }, 'query_outbound_list': { 'outbound_number': ('排舱单号',), 'direct_send': ('直送柜',), 'remark': ('备注',), 'cargo_type': ('超长超重',), 'warehouse_name': ('集货仓库',), 'status': ('状态',), 'so_number': ('SO号',), 'container_code': ('柜号',), 'seal_number': ('封条号',), 'container_type': ('柜型',), 'shipping_method': ('运输方式',), 'total_volume': ('体积(CBM)',), 'total_weight': ('重量(KG)',), 'ship_schedule': ('船期',), 'closing_time': ('截关时间',), 'est_loading_time': ('预计装柜时间',), 'operator': ('操作人',), 'operation_modes': ('拖报清方式',), 'operation_time': ('操作时间',), }, 'query_outbound_detail': { 'order_number': ('订单号',), 'cargo_type': ('超长超重',), 'customs_files': ('报关资料',), 'reference_number': ('客户参考号',), 'customer_name': ('客户名称',), 'declaration_type': ('报关方式',), 'merge_declare_number': ('合并报关单号',), 'order_remark': ('订单备注',), 'bl_number': ('提单号',), 'container_code': ('柜号',), 'customer_service': ('客户经理',), 'est_inbound_date': ('预计入库时间',), 'inbound_date': ('实际入库时间',), 'status': ('状态',), 'pieces': ('预报件数 / 入库件数',), 'weight': ('预报实重(KG) / 入库实重(KG)',), 'volume': ('预报体积(m³) / 入库体积(m³)',), 'pickup_place': ('交货地',), 'product_name': ('物流产品',), 'goods_name': ('品名',), 'closing_time': ('截关时间',), 'clearance_remark': ('报关备注',), 'import_clearance_remark': ('清关备注',), 'delivery_address': ('派送地址',), 'delivery_type': ('派送类型',), 'delivery_way': ('派送方式',), 'channel': ('派送渠道',), 'country_name': ('目的国',), 'importer_name': ('进口商',), 'vat_type': ('清关税号',), 'packing_type': ('货物类型',), 'is_abnormal': ('是否问题件',), 'package_method': ('包装类型',), 'order_reply': ('到货回复',), }, } OUTBOUND_DETAIL_SUMMARY = { 'bl_number': '提单号', 'container_code': '柜号', 'container_type': '柜型', 'total_volume': '总体积', 'total_weight': '总重量', 'total_pieces': '总件数', 'sku': 'SKU', 'buy_declaration_count': '买单报关数量', 'general_declaration_count': '一般贸易报关数量', 'must_load': '必装单量/体积', 'backup_load': '备装单量/体积', } 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': '每页数量', }, 'query_outbound_list': { 'outbound_numbers': '排舱单号', 'order_numbers': '订单号', 'container_codes': '柜号', 'so_numbers': 'SO号', 'bl_numbers': '提单号', 'outbound_status': '排舱状态', 'shipping_method': '运输方式', 'warehouse_id': '集货仓库', 'is_direct_send': '是否直送柜', 'page': '页码', 'limit': '每页数量', }, 'query_outbound_detail': { 'outbound_number': '排舱单号', 'page': '页码', 'limit': '每页数量', }, 'list_order_filter_options': { 'filter_type': '筛选项类型', 'keyword': '关键词', 'page': '页码', 'limit': '每页数量', }, 'list_outbound_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_1103': '员工账号不可用，请联系管理员', 'MCP_1104': '身份信息已失效，请重新登录或重新生成设备配置', 'MCP_1201': '工具当前不可用', 'MCP_1202': '工具当前不可用', 'MCP_1301': '没有权限使用此工具', 'MCP_1501': '目标数据不可用', 'MCP_1502': '结果过多，请缩小查询范围', 'MCP_1601': '没有可导出的数据', 'MCP_9001': '工具调用失败，请稍后重试', } NON_RETRYABLE_CODES = frozenset(( 'MCP_1101', 'MCP_1102', 'MCP_1103', 'MCP_1104', 'MCP_1201', 'MCP_1202', 'MCP_1301', 'MCP_1401', 'MCP_1501', 'MCP_1502', 'MCP_1601', )) 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.DETAIL_TOOLS: return self._present_outbound_detail( tool_name, data, tool_result.get('meta'), 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_outbound_detail(self, tool_name, data, raw_meta, meta): summary = data.get('summary') columns = data.get('columns') records = data.get('records') if not isinstance(summary, dict): return self._format_error(meta) if not isinstance(columns, list) or not columns or not isinstance(records, list): return self._format_error(meta) if any(key not in summary for key in self.OUTBOUND_DETAIL_SUMMARY): return self._format_error(meta) allowed_columns = self.TABLE_COLUMNS.get(tool_name) headers = [] keys = [] for column in columns: if not isinstance(column, dict): return self._format_error(meta) key = column.get('key') definition = allowed_columns.get(key) if isinstance(key, str) else None if not isinstance(definition, tuple) or not definition: return self._format_error(meta) keys.append(key) headers.append({'label': definition[0]}) rows = [] for record in records: if not isinstance(record, dict): return self._format_error(meta) row = [] for key in keys: value = '' if record.get(key) is None else record.get(key, '') if key == 'customs_files': if not isinstance(value, list): return self._format_error(meta) files = [] for item in value: if not isinstance(item, dict): return self._format_error(meta) files.append({ '文件名称': item.get('file_name', ''), '文件类型': item.get('file_type', ''), '文件链接': item.get('file_url', ''), }) value = files row.append(value) rows.append(row) summary_keys = list(self.OUTBOUND_DETAIL_SUMMARY.keys()) content = { 'summary': { 'headers': [ {'label': self.OUTBOUND_DETAIL_SUMMARY[key]} for key in summary_keys ], 'row': [summary.get(key, '') for key in summary_keys], }, 'details': { 'headers': headers, 'rows': rows, }, } tips = self._safe_tips(data.get('tips')) if tips: content['details']['tips'] = tips pagination = self._build_pagination(raw_meta) if pagination: content['details']['pagination'] = pagination return self._success_result( content, self._render_outbound_detail(content), meta, ) @classmethod def _render_outbound_detail(cls, content): lines = ['排舱汇总：'] summary = content['summary'] for header, value in zip(summary['headers'], summary['row']): lines.append('- {0}: {1}'.format(header['label'], value)) lines.append('订单明细：') details = content['details'] lines.append('表头共 {0} 列：'.format(len(details['headers']))) for index, header in enumerate(details['headers'], start=1): lines.append('{0}. {1}'.format(index, header['label'])) for index, row in enumerate(details['rows'], start=1): lines.append('记录 {0}：'.format(index)) for header, value in zip(details['headers'], row): if isinstance(value, (dict, list)): value = json.dumps(value, ensure_ascii=False) lines.append('- {0}: {1}'.format(header['label'], value)) tips = details.get('tips') or [] if tips: lines.append('提示：{0}'.format('；'.join(tips))) return '\n'.join(lines) 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 not in self.ERROR_MESSAGES and code != 'MCP_1401': logger.warning( 'Unknown MCP business error code', extra={ 'request_id': meta.get('request_id', ''), 'tool_code': tool_name, 'backend_code': code, 'response_code': 'MCP_9001', 'diagnostic_reason': 'UNEXPECTED_EXCEPTION', }, ) code = 'MCP_9001' if code == 'MCP_1401': label = self._find_field_label(tool_name, raw_message) message = self._parameter_message(label) else: message = self.ERROR_MESSAGES[code] retryable = code not in self.NON_RETRYABLE_CODES 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, )