erjemin/instaloader
erjemin/instaloader
1
0
Fork 0
Code Issues Wiki Activity

Implement high quality profile pic hack

Browse Source 
Use 'i.instagram.com' API to retrieve the best quality version of the
profile pic.
Related to #85.
This commit is contained in:
André Koch-Kramer 2018-03-29 19:33:05 +02:00
parent 5b6e8f1a31
commit d9b5c77e3a
1 changed files with 15 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
instaloader.py
View File

@ -588,12 +588,13 @@ class Instaloader:
self.error("[skipped by user]", repeat_at_end=False) self.error("[skipped by user]", repeat_at_end=False)
raise ConnectionException(error_string) raise ConnectionException(error_string)
def get_json(self, url: str, params: Dict[str, Any], def get_json(self, path: str, params: Dict[str, Any], host: str = 'www.instagram.com',
session: Optional[requests.Session] = None, _attempt = 1) -> Dict[str, Any]: session: Optional[requests.Session] = None, _attempt = 1) -> Dict[str, Any]:
"""JSON request to Instagram. """JSON request to Instagram.
:param url: URL, relative to www.instagram.com/ :param path: URL, relative to the given domain which defaults to www.instagram.com/
:param params: GET parameters :param params: GET parameters
:param host: Domain part of the URL from where to download the requested JSON; defaults to www.instagram.com
:param session: Session to use, or None to use self.session :param session: Session to use, or None to use self.session
:return: Decoded response dictionary :return: Decoded response dictionary
:raises QueryReturnedNotFoundException: When the server responds with a 404. :raises QueryReturnedNotFoundException: When the server responds with a 404.
@ -610,7 +611,7 @@ class Instaloader:
if len(timestamps) < 100 and not untracked_queries: if len(timestamps) < 100 and not untracked_queries:
return 0 return 0
return round(min(timestamps) + sliding_window - current_time) + 6 return round(min(timestamps) + sliding_window - current_time) + 6
is_graphql_query = 'query_id' in params and 'graphql/query' in url is_graphql_query = 'query_id' in params and 'graphql/query' in path
if is_graphql_query: if is_graphql_query:
query_id = params['query_id'] query_id = params['query_id']
waittime = graphql_query_waittime(query_id) waittime = graphql_query_waittime(query_id)
@ -625,7 +626,7 @@ class Instaloader:
sess = session if session else self.session sess = session if session else self.session
try: try:
self._sleep() self._sleep()
resp = sess.get('https://www.instagram.com/' + url, params=params) resp = sess.get('https://{0}/{1}'.format(host, path), params=params)
if resp.status_code == 404: if resp.status_code == 404:
raise QueryReturnedNotFoundException("404") raise QueryReturnedNotFoundException("404")
if resp.status_code == 429: if resp.status_code == 429:
@ -641,7 +642,7 @@ class Instaloader:
raise ConnectionException("Returned \"{}\" status.".format(resp_json['status'])) raise ConnectionException("Returned \"{}\" status.".format(resp_json['status']))
return resp_json return resp_json
except (ConnectionException, json.decoder.JSONDecodeError, requests.exceptions.RequestException) as err: except (ConnectionException, json.decoder.JSONDecodeError, requests.exceptions.RequestException) as err:
error_string = "JSON Query to {}: {}".format(url, err) error_string = "JSON Query to {}: {}".format(path, err)
if _attempt == self.max_connection_attempts: if _attempt == self.max_connection_attempts:
raise ConnectionException(error_string) raise ConnectionException(error_string)
self.error(error_string + " [retrying; skip with ^C]", repeat_at_end=False) self.error(error_string + " [retrying; skip with ^C]", repeat_at_end=False)
@ -657,7 +658,7 @@ class Instaloader:
self._log('The request will be retried in {} seconds.'.format(waittime)) self._log('The request will be retried in {} seconds.'.format(waittime))
time.sleep(waittime) time.sleep(waittime)
self._sleep() self._sleep()
return self.get_json(url, params, sess, _attempt + 1) return self.get_json(path=path, params=params, host=host, session=sess, _attempt=_attempt + 1)
except KeyboardInterrupt: except KeyboardInterrupt:
self.error("[skipped by user]", repeat_at_end=False) self.error("[skipped by user]", repeat_at_end=False)
raise ConnectionException(error_string) raise ConnectionException(error_string)
@ -900,8 +901,14 @@ class Instaloader:
def download_profilepic(self, name: str, profile_metadata: Dict[str, Any]) -> None: def download_profilepic(self, name: str, profile_metadata: Dict[str, Any]) -> None:
"""Downloads and saves profile pic.""" """Downloads and saves profile pic."""
url = profile_metadata["user"]["profile_pic_url_hd"] if "profile_pic_url_hd" in profile_metadata["user"] \ try:
else profile_metadata["user"]["profile_pic_url"] data = self.get_json(path='api/v1/users/{0}/info/'.format(profile_metadata["user"]["id"]), params={},
host='i.instagram.com')
url = data["user"]["hd_profile_pic_url_info"]["url"]
except (InstaloaderException, KeyError) as err:
self.error('{} Unable to fetch high quality profile pic.'.format(err))
url = profile_metadata["user"]["profile_pic_url_hd"] if "profile_pic_url_hd" in profile_metadata["user"] \
else profile_metadata["user"]["profile_pic_url"]
def _epoch_to_string(epoch: datetime) -> str: def _epoch_to_string(epoch: datetime) -> str:
return epoch.strftime('%Y-%m-%d_%H-%M-%S') return epoch.strftime('%Y-%m-%d_%H-%M-%S')