wehub-resource-sync
712 lines
32 KiB
Python
712 lines
32 KiB
Python
"""
|
||
Cross-app Content & Social tasks.
|
||
"""
|
||
# -- Task Index (auto-generated, do not edit) --
|
||
# 21 tasks | L2×4 L3×7 L4×10
|
||
#
|
||
# [L2] SpotifyNowPlayingToWechat 把 Spotify 当前播放的歌加入喜欢，再把歌名微信发给{contact}
|
||
# [L3] BilibiliRankingToWechat 看看B站{partition}区排行榜第{rank}名是什么视频，把标题微信发给{contact}
|
||
# [L3] RedbookSearchTitleToWechat 在小红书搜'{keyword}'，把第一篇笔记的标题微信发给{contact}
|
||
# [L4] SpotifyTodayNthPlayToRedbook 查看我今天在 Spotify 听的第{nth}首歌，在小红书发一篇推荐笔记，标题或正文包含歌名和艺人
|
||
# [L3] WechatReadingBestBookToWechat 帮我在微信读书{category}分类下找推荐值最高的书，把书名和推荐值微信发给{contact}
|
||
# [L4] WechatReadingStatsToWechat 查微信读书最近一周内阅读时长最多的一天是哪天读了多久，告诉微信好友{contact}
|
||
# [L3] RedbookAuthorFollowersToWechat 在小红书搜'{keyword}'，关注第一篇笔记的作者，并将作者名字和粉丝数微信发给{contact}
|
||
# [L4] XLatestPostToReddit_WithTitleFormat 把 X 用户 {user} 最新一条推文的文字内容，以"{user}:"开头发到 Reddit 的 r/{subreddit}。
|
||
# [L4] RedbookFollowingNoteCountToSms 查小红书我关注的'{username}'发了多少篇笔记，发短信告诉{contact}
|
||
# [L4] SpotifySongFullDetailsToRedbook 在Spotify搜'{song}'查下是谁唱的、几分钟，在小红书发一篇听歌笔记把这些写进去
|
||
# [L3] BilibiliTripleLikeThenMoments 在B站{partition}排行榜找到第{rank}名给它一键三连，然后发个纯文字朋友圈推荐这个视频
|
||
# [L2] RedbookDmThenWechatReport 给小红书上我关注的'{username}'发私信'{message}'，然后在微信告诉{contact}已经联系他了
|
||
# [L2] NotesContentToRedbookAndX 在笔记里写一段关于{topic}的想法，然后分别在小红书和X上发布出来
|
||
# [L4] DailyLogToMoments 把我笔记里最新两条笔记简单汇总一下，发一条朋友圈。
|
||
# [L4] CulturalChecklistToRedbook 看看Spotify我今天最早听的那首歌是什么，再看看微信读书热搜第一本书叫什么，在笔记里记一份'今日文化清单'，然后在小红书发一篇笔记分享
|
||
# [L3] EbayCheapToRedbook 帮我在 eBay 看看{product}里最便宜的那款，然后发一篇小红书商品推荐笔记。
|
||
# [L2] SpotifySaveCurrentSongToNotes 把 Spotify 正在播放的歌名和歌手记到笔记里
|
||
# [L3] WechatReadingShareBookList 把微信读书书架前{n}本书的名字微信发给{contact}
|
||
# [L4] ReadingPlanToNotes 看看微信读书里我正在读的书，然后在笔记里制定一个本周的阅读计划。
|
||
# [L4] FileManagerSendFileToWechatContact 我有一张图片分别在两个目录下各有一张不同名的副本，帮我找出他们，并把这两个文件名发给微信联系人{contact}
|
||
# [L4] NotesToWechatAndRedbook 把包含"{text_keyword}"的内容记到笔记后，再用微信同步发给{contact}，并发布一条对应的小红书笔记。
|
||
# -- 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 match_track_duration
|
||
from bench_env.task.bilibili.app import BILIBILI_PARTITION_PARAM, Bilibili
|
||
from bench_env.task.ebay.app import EBAY_SEARCH_CHANGES, EBAY_SEARCH_QUERY_PARAM, Ebay
|
||
from bench_env.task.judge import JudgeInput
|
||
from bench_env.task.notes.app import NOTES_CREATE_CHANGES, Notes
|
||
from bench_env.task.redbook.app import REDBOOK_FOLLOWING_USER_PARAM, REDBOOK_KEYWORD_PARAM, REDBOOK_PUBLISH_CHANGES, Redbook
|
||
from bench_env.task.reddit.app import Reddit
|
||
from bench_env.task.sms.app import SMS_RECIPIENT_PARAM, SMS_SEND_CHANGES, sms_from_input
|
||
from bench_env.task.spotify.app import SPOTIFY_ARTIST_PARAM, SPOTIFY_QUERY_CHANGES, Spotify
|
||
from bench_env.task.utils import count_titles_in_text, norm, to_simplified
|
||
from bench_env.task.wechat.app import WECHAT_CONTACT_PARAM, WECHAT_MOMENT_CHANGES, WECHAT_SEND_CHANGES, Wechat
|
||
from bench_env.task.wechat_reading.app import (
|
||
WECHAT_READING_BOOK_PARAM,
|
||
WECHAT_READING_CATEGORY_PARAM,
|
||
WECHAT_READING_UI_TO_DATA,
|
||
WechatReading,
|
||
format_words,
|
||
)
|
||
from bench_env.task.x.app import X, X_POST_CHANGES
|
||
|
||
|
||
|
||
class SpotifyNowPlayingToWechat(BaseTask):
|
||
templates = [
|
||
"把 Spotify 当前播放的歌加入喜欢，再把歌名微信发给{contact}",
|
||
"Add the currently playing song on Spotify to liked songs, then send the song name to {contact} on WeChat",
|
||
]
|
||
apps = ["spotify", "wechat"]
|
||
scope = "S2"
|
||
objective = "hybrid"
|
||
composition = "transfer"
|
||
difficulty = "L2"
|
||
capabilities = ["extract", "social", "handoff"]
|
||
parameters = {"contact": WECHAT_CONTACT_PARAM}
|
||
expected_changes = ["apps.spotify"] + WECHAT_SEND_CHANGES
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
spotify_init = Spotify(input.apps_init["spotify"])
|
||
spotify = Spotify(input.apps["spotify"])
|
||
wechat = Wechat(input.apps["wechat"], init=input.apps_init["wechat"])
|
||
title = spotify_init.current_track_title
|
||
if not title:
|
||
raise ValueError("任务设计错误：spotify.currentTrack 为空。")
|
||
return [
|
||
spotify.check_in_liked(title, field="spotify_liked"),
|
||
wechat.check_new_sent_contains(self.p.contact, title, field="spotify_now_playing"),
|
||
]
|
||
|
||
|
||
class BilibiliRankingToWechat(BaseTask):
|
||
templates = [
|
||
"看看B站{partition}区排行榜第{rank}名是什么视频，把标题微信发给{contact}",
|
||
"帮我查B站{partition}分区排行榜第{rank}名叫什么，发给微信的{contact}",
|
||
"Check what the #{rank} video is on Bilibili's {partition} ranking, then send its title to {contact} on WeChat",
|
||
"Look up the #{rank} ranked video in Bilibili's {partition} section and send the title to {contact} via WeChat",
|
||
]
|
||
apps = ["bilibili", "wechat"]
|
||
scope = "S2"
|
||
objective = "hybrid"
|
||
composition = "transfer"
|
||
difficulty = "L3"
|
||
max_steps = 30
|
||
capabilities = ["extract", "handoff"]
|
||
parameters = {
|
||
"partition": BILIBILI_PARTITION_PARAM,
|
||
"rank": {"type": "int", "default": 1},
|
||
"contact": WECHAT_CONTACT_PARAM,
|
||
}
|
||
expected_changes = WECHAT_SEND_CHANGES + ["apps.bilibili.activeVideoId"]
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
wechat = Wechat(input.apps["wechat"], init=input.apps_init["wechat"])
|
||
title = Bilibili.ranking_title(self.p.partition, int(self.p.rank))
|
||
return [wechat.check_new_sent_contains(self.p.contact, title, field="bili_ranking_title")]
|
||
|
||
|
||
class RedbookSearchTitleToWechat(BaseTask):
|
||
templates = ["在小红书搜'{keyword}'，把第一篇笔记的标题微信发给{contact}"]
|
||
apps = ["redbook", "wechat"]
|
||
scope = "S2"
|
||
objective = "hybrid"
|
||
composition = "transfer"
|
||
difficulty = "L3"
|
||
max_steps = 30
|
||
capabilities = ["search", "extract", "handoff"]
|
||
parameters = {"keyword": REDBOOK_KEYWORD_PARAM, "contact": WECHAT_CONTACT_PARAM}
|
||
expected_changes = WECHAT_SEND_CHANGES + ["redbook.searchHistory", "redbook.history"]
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
rb = Redbook(input.apps["redbook"], init=input.apps_init["redbook"])
|
||
wechat = Wechat(input.apps["wechat"], init=input.apps_init["wechat"])
|
||
note = rb.first_search_note(self.p.keyword)
|
||
return [
|
||
wechat.check_new_sent_norm_contains(
|
||
self.p.contact,
|
||
str(note["title"]),
|
||
field="redbook_title_share",
|
||
last_only=True,
|
||
)
|
||
]
|
||
class SpotifyTodayNthPlayToRedbook(BaseTask):
|
||
templates = [
|
||
"查看我今天在 Spotify 听的第{nth}首歌，在小红书发一篇推荐笔记，标题或正文包含歌名和艺人",
|
||
]
|
||
apps = ["spotify", "redbook"]
|
||
scope = "S2"
|
||
objective = "operate"
|
||
composition = "transfer"
|
||
difficulty = "L4"
|
||
max_steps = 45
|
||
capabilities = ["extract", "create", "handoff"]
|
||
parameters = {
|
||
"nth": {
|
||
"type": "enum",
|
||
"values": {"一": 1, "二": 2, "三": 3},
|
||
"default": 1,
|
||
},
|
||
}
|
||
expected_changes = ["apps.spotify"] + REDBOOK_PUBLISH_CHANGES
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
spotify = Spotify(input.apps_init["spotify"])
|
||
rb = Redbook(input.apps["redbook"], init=input.apps_init["redbook"])
|
||
# 保留自然用户话术“今天第 n 首”，但 judge 仍按最近播放页可见顺序近似。
|
||
track = spotify.nth_today_play(int(self.p.nth))
|
||
song_title = str(track["title"])
|
||
artist = str(track["artist"])
|
||
# 任务要求"标题或正文包含歌名和艺人"，用 text_keywords 在 title+desc 合并文本中检查
|
||
return [
|
||
rb.check_note_published(
|
||
text_keywords=(song_title, artist),
|
||
new_only=True,
|
||
field="redbook_today_nth_play",
|
||
)
|
||
]
|
||
class WechatReadingBestBookToWechat(BaseTask):
|
||
templates = ["帮我在微信读书{category}分类下找推荐值最高的书，把书名和推荐值微信发给{contact}"]
|
||
apps = ["wechat_reading", "wechat"]
|
||
scope = "S2"
|
||
objective = "hybrid"
|
||
composition = "transfer"
|
||
difficulty = "L3"
|
||
capabilities = ["search", "extract", "handoff"]
|
||
parameters = {"category": WECHAT_READING_CATEGORY_PARAM, "contact": WECHAT_CONTACT_PARAM}
|
||
expected_changes = WECHAT_SEND_CHANGES
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
wr = WechatReading(input.apps["wechat_reading"])
|
||
wechat = Wechat(input.apps["wechat"], init=input.apps_init["wechat"])
|
||
data_cats = WECHAT_READING_UI_TO_DATA.get(self.p.category, [self.p.category])
|
||
books = [b for b in wr.store if str(b.get("category")) in data_cats]
|
||
if not books:
|
||
raise ValueError(f"No books in UI category '{self.p.category}'")
|
||
books.sort(key=lambda b: float(b.get("rating") or 0), reverse=True)
|
||
book = books[0]
|
||
rv = str(book.get("recommendedValue", ""))
|
||
return [
|
||
wechat.check_new_sent_contains(
|
||
self.p.contact,
|
||
str(book["title"]),
|
||
rv,
|
||
field="best_book_share",
|
||
)
|
||
]
|
||
class WechatReadingStatsToWechat(BaseTask):
|
||
templates = ["查微信读书最近一周内阅读时长最多的一天是哪天读了多久，告诉微信好友{contact}"]
|
||
apps = ["wechat_reading", "wechat"]
|
||
scope = "S2"
|
||
objective = "hybrid"
|
||
composition = "transfer"
|
||
difficulty = "L4"
|
||
max_steps = 45
|
||
capabilities = ["extract", "handoff"]
|
||
parameters = {"contact": WECHAT_CONTACT_PARAM}
|
||
expected_changes = WECHAT_SEND_CHANGES
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
wr = WechatReading(input.apps["wechat_reading"])
|
||
wechat = Wechat(input.apps["wechat"], init=input.apps_init["wechat"])
|
||
date_value, minutes = wr.best_reading_day_and_duration(input.os)
|
||
labels = WechatReading.date_labels(date_value, input.os)
|
||
return [
|
||
wechat.check_new_sent_any_of(
|
||
self.p.contact,
|
||
labels,
|
||
str(minutes),
|
||
field="reading_stats",
|
||
)
|
||
]
|
||
|
||
|
||
class RedbookAuthorFollowersToWechat(BaseTask):
|
||
templates = [
|
||
"在小红书搜'{keyword}'，关注第一篇笔记的作者，并将作者名字和粉丝数微信发给{contact}",
|
||
]
|
||
apps = ["redbook", "wechat"]
|
||
scope = "S2"
|
||
objective = "hybrid"
|
||
composition = "transfer"
|
||
difficulty = "L3"
|
||
capabilities = ["search", "extract", "social", "handoff"]
|
||
parameters = {"keyword": REDBOOK_KEYWORD_PARAM, "contact": WECHAT_CONTACT_PARAM}
|
||
expected_changes = [
|
||
"redbook.searchHistory",
|
||
"redbook.user.followingIds",
|
||
"redbook.history",
|
||
] + WECHAT_SEND_CHANGES
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
rb = Redbook(input.apps["redbook"], init=input.apps_init["redbook"])
|
||
wechat = Wechat(input.apps["wechat"], init=input.apps_init["wechat"])
|
||
author = rb.note_author(rb.first_search_note(self.p.keyword))
|
||
author_name = str(author["name"])
|
||
followers = str(author["followers"])
|
||
return [
|
||
rb.check_following(str(author["id"]), field="redbook_following"),
|
||
wechat.check_new_sent_norm_contains(
|
||
self.p.contact,
|
||
author_name,
|
||
followers,
|
||
field="redbook_author_followers",
|
||
),
|
||
]
|
||
|
||
|
||
class XLatestPostToReddit_WithTitleFormat(BaseTask):
|
||
templates = [
|
||
'把 X 用户 {user} 最新一条推文的文字内容，以"{user}:"开头发到 Reddit 的 r/{subreddit}。',
|
||
"Post the text content of X user {user}'s latest tweet to Reddit's r/{subreddit}, starting with \"{user}:\".",
|
||
]
|
||
apps = ["x", "reddit"]
|
||
scope = "S2"
|
||
objective = "operate"
|
||
composition = "transfer"
|
||
difficulty = "L4"
|
||
capabilities = ["extract", "create", "social", "handoff"]
|
||
parameters = {
|
||
"user": {"type": "string", "default": "elonmusk"},
|
||
"subreddit": {
|
||
"type": "enum",
|
||
"values": {"China_irl": "China_irl", "Games": "Games", "Music": "Music", "OtherSide": "OtherSide"},
|
||
"default": "China_irl",
|
||
},
|
||
}
|
||
expected_changes = [
|
||
"reddit.comments", "reddit.createDraft",
|
||
"reddit.posts", "reddit.user.postIds", "reddit.user.postVotes", "reddit.user.commentIds",
|
||
]
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
x_app = X(input.apps_init["x"])
|
||
reddit = Reddit(input.apps["reddit"], init=input.apps_init["reddit"])
|
||
|
||
# 1) 从 X 找到该用户最新推文内容
|
||
user_lower = self.p.user.lower().lstrip("@")
|
||
if not user_lower:
|
||
raise RuntimeError("任务设计错误：X user 不能为空")
|
||
|
||
tweet_content = ""
|
||
for post in x_app.view_posts():
|
||
aid = str(post.get("authorId") or "").lower()
|
||
# authorId 格式为 "u_elonmusk"，去掉 "u_" 前缀比较
|
||
if aid.removeprefix("u_") == user_lower or user_lower in aid:
|
||
tweet_content = str(post.get("content") or "").strip()
|
||
break
|
||
if not tweet_content:
|
||
raise RuntimeError(f"任务设计错误：未找到 X 用户 {self.p.user} 的非空推文")
|
||
|
||
# 2) 从 Reddit 找用户新发布的内容，验证包含 "{user}:" 前缀和推文内容
|
||
prefix = self.p.user.strip() + ":"
|
||
return [
|
||
reddit.check_new_content_contains(
|
||
prefix,
|
||
tweet_content,
|
||
subreddit=str(self.p.subreddit).strip(),
|
||
field="reddit_post",
|
||
normalize_match=True,
|
||
)
|
||
]
|
||
|
||
|
||
class RedbookFollowingNoteCountToSms(BaseTask):
|
||
templates = [
|
||
"查小红书我关注的'{username}'发了多少篇笔记，发短信告诉{contact}",
|
||
"Check how many notes '{username}' (someone I follow on RedNote) has posted, and send the count to {contact} via SMS",
|
||
]
|
||
apps = ["redbook", "sms"]
|
||
scope = "S2"
|
||
objective = "hybrid"
|
||
composition = "transfer"
|
||
difficulty = "L4"
|
||
max_steps = 45
|
||
capabilities = ["extract", "handoff"]
|
||
parameters = {
|
||
"username": REDBOOK_FOLLOWING_USER_PARAM,
|
||
"contact": {
|
||
"type": "enum",
|
||
"values": {"张三": "张三", "李四": "李四", "王五": "王五"},
|
||
"default": "张三",
|
||
},
|
||
}
|
||
expected_changes = SMS_SEND_CHANGES
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
rb = Redbook(input.apps["redbook"], init=input.apps_init["redbook"])
|
||
sms = sms_from_input(input)
|
||
count = rb.followed_user_note_count(self.p.username)
|
||
return [
|
||
sms.check_new_sent_contains_number(
|
||
self.p.contact,
|
||
count,
|
||
field="redbook_note_count",
|
||
)
|
||
]
|
||
class SpotifySongFullDetailsToRedbook(BaseTask):
|
||
templates = ["在Spotify搜'{song}'查下是谁唱的、几分钟，在小红书发一篇听歌笔记把这些写进去"]
|
||
apps = ["spotify", "redbook"]
|
||
scope = "S2"
|
||
objective = "operate"
|
||
composition = "transfer"
|
||
difficulty = "L4"
|
||
capabilities = ["search", "extract", "create", "handoff"]
|
||
parameters = {"song": {"type": "enum", "values": {"搁浅": "搁浅", "修炼爱情": "修炼爱情"}, "default": "搁浅"}}
|
||
expected_changes = REDBOOK_PUBLISH_CHANGES + ["apps.spotify"]
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
spotify = Spotify(input.apps["spotify"])
|
||
rb = Redbook(input.apps["redbook"], init=input.apps_init["redbook"])
|
||
track = spotify.track_by_title(self.p.song)
|
||
artist = str(track["artist"])
|
||
artist_simp = to_simplified(artist)
|
||
duration = str(track["duration"])
|
||
return [
|
||
rb.check_note_published(
|
||
content_pred=lambda content: (
|
||
artist_simp in to_simplified(str(content))
|
||
and match_track_duration(duration, str(content))
|
||
),
|
||
field="song_full_details",
|
||
)
|
||
]
|
||
class BilibiliTripleLikeThenMoments(BaseTask):
|
||
templates = [
|
||
"在B站{partition}排行榜找到第{rank}名给它一键三连，然后发个纯文字朋友圈推荐这个视频",
|
||
"Find the #{rank} video on Bilibili's {partition} ranking, give it a triple-like (like + coin + favorite), then post a Moments with pure texts to recommend it",
|
||
]
|
||
apps = ["bilibili", "wechat"]
|
||
scope = "S2"
|
||
objective = "operate"
|
||
composition = "sequential"
|
||
difficulty = "L3"
|
||
capabilities = ["social", "create", "handoff"]
|
||
parameters = {"partition": BILIBILI_PARTITION_PARAM, "rank": {"type": "int", "default": 1}}
|
||
expected_changes = WECHAT_MOMENT_CHANGES + ["apps.bilibili.activeVideoId", "apps.bilibili.user"]
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
bili = Bilibili(input.apps["bilibili"])
|
||
wechat = Wechat(input.apps["wechat"], init=input.apps_init["wechat"])
|
||
entry = Bilibili.ranking_entry(self.p.partition, int(self.p.rank))
|
||
title = str(entry["title"])
|
||
bvid = str(entry["id"])
|
||
return [
|
||
bili.check_liked_bvid(bvid, video_title=title, field="liked"),
|
||
bili.check_coined_bvid(bvid, video_title=title, field="coined"),
|
||
bili.check_favored_bvid(bvid, video_title=title, field="favored"),
|
||
wechat.check_new_moment_contains(title, field="moment_share"),
|
||
wechat.check_new_moment_no_images(),
|
||
]
|
||
class RedbookDmThenWechatReport(BaseTask):
|
||
templates = ["给小红书上我关注的'{username}'发私信'{message}'，然后在微信告诉{contact}已经联系他了"]
|
||
apps = ["redbook", "wechat"]
|
||
scope = "S2"
|
||
objective = "operate"
|
||
composition = "sequential"
|
||
difficulty = "L3"
|
||
capabilities = ["social", "handoff"]
|
||
parameters = {
|
||
"username": REDBOOK_FOLLOWING_USER_PARAM,
|
||
"message": {"type": "string", "default": "你好呀"},
|
||
"contact": WECHAT_CONTACT_PARAM,
|
||
}
|
||
expected_changes = WECHAT_SEND_CHANGES + ["redbook.chats"]
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
rb = Redbook(input.apps["redbook"], init=input.apps_init["redbook"])
|
||
wechat = Wechat(input.apps["wechat"], init=input.apps_init["wechat"])
|
||
return [
|
||
rb.check_chat_sent_to(self.p.username, self.p.message, field="redbook_dm"),
|
||
wechat.check_new_sent_any_of(
|
||
self.p.contact,
|
||
["已经联系", "已联系", "联系过", "已经私信", "已私信"],
|
||
self.p.username,
|
||
field="wechat_report",
|
||
),
|
||
]
|
||
class NotesContentToRedbookAndX(BaseTask):
|
||
templates = [
|
||
"在笔记里写一段关于{topic}的想法，然后分别在小红书和X上发布出来",
|
||
"Write some thoughts about {topic} in Notes, then post them on both RedNote and X",
|
||
]
|
||
apps = ["notes", "redbook", "x"]
|
||
scope = "S3"
|
||
objective = "operate"
|
||
composition = "transfer"
|
||
difficulty = "L2"
|
||
max_steps = 45
|
||
capabilities = ["create", "handoff"]
|
||
parameters = {"topic": {"type": "string", "default": "AI代理"}}
|
||
expected_changes = NOTES_CREATE_CHANGES + REDBOOK_PUBLISH_CHANGES + X_POST_CHANGES + ["os.clipboard"]
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
notes = Notes(input.apps["notes"], init=input.apps_init["notes"])
|
||
rb = Redbook(input.apps["redbook"], init=input.apps_init["redbook"])
|
||
x_app = X(input.apps["x"], init=input.apps_init["x"])
|
||
latest = notes.latest_note
|
||
if latest is None:
|
||
raise ValueError("任务设计错误：notes.latest_note 为空。")
|
||
content = str(latest.get("content") or "")
|
||
return [
|
||
notes.check_latest_contains(self.p.topic, field="notes_content"),
|
||
rb.check_note_published(content_pred=lambda text: content and content in str(text), field="redbook_sync"),
|
||
x_app.check_new_post_contains(content, field="x_sync"),
|
||
]
|
||
|
||
|
||
class DailyLogToMoments(BaseTask):
|
||
templates = [
|
||
"把我笔记里最新两条笔记简单汇总一下，发一条朋友圈。",
|
||
"Summarize the two most recent notes in my Notes app and post a WeChat Moments update about them.",
|
||
]
|
||
apps = ["notes", "wechat"]
|
||
scope = "S2"
|
||
objective = "operate"
|
||
composition = "transfer"
|
||
difficulty = "L4"
|
||
capabilities = ["extract", "reasoning", "social", "handoff"]
|
||
parameters = {}
|
||
expected_changes = WECHAT_MOMENT_CHANGES
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
notes = Notes(input.apps_init["notes"])
|
||
wechat = Wechat(input.apps["wechat"], init=input.apps_init["wechat"])
|
||
latest_notes = notes.latest_n_notes(2)
|
||
if len(latest_notes) < 2:
|
||
raise ValueError("Insufficient notes: need at least 2 notes for DailyLogToMoments task")
|
||
title1 = str(latest_notes[0].get("title") or latest_notes[0].get("content") or "").strip()
|
||
title2 = str(latest_notes[1].get("title") or latest_notes[1].get("content") or "").strip()
|
||
return [wechat.check_new_moment_contains(title1, title2, field="daily_log")]
|
||
|
||
|
||
class CulturalChecklistToRedbook(BaseTask):
|
||
templates = ["看看Spotify我今天最早听的那首歌是什么，再看看微信读书热搜第一本书叫什么，在笔记里记一份'今日文化清单'，然后在小红书发一篇笔记分享"]
|
||
apps = ["spotify", "wechat_reading", "notes", "redbook"]
|
||
scope = "S3"
|
||
objective = "operate"
|
||
composition = "transfer"
|
||
difficulty = "L4"
|
||
capabilities = ["extract", "create", "handoff"]
|
||
parameters = {}
|
||
expected_changes = NOTES_CREATE_CHANGES + REDBOOK_PUBLISH_CHANGES + ["apps.spotify"]
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
spotify_init = Spotify(input.apps_init["spotify"])
|
||
wr_init = WechatReading(input.apps_init["wechat_reading"])
|
||
notes = Notes(input.apps["notes"], init=input.apps_init["notes"])
|
||
rb = Redbook(input.apps["redbook"], init=input.apps_init["redbook"])
|
||
# 这里的“今天最早听的那首歌”同样采用最近播放页可见顺序的近似判定。
|
||
song = str(spotify_init.nth_today_play(1)["title"])
|
||
book = wr_init.first_hot_search_title()
|
||
return [
|
||
notes.check_latest_contains(song, book, field="cultural_note"),
|
||
rb.check_note_published(
|
||
content_keywords=(song, book),
|
||
field="cultural_redbook",
|
||
),
|
||
]
|
||
|
||
|
||
class EbayCheapToRedbook(BaseTask):
|
||
templates = ["帮我在 eBay 看看{product}里最便宜的那款，然后发一篇小红书商品推荐笔记。"]
|
||
apps = ["ebay", "redbook"]
|
||
scope = "S2"
|
||
objective = "operate"
|
||
composition = "transfer"
|
||
difficulty = "L3"
|
||
capabilities = ["search", "create", "handoff"]
|
||
parameters = {"product": EBAY_SEARCH_QUERY_PARAM}
|
||
expected_changes = EBAY_SEARCH_CHANGES + REDBOOK_PUBLISH_CHANGES
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
ebay = Ebay(input.apps["ebay"])
|
||
rb = Redbook(input.apps["redbook"], init=input.apps_init["redbook"])
|
||
top = ebay.cheapest_product(query=self.p.product)
|
||
snapshot = ebay.find_latest_snapshot(query=self.p.product, sort_option="priceLow")
|
||
|
||
return [
|
||
{
|
||
"field": "ebay_search",
|
||
"expected": f"{self.p.product}/priceLow",
|
||
"actual": snapshot,
|
||
"passed": snapshot is not None,
|
||
},
|
||
rb.check_note_published(
|
||
text_keywords=(str(top.title),),
|
||
field="product_recommendation",
|
||
),
|
||
]
|
||
|
||
|
||
class SpotifySaveCurrentSongToNotes(BaseTask):
|
||
templates = [
|
||
"把 Spotify 正在播放的歌名和歌手记到笔记里",
|
||
"Write down the song name and artist of what's currently playing on Spotify into Notes",
|
||
]
|
||
apps = ["spotify", "notes"]
|
||
scope = "S2"
|
||
objective = "operate"
|
||
composition = "transfer"
|
||
difficulty = "L2"
|
||
capabilities = ["extract", "create", "handoff"]
|
||
expected_changes = NOTES_CREATE_CHANGES + ["apps.spotify"]
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
spotify = Spotify(input.apps_init["spotify"])
|
||
notes = Notes(input.apps["notes"], init=input.apps_init["notes"])
|
||
track = spotify.current_track
|
||
return [notes.check_latest_contains(track["title"], track["artist"])]
|
||
class WechatReadingShareBookList(BaseTask):
|
||
templates = [
|
||
"把微信读书书架前{n}本书的名字微信发给{contact}",
|
||
"Send the names of the first {n} books on my WeChat Reading bookshelf to {contact}",
|
||
]
|
||
apps = ["wechat_reading", "wechat"]
|
||
scope = "S2"
|
||
objective = "operate"
|
||
composition = "deep_dive"
|
||
difficulty = "L3"
|
||
capabilities = ["extract", "handoff", "reasoning"]
|
||
parameters = {
|
||
"n": {"type": "int", "default": 3},
|
||
"contact": WECHAT_CONTACT_PARAM,
|
||
}
|
||
expected_changes = WECHAT_SEND_CHANGES
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
reading = WechatReading(input.apps["wechat_reading"])
|
||
wechat = Wechat(input.apps["wechat"], init=input.apps_init["wechat"])
|
||
titles = [
|
||
str(reading.require_store_book(str(item["bookId"]))["title"])
|
||
for item in reading.shelf[: int(self.p.n)]
|
||
]
|
||
actual = wechat.joined_new_texts_to(self.p.contact)
|
||
return [
|
||
{
|
||
"field": "sent_book_list",
|
||
"expected": titles,
|
||
"actual": actual or "(none)",
|
||
"passed": bool(actual) and count_titles_in_text(actual, titles) == len(titles),
|
||
}
|
||
]
|
||
class ReadingPlanToNotes(BaseTask):
|
||
"""
|
||
新增笔记包含正在读的书即可
|
||
"""
|
||
templates = [
|
||
"看看微信读书里我正在读的书，然后在笔记里制定一个本周的阅读计划。",
|
||
"Check what books I'm currently reading on WeChat Reading, then create a weekly reading plan in Notes.",
|
||
]
|
||
apps = ["wechat_reading", "notes"]
|
||
scope = "S2"
|
||
objective = "operate"
|
||
composition = "transfer"
|
||
difficulty = "L4"
|
||
max_steps = 45
|
||
capabilities = ["extract", "create", "handoff"]
|
||
parameters = {}
|
||
expected_changes = NOTES_CREATE_CHANGES
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
wr = WechatReading(input.apps["wechat_reading"])
|
||
notes = Notes(input.apps["notes"], init=input.apps_init["notes"])
|
||
target_books = set(wr.reading_book_titles())
|
||
if not target_books:
|
||
raise ValueError("No books found in WechatReading")
|
||
note = notes.latest_note
|
||
if not note:
|
||
return [{"field": "plan_note", "expected": "Create note", "actual": "No notes", "passed": False}]
|
||
content = f'{note.get("title", "")} {note.get("content", "")}'.lower()
|
||
passed_plan = "计划" in content or "plan" in content
|
||
missing_books = [book for book in target_books if book.lower() not in content]
|
||
passed_books = len(missing_books) == 0
|
||
actual_info = content
|
||
if missing_books:
|
||
actual_info = f"Missing: {', '.join(missing_books)}. Content: {content[:50]}..."
|
||
return [
|
||
{
|
||
"field": "plan_note",
|
||
"expected": f"Note with '计划' and books: {', '.join(target_books)}",
|
||
"actual": actual_info,
|
||
"passed": passed_books and passed_plan,
|
||
}
|
||
]
|
||
class FileManagerSendFileToWechatContact(BaseTask):
|
||
"""
|
||
两个文件名副本分别是：
|
||
- /sdcard/Download/downloaded_image.jpg
|
||
- /sdcard/Pictures/downloaded_image_copy.jpg
|
||
"""
|
||
templates = [
|
||
"我有一张图片分别在两个目录下各有一张不同名的副本，帮我找出他们，并把这两个文件名发给微信联系人{contact}",
|
||
"I have two copies of the same image with different filenames in two different directories, and send both filenames to WeChat contact {contact}",
|
||
]
|
||
apps = ["file_manager", "wechat"]
|
||
scope = "S2"
|
||
objective = "operate"
|
||
composition = "transfer"
|
||
difficulty = "L4"
|
||
capabilities = ["extract", "social", "handoff"]
|
||
parameters = {"contact": WECHAT_CONTACT_PARAM}
|
||
expected_changes = WECHAT_SEND_CHANGES
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
wechat = Wechat(input.apps["wechat"], init=input.apps_init["wechat"])
|
||
# 使用完整文件名（含扩展名），避免 "downloaded_image" 是 "downloaded_image_copy.jpg" 子串的误判
|
||
return [
|
||
wechat.check_new_sent_contains(
|
||
self.p.contact,
|
||
"downloaded_image.jpg",
|
||
"downloaded_image_copy.jpg",
|
||
field="wechat_file_names",
|
||
),
|
||
]
|
||
|
||
|
||
class NotesToWechatAndRedbook(BaseTask):
|
||
templates = ['把包含"{text_keyword}"的内容记到笔记后，再用微信同步发给{contact}，并发布一条对应的小红书笔记。']
|
||
apps = ["notes", "wechat", "redbook"]
|
||
scope = "S3"
|
||
objective = "operate"
|
||
composition = "transfer"
|
||
difficulty = "L4"
|
||
capabilities = ["create", "social", "handoff"]
|
||
parameters = {
|
||
"text_keyword": {"type": "string", "default": "今天心情很好"},
|
||
"contact": {"type": "string", "default": "王芳", "source": "apps.wechat.contacts[name]"},
|
||
}
|
||
expected_changes = NOTES_CREATE_CHANGES + WECHAT_SEND_CHANGES + REDBOOK_PUBLISH_CHANGES
|
||
|
||
def check_goals(self, input: JudgeInput) -> list[dict[str, Any]]:
|
||
xn = Notes(input.apps["notes"], init=input.apps_init["notes"])
|
||
wechat = Wechat(input.apps["wechat"], init=input.apps_init["wechat"])
|
||
rb = Redbook(input.apps["redbook"], init=input.apps_init["redbook"])
|
||
has_note = any(
|
||
str(self.p.text_keyword) in (str(n.get("title") or "") + str(n.get("content") or ""))
|
||
for n in xn.notes
|
||
)
|
||
init_has_note = any(
|
||
str(self.p.text_keyword) in (str(n.get("title") or "") + str(n.get("content") or ""))
|
||
for n in xn.init.notes
|
||
)
|
||
message_check = wechat.check_new_sent_to(self.p.contact, str(self.p.text_keyword), field="wechat")
|
||
return [
|
||
{
|
||
"field": "notes",
|
||
"expected": self.p.text_keyword,
|
||
"actual": "已写入" if has_note else "未写入",
|
||
"passed": has_note and not init_has_note,
|
||
},
|
||
message_check,
|
||
rb.check_note_published(
|
||
text_keywords=(str(self.p.text_keyword),),
|
||
new_only=True,
|
||
allow_draft=True,
|
||
field="redbook",
|
||
),
|
||
]
|