115 lines
4.9 KiB
Python
115 lines
4.9 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""pipeline_llm.todos — 平台待办 provider：账号钱包余额不足（2026-09-11 用户定夺）。
|
||
|
||
钱包余额改变的两个目的（用户原话）：
|
||
1) 哪个钱包什么时候该充值了 → 本 provider 派生待办给属主机构 operator 催充值
|
||
2) 哪个钱包不能再调用 llm 了 → gateway._pick_candidate 门禁剔除（balance<阈值）
|
||
|
||
待办由**状态派生**（不建通知表）：balance < 阈值 的 active 账号即待办；充值
|
||
到账余额回升 → 待办自动消失。「同一事项只发一次」不变量天然满足（human-task-
|
||
issuance-gate 模式，与 ticket.todos 同款）。
|
||
|
||
接收人：账号属主机构（llm_account.org_id，空则回溯 llm_vendor.org_id）中持有
|
||
operator 角色的用户。角色名 params 可配（llm_wallet_todo_role，缺省 operator）
|
||
——角色/机构类型运行时可动态增删，禁硬编码（用户铁律）；roles 为
|
||
「{orgtypeid}.{name}」格式，按 . 后的名字匹配，不锁死 orgtypeid。
|
||
|
||
provider 签名（pipeline_service.human_task_capability.register_todo_provider）：
|
||
async fn(user_id, roles, limit) -> [todo dict]
|
||
"""
|
||
|
||
import logging
|
||
|
||
from .gateway import _get_db, _get_param
|
||
|
||
logger = logging.getLogger("pipeline_llm.todos")
|
||
|
||
_DEFAULT_ROLE = 'operator'
|
||
|
||
|
||
async def _wallet_threshold(sor) -> float:
|
||
v = await _get_param(sor, 'llm_wallet_balance_threshold', '5')
|
||
try:
|
||
return float(v)
|
||
except (TypeError, ValueError):
|
||
return 5.0
|
||
|
||
|
||
def _match_role(roles, want):
|
||
"""roles=['reseller.operator',...]；匹配 . 后的角色名（orgtypeid 运行时动态）。"""
|
||
for r in (roles or []):
|
||
name = str(r).split('.')[-1].strip().lower()
|
||
if name == want:
|
||
return True
|
||
return False
|
||
|
||
|
||
async def list_wallet_todos(user_id, roles=None, limit=100):
|
||
"""平台待办聚合器调用的 provider。roles 缺省时自查。"""
|
||
if not user_id:
|
||
return []
|
||
db, dbname = _get_db()
|
||
async with db.sqlorContext(dbname) as sor:
|
||
threshold = await _wallet_threshold(sor)
|
||
want_role = str(await _get_param(
|
||
sor, 'llm_wallet_todo_role', _DEFAULT_ROLE)).strip().lower()
|
||
# 用户机构（属主机构匹配用）
|
||
recs = await sor.sqlExe(
|
||
"SELECT orgid FROM users WHERE id=${u}$ LIMIT 1", {"u": str(user_id)})
|
||
await sor.sqlExe("COMMIT", {})
|
||
user_org = str(getattr(recs[0], 'orgid', '') or '') if recs else ''
|
||
if roles is None:
|
||
rrecs = await sor.sqlExe(
|
||
"SELECT r.orgtypeid, r.name FROM userrole ur JOIN role r ON ur.roleid=r.id "
|
||
"WHERE ur.userid=${u}$", {"u": str(user_id)})
|
||
await sor.sqlExe("COMMIT", {})
|
||
roles = []
|
||
for r in (rrecs or []):
|
||
o = getattr(r, 'orgtypeid', '') or ''
|
||
n = getattr(r, 'name', '') or ''
|
||
if o and n:
|
||
roles.append(f"{o}.{n}")
|
||
if not _match_role(roles, want_role):
|
||
return []
|
||
# 余额不足的 active 账号（属主机构=用户机构；账号 org_id 空回溯供应商 org_id）
|
||
recs = await sor.sqlExe(
|
||
"SELECT a.id, a.name, a.balance, a.org_id, a.vendor_id, v.name AS vendor_name, "
|
||
"COALESCE(NULLIF(a.org_id,''), v.org_id, '0') AS owner_org "
|
||
"FROM llm_account a LEFT JOIN llm_vendor v ON v.id=a.vendor_id "
|
||
"WHERE a.status='active' AND a.balance < ${t}$ "
|
||
"ORDER BY a.balance ASC LIMIT ${lim}$",
|
||
{"t": threshold, "lim": int(limit) * 4})
|
||
await sor.sqlExe("COMMIT", {})
|
||
|
||
todos = []
|
||
seen = 0
|
||
for r in (recs or []):
|
||
owner_org = str(getattr(r, 'owner_org', '') or '0')
|
||
if owner_org != (user_org or '0'):
|
||
continue # 只发给属主机构的人
|
||
if seen >= int(limit):
|
||
break
|
||
seen += 1
|
||
acc_id = str(getattr(r, 'id', '') or '')
|
||
balance = float(getattr(r, 'balance', 0) or 0)
|
||
acc_name = str(getattr(r, 'name', '') or acc_id)
|
||
vendor_name = str(getattr(r, 'vendor_name', '') or '')
|
||
todos.append({
|
||
'id': 'llm_wallet_' + acc_id,
|
||
'source': 'llm_wallet',
|
||
'task_type': 'llm_wallet_recharge',
|
||
'badge': '钱包充值',
|
||
'title': '【钱包充值】%s%s' % (
|
||
('%s·' % vendor_name) if vendor_name else '', acc_name),
|
||
'description': '模型供应商账号钱包余额 %.2f 元，低于门禁阈值 %.2f 元——'
|
||
'该账号已从模型调用轮询集剔除（充值到账后自动恢复并关闭本待办）'
|
||
% (balance, threshold),
|
||
'account_id': acc_id,
|
||
'balance': round(balance, 4),
|
||
'threshold': threshold,
|
||
'project_id': '',
|
||
'status': 'pending',
|
||
'created_at': None,
|
||
})
|
||
return todos
|