mirror of
https://github.com/instaloader/instaloader.git
synced 2026-10-16 21:52:17 -07:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
75d3ab69d1 | ||
|
|
79820bb442 | ||
|
|
f70a7eb4db | ||
|
|
2dca99b89e | ||
|
|
1074a74a59 | ||
|
|
aa59683bd1 | ||
|
|
1db182ec27 | ||
|
|
d09493e669 | ||
|
|
0570dd71d6 | ||
|
|
fb14c2699e | ||
|
|
8784ac7d9b | ||
|
|
0c21da8c18 | ||
|
|
9b928b4679 | ||
|
|
d406de4f35 | ||
|
|
862aa4e82c | ||
|
|
644fab1f5e |
+1
-5
@@ -2,11 +2,7 @@
|
||||
|
||||
.. badges-start
|
||||
|
||||
|travis| |pypi| |pyversion| |license| |aur| |contributors| |downloads|
|
||||
|
||||
.. |travis| image:: https://img.shields.io/travis/instaloader/instaloader/master.svg
|
||||
:alt: Travis-CI Build Status
|
||||
:target: https://travis-ci.org/instaloader/instaloader
|
||||
|pypi| |pyversion| |license| |aur| |contributors| |downloads|
|
||||
|
||||
.. |pypi| image:: https://img.shields.io/pypi/v/instaloader.svg
|
||||
:alt: Instaloader PyPI Project Page
|
||||
|
||||
@@ -30,7 +30,7 @@ Iterator :class:`NodeIterator` and a context manager
|
||||
|
||||
It can be serialized and deserialized with :func:`save_structure_to_file`
|
||||
and :func:`load_structure_from_file`, as well as with :mod:`json` and
|
||||
:mod:`pickle` thanks to being a :func:`~collections.namedtuple`.
|
||||
:mod:`pickle` thanks to being a :class:`~typing.NamedTuple`.
|
||||
|
||||
``resumable_iteration``
|
||||
"""""""""""""""""""""""
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
"""Download pictures (or videos) along with their captions and other metadata from Instagram."""
|
||||
|
||||
|
||||
__version__ = '4.9.6'
|
||||
__version__ = '4.10b1'
|
||||
|
||||
|
||||
try:
|
||||
|
||||
@@ -524,7 +524,7 @@ class Instaloader:
|
||||
.. versionadded:: 4.3"""
|
||||
|
||||
http_response = self.context.get_raw(url)
|
||||
date_object = None # type: Optional[datetime]
|
||||
date_object: Optional[datetime] = None
|
||||
if 'Last-Modified' in http_response.headers:
|
||||
date_object = datetime.strptime(http_response.headers["Last-Modified"], '%a, %d %b %Y %H:%M:%S GMT')
|
||||
date_object = date_object.replace(tzinfo=timezone.utc)
|
||||
@@ -585,6 +585,23 @@ class Instaloader:
|
||||
.. versionadded:: 4.4"""
|
||||
self.download_title_pic(hashtag.profile_pic_url, '#' + hashtag.name, 'profile_pic', None)
|
||||
|
||||
@_requires_login
|
||||
def save_session(self) -> dict:
|
||||
"""Saves internally stored :class:`requests.Session` object to :class:`dict`.
|
||||
|
||||
:raises LoginRequiredException: If called without being logged in.
|
||||
|
||||
.. versionadded:: 4.10
|
||||
"""
|
||||
return self.context.save_session()
|
||||
|
||||
def load_session(self, username: str, session_data: dict) -> None:
|
||||
"""Internally stores :class:`requests.Session` object from :class:`dict`.
|
||||
|
||||
.. versionadded:: 4.10
|
||||
"""
|
||||
self.context.load_session(username, session_data)
|
||||
|
||||
@_requires_login
|
||||
def save_session_to_file(self, filename: Optional[str] = None) -> None:
|
||||
"""Saves internally stored :class:`requests.Session` object.
|
||||
@@ -712,7 +729,7 @@ class Instaloader:
|
||||
post.get_sidecar_nodes(self.slide_start, self.slide_end),
|
||||
start=self.slide_start % post.mediacount + 1
|
||||
):
|
||||
suffix = str(edge_number) # type: Optional[str]
|
||||
suffix: Optional[str] = str(edge_number)
|
||||
if '{filename}' in self.filename_pattern:
|
||||
suffix = None
|
||||
if self.download_pictures and (not sidecar_node.is_video or self.download_video_thumbnails):
|
||||
@@ -945,11 +962,11 @@ class Instaloader:
|
||||
"""
|
||||
for user_highlight in self.get_highlights(user):
|
||||
name = user_highlight.owner_username
|
||||
highlight_target = (filename_target
|
||||
highlight_target: Union[str, Path] = (filename_target
|
||||
if filename_target
|
||||
else (Path(_PostPathFormatter.sanitize_path(name, self.sanitize_paths)) /
|
||||
_PostPathFormatter.sanitize_path(user_highlight.title,
|
||||
self.sanitize_paths))) # type: Union[str, Path]
|
||||
self.sanitize_paths)))
|
||||
self.context.log("Retrieving highlights \"{}\" from profile {}".format(user_highlight.title, name))
|
||||
self.download_highlight_cover(user_highlight, highlight_target)
|
||||
totalcount = user_highlight.itemcount
|
||||
|
||||
@@ -9,6 +9,7 @@ import sys
|
||||
import textwrap
|
||||
import time
|
||||
import urllib.parse
|
||||
import uuid
|
||||
from contextlib import contextmanager
|
||||
from datetime import datetime, timedelta
|
||||
from functools import partial
|
||||
@@ -36,6 +37,34 @@ def default_user_agent() -> str:
|
||||
'(KHTML, like Gecko) Chrome/92.0.4515.131 Safari/537.36'
|
||||
|
||||
|
||||
def default_iphone_headers() -> Dict[str, Any]:
|
||||
return {'User-Agent': 'Instagram 273.0.0.16.70 (iPad13,8; iOS 16_3; en_US; en-US; ' \
|
||||
'scale=2.00; 2048x2732; 452417278) AppleWebKit/420+',
|
||||
'x-ads-opt-out': '1',
|
||||
'x-bloks-is-panorama-enabled': 'true',
|
||||
'x-bloks-version-id': '01507c21540f73e2216b6f62a11a5b5e51aa85491b72475c080da35b1228ddd6',
|
||||
'x-fb-client-ip': 'True',
|
||||
'x-fb-connection-type': 'wifi',
|
||||
'x-fb-http-engine': 'Liger',
|
||||
'x-fb-server-cluster': 'True',
|
||||
'x-fb': '1',
|
||||
'x-ig-abr-connection-speed-kbps': '2',
|
||||
'x-ig-app-id': '124024574287414',
|
||||
'x-ig-app-locale': 'en-US',
|
||||
'x-ig-app-startup-country': 'US',
|
||||
'x-ig-bandwidth-speed-kbps': '0.000',
|
||||
'x-ig-capabilities': '36r/F/8=',
|
||||
'x-ig-connection-speed': '{}kbps'.format(random.randint(1000, 20000)),
|
||||
'x-ig-connection-type': 'WiFi',
|
||||
'x-ig-device-locale': 'en-US',
|
||||
'x-ig-mapped-locale': 'en-US',
|
||||
'x-ig-timezone-offset': str((datetime.now().astimezone().utcoffset() or timedelta(seconds=0)).seconds),
|
||||
'x-ig-www-claim': '0',
|
||||
'x-pigeon-session-id': str(uuid.uuid4()),
|
||||
'x-tigon-is-retry': 'False',
|
||||
'x-whatsapp': '0'}
|
||||
|
||||
|
||||
class InstaloaderContext:
|
||||
"""Class providing methods for (error) logging and low-level communication with Instagram.
|
||||
|
||||
@@ -61,6 +90,7 @@ class InstaloaderContext:
|
||||
self.request_timeout = request_timeout
|
||||
self._session = self.get_anonymous_session()
|
||||
self.username = None
|
||||
self.user_id = None
|
||||
self.sleep = sleep
|
||||
self.quiet = quiet
|
||||
self.max_connection_attempts = max_connection_attempts
|
||||
@@ -68,9 +98,10 @@ class InstaloaderContext:
|
||||
self._root_rhx_gis = None
|
||||
self.two_factor_auth_pending = None
|
||||
self.iphone_support = iphone_support
|
||||
self.iphone_headers = default_iphone_headers()
|
||||
|
||||
# error log, filled with error() and printed at the end of Instaloader.main()
|
||||
self.error_log = [] # type: List[str]
|
||||
self.error_log: List[str] = []
|
||||
|
||||
self._rate_controller = rate_controller(self) if rate_controller is not None else RateController(self)
|
||||
|
||||
@@ -81,20 +112,26 @@ class InstaloaderContext:
|
||||
self.fatal_status_codes = fatal_status_codes or []
|
||||
|
||||
# Cache profile from id (mapping from id to Profile)
|
||||
self.profile_id_cache = dict() # type: Dict[int, Any]
|
||||
self.profile_id_cache: Dict[int, Any] = dict()
|
||||
|
||||
@contextmanager
|
||||
def anonymous_copy(self):
|
||||
session = self._session
|
||||
username = self.username
|
||||
user_id = self.user_id
|
||||
iphone_headers = self.iphone_headers
|
||||
self._session = self.get_anonymous_session()
|
||||
self.username = None
|
||||
self.user_id = None
|
||||
self.iphone_headers = default_iphone_headers()
|
||||
try:
|
||||
yield self
|
||||
finally:
|
||||
self._session.close()
|
||||
self.username = username
|
||||
self._session = session
|
||||
self.user_id = user_id
|
||||
self.iphone_headers = iphone_headers
|
||||
|
||||
@property
|
||||
def is_logged_in(self) -> bool:
|
||||
@@ -170,22 +207,30 @@ class InstaloaderContext:
|
||||
session.request = partial(session.request, timeout=self.request_timeout) # type: ignore
|
||||
return session
|
||||
|
||||
def save_session_to_file(self, sessionfile):
|
||||
"""Not meant to be used directly, use :meth:`Instaloader.save_session_to_file`."""
|
||||
pickle.dump(requests.utils.dict_from_cookiejar(self._session.cookies), sessionfile)
|
||||
def save_session(self):
|
||||
"""Not meant to be used directly, use :meth:`Instaloader.save_session`."""
|
||||
return requests.utils.dict_from_cookiejar(self._session.cookies)
|
||||
|
||||
def load_session_from_file(self, username, sessionfile):
|
||||
"""Not meant to be used directly, use :meth:`Instaloader.load_session_from_file`."""
|
||||
def load_session(self, username, sessiondata):
|
||||
"""Not meant to be used directly, use :meth:`Instaloader.load_session`."""
|
||||
session = requests.Session()
|
||||
session.cookies = requests.utils.cookiejar_from_dict(pickle.load(sessionfile))
|
||||
session.cookies = requests.utils.cookiejar_from_dict(sessiondata)
|
||||
session.headers.update(self._default_http_header())
|
||||
session.headers.update({'X-CSRFToken': session.cookies.get_dict()['csrftoken']})
|
||||
# Override default timeout behavior.
|
||||
# Need to silence mypy bug for this. See: https://github.com/python/mypy/issues/2427
|
||||
session.request = partial(session.request, timeout=self.request_timeout) # type: ignore
|
||||
session.request = partial(session.request, timeout=self.request_timeout) # type: ignore
|
||||
self._session = session
|
||||
self.username = username
|
||||
|
||||
def save_session_to_file(self, sessionfile):
|
||||
"""Not meant to be used directly, use :meth:`Instaloader.save_session_to_file`."""
|
||||
pickle.dump(self.save_session(), sessionfile)
|
||||
|
||||
def load_session_from_file(self, username, sessionfile):
|
||||
"""Not meant to be used directly, use :meth:`Instaloader.load_session_from_file`."""
|
||||
self.load_session(username, pickle.load(sessionfile))
|
||||
|
||||
def test_login(self) -> Optional[str]:
|
||||
"""Not meant to be used directly, use :meth:`Instaloader.test_login`."""
|
||||
data = self.graphql_query("d6f4427fbe92d846298cf93df0b937d3", {})
|
||||
@@ -265,6 +310,7 @@ class InstaloaderContext:
|
||||
session.headers.update({'X-CSRFToken': login.cookies['csrftoken']})
|
||||
self._session = session
|
||||
self.username = user
|
||||
self.user_id = resp_json['userId']
|
||||
|
||||
def two_factor_login(self, two_factor_code):
|
||||
"""Second step of login if 2FA is enabled.
|
||||
@@ -298,7 +344,8 @@ class InstaloaderContext:
|
||||
time.sleep(min(random.expovariate(0.6), 15.0))
|
||||
|
||||
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,
|
||||
response_headers: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
||||
"""JSON request to Instagram.
|
||||
|
||||
:param path: URL, relative to the given domain which defaults to www.instagram.com/
|
||||
@@ -335,13 +382,16 @@ class InstaloaderContext:
|
||||
redirect_url.startswith('https://i.instagram.com/accounts/login')):
|
||||
if not self.is_logged_in:
|
||||
raise LoginRequiredException("Redirected to login page. Use --login.")
|
||||
# alternate rate limit exceeded behavior
|
||||
raise TooManyRequestsException("Redirected to login")
|
||||
raise AbortDownloadException("Redirected to login page. You've been logged out, please wait " +
|
||||
"some time, recreate the session and try again")
|
||||
if redirect_url.startswith('https://{}/'.format(host)):
|
||||
resp = sess.get(redirect_url if redirect_url.endswith('/') else redirect_url + '/',
|
||||
params=params, allow_redirects=False)
|
||||
else:
|
||||
break
|
||||
if response_headers is not None:
|
||||
response_headers.clear()
|
||||
response_headers.update(resp.headers)
|
||||
if resp.status_code == 400:
|
||||
raise QueryReturnedBadRequestException("400 Bad Request")
|
||||
if resp.status_code == 404:
|
||||
@@ -392,7 +442,8 @@ class InstaloaderContext:
|
||||
self._rate_controller.handle_429('iphone')
|
||||
if is_other_query:
|
||||
self._rate_controller.handle_429('other')
|
||||
return self.get_json(path=path, params=params, host=host, session=sess, _attempt=_attempt + 1)
|
||||
return self.get_json(path=path, params=params, host=host, session=sess, _attempt=_attempt + 1,
|
||||
response_headers=response_headers)
|
||||
except KeyboardInterrupt:
|
||||
self.error("[skipped by user]", repeat_at_end=False)
|
||||
raise ConnectionException(error_string) from err
|
||||
@@ -482,11 +533,57 @@ class InstaloaderContext:
|
||||
|
||||
.. versionadded:: 4.2.1"""
|
||||
with copy_session(self._session, self.request_timeout) as tempsession:
|
||||
tempsession.headers['User-Agent'] = 'Instagram 146.0.0.27.125 (iPhone12,1; iOS 13_3; en_US; en-US; ' \
|
||||
'scale=2.00; 1656x3584; 190542906)'
|
||||
for header in ['Host', 'Origin', 'X-Instagram-AJAX', 'X-Requested-With']:
|
||||
# Set headers to simulate an API request from iPad
|
||||
tempsession.headers['ig-intended-user-id'] = str(self.user_id)
|
||||
tempsession.headers['x-pigeon-rawclienttime'] = '{:.6f}'.format(time.time())
|
||||
|
||||
# Add headers obtained from previous iPad request
|
||||
tempsession.headers.update(self.iphone_headers)
|
||||
|
||||
# Extract key information from cookies if we haven't got it already from a previous request
|
||||
header_cookies_mapping = {'x-mid': 'mid',
|
||||
'ig-u-ds-user-id': 'ds_user_id',
|
||||
'x-ig-device-id': 'ig_did',
|
||||
'x-ig-family-device-id': 'ig_did',
|
||||
'family_device_id': 'ig_did'}
|
||||
|
||||
# Map the cookie value to the matching HTTP request header
|
||||
cookies = tempsession.cookies.get_dict().copy()
|
||||
for key, value in header_cookies_mapping.items():
|
||||
if value in cookies:
|
||||
if key not in tempsession.headers:
|
||||
tempsession.headers[key] = cookies[value]
|
||||
else:
|
||||
# Remove the cookie value if it's already specified as a header
|
||||
tempsession.cookies.pop(value, None)
|
||||
|
||||
# Edge case for ig-u-rur header due to special string encoding in cookie
|
||||
if 'rur' in cookies:
|
||||
if 'ig-u-rur' not in tempsession.headers:
|
||||
tempsession.headers['ig-u-rur'] = cookies['rur'].strip('\"').encode('utf-8') \
|
||||
.decode('unicode_escape')
|
||||
else:
|
||||
tempsession.cookies.pop('rur', None)
|
||||
|
||||
# Remove headers specific to Desktop version
|
||||
for header in ['Host', 'Origin', 'X-Instagram-AJAX', 'X-Requested-With', 'Referer']:
|
||||
tempsession.headers.pop(header, None)
|
||||
return self.get_json(path, params, 'i.instagram.com', tempsession)
|
||||
|
||||
# No need for cookies if we have a bearer token
|
||||
if 'authorization' in tempsession.headers:
|
||||
tempsession.cookies.clear()
|
||||
|
||||
response_headers = dict() # type: Dict[str, Any]
|
||||
response = self.get_json(path, params, 'i.instagram.com', tempsession, response_headers=response_headers)
|
||||
|
||||
# Extract the ig-set-* headers and use them in the next request
|
||||
for key, value in response_headers.items():
|
||||
if key.startswith('ig-set-'):
|
||||
self.iphone_headers[key.replace('ig-set-', '')] = value
|
||||
elif key.startswith('x-ig-set-'):
|
||||
self.iphone_headers[key.replace('x-ig-set-', 'x-ig-')] = value
|
||||
|
||||
return response
|
||||
|
||||
def write_raw(self, resp: Union[bytes, requests.Response], filename: str) -> None:
|
||||
"""Write raw response data into a file.
|
||||
@@ -582,7 +679,7 @@ class RateController:
|
||||
|
||||
def __init__(self, context: InstaloaderContext):
|
||||
self._context = context
|
||||
self._query_timestamps = dict() # type: Dict[str, List[float]]
|
||||
self._query_timestamps: Dict[str, List[float]] = dict()
|
||||
self._earliest_next_request_time = 0.0
|
||||
self._iphone_earliest_next_request_time = 0.0
|
||||
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
import configparser
|
||||
from datetime import datetime, timezone
|
||||
from typing import Optional
|
||||
from os.path import dirname
|
||||
from os import makedirs
|
||||
|
||||
|
||||
class LatestStamps:
|
||||
@@ -25,6 +27,7 @@ class LatestStamps:
|
||||
self.data.read(latest_stamps_file)
|
||||
|
||||
def _save(self):
|
||||
makedirs(dirname(self.file), exist_ok=True)
|
||||
with open(self.file, 'w') as f:
|
||||
self.data.write(f)
|
||||
|
||||
|
||||
+10
-10
@@ -10,15 +10,15 @@ from typing import Any, Callable, Dict, Iterable, Iterator, NamedTuple, Optional
|
||||
from .exceptions import AbortDownloadException, InvalidArgumentException, QueryReturnedBadRequestException
|
||||
from .instaloadercontext import InstaloaderContext
|
||||
|
||||
FrozenNodeIterator = NamedTuple('FrozenNodeIterator',
|
||||
[('query_hash', str),
|
||||
('query_variables', Dict),
|
||||
('query_referer', Optional[str]),
|
||||
('context_username', Optional[str]),
|
||||
('total_index', int),
|
||||
('best_before', Optional[float]),
|
||||
('remaining_data', Optional[Dict]),
|
||||
('first_node', Optional[Dict])])
|
||||
class FrozenNodeIterator(NamedTuple):
|
||||
query_hash: str
|
||||
query_variables: Dict
|
||||
query_referer: Optional[str]
|
||||
context_username: Optional[str]
|
||||
total_index: int
|
||||
best_before: Optional[float]
|
||||
remaining_data: Optional[Dict]
|
||||
first_node: Optional[Dict]
|
||||
FrozenNodeIterator.query_hash.__doc__ = """The GraphQL ``query_hash`` parameter."""
|
||||
FrozenNodeIterator.query_variables.__doc__ = """The GraphQL ``query_variables`` parameter."""
|
||||
FrozenNodeIterator.query_referer.__doc__ = """The HTTP referer used for the GraphQL query."""
|
||||
@@ -95,7 +95,7 @@ class NodeIterator(Iterator[T]):
|
||||
self._is_first = is_first
|
||||
|
||||
def _query(self, after: Optional[str] = None) -> Dict:
|
||||
pagination_variables = {'first': NodeIterator._graphql_page_length} # type: Dict[str, Any]
|
||||
pagination_variables: Dict[str, Any] = {'first': NodeIterator._graphql_page_length}
|
||||
if after is not None:
|
||||
pagination_variables['after'] = after
|
||||
try:
|
||||
|
||||
+139
-37
@@ -2,12 +2,11 @@ import json
|
||||
import lzma
|
||||
import re
|
||||
from base64 import b64decode, b64encode
|
||||
from collections import namedtuple
|
||||
from contextlib import suppress
|
||||
from datetime import datetime
|
||||
from itertools import islice
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable, Dict, Iterable, Iterator, List, Optional, Tuple, Union
|
||||
from typing import Any, Callable, Dict, Iterable, Iterator, List, NamedTuple, Optional, Tuple, Union
|
||||
from unicodedata import normalize
|
||||
|
||||
from . import __version__
|
||||
@@ -16,25 +15,57 @@ from .instaloadercontext import InstaloaderContext
|
||||
from .nodeiterator import FrozenNodeIterator, NodeIterator
|
||||
from .sectioniterator import SectionIterator
|
||||
|
||||
PostSidecarNode = namedtuple('PostSidecarNode', ['is_video', 'display_url', 'video_url'])
|
||||
PostSidecarNode.__doc__ = "Item of a Sidecar Post."
|
||||
|
||||
class PostSidecarNode(NamedTuple):
|
||||
"""Item of a Sidecar Post."""
|
||||
is_video: bool
|
||||
display_url: str
|
||||
video_url: str
|
||||
|
||||
|
||||
PostSidecarNode.is_video.__doc__ = "Whether this node is a video."
|
||||
PostSidecarNode.display_url.__doc__ = "URL of image or video thumbnail."
|
||||
PostSidecarNode.video_url.__doc__ = "URL of video or None."
|
||||
|
||||
PostCommentAnswer = namedtuple('PostCommentAnswer', ['id', 'created_at_utc', 'text', 'owner', 'likes_count'])
|
||||
|
||||
class PostCommentAnswer(NamedTuple):
|
||||
id: int
|
||||
created_at_utc: datetime
|
||||
text: str
|
||||
owner: 'Profile'
|
||||
likes_count: int
|
||||
|
||||
|
||||
PostCommentAnswer.id.__doc__ = "ID number of comment."
|
||||
PostCommentAnswer.created_at_utc.__doc__ = ":class:`~datetime.datetime` when comment was created (UTC)."
|
||||
PostCommentAnswer.text.__doc__ = "Comment text."
|
||||
PostCommentAnswer.owner.__doc__ = "Owner :class:`Profile` of the comment."
|
||||
PostCommentAnswer.likes_count.__doc__ = "Number of likes on comment."
|
||||
|
||||
PostComment = namedtuple('PostComment', (*PostCommentAnswer._fields, 'answers')) # type: ignore
|
||||
|
||||
class PostComment(NamedTuple):
|
||||
id: int
|
||||
created_at_utc: datetime
|
||||
text: str
|
||||
owner: 'Profile'
|
||||
likes_count: int
|
||||
answers: Iterator[PostCommentAnswer]
|
||||
|
||||
|
||||
for field in PostCommentAnswer._fields:
|
||||
getattr(PostComment, field).__doc__ = getattr(PostCommentAnswer, field).__doc__ # pylint: disable=no-member
|
||||
PostComment.answers.__doc__ = r"Iterator which yields all :class:`PostCommentAnswer`\ s for the comment." # type: ignore
|
||||
PostComment.answers.__doc__ = r"Iterator which yields all :class:`PostCommentAnswer`\ s for the comment."
|
||||
|
||||
|
||||
class PostLocation(NamedTuple):
|
||||
id: int
|
||||
name: str
|
||||
slug: str
|
||||
has_public_page: Optional[bool]
|
||||
lat: Optional[float]
|
||||
lng: Optional[float]
|
||||
|
||||
|
||||
PostLocation = namedtuple('PostLocation', ['id', 'name', 'slug', 'has_public_page', 'lat', 'lng'])
|
||||
PostLocation.id.__doc__ = "ID number of location."
|
||||
PostLocation.name.__doc__ = "Location name."
|
||||
PostLocation.slug.__doc__ = "URL friendly variant of location name."
|
||||
@@ -42,6 +73,22 @@ PostLocation.has_public_page.__doc__ = "Whether location has a public page."
|
||||
PostLocation.lat.__doc__ = "Latitude (:class:`float` or None)."
|
||||
PostLocation.lng.__doc__ = "Longitude (:class:`float` or None)."
|
||||
|
||||
# This regular expression is by MiguelX413
|
||||
_hashtag_regex = re.compile(r"(?:#)((?:\w){1,150})")
|
||||
|
||||
# This regular expression is modified from jStassen, adjusted to use Python's \w to
|
||||
# support Unicode and a word/beginning of string delimiter at the beginning to ensure
|
||||
# that no email addresses join the list of mentions.
|
||||
# http://blog.jstassen.com/2016/03/code-regex-for-instagram-username-and-hashtags/
|
||||
_mention_regex = re.compile(r"(?:^|[^\w\n]|_)(?:@)(\w(?:(?:\w|(?:\.(?!\.))){0,28}(?:\w))?)", re.ASCII)
|
||||
|
||||
|
||||
def _optional_normalize(string: Optional[str]) -> Optional[str]:
|
||||
if string is not None:
|
||||
return normalize("NFC", string)
|
||||
else:
|
||||
return None
|
||||
|
||||
|
||||
class Post:
|
||||
"""
|
||||
@@ -73,8 +120,8 @@ class Post:
|
||||
self._context = context
|
||||
self._node = node
|
||||
self._owner_profile = owner_profile
|
||||
self._full_metadata_dict = None # type: Optional[Dict[str, Any]]
|
||||
self._location = None # type: Optional[PostLocation]
|
||||
self._full_metadata_dict: Optional[Dict[str, Any]] = None
|
||||
self._location: Optional[PostLocation] = None
|
||||
self._iphone_struct_ = None
|
||||
if 'iphone_struct' in node:
|
||||
# if loaded from JSON with load_structure_from_file()
|
||||
@@ -364,16 +411,10 @@ class Post:
|
||||
@property
|
||||
def caption(self) -> Optional[str]:
|
||||
"""Caption."""
|
||||
def _normalize(string: Optional[str]) -> Optional[str]:
|
||||
if string is not None:
|
||||
return normalize("NFC", string)
|
||||
else:
|
||||
return None
|
||||
|
||||
if "edge_media_to_caption" in self._node and self._node["edge_media_to_caption"]["edges"]:
|
||||
return _normalize(self._node["edge_media_to_caption"]["edges"][0]["node"]["text"])
|
||||
return _optional_normalize(self._node["edge_media_to_caption"]["edges"][0]["node"]["text"])
|
||||
elif "caption" in self._node:
|
||||
return _normalize(self._node["caption"])
|
||||
return _optional_normalize(self._node["caption"])
|
||||
return None
|
||||
|
||||
@property
|
||||
@@ -381,22 +422,14 @@ class Post:
|
||||
"""List of all lowercased hashtags (without preceeding #) that occur in the Post's caption."""
|
||||
if not self.caption:
|
||||
return []
|
||||
# This regular expression is from jStassen, adjusted to use Python's \w to support Unicode
|
||||
# http://blog.jstassen.com/2016/03/code-regex-for-instagram-username-and-hashtags/
|
||||
hashtag_regex = re.compile(r"(?:#)(\w(?:(?:\w|(?:\.(?!\.))){0,28}(?:\w))?)")
|
||||
return re.findall(hashtag_regex, self.caption.lower())
|
||||
return _hashtag_regex.findall(self.caption.lower())
|
||||
|
||||
@property
|
||||
def caption_mentions(self) -> List[str]:
|
||||
"""List of all lowercased profiles that are mentioned in the Post's caption, without preceeding @."""
|
||||
if not self.caption:
|
||||
return []
|
||||
# This regular expression is modified from jStassen, adjusted to use Python's \w to
|
||||
# support Unicode and a word/beginning of string delimiter at the beginning to ensure
|
||||
# that no email addresses join the list of mentions.
|
||||
# http://blog.jstassen.com/2016/03/code-regex-for-instagram-username-and-hashtags/
|
||||
mention_regex = re.compile(r"(?:^|\W|_)(?:@)(\w(?:(?:\w|(?:\.(?!\.))){0,28}(?:\w))?)", re.ASCII)
|
||||
return re.findall(mention_regex, self.caption.lower())
|
||||
return _mention_regex.findall(self.caption.lower())
|
||||
|
||||
@property
|
||||
def pcaption(self) -> str:
|
||||
@@ -515,7 +548,7 @@ class Post:
|
||||
def get_comments(self) -> Iterable[PostComment]:
|
||||
r"""Iterate over all comments of the post.
|
||||
|
||||
Each comment is represented by a PostComment namedtuple with fields text (string), created_at (datetime),
|
||||
Each comment is represented by a PostComment NamedTuple with fields text (string), created_at (datetime),
|
||||
id (int), owner (:class:`Profile`) and answers (:class:`~typing.Iterator`\ [:class:`PostCommentAnswer`])
|
||||
if available.
|
||||
|
||||
@@ -625,7 +658,7 @@ class Post:
|
||||
@property
|
||||
def location(self) -> Optional[PostLocation]:
|
||||
"""
|
||||
If the Post has a location, returns PostLocation namedtuple with fields 'id', 'lat' and 'lng' and 'name'.
|
||||
If the Post has a location, returns PostLocation NamedTuple with fields 'id', 'lat' and 'lng' and 'name'.
|
||||
|
||||
.. versionchanged:: 4.2.9
|
||||
Require being logged in (as required by Instagram), return None if not logged-in.
|
||||
@@ -681,7 +714,7 @@ class Profile:
|
||||
def __init__(self, context: InstaloaderContext, node: Dict[str, Any]):
|
||||
assert 'username' in node
|
||||
self._context = context
|
||||
self._has_public_story = None # type: Optional[bool]
|
||||
self._has_public_story: Optional[bool] = None
|
||||
self._node = node
|
||||
self._has_full_metadata = False
|
||||
self._iphone_struct_ = None
|
||||
@@ -874,7 +907,29 @@ class Profile:
|
||||
|
||||
@property
|
||||
def biography(self) -> str:
|
||||
return self._metadata('biography')
|
||||
return normalize("NFC", self._metadata('biography'))
|
||||
|
||||
@property
|
||||
def biography_hashtags(self) -> List[str]:
|
||||
"""
|
||||
List of all lowercased hashtags (without preceeding #) that occur in the Profile's biography.
|
||||
|
||||
.. versionadded:: 4.10
|
||||
"""
|
||||
if not self.biography:
|
||||
return []
|
||||
return _hashtag_regex.findall(self.biography.lower())
|
||||
|
||||
@property
|
||||
def biography_mentions(self) -> List[str]:
|
||||
"""
|
||||
List of all lowercased profiles that are mentioned in the Profile's biography, without preceeding @.
|
||||
|
||||
.. versionadded:: 4.10
|
||||
"""
|
||||
if not self.biography:
|
||||
return []
|
||||
return _mention_regex.findall(self.biography.lower())
|
||||
|
||||
@property
|
||||
def blocked_by_viewer(self) -> bool:
|
||||
@@ -1248,6 +1303,53 @@ class StoryItem:
|
||||
"""Type of post, GraphStoryImage or GraphStoryVideo"""
|
||||
return self._node['__typename']
|
||||
|
||||
@property
|
||||
def caption(self) -> Optional[str]:
|
||||
"""
|
||||
Caption.
|
||||
|
||||
.. versionadded:: 4.10
|
||||
"""
|
||||
if "edge_media_to_caption" in self._node and self._node["edge_media_to_caption"]["edges"]:
|
||||
return _optional_normalize(self._node["edge_media_to_caption"]["edges"][0]["node"]["text"])
|
||||
elif "caption" in self._node:
|
||||
return _optional_normalize(self._node["caption"])
|
||||
return None
|
||||
|
||||
@property
|
||||
def caption_hashtags(self) -> List[str]:
|
||||
"""
|
||||
List of all lowercased hashtags (without preceeding #) that occur in the StoryItem's caption.
|
||||
|
||||
.. versionadded:: 4.10
|
||||
"""
|
||||
if not self.caption:
|
||||
return []
|
||||
return _hashtag_regex.findall(self.caption.lower())
|
||||
|
||||
@property
|
||||
def caption_mentions(self) -> List[str]:
|
||||
"""
|
||||
List of all lowercased profiles that are mentioned in the StoryItem's caption, without preceeding @.
|
||||
|
||||
.. versionadded:: 4.10
|
||||
"""
|
||||
if not self.caption:
|
||||
return []
|
||||
return _mention_regex.findall(self.caption.lower())
|
||||
|
||||
@property
|
||||
def pcaption(self) -> str:
|
||||
"""
|
||||
Printable caption, useful as a format specifier for --filename-pattern.
|
||||
|
||||
.. versionadded:: 4.10
|
||||
"""
|
||||
def _elliptify(caption):
|
||||
pcaption = ' '.join([s.replace('/', '\u2215') for s in caption.splitlines() if s]).strip()
|
||||
return (pcaption[:30] + "\u2026") if len(pcaption) > 31 else pcaption
|
||||
return _elliptify(self.caption) if self.caption else ''
|
||||
|
||||
@property
|
||||
def is_video(self) -> bool:
|
||||
"""True if the StoryItem is a video."""
|
||||
@@ -1313,9 +1415,9 @@ class Story:
|
||||
def __init__(self, context: InstaloaderContext, node: Dict[str, Any]):
|
||||
self._context = context
|
||||
self._node = node
|
||||
self._unique_id = None # type: Optional[str]
|
||||
self._owner_profile = None # type: Optional[Profile]
|
||||
self._iphone_struct_ = None # type: Optional[Dict[str, Any]]
|
||||
self._unique_id: Optional[str] = None
|
||||
self._owner_profile: Optional[Profile] = None
|
||||
self._iphone_struct_: Optional[Dict[str, Any]] = None
|
||||
|
||||
def __repr__(self):
|
||||
return '<Story by {} changed {:%Y-%m-%d_%H-%M-%S_UTC}>'.format(self.owner_username, self.latest_media_utc)
|
||||
@@ -1431,8 +1533,8 @@ class Highlight(Story):
|
||||
def __init__(self, context: InstaloaderContext, node: Dict[str, Any], owner: Optional[Profile] = None):
|
||||
super().__init__(context, node)
|
||||
self._owner_profile = owner
|
||||
self._items = None # type: Optional[List[Dict[str, Any]]]
|
||||
self._iphone_struct_ = None # type: Optional[Dict[str, Any]]
|
||||
self._items: Optional[List[Dict[str, Any]]] = None
|
||||
self._iphone_struct_: Optional[Dict[str, Any]] = None
|
||||
|
||||
def __repr__(self):
|
||||
return '<Highlight by {}: {}>'.format(self.owner_username, self.title)
|
||||
|
||||
@@ -23,7 +23,7 @@ PRIVATE_PROFILE_ID = 1706625676
|
||||
EMPTY_PROFILE = "not_public"
|
||||
EMPTY_PROFILE_ID = 1928659031
|
||||
|
||||
ratecontroller = None # type: Optional[instaloader.RateController]
|
||||
ratecontroller: Optional[instaloader.RateController] = None
|
||||
|
||||
|
||||
class TestInstaloaderAnonymously(unittest.TestCase):
|
||||
|
||||
Reference in New Issue
Block a user