krisolo/storyforge
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: deepen direct benchmark and analysis actions
Some checks failed
StoryForge CI / Baseline checks (push) Has been cancelled
Details
StoryForge CI / Backend tests (push) Has been cancelled
Details
StoryForge CI / Web tests (push) Has been cancelled
Details

Browse Source
This commit is contained in:
kris
2026-04-05 06:55:13 +08:00
parent 88ccc62c71
commit 905c3adabe
3 changed files with 527 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
CHANGELOG.md
View File

@@ -4,12 +4,23 @@
## 2026-04-05
### 主 Agent 抖音相似搜索与对标关系 live 修复
- 修复 `search-similar-accounts` / `save-benchmark-link` 在抖音 live 数据上错误按 `project_id` 查询账号导致的 500。
- `OneLiner` 现在会按抖音真实表结构解析目标账号，和国内平台 `content_sources` 路径分开处理。
- 新增抖音专用治理回归，锁住“查相似账号 -> 存对标关系”这条真实执行链。
### OneLiner 对话里的直接执行建议保留完整上下文
- OneLiner 助手消息里的 `suggested_actions` 现在不再只是渲染成一个裸 `data-action` 标签。
- 前端会把每条建议对应的 `executor_key / platform / payload / session_id` 一起带上，所以“直接分析账号 / 直接同步跟踪池 / 直接创建 AI 视频”这类建议从对话里点下去时，会真正走当前 live 执行器。
- 这让 OneLiner 对话、运行卡、结果卡三条链的“直接执行”行为终于统一，不会再出现运行卡能跑、对话建议却丢上下文的断层。
### 主页导入和高分分析的落点改成真正直达
- `直接导入主页` 现在不再把人扔回 `找对标` 总览，而是直接落到新建同步任务的详情页，方便立即看同步进度。
- `直接分析高分作品` 现在会直接回到当前对象，而不是回到整个 `找对标` 首页，让高分拆解结论和相似账号建议更容易接着看。
### 主 Agent 可直接执行分析账号、加入跟踪、创建 Agent
- `OneLiner / 主 Agent` 的动作执行器现在新增了三条真实动作：
@@ -428,3 +439,9 @@
- 依赖健康卡片在“未配置地址”时，管理员可以直接点 `去管理员配置台` 继续配置。
- 探测地址缺失文案改成“等待配置探测地址”，不再让人误以为系统异常。
### 主 Agent 可直接查相似与存对标
- `OneLiner / 主 Agent` 现在新增了 `直接查相似账号` 和 `直接存对标关系` 两条真实执行动作，不再只停留在“建议后跳回找对标”。
- `直接查相似账号` 会调用当前平台的相似搜索接口，返回真实候选数量，并在有候选账号时直接落到该账号详情。
- `直接存对标关系` 会优先复用最近一次相似搜索的候选，把它直接写入当前平台的对标关系，并把结果回写到找对标工作区。

290
collector-service/app/oneliner_features.py
View File

