mirror of
https://github.com/instaloader/instaloader.git
synced 2024-11-04 09:22:29 +01:00
graphql_query() method for GraphQL queries
This commit is contained in:
Alexander Graf
parent
d614d9f015
commit
5068c9453e
198
instaloader.py
198
instaloader.py
@ -180,34 +180,49 @@ class Instaloader:
|
|||||||
session.headers.update(self.default_http_header(empty_session_only=True))
|
session.headers.update(self.default_http_header(empty_session_only=True))
|
||||||
return session
|
return session
|
||||||
|
|
||||||
|
def graphql_query(self, query_id: int, variables: Dict[str, Any],
|
||||||
|
referer: Optional[str] = None) -> Dict[str, Any]:
|
||||||
|
"""
|
||||||
|
Do a GraphQL Query.
|
||||||
|
|
||||||
|
:param query_id: Query ID.
|
||||||
|
:param variables: Variables for the Query.
|
||||||
|
:param referer: HTTP Referer, or None.
|
||||||
|
:return: The server's response dictionary.
|
||||||
|
"""
|
||||||
|
tmpsession = copy_session(self.session)
|
||||||
|
tmpsession.headers.update(self.default_http_header(empty_session_only=True))
|
||||||
|
del tmpsession.headers['Connection']
|
||||||
|
del tmpsession.headers['Content-Length']
|
||||||
|
tmpsession.headers['authority'] = 'www.instagram.com'
|
||||||
|
tmpsession.headers['scheme'] = 'https'
|
||||||
|
tmpsession.headers['accept'] = '*/*'
|
||||||
|
if referer is not None:
|
||||||
|
tmpsession.headers['referer'] = referer
|
||||||
|
response = tmpsession.get('https://www.instagram.com/graphql/query',
|
||||||
|
params={'query_id': query_id,
|
||||||
|
'variables': json.dumps(variables, separators=(',', ':'))})
|
||||||
|
if response.status_code != 200:
|
||||||
|
raise ConnectionException("GraphQL query returned HTTP error code {}.".format(response.status_code))
|
||||||
|
return response.json()
|
||||||
|
|
||||||
def get_username_by_id(self, profile_id: int) -> str:
|
def get_username_by_id(self, profile_id: int) -> str:
|
||||||
"""To get the current username of a profile, given its unique ID, this function can be used."""
|
"""To get the current username of a profile, given its unique ID, this function can be used."""
|
||||||
tmpsession = copy_session(self.session)
|
data = self.graphql_query(17862015703145017, {'id': str(profile_id), 'first': 1})['data']['user']
|
||||||
tmpsession.headers.update(self.default_http_header())
|
if data:
|
||||||
del tmpsession.headers['Referer']
|
data = data["edge_owner_to_timeline_media"]
|
||||||
del tmpsession.headers['X-Instagram-AJAX']
|
|
||||||
resp = tmpsession.get('https://www.instagram.com/graphql/query/',
|
|
||||||
params={'query_id': '17862015703145017',
|
|
||||||
'variables': '{"id":"' + str(profile_id) + '","first":1}'})
|
|
||||||
if resp.status_code == 200:
|
|
||||||
data = json.loads(resp.text)["data"]["user"]
|
|
||||||
if data:
|
|
||||||
data = data["edge_owner_to_timeline_media"]
|
|
||||||
else:
|
|
||||||
raise ProfileNotExistsException("No profile found, the user may have blocked you (ID: " +
|
|
||||||
str(profile_id) + ").")
|
|
||||||
if not data['edges']:
|
|
||||||
if data['count'] == 0:
|
|
||||||
raise ProfileHasNoPicsException("Profile with ID {0}: no pics found.".format(str(profile_id)))
|
|
||||||
else:
|
|
||||||
raise LoginRequiredException("Login required to determine username (ID: " + str(profile_id) + ").")
|
|
||||||
else:
|
|
||||||
shortcode = mediaid_to_shortcode(int(data['edges'][0]["node"]["id"]))
|
|
||||||
data = self.get_json("p/" + shortcode)
|
|
||||||
return data['entry_data']['PostPage'][0]['graphql']['shortcode_media']['owner']['username']
|
|
||||||
else:
|
else:
|
||||||
raise ProfileAccessDeniedException("Username could not be determined due to error {0} (ID: {1})."
|
raise ProfileNotExistsException("No profile found, the user may have blocked you (ID: " +
|
||||||
.format(str(resp.status_code), str(profile_id)))
|
str(profile_id) + ").")
|
||||||
|
if not data['edges']:
|
||||||
|
if data['count'] == 0:
|
||||||
|
raise ProfileHasNoPicsException("Profile with ID {0}: no pics found.".format(str(profile_id)))
|
||||||
|
else:
|
||||||
|
raise LoginRequiredException("Login required to determine username (ID: " + str(profile_id) + ").")
|
||||||
|
else:
|
||||||
|
shortcode = mediaid_to_shortcode(int(data['edges'][0]["node"]["id"]))
|
||||||
|
data = self.get_json("p/" + shortcode)
|
||||||
|
return data['entry_data']['PostPage'][0]['graphql']['shortcode_media']['owner']['username']
|
||||||
|
|
||||||
def get_id_by_username(self, profile: str) -> int:
|
def get_id_by_username(self, profile: str) -> int:
|
||||||
"""Each Instagram profile has its own unique ID which stays unmodified even if a user changes
|
"""Each Instagram profile has its own unique ID which stays unmodified even if a user changes
|
||||||
@ -226,36 +241,22 @@ class Instaloader:
|
|||||||
:param profile: Name of profile to lookup followers.
|
:param profile: Name of profile to lookup followers.
|
||||||
:return: List of followers (list of dictionaries).
|
:return: List of followers (list of dictionaries).
|
||||||
"""
|
"""
|
||||||
tmpsession = copy_session(self.session)
|
|
||||||
header = self.default_http_header(empty_session_only=True)
|
|
||||||
del header['Connection']
|
|
||||||
del header['Content-Length']
|
|
||||||
header['authority'] = 'www.instagram.com'
|
|
||||||
header['scheme'] = 'https'
|
|
||||||
header['accept'] = '*/*'
|
|
||||||
header['referer'] = 'https://www.instagram.com/' + profile + '/'
|
|
||||||
tmpsession.headers = header
|
|
||||||
profile_id = self.get_id_by_username(profile)
|
profile_id = self.get_id_by_username(profile)
|
||||||
resp = tmpsession.get('https://www.instagram.com/graphql/query/',
|
data = self.graphql_query(17851374694183129, {'id': str(profile_id),
|
||||||
params={'query_id': 17851374694183129,
|
'first': 500},
|
||||||
'variables': '{"id":"' + str(profile_id) + '","first":500}'})
|
referer='https://www.instagram.com/' + profile + '/')
|
||||||
if resp.status_code == 200:
|
followers = []
|
||||||
data = resp.json()
|
while True:
|
||||||
followers = []
|
edge_followed_by = data['data']['user']['edge_followed_by']
|
||||||
while True:
|
followers.extend([follower['node'] for follower in edge_followed_by['edges']])
|
||||||
edge_followed_by = data['data']['user']['edge_followed_by']
|
page_info = edge_followed_by['page_info']
|
||||||
followers.extend([follower['node'] for follower in edge_followed_by['edges']])
|
if page_info['has_next_page']:
|
||||||
page_info = edge_followed_by['page_info']
|
data = self.graphql_query(17851374694183129, {'id': str(profile_id),
|
||||||
if page_info['has_next_page']:
|
'first': 500,
|
||||||
resp = tmpsession.get('https://www.instagram.com/graphql/query/',
|
'after': page_info['end_cursor']},
|
||||||
params={'query_id': 17851374694183129,
|
referer='https://www.instagram.com/' + profile + '/')
|
||||||
'variables': '{"id":"' + str(profile_id) + '","first":500,"after":"' +
|
else:
|
||||||
page_info['end_cursor'] + '"}'})
|
break
|
||||||
data = resp.json()
|
|
||||||
else:
|
|
||||||
break
|
|
||||||
else:
|
|
||||||
raise ConnectionException("ConnectionError({0}): unable to gather followers.".format(resp.status_code))
|
|
||||||
return followers
|
return followers
|
||||||
|
|
||||||
def get_followees(self, profile: str) -> List[Dict[str, Any]]:
|
def get_followees(self, profile: str) -> List[Dict[str, Any]]:
|
||||||
@ -267,68 +268,41 @@ class Instaloader:
|
|||||||
:param profile: Name of profile to lookup followers.
|
:param profile: Name of profile to lookup followers.
|
||||||
:return: List of followees (list of dictionaries).
|
:return: List of followees (list of dictionaries).
|
||||||
"""
|
"""
|
||||||
tmpsession = copy_session(self.session)
|
|
||||||
header = self.default_http_header(empty_session_only=True)
|
|
||||||
del header['Connection']
|
|
||||||
del header['Content-Length']
|
|
||||||
header['authority'] = 'www.instagram.com'
|
|
||||||
header['scheme'] = 'https'
|
|
||||||
header['accept'] = '*/*'
|
|
||||||
header['referer'] = 'https://www.instagram.com/' + profile + '/'
|
|
||||||
tmpsession.headers = header
|
|
||||||
profile_id = self.get_id_by_username(profile)
|
profile_id = self.get_id_by_username(profile)
|
||||||
resp = tmpsession.get('https://www.instagram.com/graphql/query/',
|
data = self.graphql_query(17874545323001329, {'id': profile_id,
|
||||||
params={'query_id': 17874545323001329,
|
'first': 500},
|
||||||
'variables': '{"id":"' + str(profile_id) + '","first":500}'})
|
referer='https://www.instagram.com/' + profile + '/')
|
||||||
if resp.status_code == 200:
|
followees = []
|
||||||
data = resp.json()
|
while True:
|
||||||
followees = []
|
edge_follow = data['data']['user']['edge_follow']
|
||||||
while True:
|
followees.extend([followee['node'] for followee in edge_follow['edges']])
|
||||||
edge_follow = data['data']['user']['edge_follow']
|
page_info = edge_follow['page_info']
|
||||||
followees.extend([followee['node'] for followee in edge_follow['edges']])
|
if page_info['has_next_page']:
|
||||||
page_info = edge_follow['page_info']
|
data = self.graphql_query(17874545323001329, {'id': profile_id,
|
||||||
if page_info['has_next_page']:
|
'first': 500,
|
||||||
resp = tmpsession.get('https://www.instagram.com/graphql/query/',
|
'after': page_info['end_cursor']},
|
||||||
params={'query_id': 17874545323001329,
|
referer='https://www.instagram.com/' + profile + '/')
|
||||||
'variables': '{"id":"' + str(profile_id) + '","first":500,"after":"' +
|
else:
|
||||||
page_info['end_cursor'] + '"}'})
|
break
|
||||||
data = resp.json()
|
|
||||||
else:
|
|
||||||
break
|
|
||||||
else:
|
|
||||||
raise ConnectionException("ConnectionError({0}): unable to gather followees.".format(resp.status_code))
|
|
||||||
return followees
|
return followees
|
||||||
|
|
||||||
def get_comments(self, shortcode: str) -> List[Dict[str, Any]]:
|
def get_comments(self, shortcode: str) -> List[Dict[str, Any]]:
|
||||||
tmpsession = copy_session(self.session)
|
"""Retrieve comments of node with given shortcode."""
|
||||||
header = self.default_http_header(empty_session_only=True)
|
data = self.graphql_query(17852405266163336, {'shortcode': shortcode,
|
||||||
del header['Connection']
|
'first': 500},
|
||||||
del header['Content-Length']
|
referer='https://www.instagram.com/p/' + shortcode + '/')
|
||||||
header['authority'] = 'www.instagram.com'
|
comments = []
|
||||||
header['scheme'] = 'https'
|
while True:
|
||||||
header['accept'] = '*/*'
|
edge_media_to_comment = data['data']['shortcode_media']['edge_media_to_comment']
|
||||||
header['referer'] = 'https://www.instagram.com/p/' + shortcode + '/'
|
comments.extend([comment['node'] for comment in edge_media_to_comment['edges']])
|
||||||
tmpsession.headers = header
|
page_info = edge_media_to_comment['page_info']
|
||||||
resp = tmpsession.get('https://www.instagram.com/graphql/query/',
|
if page_info['has_next_page']:
|
||||||
params={'query_id': 17852405266163336,
|
data = self.graphql_query(17852405266163336, {'shortcode': shortcode,
|
||||||
'variables': '{"shortcode":"' + shortcode + '","first":500}'})
|
'first': 500,
|
||||||
if resp.status_code == 200:
|
'after': page_info['end_cursor']},
|
||||||
data = resp.json()
|
referer='https://www.instagram.com/p/' + shortcode + '/')
|
||||||
comments = []
|
else:
|
||||||
while True:
|
break
|
||||||
edge_media_to_comment = data['data']['shortcode_media']['edge_media_to_comment']
|
|
||||||
comments.extend([comment['node'] for comment in edge_media_to_comment['edges']])
|
|
||||||
page_info = edge_media_to_comment['page_info']
|
|
||||||
if page_info['has_next_page']:
|
|
||||||
resp = tmpsession.get('https://www.instagram.com/graphql/query/',
|
|
||||||
params={'query_id': 17852405266163336,
|
|
||||||
'variables': '{"shortcode":"' + shortcode + '","first":500,"after":"'
|
|
||||||
+ page_info['end_cursor'] + '"}'})
|
|
||||||
data = resp.json()
|
|
||||||
else:
|
|
||||||
break
|
|
||||||
else:
|
|
||||||
raise ConnectionException("ConnectionError({0}): unable to gather comments.".format(resp.status_code))
|
|
||||||
return comments
|
return comments
|
||||||
|
|
||||||
def download_pic(self, name: str, url: str, date_epoch: float, outputlabel: Optional[str] = None,
|
def download_pic(self, name: str, url: str, date_epoch: float, outputlabel: Optional[str] = None,
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user