wehub-resource-sync
293 lines
10 KiB
Python
293 lines
10 KiB
Python
"""
|
||
SMS app task definitions.
|
||
"""
|
||
# -- Task Index (auto-generated, do not edit) --
|
||
# 10 tasks | L1×2 L2×3 L3×3 L4×2
|
||
#
|
||
# [L2] ToggleMainSetting 把短信的{setting_key}设为{enabled}
|
||
# [L1] OpenConversationBySender 打开来自{conversation_id}的短信会话
|
||
# [L2] ReadUnreadConversationCount 数一下短信里现在有几个未读会话
|
||
# [L3] ReplyToConversation 给{sender}回复一条短信，内容是"{content}"
|
||
# [L3] MarkAllConversationsRead 把短信里的所有会话都标成已读
|
||
# [L1] ToggleFreeNetworkSetting 把短信里{setting_key}设为{enabled}
|
||
# [L2] CompareConversationMessageCount {sender1}和{sender2}这两个短信会话里，哪个消息更多
|
||
# [L4] DeleteConversation 帮我把{sender}的短信会话删掉
|
||
# [L4] ReplyToLatestUnread 看看最新的未读短信是哪个发来的，帮我回复他「{content}」
|
||
# [L3] FindAndReplySendersByKeyword 把之前给我发过提到{keyword}短信的人都找出来，统一回一句{reply}
|
||
# -- End Task Index --
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
from bench_env.task.base import BaseTask
|
||
from bench_env.task.common_tasks import AnswerTask, CriteriaTask
|
||
from bench_env.task.judge import JudgeInput
|
||
from bench_env.task.sms.app import (
|
||
SMS_EXISTING_SENDERS,
|
||
SMS_NEW_RECIPIENTS,
|
||
Sms,
|
||
sms_from_input,
|
||
sms_init_from_input,
|
||
)
|
||
class ToggleMainSetting(CriteriaTask):
|
||
templates = ["把短信的{setting_key}设为{enabled}"]
|
||
apps = ["sms"]
|
||
scope = "S1"
|
||
objective = "operate"
|
||
composition = "sequential"
|
||
difficulty = "L2"
|
||
capabilities = ["settings"]
|
||
parameters = {
|
||
"setting_key": {
|
||
"type": "enum",
|
||
"values": {
|
||
"免费网络短信(SIM卡1)": "free_sms_sim1",
|
||
"免费网络短信(SIM卡2)": "free_sms_sim2",
|
||
"5G消息": "5g_message",
|
||
"列表中显示头像": "show_avatar",
|
||
"文字头像": "text_avatar",
|
||
},
|
||
"default": "show_avatar",
|
||
},
|
||
"enabled": {
|
||
"type": "bool",
|
||
"values": {"开启": True, "关闭": False},
|
||
"default": False,
|
||
},
|
||
}
|
||
criteria = {"settings.{setting_key}": "{enabled}"}
|
||
|
||
async def _post_sample(self, env) -> None:
|
||
await self._invert_criteria(env)
|
||
|
||
|
||
class OpenConversationBySender(CriteriaTask):
|
||
templates = ["打开来自{conversation_id}的短信会话"]
|
||
apps = ["sms"]
|
||
scope = "S1"
|
||
objective = "operate"
|
||
composition = "atomic"
|
||
difficulty = "L1"
|
||
capabilities = ["nav"]
|
||
expected_changes = ["os.providers.sms.conversations"]
|
||
parameters = {
|
||
"conversation_id": {
|
||
"type": "enum",
|
||
"values": {
|
||
"华为云": "huawei-cloud",
|
||
"抖音月付": "douyin-pay",
|
||
"10690955998": "abc-bank",
|
||
"建设银行": "ccb-bank",
|
||
"中国电信": "china-telecom",
|
||
"中国联通": "china-unicom",
|
||
},
|
||
"default": "china-telecom",
|
||
},
|
||
}
|
||
criteria = {"route": "/conversation/{conversation_id}"}
|
||
|
||
|
||
class ReadUnreadConversationCount(AnswerTask):
|
||
templates = ["数一下短信里现在有几个未读会话"]
|
||
apps = ["sms"]
|
||
scope = "S1"
|
||
objective = "query"
|
||
composition = "atomic"
|
||
difficulty = "L2"
|
||
capabilities = ["extract"]
|
||
answer_fields = [{"type": "number", "label": "未读会话数"}]
|
||
|
||
def get_answer(self, input: JudgeInput) -> Any:
|
||
return sms_init_from_input(input).unread_conversation_count
|
||
class ReplyToConversation(BaseTask):
|
||
templates = [
|
||
'给{sender}回复一条短信，内容是"{content}"',
|
||
'Reply to {sender} with a message saying "{content}"',
|
||
]
|
||
apps = ["sms"]
|
||
scope = "S1"
|
||
objective = "operate"
|
||
composition = "sequential"
|
||
difficulty = "L3"
|
||
capabilities = ["create"]
|
||
parameters = {
|
||
"sender": {
|
||
"type": "enum",
|
||
"values": SMS_EXISTING_SENDERS,
|
||
"default": "中国联通",
|
||
},
|
||
"content": {
|
||
"type": "string",
|
||
"default": "我知道了",
|
||
},
|
||
}
|
||
expected_changes = ["os.providers.sms.conversations", "os.providers.sms.messagesByConversationId"]
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
sms = sms_from_input(input)
|
||
return [
|
||
sms.check_new_sent_to(self.p.sender, self.p.content, field="reply_message"),
|
||
]
|
||
|
||
|
||
class MarkAllConversationsRead(BaseTask):
|
||
templates = [
|
||
"把短信里的所有会话都标成已读",
|
||
"Mark all SMS conversations as read",
|
||
]
|
||
apps = ["sms"]
|
||
scope = "S1"
|
||
objective = "operate"
|
||
composition = "atomic"
|
||
difficulty = "L3"
|
||
capabilities = ["edit"]
|
||
expected_changes = ["os.providers.sms.conversations"]
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
sms = sms_from_input(input)
|
||
return [sms.check_all_read()]
|
||
|
||
|
||
class ToggleFreeNetworkSetting(CriteriaTask):
|
||
templates = ["把短信里{setting_key}设为{enabled}"]
|
||
apps = ["sms"]
|
||
scope = "S1"
|
||
objective = "operate"
|
||
composition = "sequential"
|
||
difficulty = "L2"
|
||
capabilities = ["settings"]
|
||
parameters = {
|
||
"setting_key": {
|
||
"type": "enum",
|
||
"values": {
|
||
"自动转为短信": "auto_convert_sms",
|
||
"自动转为彩信": "auto_convert_mms",
|
||
"屏蔽陌生人的网络短信": "block_strangers",
|
||
},
|
||
"default": "block_strangers",
|
||
},
|
||
"enabled": {
|
||
"type": "bool",
|
||
"values": {"开启": True, "关闭": False},
|
||
"default": False,
|
||
},
|
||
}
|
||
criteria = {"settings.{setting_key}": "{enabled}"}
|
||
|
||
async def _post_sample(self, env) -> None:
|
||
await self._invert_criteria(env)
|
||
class CompareConversationMessageCount(AnswerTask):
|
||
templates = [
|
||
"{sender1}和{sender2}这两个短信会话里，哪个消息更多",
|
||
"帮我看看{sender1}和{sender2}的短信会话，谁那边的消息条数更多",
|
||
]
|
||
apps = ["sms"]
|
||
scope = "S1"
|
||
objective = "query"
|
||
composition = "deep_dive"
|
||
difficulty = "L2"
|
||
capabilities = ["extract", "reasoning"]
|
||
expected_changes = ["os.providers.sms.conversations"]
|
||
answer_fields = [{"type": "choice", "label": "消息更多的会话", "options": ["{sender1}", "{sender2}"]}]
|
||
parameters = {
|
||
"sender1": {
|
||
"type": "string",
|
||
"default": "中国电信",
|
||
},
|
||
"sender2": {
|
||
"type": "string",
|
||
"default": "中国联通",
|
||
},
|
||
"_pair": {
|
||
"sampler": Sms.sample_compare_pair,
|
||
"fields": {"sender1": "sender1", "sender2": "sender2"},
|
||
},
|
||
}
|
||
|
||
def get_answer(self, input: JudgeInput) -> Any:
|
||
sms = sms_init_from_input(input)
|
||
return sms.sender_with_more_messages(self.p.sender1, self.p.sender2)
|
||
class DeleteConversation(BaseTask):
|
||
templates = [
|
||
"帮我把{sender}的短信会话删掉",
|
||
"删除{sender}的短信会话",
|
||
"Delete the SMS conversation with {sender}",
|
||
"Remove the SMS conversation from {sender}",
|
||
]
|
||
apps = ["sms"]
|
||
scope = "S1"
|
||
objective = "operate"
|
||
composition = "deep_dive"
|
||
difficulty = "L4"
|
||
capabilities = ["edit"]
|
||
parameters = {
|
||
"sender": {
|
||
"type": "enum",
|
||
"values": SMS_EXISTING_SENDERS,
|
||
"default": "建设银行",
|
||
},
|
||
}
|
||
expected_changes = ["os.providers.sms.conversations", "os.providers.sms.messagesByConversationId"]
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
sms = sms_from_input(input)
|
||
return [sms.check_conversation_deleted(self.p.sender)]
|
||
|
||
|
||
class ReplyToLatestUnread(BaseTask):
|
||
templates = [
|
||
"看看最新的未读短信是哪个发来的，帮我回复他「{content}」",
|
||
"帮我打开最新的未读短信会话，回复“{content}”",
|
||
'Find the latest unread SMS and reply with "{content}"',
|
||
'Open the most recent unread SMS conversation and reply "{content}"',
|
||
]
|
||
apps = ["sms"]
|
||
scope = "S1"
|
||
objective = "operate"
|
||
composition = "deep_dive"
|
||
difficulty = "L4"
|
||
capabilities = ["extract", "create"]
|
||
parameters = {
|
||
"content": {
|
||
"type": "string",
|
||
"default": "好的收到",
|
||
},
|
||
}
|
||
expected_changes = ["os.providers.sms.conversations", "os.providers.sms.messagesByConversationId"]
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
sms = sms_from_input(input)
|
||
target_sender = sms_init_from_input(input).latest_unread_sender()
|
||
return [sms.check_new_sent_to(target_sender, self.p.content, field="reply_to_unread")]
|
||
|
||
|
||
class FindAndReplySendersByKeyword(BaseTask):
|
||
"""验证 Agent 是否找出所有含关键词短信的发送方并统一回复。
|
||
|
||
判定：使用 Sms.senders_with_keyword 获取目标发送方列表，
|
||
再用 Sms.check_replied_to_all 验证每个目标都收到了包含 reply 的新消息。
|
||
"""
|
||
|
||
templates = [
|
||
"把之前给我发过提到{keyword}短信的人都找出来，统一回一句{reply}",
|
||
"帮我找出短信里提到过{keyword}的发送方，给每个人回复{reply}",
|
||
"Find all senders who previously sent me SMS messages mentioning {keyword}, and reply to each of them with {reply}",
|
||
"Look through SMS for messages containing {keyword}, and send {reply} to each of those senders",
|
||
]
|
||
apps = ["sms"]
|
||
scope = "S1"
|
||
objective = "operate"
|
||
composition = "deep_dive"
|
||
difficulty = "L3"
|
||
capabilities = ["extract", "create"]
|
||
parameters = {
|
||
"keyword": {"type": "string", "default": "套餐"},
|
||
"reply": {"type": "string", "default": "拒收"},
|
||
}
|
||
expected_changes = ["os.providers.sms.messagesByConversationId", "os.providers.sms.conversations"]
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
sms = sms_from_input(input)
|
||
senders = sms.senders_with_keyword(self.p.keyword)
|
||
return [sms.check_replied_to_all(senders, self.p.reply)] |