mcp/tools/query_receivable_cost_list.py

import re
from datetime import date


class QueryReceivableCostListTool:
    name = 'query_receivable_cost_list'
    route_path = '/mcp/tools/queryReceivableCostList'
    NUMBER_FIELDS = (
        'reference_numbers',
        'tracking_numbers',
        'order_numbers',
        'bill_numbers',
    )

    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}$',
        }
        return {
            'name': self.name,
            'description': (
                '分页查询当前员工有权查看的应收费用列表。仅可按已明确业务类型的参考号、'
                '跟踪号、订单号或账单编号精确查询；禁止根据格式猜测号码类型，禁止跨字段'
                '试查、并行试查或失败后改用其他号码字段。没有明确号码时必须提供不超过31个'
                '日历日的业务日期闭区间。客户、出账状态、核销状态、单据类型和费用项筛选值'
                '必须先调用list_receivable_cost_filter_options选择并使用其返回的value，'
                '不得猜测内部ID。'
            ),
            'input_schema': {
                'type': 'object',
                'properties': {
                    'reference_numbers': dict(number_array),
                    'tracking_numbers': dict(number_array),
                    'order_numbers': dict(number_array),
                    'bill_numbers': dict(number_array),
                    'business_date_start': dict(date_field),
                    'business_date_end': dict(date_field),
                    'customer_id': {'type': 'integer', 'minimum': 1},
                    'sub_customer_id': {'type': 'integer', 'minimum': 1},
                    'billing_status': {'type': 'integer', 'enum': [0, 1]},
                    'verification_status': {
                        'type': 'integer', 'enum': [-1, 0, 1],
                    },
                    'document_type': {'type': 'integer', 'minimum': 0},
                    'cost_type_id': {'type': 'integer', 'minimum': 1},
                    'page': {
                        'type': 'integer', 'minimum': 1, 'maximum': 100,
                        'default': 1,
                    },
                    'limit': {
                        'type': 'integer', 'minimum': 1, 'maximum': 100,
                        'default': 20,
                    },
                },
                'required': [],
                'allOf': [
                    {
                        'anyOf': [
                            {'required': ['reference_numbers']},
                            {'required': ['tracking_numbers']},
                            {'required': ['order_numbers']},
                            {'required': ['bill_numbers']},
                            {
                                'required': [
                                    'business_date_start',
                                    'business_date_end',
                                ],
                            },
                        ],
                    },
                    {
                        'if': {'required': ['sub_customer_id']},
                        'then': {'required': ['customer_id']},
                    },
                ],
                'additionalProperties': False,
            },
        }

    def call(
        self,
        reference_numbers=None,
        tracking_numbers=None,
        order_numbers=None,
        bill_numbers=None,
        business_date_start=None,
        business_date_end=None,
        customer_id=None,
        sub_customer_id=None,
        billing_status=None,
        verification_status=None,
        document_type=None,
        cost_type_id=None,
        page=1,
        limit=20,
        request_id='rq_query_receivable_cost_list',
    ):
        if self.api_client is None:
            raise RuntimeError(
                'api client is required for query_receivable_cost_list'
            )

        supplied_numbers = {}
        total_numbers = 0
        for field, values in (
            ('reference_numbers', reference_numbers),
            ('tracking_numbers', tracking_numbers),
            ('order_numbers', order_numbers),
            ('bill_numbers', bill_numbers),
        ):
            if values is not None:
                normalized = self._number_list(values, field)
                supplied_numbers[field] = normalized
                total_numbers += len(normalized)
        if total_numbers > 200:
            raise ValueError('combined exact number count is invalid')

        start = self._optional_date(business_date_start, 'business_date_start')
        end = self._optional_date(business_date_end, 'business_date_end')
        if bool(start) != bool(end):
            raise ValueError('business date range is invalid')
        if start and end:
            span = (date.fromisoformat(end) - date.fromisoformat(start)).days
            if span < 0 or span > 30:
                raise ValueError('business date range is invalid')
        if not supplied_numbers and not (start and end):
            raise ValueError('exact numbers or business date range is required')
        if sub_customer_id is not None and customer_id is None:
            raise ValueError('customer_id is required with sub_customer_id')

        payload = dict(supplied_numbers)
        if start:
            payload['business_date_start'] = start
            payload['business_date_end'] = end
        for field, value, minimum in (
            ('customer_id', customer_id, 1),
            ('sub_customer_id', sub_customer_id, 1),
            ('document_type', document_type, 0),
            ('cost_type_id', cost_type_id, 1),
        ):
            if value is not None:
                payload[field] = self._integer(value, field, minimum)
        for field, value, allowed in (
            ('billing_status', billing_status, (0, 1)),
            ('verification_status', verification_status, (-1, 0, 1)),
        ):
            if value is not None:
                value = self._integer(value, field)
                if value not in allowed:
                    raise ValueError('{0} is invalid'.format(field))
                payload[field] = value
        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 1 <= len(values) <= 200:
            raise ValueError('{0} is invalid'.format(field))
        result = []
        for value in values:
            if not isinstance(value, str):
                raise ValueError('{0} is invalid'.format(field))
            value = value.strip()
            if not value or len(value) > 100:
                raise ValueError('{0} is invalid'.format(field))
            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('{0} is invalid'.format(field))
        if re.fullmatch(r'[0-9]{4}-[0-9]{2}-[0-9]{2}', value) is None:
            raise ValueError('{0} is invalid'.format(field))
        try:
            date.fromisoformat(value)
        except ValueError:
            raise ValueError('{0} is invalid'.format(field))
        return value

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

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