From b0866ed8d75898db763f8a982ac7def3945ef4e3 Mon Sep 17 00:00:00 2001 From: jackwener Date: Sat, 7 Mar 2026 19:15:37 +0800 Subject: [PATCH] feat: add search and likes commands - Add 'twitter search' command with --type (Top/Latest/Photos/Videos), --max, --json, --filter - Add 'twitter likes' command to view tweets liked by a user - Add SearchTimeline and Likes GraphQL operations with fallback queryIds - Update README with new command examples (EN + CN) --- README.md | 17 ++++++++-- twitter_cli/cli.py | 76 +++++++++++++++++++++++++++++++++++++++++++ twitter_cli/client.py | 40 +++++++++++++++++++++++ 3 files changed, 131 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 008b6ec..d82e4a8 100644 --- a/README.md +++ b/README.md @@ -19,7 +19,8 @@ A terminal-first CLI for Twitter/X: read timelines, bookmarks, and user profiles - Timeline: fetch `for-you` and `following` feeds - Bookmarks: list saved tweets from your account -- User lookup: fetch user profile and recent tweets +- Search: find tweets by keyword with Top/Latest/Photos/Videos tabs +- User lookup: fetch user profile, tweets, and likes - JSON output: export feed/bookmarks/user tweets for scripting - Optional scoring filter: rank tweets by engagement weights - Cookie auth: use browser cookies or environment variables @@ -67,9 +68,15 @@ twitter feed --input tweets.json twitter favorite twitter favorite --max 30 --json +# Search +twitter search "Claude Code" +twitter search "AI agent" -t Latest --max 50 +twitter search "机器学习" --json + # User twitter user elonmusk twitter user-posts elonmusk --max 20 +twitter likes elonmusk --max 30 ``` ### Authentication @@ -198,7 +205,8 @@ After installation, OpenClaw can call `twitter-cli` commands directly. - 时间线读取：支持 `for-you` 和 `following` - 收藏读取：查看账号书签推文 -- 用户查询：查看用户资料和用户推文 +- 搜索：按关键词搜索推文，支持 Top/Latest/Photos/Videos +- 用户查询：查看用户资料、推文和点赞 - JSON 输出：便于脚本处理 - 可选筛选：按 engagement score 排序 - Cookie 认证：支持环境变量和浏览器自动提取 @@ -228,9 +236,14 @@ twitter feed --filter # 收藏 twitter favorite +# 搜索 +twitter search "Claude Code" +twitter search "AI agent" -t Latest --max 50 + # 用户 twitter user elonmusk twitter user-posts elonmusk --max 20 +twitter likes elonmusk --max 30 ``` ### 认证说明 diff --git a/twitter_cli/cli.py b/twitter_cli/cli.py index 1bb7149..55673ee 100644 --- a/twitter_cli/cli.py +++ b/twitter_cli/cli.py @@ -244,5 +244,81 @@ def user_posts(screen_name, max_count, as_json): console.print() +SEARCH_PRODUCTS = ["Top", "Latest", "Photos", "Videos"] + + +@cli.command() +@click.argument("query") +@click.option( + "--type", + "-t", + "product", + type=click.Choice(SEARCH_PRODUCTS, case_sensitive=False), + default="Top", + help="Search tab: Top, Latest, Photos, or Videos.", +) +@click.option("--max", "-n", "max_count", type=int, default=20, help="Max number of tweets to fetch.") +@click.option("--json", "as_json", is_flag=True, help="Output as JSON.") +@click.option("--filter", "do_filter", is_flag=True, help="Enable score-based filtering.") +def search(query, product, max_count, as_json, do_filter): + # type: (str, str, int, bool, bool) -> None + """Search tweets by QUERY string.""" + config = load_config() + try: + fetch_count = _resolve_fetch_count(max_count, 20) + client = _get_client(config) + console.print("🔍 Searching '%s' (%s, %d tweets)...\n" % (query, product, fetch_count)) + start = time.time() + tweets = client.fetch_search(query, fetch_count, product) + elapsed = time.time() - start + console.print("✅ Found %d tweets in %.1fs\n" % (len(tweets), elapsed)) + except RuntimeError as exc: + console.print("[red]❌ %s[/red]" % exc) + sys.exit(1) + + filtered = _apply_filter(tweets, do_filter, config) + + if as_json: + click.echo(tweets_to_json(filtered)) + return + + print_tweet_table(filtered, console, title="🔍 '%s' — %d tweets" % (query, len(filtered))) + console.print() + + +@cli.command() +@click.argument("screen_name") +@click.option("--max", "-n", "max_count", type=int, default=20, help="Max number of tweets to fetch.") +@click.option("--json", "as_json", is_flag=True, help="Output as JSON.") +@click.option("--filter", "do_filter", is_flag=True, help="Enable score-based filtering.") +def likes(screen_name, max_count, as_json, do_filter): + # type: (str, int, bool, bool) -> None + """Show tweets liked by a user. SCREEN_NAME is the @handle (without @).""" + screen_name = screen_name.lstrip("@") + config = load_config() + try: + fetch_count = _resolve_fetch_count(max_count, 20) + client = _get_client(config) + console.print("👤 Fetching @%s's profile..." % screen_name) + profile = client.fetch_user(screen_name) + console.print("❤️ Fetching likes (%d)...\n" % fetch_count) + start = time.time() + tweets = client.fetch_user_likes(profile.id, fetch_count) + elapsed = time.time() - start + console.print("✅ Fetched %d liked tweets in %.1fs\n" % (len(tweets), elapsed)) + except RuntimeError as exc: + console.print("[red]❌ %s[/red]" % exc) + sys.exit(1) + + filtered = _apply_filter(tweets, do_filter, config) + + if as_json: + click.echo(tweets_to_json(filtered)) + return + + print_tweet_table(filtered, console, title="❤️ @%s likes — %d tweets" % (screen_name, len(filtered))) + console.print() + + if __name__ == "__main__": cli() diff --git a/twitter_cli/client.py b/twitter_cli/client.py index cbc6231..b111104 100644 --- a/twitter_cli/client.py +++ b/twitter_cli/client.py @@ -28,6 +28,8 @@ FALLBACK_QUERY_IDS = { "Bookmarks": "VFdMm9iVZxlU6hD86gfW_A", "UserByScreenName": "1VOOyvKkiI3FMmkeDNxM9A", "UserTweets": "E3opETHurmVJflFsUBVuUQ", + "SearchTimeline": "VhUd6vHVmLBcw0uX-6jMLA", + "Likes": "lIDpu_NWL7_VhimGGt0o6A", } TWITTER_OPENAPI_URL = ( @@ -312,6 +314,44 @@ class TwitterClient: }, ) + def fetch_user_likes(self, user_id, count=20): + # type: (str, int) -> List[Tweet] + """Fetch tweets liked by a user.""" + return self._fetch_timeline( + "Likes", + count, + lambda data: _deep_get(data, "data", "user", "result", "timeline_v2", "timeline", "instructions"), + extra_variables={ + "userId": user_id, + "includePromotedContent": False, + "withClientEventToken": False, + "withBirdwatchNotes": False, + "withVoice": True, + }, + ) + + def fetch_search(self, query, count=20, product="Top"): + # type: (str, int, str) -> List[Tweet] + """Search tweets by query. + + Args: + query: Search query string. + count: Max number of tweets to return. + product: Search tab — "Top", "Latest", "People", "Photos", "Videos". + """ + return self._fetch_timeline( + "SearchTimeline", + count, + lambda data: _deep_get( + data, "data", "search_by_raw_query", "search_timeline", "timeline", "instructions", + ), + extra_variables={ + "rawQuery": query, + "querySource": "typed_query", + "product": product, + }, + ) + def _fetch_timeline(self, operation_name, count, get_instructions, extra_variables=None): # type: (str, int, Callable[[Any], Any], Optional[Dict[str, Any]]) -> List[Tweet] """Generic timeline fetcher with pagination and deduplication."""