Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
.. versionadded:: 4.4
"""
for hashtag in self._node.get('hashtags', []):
node = hashtag.get('hashtag', {})
if 'name' in node:
yield Hashtag(self._context, node)
@property
def searchstring(self) -> str:
"""
The string that was searched for on Instagram to produce this :class:`TopSearchResults` instance.
"""
return self._searchstring
JsonExportable = Union[Post, Profile, StoryItem, Hashtag]
def save_structure_to_file(structure: JsonExportable, filename: str) -> None:
"""Saves a :class:`Post`, :class:`Profile`, :class:`StoryItem` or :class:`Hashtag` to a '.json' or '.json.xz' file
such that it can later be loaded by :func:`load_structure_from_file`.
If the specified filename ends in '.xz', the file will be LZMA compressed. Otherwise, a pretty-printed JSON file
will be created.
:param structure: :class:`Post`, :class:`Profile`, :class:`StoryItem` or :class:`Hashtag`
:param filename: Filename, ends in '.json' or '.json.xz'
"""
json_structure = {'node': structure._asdict(),
'instaloader': {'version': __version__, 'node_type': structure.__class__.__name__}}
compress = filename.endswith('.xz')
if compress:
:param context: :attr:`Instaloader.context` linked to the new object, used for additional queries if neccessary.
:param filename: Filename, ends in '.json' or '.json.xz'
"""
compressed = filename.endswith('.xz')
if compressed:
fp = lzma.open(filename, 'rt')
else:
fp = open(filename, 'rt')
json_structure = json.load(fp)
fp.close()
if 'node' in json_structure and 'instaloader' in json_structure and \
'node_type' in json_structure['instaloader']:
node_type = json_structure['instaloader']['node_type']
if node_type == "Post":
return Post(context, json_structure['node'])
elif node_type == "Profile":
return Profile(context, json_structure['node'])
elif node_type == "StoryItem":
return StoryItem(context, json_structure['node'])
elif node_type == "Hashtag":
return Hashtag(context, json_structure['node'])
else:
raise InvalidArgumentException("{}: Not an Instaloader JSON.".format(filename))
elif 'shortcode' in json_structure:
# Post JSON created with Instaloader v3
return Post.from_shortcode(context, json_structure['shortcode'])
else:
raise InvalidArgumentException("{}: Not an Instaloader JSON.".format(filename))
def save_metadata_json(self, filename: str, structure: JsonExportable) -> None:
"""Saves metadata JSON file of a structure."""
if self.compress_json:
filename += '.json.xz'
else:
filename += '.json'
os.makedirs(os.path.dirname(filename), exist_ok=True)
save_structure_to_file(structure, filename)
if isinstance(structure, (Post, StoryItem)):
# log 'json ' message when saving Post or StoryItem
self.context.log('json', end=' ', flush=True)
def get_igtv_posts(self) -> Iterator[Post]:
"""Retrieve all IGTV posts.
.. versionadded:: 4.3"""
self._obtain_metadata()
yield from (Post(self._context, node, self) for node in
self._context.graphql_node_list('bc78b344a68ed16dd5d7f264681c4c76',
{'id': self.userid},
'https://www.instagram.com/{0}/channel/'.format(self.username),
lambda d: d['data']['user']['edge_felix_video_timeline'],
first_data=self._metadata('edge_felix_video_timeline')))
def get_feed_posts(self) -> Iterator[Post]:
"""Get Posts of the user's feed.
:return: Iterator over Posts of the user's feed.
:raises LoginRequiredException: If called without being logged in.
"""
data = self.context.graphql_query("d6f4427fbe92d846298cf93df0b937d3", {})["data"]
while True:
feed = data["user"]["edge_web_feed_timeline"]
for edge in feed["edges"]:
node = edge["node"]
if node.get("__typename") in Post.supported_graphql_types() and node.get("shortcode") is not None:
yield Post(self.context, node)
if not feed["page_info"]["has_next_page"]:
break
data = self.context.graphql_query("d6f4427fbe92d846298cf93df0b937d3",
{'fetch_media_item_count': 12,
'fetch_media_item_cursor': feed["page_info"]["end_cursor"],
'fetch_comment_count': 4,
'fetch_like': 10,
'has_stories': False})["data"]
.. versionadded:: 4.2
.. versionchanged:: 4.2.9
Require being logged in (as required by Instagram)
"""
has_next_page = True
end_cursor = None
while has_next_page:
if end_cursor:
params = {'__a': 1, 'max_id': end_cursor}
else:
params = {'__a': 1}
location_data = self.context.get_json('explore/locations/{0}/'.format(location),
params)['graphql']['location']['edge_location_to_media']
yield from (Post(self.context, edge['node']) for edge in location_data['edges'])
has_next_page = location_data['page_info']['has_next_page']
end_cursor = location_data['page_info']['end_cursor']
def __eq__(self, o: object) -> bool:
if isinstance(o, Post):
return self.shortcode == o.shortcode
return NotImplemented
def get_posts(self) -> Iterator[Post]:
"""Yields the posts associated with this hashtag."""
self._metadata("edge_hashtag_to_media", "edges")
self._metadata("edge_hashtag_to_media", "page_info")
conn = self._metadata("edge_hashtag_to_media")
yield from (Post(self._context, edge["node"]) for edge in conn["edges"])
while conn["page_info"]["has_next_page"]:
data = self._query({'__a': 1, 'max_id': conn["page_info"]["end_cursor"]})
conn = data["edge_hashtag_to_media"]
yield from (Post(self._context, edge["node"]) for edge in conn["edges"])
if 'node' in json_structure and 'instaloader' in json_structure and \
'node_type' in json_structure['instaloader']:
node_type = json_structure['instaloader']['node_type']
if node_type == "Post":
return Post(context, json_structure['node'])
elif node_type == "Profile":
return Profile(context, json_structure['node'])
elif node_type == "StoryItem":
return StoryItem(context, json_structure['node'])
elif node_type == "Hashtag":
return Hashtag(context, json_structure['node'])
else:
raise InvalidArgumentException("{}: Not an Instaloader JSON.".format(filename))
elif 'shortcode' in json_structure:
# Post JSON created with Instaloader v3
return Post.from_shortcode(context, json_structure['shortcode'])
else:
raise InvalidArgumentException("{}: Not an Instaloader JSON.".format(filename))