diff --git a/.github/workflows/release.yaml b/.github/workflows/release.yaml index 73d6f9e..74c7f16 100644 --- a/.github/workflows/release.yaml +++ b/.github/workflows/release.yaml @@ -9,7 +9,7 @@ jobs: steps: - uses: actions/checkout@v2 - name: Build and publish to pypi - uses: JRubics/poetry-publish@v1.8 + uses: JRubics/poetry-publish@v1.17 with: pypi_token: ${{ secrets.PYPI_TOKEN }} diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml index f902b4f..51bf30b 100644 --- a/.github/workflows/test.yaml +++ b/.github/workflows/test.yaml @@ -11,9 +11,9 @@ jobs: runs-on: ubuntu-latest strategy: matrix: - python-version: ['3.7', '3.8', '3.9', '3.10', '3.11'] + python-version: ['3.7', '3.8', '3.9', '3.10', '3.11', '3.12'] include: - - python-version: 3.8 + - python-version: '3.10' update-coverage: true steps: @@ -22,25 +22,18 @@ jobs: uses: actions/setup-python@v4 with: python-version: ${{ matrix.python-version }} - - name: Install Poetry - uses: snok/install-poetry@v1 - with: - virtualenvs-create: true - virtualenvs-in-project: true - installer-parallel: true - - name: Load cached venv - id: cached-poetry-dependencies + - name: Cache pip uses: actions/cache@v2 with: - path: .venv + path: ~/.cache/pip key: ${{ matrix.python-version }}-poetry-${{ hashFiles('pyproject.toml') }} - name: Install dependencies - if: steps.cached-poetry-dependencies.outputs.cache-hit != 'true' - run: poetry install --no-interaction --no-root + run: | + python -m pip install --upgrade pip poetry + poetry install - name: Test with pytest run: | - source .venv/bin/activate - pytest + poetry run pytest - name: Upload coverage to Codecov if: ${{ matrix.update-coverage }} uses: codecov/codecov-action@v3 @@ -58,7 +51,7 @@ jobs: - uses: actions/checkout@v3 - uses: actions/setup-python@v4 with: - python-version: 3.8 + python-version: '3.10' - name: Cache pip uses: actions/cache@v2 with: @@ -67,4 +60,4 @@ jobs: - name: Install dependencies run: python -m pip install --upgrade pip black - name: Black test - run: black --check . + run: make lint-check diff --git a/README.rst b/README.rst index fa64ba2..15c7037 100644 --- a/README.rst +++ b/README.rst @@ -123,6 +123,10 @@ Now covers these features: - Direct Messages lookup - Manage Direct Messages +- Media Upload + - Media Simple upload + - Media Chunked upload + ----------- INSTANTIATE ----------- diff --git a/docs/docs/usage/media-upload/chunked-upload.md b/docs/docs/usage/media-upload/chunked-upload.md new file mode 100644 index 0000000..129f49d --- /dev/null +++ b/docs/docs/usage/media-upload/chunked-upload.md @@ -0,0 +1,69 @@ +Chunked upload is the recommended method for uploading media files. It allows you to upload media files up to 512MB. The chunked upload endpoint can be used to upload both images and videos. + +You can get more information for this at [docs](https://developer.twitter.com/en/docs/twitter-api/v1/media/upload-media/api-reference/post-media-upload-init). + +follow the steps below to upload a video by chunked upload: + +### Step 1: Initialize the upload + +```python +resp = my_api.upload_media_chunked_init( + total_bytes=1234561, + media_type="video/mp4", +) +media_id = resp.media_id_string +``` + +### Step 2: Append the file + +Assume the file has split into 3 parts + +```python +video_parts = [ + "path/to/video/part1.mp4", + "path/to/video/part2.mp4", + "path/to/video/part3.mp4", +] + +for idx, part in enumerate(video_parts): + with open(part, "rb") as media: + status = my_api.upload_media_chunked_append( + media_id=media_id, + media=media, + segment_index=idx, + ) + print(status) +``` + +### Step 3: Finalize the upload + +Once you have appended all the file parts, you need to finalize the upload. + +```python +resp = my_api.upload_media_chunked_finalize(media_id=media_id) +print(resp) +``` + +### Step 4 (Optional): Check the status + +If the finalize response show the video is processing, you can check the status by using the following code: + +```python +resp = my_api.upload_media_chunked_status(media_id=media_id) +print(resp) +``` + +Note, only the media is in processing status, you can get the status. + +### Step 5: Create a tweet with video + +Once the processing is complete, you can use the `media_id` to create a tweet with video. + +```python +my_api.create_tweet( + text="Hello World", + media_media_ids=[media_id], +) +``` + +Enjoy it! diff --git a/docs/docs/usage/media-upload/simple-upload.md b/docs/docs/usage/media-upload/simple-upload.md new file mode 100644 index 0000000..b58a2f2 --- /dev/null +++ b/docs/docs/usage/media-upload/simple-upload.md @@ -0,0 +1,23 @@ +The simple upload endpoint can only be used to upload images(gifs). + +You can get more information for this at [docs](https://developer.twitter.com/en/docs/twitter-api/v1/media/upload-media/api-reference/post-media-upload) + +follow the steps below to upload an image: + +```python + +with open("path/to/image", "rb") as media: + resp = my_api.upload_media_simple(media=media) + print(resp) +``` + +also you can upload with base64 encoded image: + +```python +import base64 + +with open("path/to/image", "rb") as f: + media_data = base64.b64encode(f.read()).decode("utf-8") + resp = my_api.upload_media_simple(media_data=media_data) + print(resp) +``` diff --git a/docs/mkdocs.yml b/docs/mkdocs.yml index a4371bc..20acace 100644 --- a/docs/mkdocs.yml +++ b/docs/mkdocs.yml @@ -34,6 +34,9 @@ nav: - Authorization OAuth2.0: authorization_oauth2.md - Usage: - Preparation: usage/preparation.md + - Media Upload: + - Simple Upload: usage/media-upload/simple-upload.md + - Chunked Upload: usage/media-upload/chunked-upload.md - Tweets: - Tweet Lookup: usage/tweets/tweet-lookup.md - Manage Tweets: usage/tweets/tweet-manage.md diff --git a/pytwitter/api.py b/pytwitter/api.py index 73b628c..35182e4 100644 --- a/pytwitter/api.py +++ b/pytwitter/api.py @@ -6,7 +6,7 @@ import os import re import time -from typing import List, Optional, Tuple, Union +from typing import List, Optional, Tuple, Union, IO import requests from requests.models import Response @@ -33,6 +33,7 @@ class Api: DEFAULT_CALLBACK_URI = "https://localhost/" BASE_OAUTH2_AUTHORIZE_URL = "https://twitter.com/i/oauth2/authorize" BASE_OAUTH2_ACCESS_TOKEN_URL = "https://api.twitter.com/2/oauth2/token" + BASE_UPLOAD_URL = "https://upload.twitter.com/1.1" DEFAULT_SCOPES = ["users.read", "tweet.read"] def __init__( @@ -132,16 +133,23 @@ def get_uid_from_access_token_key(access_token: str) -> str: return uid def _request( - self, url, verb="GET", params=None, data=None, json=None, enforce_auth=True + self, + url, + verb="GET", + params=None, + data=None, + json=None, + files=None, + enforce_auth=True, ) -> Response: """ - Request for twitter api url + Request for Twitter api url :param url: The api location for twitter :param verb: HTTP Method, like GET,POST,PUT. :param params: The url params to send in the body of the request. :param data: The form data to send in the body of the request. :param json: The json data to send in the body of the request. - :param enforce_auth: Whether need auth + :param enforce_auth: Whether api need auth :return: A json object """ auth = None @@ -167,6 +175,7 @@ def _request( data=data, auth=auth, json=json, + files=files, timeout=self.timeout, proxies=self.proxies, ) @@ -546,6 +555,209 @@ def get_tweet( return_json=return_json, ) + def upload_media_simple( + self, + media: Optional[IO] = None, + media_data: Optional[str] = None, + media_category: Optional[str] = None, + additional_owners: Optional[List[str]] = None, + return_json: bool = False, + ) -> Union[dict, md.MediaUploadResponse]: + """ + Simple Upload, Use this endpoint to upload images to Twitter. + + Note: The simple upload endpoint can only be used to upload images. + + :param media: The raw binary file content being uploaded. Cannot be used with `media_data`. + :param media_data: The base64-encoded file content being uploaded. Cannot be used with `media`. + :param media_category: The category that represents how the media will be used. + This field is required when using the media with the Ads API. + Possible values: + - tweet_image + - tweet_gif + - tweet_video + - amplify_video + :param additional_owners: A comma-separated list of user IDs to set as additional owners + allowed to use the returned media_id in Tweets or Cards. + Up to 100 additional owners may be specified. + :param return_json: Type for returned data. If you set True JSON data will be returned. + :return: Media upload response. + """ + + files, args = {}, {} + if media: + files["media"] = media + elif media_data: + args["media_data"] = media_data + else: + raise PyTwitterError("Need media or media_data") + if media_category: + args["media_category"] = media_category + if additional_owners: + args["additional_owners"] = enf_comma_separated( + name="additional_owners", value=additional_owners + ) + + resp = self._request( + url=f"{self.BASE_UPLOAD_URL}/media/upload.json", + verb="POST", + data=args, + files=files, + ) + data = self._parse_response(resp=resp) + if return_json: + return data + else: + return md.MediaUploadResponse.new_from_json_dict(data=data) + + def upload_media_chunked_init( + self, + total_bytes: int, + media_type: str, + media_category: Optional[str] = None, + additional_owners: Optional[List[str]] = None, + return_json: bool = False, + ) -> Union[dict, md.MediaUploadResponse]: + """ + Chunked Upload, Use this endpoint to upload videos and images to Twitter. + + Note: The chunked upload endpoint can be used to upload both images and videos. + Videos must be sent as chunked media containers, which means that you must send the + raw chunked media data and the media category separately. + + :param total_bytes: The total size of the media being uploaded in bytes. + :param media_type: The MIME type of the media being uploaded. example: image/jpeg, image/gif, and video/mp4. + :param media_category: The category that represents how the media will be used. + This field is required when using the media with the Ads API. + Possible values: + - tweet_image + - tweet_gif + - tweet_video + - amplify_video + :param additional_owners: A comma-separated list of user IDs to set as additional owners + allowed to use the returned media_id in Tweets or Cards. + Up to 100 additional owners may be specified. + :param return_json: Type for returned data. If you set True JSON data will be returned. + :return: Media upload response. + """ + + args = { + "command": "INIT", + "total_bytes": total_bytes, + "media_type": media_type, + } + if media_category: + args["media_category"] = media_category + if additional_owners: + args["additional_owners"] = enf_comma_separated( + name="additional_owners", value=additional_owners + ) + + resp = self._request( + url=f"{self.BASE_UPLOAD_URL}/media/upload.json", + verb="POST", + data=args, + ) + data = self._parse_response(resp=resp) + if return_json: + return data + else: + return md.MediaUploadResponse.new_from_json_dict(data=data) + + def upload_media_chunked_append( + self, + media_id: str, + segment_index: int, + media: Optional[IO] = None, + media_data: Optional[str] = None, + ) -> bool: + """ + Used to upload a chunk (consecutive byte range) of the media file. + + :param media_id: The `media_id` returned from the INIT step. + :param segment_index: An ordered index of file chunk. It must be between 0-999 inclusive. + The first segment has index 0, second segment has index 1, and so on. + :param media: The raw binary file content being uploaded. Cannot be used with `media_data`. + :param media_data: The base64-encoded file content being uploaded. Cannot be used with `media`. + :return: True if upload success. + """ + args = { + "command": "APPEND", + "media_id": media_id, + "segment_index": segment_index, + } + files = {} + if media: + files["media"] = media + elif media_data: + args["media_data"] = media_data + else: + raise PyTwitterError("Need media or media_data") + + resp = self._request( + url=f"{self.BASE_UPLOAD_URL}/media/upload.json", + verb="POST", + data=args, + files=files, + ) + if resp.ok: + return True + raise PyTwitterError(resp.json()) + + def upload_media_chunked_finalize( + self, + media_id: str, + return_json: bool = False, + ) -> Union[dict, md.MediaUploadResponse]: + """ + Finalize the chunk upload. + :param media_id: The `media_id` returned from the INIT step. + :param return_json: Type for returned data. If you set True JSON data will be returned. + :return: Media upload response. + """ + + resp = self._request( + url=f"{self.BASE_UPLOAD_URL}/media/upload.json", + verb="POST", + data={ + "command": "FINALIZE", + "media_id": media_id, + }, + ) + data = self._parse_response(resp=resp) + if return_json: + return data + else: + return md.MediaUploadResponse.new_from_json_dict(data=data) + + def upload_media_chunked_status( + self, + media_id: str, + return_json: bool = False, + ) -> Union[dict, md.MediaUploadResponse]: + """ + Check the status of the chunk upload. + + Note: Can only call after the FINALIZE step. If chunked upload not sync mode will return error. + + :param media_id: The `media_id` returned from the INIT step. + :param return_json: Type for returned data. If you set True JSON data will be returned. + :return: Media upload response. + """ + resp = self._request( + url=f"{self.BASE_UPLOAD_URL}/media/upload.json", + verb="GET", + params={ + "command": "STATUS", + "media_id": media_id, + }, + ) + data = self._parse_response(resp=resp) + if return_json: + return data + else: + return md.MediaUploadResponse.new_from_json_dict(data=data) + def create_tweet( self, *, diff --git a/pytwitter/models/__init__.py b/pytwitter/models/__init__.py index 4e37f1d..74bfcbd 100644 --- a/pytwitter/models/__init__.py +++ b/pytwitter/models/__init__.py @@ -1,6 +1,7 @@ from .user import * # noqa from .tweet import * # noqa from .media import * # noqa +from .media_upload import * # noqa from .poll import * # noqa from .place import * # noqa from .compliance import * # noqa diff --git a/pytwitter/models/media_upload.py b/pytwitter/models/media_upload.py new file mode 100644 index 0000000..8e47776 --- /dev/null +++ b/pytwitter/models/media_upload.py @@ -0,0 +1,69 @@ +""" + Media upload response object: + + Refer: https://developer.twitter.com/en/docs/twitter-api/v1/media/upload-media/api-reference/post-media-upload +""" + +from dataclasses import dataclass, field +from typing import Optional + +from .base import BaseModel + + +@dataclass +class MediaUploadResponseProcessingInfoError(BaseModel): + """ + A class representing the media upload response processing info error object. + """ + + code: Optional[int] = field(default=None) + name: Optional[str] = field(default=None) + message: Optional[str] = field(default=None) + + +@dataclass +class MediaUploadResponseProcessingInfo(BaseModel): + """ + A class representing the media upload response processing info object. + """ + + state: Optional[str] = field(default=None) + check_after_secs: Optional[int] = field(default=None) + progress_percent: Optional[int] = field(default=None) + error: Optional[MediaUploadResponseProcessingInfoError] = field(default=None) + + +@dataclass +class MediaUploadResponseImage(BaseModel): + """ + A class representing the media upload response image object. + """ + + image_type: Optional[str] = field(default=None) + w: Optional[int] = field(default=None) + h: Optional[int] = field(default=None) + + +@dataclass +class MediaUploadResponseVideo(BaseModel): + """ + A class representing the media upload response video object. + """ + + video_type: Optional[str] = field(default=None) + + +@dataclass +class MediaUploadResponse(BaseModel): + """ + A class representing the media upload response object. + """ + + media_id: Optional[int] = field(default=None) + media_id_string: Optional[str] = field(default=None) + media_key: Optional[str] = field(default=None, repr=False) + size: Optional[int] = field(default=None, repr=False) + expires_after_secs: Optional[int] = field(default=None, repr=False) + processing_info: Optional[MediaUploadResponseProcessingInfo] = field(default=None) + image: Optional[MediaUploadResponseImage] = field(default=None) + video: Optional[MediaUploadResponseVideo] = field(default=None) diff --git a/testdata/apis/media_upload/upload_chunk_finalize_resp.json b/testdata/apis/media_upload/upload_chunk_finalize_resp.json new file mode 100644 index 0000000..1d96530 --- /dev/null +++ b/testdata/apis/media_upload/upload_chunk_finalize_resp.json @@ -0,0 +1 @@ +{"media_id":1726870404957175808,"media_id_string":"1726870404957175808","media_key":"7_1726870404957175808","size":2111839,"expires_after_secs":86400,"processing_info":{"state":"pending","check_after_secs":1}} \ No newline at end of file diff --git a/testdata/apis/media_upload/upload_chunk_init_resp.json b/testdata/apis/media_upload/upload_chunk_init_resp.json new file mode 100644 index 0000000..e76d584 --- /dev/null +++ b/testdata/apis/media_upload/upload_chunk_init_resp.json @@ -0,0 +1 @@ +{"media_id":1726870404957175808,"media_id_string":"1726870404957175808","expires_after_secs":86400,"media_key":"7_1726870404957175808"} \ No newline at end of file diff --git a/testdata/apis/media_upload/upload_chunk_status_resp.json b/testdata/apis/media_upload/upload_chunk_status_resp.json new file mode 100644 index 0000000..dae4e90 --- /dev/null +++ b/testdata/apis/media_upload/upload_chunk_status_resp.json @@ -0,0 +1 @@ +{"media_id":1726870404957175808,"media_id_string":"1726870404957175808","media_key":"7_1726870404957175808","size":2111839,"expires_after_secs":85119,"video":{"video_type":"video/mp4"},"processing_info":{"state":"succeeded","progress_percent":100}} \ No newline at end of file diff --git a/testdata/apis/media_upload/upload_simple_resp.json b/testdata/apis/media_upload/upload_simple_resp.json new file mode 100644 index 0000000..453adb2 --- /dev/null +++ b/testdata/apis/media_upload/upload_simple_resp.json @@ -0,0 +1 @@ +{"media_id":1726817595448610817,"media_id_string":"1726817595448610817","media_key":"3_1726817595448610817","size":1864939,"expires_after_secs":86400,"image":{"image_type":"image/jpeg","w":4928,"h":3280}} \ No newline at end of file diff --git a/testdata/apis/media_upload/x-logo.png b/testdata/apis/media_upload/x-logo.png new file mode 100644 index 0000000..4ad8abf Binary files /dev/null and b/testdata/apis/media_upload/x-logo.png differ diff --git a/testdata/apis/media_upload/xaa b/testdata/apis/media_upload/xaa new file mode 100644 index 0000000..9e464b4 Binary files /dev/null and b/testdata/apis/media_upload/xaa differ diff --git a/testdata/apis/media_upload/xab b/testdata/apis/media_upload/xab new file mode 100644 index 0000000..9fc778c Binary files /dev/null and b/testdata/apis/media_upload/xab differ diff --git a/tests/apis/test_media_upload.py b/tests/apis/test_media_upload.py new file mode 100644 index 0000000..0aca4f9 --- /dev/null +++ b/tests/apis/test_media_upload.py @@ -0,0 +1,156 @@ +""" + Tests for media upload API +""" +import base64 + +import pytest +import responses + +from pytwitter import PyTwitterError + + +@responses.activate +def test_media_upload_simple(api_with_user, helpers): + with pytest.raises(PyTwitterError): + api_with_user.upload_media_simple() + + responses.add( + responses.POST, + url="https://upload.twitter.com/1.1/media/upload.json", + json=helpers.load_json_data( + "testdata/apis/media_upload/upload_simple_resp.json" + ), + ) + + with open("testdata/apis/media_upload/x-logo.png", "rb") as media: + resp = api_with_user.upload_media_simple( + media=media, + media_category="tweet_image", + additional_owners=["123456789"], + ) + assert resp.media_id_string == "1726817595448610817" + + with open("testdata/apis/media_upload/x-logo.png", "rb") as f: + media_data = base64.b64encode(f.read()).decode("utf-8") + resp_json = api_with_user.upload_media_simple( + media_data=media_data, + return_json=True, + ) + assert resp_json["media_id_string"] == "1726817595448610817" + + +@responses.activate +def test_upload_media_chunked_init(api_with_user, helpers): + responses.add( + responses.POST, + url="https://upload.twitter.com/1.1/media/upload.json", + json=helpers.load_json_data( + "testdata/apis/media_upload/upload_chunk_init_resp.json" + ), + ) + + resp = api_with_user.upload_media_chunked_init( + total_bytes=1000000, + media_type="video/mp4", + media_category="tweet_video", + additional_owners=["123456789"], + ) + assert resp.media_id_string == "1726870404957175808" + + resp_json = api_with_user.upload_media_chunked_init( + total_bytes=1000000, + media_type="video/mp4", + return_json=True, + ) + assert resp_json["media_id"] == 1726870404957175808 + + +@responses.activate +def test_upload_media_chunked_append(api_with_user, helpers): + media_id = "1726870404957175808" + + with pytest.raises(PyTwitterError): + api_with_user.upload_media_chunked_append(media_id=media_id, segment_index=0) + + responses.add( + responses.POST, + url="https://upload.twitter.com/1.1/media/upload.json", + ) + + with open("testdata/apis/media_upload/xaa", "rb") as media: + status = api_with_user.upload_media_chunked_append( + media_id=media_id, + media=media, + segment_index=0, + ) + assert status + + with open("testdata/apis/media_upload/xab", "rb") as f: + media_data = base64.b64encode(f.read()).decode("utf-8") + status = api_with_user.upload_media_chunked_append( + media_id=media_id, + media_data=media_data, + segment_index=1, + ) + assert status + + responses.add( + responses.POST, + url="https://upload.twitter.com/1.1/media/upload.json", + status=401, + json={"errors": [{"code": 32, "message": "Could not authenticate you."}]}, + ) + with pytest.raises(PyTwitterError): + api_with_user.upload_media_chunked_append( + media_id=media_id, + media_data=media_data, + segment_index=1, + ) + + +@responses.activate +def test_upload_media_chunked_finalize(api_with_user, helpers): + media_id = "1726870404957175808" + + responses.add( + responses.POST, + url="https://upload.twitter.com/1.1/media/upload.json", + json=helpers.load_json_data( + "testdata/apis/media_upload/upload_chunk_finalize_resp.json" + ), + ) + + resp = api_with_user.upload_media_chunked_finalize( + media_id=media_id, + ) + assert resp.media_id_string == media_id + + resp_json = api_with_user.upload_media_chunked_finalize( + media_id=media_id, + return_json=True, + ) + assert resp_json["media_id_string"] == media_id + + +@responses.activate +def test_upload_media_chunked_status(api_with_user, helpers): + media_id = "1726870404957175808" + + responses.add( + responses.GET, + url="https://upload.twitter.com/1.1/media/upload.json", + json=helpers.load_json_data( + "testdata/apis/media_upload/upload_chunk_status_resp.json" + ), + ) + + resp = api_with_user.upload_media_chunked_status( + media_id=media_id, + ) + assert resp.processing_info.state == "succeeded" + + resp_json = api_with_user.upload_media_chunked_status( + media_id=media_id, + return_json=True, + ) + assert resp_json["processing_info"]["state"] == "succeeded"