diff --git a/ensta/Authentication.py b/ensta/Authentication.py index d57a5b1..dbc3422 100644 --- a/ensta/Authentication.py +++ b/ensta/Authentication.py @@ -10,6 +10,7 @@ from .PasswordEncryption import PasswordEncryption from .lib.Exceptions import (AuthenticationError, NetworkError) from .WebSession import WebSession +from .lib.WebHeaders import USER_AGENT, client_hints def new_session_id( @@ -20,8 +21,7 @@ def new_session_id( ) -> str: request_session: Session = requests.Session() - request_session.headers["user-agent"] = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " \ - "(KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36" + request_session.headers["user-agent"] = USER_AGENT if proxy is not None: request_session.proxies.update(proxy) @@ -44,14 +44,7 @@ def new_session_id( "content-type": "application/x-www-form-urlencoded", "dpr": "1.30208", "sec-ch-prefers-color-scheme": "dark", - "sec-ch-ua": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) " - "Chrome/119.0.0.0 Safari/537.36", - "sec-ch-ua-full-version-list": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " - "(KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36", - "sec-ch-ua-mobile": "?0", - "sec-ch-ua-model": "\"\"", - "sec-ch-ua-platform": "\"Windows\"", - "sec-ch-ua-platform-version": "\"15.0.0\"", + **client_hints(), "sec-fetch-dest": "empty", "sec-fetch-mode": "cors", "sec-fetch-site": "same-origin", diff --git a/ensta/Guest.py b/ensta/Guest.py index 5692059..dcbc847 100644 --- a/ensta/Guest.py +++ b/ensta/Guest.py @@ -6,6 +6,7 @@ from .containers.Profile import Profile from .containers.ProfileHost import ProfileHost from .lib.Exceptions import APIError, NetworkError, RateLimitedError +from .lib.WebHeaders import USER_AGENT, IG_APP_ID, ASBD_ID, api_headers, client_hints, csrf_token_from, describe_response from collections.abc import Generator from .containers.Post import Post from .containers.PostUser import PostUser @@ -14,18 +15,19 @@ class Guest: request_session: requests.Session = None homepage_source: str = None - insta_app_id: str = "936619743392459" + insta_app_id: str = IG_APP_ID preferred_color_scheme: str = "dark" - x_ig_www_claim: str + # Instagram issues the real claim via the "x-ig-set-www-claim" response + # header. Until we have seen one, "0" is what a fresh browser sends. + x_ig_www_claim: str = "0" csrf_token: str = None - user_agent: str = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " \ - "(KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36" + user_agent: str = USER_AGENT def __init__(self, proxy: dict[str, str] | None = None) -> None: self.request_session = requests.Session() - self.x_ig_www_claim = "hmac." + "".join(random.choices(string.ascii_letters + string.digits + "_-", k=48)) + self.request_session.headers["user-agent"] = self.user_agent self.csrf_token = "".join(random.choices(string.ascii_letters + string.digits, k=32)) - self.request_session.cookies.set("csrftoken", self.csrf_token) + self.request_session.cookies.set("csrftoken", self.csrf_token, domain=".instagram.com", path="/") if proxy is not None: self.request_session.proxies.update(proxy) @@ -43,16 +45,12 @@ def username_availability(self, username: str) -> bool | None: "accept-language": "en-US,en;q=0.9", "content-type": "application/x-www-form-urlencoded", "sec-ch-prefers-color-scheme": self.preferred_color_scheme, - "sec-ch-ua": self.user_agent, - "sec-ch-ua-full-version-list": self.user_agent, - "sec-ch-ua-mobile": "?0", - "sec-ch-ua-platform": "\"Windows\"", - "sec-ch-ua-platform-version": "\"15.0.0\"", + **client_hints(), "sec-fetch-dest": "empty", "sec-fetch-mode": "cors", "sec-fetch-site": "same-origin", "viewport-width": "1475", - "x-asbd-id": "198387", + "x-asbd-id": ASBD_ID, "x-csrftoken": self.csrf_token, "x-ig-app-id": self.insta_app_id, "x-ig-www-claim": "0", @@ -78,31 +76,27 @@ def username_availability(self, username: str) -> bool | None: def profile(self, username: str, __session__: requests.Session | None = None) -> Profile | ProfileHost | None: username: str = username.replace(" ", "").lower() + session: requests.Session = __session__ + if __session__ is None: session: requests.Session = self.request_session + request_headers: dict = { "accept": "*/*", "accept-language": "en-US,en;q=0.9", "sec-ch-prefers-color-scheme": self.preferred_color_scheme, - "sec-ch-ua": self.user_agent, - "sec-ch-ua-full-version-list": self.user_agent, - "sec-ch-ua-mobile": "?0", - "sec-ch-ua-platform": "\"Windows\"", - "sec-ch-ua-platform-version": "\"15.0.0\"", + **client_hints(), "sec-fetch-dest": "empty", "sec-fetch-mode": "cors", "sec-fetch-site": "same-origin", "viewport-width": "1475", - "x-asbd-id": "198387", - "x-csrftoken": self.csrf_token, + "x-asbd-id": ASBD_ID, + "x-csrftoken": csrf_token_from(session, self.csrf_token), "x-ig-app-id": self.insta_app_id, - "x-ig-www-claim": "0", + "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" } - session: requests.Session = __session__ - if __session__ is None: session: requests.Session = self.request_session - http_response: requests.Response = session.get( f"https://www.instagram.com/api/v1/users/web_profile_info/?username={username}", headers=request_headers, @@ -188,7 +182,7 @@ def profile(self, username: str, __session__: requests.Session | None = None) -> except KeyError: raise APIError() except JSONDecodeError: - raise NetworkError("HTTP Response is not a valid JSON.") + raise NetworkError("HTTP Response is not a valid JSON. " + describe_response(http_response)) def get_uid(self, username: str, __session__: requests.Session | None = None) -> str | None: username: str = username.strip().lower().replace(" ", "") @@ -222,38 +216,32 @@ def get_username(self, uid: str | int, __session__: requests.Session | None = No except JSONDecodeError: raise NetworkError("HTTP Response is not a valid JSON.") - def posts(self, username: str, count: int = 0, __session__: requests.Session | None = None) -> Generator[Post, None, None]: + def posts( + self, + username: str, + count: int = 0, + __session__: requests.Session | None = None, + user_id: str | int | None = None, + ) -> Generator[Post, None, None]: """ Generates a list of target's posts of specified size. :param username: Target's Username :param count: Amount of posts to fetch :param __session__: (Optional) Custom request session object + :param user_id: (Optional) Target's UserID. When given, the numeric feed endpoint is used + instead of the username one, which is what the Instagram web app itself calls. :return: Generator which yields each post's data """ username = username.replace(" ", "").lower() - request_headers = { - "accept": "*/*", - "accept-language": "en-US,en;q=0.9", - "sec-ch-prefers-color-scheme": self.preferred_color_scheme, - "sec-ch-ua": self.user_agent, - "sec-ch-ua-full-version-list": self.user_agent, - "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" - } + session: requests.Session = __session__ + if __session__ is None: session: requests.Session = self.request_session + + if user_id is not None: + feed_url = f"https://www.instagram.com/api/v1/feed/user/{str(user_id).strip()}/" + else: + feed_url = f"https://www.instagram.com/api/v1/feed/user/{username}/username/" current_max_id = "" generated_count = 0 @@ -270,20 +258,46 @@ def posts(self, username: str, count: int = 0, __session__: requests.Session | N if count < 35: count_text = count - session: requests.Session = __session__ - if __session__ is None: session: requests.Session = self.request_session + request_headers = api_headers( + csrf_token=csrf_token_from(session, self.csrf_token), + www_claim=self.x_ig_www_claim, + referer=f"https://www.instagram.com/{username}/", + ) http_response = session.get( - f"https://www.instagram.com/api/v1/feed/user/{username}/username/?count={count_text}" - f"{current_max_id_text}", - headers=request_headers + f"{feed_url}?count={count_text}{current_max_id_text}", + headers=request_headers, + allow_redirects=False, ) - - response_json = http_response.json() + + claim = http_response.headers.get("x-ig-set-www-claim") + if claim: self.x_ig_www_claim = claim + + if http_response.status_code in (301, 302, 303, 307, 308): + yield None + raise RateLimitedError( + "Instagram redirected the feed request to " + f"{http_response.headers.get('location', '?')} — the session or IP is being " + "login-walled. " + describe_response(http_response) + ) + + try: + response_json = http_response.json() + except JSONDecodeError: + yield None + raise NetworkError("HTTP Response is not a valid JSON. " + describe_response(http_response)) + + if response_json.get("require_login") or http_response.status_code in (401, 403): + yield None + raise RateLimitedError( + "Instagram requires a logged-in session for this feed. " + describe_response(http_response) + ) if "status" not in response_json or "items" not in response_json: yield None - raise NetworkError("HTTP response doesn't include 'status' or 'items' node.") + raise NetworkError( + "HTTP response doesn't include 'status' or 'items' node. " + describe_response(http_response) + ) if response_json["status"] != "ok": yield None @@ -294,7 +308,7 @@ def posts(self, username: str, count: int = 0, __session__: requests.Session | N "to a different WiFi Network, or use proxies." ) - raise NetworkError("Request failed.") + raise NetworkError("Request failed. " + describe_response(http_response)) for each_item in response_json["items"]: if generated_count < count or count == 0: @@ -330,16 +344,12 @@ def reels( "accept": "*/*", "accept-language": "en-US,en;q=0.9", "sec-ch-prefers-color-scheme": self.preferred_color_scheme, - "sec-ch-ua": self.user_agent, - "sec-ch-ua-full-version-list": self.user_agent, - "sec-ch-ua-mobile": "?0", - "sec-ch-ua-platform": '"Windows"', - "sec-ch-ua-platform-version": '"15.0.0"', + **client_hints(), "sec-fetch-dest": "empty", "sec-fetch-mode": "cors", "sec-fetch-site": "same-origin", "viewport-width": "1475", - "x-asbd-id": "129477", + "x-asbd-id": ASBD_ID, "x-csrftoken": self.csrf_token, "x-ig-app-id": self.insta_app_id, "x-ig-www-claim": self.x_ig_www_claim, diff --git a/ensta/WebSession.py b/ensta/WebSession.py index d7db9e5..c408df5 100644 --- a/ensta/WebSession.py +++ b/ensta/WebSession.py @@ -25,12 +25,17 @@ IdentifierError, DevelopmentError, APIError, - ConversionError + ConversionError, + RateLimitedError ) from PIL import Image from ensta.lib.Searcher import create_search_obj, search_comments from urllib.parse import urlparse, parse_qs from .Utils import time_id, fb_uploader +from .lib.WebHeaders import ( + USER_AGENT, IG_APP_ID, ASBD_ID, PROFILE_POSTS_DOC_ID, PROFILE_POSTS_CONNECTION, + client_hints, csrf_token_from, describe_response, +) from pyquery import PyQuery USERNAME, UID = 0, 1 @@ -61,16 +66,15 @@ class WebSession: session_data: str request_session: requests.Session - insta_app_id: str = "936619743392459" + insta_app_id: str = IG_APP_ID preferred_color_scheme: str = "dark" - x_ig_www_claim: str + x_ig_www_claim: str = "0" csrf_token: str guest: Guest user_id: str username: str identifier: str - user_agent: str = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " \ - "(KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36" + user_agent: str = USER_AGENT private_user_agent: str = "Instagram 269.0.0.18.75 Android (26/8.0.0; 480dpi; 1080x1920; " \ "OnePlus; 6T Dev; devitron; qcom; en_US; 314665256)" media_resolver: MediaResolver = None @@ -84,27 +88,39 @@ def __init__( ) -> None: self.session_data = session_data - self.x_ig_www_claim = "hmac." + "".join(random.choices(string.ascii_letters + string.digits + "_-", k=48)) self.csrf_token = "".join(random.choices(string.ascii_letters + string.digits, k=32)) self.request_session = requests.Session() self.request_session.headers["user-agent"] = self.user_agent + # Instagram hands out the real www-claim on any authenticated response; + # remember it so subsequent requests look like the same browser tab. + self.request_session.hooks["response"].append(self._remember_www_claim) if proxy is not None: self.request_session.proxies.update(proxy) session_data_json: dict = json.loads(session_data) self.guest = Guest(proxy=proxy) + self.guest.x_ig_www_claim = self.x_ig_www_claim self.user_id = session_data_json.get("user_id") self.username = session_data_json.get("username") self.identifier = session_data_json.get("identifier") - self.request_session.cookies.set("sessionid", session_data_json.get("session_id")) - self.request_session.cookies.set("rur", session_data_json.get("rur")) - self.request_session.cookies.set("mid", session_data_json.get("mid")) - self.request_session.cookies.set("ds_user_id", session_data_json.get("user_id")) - self.request_session.cookies.set("ig_did", session_data_json.get("ig_did")) - self.request_session.cookies.set("csrftoken", self.csrf_token) + # Scope the cookies to .instagram.com exactly as the browser holds them. + # Set without a domain, they would linger next to the fresh + # domain-scoped copies Instagram sends back, and every later request + # would carry two "sessionid"/"csrftoken"/... cookies — which Instagram + # answers with a redirect to the home page or an HTML app shell. + for name, value in ( + ("sessionid", session_data_json.get("session_id")), + ("rur", session_data_json.get("rur")), + ("mid", session_data_json.get("mid")), + ("ds_user_id", session_data_json.get("user_id")), + ("ig_did", session_data_json.get("ig_did")), + ("csrftoken", self.csrf_token), + ): + if value is not None: + self.request_session.cookies.set(name, value, domain=".instagram.com", path="/") if not skip_auth_verification and not self.authenticated(): raise SessionError( @@ -113,6 +129,12 @@ def __init__( self.media_resolver = media_resolver if media_resolver is not None else MediaResolver() + def _remember_www_claim(self, response: requests.Response, *args, **kwargs) -> None: + claim = response.headers.get("x-ig-set-www-claim") + if claim: + self.x_ig_www_claim = claim + self.guest.x_ig_www_claim = claim + def authenticated(self) -> bool: """ Returns whether user is logged in or not. @@ -124,16 +146,12 @@ def authenticated(self) -> bool: "accept": "*/*", "accept-language": "en-US,en;q=0.9", "sec-ch-prefers-color-scheme": self.preferred_color_scheme, - "sec-ch-ua": self.user_agent, - "sec-ch-ua-full-version-list": self.user_agent, - "sec-ch-ua-mobile": "?0", - "sec-ch-ua-platform": "\"Windows\"", - "sec-ch-ua-platform-version": "\"15.0.0\"", + **client_hints(), "sec-fetch-dest": "empty", "sec-fetch-mode": "cors", "sec-fetch-site": "same-origin", "viewport-width": "1475", - "x-asbd-id": "198387", + "x-asbd-id": ASBD_ID, "x-csrftoken": self.csrf_token, "x-ig-app-id": self.insta_app_id, "x-ig-www-claim": self.x_ig_www_claim, @@ -173,16 +191,12 @@ def follow(self, identifier: str | int) -> FollowedStatus | None: "accept-language": "en-US,en;q=0.9", "content-type": "application/x-www-form-urlencoded", "sec-ch-prefers-color-scheme": self.preferred_color_scheme, - "sec-ch-ua": self.user_agent, - "sec-ch-ua-full-version-list": self.user_agent, - "sec-ch-ua-mobile": "?0", - "sec-ch-ua-platform": "\"Windows\"", - "sec-ch-ua-platform-version": "\"15.0.0\"", + **client_hints(), "sec-fetch-dest": "empty", "sec-fetch-mode": "cors", "sec-fetch-site": "same-origin", "viewport-width": "1475", - "x-asbd-id": "198387", + "x-asbd-id": ASBD_ID, "x-csrftoken": self.csrf_token, "x-ig-app-id": self.insta_app_id, "x-ig-www-claim": self.x_ig_www_claim, @@ -248,16 +262,12 @@ def unfollow(self, identifier: str | int) -> UnfollowedStatus | None: "accept-language": "en-US,en;q=0.9", "content-type": "application/x-www-form-urlencoded", "sec-ch-prefers-color-scheme": self.preferred_color_scheme, - "sec-ch-ua": self.user_agent, - "sec-ch-ua-full-version-list": self.user_agent, - "sec-ch-ua-mobile": "?0", - "sec-ch-ua-platform": "\"Windows\"", - "sec-ch-ua-platform-version": "\"15.0.0\"", + **client_hints(), "sec-fetch-dest": "empty", "sec-fetch-mode": "cors", "sec-fetch-site": "same-origin", "viewport-width": "1475", - "x-asbd-id": "198387", + "x-asbd-id": ASBD_ID, "x-csrftoken": self.csrf_token, "x-ig-app-id": self.insta_app_id, "x-ig-www-claim": self.x_ig_www_claim, @@ -316,16 +326,12 @@ def followers(self, identifier: str | int, count: int = 0) -> Generator[FollowPe "accept": "*/*", "accept-language": "en-US,en;q=0.9", "sec-ch-prefers-color-scheme": self.preferred_color_scheme, - "sec-ch-ua": self.user_agent, - "sec-ch-ua-full-version-list": self.user_agent, - "sec-ch-ua-mobile": "?0", - "sec-ch-ua-platform": "\"Windows\"", - "sec-ch-ua-platform-version": "\"15.0.0\"", + **client_hints(), "sec-fetch-dest": "empty", "sec-fetch-mode": "cors", "sec-fetch-site": "same-origin", "viewport-width": "1475", - "x-asbd-id": "198387", + "x-asbd-id": ASBD_ID, "x-csrftoken": self.csrf_token, "x-ig-app-id": self.insta_app_id, "x-ig-www-claim": self.x_ig_www_claim, @@ -411,16 +417,12 @@ def followings(self, identifier: str | int, count: int = 0) -> Generator[FollowP "accept": "*/*", "accept-language": "en-US,en;q=0.9", "sec-ch-prefers-color-scheme": self.preferred_color_scheme, - "sec-ch-ua": self.user_agent, - "sec-ch-ua-full-version-list": self.user_agent, - "sec-ch-ua-mobile": "?0", - "sec-ch-ua-platform": "\"Windows\"", - "sec-ch-ua-platform-version": "\"15.0.0\"", + **client_hints(), "sec-fetch-dest": "empty", "sec-fetch-mode": "cors", "sec-fetch-site": "same-origin", "viewport-width": "1475", - "x-asbd-id": "198387", + "x-asbd-id": ASBD_ID, "x-csrftoken": self.csrf_token, "x-ig-app-id": self.insta_app_id, "x-ig-www-claim": self.x_ig_www_claim, @@ -537,16 +539,12 @@ def _set_account_privacy(self, privacy: str) -> bool: "accept-language": "en-US,en;q=0.9", "content-type": "application/x-www-form-urlencoded", "sec-ch-prefers-color-scheme": self.preferred_color_scheme, - "sec-ch-ua": self.user_agent, - "sec-ch-ua-full-version-list": self.user_agent, - "sec-ch-ua-mobile": "?0", - "sec-ch-ua-platform": "\"Windows\"", - "sec-ch-ua-platform-version": "\"15.0.0\"", + **client_hints(), "sec-fetch-dest": "empty", "sec-fetch-mode": "cors", "sec-fetch-site": "same-origin", "viewport-width": "1475", - "x-asbd-id": "198387", + "x-asbd-id": ASBD_ID, "x-csrftoken": self.csrf_token, "x-ig-app-id": self.insta_app_id, "x-ig-www-claim": self.x_ig_www_claim, @@ -617,15 +615,120 @@ def get_uid(self, username: str) -> str | None: return self.guest.get_uid(username, __session__=self.request_session) - def posts(self, username: str, count: int = 0) -> Generator[Post, None, None]: + def posts( + self, + username: str, + count: int = 0, + user_id: str | int | None = None, + doc_id: str = PROFILE_POSTS_DOC_ID, + ) -> Generator[Post, None, None]: """ Generates a list of target's posts of specified size. + + Logged-in sessions use the GraphQL profile-timeline query the Instagram + web app itself issues; the older /api/v1/feed/user/ REST endpoint now + redirects logged-in sessions to the home page. + :param username: Target's Username - :param count: Amount of posts to fetch + :param count: Amount of posts to fetch (0 = all) + :param user_id: Unused for logged-in sessions; kept for signature compatibility with Guest.posts + :param doc_id: (Optional) Override the persisted GraphQL query id :return: Generator which yields each post's data """ - return self.guest.posts(username, count, __session__=self.request_session) + username = username.replace(" ", "").lower() + page_size = 12 + generated = 0 + cursor: str | None = None + + while True: + variables = { + "data": { + "count": page_size, + "include_relationship_info": True, + "latest_besties_reel_media": True, + "latest_reel_media": True, + }, + "username": username, + } + if cursor: + variables["after"] = cursor + variables["first"] = page_size + + # Deliberately minimal: this is the request shape that Instagram + # answers with JSON. Adding the web app's own markers (x-fb-friendly-name, + # fb_api_req_friendly_name, x-ig-www-claim, sec-fetch-*, ...) without the + # per-page LSD/DTSG tokens makes Instagram answer with the HTML app + # shell instead. + headers = { + "accept": "*/*", + "accept-language": "en-US,en;q=0.8", + "x-ig-app-id": self.insta_app_id, + "x-csrftoken": csrf_token_from(self.request_session, self.csrf_token), + "referer": f"https://www.instagram.com/{username}/", + } + + http_response = self.request_session.post( + "https://www.instagram.com/graphql/query", + data={ + "variables": json.dumps(variables, separators=(",", ":")), + "doc_id": doc_id, + "server_timestamps": "true", + }, + headers=headers, + allow_redirects=False, + ) + + if http_response.status_code in (301, 302, 303, 307, 308): + yield None + raise RateLimitedError( + "Instagram redirected the timeline query to " + f"{http_response.headers.get('location', '?')} — the session is being login-walled. " + + describe_response(http_response) + ) + + try: + payload = http_response.json() + except JSONDecodeError: + yield None + raise NetworkError("Timeline query did not return JSON. " + describe_response(http_response)) + + if http_response.status_code in (401, 403, 429) or payload.get("require_login"): + yield None + raise RateLimitedError("Instagram refused the timeline query. " + describe_response(http_response)) + + connection = (payload.get("data") or {}).get(PROFILE_POSTS_CONNECTION) + if not isinstance(connection, dict) or "edges" not in connection: + yield None + error_items = [e for e in payload.get("errors", []) if isinstance(e, dict)] + descriptions = "; ".join( + f"{e.get('message', '')}: {e.get('description', '')}".strip(": ") for e in error_items + ) + # Instagram answers a missing/renamed/deleted account with + # "execution error" + description "User lookup returned null". + if any("user lookup returned null" in str(e.get("description", "")).lower() for e in error_items): + raise IdentifierError( + f"Instagram has no account '{username}' (deleted, renamed or blocked). " + f"errors: {descriptions}. " + describe_response(http_response) + ) + raise NetworkError( + f"Timeline query returned no '{PROFILE_POSTS_CONNECTION}' (doc_id {doc_id} may be retired). " + f"errors: {descriptions or 'none'}. " + describe_response(http_response) + ) + + for edge in connection["edges"]: + node = edge.get("node") if isinstance(edge, dict) else None + if not node: + continue + if count and generated >= count: + return None + yield self.guest._Guest__process_post_data(node) + generated += 1 + + page_info = connection.get("page_info") or {} + if (count and generated >= count) or not page_info.get("has_next_page") or not page_info.get("end_cursor"): + return None + cursor = page_info["end_cursor"] def get_raw_post(self, share_url: str) -> str: share_url: str = share_url.strip() @@ -636,11 +739,7 @@ def get_raw_post(self, share_url: str) -> str: "accept-language": "en-US,en;q=0.9", "cache-control": "max-age=0", "sec-ch-prefers-color-scheme": "dark", - "sec-ch-ua": self.user_agent, - "sec-ch-ua-full-version-list": self.user_agent, - "sec-ch-ua-mobile": "?0", - "sec-ch-ua-platform": "\"Windows\"", - "sec-ch-ua-platform-version": "\"15.0.0\"", + **client_hints(), "sec-fetch-dest": "document", "sec-fetch-mode": "navigate", "sec-fetch-site": "same-origin", @@ -711,14 +810,9 @@ def private_info(self) -> PrivateInfo: "accept": "*/*", "dpr": "1.30208", "sec-ch-prefers-color-scheme": "dark", - "sec-ch-ua": self.user_agent, - "sec-ch-ua-full-version-list": self.user_agent, - "sec-ch-ua-mobile": "?0", - "sec-ch-ua-model": "\"\"", - "sec-ch-ua-platform": "\"Windows\"", - "sec-ch-ua-platform-version": "\"15.0.0\"", + **client_hints(), "viewport-width": "1475", - "x-asbd-id": "129477", + "x-asbd-id": ASBD_ID, "x-csrftoken": self.csrf_token, "x-ig-app-id": self.insta_app_id, "x-ig-www-claim": self.x_ig_www_claim, @@ -780,14 +874,9 @@ def change_bio(self, biography: str) -> bool: "content-type": "application/x-www-form-urlencoded", "dpr": "1.30208", "sec-ch-prefers-color-scheme": "dark", - "sec-ch-ua": self.user_agent, - "sec-ch-ua-full-version-list": self.user_agent, - "sec-ch-ua-mobile": "?0", - "sec-ch-ua-model": "\"\"", - "sec-ch-ua-platform": "\"Windows\"", - "sec-ch-ua-platform-version": "\"15.0.0\"", + **client_hints(), "viewport-width": "1475", - "x-asbd-id": "129477", + "x-asbd-id": ASBD_ID, "x-csrftoken": self.csrf_token, "x-ig-app-id": self.insta_app_id, "x-ig-www-claim": self.x_ig_www_claim, @@ -838,14 +927,9 @@ def change_display_name(self, display_name: str) -> bool: "content-type": "application/x-www-form-urlencoded", "dpr": "1.30208", "sec-ch-prefers-color-scheme": "dark", - "sec-ch-ua": self.user_agent, - "sec-ch-ua-full-version-list": self.user_agent, - "sec-ch-ua-mobile": "?0", - "sec-ch-ua-model": "\"\"", - "sec-ch-ua-platform": "\"Windows\"", - "sec-ch-ua-platform-version": "\"15.0.0\"", + **client_hints(), "viewport-width": "1475", - "x-asbd-id": "129477", + "x-asbd-id": ASBD_ID, "x-csrftoken": self.csrf_token, "x-ig-app-id": self.insta_app_id, "x-ig-www-claim": self.x_ig_www_claim, @@ -1037,17 +1121,12 @@ def pub_photo( "content-type": "application/x-www-form-urlencoded", "dpr": "1.30208", "sec-ch-prefers-color-scheme": "dark", - "sec-ch-ua": self.user_agent, - "sec-ch-ua-full-version-list": self.user_agent, - "sec-ch-ua-mobile": "?0", - "sec-ch-ua-model": "\"\"", - "sec-ch-ua-platform": "\"Windows\"", - "sec-ch-ua-platform-version": "\"15.0.0\"", + **client_hints(), "sec-fetch-dest": "empty", "sec-fetch-mode": "cors", "sec-fetch-site": "same-origin", "viewport-width": "1475", - "x-asbd-id": "129477", + "x-asbd-id": ASBD_ID, "x-csrftoken": self.csrf_token, "x-ig-app-id": self.insta_app_id, "x-ig-www-claim": self.x_ig_www_claim, @@ -1111,17 +1190,12 @@ def pub_carousel( "content-type": "application/x-www-form-urlencoded", "dpr": "1.30208", "sec-ch-prefers-color-scheme": "dark", - "sec-ch-ua": self.user_agent, - "sec-ch-ua-full-version-list": self.user_agent, - "sec-ch-ua-mobile": "?0", - "sec-ch-ua-model": "\"\"", - "sec-ch-ua-platform": "\"Windows\"", - "sec-ch-ua-platform-version": "\"15.0.0\"", + **client_hints(), "sec-fetch-dest": "empty", "sec-fetch-mode": "cors", "sec-fetch-site": "same-origin", "viewport-width": "1475", - "x-asbd-id": "129477", + "x-asbd-id": ASBD_ID, "x-csrftoken": self.csrf_token, "x-ig-app-id": self.insta_app_id, "x-ig-www-claim": self.x_ig_www_claim, @@ -1183,17 +1257,12 @@ def pub_reel( "content-type": "application/x-www-form-urlencoded", "dpr": "1.30208", "sec-ch-prefers-color-scheme": "dark", - "sec-ch-ua": self.user_agent, - "sec-ch-ua-full-version-list": self.user_agent, - "sec-ch-ua-mobile": "?0", - "sec-ch-ua-model": "\"\"", - "sec-ch-ua-platform": "\"Windows\"", - "sec-ch-ua-platform-version": "\"15.0.0\"", + **client_hints(), "sec-fetch-dest": "empty", "sec-fetch-mode": "cors", "sec-fetch-site": "same-origin", "viewport-width": "1475", - "x-asbd-id": "129477", + "x-asbd-id": ASBD_ID, "x-csrftoken": self.csrf_token, "x-ig-app-id": self.insta_app_id, "x-ig-www-claim": self.x_ig_www_claim, @@ -1255,7 +1324,7 @@ def comment(self, text: str, post_id: str) -> bool: "sec-fetch-mode": "cors", "sec-fetch-site": "same-origin", "viewport-width": "1475", - "x-asbd-id": "129477", + "x-asbd-id": ASBD_ID, "x-csrftoken": self.csrf_token, "x-ig-app-id": self.insta_app_id, "x-ig-www-claim": self.x_ig_www_claim, @@ -1297,7 +1366,7 @@ def __like_action(self, post_id: str, action: str = "like") -> bool: "sec-fetch-mode": "cors", "sec-fetch-site": "same-origin", "viewport-width": "1475", - "x-asbd-id": "129477", + "x-asbd-id": ASBD_ID, "x-csrftoken": self.csrf_token, "x-ig-app-id": self.insta_app_id, "x-ig-www-claim": self.x_ig_www_claim, @@ -1361,7 +1430,7 @@ def likers(self, post_id: str) -> Likers | None: "sec-fetch-mode": "cors", "sec-fetch-site": "same-origin", "viewport-width": "1475", - "x-asbd-id": "129477", + "x-asbd-id": ASBD_ID, "x-csrftoken": self.csrf_token, "x-ig-app-id": self.insta_app_id, "x-ig-www-claim": self.x_ig_www_claim, diff --git a/ensta/lib/WebHeaders.py b/ensta/lib/WebHeaders.py new file mode 100644 index 0000000..708be6b --- /dev/null +++ b/ensta/lib/WebHeaders.py @@ -0,0 +1,95 @@ +"""Browser identity headers shared by the web (www.instagram.com) clients. + +Instagram validates that the ``user-agent`` and the ``sec-ch-ua*`` client +hints describe the same browser. The values below must therefore be kept in +sync with each other; bump ``CHROME_MAJOR`` to move to a newer Chrome. +""" + +CHROME_MAJOR: str = "140" + +USER_AGENT: str = ( + "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 " + f"(KHTML, like Gecko) Chrome/{CHROME_MAJOR}.0.0.0 Safari/537.36" +) + +# Instagram's public web app id. Unchanged for years, so it is safe to embed. +IG_APP_ID: str = "936619743392459" + +# Current value sent by the Instagram web app. +ASBD_ID: str = "198387" + + +# Persisted GraphQL query id for the profile timeline ("PolarisProfilePostsQuery"), +# as used by the Instagram web app for logged-in viewers. Instagram retires +# these occasionally (7898261790222653 stopped working in 2026); when it does, +# the request answers with an "execution error" payload instead of the timeline +# connection and the exception says so. +PROFILE_POSTS_DOC_ID: str = "34579740524958711" +PROFILE_POSTS_FRIENDLY_NAME: str = "PolarisProfilePostsQuery" +PROFILE_POSTS_CONNECTION: str = "xdt_api__v1__feed__user_timeline_graphql_connection" + + +def client_hints() -> dict[str, str]: + """Return the ``sec-ch-ua*`` client hint headers matching ``USER_AGENT``.""" + + brands = ( + f'"Chromium";v="{CHROME_MAJOR}", "Not=A?Brand";v="24", "Google Chrome";v="{CHROME_MAJOR}"' + ) + full_brands = ( + f'"Chromium";v="{CHROME_MAJOR}.0.0.0", "Not=A?Brand";v="24.0.0.0", ' + f'"Google Chrome";v="{CHROME_MAJOR}.0.0.0"' + ) + return { + "sec-ch-ua": brands, + "sec-ch-ua-full-version-list": full_brands, + "sec-ch-ua-mobile": "?0", + "sec-ch-ua-model": '""', + "sec-ch-ua-platform": '"macOS"', + "sec-ch-ua-platform-version": '"14.0.0"', + } + + +def api_headers(csrf_token: str, www_claim: str, referer: str) -> dict[str, str]: + """Return the header set the Instagram web app sends to ``/api/v1/`` endpoints.""" + + return { + "accept": "*/*", + "accept-language": "en-US,en;q=0.9", + "sec-ch-prefers-color-scheme": "dark", + **client_hints(), + "sec-fetch-dest": "empty", + "sec-fetch-mode": "cors", + "sec-fetch-site": "same-origin", + "viewport-width": "1475", + "x-asbd-id": ASBD_ID, + "x-csrftoken": csrf_token, + "x-ig-app-id": IG_APP_ID, + "x-ig-www-claim": www_claim, + "x-requested-with": "XMLHttpRequest", + "Referer": referer, + "Referrer-Policy": "strict-origin-when-cross-origin", + } + + +def csrf_token_from(session, fallback: str) -> str: + """Return the csrftoken Instagram set on this session, preferring the + instagram.com-scoped cookie over any placeholder set without a domain. + + ``RequestsCookieJar.get`` raises ``CookieConflictError`` once both exist, + which happens after the first real Instagram response.""" + + placeholder = None + for cookie in session.cookies: + if cookie.name != "csrftoken": + continue + if "instagram.com" in (cookie.domain or ""): + return cookie.value + placeholder = cookie.value + return placeholder or fallback + + +def describe_response(response) -> str: + """Short, loggable description of an unexpected Instagram response.""" + + body = response.text[:200].replace("\n", " ") + return f"HTTP {response.status_code} from {response.url}: {body!r}"