跳转至

payment_accounts

feishu.agent.payment_accounts

收款账户（payment account）解析：从用户本人历史审批实例中恢复账户控件值，句柄优先、内存暂存、严格按用户隔离。

飞书开放平台文档把 account（收款账户）列为创建审批实例 API 不直接支持的控件；但用户本人过去提交的报销实例 里会带完整账户控件值。这个模块只从请求用户本人历史实例读取账户值，向模型只暴露不可逆句柄和脱敏标签；完整 账户值仅保存在内存中，并只在提交瞬间还原进 account 控件。

PaymentAccount dataclass

一个收款账户：模型可见的句柄和脱敏标签，外加仅服务端可见的完整控件值。

源代码位于： feishu/agent/payment_accounts.py
Python
73
74
75
76
77
78
79
80
81
82
83
84
@dataclass
class PaymentAccount:
    r"""一个收款账户：模型可见的句柄和脱敏标签，外加仅服务端可见的完整控件值。"""

    account_id: str
    label: str
    account_value: dict[str, Any]
    user_keys: tuple[str, ...] = ()

    def summary(self) -> dict[str, str]:
        r"""模型安全视图：仅句柄和脱敏标签，绝不含完整账户值或卡号。"""
        return {"account_id": self.account_id, "label": self.label}

summary

Python
summary() -> dict[str, str]

模型安全视图：仅句柄和脱敏标签，绝不含完整账户值或卡号。

源代码位于： feishu/agent/payment_accounts.py
Python
82
83
84
def summary(self) -> dict[str, str]:
    r"""模型安全视图：仅句柄和脱敏标签，绝不含完整账户值或卡号。"""
    return {"account_id": self.account_id, "label": self.label}

PaymentAccountResolver

从请求用户本人的历史审批实例解析收款账户，句柄优先、内存暂存、严格按用户隔离。

recent 经 feishu.approval.instances.InstancesNamespace.query 按用户 open_id 拉取本人实例，再抽取账户值、 去重、生成句柄与脱敏标签；完整账户值仅缓存在内存。resolve 仅在提交瞬间把句柄还原为完整值，供 feishu.agent.toolkit.approvals.create_approval_instance 填入账户控件。

参数：

名称 类型 描述 默认

client

Any

飞书客户端，提供 approval.instances.query / approval.instances.get 以拉取并读取本人历史实例。

 必需

lookback_ms

int

历史实例的回溯窗口（毫秒），即 query 的 start_time 相对当前时刻的回看跨度。 默认为 730 * 24 * 3600 * 1000，即约 730 天（约 2 年）的历史回溯窗口。

 730 * 24 * 3600 * 1000
源代码位于： feishu/agent/payment_accounts.py
Python
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
class PaymentAccountResolver:
    r"""
    从请求用户本人的历史审批实例解析收款账户，句柄优先、内存暂存、严格按用户隔离。

    `recent` 经 [feishu.approval.instances.InstancesNamespace.query][] 按用户 `open_id` 拉取本人实例，再抽取账户值、
    去重、生成句柄与脱敏标签；完整账户值仅缓存在内存。`resolve` 仅在提交瞬间把句柄还原为完整值，供
    [feishu.agent.toolkit.approvals.create_approval_instance][] 填入账户控件。

    Args:
        client: 飞书客户端，提供 `approval.instances.query` / `approval.instances.get` 以拉取并读取本人历史实例。
        lookback_ms: 历史实例的回溯窗口（毫秒），即 `query` 的 `start_time` 相对当前时刻的回看跨度。
            默认为 `730 * 24 * 3600 * 1000`，即约 730 天（约 2 年）的历史回溯窗口。
    """

    def __init__(self, client: Any, *, lookback_ms: int = 730 * 24 * 3600 * 1000) -> None:
        self._client = client
        self._lookback_ms = lookback_ms
        self._cache: dict[str, dict[str, PaymentAccount]] = {}

    async def recent(
        self, user: Mapping[str, Any], *, approval_code: str | None = None, limit: int = 10
    ) -> list[PaymentAccount]:
        r"""返回请求用户本人历史里出现过的去重收款账户（句柄 + 脱敏标签）。"""
        user_keys = payment_account_keys(user)
        open_id = user.get("open_id") if isinstance(user, Mapping) else None
        if not user_keys or not open_id:
            return []
        end = int(time.time() * 1000)
        start = end - self._lookback_ms
        try:
            items = await self._client.approval.instances.query(
                user_id=open_id,
                approval_code=approval_code,
                user_id_type="open_id",
                start_time=str(start),
                end_time=str(end),
                max_items=50,
            )
        except Exception:  # noqa: BLE001 - account discovery must not crash the tool; degrade to no results.
            logging.getLogger("feishu").debug("payment account query failed for the requesting user", exc_info=True)
            return []

        accounts: dict[str, PaymentAccount] = {}
        for item in items:
            instance_node = item.get("instance") if isinstance(item, Mapping) else None
            code = instance_node.get("code") if isinstance(instance_node, Mapping) else None
            if not code:
                continue
            try:
                instance = await self._client.approval.instances.get(code)
            except Exception:  # noqa: BLE001 - skip historical instances that cannot be read.
                logging.getLogger("feishu").debug(
                    "could not read instance %s for payment accounts", code, exc_info=True
                )
                continue
            if not _instance_belongs_to(instance, user):
                continue
            for widget in approval_account_widgets(instance):
                value = widget.get("value")
                if not _is_reusable_account_value(value):
                    logging.getLogger("feishu").debug(
                        "skipping non-reusable payment account value from instance %s", code
                    )
                    continue
                number = approval_account_number(value)
                if not number:
                    continue
                handle = payment_account_handle(number)
                if handle not in accounts:
                    accounts[handle] = PaymentAccount(
                        account_id=handle,
                        label=approval_account_label(value),
                        account_value=dict(value),
                        user_keys=user_keys,
                    )
                if len(accounts) >= limit:
                    break
            if len(accounts) >= limit:
                break
        self._cache.setdefault(user_keys[0], {}).update(accounts)
        return list(accounts.values())

    async def resolve(self, user: Mapping[str, Any], account_id: str) -> dict[str, Any] | None:
        r"""把句柄还原为完整账户控件值，严格限定为请求用户本人。"""
        user_keys = payment_account_keys(user)
        if not user_keys:
            return None
        primary = user_keys[0]
        account = self._cache.get(primary, {}).get(account_id)
        if account is None:
            await self.recent(user)
            account = self._cache.get(primary, {}).get(account_id)
        if account is None:
            return None
        if not set(user_keys) & set(account.user_keys):
            return None
        return dict(account.account_value)

