mcp/tools/export_pallet_data.py

from datetime import datetime
import re


class ExportPalletDataTool:
    name = 'export_pallet_data'
    route_path = '/mcp/tools/exportPalletData'
    DATE_RE = re.compile(r'^\d{4}-\d{2}-\d{2}$')
    DATETIME_RE = re.compile(r'^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$')

    def __init__(self, api_client=None):
        self.api_client = api_client

    def metadata(self):
        number_items = {
            'type': 'string',
            'minLength': 1,
            'maxLength': 100,
            'pattern': '.*\\S.*',
        }
        number_array = {
            'type': 'array',
            'minItems': 1,
            'maxItems': 200,
            'items': dict(number_items),
        }
        time_field = {
            'type': 'string',
            'description': (
                '海外仓入库时间，格式 YYYY-MM-DD 或 YYYY-MM-DD HH:MM:SS。'
                '只填日期时开始日 00:00:00、结束日 23:59:59。'
            ),
        }
        return {
            'name': self.name,
            'description': (
                '本工具只提交异步导出任务，不会在本次调用中等待文件生成，也禁止在一次调用内'
                '轮询任务状态。成功后返回任务引用(task_ref)和建议等待时间'
                '(retry_after_seconds)。稍后单独使用 query_export_task 查询任务状态或下载链接。'
                '使用场景：只有用户明确要求导出打托数据并需要下载文件时才可调用。'
                '必须提供柜号数组、提单号数组或海外仓入库时间起止，三者严格三选一，'
                '不可混用。号码类型不明确时必须先询问用户；用户没有明确号码类型时必须先询问：'
                '“请确认使用哪种号码导出：柜号还是提单号？”确认前不得调用。'
                '禁止使用：查看入仓列表、按订单号或打托批次号查询、勾选内部 ID、'
                '混用柜号/提单号/入库时间、一次调用内轮询 query_export_task。'
                '不得根据号码格式猜测，不得跨字段或跨工具试查。'
                '参数名仅用于工具调用；向用户回答时只能使用中文业务名称，'
                '不得展示内部参数名。'
            ),
            'input_schema': {
                'type': 'object',
                'properties': {
                    'container_codes': dict(
                        number_array,
                        description=(
                            '柜号数组。仅当用户明确说柜号时使用；'
                            '不得放入提单号、订单号或打托批次号。'
                        ),
                    ),
                    'bl_numbers': dict(
                        number_array,
                        description=(
                            '提单号数组。仅当用户明确说提单号时使用；'
                            '不得放入柜号、订单号或打托批次号。'
                        ),
                    ),
                    'inbound_time_start': dict(time_field),
                    'inbound_time_end': dict(time_field),
                },
                'required': [],
                'additionalProperties': False,
            },
        }

    def call(
        self,
        container_codes=None,
        bl_numbers=None,
        inbound_time_start=None,
        inbound_time_end=None,
        request_id='rq_export_pallet_data',
    ):
        if self.api_client is None:
            raise RuntimeError('api client is required for export_pallet_data')

        payload = {}
        effective_container = self._is_effective_number_selector(container_codes)
        effective_bl = self._is_effective_number_selector(bl_numbers)
        has_start = inbound_time_start is not None
        has_end = inbound_time_end is not None
        if has_start != has_end:
            raise ValueError('inbound time range requires both start and end')
        time_mode = has_start and has_end
        number_modes = int(effective_container) + int(effective_bl)
        if number_modes > 1:
            raise ValueError('cannot mix container codes and bl numbers')
        if number_modes >= 1 and time_mode:
            raise ValueError('cannot mix numbers and inbound time')
        if effective_container:
            payload['container_codes'] = self._clean_number_list(
                'container_codes', container_codes
            )
        elif effective_bl:
            payload['bl_numbers'] = self._clean_number_list(
                'bl_numbers', bl_numbers
            )
        elif time_mode:
            start_bound = self._parse_bound(inbound_time_start)
            end_bound = self._parse_bound(inbound_time_end)
            if end_bound.date() < start_bound.date() or (
                end_bound.date() - start_bound.date()
            ).days > 30:
                raise ValueError('inbound time range must be within 31 days')
            payload['inbound_time_start'] = str(inbound_time_start).strip()
            payload['inbound_time_end'] = str(inbound_time_end).strip()
        elif container_codes is not None:
            self._clean_number_list('container_codes', container_codes)
        elif bl_numbers is not None:
            self._clean_number_list('bl_numbers', bl_numbers)
        else:
            raise ValueError(
                'provide container codes, bl numbers, or an inbound time range'
            )
        return self.api_client.call_tool(
            self.name, self.route_path, payload, request_id
        )

    @staticmethod
    def _is_effective_number_selector(values):
        return isinstance(values, list) and len(values) > 0

    def _clean_number_list(self, field_name, values):
        if not isinstance(values, list) or not values:
            raise ValueError('{0} must be a non-empty list'.format(field_name))
        cleaned = []
        for value in values:
            if not isinstance(value, str):
                raise ValueError('{0} items must be strings'.format(field_name))
            item = value.strip()
            if not item or len(item) > 100:
                raise ValueError(
                    '{0} items must be 1 to 100 chars'.format(field_name)
                )
            if item not in cleaned:
                cleaned.append(item)
        if len(cleaned) > 200:
            raise ValueError('at most 200 {0}'.format(field_name))
        return cleaned

    def _parse_bound(self, value):
        text = str(value).strip()
        if self.DATE_RE.match(text):
            return datetime.strptime(text, '%Y-%m-%d')
        if self.DATETIME_RE.match(text):
            return datetime.strptime(text, '%Y-%m-%d %H:%M:%S')
        raise ValueError('inbound time must be date or datetime')