Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
The table of contents is too big for display.
Diff view
Diff view
  •  
  •  
  •  
4 changes: 2 additions & 2 deletions twitter_openapi_python/pyproject.toml
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[project]
name = "twitter_openapi_python"
version = "0.1.1"
version = "0.1.2"
description = "Twitter OpenAPI"
authors = [{ name = "fa0311", email = "yuki@yuki0311.com" }]
readme = "README.md"
Expand All @@ -9,7 +9,7 @@ license-files = ["LICENSE", "LICENSE.AGPL"]
keywords = ["twitter", "api", "python", "scraper"]
requires-python = ">=3.10"
dependencies = [
"twitter_openapi_python_generated==0.1.1",
"twitter_openapi_python_generated==0.1.2",
"pydantic>=2.6",
"requests>=2.31.0,<3.0.0",
"urllib3>=2.1.0,<3.0.0",
Expand Down
37 changes: 37 additions & 0 deletions twitter_openapi_python/test/api/test_1_birdwatch_api.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,37 @@
import logging
import unittest

import twitter_openapi_python as api
from test.api import get_client, print_tweet


class TestBirdwatchApi(unittest.TestCase):
client: api.BirdwatchApiUtils

def setUp(self):
self.client = get_client().get_birdwatch_api()

def test_get_birdwatch_fetch_alias_self_select_status(self):
result = self.client.get_birdwatch_fetch_alias_self_select_status()
logging.info(f"can_select: {result.data.can_select} has_selected: {result.data.has_selected}")

def test_get_birdwatch_fetch_authenticated_user_profile(self):
result = self.client.get_birdwatch_fetch_authenticated_user_profile()
logging.info(result.data.alias)

def test_get_birdwatch_fetch_birdwatch_profile(self):
result = self.client.get_birdwatch_fetch_birdwatch_profile(alias="me")
logging.info(result.data.alias)

def test_get_birdwatch_fetch_global_timeline(self):
result = self.client.get_birdwatch_fetch_global_timeline()
for tweet in result.data.data:
print_tweet(tweet)

def test_get_birdwatch_fetch_public_data(self):
result = self.client.get_birdwatch_fetch_public_data()
logging.info(f"notes files: {len(result.data.notes.urls)}")


if __name__ == "__main__":
unittest.main()
44 changes: 44 additions & 0 deletions twitter_openapi_python/test/api/test_1_community_api.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
import logging
import unittest

import twitter_openapi_python as api
from test.api import get_client, print_tweet


class TestCommunityApi(unittest.TestCase):
client: api.CommunityApiUtils

def setUp(self):
self.client = get_client().get_community_api()

def test_get_communities_fetch_one_query(self):
result = self.client.get_communities_fetch_one_query(community_id="1738874426798870555")
logging.info(f"{result.data.name}: {result.data.member_count} members")

def test_get_community_query(self):
result = self.client.get_community_query(community_id="1493446837214187523")
logging.info(f"{result.data.name}: {result.data.member_count} members")

def test_get_communities_search_query(self):
result = self.client.get_communities_search_query(query="test")
for item in result.data.items_results:
logging.info(item.result.name)

def test_get_topic_carousel_query(self):
result = self.client.get_topic_carousel_query()
for topic in result.data:
logging.info(topic.topic_name)

def test_get_global_communities_post_search_timeline(self):
result = self.client.get_global_communities_post_search_timeline(raw_query="test")
for tweet in result.data.data:
print_tweet(tweet)

def test_get_global_communities_latest_post_search_timeline(self):
result = self.client.get_global_communities_latest_post_search_timeline(raw_query="test")
for tweet in result.data.data:
print_tweet(tweet)


if __name__ == "__main__":
unittest.main()
20 changes: 20 additions & 0 deletions twitter_openapi_python/test/api/test_1_default_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,26 @@ def test_get_tweet_result_by_rest_id(self):
assert result.data.tweet.legacy is not None
logging.info(result.data.tweet.legacy.full_text)

def test_get_authenticate_periscope(self):
result = self.client.get_authenticate_periscope()
logging.info(f"token length: {len(result.data)}")

def test_get_creator_studio_tab_bar_item_query(self):
result = self.client.get_creator_studio_tab_bar_item_query()
logging.info(f"tasks: {len(result.data.xpayments_audience_rewards_tasks)}")

def test_get_data_saver_mode(self):
result = self.client.get_data_saver_mode()
logging.info(result.data.viewer)

def test_get_grok_home(self):
result = self.client.get_grok_home()
logging.info(f"grok {result.data.grok_version}: {[m.name for m in result.data.grok_model_options]}")

def test_get_pinned_timelines(self):
result = self.client.get_pinned_timelines()
logging.info(f"pinned: {len(result.data.pinned_timelines)}")


if __name__ == "__main__":
unittest.main()
10 changes: 10 additions & 0 deletions twitter_openapi_python/test/api/test_1_tweet_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -131,6 +131,16 @@ def test_get_connect_tab_timeline(self):
for user in result.data.data:
print_user(user)

def test_get_bookmark_search_timeline(self):
result = self.client.get_bookmark_search_timeline(raw_query="test")
for tweet in list(filter(self.ad_fillter, result.data.data)):
print_tweet(tweet)

def test_get_explore_sidebar(self):
result = self.client.get_explore_sidebar()
for trend in result.data.data:
logging.info(trend.name)


if __name__ == "__main__":
unittest.main()
15 changes: 15 additions & 0 deletions twitter_openapi_python/test/api/test_1_user_list_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,21 @@ def test_get_list_by_rest_id(self):
result = self.client.get_list_by_rest_id(list_id="1141162794290520064")
logging.info(f"{result.data.name}: {result.data.member_count} members")

def test_get_list_search_timeline(self):
result = self.client.get_list_search_timeline(raw_query="test")
for twitter_list in result.data.data:
logging.info(twitter_list.name)

def test_get_combined_lists(self):
result = self.client.get_combined_lists(user_id="44196397")
for twitter_list in result.data.data:
logging.info(twitter_list.name)

def test_get_lists_management_page_timeline(self):
result = self.client.get_lists_management_page_timeline()
for twitter_list in result.data.data:
logging.info(twitter_list.name)


if __name__ == "__main__":
unittest.main()
6 changes: 6 additions & 0 deletions twitter_openapi_python/test/api/test_1_users_api.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import logging
import unittest
from test.api import get_client, print_legacy_user

Expand All @@ -16,6 +17,11 @@ def test_get_users_by_rest_ids(self):
for user in result.data:
print_legacy_user(user.user.legacy)

def test_get_users_verified_avatars(self):
result = self.client.get_users_verified_avatars(user_ids=["44196397"])
for user in result.data:
logging.info(f"{user.rest_id}: blue={user.is_blue_verified} shape={user.profile_image_shape}")


if __name__ == "__main__":
unittest.main()
2 changes: 2 additions & 0 deletions twitter_openapi_python/twitter_openapi_python/api/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,8 @@

# flake8: noqa

from twitter_openapi_python.api.birdwatch_api import BirdwatchApiUtils
from twitter_openapi_python.api.community_api import CommunityApiUtils
from twitter_openapi_python.api.default_api import DefaultApiUtils
from twitter_openapi_python.api.initial_state_api import InitialStateApiUtils
from twitter_openapi_python.api.post_api import PostApiUtils
Expand Down
116 changes: 116 additions & 0 deletions twitter_openapi_python/twitter_openapi_python/api/birdwatch_api.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,116 @@
from typing import Any, Optional

import twitter_openapi_python_generated as twitter
import twitter_openapi_python_generated.models as models
from x_client_transaction import ClientTransaction

from twitter_openapi_python.models import (
TimelineApiUtilsResponse,
TweetApiUtilsData,
TwitterApiUtilsResponse,
)
from twitter_openapi_python.utils import (
data_request,
error_check,
timeline_request,
tweet_entries_converter,
)

ResponseType = TwitterApiUtilsResponse[TimelineApiUtilsResponse[TweetApiUtilsData]]
ParamType = dict[str, Any]


class BirdwatchApiUtils:
"""Community Notes."""

api: twitter.BirdwatchApi
flag: ParamType

def __init__(self, api: twitter.BirdwatchApi, flag: ParamType, ct: ClientTransaction):
self.api = api
self.flag = flag
self.ct = ct

def get_birdwatch_fetch_alias_self_select_status(
self,
extra_param: Optional[ParamType] = None,
) -> TwitterApiUtilsResponse[models.BirdwatchAliasSelfSelectStatus]:
param: ParamType = {}
if extra_param is not None:
param.update(extra_param)
return data_request(
self.api.get_birdwatch_fetch_alias_self_select_status_with_http_info,
self.flag["BirdwatchFetchAliasSelfSelectStatus"],
self.ct,
lambda e: error_check(e.data, e.errors).authenticated_user_birdwatch_alias_self_select_status,
param,
)

def get_birdwatch_fetch_authenticated_user_profile(
self,
extra_param: Optional[ParamType] = None,
) -> TwitterApiUtilsResponse[models.BirdwatchProfile]:
param: ParamType = {}
if extra_param is not None:
param.update(extra_param)
return data_request(
self.api.get_birdwatch_fetch_authenticated_user_profile_with_http_info,
self.flag["BirdwatchFetchAuthenticatedUserProfile"],
self.ct,
lambda e: error_check(e.data, e.errors).authenticated_user_birdwatch_profile,
param,
)

def get_birdwatch_fetch_birdwatch_profile(
self,
alias: str,
extra_param: Optional[ParamType] = None,
) -> TwitterApiUtilsResponse[models.BirdwatchProfile]:
param: ParamType = {"alias": alias}
if extra_param is not None:
param.update(extra_param)
return data_request(
self.api.get_birdwatch_fetch_birdwatch_profile_with_http_info,
self.flag["BirdwatchFetchBirdwatchProfile"],
self.ct,
lambda e: error_check(e.data, e.errors).birdwatch_profile_by_alias,
param,
)

def get_birdwatch_fetch_global_timeline(
self,
cursor: Optional[str] = None,
extra_param: Optional[ParamType] = None,
) -> ResponseType:
"""Posts with notes on the Community Notes home page (first tab)."""
param: ParamType = {}
if cursor is not None:
param["cursor"] = cursor
if extra_param is not None:
param.update(extra_param)
return timeline_request(
self.api.get_birdwatch_fetch_global_timeline_with_http_info,
self.flag["BirdwatchFetchGlobalTimeline"],
self.ct,
lambda e: error_check(
e.data.viewer.birdwatch_home_page.body.initial_timeline.timeline.timeline, e.errors
).instructions,
param,
tweet_entries_converter,
)

def get_birdwatch_fetch_public_data(
self,
extra_param: Optional[ParamType] = None,
) -> TwitterApiUtilsResponse[models.BirdwatchPublicDataFileBundle]:
"""Download urls of the latest public data files."""
param: ParamType = {}
if extra_param is not None:
param.update(extra_param)
return data_request(
self.api.get_birdwatch_fetch_public_data_with_http_info,
self.flag["BirdwatchFetchPublicData"],
self.ct,
lambda e: error_check(e.data, e.errors).birdwatch_latest_public_data_file_bundle,
param,
)
Loading