From 74a83e3a6566e1766d9a509b6bebcd906ad3466c Mon Sep 17 00:00:00 2001 From: d60 Date: Sun, 19 May 2024 01:53:50 +0900 Subject: [PATCH] Geo update --- docs/twikit.rst | 8 +++ docs/twikit.twikit_async.rst | 9 +++ twikit/__init__.py | 3 +- twikit/client.py | 116 ++++++++++++++++++++++++++++++++ twikit/geo.py | 81 ++++++++++++++++++++++ twikit/twikit_async/__init__.py | 1 + twikit/twikit_async/client.py | 116 ++++++++++++++++++++++++++++++++ twikit/twikit_async/geo.py | 81 ++++++++++++++++++++++ twikit/utils.py | 3 + 9 files changed, 417 insertions(+), 1 deletion(-) create mode 100644 twikit/geo.py create mode 100644 twikit/twikit_async/geo.py diff --git a/docs/twikit.rst b/docs/twikit.rst index 0974a98a..b22eb7b7 100644 --- a/docs/twikit.rst +++ b/docs/twikit.rst @@ -122,6 +122,14 @@ Notification :show-inheritance: :member-order: bysource +Geo +------------------- + +.. automodule:: twikit.geo + :members: + :undoc-members: + :show-inheritance: + :member-order: bysource Utils ------------------- diff --git a/docs/twikit.twikit_async.rst b/docs/twikit.twikit_async.rst index 418e321f..e81962b3 100644 --- a/docs/twikit.twikit_async.rst +++ b/docs/twikit.twikit_async.rst @@ -122,6 +122,15 @@ Notification :show-inheritance: :member-order: bysource +Geo +------------------- + +.. automodule:: twikit.geo + :members: + :undoc-members: + :show-inheritance: + :member-order: bysource + Utils --------------------------------- diff --git a/twikit/__init__.py b/twikit/__init__.py index 29f11ec9..1a6a226c 100644 --- a/twikit/__init__.py +++ b/twikit/__init__.py @@ -7,13 +7,14 @@ A Python library for interacting with the Twitter API. """ -__version__ = '1.6.3' +__version__ = '1.6.4' from .bookmark import BookmarkFolder from .client import Client from .community import (Community, CommunityCreator, CommunityMember, CommunityRule) from .errors import * +from .geo import Place from .group import Group, GroupMessage from .list import List from .message import Message diff --git a/twikit/client.py b/twikit/client.py index 1b510923..41eab8f9 100644 --- a/twikit/client.py +++ b/twikit/client.py @@ -16,11 +16,13 @@ from .errors import ( CouldNotTweet, InvalidMedia, + TweetNotAvailable, TwitterException, UserNotFound, UserUnavailable, raise_exceptions_from_response ) +from .geo import Place, _places_from_response from .group import Group, GroupMessage from .http import HTTPClient from .list import List @@ -1307,6 +1309,117 @@ def get_user_by_id(self, user_id: str) -> User: raise UserUnavailable(user_data.get('message')) return User(self, user_data) + def reverse_geocode( + self, lat: float, long: float, accuracy: str | float | None = None, + granularity: str | None = None, max_results: int | None = None + ) -> list[Place]: + """ + Given a latitude and a longitude, searches for up to 20 places that + + Parameters + ---------- + lat : :class:`float` + The latitude to search around. + long : :class:`float` + The longitude to search around. + accuracy : :class:`str` | :class:`float` None, default=None + A hint on the "region" in which to search. + granularity : :class:`str` | None, default=None + This is the minimal granularity of place types to return and must + be one of: `neighborhood`, `city`, `admin` or `country`. + max_results : :class:`int` | None, default=None + A hint as to the number of results to return. + + Returns + ------- + list[:class:`.Place`] + """ + params = { + 'lat': lat, + 'long': long, + 'accuracy': accuracy, + 'granularity': granularity, + 'max_results': max_results + } + for k, v in tuple(params.items()): + if v is None: + params.pop(k) + + response = self.http.get( + Endpoint.REVERSE_GEOCODE, + params=params, + headers=self._base_headers + ).json() + return _places_from_response(self, response) + + def search_geo( + self, lat: float | None = None, long: float | None = None, + query: str | None = None, ip: str | None = None, + granularity: str | None = None, max_results: int | None = None + ) -> list[Place]: + """ + Search for places that can be attached to a Tweet via POST + statuses/update. + + Parameters + ---------- + lat : :class:`float` | None + The latitude to search around. + long : :class:`float` | None + The longitude to search around. + query : :class:`str` | None + Free-form text to match against while executing a geo-based query, + best suited for finding nearby locations by name. + Remember to URL encode the query. + ip : :class:`str` | None + An IP address. Used when attempting to + fix geolocation based off of the user's IP address. + granularity : :class:`str` | None + This is the minimal granularity of place types to return and must + be one of: `neighborhood`, `city`, `admin` or `country`. + max_results : :class:`int` | None + A hint as to the number of results to return. + + Returns + ------- + list[:class:`.Place`] + """ + params = { + 'lat': lat, + 'long': long, + 'query': query, + 'ip': ip, + 'granularity': granularity, + 'max_results': max_results + } + for k, v in tuple(params.items()): + if v is None: + params.pop(k) + + response = self.http.get( + Endpoint.SEARCH_GEO, + params=params, + headers=self._base_headers + ).json() + return _places_from_response(self, response) + + def get_place(self, id: str) -> Place: + """ + Parameters + ---------- + id : :class:`str` + The ID of the place. + + Returns + ------- + :class:`.Place` + """ + response = self.http.get( + Endpoint.PLACE_BY_ID.format(id), + headers=self._base_headers + ).json() + return Place(self, response) + def _get_tweet_detail(self, tweet_id: str, cursor: str | None): variables = { 'focalTweetId': tweet_id, @@ -1395,6 +1508,9 @@ def get_tweet_by_id( """ response = self._get_tweet_detail(tweet_id, cursor) + if 'errors' in response: + raise TweetNotAvailable(response['errors'][0]['message']) + entries = find_dict(response, 'entries')[0] reply_to = [] replies_list = [] diff --git a/twikit/geo.py b/twikit/geo.py new file mode 100644 index 00000000..ca26b43b --- /dev/null +++ b/twikit/geo.py @@ -0,0 +1,81 @@ +from __future__ import annotations + +import warnings +from typing import TYPE_CHECKING +from .errors import TwitterException + +if TYPE_CHECKING: + from .client import Client + + +class Place: + """ + Attributes + ---------- + id : :class:`str` + The ID of the place. + name : :class:`str` + The name of the place. + full_name : :class:`str` + The full name of the place. + country : :class:`str` + The country where the place is located. + country_code : :class:`str` + The ISO 3166-1 alpha-2 country code of the place. + url : :class:`str` + The URL providing more information about the place. + place_type : :class:`str` + The type of place. + attributes : :class:`dict` + bounding_box : :class:`dict` + The bounding box that defines the geographical area of the place. + centroid : list[:class:`float`] + The geographical center of the place, represented by latitude and + longitude. + contained_within : list[:class:`.Place`] + A list of places that contain this place. + """ + + def __init__(self, client: Client, data: dict) -> None: + self._client = client + + self.id: str = data['id'] + self.name: str = data['name'] + self.full_name: str = data['full_name'] + self.country: str = data['country'] + self.country_code: str = data['country_code'] + self.url: str = data['url'] + self.place_type: str = data['place_type'] + self.attributes: dict = data['attributes'] + self.bounding_box: dict = data['bounding_box'] + self.centroid: list[float] = data['centroid'] + + self.contained_within: list[Place] = [ + Place(client, place) for place in data.get('contained_within', []) + ] + + def update(self) -> None: + new = self._client.get_place(self.id) + self.__dict__.update(new.__dict__) + + def __repr__(self) -> str: + return f'' + + def __eq__(self, __value: object) -> bool: + return isinstance(__value, Place) and self.id == __value.id + + def __ne__(self, __value: object) -> bool: + return not self == __value + + +def _places_from_response(client: Client, response: dict) -> list[Place]: + if 'errors' in response: + e = response['errors'][0] + # No data available for the given coordinate. + if e['code'] == 6: + warnings.warn(e['message']) + else: + raise TwitterException(e['message']) + + places = response['result']['places'] if 'result' in response else [] + return [Place(client, place) for place in places] diff --git a/twikit/twikit_async/__init__.py b/twikit/twikit_async/__init__.py index dd0f0fa6..755c8aa6 100644 --- a/twikit/twikit_async/__init__.py +++ b/twikit/twikit_async/__init__.py @@ -10,6 +10,7 @@ from .client import Client from .community import (Community, CommunityCreator, CommunityMember, CommunityRule) +from .geo import Place from .group import Group, GroupMessage from .list import List from .message import Message diff --git a/twikit/twikit_async/client.py b/twikit/twikit_async/client.py index 670af472..ad851811 100644 --- a/twikit/twikit_async/client.py +++ b/twikit/twikit_async/client.py @@ -15,6 +15,7 @@ CouldNotTweet, InvalidMedia, TwitterException, + TweetNotAvailable, UserNotFound, UserUnavailable, raise_exceptions_from_response @@ -43,6 +44,7 @@ Community, CommunityMember ) +from .geo import Place, _places_from_response from .group import Group, GroupMessage from .http import HTTPClient from .list import List @@ -1320,6 +1322,117 @@ async def get_user_by_id(self, user_id: str) -> User: raise UserUnavailable(user_data.get('message')) return User(self, user_data) + async def reverse_geocode( + self, lat: float, long: float, accuracy: str | float | None = None, + granularity: str | None = None, max_results: int | None = None + ) -> list[Place]: + """ + Given a latitude and a longitude, searches for up to 20 places that + + Parameters + ---------- + lat : :class:`float` + The latitude to search around. + long : :class:`float` + The longitude to search around. + accuracy : :class:`str` | :class:`float` None, default=None + A hint on the "region" in which to search. + granularity : :class:`str` | None, default=None + This is the minimal granularity of place types to return and must + be one of: `neighborhood`, `city`, `admin` or `country`. + max_results : :class:`int` | None, default=None + A hint as to the number of results to return. + + Returns + ------- + list[:class:`.Place`] + """ + params = { + 'lat': lat, + 'long': long, + 'accuracy': accuracy, + 'granularity': granularity, + 'max_results': max_results + } + for k, v in tuple(params.items()): + if v is None: + params.pop(k) + + response = (await self.http.get( + Endpoint.REVERSE_GEOCODE, + params=params, + headers=self._base_headers + )).json() + return _places_from_response(self, response) + + async def search_geo( + self, lat: float | None = None, long: float | None = None, + query: str | None = None, ip: str | None = None, + granularity: str | None = None, max_results: int | None = None + ) -> list[Place]: + """ + Search for places that can be attached to a Tweet via POST + statuses/update. + + Parameters + ---------- + lat : :class:`float` | None + The latitude to search around. + long : :class:`float` | None + The longitude to search around. + query : :class:`str` | None + Free-form text to match against while executing a geo-based query, + best suited for finding nearby locations by name. + Remember to URL encode the query. + ip : :class:`str` | None + An IP address. Used when attempting to + fix geolocation based off of the user's IP address. + granularity : :class:`str` | None + This is the minimal granularity of place types to return and must + be one of: `neighborhood`, `city`, `admin` or `country`. + max_results : :class:`int` | None + A hint as to the number of results to return. + + Returns + ------- + list[:class:`.Place`] + """ + params = { + 'lat': lat, + 'long': long, + 'query': query, + 'ip': ip, + 'granularity': granularity, + 'max_results': max_results + } + for k, v in tuple(params.items()): + if v is None: + params.pop(k) + + response = (await self.http.get( + Endpoint.SEARCH_GEO, + params=params, + headers=self._base_headers + )).json() + return _places_from_response(self, response) + + async def get_place(self, id: str) -> Place: + """ + Parameters + ---------- + id : :class:`str` + The ID of the place. + + Returns + ------- + :class:`.Place` + """ + response = (await self.http.get( + Endpoint.PLACE_BY_ID.format(id), + headers=self._base_headers + )).json() + return Place(self, response) + async def _get_tweet_detail(self, tweet_id: str, cursor: str | None): variables = { 'focalTweetId': tweet_id, @@ -1412,6 +1525,9 @@ async def get_tweet_by_id( """ response = await self._get_tweet_detail(tweet_id, cursor) + if 'errors' in response: + raise TweetNotAvailable(response['errors'][0]['message']) + entries = find_dict(response, 'entries')[0] reply_to = [] replies_list = [] diff --git a/twikit/twikit_async/geo.py b/twikit/twikit_async/geo.py new file mode 100644 index 00000000..d1f92b02 --- /dev/null +++ b/twikit/twikit_async/geo.py @@ -0,0 +1,81 @@ +from __future__ import annotations + +import warnings +from typing import TYPE_CHECKING +from .errors import TwitterException + +if TYPE_CHECKING: + from .client import Client + + +class Place: + """ + Attributes + ---------- + id : :class:`str` + The ID of the place. + name : :class:`str` + The name of the place. + full_name : :class:`str` + The full name of the place. + country : :class:`str` + The country where the place is located. + country_code : :class:`str` + The ISO 3166-1 alpha-2 country code of the place. + url : :class:`str` + The URL providing more information about the place. + place_type : :class:`str` + The type of place. + attributes : :class:`dict` + bounding_box : :class:`dict` + The bounding box that defines the geographical area of the place. + centroid : list[:class:`float`] + The geographical center of the place, represented by latitude and + longitude. + contained_within : list[:class:`.Place`] + A list of places that contain this place. + """ + + def __init__(self, client: Client, data: dict) -> None: + self._client = client + + self.id: str = data['id'] + self.name: str = data['name'] + self.full_name: str = data['full_name'] + self.country: str = data['country'] + self.country_code: str = data['country_code'] + self.url: str = data['url'] + self.place_type: str = data['place_type'] + self.attributes: dict = data['attributes'] + self.bounding_box: dict = data['bounding_box'] + self.centroid: list[float] = data['centroid'] + + self.contained_within: list[Place] = [ + Place(client, place) for place in data.get('contained_within', []) + ] + + async def update(self) -> None: + new = self._client.get_place(self.id) + await self.__dict__.update(new.__dict__) + + def __repr__(self) -> str: + return f'' + + def __eq__(self, __value: object) -> bool: + return isinstance(__value, Place) and self.id == __value.id + + def __ne__(self, __value: object) -> bool: + return not self == __value + + +def _places_from_response(client: Client, response: dict) -> list[Place]: + if 'errors' in response: + e = response['errors'][0] + # No data available for the given coordinate. + if e['code'] == 6: + warnings.warn(e['message']) + else: + raise TwitterException(e['message']) + + places = response['result']['places'] if 'result' in response else [] + return [Place(client, place) for place in places] diff --git a/twikit/utils.py b/twikit/utils.py index c79b5d02..09f7a461 100644 --- a/twikit/utils.py +++ b/twikit/utils.py @@ -297,6 +297,9 @@ class Endpoint: AVAILABLE_LOCATIONS = 'https://api.twitter.com/1.1/trends/available.json' EVENTS = 'https://api.twitter.com/live_pipeline/events' UPDATE_SUBSCRIPTIONS = 'https://api.twitter.com/1.1/live_pipeline/update_subscriptions' + REVERSE_GEOCODE = 'https://api.twitter.com/1.1/geo/reverse_geocode.json' + SEARCH_GEO = 'https://api.twitter.com/1.1/geo/search.json' + PLACE_BY_ID = 'https://api.twitter.com/1.1/geo/id/{}.json' T = TypeVar('T')