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