namedtuples PostLocation and PostSidecarNode
Post.get_sidecar_edges() has been renamed to Post.get_sidecar_nodes(). Post.get_location() has been made to a property Post.location and it now internally stores location struct and makes it JSON store-/loadable.
This commit is contained in:
@@ -11,11 +11,11 @@ from contextlib import contextmanager, suppress
|
|||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from functools import wraps
|
from functools import wraps
|
||||||
from io import BytesIO
|
from io import BytesIO
|
||||||
from typing import Callable, Dict, Iterator, List, Optional, Any
|
from typing import Callable, Iterator, List, Optional, Any
|
||||||
|
|
||||||
from .exceptions import *
|
from .exceptions import *
|
||||||
from .instaloadercontext import InstaloaderContext
|
from .instaloadercontext import InstaloaderContext
|
||||||
from .structures import JsonExportable, Post, Profile, Story, StoryItem, save_structure_to_file
|
from .structures import JsonExportable, Post, PostLocation, Profile, Story, StoryItem, save_structure_to_file
|
||||||
|
|
||||||
|
|
||||||
def get_default_session_filename(username: str) -> str:
|
def get_default_session_filename(username: str) -> str:
|
||||||
@@ -244,12 +244,12 @@ class Instaloader:
|
|||||||
shutil.copyfileobj(BytesIO(caption), text_file)
|
shutil.copyfileobj(BytesIO(caption), text_file)
|
||||||
os.utime(filename, (datetime.now().timestamp(), mtime.timestamp()))
|
os.utime(filename, (datetime.now().timestamp(), mtime.timestamp()))
|
||||||
|
|
||||||
def save_location(self, filename: str, location_json: Dict[str, str], mtime: datetime) -> None:
|
def save_location(self, filename: str, location: PostLocation, mtime: datetime) -> None:
|
||||||
"""Save post location name and Google Maps link."""
|
"""Save post location name and Google Maps link."""
|
||||||
filename += '_location.txt'
|
filename += '_location.txt'
|
||||||
location_string = (location_json["name"] + "\n" +
|
location_string = (location.name + "\n" +
|
||||||
"https://maps.google.com/maps?q={0},{1}&ll={0},{1}\n".format(location_json["lat"],
|
"https://maps.google.com/maps?q={0},{1}&ll={0},{1}\n".format(location.lat,
|
||||||
location_json["lng"]))
|
location.lng))
|
||||||
with open(filename, 'wb') as text_file:
|
with open(filename, 'wb') as text_file:
|
||||||
shutil.copyfileobj(BytesIO(location_string.encode()), text_file)
|
shutil.copyfileobj(BytesIO(location_string.encode()), text_file)
|
||||||
os.utime(filename, (datetime.now().timestamp(), mtime.timestamp()))
|
os.utime(filename, (datetime.now().timestamp(), mtime.timestamp()))
|
||||||
@@ -335,14 +335,14 @@ class Instaloader:
|
|||||||
downloaded = False
|
downloaded = False
|
||||||
if post.typename == 'GraphSidecar':
|
if post.typename == 'GraphSidecar':
|
||||||
edge_number = 1
|
edge_number = 1
|
||||||
for edge in post.get_sidecar_edges():
|
for sidecar_node in post.get_sidecar_nodes():
|
||||||
# Download picture or video thumbnail
|
# Download picture or video thumbnail
|
||||||
if not edge['node']['is_video'] or self.download_video_thumbnails is True:
|
if not sidecar_node.is_video or self.download_video_thumbnails is True:
|
||||||
downloaded |= self.download_pic(filename=filename, url=edge['node']['display_url'],
|
downloaded |= self.download_pic(filename=filename, url=sidecar_node.display_url,
|
||||||
mtime=post.date_local, filename_suffix=str(edge_number))
|
mtime=post.date_local, filename_suffix=str(edge_number))
|
||||||
# Additionally download video if available and desired
|
# Additionally download video if available and desired
|
||||||
if edge['node']['is_video'] and self.download_videos is True:
|
if sidecar_node.is_video and self.download_videos is True:
|
||||||
downloaded |= self.download_pic(filename=filename, url=edge['node']['video_url'],
|
downloaded |= self.download_pic(filename=filename, url=sidecar_node.video_url,
|
||||||
mtime=post.date_local, filename_suffix=str(edge_number))
|
mtime=post.date_local, filename_suffix=str(edge_number))
|
||||||
edge_number += 1
|
edge_number += 1
|
||||||
elif post.typename == 'GraphImage':
|
elif post.typename == 'GraphImage':
|
||||||
@@ -366,10 +366,8 @@ class Instaloader:
|
|||||||
downloaded |= self.download_pic(filename=filename, url=post.video_url, mtime=post.date_local)
|
downloaded |= self.download_pic(filename=filename, url=post.video_url, mtime=post.date_local)
|
||||||
|
|
||||||
# Download geotags if desired
|
# Download geotags if desired
|
||||||
if self.download_geotags is True:
|
if self.download_geotags and post.location:
|
||||||
location = post.get_location()
|
self.save_location(filename, post.location, post.date_local)
|
||||||
if location:
|
|
||||||
self.save_location(filename, location, post.date_local)
|
|
||||||
|
|
||||||
# Update comments if desired
|
# Update comments if desired
|
||||||
if self.download_comments is True:
|
if self.download_comments is True:
|
||||||
|
@@ -2,6 +2,7 @@ import json
|
|||||||
import lzma
|
import lzma
|
||||||
import re
|
import re
|
||||||
from base64 import b64decode, b64encode
|
from base64 import b64decode, b64encode
|
||||||
|
from collections import namedtuple
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from typing import Any, Dict, Iterator, List, Optional, Union
|
from typing import Any, Dict, Iterator, List, Optional, Union
|
||||||
|
|
||||||
@@ -23,6 +24,10 @@ def mediaid_to_shortcode(mediaid: int) -> str:
|
|||||||
return b64encode(mediaid.to_bytes(9, 'big'), b'-_').decode().replace('A', ' ').lstrip().replace(' ', 'A')
|
return b64encode(mediaid.to_bytes(9, 'big'), b'-_').decode().replace('A', ' ').lstrip().replace(' ', 'A')
|
||||||
|
|
||||||
|
|
||||||
|
PostSidecarNode = namedtuple('PostSidecarNode', ['is_video', 'display_url', 'video_url'])
|
||||||
|
PostLocation = namedtuple('PostLocation', ['id', 'name', 'slug', 'has_public_page', 'lat', 'lng'])
|
||||||
|
|
||||||
|
|
||||||
class Post:
|
class Post:
|
||||||
"""
|
"""
|
||||||
Structure containing information about an Instagram post.
|
Structure containing information about an Instagram post.
|
||||||
@@ -55,6 +60,7 @@ class Post:
|
|||||||
self._owner_profile = owner_profile
|
self._owner_profile = owner_profile
|
||||||
self._full_metadata_dict = None
|
self._full_metadata_dict = None
|
||||||
self._rhx_gis_str = None
|
self._rhx_gis_str = None
|
||||||
|
self._location = None
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def from_shortcode(cls, context: InstaloaderContext, shortcode: str):
|
def from_shortcode(cls, context: InstaloaderContext, shortcode: str):
|
||||||
@@ -76,6 +82,8 @@ class Post:
|
|||||||
node = self._node
|
node = self._node
|
||||||
if self._owner_profile:
|
if self._owner_profile:
|
||||||
node['owner'] = self.owner_profile.get_node()
|
node['owner'] = self.owner_profile.get_node()
|
||||||
|
if self._location:
|
||||||
|
node['location'] = self._location._asdict()
|
||||||
return node
|
return node
|
||||||
|
|
||||||
@property
|
@property
|
||||||
@@ -189,8 +197,14 @@ class Post:
|
|||||||
# if __typename is not in node, it is an old image or video
|
# if __typename is not in node, it is an old image or video
|
||||||
return 'GraphImage'
|
return 'GraphImage'
|
||||||
|
|
||||||
def get_sidecar_edges(self) -> List[Dict[str, Any]]:
|
def get_sidecar_nodes(self) -> Iterator[PostSidecarNode]:
|
||||||
return self._field('edge_sidecar_to_children', 'edges')
|
"""Sidecar nodes of a Post with typename==GraphSidecar."""
|
||||||
|
if self.typename == 'GraphSidecar':
|
||||||
|
for edge in self._field('edge_sidecar_to_children', 'edges'):
|
||||||
|
node = edge['node']
|
||||||
|
is_video = node['is_video']
|
||||||
|
yield PostSidecarNode(is_video=is_video, display_url=node['display_url'],
|
||||||
|
video_url=node['video_url'] if is_video else None)
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def caption(self) -> Optional[str]:
|
def caption(self) -> Optional[str]:
|
||||||
@@ -298,13 +312,19 @@ class Post:
|
|||||||
lambda d: d['data']['shortcode_media']['edge_liked_by'],
|
lambda d: d['data']['shortcode_media']['edge_liked_by'],
|
||||||
self._rhx_gis)
|
self._rhx_gis)
|
||||||
|
|
||||||
def get_location(self) -> Optional[Dict[str, str]]:
|
@property
|
||||||
"""If the Post has a location, returns a dictionary with fields 'lat' and 'lng' and 'name'."""
|
def location(self) -> Optional[PostLocation]:
|
||||||
loc_dict = self._field("location")
|
"""If the Post has a location, returns PostLocation namedtuple with fields 'id', 'lat' and 'lng' and 'name'."""
|
||||||
if loc_dict is not None:
|
loc = self._field("location")
|
||||||
location_json = self._context.get_json("explore/locations/{0}/".format(loc_dict["id"]),
|
if self._location or not loc:
|
||||||
params={'__a': 1})
|
return self._location
|
||||||
return location_json["location"] if "location" in location_json else location_json['graphql']['location']
|
location_id = int(loc['id'])
|
||||||
|
if any(k not in loc for k in ('name', 'slug', 'has_public_page', 'lat', 'lng')):
|
||||||
|
loc = self._context.get_json("explore/locations/{0}/".format(location_id),
|
||||||
|
params={'__a': 1})['graphql']['location']
|
||||||
|
self._location = PostLocation(location_id, loc['name'], loc['slug'], loc['has_public_page'],
|
||||||
|
loc['lat'], loc['lng'])
|
||||||
|
return self._location
|
||||||
|
|
||||||
|
|
||||||
class Profile:
|
class Profile:
|
||||||
|
Reference in New Issue
Block a user