recent async

Python
recent(user: Mapping[str, Any], *, approval_code: str | None = None, limit: int = 10) -> list[PaymentAccount]

返回请求用户本人历史里出现过的去重收款账户（句柄 + 脱敏标签）。

源代码位于： feishu/agent/payment_accounts.py
Python
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
async def recent(
    self, user: Mapping[str, Any], *, approval_code: str | None = None, limit: int = 10
) -> list[PaymentAccount]:
    r"""返回请求用户本人历史里出现过的去重收款账户（句柄 + 脱敏标签）。"""
    user_keys = payment_account_keys(user)
    open_id = user.get("open_id") if isinstance(user, Mapping) else None
    if not user_keys or not open_id:
        return []
    end = int(time.time() * 1000)
    start = end - self._lookback_ms
    try:
        items = await self._client.approval.instances.query(
            user_id=open_id,
            approval_code=approval_code,
            user_id_type="open_id",
            start_time=str(start),
            end_time=str(end),
            max_items=50,
        )
    except Exception:  # noqa: BLE001 - account discovery must not crash the tool; degrade to no results.
        logging.getLogger("feishu").debug("payment account query failed for the requesting user", exc_info=True)
        return []

    accounts: dict[str, PaymentAccount] = {}
    for item in items:
        instance_node = item.get("instance") if isinstance(item, Mapping) else None
        code = instance_node.get("code") if isinstance(instance_node, Mapping) else None
        if not code:
            continue
        try:
            instance = await self._client.approval.instances.get(code)
        except Exception:  # noqa: BLE001 - skip historical instances that cannot be read.
            logging.getLogger("feishu").debug(
                "could not read instance %s for payment accounts", code, exc_info=True
            )
            continue
        if not _instance_belongs_to(instance, user):
            continue
        for widget in approval_account_widgets(instance):
            value = widget.get("value")
            if not _is_reusable_account_value(value):
                logging.getLogger("feishu").debug(
                    "skipping non-reusable payment account value from instance %s", code
                )
                continue
            number = approval_account_number(value)
            if not number:
                continue
            handle = payment_account_handle(number)
            if handle not in accounts:
                accounts[handle] = PaymentAccount(
                    account_id=handle,
                    label=approval_account_label(value),
                    account_value=dict(value),
                    user_keys=user_keys,
                )
            if len(accounts) >= limit:
                break
        if len(accounts) >= limit:
            break
    self._cache.setdefault(user_keys[0], {}).update(accounts)
    return list(accounts.values())

resolve async

Python
resolve(user: Mapping[str, Any], account_id: str) -> dict[str, Any] | None

把句柄还原为完整账户控件值，严格限定为请求用户本人。

源代码位于： feishu/agent/payment_accounts.py
Python
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
async def resolve(self, user: Mapping[str, Any], account_id: str) -> dict[str, Any] | None:
    r"""把句柄还原为完整账户控件值，严格限定为请求用户本人。"""
    user_keys = payment_account_keys(user)
    if not user_keys:
        return None
    primary = user_keys[0]
    account = self._cache.get(primary, {}).get(account_id)
    if account is None:
        await self.recent(user)
        account = self._cache.get(primary, {}).get(account_id)
    if account is None:
        return None
    if not set(user_keys) & set(account.user_keys):
        return None
    return dict(account.account_value)

payment_account_keys

Python
payment_account_keys(user: Mapping[str, Any]) -> tuple[str, ...]

用户的稳定多别名键；feishu.auth.user_tokens.user_identity_keys 的别名。

源代码位于： feishu/agent/payment_accounts.py
Python
43
44
45
def payment_account_keys(user: Mapping[str, Any]) -> tuple[str, ...]:
    r"""用户的稳定多别名键；[feishu.auth.user_tokens.user_identity_keys][] 的别名。"""
    return user_identity_keys(user)

payment_account_handle

Python
payment_account_handle(number: str) -> str

把卡号映射为稳定、不可逆的句柄（pa_<sha256[:16]>）；模型只见句柄，绝不见卡号。

源代码位于： feishu/agent/payment_accounts.py
Python
48
49
50
def payment_account_handle(number: str) -> str:
    r"""把卡号映射为稳定、不可逆的句柄（`pa_<sha256[:16]>`）；模型只见句柄，绝不见卡号。"""
    return "pa_" + hashlib.sha256(number.encode("utf-8")).hexdigest()[:16]