mcp/tools/query_payable_cost_list.py

import re
from datetime import date


class QueryPayableCostListTool:
    name = 'query_payable_cost_list'
    route_path = '/mcp/tools/queryPayableCostList'
    NUMBER_FIELDS = (
        'order_numbers', 'tracking_numbers', 'container_codes',
        'bl_numbers', 'so_numbers',
    )
    BUSINESS_TYPES = (1, 2, 3, 4, 5, 7)

    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 = {
            'business_type': {'type': 'integer', 'enum': list(self.BUSINESS_TYPES)},
            'order_numbers': dict(number_array),
            'tracking_numbers': dict(number_array),
            'container_codes': dict(number_array),
            'bl_numbers': dict(number_array),
            'so_numbers': dict(number_array),
            'business_date_start': dict(date_field),
            'business_date_end': dict(date_field),
            'cost_date_start': dict(date_field),
            'cost_date_end': dict(date_field),
            'operation_date_start': dict(date_field),
            'operation_date_end': dict(date_field),
            'business_node_id': {'type': 'integer', 'minimum': 1},
            'provider_id': {'type': 'integer', 'minimum': 1},
            'cost_type_ids': {
                'type': 'array', 'minItems': 1, 'maxItems': 200,
                'items': {'type': 'integer', 'minimum': 1},
            },
            'billing_status': {'type': 'integer', 'enum': [0, 1]},
            'payment_status': {'type': 'integer', 'enum': [-1, 1]},
            'verification_status': {'type': 'integer', 'enum': [0, 1, 2]},
            'document_type': {'type': 'integer', 'minimum': 0},
            'page': {'type': 'integer', 'minimum': 1, 'maximum': 100, 'default': 1},
            'limit': {'type': 'integer', 'minimum': 1, 'maximum': 100, 'default': 20},
        }
        return {
            'name': self.name,
            'description': (
                '按结算新版页签查询当前员工公司内的应付费用列表。五类号码字段分别精确筛选，'
                '不同字段与其他条件均为AND；禁止混合单号猜测、跨字段试查或使用未确认的费用项。'
                '头程(business_type=1)的order_numbers同时匹配单号与展示订单号(sub_number)。'
                '不传号码时至少提供一组不超过31日的业务、费用或操作日期闭区间。'
                '筛选值必须先调用list_payable_cost_filter_options取得，禁止猜测内部ID。'
            ),
            'input_schema': {
                'type': 'object', 'properties': properties,
                'required': ['business_type'], 'additionalProperties': False,
            },
        }

    def call(
        self, business_type, order_numbers=None, tracking_numbers=None,
        container_codes=None, bl_numbers=None, so_numbers=None,
        business_date_start=None, business_date_end=None,
        cost_date_start=None, cost_date_end=None,
        operation_date_start=None, operation_date_end=None,
        business_node_id=None, provider_id=None, cost_type_ids=None,
        billing_status=None, payment_status=None, verification_status=None,
        document_type=None, page=1, limit=20,
        request_id='rq_query_payable_cost_list',
    ):
        if self.api_client is None:
            raise RuntimeError('api client is required for query_payable_cost_list')
        business_type = self._integer(business_type, 'business_type')
        if business_type not in self.BUSINESS_TYPES:
            raise ValueError('business_type is invalid')
        payload = self._filters(
            business_type, order_numbers, tracking_numbers, container_codes,
            bl_numbers, so_numbers, business_date_start, business_date_end,
            cost_date_start, cost_date_end, operation_date_start,
            operation_date_end, business_node_id, provider_id, cost_type_ids,
            billing_status, payment_status, verification_status, document_type,
        )
        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 _filters(
        cls, business_type, order_numbers, tracking_numbers, container_codes,
        bl_numbers, so_numbers, business_date_start, business_date_end,
        cost_date_start, cost_date_end, operation_date_start, operation_date_end,
        business_node_id, provider_id, cost_type_ids, billing_status,
        payment_status, verification_status, document_type,
    ):
        values = {
            'order_numbers': order_numbers, 'tracking_numbers': tracking_numbers,
            'container_codes': container_codes, 'bl_numbers': bl_numbers,
            'so_numbers': so_numbers,
        }
        payload = {}
        payload['business_type'] = business_type
        total = 0
        for field in cls.NUMBER_FIELDS:
            if values[field] is None:
                continue
            cleaned = cls._number_list(values[field], field)
            payload[field] = cleaned
            total += len(values[field])
        if total > 200:
            raise ValueError('at most 200 numbers are allowed')

        date_pairs = (
            ('business_date_start', business_date_start, business_date_end),
            ('cost_date_start', cost_date_start, cost_date_end),
            ('operation_date_start', operation_date_start, operation_date_end),
        )
        complete_dates = False
        for field, start_value, end_value in date_pairs:
            start = cls._optional_date(start_value, field)
            end = cls._optional_date(end_value, field.replace('_start', '_end'))
            if bool(start) != bool(end):
                raise ValueError(field + ' range is invalid')
            if start:
                if (date.fromisoformat(end) - date.fromisoformat(start)).days > 30:
                    raise ValueError(field + ' range is invalid')
                if end < start:
                    raise ValueError(field + ' range is invalid')
                payload[field] = start
                payload[field.replace('_start', '_end')] = end
                complete_dates = True
        if total == 0 and not complete_dates:
            raise ValueError('number or date filters are required')

        for field, value, minimum in (
            ('business_node_id', business_node_id, 1),
            ('provider_id', provider_id, 1),
            ('document_type', document_type, 0),
        ):
            if value is not None:
                payload[field] = cls._integer(value, field, minimum)
        if cost_type_ids is not None:
            if not isinstance(cost_type_ids, list) or not cost_type_ids:
                raise ValueError('cost_type_ids is invalid')
            if len(cost_type_ids) > 200:
                raise ValueError('cost_type_ids is invalid')
            payload['cost_type_ids'] = [cls._integer(item, 'cost_type_ids', 1) for item in cost_type_ids]
        for field, value, allowed in (
            ('billing_status', billing_status, (0, 1)),
            ('payment_status', payment_status, (-1, 1)),
            ('verification_status', verification_status, (0, 1, 2)),
        ):
            if value is not None:
                checked = cls._integer(value, field)
                if checked not in allowed:
                    raise ValueError(field + ' is invalid')
                payload[field] = checked
        return payload

    @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')
        if re.fullmatch(r'[0-9]{4}-[0-9]{2}-[0-9]{2}', value) is None:
            raise ValueError(field + ' is invalid')
        try:
            date.fromisoformat(value)
        except ValueError:
            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

    @classmethod
    def _bounded_integer(cls, value, field):
        value = cls._integer(value, field, 1)
        if value > 100:
            raise ValueError(field + ' is invalid')
        return value