mcp/tools/query_container_timeliness_list.py

from datetime import datetime


COLUMNS = (
    ('ship_company', '船公司'),
    ('bl_number', '提单号'),
    ('container_code', '柜号'),
    ('container_type', '柜型'),
    ('is_direct_send_name', '是否直送'),
    ('shipping_method', '运输方式'),
    ('route', '航线'),
    ('vessel_name', '航名航次'),
    ('wharf_name', '到港码头'),
    ('providers_name', '拖车行'),
    ('dt_status_name', '柜子状态'),
    ('warehouse_name', '交货仓库'),
    ('order_total', '件数'),
    ('total_volume', '体积'),
    ('total_weight', '重量'),
    ('loading_name', '起运港'),
    ('destination_name', '目的港'),
    ('delivery_address', '派送地址'),
    ('has_throw_off_name', '甩柜'),
    ('throw_off_time', '甩柜时间'),
    ('ed_check_name', '报关查验'),
    ('ed_check_time', '报关查验时间'),
    ('qg_check_name', '清关查验'),
    ('check_time', '清关查验时间'),
    ('in_inspection_site', '进查验场时间'),
    ('out_inspection_site', '出查验场时间'),
    ('is_close_area_name', '封闭区'),
    ('start_close_area', '进封闭区时间'),
    ('end_close_area', '出封闭区时间'),
    ('outbound_date', '出库时间'),
    ('export_release_time', '出口报关放行时间'),
    ('etd', '干线预计出发时间'),
    ('eta', '干线预计到达时间'),
    ('gxyjsx', '干线预计时效'),
    ('departure_time', '干线实际出发时间'),
    ('arrival_time', '干线实际到达时间'),
    ('gxsjsx', '干线实际时效'),
    ('hcsx', '航程时效'),
    ('start_clearance_time', '开始清关时间'),
    ('import_release_time', '海外清关放行时间'),
    ('wharf_wait_time', '码头等待时间'),
    ('pickup_time', '码头提柜时间'),
    ('second_pickup_time', '二次提柜时间'),
    ('tgsx', '提柜时效'),
    ('max_appointment_delivery_time', 'APPT时间'),
    ('delivery_end_time', '卡车实际派送时间'),
    ('oversea_warehouse_name', '海外仓库名称'),
    ('inbound_date', '到仓时间'),
    ('inbound_time', '海外仓入库时间'),
    ('container_return_date', '还空柜时间'),
    ('hksx', '还空时效'),
    ('amazon_wait_time', '亚马逊等待时间'),
    ('inner_remark', '内部备注'),
    ('financial_remark', '财务备注'),
)


