diff --git a/.github/workflows/docs.yaml b/.github/workflows/docs.yaml index 0d66de71..bf7e3368 100644 --- a/.github/workflows/docs.yaml +++ b/.github/workflows/docs.yaml @@ -7,9 +7,9 @@ jobs: deploy: runs-on: ubuntu-latest steps: - - uses: actions/checkout@v2 - - uses: actions/setup-python@v2 + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 with: - python-version: 3.8 + python-version: 3.12 - run: pip install mkdocs-material - run: mkdocs gh-deploy -f docs/mkdocs.yml --force \ No newline at end of file diff --git a/.github/workflows/release.yaml b/.github/workflows/release.yaml index 74c7f169..7ef81fd6 100644 --- a/.github/workflows/release.yaml +++ b/.github/workflows/release.yaml @@ -7,9 +7,9 @@ jobs: build: runs-on: ubuntu-latest steps: - - uses: actions/checkout@v2 + - uses: actions/checkout@v4 - name: Build and publish to pypi - uses: JRubics/poetry-publish@v1.17 + uses: JRubics/poetry-publish@v2.1 with: pypi_token: ${{ secrets.PYPI_TOKEN }} @@ -19,7 +19,7 @@ jobs: .github/hack/changelog.sh $VERSION > NEW-VERSION-CHANGELOG.md - name: Publish - uses: softprops/action-gh-release@v1 + uses: softprops/action-gh-release@v2 with: body_path: NEW-VERSION-CHANGELOG.md files: 'python-twitter-*' diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml index 77306d04..4430706a 100644 --- a/.github/workflows/test.yaml +++ b/.github/workflows/test.yaml @@ -11,19 +11,19 @@ jobs: runs-on: ubuntu-latest strategy: matrix: - python-version: ['3.7', '3.8', '3.9', '3.10', '3.11', '3.12'] + python-version: ['3.8', '3.9', '3.10', '3.11', '3.12'] include: - - python-version: '3.10' + - python-version: '3.12' update-coverage: true steps: - - uses: actions/checkout@v3 + - uses: actions/checkout@v4 - name: Set up Python ${{ matrix.python-version }} - uses: actions/setup-python@v4 + uses: actions/setup-python@v5 with: python-version: ${{ matrix.python-version }} - name: Cache pip - uses: actions/cache@v2 + uses: actions/cache@v4 with: path: ~/.cache/pip key: ${{ matrix.python-version }}-poetry-${{ hashFiles('pyproject.toml') }} @@ -48,15 +48,15 @@ jobs: runs-on: ubuntu-latest steps: - - uses: actions/checkout@v3 - - uses: actions/setup-python@v4 + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 with: - python-version: '3.10' + python-version: '3.8' - name: Cache pip - uses: actions/cache@v2 + uses: actions/cache@v4 with: path: ~/.cache/pip - key: lintenv-v2 + key: lintenv-v2-38 - name: Install dependencies run: python -m pip install --upgrade pip black - name: Black test diff --git a/docs/docs/usage/media-upload-v2/chunked-upload.md b/docs/docs/usage/media-upload-v2/chunked-upload.md new file mode 100644 index 00000000..bda4c592 --- /dev/null +++ b/docs/docs/usage/media-upload-v2/chunked-upload.md @@ -0,0 +1,91 @@ +This guide will help you make your first requests to upload media using the X API v2 media upload endpoint(s). + +You can get more information for this at [docs](https://docs.x.com/x-api/media/quickstart/media-upload-chunked) + +For video or chunked uploads, you must: + +1. Initialize the upload using the `INIT` command +2. Upload each chunk of bytes using the `APPEND` command +3. Complete the upload using the `FINALIZE` command + +let's do it, Now we need to upload a big video with a filename `/path/to/video.mp4` + +### Step 1: Initialize the upload + +As first step, you need to initialize the upload. + +```python + +import os + +filename = "/path/to/video.mp4" + +init_resp = myapi.upload_media_chunked_init_v2( + total_bytes=os.path.getsize(filename), + media_type="video/mp4", + media_category="tweet_video", +) +print(init_resp) +# Response(data=MediaUpload(id='1912334964932374529', media_key='7_1912334964932374529', processing_info=None, image=None, video=None)) +``` + +### Step 2: Append the file by chunks + +Once we have the media identifiers `id` from the `init_resp`, we can start uploading the file by chunks. + +```python + +media_id = init_resp.data.id + +chunk_size = 2 * 1024 * 1024 +segment_index = 0 +with open(filename, "rb") as f: + while True: + chunk = f.read(chunk_size) + if not chunk: + break + + chunk_resp = myapi.upload_media_chunked_append_v2( + media_id=media_id, + media=chunk, + segment_index=segment_index, + ) + print(chunk_resp) + segment_index += 1 + +# True +``` + +### Step 3: Finalize the upload + +Everything is ok, we need finalize the upload. + +```python +finalize_resp = myapi.upload_media_chunked_finalize_v2(media_id=media_id) +print(finalize_resp) +# Response(data=MediaUpload(id='1912090619981471744', media_key='7_1912090619981471744', processing_info=MediaUploadResponseProcessingInfo(state='succeeded', check_after_secs=None, progress_percent=None, error=None), image=None, video=None)) +``` + +### Step 4 (Optional): Check the processing status + +Once you have finalized the upload, you can check the processing status. + +```python +status_resp = myapi.upload_media_chunked_status_v2(media_id=media_id) +print(status_resp) +# Response(data=MediaUpload(id='1912090619981471744', media_key='7_1912090619981471744', processing_info=MediaUploadResponseProcessingInfo(state='succeeded', check_after_secs=None, progress_percent=100, error=None), image=None, video=None)) +``` + +### Step 5: Create tweet with media + +Congratulations, you have uploaded a video using the X API v2 media upload endpoint(s). + +Now we can create a tweet with this video. + +```python +tweet_resp = myapi.create_tweet(text="My first tweet with a video", media_media_ids=[media_id]) + +# Tweet(id=1912338879258194343, text=My first tweet with a video...) +``` + +Enjoy it! diff --git a/docs/docs/usage/media-upload-v2/simple-upload.md b/docs/docs/usage/media-upload-v2/simple-upload.md new file mode 100644 index 00000000..7f488d65 --- /dev/null +++ b/docs/docs/usage/media-upload-v2/simple-upload.md @@ -0,0 +1,13 @@ +You can use media upload endpoint to upload simple media, images(gifs). + +You can get more information for this at [docs](https://docs.x.com/x-api/media/media-upload) + +## upload simple + +```python + +with open("path/to/image", "rb") as media: + resp = my_api.upload_media_simple_v2(media=media) + print(resp) +``` + diff --git a/docs/mkdocs.yml b/docs/mkdocs.yml index 8bd572b7..21cbc3f4 100644 --- a/docs/mkdocs.yml +++ b/docs/mkdocs.yml @@ -37,6 +37,10 @@ nav: - Media Upload: - Simple Upload: usage/media-upload/simple-upload.md - Chunked Upload: usage/media-upload/chunked-upload.md + + - Media Upload V2: + - Simple Upload: usage/media-upload-v2/simple-upload.md + - Chunked Upload: usage/media-upload-v2/chunked-upload.md - Tweets: - Tweet Lookup: usage/tweets/tweet-lookup.md - Manage Tweets: usage/tweets/tweet-manage.md diff --git a/examples/media_upload_v2.py b/examples/media_upload_v2.py new file mode 100644 index 00000000..f564f9c6 --- /dev/null +++ b/examples/media_upload_v2.py @@ -0,0 +1,64 @@ +""" + Upload a video with media upload v2 +""" + +import os + +from pytwitter import Api + +consumer_key = "your app consumer key" +consumer_secret = "your app consumer secret" +access_token = "your access token" +access_secret = "your access token secret" + +# init api with OAuth1.0 +api = Api( + consumer_key=consumer_key, + consumer_secret=consumer_secret, + access_token=access_token, + access_secret=access_secret, +) + +video_path = "path/to/video.mp4" + +# init media upload + +init_resp = api.upload_media_chunked_init_v2( + total_bytes=os.path.getsize(video_path), + media_type="video/mp4", + media_category="tweet_video", +) + +print(f"Init response: {init_resp}") + +# upload by chunk +chunk_size = 1024 * 1024 * 1 +segment_index = 0 + +with open(video_path, "rb") as f: + while True: + chunk = f.read(chunk_size) + if not chunk: + break + + chunk_resp = api.upload_media_chunked_append_v2( + media_id=init_resp.data.id, + segment_index=segment_index, + media=chunk, + ) + print(f"Chunk response: {chunk_resp}") + +print("Finished chunk upload") + +# finalize upload +finalize_resp = api.upload_media_chunked_finalize_v2( + media_id=init_resp.data.id, +) +print(f"Finalize response: {finalize_resp}") + +# Now you can use the media to create tweet +tweet_resp = api.create_tweet( + text="Tweet with video", media_media_ids=[init_resp.data.id] +) + +print(f"Tweet response: {tweet_resp}") diff --git a/pytwitter/api.py b/pytwitter/api.py index 08eeecb5..4ad80307 100644 --- a/pytwitter/api.py +++ b/pytwitter/api.py @@ -410,6 +410,32 @@ def _parse_response(resp: Response) -> dict: return data + @staticmethod + def _format_response(resp_json, cls, multi=False) -> md.Response: + data, includes, meta, errors = ( + resp_json.get("data", []), + resp_json.get("includes"), + resp_json.get("meta"), + resp_json.get("errors"), + ) + if multi: + data = [cls.new_from_json_dict(item) for item in data] + else: + data = cls.new_from_json_dict(data) + + res = md.Response( + data=data, + includes=md.Includes.new_from_json_dict(includes), + meta=md.Meta.new_from_json_dict(meta), + errors=( + [md.Error.new_from_json_dict(err) for err in errors] + if errors is not None + else None + ), + _json=resp_json, + ) + return res + def _get( self, url: str, @@ -434,29 +460,7 @@ def _get( if return_json: return resp_json else: - data, includes, meta, errors = ( - resp_json.get("data", []), - resp_json.get("includes"), - resp_json.get("meta"), - resp_json.get("errors"), - ) - if multi: - data = [cls.new_from_json_dict(item) for item in data] - else: - data = cls.new_from_json_dict(data) - - res = md.Response( - data=data, - includes=md.Includes.new_from_json_dict(includes), - meta=md.Meta.new_from_json_dict(meta), - errors=( - [md.Error.new_from_json_dict(err) for err in errors] - if errors is not None - else None - ), - _json=resp_json, - ) - return res + return self._format_response(resp_json, cls, multi) def get_tweets( self, @@ -762,6 +766,202 @@ def upload_media_chunked_status( else: return md.MediaUploadResponse.new_from_json_dict(data=data) + def upload_media_simple_v2( + self, + media: Optional[bytes] = None, + media_category: Optional[str] = None, + additional_owners: Optional[List[str]] = None, + return_json: bool = False, + ) -> Union[dict, md.MediaUpload]: + """ + 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. + :param media_category: A string enum value which identifies a media use-case. + This identifier is used to enforce use-case specific constraints (e.g. file size, video duration) and enable advanced features. + Possible values: + - tweet_image + - tweet_gif + - tweet_video + - amplify_video + - dm_video + - subtitles + :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. + Unique identifier of this User. This is returned as a string in order to avoid complications with + languages and tools that cannot handle large integers. + :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 + 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_URL_V2}/media/upload", + verb="POST", + data=args, + files=files, + ) + data = self._parse_response(resp=resp) + if return_json: + return data + else: + return md.MediaUpload.new_from_json_dict(data=data) + + def upload_media_chunked_init_v2( + 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.Response]: + """ + 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: A string enum value which identifies a media use-case. + This identifier is used to enforce use-case specific constraints (e.g. file size, video duration) and enable advanced features. + Possible values: + - tweet_image + - tweet_gif + - tweet_video + - amplify_video + - dm_video + - subtitles + :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. + Unique identifier of this User. This is returned as a string in order to avoid complications with + languages and tools that cannot handle large integers. + :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_URL_V2}/media/upload", + verb="POST", + data=args, + ) + data = self._parse_response(resp=resp) + if return_json: + return data + else: + return self._format_response(resp_json=data, cls=md.MediaUpload) + + def upload_media_chunked_append_v2( + self, + media_id: str, + segment_index: int, + media: Optional[bytes], + ) -> 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. + :return: True if upload success. + """ + resp = self._request( + url=f"{self.BASE_URL_V2}/media/upload", + verb="POST", + params={ + "command": "APPEND", + "media_id": media_id, + }, + data={"segment_index": segment_index}, + files={"media": media}, + ) + if resp.ok: + return True + raise PyTwitterError(resp.json()) + + def upload_media_chunked_finalize_v2( + self, + media_id: str, + return_json: bool = False, + ) -> Union[dict, md.Response]: + """ + 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_URL_V2}/media/upload", + verb="POST", + params={ + "command": "FINALIZE", + "media_id": media_id, + }, + ) + data = self._parse_response(resp=resp) + if return_json: + return data + else: + return self._format_response(resp_json=data, cls=md.MediaUpload) + + def upload_media_chunked_status_v2( + self, + media_id: str, + return_json: bool = False, + ) -> Union[dict, md.Response]: + """ + 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_URL_V2}/media/upload", + verb="GET", + params={ + "command": "STATUS", + "media_id": media_id, + }, + ) + data = self._parse_response(resp=resp) + if return_json: + return data + else: + return self._format_response(resp_json=data, cls=md.MediaUpload) + def create_tweet( self, *, diff --git a/pytwitter/models/ext.py b/pytwitter/models/ext.py index 3813a486..2f6f92d1 100644 --- a/pytwitter/models/ext.py +++ b/pytwitter/models/ext.py @@ -10,6 +10,7 @@ from . import ( BaseModel, Media, + MediaUpload, Place, Poll, Tweet, @@ -107,6 +108,7 @@ class Response: User, Tweet, Media, + MediaUpload, Poll, Place, Space, diff --git a/pytwitter/models/media_upload.py b/pytwitter/models/media_upload.py index 8e477763..b73e4c3d 100644 --- a/pytwitter/models/media_upload.py +++ b/pytwitter/models/media_upload.py @@ -67,3 +67,22 @@ class MediaUploadResponse(BaseModel): processing_info: Optional[MediaUploadResponseProcessingInfo] = field(default=None) image: Optional[MediaUploadResponseImage] = field(default=None) video: Optional[MediaUploadResponseVideo] = field(default=None) + + +@dataclass +class MediaUploadImage(MediaUploadResponseImage): ... + + +@dataclass +class MediaUploadVideo(MediaUploadResponseVideo): ... + + +@dataclass +class MediaUpload(BaseModel): + id: Optional[str] = field(default=None) + media_key: Optional[str] = field(default=None) + expires_after_secs: Optional[int] = field(default=None, repr=False) + processing_info: Optional[MediaUploadResponseProcessingInfo] = field(default=None) + size: Optional[int] = field(default=None, repr=False) + image: Optional[MediaUploadImage] = field(default=None) + video: Optional[MediaUploadVideo] = field(default=None) diff --git a/testdata/apis/media_upload_v2/upload_chunk_finalize_resp.json b/testdata/apis/media_upload_v2/upload_chunk_finalize_resp.json new file mode 100644 index 00000000..c39d0e4b --- /dev/null +++ b/testdata/apis/media_upload_v2/upload_chunk_finalize_resp.json @@ -0,0 +1 @@ +{"data":{"id":"1912090619981471744","media_key":"7_1912090619981471744","size":19048761,"expires_after_secs":86400,"processing_info":{"state":"succeeded"}}} \ No newline at end of file diff --git a/testdata/apis/media_upload_v2/upload_chunk_init_resp.json b/testdata/apis/media_upload_v2/upload_chunk_init_resp.json new file mode 100644 index 00000000..9c3cf520 --- /dev/null +++ b/testdata/apis/media_upload_v2/upload_chunk_init_resp.json @@ -0,0 +1 @@ +{"data":{"id":"1912103767639719936","expires_after_secs":86400,"media_key":"3_1912103767639719936"}} \ No newline at end of file diff --git a/testdata/apis/media_upload_v2/upload_chunk_status_resp.json b/testdata/apis/media_upload_v2/upload_chunk_status_resp.json new file mode 100644 index 00000000..146619db --- /dev/null +++ b/testdata/apis/media_upload_v2/upload_chunk_status_resp.json @@ -0,0 +1 @@ +{"data":{"expires_after_secs":83800,"id":"1912090619981471744","media_key":"7_1912090619981471744","processing_info":{"progress_percent":100,"state":"succeeded"},"size":19048761}} \ No newline at end of file diff --git a/testdata/apis/media_upload_v2/upload_simple_resp.json b/testdata/apis/media_upload_v2/upload_simple_resp.json new file mode 100644 index 00000000..b80ea3b0 --- /dev/null +++ b/testdata/apis/media_upload_v2/upload_simple_resp.json @@ -0,0 +1 @@ +{"id":"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/tests/apis/test_media_upload_v2.py b/tests/apis/test_media_upload_v2.py new file mode 100644 index 00000000..2b117499 --- /dev/null +++ b/tests/apis/test_media_upload_v2.py @@ -0,0 +1,148 @@ +""" + Tests for media upload API v2 +""" + +import pytest +import responses + +from pytwitter import PyTwitterError + + +@responses.activate +def test_media_upload_simple_v2(api_with_user, helpers): + with pytest.raises(PyTwitterError): + api_with_user.upload_media_simple_v2() + + responses.add( + responses.POST, + url="https://api.twitter.com/2/media/upload", + json=helpers.load_json_data( + "testdata/apis/media_upload_v2/upload_simple_resp.json" + ), + ) + + with open("testdata/apis/media_upload/x-logo.png", "rb") as media: + resp = api_with_user.upload_media_simple_v2( + media=media, + media_category="tweet_image", + additional_owners=["123456789"], + ) + assert resp.id == "1726817595448610817" + + with open("testdata/apis/media_upload/x-logo.png", "rb") as media: + resp = api_with_user.upload_media_simple_v2( + media=media, media_category="tweet_image", return_json=True + ) + assert resp["id"] == "1726817595448610817" + + +@responses.activate +def test_upload_media_chunked_init_v2(api_with_user, helpers): + responses.add( + responses.POST, + url="https://api.twitter.com/2/media/upload", + json=helpers.load_json_data( + "testdata/apis/media_upload_v2/upload_chunk_init_resp.json" + ), + ) + + resp = api_with_user.upload_media_chunked_init_v2( + total_bytes=1000000, + media_type="video/mp4", + media_category="tweet_video", + additional_owners=["123456789"], + ) + assert resp.data.id == "1912103767639719936" + + resp_json = api_with_user.upload_media_chunked_init_v2( + total_bytes=1000000, + media_type="video/mp4", + return_json=True, + ) + assert resp_json["data"]["id"] == "1912103767639719936" + + +@responses.activate +def test_upload_media_chunked_append_v2(api_with_user, helpers): + media_id = "1912090619981471744" + + responses.add( + responses.POST, + url="https://api.twitter.com/2/media/upload", + ) + + with open("testdata/apis/media_upload/x-logo.png", "rb") as media: + segment_index = 0 + while True: + chunk = media.read(1 * 1024 * 1024) + if not chunk: + break + status = api_with_user.upload_media_chunked_append_v2( + media_id=media_id, + media=media, + segment_index=segment_index, + ) + assert status + + segment_index += 1 + + responses.add( + responses.POST, + url="https://api.twitter.com/2/media/upload", + status=401, + json={"errors": [{"code": 32, "message": "Could not authenticate you."}]}, + ) + with pytest.raises(PyTwitterError): + api_with_user.upload_media_chunked_append_v2( + media_id=media_id, + media=b"", + segment_index=1, + ) + + +@responses.activate +def test_upload_media_chunked_finalize_v2(api_with_user, helpers): + media_id = "1912090619981471744" + + responses.add( + responses.POST, + url="https://api.twitter.com/2/media/upload", + json=helpers.load_json_data( + "testdata/apis/media_upload_v2/upload_chunk_finalize_resp.json" + ), + ) + + resp = api_with_user.upload_media_chunked_finalize_v2( + media_id=media_id, + ) + assert resp.data.id == media_id + + resp_json = api_with_user.upload_media_chunked_finalize_v2( + media_id=media_id, + return_json=True, + ) + assert resp_json["data"]["id"] == media_id + + +@responses.activate +def test_upload_media_chunked_status_v2(api_with_user, helpers): + media_id = "1912090619981471744" + + responses.add( + responses.GET, + url="https://api.twitter.com/2/media/upload", + json=helpers.load_json_data( + "testdata/apis/media_upload_v2/upload_chunk_status_resp.json" + ), + ) + + resp = api_with_user.upload_media_chunked_status_v2( + media_id=media_id, + ) + assert resp.data.processing_info.state == "succeeded" + + resp_json = api_with_user.upload_media_chunked_status_v2( + media_id=media_id, + return_json=True, + ) + assert resp_json["data"]["processing_info"]["state"] == "succeeded"