@@ -356,6 +356,26 @@ ACTION_REGISTRY_DEFAULTS: dict[str, dict[str, Any]] = {
"requires_platform": True,
"config": {},
},
"search-similar-accounts": {
"label": "直接查相似账号",
"description": "基于当前平台账号直接生成一批相似候选，并沉淀到当前项目。",
"category": "analysis",
"handler_key": "search-similar-accounts",
"status": "enabled",
"admin_only": False,
"requires_platform": True,
"config": {"max_candidates": 8},
},
"save-benchmark-link": {
"label": "直接存对标关系",
"description": "把当前相似候选直接加入对标关系，便于后续持续跟踪和拆解。",
"category": "analysis",
"handler_key": "save-benchmark-link",
"status": "enabled",
"admin_only": False,
"requires_platform": True,
"config": {"relation_type": "benchmark"},
},
"create-assistant": {
"label": "直接创建 Agent",
"description": "根据当前项目和平台上下文，直接创建可继续编辑的 Agent。",
@@ -458,6 +478,7 @@ ACTION_USAGE_KEYS: dict[str, str] = {
"analyze-account": "analysis",
"track-account": "content_source_sync",
"refresh-tracking": "content_source_sync",
"search-similar-accounts": "analysis",
"create-ai-video": "ai_video",
"create-real-cut": "real_cut",
"save-live-recorder-source": "live_recorder",
@@ -3235,11 +3256,11 @@ def register_oneliner_routes(app: Any, legacy: Any) -> None:
return legacy.db.fetch_one(
"""
SELECT * FROM douyin_accounts
WHERE user_id = ? AND project_id = ?
WHERE user_id = ?
ORDER BY updated_at DESC, created_at DESC
LIMIT 1
""",
(account["id"], project_id),
(account["id"],),
)
def _resolve_platform_target_account(
@@ -3254,8 +3275,8 @@ def register_oneliner_routes(app: Any, legacy: Any) -> None:
if normalized_platform == "douyin":
if normalized_requested:
return legacy.db.fetch_one(
"SELECT * FROM douyin_accounts WHERE id = ? AND user_id = ? AND project_id = ?",
(normalized_requested, account["id"], project_id),
"SELECT * FROM douyin_accounts WHERE id = ? AND user_id = ?",
(normalized_requested, account["id"]),
)
return _latest_douyin_account(account, project_id=project_id)
if normalized_requested:
@@ -3268,6 +3289,51 @@ def register_oneliner_routes(app: Any, legacy: Any) -> None:
)
return _latest_platform_account(account, project_id=project_id, platform=normalized_platform)
def _latest_similarity_candidate(
account: dict[str, Any],
*,
platform: str,
source_account_id: str,
) -> dict[str, Any] | None:
normalized_platform = _safe_platform(platform, fallback="douyin")
normalized_source_id = str(source_account_id or "").strip()
if not normalized_source_id:
return None
table_prefix = "douyin" if normalized_platform == "douyin" else normalized_platform
search_row = legacy.db.fetch_one(
f"""
SELECT * FROM {table_prefix}_similarity_searches
WHERE user_id = ? AND source_account_id = ?
ORDER BY created_at DESC
LIMIT 1
""",
(account["id"], normalized_source_id),
)
if not search_row:
return None
candidate_row = legacy.db.fetch_one(
f"""
SELECT * FROM {table_prefix}_similarity_candidates
WHERE search_id = ?
ORDER BY rank_index ASC
LIMIT 1
""",
(search_row["id"],),
)
if not candidate_row:
return {"search_id": search_row["id"], "candidate": {}}
candidate_payload = _parse_json(candidate_row.get("raw_output_json") or "{}", {})
candidate_payload.setdefault("candidate_account_id", candidate_row.get("candidate_account_id", ""))
candidate_payload.setdefault("candidate_profile_url", candidate_row.get("candidate_profile_url", ""))
candidate_payload.setdefault("candidate_nickname", candidate_row.get("candidate_nickname", ""))
candidate_payload.setdefault("rationale_text", candidate_row.get("rationale_text", ""))
candidate_payload.setdefault("agent_score", candidate_row.get("agent_score", 0))
candidate_payload.setdefault("heuristic_score", candidate_row.get("heuristic_score", 0))
return {
"search_id": search_row["id"],
"candidate": candidate_payload,
}
async def _call_local_api(
account: dict[str, Any],
*,
@@ -4549,6 +4615,18 @@ def register_oneliner_routes(app: Any, legacy: Any) -> None:
},
}
)
secondary_actions.append(
{
"key": "run-oneliner-action",
"label": "直接查相似账号",
"kind": "api_action",
"executor_key": "search-similar-accounts",
"platform": plan.get("platform", ""),
"payload": {
"target_account_id": latest_platform_account["id"],
},
}
)
if plan.get("platform") and latest_platform_account and plan.get("intent_key") in {"track_account", "custom"}:
secondary_actions.append(
{
@@ -4563,6 +4641,37 @@ def register_oneliner_routes(app: Any, legacy: Any) -> None:
},
}
)
latest_similarity = (
_latest_similarity_candidate(
account,
platform=plan.get("platform", ""),
source_account_id=latest_platform_account["id"],
)
if plan.get("platform") and latest_platform_account
else None
)
latest_similarity_candidate = (latest_similarity or {}).get("candidate") or {}
if (
plan.get("platform")
and latest_platform_account
and latest_similarity_candidate
and plan.get("intent_key") in {"analyze_top_videos", "analyze_account", "custom", "track_account"}
):
secondary_actions.append(
{
"key": "run-oneliner-action",
"label": "直接存对标关系",
"kind": "api_action",
"executor_key": "save-benchmark-link",
"platform": plan.get("platform", ""),
"payload": {
"source_account_id": latest_platform_account["id"],
"target_account_id": latest_similarity_candidate.get("candidate_account_id") or "",
"target_profile_url": latest_similarity_candidate.get("candidate_profile_url") or "",
"search_id": (latest_similarity or {}).get("search_id") or "",
},
}
)
if plan.get("platform") and plan.get("intent_key") in {"track_account", "custom"}:
secondary_actions.append(
{
@@ -5732,10 +5841,10 @@ def register_oneliner_routes(app: Any, legacy: Any) -> None:
"existing_source_id": (existing_source or {}).get("id", ""),
},
"recommended_action": _recommended_action(
"goto-discovery",
label="去找对标",
summary=f"继续查看 {legacy.platform_label(inferred_platform)} 主页导入后的账号分析和候选对标。",
screen="discovery",
"open-job-detail",
label="看任务详情",
summary=f"继续查看 {legacy.platform_label(inferred_platform)} 主页导入任务的同步进度和后续分析结果。",
screen="production",
platform=inferred_platform,
job_id=sync_job.get("id", ""),
),
@@ -5918,6 +6027,162 @@ def register_oneliner_routes(app: Any, legacy: Any) -> None:
),
}
async def _run_search_similar_accounts() -> dict[str, Any]:
if not normalized_platform:
raise HTTPException(status_code=400, detail="Platform is required for similarity search")
source_account = _resolve_platform_target_account(
account,
project_id=project["id"],
platform=normalized_platform,
requested_account_id=str(
requested_payload.get("target_account_id")
or requested_payload.get("targetAccountId")
or requested_payload.get("source_account_id")
or requested_payload.get("sourceAccountId")
or ""
),
)
if not source_account:
raise HTTPException(status_code=404, detail="No platform account available for similarity search")
search_payload = await _call_local_api(
account,
method="POST",
path=f"/v2/{normalized_platform}/similar-searches",
json_body={
"source_account_id": source_account["id"],
"candidate_urls": list(requested_payload.get("candidate_urls") or requested_payload.get("candidateUrls") or []),
"seed_linked_accounts": bool(requested_payload.get("seed_linked_accounts", requested_payload.get("seedLinkedAccounts", True))),
"search_public_pages": bool(requested_payload.get("search_public_pages", requested_payload.get("searchPublicPages", True))),
"model_profile_id": str(requested_payload.get("model_profile_id") or requested_payload.get("modelProfileId") or ""),
"max_candidates": max(1, min(int(requested_payload.get("max_candidates") or requested_payload.get("maxCandidates") or 8), 20)),
"extra_requirements": str(requested_payload.get("extra_requirements") or requested_payload.get("extraRequirements") or latest_user_message or ""),
},
)
search_id = str(search_payload.get("search_id") or search_payload.get("id") or "").strip()
detail = await _call_local_api(
account,
method="GET",
path=f"/v2/{normalized_platform}/similar-searches/{search_id}",
) if search_id else {"candidates": []}
candidates = list(detail.get("candidates") or [])
top_candidate = candidates[0] if candidates else {}
top_candidate_account_id = str(top_candidate.get("candidate_account_id") or "").strip()
recommended_action = (
_recommended_action(
"select-account",
label="打开当前候选",
summary=f"继续查看 {legacy.platform_label(normalized_platform)} 相似候选的详情、拆解和对标关系。",
screen="discovery",
account_id=top_candidate_account_id,
search_id=search_id,
)
if top_candidate_account_id
else _recommended_action(
"goto-discovery",
label="回到找对标",
summary=f"继续查看 {legacy.platform_label(normalized_platform)} 相似候选列表和高分样本。",
screen="discovery",
search_id=search_id,
)
)
return {
"title": "OneLiner 已查到相似账号",
"summary": f"已为 {legacy.platform_label(normalized_platform)} 当前账号生成 {len(candidates)} 个相似候选。",
"payload": {
"platform": normalized_platform,
"source_account_id": source_account["id"],
"search": {
"id": search_id,
"candidate_count": len(candidates),
"top_candidate_account_id": top_candidate_account_id,
},
"detail": detail,
},
"recommended_action": recommended_action,
}
async def _run_save_benchmark_link() -> dict[str, Any]:
if not normalized_platform:
raise HTTPException(status_code=400, detail="Platform is required for benchmark linking")
source_account = _resolve_platform_target_account(
account,
project_id=project["id"],
platform=normalized_platform,
requested_account_id=str(
requested_payload.get("source_account_id")
or requested_payload.get("sourceAccountId")
or requested_payload.get("target_account_id")
or requested_payload.get("targetAccountId")
or ""
),
)
if not source_account:
raise HTTPException(status_code=404, detail="No platform account available for benchmark linking")
latest_similarity = _latest_similarity_candidate(
account,
platform=normalized_platform,
source_account_id=source_account["id"],
) or {}
candidate_payload = latest_similarity.get("candidate") or {}
target_account_id = str(
requested_payload.get("target_account_id")
or requested_payload.get("targetAccountId")
or candidate_payload.get("candidate_account_id")
or ""
).strip()
target_profile_url = str(
requested_payload.get("target_profile_url")
or requested_payload.get("targetProfileUrl")
or ("" if target_account_id else candidate_payload.get("candidate_profile_url"))
or ""
).strip()
if not target_account_id and not target_profile_url:
raise HTTPException(status_code=404, detail="No benchmark candidate available to save")
created = await _call_local_api(
account,
method="POST",
path=f"/v2/{normalized_platform}/accounts/{source_account['id']}/benchmark-links",
json_body={
"target_account_ids": [target_account_id] if target_account_id else [],
"target_profile_urls": [] if target_account_id else [target_profile_url],
"relation_type": str(requested_payload.get("relation_type") or requested_payload.get("relationType") or "benchmark"),
"note": str(requested_payload.get("note") or candidate_payload.get("rationale_text") or "由 OneLiner 直接加入对标库。"),
"search_id": str(requested_payload.get("search_id") or requested_payload.get("searchId") or latest_similarity.get("search_id") or ""),
},
)
links = list(created.get("links") or [])
latest_link = links[0] if links else {}
recommended_action = (
_recommended_action(
"select-account",
label="打开当前对标",
summary=f"继续查看 {legacy.platform_label(normalized_platform)} 当前对标账号的详情、关系和高分样本。",
screen="discovery",
account_id=target_account_id,
)
if target_account_id
else _recommended_action(
"goto-discovery",
label="回到找对标",
summary=f"继续查看 {legacy.platform_label(normalized_platform)} 对标关系和相似候选。",
screen="discovery",
)
)
return {
"title": "OneLiner 已存对标关系",
"summary": f"已把当前候选加入 {legacy.platform_label(normalized_platform)} 对标关系。",
"payload": {
"platform": normalized_platform,
"source_account_id": source_account["id"],
"target_account_id": target_account_id,
"target_profile_url": target_profile_url,
"search_id": str(latest_similarity.get("search_id") or ""),
"link": latest_link,
"links": links,
},
"recommended_action": recommended_action,
}
async def _run_analyze_top_videos() -> dict[str, Any]:
if not normalized_platform:
raise HTTPException(status_code=400, detail="Platform is required for top video analysis")
@@ -6012,10 +6277,11 @@ def register_oneliner_routes(app: Any, legacy: Any) -> None:
"memory": memory,
},
"recommended_action": _recommended_action(
"goto-discovery",
label="去找对标",
summary=f"继续查看 {legacy.platform_label(normalized_platform)} 高分作品拆解和相似账号。",
"select-account",
label="打开当前对象",
summary=f"继续查看 {legacy.platform_label(normalized_platform)} 当前对象的高分作品拆解和相似账号。",
screen="discovery",
account_id=account_payload.get("id", ""),
platform=normalized_platform,
),
}
@@ -6224,6 +6490,8 @@ def register_oneliner_routes(app: Any, legacy: Any) -> None:
"track-account": _run_track_account,
"refresh-tracking": _run_refresh_tracking,
"mark-tracking-read": _run_mark_tracking_read,
"search-similar-accounts": _run_search_similar_accounts,
"save-benchmark-link": _run_save_benchmark_link,
"analyze-top-videos": _run_analyze_top_videos,
"create-assistant": _run_create_assistant,
"create-ai-video": _run_create_ai_video,

