From cd99cd926e409eb5f342d19cd6222646937bac8d Mon Sep 17 00:00:00 2001 From: Deepak Soni Date: Tue, 14 Nov 2023 18:22:10 +0530 Subject: [PATCH] added feature to upload reels --- README.md | 18 ++++++ ensta/BaseHost.py | 147 +++++++++++++++++++++++++++++++++++++++++----- requirements.txt | 2 +- setup.py | 4 +- 4 files changed, 153 insertions(+), 18 deletions(-) diff --git a/README.md b/README.md index cd21013..8fd8e01 100644 --- a/README.md +++ b/README.md @@ -60,6 +60,24 @@ host.upload_post(
+Upload Reels
+ +```python +from ensta import Host + +host = Host(username, password) + +host.upload_reel( + video_path="Video.mp4", + thumbnail_path="Thumbnail.jpg", + caption="Enjoying the winter! ⛄" +) +``` + +
+ +
+ Check Username Availability
```python diff --git a/ensta/BaseHost.py b/ensta/BaseHost.py index 494d8fa..9b89a81 100644 --- a/ensta/BaseHost.py +++ b/ensta/BaseHost.py @@ -2,9 +2,9 @@ import random import string import json -from PIL import Image +import moviepy.editor from pathlib import Path -from time import time +import time from uuid import uuid4 from json import JSONDecodeError from collections.abc import Generator @@ -45,6 +45,7 @@ class BaseHost: x_ig_www_claim: str = None csrf_token: str = None guest: Guest = None + own_username: str = None def __init__(self, session_id: str, proxy: dict[str, str] | None = None) -> None: self.x_ig_www_claim = "hmac." + "".join(random.choices(string.ascii_letters + string.digits + "_-", k=48)) @@ -100,7 +101,7 @@ def authenticated(self) -> bool: http_response = self.request_session.get("https://www.instagram.com/api/v1/accounts/edit/web_form_data/", headers=request_headers) try: - http_response.json() + self.own_username = str(http_response.json()["form_data"]["username"]) return True except JSONDecodeError: return False @@ -844,13 +845,13 @@ def change_display_name(self, display_name: str) -> bool: except JSONDecodeError: raise NetworkError("HTTP Response is not a valid JSON.") - def __upload_photo(self, path: str) -> tuple[str, int, int]: + def __upload_photo(self, path: str, arg_upload_id: str | None = None) -> str: refresh_csrf_token(self) path: Path = Path(path) if path.suffix not in (".jpg", ".jpeg"): raise FileTypeError("Only jpg and jpeg image types are allowed to post.") - upload_id = str(int(time()) * 1000) + upload_id = arg_upload_id if arg_upload_id is not None else str(int(time.time()) * 1000) waterfall_id = str(uuid4()) upload_name = f"{upload_id}_0_{random.randint(1000000000, 9999999999)}" @@ -886,16 +887,72 @@ def __upload_photo(self, path: str) -> tuple[str, int, int]: if response_json.get("status", "") != "ok": raise NetworkError("Response json key 'status' not ok.") if response_json.get("upload_id", "") == "": raise NetworkError("Key 'upload_id' in response json doesn't exist or is invalid.") - image_dimensions: tuple[int, int] = Image.open(path).size + return str(response_json.get("upload_id")) - return str(response_json.get("upload_id")), image_dimensions[0], image_dimensions[1] + except JSONDecodeError: + raise NetworkError("Response not a valid json.") + + def __upload_video(self, path: str, arg_upload_id: str | None = None) -> tuple[bool, any, any, any]: + refresh_csrf_token(self) + video_editor = moviepy.editor.VideoFileClip(path) + + path: Path = Path(path) + waterfall_id = str(uuid4()) + + upload_id = arg_upload_id if arg_upload_id is not None else str(int(time.time()) * 1000) + upload_name = f"{upload_id}_0_{random.randint(1000000000, 9999999999)}" + + own_uid = self.profile(self.own_username).user_id + + rupload_params = { + "is_clips_video": "1", + "retry_context": "{\"num_step_auto_retry\": 0, \"num_reupload\": 0, \"num_step_manual_retry\": 0}", + "media_type": "2", + "xsharing_user_ids": json.dumps([own_uid]), + "upload_id": upload_id, + "upload_media_duration_ms": str(int(video_editor.duration * 1000)), + "upload_media_width": str(video_editor.size[0]), + "upload_media_height": str(video_editor.size[1]) + } + + request_headers__get = { + "accept-encoding": "gzip", + "x-instagram-rupload-params": json.dumps(rupload_params), + "x_fb_video_waterfall_id": waterfall_id, + "x-entity-type": "video/mp4" + } + + http_response__get = self.request_session.get(f"https://i.instagram.com/rupload_igvideo/{upload_name}", headers=request_headers__get) + if http_response__get.status_code != 200: raise NetworkError("Video Upload 'GET' Request failed. Status code not 200.") + + # POST Request + + with open(path, "rb") as file: + video_data = file.read() + video_length = str(len(video_data)) + + request_headers = { + "offset": "0", + "x-entity-name": upload_name, + "x-entity-length": video_length, + "content-type": "application/octet-stream", + "content-length": video_length, + **request_headers__get + } + + http_response = self.request_session.post(f"https://i.instagram.com/rupload_igvideo/{upload_name}", data=video_data, headers=request_headers) + + try: + response_json: dict = http_response.json() + + return response_json.get("status", "") == "ok", video_editor.duration, video_editor.size[0], video_editor.size[1] except JSONDecodeError: raise NetworkError("Response not a valid json.") def upload_post(self, photo_path: str, caption: str = "", archive_only: bool = False, disable_comments: bool = False, like_and_view_counts_disabled: bool = False, video_subtitles_enabled: bool = False) -> bool: # TODO: Implement Return Value refresh_csrf_token(self) - upload_id, width, height = self.__upload_photo(photo_path) + upload_id = self.__upload_photo(photo_path) request_headers: dict = { "accept": "*/*", @@ -926,16 +983,16 @@ def upload_post(self, photo_path: str, caption: str = "", archive_only: bool = F body_json = { "archive_only": archive_only, "caption": caption, - "clips_share_preview_to_feed": 1, - "disable_comments": 1 if disable_comments else 0, + "clips_share_preview_to_feed": "1", + "disable_comments": "1" if disable_comments else "0", "disable_oa_reuse": False, - "igtv_share_preview_to_feed": 1, - "is_meta_only_post": 0, - "is_unified_video": 1, - "like_and_view_counts_disabled": 1 if like_and_view_counts_disabled else 0, + "igtv_share_preview_to_feed": "1", + "is_meta_only_post": "0", + "is_unified_video": "1", + "like_and_view_counts_disabled": "1" if like_and_view_counts_disabled else "0", "source_type": "library", "upload_id": upload_id, - "video_subtitles_enabled": 1 if video_subtitles_enabled else 0 + "video_subtitles_enabled": "1" if video_subtitles_enabled else "0" } http_response = self.request_session.post("https://www.instagram.com/api/v1/media/configure/", headers=request_headers, data=body_json) @@ -947,3 +1004,63 @@ def upload_post(self, photo_path: str, caption: str = "", archive_only: bool = F except JSONDecodeError: raise NetworkError("Response not a valid json.") + + def upload_reel(self, video_path: str, thumbnail_path: str, caption: str = "", archive_only: bool = False, disable_comments: bool = False, like_and_view_counts_disabled: bool = False, video_subtitles_enabled: bool = False) -> bool: # TODO: Implement Return Value + upload_id = str(int(time.time()) * 1000) + + video_success, video_duration, video_width, video_height = self.__upload_video(video_path, upload_id) + + if not video_success: return False + if not self.__upload_photo(thumbnail_path, upload_id): return False + + refresh_csrf_token(self) + request_headers: dict = { + "accept": "*/*", + "accept-language": "en-US,en;q=0.9", + "content-type": "application/x-www-form-urlencoded", + "dpr": "1.30208", + "sec-ch-prefers-color-scheme": "dark", + "sec-ch-ua": "\"Google Chrome\";v=\"119\", \"Chromium\";v=\"119\", \"Not?A_Brand\";v=\"24\"", + "sec-ch-ua-full-version-list": "\"Google Chrome\";v=\"119.0.6045.124\", \"Chromium\";v=\"119.0.6045.124\", \"Not?A_Brand\";v=\"24.0.0.0\"", + "sec-ch-ua-mobile": "?0", + "sec-ch-ua-model": "\"\"", + "sec-ch-ua-platform": "\"Windows\"", + "sec-ch-ua-platform-version": "\"15.0.0\"", + "sec-fetch-dest": "empty", + "sec-fetch-mode": "cors", + "sec-fetch-site": "same-origin", + "viewport-width": "1475", + "x-asbd-id": "129477", + "x-csrftoken": self.csrf_token, + "x-ig-app-id": self.insta_app_id, + "x-ig-www-claim": self.x_ig_www_claim, + "x-instagram-ajax": "1009848613", + "x-requested-with": "XMLHttpRequest", + "Referer": "https://www.instagram.com/", + "Referrer-Policy": "strict-origin-when-cross-origin" + } + + body_json = { + "archive_only": archive_only, + "caption": caption, + "clips_share_preview_to_feed": "1", + "disable_comments": "1" if disable_comments else "0", + "disable_oa_reuse": "0", + "igtv_share_preview_to_feed": "1", + "is_meta_only_post": "0", + "is_unified_video": "1", + "like_and_view_counts_disabled": "1" if like_and_view_counts_disabled else "0", + "source_type": "library", + "upload_id": upload_id, + "video_subtitles_enabled": "1" if video_subtitles_enabled else "0" + } + + http_response = self.request_session.post("https://www.instagram.com/api/v1/media/configure_to_clips/", headers=request_headers, data=body_json) + + try: + response_json: dict = http_response.json() + + return response_json.get("status", "") == "ok" + + except JSONDecodeError: + raise NetworkError("Response not a valid json.") diff --git a/requirements.txt b/requirements.txt index 84d0062..8c70e7d 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,4 +2,4 @@ requests selenium browser-cookie3 fake-useragent -pillow +moviepy diff --git a/setup.py b/setup.py index eb28a14..48d7b44 100644 --- a/setup.py +++ b/setup.py @@ -2,7 +2,7 @@ from pathlib import Path long_description = (Path(__file__).parent / "README.md").read_text(encoding="utf-8") -version = "4.9" +version = "5.0.0" setup( name="ensta", @@ -17,7 +17,7 @@ url="https://github.com/diezo/ensta", download_url=f"https://github.com/diezo/ensta/archive/refs/tags/v{version}.tar.gz", keywords=["instagram-client", "instagram", "api-wrapper", "instagram-scraper", "instagram-api", "instagram-sdk", "instagram-photos", "instagram-api-python", "instabot", "instagram-stories", "instagram-bot", "instapy", "instagram-downloader", "instagram-account", "instagram-crawler", "instagram-private-api", "igtv", "instagram-automation", "reels", "instagram-feed"], - install_requires=["requests", "selenium", "fake-useragent", "pillow"], + install_requires=["requests", "selenium", "fake-useragent", "moviepy"], classifiers=[ "Development Status :: 5 - Production/Stable", "Intended Audience :: Developers",