twitter : new stream to extract tweets tagging us #35
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,24 +1,12 @@ | ||
| { | ||
| "credentials":{ | ||
| "client_id": "your_client_id", | ||
| "client_secret": "your_client_secret", | ||
| "access_token": "your_access_token", | ||
| "refresh_token": "your_refresh_token", | ||
| "token_expiry_date": "2024-12-31T00:00:00Z" | ||
| }, | ||
| "account_id": "your_account_id", | ||
| "start_time": "2024-01-01T00:00:00Z", | ||
| "tags": ["@IFT", "@status", "@Airbyte"] | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,77 @@ | ||
| { | ||
| "$schema": "http://json-schema.org/draft-07/schema#", | ||
| "type": "object", | ||
| "properties": { | ||
| "id": { | ||
| "type": ["null", "string"] | ||
| }, | ||
| "text": { | ||
| "type": ["null", "string"] | ||
| }, | ||
| "created_at": { | ||
| "type": ["null", "string"] | ||
| }, | ||
| "author_id": { | ||
| "type": ["null", "string"] | ||
| }, | ||
| "author_username": { | ||
| "type": ["null", "string"], | ||
| "description": "The Twitter handle/username of the tweet author (e.g., 'john_doe')" | ||
| }, | ||
| "author_name": { | ||
| "type": ["null", "string"], | ||
| "description": "The display name of the tweet author (e.g., 'John Doe')" | ||
| }, | ||
| "author_verified": { | ||
| "type": ["null", "boolean"], | ||
| "description": "Whether the tweet author is verified" | ||
| }, | ||
| "conversation_id": { | ||
| "type": ["null", "string"] | ||
| }, | ||
| "reply_settings": { | ||
| "type": ["null", "string"] | ||
| }, | ||
| "matched_tag": { | ||
| "type": ["null", "string"], | ||
| "description": "The tag that matched this tweet (e.g., '@IFT' or '@status')" | ||
| }, | ||
| "referenced_tweets": { | ||
| "type": ["null", "array"], | ||
| "items": { | ||
| "type": ["object"], | ||
| "properties":{ | ||
| "type": { | ||
| "type": ["null", "string"] | ||
| }, | ||
| "id": { | ||
| "type": ["null", "string"] | ||
| } | ||
| } | ||
| } | ||
| }, | ||
| "public_metrics": { | ||
| "type": ["null", "object"], | ||
| "properties": { | ||
| "retweet_count": { | ||
| "type": ["null", "number"] | ||
| }, | ||
| "reply_count": { | ||
| "type": ["null", "number"] | ||
| }, | ||
| "like_count": { | ||
| "type": ["null", "number"] | ||
| }, | ||
| "quote_count": { | ||
| "type": ["null", "number"] | ||
| }, | ||
| "impression_count": { | ||
| "type": ["null", "number"] | ||
| }, | ||
| "bookmark_count": { | ||
| "type": ["null", "number"] | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,94 @@ | ||
| from typing import Any, Iterable, Mapping, MutableMapping, Optional, List | ||
| import logging | ||
| import requests | ||
| import time | ||
| from datetime import datetime, timedelta | ||
| from airbyte_cdk.sources.streams import Stream | ||
| from airbyte_cdk.sources.streams.http import HttpStream | ||
|
|
||
| from .tweets_stream import TwitterStream | ||
|
|
||
| logger = logging.getLogger("airbyte") | ||
|
|
||
| class TagsStream(TwitterStream): | ||
| primary_key = "id" | ||
|
|
||
| def __init__(self, start_time: str = None, account_id: str = None, tags: List[str] = None, tags_frequent_extractions: bool = False, **kwargs): | ||
| super().__init__(start_time=start_time, account_id=account_id, **kwargs) | ||
|
|
||
| if not self.start_time: | ||
| if tags_frequent_extractions: | ||
| # Default to 1 hour 15 minutes before current time | ||
| self.start_time = datetime.utcnow() - timedelta(hours=1, minutes=15) | ||
| else: | ||
| # Default to 5 days before current time | ||
| self.start_time = datetime.utcnow() - timedelta(days=5) | ||
|
|
||
| self.tags = tags or [] | ||
|
|
||
| def stream_slices(self, **kwargs) -> Iterable[Optional[Mapping[str, Any]]]: | ||
| for tag in self.tags: | ||
| yield {"tag": tag} | ||
|
|
||
| def path( | ||
| self, | ||
| stream_state: Mapping[str, Any] = None, | ||
| stream_slice: Mapping[str, Any] = None, | ||
| next_page_token: Mapping[str, Any] = None | ||
| ) -> str: | ||
| return "tweets/search/recent" # this endpoint fetches data from the last 7 days | ||
|
|
||
| def next_page_token(self, response: requests.Response) -> Optional[Mapping[str, Any]]: | ||
| if 'meta' in response.json() and 'next_token' in response.json()['meta'] and response.json()['meta']['result_count'] > 0: | ||
| logger.debug('DBG-NT: %s', response.json()['meta']['next_token']) | ||
| return {"next_token": response.json()['meta']['next_token']} | ||
|
Comment on lines
+41
to
+44
There was a problem hiding this comment. This should probably be mutualized in TwitterStream There was a problem hiding this comment. In the end it can't be mutulalized here because :
|
||
|
|
||
| def request_params( | ||
| self, | ||
| next_page_token: Optional[Mapping[str, Any]] = None, | ||
| stream_state: Mapping[str, Any] = None, | ||
| stream_slice: Mapping[str, Any] = None | ||
| ) -> MutableMapping[str, Any]: | ||
| tag = stream_slice["tag"] | ||
| params = { | ||
| "query": tag, | ||
| "tweet.fields": "text,public_metrics,author_id,referenced_tweets,created_at", | ||
| "expansions": "author_id", | ||
| "user.fields": "username,name,verified,public_metrics", | ||
| "max_results": 100 | ||
| } | ||
| params.update({"start_time": self.start_time.strftime("%Y-%m-%dT%H:%M:%SZ")}) | ||
| if next_page_token: | ||
| params.update(**next_page_token) | ||
| return params | ||
|
|
||
| def parse_response( | ||
| self, | ||
| response: requests.Response, | ||
| stream_slice: Mapping[str, Any] = None, | ||
| **kwargs | ||
| ) -> Iterable[Mapping]: | ||
| logger.debug("Full response %s", response.json()) | ||
| response_data = response.json() | ||
|
|
||
| # Create a mapping of user_id to user info for quick lookup because user data is returned separately in the includes.users array, you need to manually join them using the author_id as the key | ||
| users_map = {} | ||
| if 'includes' in response_data and 'users' in response_data['includes']: | ||
| for user in response_data['includes']['users']: | ||
| users_map[user['id']] = user | ||
|
|
||
| if 'data' in response_data: | ||
| data = response_data['data'] | ||
| for t in data: | ||
| t["matched_tag"] = stream_slice["tag"] | ||
|
|
||
| if t.get('author_id') and t['author_id'] in users_map: | ||
| user_info = users_map[t['author_id']] | ||
| t["author_username"] = user_info.get('username') | ||
| t["author_name"] = user_info.get('name') | ||
| t["author_verified"] = user_info.get('verified') | ||
|
|
||
| yield t | ||
| self._apply_rate_limiting() | ||
|
|
||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Why did you add a whitespace at the end ?