241
tests/test_main_agent_governance.py
View File

@@ -156,16 +156,7 @@ class MainAgentGovernanceTests(unittest.TestCase):
result_json: str = '{"summary":"done"}',
) -> str:
now = self.db_module.utc_now()
knowledge_base_id = "kb_member_default"
existing_kb = self.core.db.fetch_one("SELECT id FROM knowledge_bases WHERE id = ?", (knowledge_base_id,))
if not existing_kb:
self.core.db.execute(
"""
INSERT INTO knowledge_bases (id, user_id, project_id, name, description, sync_status, created_at, updated_at)
VALUES (?, ?, ?, 'Default KB', '', 'ready', ?, ?)
""",
(knowledge_base_id, self.ctx["member_id"], self.ctx["project_id"], now, now),
)
knowledge_base_id = self._ensure_default_knowledge_base(now)
self.core.db.execute(
"""
INSERT INTO jobs (
@@ -191,6 +182,42 @@ class MainAgentGovernanceTests(unittest.TestCase):
)
return job_id
def _ensure_default_knowledge_base(self, now: str | None = None) -> str:
knowledge_base_id = "kb_member_default"
existing_kb = self.core.db.fetch_one("SELECT id FROM knowledge_bases WHERE id = ?", (knowledge_base_id,))
if existing_kb:
return knowledge_base_id
ts = now or self.db_module.utc_now()
self.core.db.execute(
"""
INSERT INTO knowledge_bases (id, user_id, project_id, name, description, sync_status, created_at, updated_at)
VALUES (?, ?, ?, 'Default KB', '', 'ready', ?, ?)
""",
(knowledge_base_id, self.ctx["member_id"], self.ctx["project_id"], ts, ts),
)
return knowledge_base_id
def _insert_douyin_account(
self,
*,
account_id: str,
profile_url: str,
nickname: str,
) -> str:
now = self.db_module.utc_now()
self.core.db.execute(
"""
INSERT INTO douyin_accounts (
id, user_id, profile_url, canonical_profile_url, sec_uid, douyin_uid, douyin_id,
nickname, signature, avatar_url, tags_json, profile_stats_json, raw_profile_json,
source_mode, sync_status, last_public_sync_at, last_creator_sync_at, last_analysis_at,
created_at, updated_at
) VALUES (?, ?, ?, ?, '', '', '', ?, '', '', '[]', '{}', '{}', 'public', 'ready', NULL, NULL, NULL, ?, ?)
""",
(account_id, self.ctx["member_id"], profile_url, profile_url, nickname, now, now),
)
return account_id
def _insert_assistant(
self,
*,
@@ -802,6 +829,8 @@ class MainAgentGovernanceTests(unittest.TestCase):
self.assertIn("refresh-tracking", action_keys)
self.assertIn("mark-tracking-read", action_keys)
self.assertIn("create-assistant", action_keys)
self.assertIn("search-similar-accounts", action_keys)
self.assertIn("save-benchmark-link", action_keys)
save_registry = self.client.put(
"/v2/oneliner/action-registry/generate-copy",
@@ -974,6 +1003,91 @@ class MainAgentGovernanceTests(unittest.TestCase):
self.assertEqual(copy_payload["recommended_action"]["screen"], "playbook")
self.assertEqual(copy_payload["recommended_action"]["platform"], "douyin")
with patch.object(
self.core,
"create_content_source_sync_job",
new=AsyncMock(return_value={"id": "job_sync_import", "title": "Import Sync Job"}),
):
import_response = self.client.post(
"/v2/oneliner/actions/execute",
headers=self.ctx["member_headers"],
json={
"action_key": "import-homepage",
"project_id": self.ctx["project_id"],
"platform": "douyin",
"payload": {"source_url": "https://www.douyin.com/user/test-homepage"},
},
)
self.assertEqual(import_response.status_code, 200, import_response.text)
import_payload = import_response.json()
self.assertEqual(import_payload["recommended_action"]["action"], "open-job-detail")
self.assertEqual(import_payload["recommended_action"]["screen"], "production")
self.assertEqual(import_payload["recommended_action"]["job_id"], "job_sync_import")
source_id = self._insert_content_source_account(
platform="kuaishou",
title="高分拆解账号",
source_url="https://www.kuaishou.com/profile/top-video-account",
)
video_source = self.core.create_content_source(
account_id=self.ctx["member_id"],
project_id=self.ctx["project_id"],
source_kind="video_link",
platform="kuaishou",
source_url="https://www.kuaishou.com/video/top-video-1",
title="高分作品 1",
metadata={
"origin_content_source_id": source_id,
"source_account_url": "https://www.kuaishou.com/profile/top-video-account",
},
)
now = self.db_module.utc_now()
knowledge_base_id = self._ensure_default_knowledge_base(now)
self.core.db.execute(
"""
INSERT INTO jobs (
id, user_id, project_id, parent_job_id, assistant_id, knowledge_base_id, content_source_id,
source_type, line_type, workflow_key, orchestrator, provider_name, provider_task_id,
source_url, title, language, status, transcript_text, style_summary, upload_status,
error, artifacts_json, result_json, analysis_model_profile_id, created_at, updated_at
) VALUES (?, ?, ?, '', NULL, ?, ?, ?, ?, ?, 'n8n', 'collector', '', '', ?, 'auto', 'completed', '', '', 'completed', '', '{}', ?, '', ?, ?)
""",
(
"job_video_1",
self.ctx["member_id"],
self.ctx["project_id"],
knowledge_base_id,
video_source["id"],
"video_link",
"analysis",
"analysis_pipeline",
"高分作品 1",
'{"performance_score":91,"summary":"高分作品摘要"}',
now,
now,
),
)
with patch.object(self.core, "call_model", new=AsyncMock(return_value='{"summary":"保留开头 3 秒抓人结构"}')):
analyze_top_response = self.client.post(
"/v2/oneliner/actions/execute",
headers=self.ctx["member_headers"],
json={
"action_key": "analyze-top-videos",
"project_id": self.ctx["project_id"],
"platform": "kuaishou",
"payload": {
"target_account_id": source_id,
"top_video_count": 1,
"min_score": 0,
},
},
)
self.assertEqual(analyze_top_response.status_code, 200, analyze_top_response.text)
analyze_top_payload = analyze_top_response.json()
self.assertEqual(analyze_top_payload["recommended_action"]["action"], "select-account")
self.assertEqual(analyze_top_payload["recommended_action"]["screen"], "discovery")
self.assertEqual(analyze_top_payload["recommended_action"]["account_id"], source_id)
def test_create_ai_video_action_passes_provider_and_model_through_oneliner(self) -> None:
self._insert_completed_job(job_id="job_ai_video_source", title="AI Video Source Job")
self._insert_assistant()
@@ -1040,6 +1154,11 @@ class MainAgentGovernanceTests(unittest.TestCase):
title="快手测试账号",
source_url="https://www.kuaishou.com/profile/test-account",
)
candidate_id = self._insert_content_source_account(
platform="kuaishou",
title="快手候选账号",
source_url="https://www.kuaishou.com/profile/candidate-account",
)
captured_model_calls: list[dict[str, Any]] = []
async def fake_call_model(profile: dict[str, Any], *, system_prompt: str, user_prompt: str, temperature: float = 0.3, **_: Any) -> str:
@@ -1161,6 +1280,108 @@ class MainAgentGovernanceTests(unittest.TestCase):
self.assertTrue(create_payload["recommended_action"]["assistant_id"])
self.assertEqual(create_payload["payload"]["assistant"]["name"], "快手增长 Agent")
similar_response = self.client.post(
"/v2/oneliner/actions/execute",
headers=self.ctx["member_headers"],
json={
"action_key": "search-similar-accounts",
"project_id": self.ctx["project_id"],
"platform": "kuaishou",
"payload": {
"target_account_id": source_id,
"max_candidates": 3,
"extra_requirements": "优先找商业化和知识付费方向相近的账号",
},
},
)
self.assertEqual(similar_response.status_code, 200, similar_response.text)
similar_payload = similar_response.json()
self.assertEqual(similar_payload["recommended_action"]["action"], "select-account")
self.assertEqual(similar_payload["recommended_action"]["screen"], "discovery")
self.assertTrue(similar_payload["recommended_action"]["account_id"])
self.assertNotEqual(similar_payload["recommended_action"]["account_id"], source_id)
self.assertEqual(similar_payload["payload"]["platform"], "kuaishou")
self.assertEqual(similar_payload["payload"]["source_account_id"], source_id)
self.assertGreaterEqual(int(similar_payload["payload"]["search"]["candidate_count"] or 0), 1)
save_benchmark_response = self.client.post(
"/v2/oneliner/actions/execute",
headers=self.ctx["member_headers"],
json={
"action_key": "save-benchmark-link",
"project_id": self.ctx["project_id"],
"platform": "kuaishou",
"payload": {
"source_account_id": source_id,
"note": "由主 Agent 直接加入对标库",
},
},
)
self.assertEqual(save_benchmark_response.status_code, 200, save_benchmark_response.text)
save_benchmark_payload = save_benchmark_response.json()
self.assertEqual(save_benchmark_payload["recommended_action"]["action"], "select-account")
self.assertEqual(save_benchmark_payload["recommended_action"]["screen"], "discovery")
self.assertTrue(save_benchmark_payload["recommended_action"]["account_id"])
self.assertNotEqual(save_benchmark_payload["recommended_action"]["account_id"], source_id)
self.assertEqual(save_benchmark_payload["payload"]["platform"], "kuaishou")
self.assertEqual(save_benchmark_payload["payload"]["source_account_id"], source_id)
self.assertTrue(save_benchmark_payload["payload"]["link"]["id"])
def test_direct_oneliner_similarity_and_benchmark_actions_execute_real_douyin_flows(self) -> None:
source_id = self._insert_douyin_account(
account_id="dyacct_source",
profile_url="https://www.douyin.com/user/source-account",
nickname="源账号",
)
candidate_id = self._insert_douyin_account(
account_id="dyacct_candidate",
profile_url="https://www.douyin.com/user/candidate-account",
nickname="候选账号",
)
similar_response = self.client.post(
"/v2/oneliner/actions/execute",
headers=self.ctx["member_headers"],
json={
"action_key": "search-similar-accounts",
"project_id": self.ctx["project_id"],
"platform": "douyin",
"payload": {
"target_account_id": source_id,
"max_candidates": 3,
},
},
)
self.assertEqual(similar_response.status_code, 200, similar_response.text)
similar_payload = similar_response.json()
self.assertEqual(similar_payload["recommended_action"]["action"], "select-account")
self.assertEqual(similar_payload["recommended_action"]["screen"], "discovery")
self.assertEqual(similar_payload["payload"]["platform"], "douyin")
self.assertEqual(similar_payload["payload"]["source_account_id"], source_id)
self.assertEqual(similar_payload["payload"]["search"]["top_candidate_account_id"], candidate_id)
save_benchmark_response = self.client.post(
"/v2/oneliner/actions/execute",
headers=self.ctx["member_headers"],
json={
"action_key": "save-benchmark-link",
"project_id": self.ctx["project_id"],
"platform": "douyin",
"payload": {
"source_account_id": source_id,
"note": "由主 Agent 直接加入对标库",
},
},
)
self.assertEqual(save_benchmark_response.status_code, 200, save_benchmark_response.text)
save_benchmark_payload = save_benchmark_response.json()
self.assertEqual(save_benchmark_payload["recommended_action"]["action"], "select-account")
self.assertEqual(save_benchmark_payload["recommended_action"]["screen"], "discovery")
self.assertEqual(save_benchmark_payload["payload"]["platform"], "douyin")
self.assertEqual(save_benchmark_payload["payload"]["source_account_id"], source_id)
self.assertEqual(save_benchmark_payload["payload"]["target_account_id"], candidate_id)
self.assertTrue(save_benchmark_payload["payload"]["link"]["relation_id"])
def test_platform_agent_routes_are_live(self) -> None:
save_profile = self.client.put(
"/v2/platform-agents/douyin/profile",