Skip to content

Commit

Permalink
updates
Browse files Browse the repository at this point in the history
  • Loading branch information
MoonStorage committed Jun 17, 2023
1 parent 0c879f2 commit 4db9ee8
Show file tree
Hide file tree
Showing 7 changed files with 329 additions and 4 deletions.
208 changes: 206 additions & 2 deletions ensta/Host.py
Original file line number Diff line number Diff line change
@@ -1,15 +1,18 @@
import requests
from json import JSONDecodeError
import random
import string
import json
from json import JSONDecodeError
from collections.abc import Generator
from .Guest import Guest
from .lib.Commons import (
refresh_csrf_token,
update_app_id,
update_homepage_source,
update_session,
format_identifier
format_identifier,
format_url,
format_username
)
from .lib import (
SessionError,
Expand All @@ -20,6 +23,8 @@
)
from .containers import (FollowedStatus, UnfollowedStatus, FollowPerson)
from .containers.ProfileHost import ProfileHost
from .containers.Post import Post
from .containers.PostUser import PostUser

USERNAME = 0
UID = 1
Expand Down Expand Up @@ -366,6 +371,72 @@ def followings(self, identifier: str | int, count: int = 0) -> Generator[FollowP
yield None
return None

def posts(self, username: str, count: int = 0) -> Generator[Post, None, None]:
username = format_username(username)

refresh_csrf_token(self)
request_headers = {
"accept": "*/*",
"accept-language": "en-US,en;q=0.9",
"sec-ch-prefers-color-scheme": self.preferred_color_scheme,
"sec-ch-ua": "\"Not.A/Brand\";v=\"8\", \"Chromium\";v=\"114\", \"Google Chrome\";v=\"114\"",
"sec-ch-ua-full-version-list": "\"Not.A/Brand\";v=\"8.0.0.0\", \"Chromium\";v=\"114.0.5735.110\", \"Google Chrome\";v=\"114.0.5735.110\"",
"sec-ch-ua-mobile": "?0",
"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-requested-with": "XMLHttpRequest",
"Referer": f"https://www.instagram.com/{username}/",
"Referrer-Policy": "strict-origin-when-cross-origin"
}

current_max_id = ""
generated_count = 0

while True:
current_max_id_text = ""

if current_max_id != "":
current_max_id_text = f"&max_id={current_max_id}"

try:
count_text = 35

if count < 35:
count_text = count

http_response = self.request_session.get(f"https://www.instagram.com/api/v1/feed/user/{username}/username/?count={count_text}{current_max_id_text}", headers=request_headers)
response_json = http_response.json()

if "status" not in response_json or "items" not in response_json:
yield None
return None

if response_json["status"] != "ok":
yield None
return None

for each_item in response_json["items"]:
if generated_count < count or count == 0:

yield self._process_post_data(each_item)
generated_count += 1

if (generated_count < count or count == 0) and "next_max_id" in response_json:
current_max_id = response_json["next_max_id"]
else:
return None
except JSONDecodeError:
yield None
return None

def _identifier(self, identifier: str | int, required: str | int):
identifier = format_identifier(identifier)

Expand Down Expand Up @@ -463,3 +534,136 @@ def profile(self, username: str) -> ProfileHost | None:

def get_username(self, uid: str | int) -> str | None:
return self.guest.get_username(uid, __session__=self.request_session)

def post(self, share_url: str) -> Post | None:
share_url: str = format_url(share_url)
request_headers = {
"accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7",
"accept-language": "en-US,en;q=0.9",
"cache-control": "max-age=0",
"sec-ch-prefers-color-scheme": "dark",
"sec-ch-ua": "\"Not.A/Brand\";v=\"8\", \"Chromium\";v=\"114\", \"Google Chrome\";v=\"114\"",
"sec-ch-ua-full-version-list": "\"Not.A/Brand\";v=\"8.0.0.0\", \"Chromium\";v=\"114.0.5735.110\", \"Google Chrome\";v=\"114.0.5735.110\"",
"sec-ch-ua-mobile": "?0",
"sec-ch-ua-platform": "\"Windows\"",
"sec-ch-ua-platform-version": "\"15.0.0\"",
"sec-fetch-dest": "document",
"sec-fetch-mode": "navigate",
"sec-fetch-site": "same-origin",
"sec-fetch-user": "?1",
"upgrade-insecure-requests": "1",
"viewport-width": "1475",
"referrerPolicy": "strict-origin-when-cross-origin"
}

http_response = self.request_session.get(share_url, headers=request_headers)
response_text = http_response.text

required_text = "{\"response\":\"{\\\"items\\\":"

initial_index = response_text.rfind(required_text)
if initial_index == -1: raise APIError()

rest_text = response_text[initial_index - len(required_text) + len(required_text): len(response_text)]
end_text = "\\\"auto_load_more_enabled\\\":false}\",\"status_code\":200}"
end_index = rest_text.find(end_text)

try:
data: dict = json.loads(rest_text[:end_index + len(end_text)])

if "response" in data:
response_data: dict = json.loads(data["response"])

if "items" in response_data:
items_data: list = response_data["items"]

if len(items_data) > 0:
selected_post: dict = items_data[len(items_data) - 1]

if selected_post is not None:
return self._process_post_data(selected_post)

except JSONDecodeError:
return None

def _process_post_data(self, data: dict) -> Post:

caption: dict = data.get("caption", None)

caption_text = ""
is_caption_covered = False
caption_created_at = 0
caption_share_enabled = False
caption_did_report_as_spam = False

if caption is not None:
caption_text = caption.get("text", "")
is_caption_covered = caption.get("is_covered", False)
caption_created_at = caption.get("created_at", 0)
caption_share_enabled = caption.get("share_enabled", False)
caption_did_report_as_spam = caption.get("did_report_as_spam", False)

user: PostUser = PostUser()
user_data: dict = data.get("user", None)

if user_data is not None:
user: PostUser = PostUser(
has_anonymous_profile_picture=user_data.get("has_anonymous_profile_picture", False),
fbid_v2=user_data.get("fbid_v2", ""),
transparency_product_enabled=user_data.get("transparency_product_enabled", False),
is_favorite=user_data.get("is_favorite", False),
is_unpublished=user_data.get("is_unpublished", False),
uid=user_data.get("pk", ""),
username=user_data.get("username", ""),
full_name=user_data.get("full_name", ""),
is_private=user_data.get("is_private", False),
is_verified=user_data.get("is_verified", False),
profile_picture_id=user_data.get("profile_pic_id", ""),
profile_picture_url=user_data.get("profile_pic_url", ""),
account_badges=user_data.get("account_badges", []),
feed_post_reshare_disabled=user_data.get("feed_post_reshare_disabled", False),
show_account_transparency_details=user_data.get("show_account_transparency_details", False),
third_party_downloads_enabled=user_data.get("third_party_downloads_enabled", 0),
latest_reel_media=user_data.get("latest_reel_media", 0)
)

return Post(
_instance=self.__class__,
share_url=f"https://www.instagram.com/p/{data.get('code', '')}",
taken_at=data.get("taken_at", 0),
unique_key=data.get("pk", ""),
media_type=data.get("media_type", 0),
code=data.get("code", ""),
caption_is_edited=data.get("caption_is_edited", False),
original_media_has_visual_reply_media=data.get("original_media_has_visual_reply_media", False),
like_and_view_counts_disabled=data.get("like_and_view_counts_disabled", False),
can_viewer_save=data.get("can_viewer_save", False),
profile_grid_control_enabled=data.get("profile_grid_control_enabled", False),
is_comments_gif_composer_enabled=data.get("is_comments_gif_composer_enabled", False),
comment_threading_enabled=data.get("comment_threading_enabled", False),
comment_count=data.get("comment_count", 0),
has_liked=data.get("has_liked", False),
user=user,
can_viewer_reshare=data.get("can_viewer_reshare", False),
like_count=data.get("like_count", 0),
top_likers=data.get("top_likers", []),
caption_text=caption_text,
is_caption_covered=is_caption_covered,
caption_created_at=caption_created_at,
caption_share_enabled=caption_share_enabled,
caption_did_report_as_spam=caption_did_report_as_spam,
is_paid_partnership=data.get("is_paid_partnership", False),
show_shop_entrypoint=data.get("show_shop_entrypoint", False),
deleted_reason=data.get("deleted_reason", 0),
integrity_review_decision=data.get("integrity_review_decision", ""),
ig_media_sharing_disabled=data.get("ig_media_sharing_disabled", False),
has_shared_to_fb=data.get("has_shared_to_fb", False),
is_unified_video=data.get("is_unified_video", False),
should_request_ads=data.get("should_request_ads", False),
is_visual_reply_commenter_notice_enabled=data.get("is_visual_reply_commenter_notice_enabled", False),
commerciality_status=data.get("commerciality_status", ""),
explore_hide_comments=data.get("explore_hide_comments", False),
has_delayed_metadata=data.get("has_delayed_metadata", False),
location_latitude=data.get("lat", 0),
location_longitude=data.get("lng", 0)
)
1 change: 1 addition & 0 deletions ensta/__init__.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
from ensta.Guest import Guest
from ensta.Host import Host
from ensta.Authentication import NewSessionID
from .lib import Exceptions
92 changes: 92 additions & 0 deletions ensta/containers/Post.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,92 @@
from ensta.containers import PostUser
from dataclasses import dataclass, field
from ensta import Host
from ensta.lib.Commons import refresh_csrf_token
from json import JSONDecodeError


@dataclass(frozen=False)
class Post:
_instance: type[Host] | None = None
share_url: str = ""
taken_at: int = 0
unique_key: str = ""
media_type: int = 0
code: str = ""
caption_is_edited: bool = False
original_media_has_visual_reply_media: bool = False
like_and_view_counts_disabled: bool = False
can_viewer_save: bool = False
profile_grid_control_enabled: bool = False
is_comments_gif_composer_enabled: bool = False
comment_threading_enabled: bool = False
comment_count: int = 0
has_liked: bool = False
user: PostUser = field(default_factory=PostUser)
can_viewer_reshare: bool = False
like_count: int = 0
top_likers: list[str] = field(default_factory=list)
caption_text: str = ""
is_caption_covered: bool = False
caption_created_at: int = 0
caption_share_enabled: bool = False
caption_did_report_as_spam: bool = False
is_paid_partnership: bool = False
show_shop_entrypoint: bool = False
deleted_reason: int = 0
integrity_review_decision: str = ""
ig_media_sharing_disabled: bool = False
has_shared_to_fb: int = 0
is_unified_video: bool = False
should_request_ads: bool = False
is_visual_reply_commenter_notice_enabled: bool = False
commerciality_status: str = ""
explore_hide_comments: bool = False
has_delayed_metadata: bool = False
location_latitude: float = 0
location_longitude: float = 0

def _like_action(self, action: str = "like") -> bool:
if self.unique_key == "": return False

refresh_csrf_token(self._instance)
request_headers = {
"accept": "*/*",
"accept-language": "en-US,en;q=0.9",
"content-type": "application/x-www-form-urlencoded",
"sec-ch-prefers-color-scheme": self._instance.preferred_color_scheme,
"sec-ch-ua": "\"Not.A/Brand\";v=\"8\", \"Chromium\";v=\"114\", \"Google Chrome\";v=\"114\"",
"sec-ch-ua-full-version-list": "\"Not.A/Brand\";v=\"8.0.0.0\", \"Chromium\";v=\"114.0.5735.110\", \"Google Chrome\";v=\"114.0.5735.110\"",
"sec-ch-ua-mobile": "?0",
"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._instance.csrf_token,
"x-ig-app-id": self._instance.insta_app_id,
"x-ig-www-claim": self._instance.x_ig_www_claim,
"x-instagram-ajax": "1007670408",
"x-requested-with": "XMLHttpRequest",
"Referer": f"https://www.instagram.com/p/{self.code}/",
"Referrer-Policy": "strict-origin-when-cross-origin"
}

try:
http_response = self._instance.request_session.post(f"https://www.instagram.com/api/v1/web/likes/{self.unique_key}/{action}/", headers=request_headers)
response_json = http_response.json()

if "status" in response_json:
return response_json["status"] == "ok"
else:
return False
except JSONDecodeError:
return False

def like(self) -> bool:
return self._like_action("like")

def unlike(self) -> bool:
return self._like_action("unlike")
22 changes: 22 additions & 0 deletions ensta/containers/PostUser.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
from dataclasses import dataclass, field


@dataclass(frozen=False)
class PostUser:
has_anonymous_profile_picture: bool = False
fbid_v2: str = ""
transparency_product_enabled: bool = False
is_favorite: bool = False
is_unpublished: bool = False
uid: str = ""
username: str = ""
full_name: str = ""
is_private: bool = False
is_verified: bool = False
profile_picture_id: str = ""
profile_picture_url: str = ""
account_badges: list = field(default_factory=list)
feed_post_reshare_disabled: bool = False
show_account_transparency_details: bool = False
third_party_downloads_enabled: int = 0
latest_reel_media: int = 0
2 changes: 2 additions & 0 deletions ensta/containers/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,3 +3,5 @@
from .ProfileHost import ProfileHost
from .FollowedStatus import FollowedStatus
from .UnfollowedStatus import UnfollowedStatus
from .PostUser import PostUser
from .Post import Post
4 changes: 4 additions & 0 deletions ensta/lib/Commons.py
Original file line number Diff line number Diff line change
Expand Up @@ -71,3 +71,7 @@ def format_uid(uid: str) -> str:

def format_identifier(identifier: str | int) -> str:
return str(identifier).lower().replace(" ", "")


def format_url(url: str) -> str:
return url.strip()
4 changes: 2 additions & 2 deletions setup.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,15 +6,15 @@
setup(
name="ensta",
packages=["ensta", "ensta.lib", "ensta.containers"],
version="2.5",
version="2.6",
license="MIT",
description="🔥 Fastest & Simplest Python Package For Instagram Automation",
long_description=long_description,
long_description_content_type="text/markdown",
author="Deepak Soni",
author_email="[email protected]",
url="https://github.com/diezo/ensta",
download_url="https://github.com/diezo/ensta/archive/refs/tags/v2.5.tar.gz",
download_url="https://github.com/diezo/ensta/archive/refs/tags/v2.6.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"],
classifiers=[
Expand Down

0 comments on commit 4db9ee8

Please sign in to comment.