mirrors / purewhiter--mobilegym

项目文件夹

purewhiter--mobilegym
文件
main
T
wehub-resource-sync 2114b14ee0
Sync main into demo / sync (push) Has been cancelled
详情
chore: import upstream snapshot with attribution
2026-07-13 12:35:26 +08:00

293 行
10 KiB
Python
原始文件 永久链接 Blame 文件历史

此文件含有模棱两可的 Unicode 字符
此文件含有可能会与其他字符混淆的 Unicode 字符。 如果您是想特意这样的，可以安全地忽略该警告。 使用 Escape 按钮显示他们。
"""
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)]
在新工单中引用 查看 Git Blame 复制永久链接