140fc56f18
Some checks failed
Release / semantic-release (push) Successful in 22s
tests / Unit tests (Linux, Python 3.11) (push) Failing after 5m16s
Release / build-linux (push) Failing after 7m10s
Release / build-windows (push) Has been cancelled
Release / build-macos (arm64, macos-latest) (push) Has been cancelled
Release / build-macos (x64, macos-15-intel) (push) Has been cancelled
Release / release-main (push) Has been cancelled
Release / release-develop (push) Has been cancelled
agents/llm.md promises "play the Nth video from the top", but browseAndPlay only ever clicked the first result. Add an optional 1-based index argument (default 1, backward-compatible) threaded to the Node helper, which now clicks the Nth a#video-title and clamps to the number of results returned so asking beyond the list plays the last available video instead of failing. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
102 lines
4.0 KiB
Python
102 lines
4.0 KiB
Python
"""Play a YouTube video on the shared screen (browser/screen-share mode).
|
|
|
|
Only meaningful when ``STREAM_BROWSER`` is true: it drives the on-screen Chrome
|
|
(via the Node CDP helper) to search YouTube and play the first result, which is
|
|
visible on the Go-Live broadcast. In voice-only mode (false) there is nothing to
|
|
show, so the tool reports that and does nothing.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
from typing import Dict, Any, Optional
|
|
|
|
from ..base import Tool, ToolContext
|
|
from ..types import ToolExecutionResult
|
|
from ...debug import debug_log
|
|
from .realtime_search import _NODE_SCRIPT
|
|
|
|
|
|
class BrowseAndPlayTool(Tool):
|
|
"""Play a YouTube video on the shared screen."""
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return "browseAndPlay"
|
|
|
|
@property
|
|
def description(self) -> str:
|
|
return (
|
|
"Play a song / music video / clip on the shared screen by searching YouTube "
|
|
"and playing a result. Use when the user asks you to play or watch "
|
|
"something. Plays the first result by default; pass 'index' to play the "
|
|
"Nth result from the top of the search list (e.g. 'play the 3rd video' -> "
|
|
"index=3). Only available in screen-share mode."
|
|
)
|
|
|
|
@property
|
|
def inputSchema(self) -> Dict[str, Any]:
|
|
return {
|
|
"type": "object",
|
|
"properties": {
|
|
"query": {
|
|
"type": "string",
|
|
"description": "What to play, e.g. 'IU Good Day' or 'lofi hip hop'.",
|
|
},
|
|
"index": {
|
|
"type": "integer",
|
|
"description": (
|
|
"1-based position of the video to play in the search results, "
|
|
"counted from the top of the list. Defaults to 1 (first result). "
|
|
"Use for 'play the Nth video' / 'play the second one'."
|
|
),
|
|
"minimum": 1,
|
|
},
|
|
},
|
|
"required": ["query"],
|
|
}
|
|
|
|
def run(self, args: Optional[Dict[str, Any]], context: ToolContext) -> ToolExecutionResult:
|
|
cfg = context.cfg
|
|
if not getattr(cfg, "stream_browser", True):
|
|
return ToolExecutionResult(
|
|
success=False,
|
|
reply_text="화면 공유 모드(STREAM_BROWSER=true)에서만 영상을 재생할 수 있습니다.",
|
|
)
|
|
query = ""
|
|
index = 1
|
|
if args and isinstance(args, dict):
|
|
query = str(args.get("query", "")).strip()
|
|
try:
|
|
index = int(args.get("index", 1) or 1)
|
|
except (TypeError, ValueError):
|
|
index = 1
|
|
if index < 1:
|
|
index = 1
|
|
if not query:
|
|
return ToolExecutionResult(success=False, reply_text="재생할 내용을 알려주세요.")
|
|
if not _NODE_SCRIPT.exists():
|
|
return ToolExecutionResult(success=False, reply_text="브라우저 재생 도구를 찾을 수 없습니다.")
|
|
|
|
context.user_print(f"▶️ 화면에서 '{query}' 재생 중… (#{index})")
|
|
debug_log(f" ▶️ browseAndPlay '{query}' index={index}", "tools")
|
|
try:
|
|
proc = subprocess.run(
|
|
["node", str(_NODE_SCRIPT), query, "youtube", str(index)],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=40,
|
|
env={**os.environ, "CDP_PORT": os.environ.get("CDP_PORT", "9222")},
|
|
)
|
|
data = json.loads((proc.stdout or "").strip() or "{}")
|
|
except Exception as e:
|
|
return ToolExecutionResult(success=False, reply_text=f"재생에 실패했습니다: {e}")
|
|
if not data.get("ok"):
|
|
return ToolExecutionResult(
|
|
success=False, reply_text=f"재생에 실패했습니다: {data.get('error', 'unknown')}"
|
|
)
|
|
title = data.get("title") or query
|
|
return ToolExecutionResult(success=True, reply_text=f"화면에서 '{title}' 재생을 시작했습니다.")
|