Merge pull request #1 from Huxingyu/feat/likes
feat: add likes command to fetch authenticated user's liked tweets
This commit is contained in:
commit
e675e879fb
5 files changed
+95
No files matched your search
@@ -19,6 +19,7 @@ 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
|
||||
- Likes: view your liked tweets
|
||||
- User lookup: fetch user profile and recent tweets
|
||||
- JSON output: export feed/bookmarks/user tweets for scripting
|
||||
- Optional scoring filter: rank tweets by engagement weights
|
||||
@@ -67,6 +68,10 @@ twitter feed --input tweets.json
|
||||
twitter favorite
|
||||
twitter favorite --max 30 --json
|
||||
|
||||
# Likes (your own liked tweets)
|
||||
twitter likes
|
||||
twitter likes --max 20 --json
|
||||
|
||||
# User
|
||||
twitter user elonmusk
|
||||
twitter user-posts elonmusk --max 20
|
||||
@@ -198,6 +203,7 @@ After installation, OpenClaw can call `twitter-cli` commands directly.
|
||||
|
||||
- 时间线读取:支持 `for-you` 和 `following`
|
||||
- 收藏读取:查看账号书签推文
|
||||
- 点赞列表:查看自己的点赞推文
|
||||
- 用户查询:查看用户资料和用户推文
|
||||
- JSON 输出:便于脚本处理
|
||||
- 可选筛选:按 engagement score 排序
|
||||
@@ -228,6 +234,9 @@ twitter feed --filter
|
||||
# 收藏
|
||||
twitter favorite
|
||||
|
||||
# 点赞列表
|
||||
twitter likes
|
||||
|
||||
# 用户
|
||||
twitter user elonmusk
|
||||
twitter user-posts elonmusk --max 20
|
||||
|
||||
@@ -34,6 +34,9 @@ twitter feed -t following
|
||||
# Bookmarks
|
||||
twitter favorite
|
||||
|
||||
# Likes (own liked tweets)
|
||||
twitter likes
|
||||
|
||||
# User profile and posts
|
||||
twitter user <screen_name>
|
||||
twitter user-posts <screen_name> --max 20
|
||||
@@ -56,6 +59,7 @@ Filtering is opt-in (disabled by default). Enable with `--filter`.
|
||||
```bash
|
||||
twitter feed --filter
|
||||
twitter favorite --filter
|
||||
twitter likes --filter
|
||||
```
|
||||
|
||||
The scoring formula:
|
||||
|
||||
@@ -26,3 +26,21 @@ def test_cli_feed_json_input_path(tmp_path, tweet_factory) -> None:
|
||||
result = runner.invoke(cli, ["feed", "--input", str(json_path), "--json"])
|
||||
assert result.exit_code == 0
|
||||
assert '"id": "1"' in result.output
|
||||
|
||||
|
||||
def test_cli_likes_command(monkeypatch, tweet_factory) -> None:
|
||||
fake_tweets = [tweet_factory("10"), tweet_factory("11")]
|
||||
|
||||
class FakeClient:
|
||||
def fetch_current_user_id(self) -> str:
|
||||
return "999"
|
||||
|
||||
def fetch_likes(self, user_id: str, count: int = 20):
|
||||
return fake_tweets
|
||||
|
||||
monkeypatch.setattr("twitter_cli.cli._get_client", lambda: FakeClient())
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli, ["likes", "--max", "5", "--json"])
|
||||
assert result.exit_code == 0
|
||||
assert '"id": "10"' in result.output
|
||||
assert '"id": "11"' in result.output
|
||||
@@ -193,6 +193,43 @@ def favorite(max_count, as_json, output_file, do_filter):
|
||||
console.print()
|
||||
|
||||
|
||||
@cli.command()
|
||||
@click.option("--max", "-n", "max_count", type=int, default=None, help="Max number of tweets to fetch.")
|
||||
@click.option("--json", "as_json", is_flag=True, help="Output as JSON.")
|
||||
@click.option("--output", "-o", "output_file", type=str, default=None, help="Save tweets to JSON file.")
|
||||
@click.option("--filter", "do_filter", is_flag=True, help="Enable score-based filtering.")
|
||||
def likes(max_count, as_json, output_file, do_filter):
|
||||
# type: (Optional[int], bool, Optional[str], bool) -> None
|
||||
"""Fetch your liked tweets (only available for the authenticated user)."""
|
||||
config = load_config()
|
||||
try:
|
||||
fetch_count = _resolve_fetch_count(max_count, config.get("fetch", {}).get("count", 50))
|
||||
client = _get_client()
|
||||
console.print("👤 Resolving your user ID...")
|
||||
user_id = client.fetch_current_user_id()
|
||||
console.print("❤️ Fetching likes (%d tweets)...\n" % fetch_count)
|
||||
start = time.time()
|
||||
tweets = client.fetch_likes(user_id, fetch_count)
|
||||
elapsed = time.time() - start
|
||||
console.print("✅ Fetched %d likes 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 output_file:
|
||||
Path(output_file).write_text(tweets_to_json(filtered), encoding="utf-8")
|
||||
console.print("💾 Saved to %s\n" % output_file)
|
||||
|
||||
if as_json:
|
||||
click.echo(tweets_to_json(filtered))
|
||||
return
|
||||
|
||||
print_tweet_table(filtered, console, title="❤️ Likes — %d tweets" % len(filtered))
|
||||
console.print()
|
||||
|
||||
|
||||
@cli.command()
|
||||
@click.argument("screen_name")
|
||||
def user(screen_name):
|
||||
|
||||
@@ -27,6 +27,8 @@ FALLBACK_QUERY_IDS = {
|
||||
"Bookmarks": "VFdMm9iVZxlU6hD86gfW_A",
|
||||
"UserByScreenName": "1VOOyvKkiI3FMmkeDNxM9A",
|
||||
"UserTweets": "E3opETHurmVJflFsUBVuUQ",
|
||||
"Likes": "yrFhBQCFo96rIBH28a4T2A",
|
||||
"Viewer": "zWQLM9HIVahRSUvzUH4lDw",
|
||||
}
|
||||
|
||||
TWITTER_OPENAPI_URL = (
|
||||
@@ -302,6 +304,31 @@ class TwitterClient:
|
||||
},
|
||||
)
|
||||
|
||||
def fetch_current_user_id(self):
|
||||
# type: () -> str
|
||||
"""Fetch the authenticated user's ID via the Viewer endpoint."""
|
||||
features = {
|
||||
"responsive_web_graphql_exclude_directive_enabled": True,
|
||||
"verified_phone_label_enabled": False,
|
||||
"responsive_web_graphql_skip_user_profile_image_extensions_enabled": False,
|
||||
"responsive_web_graphql_timeline_navigation_enabled": True,
|
||||
}
|
||||
data = self._graphql_get("Viewer", {}, features)
|
||||
user_id = _deep_get(data, "data", "viewer", "user_results", "result", "rest_id")
|
||||
if not user_id:
|
||||
raise RuntimeError("Could not resolve current user ID")
|
||||
return user_id
|
||||
|
||||
def fetch_likes(self, user_id, count=20):
|
||||
# type: (str, int) -> List[Tweet]
|
||||
"""Fetch liked tweets for a user (only works for the authenticated user)."""
|
||||
return self._fetch_timeline(
|
||||
"Likes",
|
||||
count,
|
||||
lambda data: _deep_get(data, "data", "user", "result", "timeline", "timeline", "instructions"),
|
||||
extra_variables={"userId": user_id},
|
||||
)
|
||||
|
||||
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."""
|
||||
|
||||
Reference in new issue
Block a user