187 lines
7.2 KiB
Python
187 lines
7.2 KiB
Python
"""BaseAdapter - 适配器基类，封装 Worker 的完整生命周期"""
|
||
import asyncio
|
||
|
||
from .client import ArrayClient
|
||
from .models import Task, Event
|
||
|
||
|
||
class BaseAdapter:
|
||
"""适配器基类，封装 Worker 的完整生命周期：连接 -> 监听 -> 处理
|
||
|
||
关于 Worker 描述（description）的推荐格式：
|
||
在注册 Worker 或更新简历时，description 字段建议按以下结构编写，
|
||
方便其他 Worker 和人类快速理解你的能力。
|
||
|
||
推荐格式：
|
||
[一句话功能概述]
|
||
|
||
能力：
|
||
- 能力1
|
||
- 能力2
|
||
|
||
使用方式：
|
||
[说明怎么跟这个 Worker 交互]
|
||
|
||
限制：（可选）
|
||
- 限制1
|
||
|
||
示例：
|
||
"多语言实时翻译工具，支持中英日韩互译。\n\n"
|
||
"能力：\n"
|
||
"- 文本翻译（中/英/日/韩）\n"
|
||
"- 自动检测源语言\n\n"
|
||
"使用方式：\n"
|
||
"@提及并发送需要翻译的文本，默认翻译为中文。\n\n"
|
||
"限制：\n"
|
||
"- 单次最多翻译 5000 字"
|
||
|
||
实际使用时，可通过 client.update_profile(description="...") 或注册时传入。
|
||
|
||
关于组织（Organization）的说明：
|
||
Worker 可以属于多个组织，组织提供了资源隔离和协作边界。如果你的 Worker
|
||
属于多个组织，建议在适配器初始化时通过参数指定当前工作组织上下文：
|
||
|
||
class MyAdapter(BaseAdapter):
|
||
def __init__(self, server_url: str, api_key: str, org_id: str | None = None):
|
||
super().__init__(server_url, api_key)
|
||
self.org_id = org_id # 当前工作组织
|
||
|
||
创建任务时，可以通过 org_id 参数指定任务所属组织：
|
||
|
||
await self.client.create_task(
|
||
title="任务标题",
|
||
org_id=self.org_id # 指定组织
|
||
)
|
||
|
||
通过邀请码加入组织：
|
||
|
||
await self.client.join_organization(invite_code="xxx")
|
||
"""
|
||
|
||
def __init__(self, server_url: str, api_key: str):
|
||
self.server_url = server_url
|
||
self.api_key = api_key
|
||
self.client: ArrayClient | None = None
|
||
|
||
async def on_task_invited(self, task: Task):
|
||
"""被邀请加入新任务时的处理逻辑。子类必须实现。"""
|
||
raise NotImplementedError
|
||
|
||
async def on_task_assigned(self, task: Task):
|
||
"""兼容别名，推荐使用 on_task_invited"""
|
||
await self.on_task_invited(task)
|
||
|
||
async def on_task_message(self, task_id: str, message: dict):
|
||
"""收到任务消息时的处理逻辑。子类可选覆盖。"""
|
||
pass
|
||
|
||
async def on_task_updated(self, task_id: str, changes: dict):
|
||
"""任务状态/内容变更时的处理逻辑。子类可选覆盖。"""
|
||
pass
|
||
|
||
async def on_pending_recovery(self, tasks: list[dict]):
|
||
"""WebSocket 连上后收到待恢复任务列表时调用。子类可覆盖。
|
||
|
||
tasks 是 [{"id": "...", "title": "...", "status": "..."}] 形式的列表。
|
||
适配器可在此恢复之前中断的任务处理。
|
||
"""
|
||
pass
|
||
|
||
async def on_connected(self):
|
||
"""连接成功后的回调。子类可选覆盖。"""
|
||
pass
|
||
|
||
async def on_disconnected(self):
|
||
"""WebSocket 断开时的回调，子类可覆盖"""
|
||
pass
|
||
|
||
async def run(self):
|
||
"""启动 Worker 完整生命周期。
|
||
|
||
外层重连：当 HTTP 连接或整个 WebSocket 会话断开时（如服务端宕机），
|
||
在这里做指数退避重连（5s 起步，最大 120s），每次重连会重新认证、
|
||
处理离线通知、再启动 WebSocket 监听。
|
||
|
||
内层重连由 client.listen() 负责：当 WebSocket 连接断开但 HTTP
|
||
会话仍然有效时，listen() 内部自动重连（2s 起步，最大 60s）。
|
||
|
||
两层的区别：
|
||
- 内层（listen）：仅 WebSocket 断线重连，不涉及重新认证
|
||
- 外层（run）：完整的会话重建，包括重新认证和离线通知恢复
|
||
"""
|
||
import logging
|
||
logger = logging.getLogger(__name__)
|
||
|
||
self._running = True
|
||
delay = 5.0
|
||
while self._running:
|
||
try:
|
||
async with ArrayClient(self.server_url, self.api_key) as client:
|
||
self.client = client
|
||
delay = 5.0 # 重置退避
|
||
print(f"Worker {client.worker_id} 已连接到平台")
|
||
|
||
# 处理离线期间的通知
|
||
await self._process_offline_notifications()
|
||
|
||
# 启动 WebSocket 监听（在后台运行）
|
||
print("开始监听实时事件...")
|
||
listen_task = asyncio.create_task(self._listen_loop(client))
|
||
|
||
# 等待 WebSocket 连接建立
|
||
await client.wait_for_ws_connection()
|
||
await self.on_connected()
|
||
|
||
# 等待监听任务完成（通常不会完成，除非连接断开）
|
||
await listen_task
|
||
# 正常断开（非异常）
|
||
await self.on_disconnected()
|
||
except asyncio.CancelledError:
|
||
break
|
||
except Exception as e:
|
||
logger.warning(f"连接失败: {e}，{delay:.0f}s 后重试")
|
||
await self.on_disconnected()
|
||
await asyncio.sleep(delay)
|
||
delay = min(delay * 2, 120.0)
|
||
|
||
async def _listen_loop(self, client: ArrayClient):
|
||
"""WebSocket 监听循环"""
|
||
async for event in client.listen():
|
||
await self._handle_event(event)
|
||
|
||
async def _process_offline_notifications(self):
|
||
"""处理离线期间积累的通知"""
|
||
notifications, total = await self.client.get_notifications(is_read=False)
|
||
if notifications:
|
||
print(f"处理 {len(notifications)} 条离线通知...")
|
||
for n in notifications:
|
||
event = Event(event_type=n.event_type, payload=n.payload or {}, timestamp=str(n.created_at))
|
||
await self._handle_event(event)
|
||
await self.client.mark_notification_read(n.id)
|
||
|
||
async def _handle_event(self, event: Event):
|
||
"""分发事件到对应的处理方法"""
|
||
if event.event_type == "task.assigned":
|
||
task_id = event.payload.get("task_id")
|
||
if task_id:
|
||
task = await self.client.get_task(task_id)
|
||
await self.on_task_invited(task)
|
||
elif event.event_type == "participant.joined":
|
||
task_id = event.payload.get("task_id")
|
||
if task_id:
|
||
task = await self.client.get_task(task_id)
|
||
await self.on_task_invited(task)
|
||
elif event.event_type == "tasks.pending_recovery":
|
||
tasks = event.payload.get("tasks", [])
|
||
await self.on_pending_recovery(tasks)
|
||
elif event.event_type == "task.message":
|
||
task_id = event.payload.get("task_id")
|
||
if task_id:
|
||
await self.on_task_message(task_id, event.payload)
|
||
elif event.event_type == "task.updated":
|
||
task_id = event.payload.get("task_id")
|
||
changes = event.payload.get("changes", {})
|
||
if task_id:
|
||
await self.on_task_updated(task_id, changes)
|
||
|