mcp/tools/query_destination_trailer_list.py

from typing import Literal


from datetime import date


class QueryDestinationTrailerListTool:
    name = 'query_destination_trailer_list'
    route_path = '/mcp/tools/queryDestinationTrailerList'
    STAGES = (
        '待安排拖车', '待提柜', '待还柜(海)', '已还柜(海)', '已提货(空)', 'DO单制作',
    )
    DO_FORBIDDEN = (
        'providers_id', 'warehouse_id', 'is_direct_send', 'bill_status',
        'cabinet_type', 'shipping_method', 'eta_start', 'eta_end',
        'pickup_time_start', 'pickup_time_end',
        'container_return_time_start', 'container_return_time_end',
    )

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

    def metadata(self):
        number_array = {
            'type': 'array', 'minItems': 1, 'maxItems': 200,
            'items': {
                'type': 'string', 'minLength': 1, 'maxLength': 100,
                'pattern': '.*\\S.*',
            },
        }
        date_field = {
            'type': 'string', 'format': 'date',
            'pattern': '^\\d{4}-\\d{2}-\\d{2}$',
        }
        properties = {
            'trailer_stage': {
                'type': 'string',
                'enum': list[Literal['待安排拖车', '待提柜', '待还柜(海)', '已还柜(海)', '已提货(空)', 'DO单制作']](self.STAGES),
                'description': (
                    '海外提柜类型。必须先调用list_destination_trailer_filter_options并选择'
                    '“海外提柜类型”取得六个中文值之一；用户未明确时必须先询问，不得默认某一类'
                    '或六类一起查，也不得改走排舱列表。'
                ),
            },
            'bl_numbers': dict(
                number_array,
                description='提单号数组。仅当用户明确说提单号时使用；不得放入柜号或其他号码。',
            ),
            'container_codes': dict(
                number_array,
                description='柜号数组。仅当用户明确说柜号时使用；不得放入提单号或其他号码。',
            ),
            'providers_id': {
                'type': 'integer', 'minimum': 1,
                'description': '拖车行筛选值，必须先通过筛选项工具取得。',
            },
            'warehouse_id': {
                'type': 'integer', 'minimum': 1,
                'description': '海外仓筛选值，必须先通过筛选项工具取得；只命中非直送。',
            },
            'is_direct_send': {
                'type': 'integer', 'enum': [0, 1],
                'description': '直送柜筛选值，先通过筛选项工具取得。',
            },
            'bill_status': {
                'type': 'integer', 'enum': [10, 20, 30, 35, 40],
                'description': '应付费用状态筛选值，先通过筛选项工具取得。',
            },
            'cabinet_type': {
                'type': 'integer', 'enum': [1, 2, 3, 4],
                'description': '整拼类型筛选值，先通过筛选项工具取得。',
            },
            'shipping_method': {
                'type': 'integer', 'enum': [1, 2, 3],
                'description': '运输方式筛选值，先通过筛选项工具取得。',
            },
            'eta_start': dict(date_field),
            'eta_end': dict(date_field),
            'pickup_time_start': dict(date_field),
            'pickup_time_end': dict(date_field),
            'container_return_time_start': dict(date_field),
            'container_return_time_end': dict(date_field),
            'page': {'type': 'integer', 'minimum': 1, 'maximum': 100, 'default': 1},
            'limit': {'type': 'integer', 'minimum': 1, 'maximum': 100, 'default': 20},
        }
        return {
            'name': self.name,
            'description': (
                '使用场景：用户要看海外提柜进度或DO单制作列表时，先确认六个类型之一即可查询当前页签；'
                '提单号、柜号、日期和其他筛选均为可选。传了号码则类型必须明确，'
                '号码类型不明确时必须先询问用户，'
                '不得根据号码格式猜测，不得跨字段或跨工具试查，也不得把海外提柜当成排舱列表调用'
                'query_outbound_list。日期若传入必须成对且最多31个目的港当地日历日。'
                '禁止使用：安排拖车、确认还柜、港后数据/应付/入仓单/POD/'
                '装柜顺序导出，以及混查订单号、拖车单号、排舱单号或SO。参数只用于工具内部调用；'
                '最终回答只能展示中文业务名称；描述筛选条件时不得展示筛选字段的英文参数名。'
            ),
            'input_schema': {
                'type': 'object',
                'properties': properties,
                'required': ['trailer_stage'],
                'additionalProperties': False,
            },
        }

    def call(
        self, trailer_stage, bl_numbers=None, container_codes=None,
        providers_id=None, warehouse_id=None, is_direct_send=None,
        bill_status=None, cabinet_type=None, shipping_method=None,
        eta_start=None, eta_end=None, pickup_time_start=None,
        pickup_time_end=None, container_return_time_start=None,
        container_return_time_end=None, page=1, limit=20,
        request_id='rq_query_destination_trailer_list',
    ):
        if self.api_client is None:
            raise RuntimeError(
                'api client is required for query_destination_trailer_list'
            )
        stage = str(trailer_stage or '').strip()
        if stage not in self.STAGES:
            raise ValueError('trailer_stage is invalid')
        payload = {'trailer_stage': stage}
        total = 0
        for field, values in (
            ('bl_numbers', bl_numbers),
            ('container_codes', container_codes),
        ):
            if values is None:
                continue
            cleaned = self._number_list(values, field)
            payload[field] = cleaned
            total += len(values)
        if total > 200:
            raise ValueError('at most 200 numbers are allowed')

        for start_name, start_value, end_value in (
            ('eta_start', eta_start, eta_end),
            ('pickup_time_start', pickup_time_start, pickup_time_end),
            (
                'container_return_time_start',
                container_return_time_start,
                container_return_time_end,
            ),
        ):
            end_name = start_name.replace('_start', '_end')
            start = self._optional_date(start_value, start_name)
            end = self._optional_date(end_value, end_name)
            if bool(start) != bool(end):
                raise ValueError(start_name + ' range is invalid')
            if start:
                if (date.fromisoformat(end) - date.fromisoformat(start)).days > 30:
                    raise ValueError(start_name + ' range is invalid')
                if end < start:
                    raise ValueError(start_name + ' range is invalid')
                payload[start_name] = start
                payload[end_name] = end

        for field, value, allowed in (
            ('providers_id', providers_id, None),
            ('warehouse_id', warehouse_id, None),
            ('is_direct_send', is_direct_send, (0, 1)),
            ('bill_status', bill_status, (10, 20, 30, 35, 40)),
            ('cabinet_type', cabinet_type, (1, 2, 3, 4)),
            ('shipping_method', shipping_method, (1, 2, 3)),
        ):
            if value is None:
                continue
            checked = self._integer(value, field, 0 if field == 'is_direct_send' else 1)
            if allowed is not None and checked not in allowed:
                raise ValueError(field + ' is invalid')
            payload[field] = checked
        if stage == 'DO单制作':
            for field in self.DO_FORBIDDEN:
                if field in payload:
                    raise ValueError('DO stage does not accept extra filters')
        payload['page'] = self._bounded_integer(page, 'page')
        payload['limit'] = self._bounded_integer(limit, 'limit')
        return self.api_client.call_tool(
            self.name, self.route_path, payload, request_id,
        )

    @staticmethod
    def _number_list(values, field):
        if not isinstance(values, list) or not values:
            raise ValueError(field + ' is invalid')
        result = []
        for value in values:
            if not isinstance(value, str):
                raise ValueError(field + ' is invalid')
            value = value.strip()
            if not value or len(value) > 100:
                raise ValueError(field + ' is invalid')
            if value not in result:
                result.append(value)
        return result

    @staticmethod
    def _optional_date(value, field):
        if value is None or value == '':
            return ''
        if not isinstance(value, str):
            raise ValueError(field + ' is invalid')
        try:
            parsed = date.fromisoformat(value)
        except ValueError:
            raise ValueError(field + ' is invalid')
        if parsed.isoformat() != value:
            raise ValueError(field + ' is invalid')
        return value

    @staticmethod
    def _integer(value, field, minimum=None):
        if isinstance(value, bool) or not isinstance(value, int):
            raise ValueError(field + ' is invalid')
        if minimum is not None and value < minimum:
            raise ValueError(field + ' is invalid')
        return value

    @staticmethod
    def _bounded_integer(value, field):
        if isinstance(value, bool) or not isinstance(value, int):
            raise ValueError(field + ' is invalid')
        if value < 1 or value > 100:
            raise ValueError(field + ' is invalid')
        return value