class QueryContainerTimelinessListTool:
    name = 'query_container_timeliness_list'
    route_path = '/mcp/tools/queryContainerTimelinessList'

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

    def metadata(self):
        date_field = {
            'type': 'string',
            'format': 'date',
            'pattern': '^\\d{4}-\\d{2}-\\d{2}$',
        }
        number_array = {
            'type': 'array',
            'minItems': 1,
            'maxItems': 200,
            'items': {
                'type': 'string',
                'minLength': 1,
                'maxLength': 100,
                'pattern': '.*\\S.*',
            },
        }
        return {
            'name': self.name,
            'description': (
                '使用场景：用户要看柜子时效统计报表表格（已出库排舱单一行一柜）时，'
                '对照后台admin/Report/outboundReport导出列，固定54列，不含序号。'
                '必须先确认柜号、提单号或最多31个起运港当地日历日的干线实际出发时间闭区间；'
                '没有号码时干线实际出发时间必填，有号码时时间可选。柜号与提单号可同时传入并按AND收窄。'
                '号码类型不明确时必须先询问用户：“请确认使用哪种号码查询：柜号还是提单号？”确认前不得调用。'
                '不得根据号码格式猜测，不得跨字段或跨工具试查。'
                '禁止使用：导出文件、排舱列表query_outbound_list、海外提柜列表、后台页面筛选'
                '（船公司、港口、出库时间、查验勾选等）。看表格用本工具；用户明确要下载文件时改走'
                'export_container_timeliness_report。'
                '参数只用于工具内部调用；最终回答只能展示中文业务名称；描述筛选条件时不得展示筛选字段的'
                '英文参数名。'
            ),
            'input_schema': {
                'type': 'object',
                'properties': {
                    'container_codes': dict(
                        number_array,
                        description=(
                            '柜号数组。仅当用户明确说柜号时使用；'
                            '不得放入其他类型号码。'
                        ),
                    ),
                    'bl_numbers': dict(
                        number_array,
                        description=(
                            '提单号数组。仅当用户明确说提单号时使用；'
                            '不得放入其他类型号码。'
                        ),
                    ),
                    'departure_time_start': dict(
                        date_field,
                        description='干线实际出发时间开始日期，起运港当地日历日 YYYY-MM-DD。',
                    ),
                    'departure_time_end': dict(
                        date_field,
                        description=(
                            '干线实际出发时间结束日期，起运港当地日历日 YYYY-MM-DD，'
                            '且与开始日期跨度不超过31个日历日。'
                        ),
                    ),
                    'page': {
                        'type': 'integer',
                        'minimum': 1,
                        'maximum': 100,
                        'default': 1,
                    },
                    'limit': {
                        'type': 'integer',
                        'minimum': 1,
                        'maximum': 100,
                        'default': 20,
                    },
                },
                'required': [],
                'additionalProperties': False,
            },
        }

    def call(
        self,
        container_codes=None,
        bl_numbers=None,
        departure_time_start=None,
        departure_time_end=None,
        page=1,
        limit=20,
        request_id='rq_query_container_timeliness_list',
    ):
        if self.api_client is None:
            raise RuntimeError(
                'api client is required for query_container_timeliness_list'
            )
        payload = self._locator_payload(
            container_codes,
            bl_numbers,
            departure_time_start,
            departure_time_end,
        )
        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,
        )

    @classmethod
    def _locator_payload(
        cls,
        container_codes,
        bl_numbers,
        departure_time_start,
        departure_time_end,
    ):
        payload = {}
        if container_codes is not None:
            payload['container_codes'] = cls._number_list(
                'container_codes', container_codes
            )
        if bl_numbers is not None:
            payload['bl_numbers'] = cls._number_list('bl_numbers', bl_numbers)
        total = len(payload.get('container_codes', [])) + len(
            payload.get('bl_numbers', [])
        )
        if total > 200:
            raise ValueError('at most 200 container codes and bl numbers combined')
        has_start = departure_time_start is not None
        has_end = departure_time_end is not None
        if has_start != has_end:
            raise ValueError('departure time range requires both start and end')
        if has_start:
            start = cls._date(departure_time_start, 'departure_time_start')
            end = cls._date(departure_time_end, 'departure_time_end')
            cls._assert_window(start, end)
            payload['departure_time_start'] = start
            payload['departure_time_end'] = end
        elif 'container_codes' not in payload and 'bl_numbers' not in payload:
            raise ValueError(
                'provide container codes, bl numbers, or a departure time range'
            )
        return payload

    @staticmethod
    def _date(value, field):
        text = str(value or '').strip()
        try:
            parsed = datetime.strptime(text, '%Y-%m-%d').date()
        except ValueError:
            raise ValueError(field + ' is invalid')
        return parsed.strftime('%Y-%m-%d')

    @staticmethod
    def _assert_window(start, end):
        start_date = datetime.strptime(start, '%Y-%m-%d').date()
        end_date = datetime.strptime(end, '%Y-%m-%d').date()
        if end_date < start_date or (end_date - start_date).days > 30:
            raise ValueError('departure time window is invalid')

    @staticmethod
    def _number_list(field, values):
        if not isinstance(values, list) or not values:
            raise ValueError(field + ' must be a non-empty list')
        cleaned = []
        for value in values:
            if not isinstance(value, str):
                raise ValueError(field + ' items must be strings')
            item = value.strip()
            if not item or len(item) > 100:
                raise ValueError(field + ' items must be 1 to 100 chars')
            if item not in cleaned:
                cleaned.append(item)
        if len(cleaned) > 200:
            raise ValueError('at most 200 ' + field)
        return cleaned

    @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