Compare commits

..
26 changed files with 1043 additions and 1617 deletions
-31
View File
@@ -1,31 +0,0 @@
name: "Documentation"
on: [push, pull_request]
jobs:
docs:
name: "Documentation"
runs-on: ubuntu-latest
steps:
- name: "Checkout Instaloader Repository"
uses: actions/checkout@v2
with:
fetch-depth: 0 # needed for building docs
- name: "Setup Python"
uses: actions/setup-python@v2
with:
python-version: 3.8
- name: "Install Dependencies"
run: |
python -m pip install pipenv==2023.2.4
pipenv --python `python --version | grep -Eo '3\.[0-9]+'` sync --dev
- name: "Build Documentation"
run: pipenv --python `python --version | grep -Eo '3\.[0-9]+'` run make -C docs html SPHINXOPTS="-W -n"
- name: "Deploy Documentation"
if: github.event_name == 'push' && github.ref == 'refs/heads/master'
uses: JamesIves/github-pages-deploy-action@v4.2.5
with:
branch: master
repository-name: instaloader/instaloader.github.io
folder: docs/_build/html
token: ${{ secrets.PAGES_DEPLOYMENT_TOKEN }}
+10 -6
View File
@@ -7,21 +7,25 @@ jobs:
name: PyLint and MyPy
runs-on: ubuntu-latest
strategy:
fail-fast: false
matrix:
python-version: ["3.8", "3.9", "3.10", "3.11"]
python-version: [3.8, 3.9]
steps:
- name: Checkout Instaloader Repository
uses: actions/checkout@v2
with:
fetch-depth: 0 # needed for building docs
- name: Setup Python
uses: actions/setup-python@v2
with:
python-version: ${{ matrix.python-version }}
- name: Install Dependencies
run: |
python -m pip install pipenv==2023.2.4
pipenv --python `python --version | grep -Eo '3\.[0-9]+'` sync --dev
python -m pip install pipenv==2020.11.15
pipenv sync --dev
- name: PyLint
run: pipenv --python `python --version | grep -Eo '3\.[0-9]+'` run pylint instaloader
run: pipenv run pylint instaloader
- name: MyPy
run: pipenv --python `python --version | grep -Eo '3\.[0-9]+'` run mypy -m instaloader
run: pipenv run mypy -m instaloader
- name: Build Documentation
if: matrix.python-version == '3.8'
run: pipenv run make -C docs html SPHINXOPTS="-W -n"
-36
View File
@@ -1,36 +0,0 @@
name: "Publish Packages"
on:
push:
tags:
- v*
jobs:
publish:
name: "Publish Package"
runs-on: ubuntu-latest
steps:
- name: "Checkout Repository"
uses: actions/checkout@v2
- name: "Setup Python"
uses: actions/setup-python@v2
with:
python-version: 3.8
- name: "Get the tagged version"
id: get_version
env:
GITHUB_REF: ${{ github.ref }}
run: echo ::set-output name=VERSION::${GITHUB_REF/refs\/tags\/v/}
shell: bash
- name: "Generate Python Package"
run: python setup.py sdist
- name: "Publish Distribution to PyPI"
uses: pypa/gh-action-pypi-publish@master
with:
password: ${{ secrets.PYPI_PASSWORD }}
- name: "Publish Distribution to AUR"
run: deploy/arch/deploy.sh $VERSION_TAG
env:
VERSION_TAG: v${{ steps.get_version.outputs.VERSION }}
AUR_KEY: ${{ secrets.AUR_KEY }}
AUR_IV: ${{ secrets.AUR_IV }}
+2 -2
View File
@@ -18,7 +18,7 @@ jobs:
stale-issue-message: 'There has been no activity on this question for an extended period of time. This issue will be closed after further 14 days of inactivity.'
stale-issue-label: 'stale'
exempt-issue-labels: 'leave open'
days-before-stale: 21
days-before-stale: 15
days-before-close: -1
remove-stale-when-updated: false
- uses: actions/stale@v1
@@ -30,5 +30,5 @@ jobs:
stale-pr-label: 'stale'
exempt-issue-label: 'leave open'
exempt-pr-label: 'leave open'
days-before-stale: 189
days-before-stale: 135
days-before-close: 14
+2 -2
View File
@@ -10,9 +10,9 @@ jobs:
- name: Checkout Instaloader repository
uses: actions/checkout@v2
- name: Setup Python
uses: actions/setup-python@v2
uses: actions/setup-python@v1
with:
python-version: "3.9"
python-version: 3.9
architecture: x64
- name: Get the tagged version
id: get_version
+405 -467
View File
File diff suppressed because it is too large Load Diff
+35
View File
@@ -0,0 +1,35 @@
dist: xenial
language: python
python:
- "3.6"
- "3.7"
install:
- pipenv sync --dev
script:
- pylint instaloader
- mypy -m instaloader
- make -C docs html SPHINXOPTS="-W -n"
addons:
ssh_known_hosts: aur.archlinux.org
deploy:
- provider: pypi
user: aandergr
on:
tags: true
python: 3.6
password:
secure: KBsATtXNnEzqr0X2ALdyx2zaYj2EUfK8YjP4jEOilBELpJ4evHYXVgRkn9CKXGsD7LKbWmR3gz2rr8NZ6U42pHw0fpQHizz91MpJSKbNY7RZZh4UVWMDh81izeglchWAF+m3Wdpxpo/GBDbrZIbRQFhPgB+kjSzJo3uiCS+yVXtGG+6csJHzd0WkJjsVHz+XSdxyOiLy/qFF3nmY6ZrVrgw6WbX2YLbx9je6DxK3YRrnBqgMgH7Ui2xWWbgJCk3KKpQTMiMKnTa9mdeMJAd26YMCz1gG/X7ck6VNly/6bs5+Zq7/bU0dSolhPBhNthyccYdS0IeGPdPF3YqatblJ4fLGTfHn8oJs3/T11yIW9pvOT3UftEynPOUX+en2Qnb/3WfgwUfVts8i8DAYgEFLGEzbF5XMGy4gwAD/bZ0JW0pmgAG2PZsdOokAt+ydlNj/uxq1GFW2doZbul6k4t03p1mik6itYscvuhKoEm8Yw6wr1P9aDvMR6FFpSu4gDt2XrKmRmWkSXxpNeCfZujTjyAnISxLVQGSY7BG5K0Z2HbDtqqNz9nj4WgNGzZmngiRofCevWR+vPIDUDW7NCX03csfcFS9RRllBO/BrL7NVUV+Qp9dfTQYp+8SqYLWgHtDoXdgyMREjJaAmbx2u/TYTY8bADZ5F3vGUpv2j5hP3NTU=
- provider: pages
skip-cleanup: true
github-token: $GITHUB_TOKEN
local-dir: docs/_build/html
repo: instaloader/instaloader.github.io
target-branch: master
on:
branch: master
python: 3.6
- provider: script
script: deploy/arch/deploy.sh $TRAVIS_TAG
on:
tags: true
python: 3.6
+2 -6
View File
@@ -12,12 +12,8 @@ pefile = "*"
pywin32-ctypes = "*"
psutil = "*"
typing-extensions = "*"
types-requests = "*"
typed-ast = "*"
jinja2 = "~=3.0.1"
wrapt = "*" # Needed for Python<3.11
tomli = "*" # Needed for Python<3.11
dill = "*" # Needed for Python<3.11
[packages]
requests = "*"
pycryptodome = "*"
pynacl = "*"
Generated
+378 -576
View File
File diff suppressed because it is too large Load Diff
-3
View File
@@ -127,9 +127,6 @@ Supporters
.. current-sponsors-start
| Instaloader is proudly sponsored by
| `@socialmethod <https://github.com/socialmethod>`__
See `Alex' GitHub Sponsors <https://github.com/sponsors/aandergr>`__ page for
how you can sponsor the development of Instaloader!
+1 -1
View File
@@ -6,7 +6,7 @@ pkgbase = instaloader
arch = any
license = MIT
makedepends = python-setuptools
depends = python>=3.8
depends = python>=3.6
depends = python-requests>=2.4
options = !emptydirs
source = instaloader-{{version}}.tar.gz::https://codeload.github.com/instaloader/instaloader/tar.gz/v{{version}}
+1 -1
View File
@@ -8,7 +8,7 @@ arch=('any')
url="https://instaloader.github.io/"
license=('MIT')
groups=()
depends=('python>=3.8' 'python-requests>=2.4')
depends=('python>=3.6' 'python-requests>=2.4')
makedepends=('python-setuptools')
options=('!emptydirs')
source=($pkgname-$pkgver.tar.gz::https://codeload.github.com/instaloader/instaloader/tar.gz/v$pkgver)
+1 -3
View File
@@ -8,12 +8,10 @@ echo $VERSION | grep -qe "[abc]" && exit 0
cd $(dirname $0)
# decrypt and add ssh key
openssl aes-256-cbc -K $AUR_KEY -iv $AUR_IV -in id_rsa_AUR.enc -out /tmp/AUR_openssh -d
openssl aes-256-cbc -K $encrypted_354637631c28_key -iv $encrypted_354637631c28_iv -in id_rsa_AUR.enc -out /tmp/AUR_openssh -d
eval "$(ssh-agent -s)"
chmod 600 /tmp/AUR_openssh
ssh-add /tmp/AUR_openssh
mkdir -p ~/.ssh
ssh-keyscan -t rsa aur.archlinux.org >> ~/.ssh/known_hosts
# clone and modify AUR repo
git clone --depth 1 ssh://aur@aur.archlinux.org/instaloader.git
Binary file not shown.
+2 -2
View File
@@ -44,8 +44,8 @@ with open('__main__.py', 'w+') as f:
f.writelines(lines)
# install dependencies and invoke PyInstaller
commands = ["pip install pipenv==2023.2.4",
f"pipenv --python {sys.version_info.major}.{sys.version_info.minor} sync --dev",
commands = ["pip install pipenv==2020.11.15",
"pipenv sync --dev",
"pipenv run pyinstaller --log-level=DEBUG instaloader.spec"]
for command in commands:
+3 -12
View File
@@ -239,13 +239,6 @@ How to Download
.. versionadded:: 4.8
.. option:: --sanitize-paths
Force sanitization of paths so that the resulting file and directory names
are valid on both Windows and Unix.
.. versionadded:: 4.9
.. option:: --resume-prefix prefix
For many targets, Instaloader is capable of resuming a previously-aborted
@@ -258,8 +251,7 @@ How to Download
- Profile posts,
- Profile IGTV posts (:option:`--igtv`),
- Profile tagged posts (:option:`--tagged`),
- Saved posts (``:saved``),
- Hashtags.
- Saved posts (``:saved``).
This feature is enabled by default for targets where it is supported;
:option:`--resume-prefix` only changes the name of the iterator files.
@@ -281,7 +273,7 @@ How to Download
.. option:: --user-agent USER_AGENT
User Agent to use for HTTP requests. Per default, Instaloader pretends being
Chrome/92 on Linux.
Chrome/89 on Linux.
.. option:: --max-connection-attempts N
@@ -329,13 +321,12 @@ Miscellaneous Options
Read arguments from file `args.txt`, a shortcut to provide arguments from
file rather than command-line. This provides a convenient way to hide login
info from CLI, and can also be used to simplify management of long arguments.
You can provide more than one file at once, e.g.: ``+args1.txt +args2.txt``.
.. note::
Text file should separate arguments with line breaks.
`args.txt` example::
args.txt example::
--login=MYUSERNAME
--password=MYPASSWORD
+1 -1
View File
@@ -48,7 +48,7 @@ autodoc_default_flags = ['show-inheritance', 'members', 'undoc-members']
autodoc_member_order = 'bysource'
intersphinx_mapping = {'python': ('https://docs.python.org/3', None),
'requests': ('https://requests.readthedocs.io/en/latest/', None)}
'requests': ('https://requests.kennethreitz.org/en/master/', None)}
nitpick_ignore = [('py:class', 'typing.Tuple')]
+1 -1
View File
@@ -8,7 +8,7 @@ Install Instaloader
To **install Instaloader**,
#. Ensure that you have `Python <https://www.python.org/>`__, at least
version 3.8, and `pip <https://pypi.python.org/pypi/pip>`__
version 3.6, and `pip <https://pypi.python.org/pypi/pip>`__
installed.
#. Then, install Instaloader using::
+1 -1
View File
@@ -1,7 +1,7 @@
"""Download pictures (or videos) along with their captions and other metadata from Instagram."""
__version__ = '4.9.6'
__version__ = '4.8.1'
try:
+2 -5
View File
@@ -45,6 +45,7 @@ def filterstr_to_filterfunc(filter_str: str, item_type: type):
class TransformFilterAst(ast.NodeTransformer):
def visit_Name(self, node: ast.Name):
# pylint:disable=no-self-use
if not isinstance(node.ctx, ast.Load):
raise InvalidArgumentException("Invalid filter: Modifying variables ({}) not allowed.".format(node.id))
if node.id == "datetime":
@@ -382,9 +383,6 @@ def main():
g_how.add_argument('--resume-prefix', metavar='PREFIX',
help='Prefix for filenames that are used to save the information to resume an interrupted '
'download.')
g_how.add_argument('--sanitize-paths', action='store_true',
help='Sanitize paths so that the resulting file and directory names are valid on both '
'Windows and Unix.')
g_how.add_argument('--no-resume', action='store_true',
help='Do not resume a previously-aborted download iteration, and do not save such information '
'when interrupted.')
@@ -465,8 +463,7 @@ def main():
slide=args.slide,
fatal_status_codes=args.abort_on,
iphone_support=not args.no_iphone,
title_pattern=args.title_pattern,
sanitize_paths=args.sanitize_paths)
title_pattern=args.title_pattern)
_main(loader,
args.profile,
username=args.login.lower() if args.login is not None else None,
+62 -105
View File
@@ -22,7 +22,6 @@ from .exceptions import *
from .instaloadercontext import InstaloaderContext, RateController
from .lateststamps import LatestStamps
from .nodeiterator import NodeIterator, resumable_iteration
from .sectioniterator import SectionIterator
from .structures import (Hashtag, Highlight, JsonExportable, Post, PostLocation, Profile, Story, StoryItem,
load_structure_from_file, save_structure_to_file, PostSidecarNode, TitlePic)
@@ -137,38 +136,20 @@ class _ArbitraryItemFormatter(string.Formatter):
class _PostPathFormatter(_ArbitraryItemFormatter):
RESERVED: set = {'CON', 'PRN', 'AUX', 'NUL',
'COM1', 'COM2', 'COM3', 'COM4', 'COM5', 'COM6', 'COM7', 'COM8', 'COM9',
'LPT1', 'LPT2', 'LPT3', 'LPT4', 'LPT5', 'LPT6', 'LPT7', 'LPT8', 'LPT9'}
def __init__(self, item: Any, force_windows_path: bool = False):
super().__init__(item)
self.force_windows_path = force_windows_path
def get_value(self, key, args, kwargs):
ret = super().get_value(key, args, kwargs)
if not isinstance(ret, str):
return ret
return self.sanitize_path(ret, self.force_windows_path)
return self.sanitize_path(ret)
@staticmethod
def sanitize_path(ret: str, force_windows_path: bool = False) -> str:
def sanitize_path(ret: str) -> str:
"""Replaces '/' with similar looking Division Slash and some other illegal filename characters on Windows."""
ret = ret.replace('/', '\u2215')
if ret.startswith('.'):
ret = ret.replace('.', '\u2024', 1)
if force_windows_path or platform.system() == 'Windows':
if platform.system() == 'Windows':
ret = ret.replace(':', '\uff1a').replace('<', '\ufe64').replace('>', '\ufe65').replace('\"', '\uff02')
ret = ret.replace('\\', '\ufe68').replace('|', '\uff5c').replace('?', '\ufe16').replace('*', '\uff0a')
ret = ret.replace('\n', ' ').replace('\r', ' ')
root, ext = os.path.splitext(ret)
if root.upper() in _PostPathFormatter.RESERVED:
root += '_'
if ext == '.':
ext = '\u2024'
ret = root + ext
return ret
@@ -201,7 +182,6 @@ class Instaloader:
:param slide: :option:`--slide`
:param fatal_status_codes: :option:`--abort-on`
:param iphone_support: not :option:`--no-iphone`
:param sanitize_paths: :option:`--sanitize-paths`
.. attribute:: context
@@ -221,8 +201,8 @@ class Instaloader:
download_comments: bool = False,
save_metadata: bool = True,
compress_json: bool = True,
post_metadata_txt_pattern: Optional[str] = None,
storyitem_metadata_txt_pattern: Optional[str] = None,
post_metadata_txt_pattern: str = None,
storyitem_metadata_txt_pattern: str = None,
max_connection_attempts: int = 3,
request_timeout: float = 300.0,
rate_controller: Optional[Callable[[InstaloaderContext], RateController]] = None,
@@ -231,8 +211,7 @@ class Instaloader:
slide: Optional[str] = None,
fatal_status_codes: Optional[List[int]] = None,
iphone_support: bool = True,
title_pattern: Optional[str] = None,
sanitize_paths: bool = False):
title_pattern: Optional[str] = None):
self.context = InstaloaderContext(sleep, quiet, user_agent, max_connection_attempts,
request_timeout, rate_controller, fatal_status_codes,
@@ -249,7 +228,6 @@ class Instaloader:
self.title_pattern = '{date_utc}_UTC_{typename}'
else:
self.title_pattern = '{target}_{date_utc}_UTC_{typename}'
self.sanitize_paths = sanitize_paths
self.download_pictures = download_pictures
self.download_videos = download_videos
self.download_video_thumbnails = download_video_thumbnails
@@ -313,8 +291,7 @@ class Instaloader:
check_resume_bbd=self.check_resume_bbd,
slide=self.slide,
fatal_status_codes=self.context.fatal_status_codes,
iphone_support=self.context.iphone_support,
sanitize_paths=self.sanitize_paths)
iphone_support=self.context.iphone_support)
yield new_loader
self.context.error_log.extend(new_loader.context.error_log)
new_loader.context.error_log = [] # avoid double-printing of errors
@@ -335,25 +312,15 @@ class Instaloader:
filename_suffix: Optional[str] = None, _attempt: int = 1) -> bool:
"""Downloads and saves picture with given url under given directory with given timestamp.
Returns true, if file was actually downloaded, i.e. updated."""
if filename_suffix is not None:
filename += '_' + filename_suffix
urlmatch = re.search('\\.[a-z0-9]*\\?', url)
file_extension = url[-3:] if urlmatch is None else urlmatch.group(0)[1:-1]
nominal_filename = filename + '.' + file_extension
if os.path.isfile(nominal_filename):
self.context.log(nominal_filename + ' exists', end=' ', flush=True)
return False
resp = self.context.get_raw(url)
if 'Content-Type' in resp.headers and resp.headers['Content-Type']:
header_extension = '.' + resp.headers['Content-Type'].split(';')[0].split('/')[-1]
header_extension = header_extension.lower().replace('jpeg', 'jpg')
filename += header_extension
else:
filename = nominal_filename
if filename != nominal_filename and os.path.isfile(filename):
if filename_suffix is not None:
filename += '_' + filename_suffix
filename += '.' + file_extension
if os.path.isfile(filename):
self.context.log(filename + ' exists', end=' ', flush=True)
return False
self.context.write_raw(resp, filename)
self.context.get_and_write_raw(url, filename)
os.utime(filename, (datetime.now().timestamp(), mtime.timestamp()))
return True
@@ -447,7 +414,7 @@ class Instaloader:
"""Updates picture caption / Post metadata info"""
def _elliptify(caption):
pcaption = caption.replace('\n', ' ').strip()
return '[' + ((pcaption[:29] + "\u2026") if len(pcaption) > 31 else pcaption) + ']'
return '[' + ((pcaption[:29] + u"\u2026") if len(pcaption) > 31 else pcaption) + ']'
filename += '.txt'
caption += '\n'
pcaption = _elliptify(caption)
@@ -478,19 +445,17 @@ class Instaloader:
self.context.log(pcaption, end=' ', flush=True)
except UnicodeEncodeError:
self.context.log('txt', end=' ', flush=True)
with open(filename, 'w', encoding='UTF-8') as fio:
fio.write(caption)
with open(filename, 'wb') as text_file:
with BytesIO(bcaption) as bio:
shutil.copyfileobj(cast(IO, bio), text_file)
os.utime(filename, (datetime.now().timestamp(), mtime.timestamp()))
def save_location(self, filename: str, location: PostLocation, mtime: datetime) -> None:
"""Save post location name and Google Maps link."""
filename += '_location.txt'
if location.lat is not None and location.lng is not None:
location_string = (location.name + "\n" +
"https://maps.google.com/maps?q={0},{1}&ll={0},{1}\n".format(location.lat,
location.lng))
else:
location_string = location.name
location_string = (location.name + "\n" +
"https://maps.google.com/maps?q={0},{1}&ll={0},{1}\n".format(location.lat,
location.lng))
with open(filename, 'wb') as text_file:
with BytesIO(location_string.encode()) as bio:
shutil.copyfileobj(cast(IO, bio), text_file)
@@ -533,10 +498,9 @@ class Instaloader:
pic_bytes = http_response.content
ig_filename = url.split('/')[-1].split('?')[0]
pic_data = TitlePic(owner_profile, target, name_suffix, ig_filename, date_object)
dirname = _PostPathFormatter(pic_data, self.sanitize_paths).format(self.dirname_pattern, target=target)
filename_template = os.path.join(
dirname,
_PostPathFormatter(pic_data, self.sanitize_paths).format(self.title_pattern, target=target))
dirname = _PostPathFormatter(pic_data).format(self.dirname_pattern, target=target)
filename_template = os.path.join(dirname,
_PostPathFormatter(pic_data).format(self.title_pattern, target=target))
filename = self.__prepare_filename(filename_template, lambda: url) + ".jpg"
content_length = http_response.headers.get('Content-Length', None)
if os.path.isfile(filename) and (not self.context.is_logged_in or
@@ -562,7 +526,7 @@ class Instaloader:
if latest_stamps is None:
self.download_profilepic(profile)
return
profile_pic_basename = profile.profile_pic_url_no_iphone.split('/')[-1].split('?')[0]
profile_pic_basename = profile.profile_pic_url.split('/')[-1].split('?')[0]
saved_basename = latest_stamps.get_profile_pic(profile.username)
if saved_basename == profile_pic_basename:
return
@@ -661,7 +625,7 @@ class Instaloader:
"""Format filename of a :class:`Post` or :class:`StoryItem` according to ``filename-pattern`` parameter.
.. versionadded:: 4.1"""
return _PostPathFormatter(item, self.sanitize_paths).format(self.filename_pattern, target=target)
return _PostPathFormatter(item).format(self.filename_pattern, target=target)
def download_post(self, post: Post, target: Union[str, Path]) -> bool:
"""
@@ -693,7 +657,7 @@ class Instaloader:
return False
return True
dirname = _PostPathFormatter(post, self.sanitize_paths).format(self.dirname_pattern, target=target)
dirname = _PostPathFormatter(post).format(self.dirname_pattern, target=target)
filename_template = os.path.join(dirname, self.format_filename(post, target=target))
filename = self.__prepare_filename(filename_template, lambda: post.url)
@@ -844,7 +808,7 @@ class Instaloader:
last_scraped = latest_stamps.get_last_story_timestamp(name)
scraped_timestamp = datetime.now().astimezone()
for item in user_story.get_items():
if latest_stamps is not None and item.date_local <= last_scraped:
if latest_stamps is not None and item.date_utc.replace(tzinfo=timezone.utc) <= last_scraped:
break
if storyitem_filter is not None and not storyitem_filter(item):
self.context.log("<{} skipped>".format(item), flush=True)
@@ -874,22 +838,17 @@ class Instaloader:
return True
date_local = item.date_local
dirname = _PostPathFormatter(item, self.sanitize_paths).format(self.dirname_pattern, target=target)
dirname = _PostPathFormatter(item).format(self.dirname_pattern, target=target)
filename_template = os.path.join(dirname, self.format_filename(item, target=target))
filename = self.__prepare_filename(filename_template, lambda: item.url)
downloaded = False
video_url_fetch_failed = False
if item.is_video and self.download_videos is True:
video_url = item.video_url
if video_url:
filename = self.__prepare_filename(filename_template, lambda: str(video_url))
downloaded |= (not _already_downloaded(filename + ".mp4") and
self.download_pic(filename=filename, url=video_url, mtime=date_local))
else:
video_url_fetch_failed = True
if video_url_fetch_failed or not item.is_video or self.download_video_thumbnails is True:
if not item.is_video or self.download_video_thumbnails is True:
downloaded = (not _already_downloaded(filename + ".jpg") and
self.download_pic(filename=filename, url=item.url, mtime=date_local))
if item.is_video and self.download_videos is True:
filename = self.__prepare_filename(filename_template, lambda: str(item.video_url))
downloaded |= (not _already_downloaded(filename + ".mp4") and
self.download_pic(filename=filename, url=item.video_url, mtime=date_local))
# Save caption if desired
metadata_string = _ArbitraryItemFormatter(item).format(self.storyitem_metadata_txt_pattern).strip()
if metadata_string:
@@ -947,9 +906,8 @@ class Instaloader:
name = user_highlight.owner_username
highlight_target = (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]
else (Path(_PostPathFormatter.sanitize_path(name)) /
_PostPathFormatter.sanitize_path(user_highlight.title))) # type: Union[str, Path]
self.context.log("Retrieving highlights \"{}\" from profile {}".format(user_highlight.title, name))
self.download_highlight_cover(user_highlight, highlight_target)
totalcount = user_highlight.itemcount
@@ -999,7 +957,7 @@ class Instaloader:
else total_count)
sanitized_target = target
if isinstance(target, str):
sanitized_target = _PostPathFormatter.sanitize_path(target, self.sanitize_paths)
sanitized_target = _PostPathFormatter.sanitize_path(target)
if takewhile is None:
takewhile = lambda _: True
with resumable_iteration(
@@ -1014,10 +972,7 @@ class Instaloader:
enabled=self.resume_prefix is not None
) as (is_resuming, start_index):
for number, post in enumerate(posts, start=start_index + 1):
should_stop = not takewhile(post)
if should_stop and post.is_pinned:
continue
if (max_count is not None and number > max_count) or should_stop:
if (max_count is not None and number > max_count) or not takewhile(post):
break
if displayed_count is not None:
self.context.log("[{0:{w}d}/{1:{w}d}] ".format(number, displayed_count,
@@ -1049,7 +1004,7 @@ class Instaloader:
except PostChangedException:
post_changed = True
continue
if fast_update and not downloaded and not post_changed and not post.is_pinned:
if fast_update and not downloaded and not post_changed:
# disengage fast_update for first post when resuming
if not is_resuming or number > 0:
break
@@ -1080,7 +1035,7 @@ class Instaloader:
'has_stories': False})["data"]
@_requires_login
def download_feed_posts(self, max_count: Optional[int] = None, fast_update: bool = False,
def download_feed_posts(self, max_count: int = None, fast_update: bool = False,
post_filter: Optional[Callable[[Post], bool]] = None) -> None:
"""
Download pictures from the user's feed.
@@ -1101,7 +1056,7 @@ class Instaloader:
self.posts_download_loop(self.get_feed_posts(), ":feed", fast_update, post_filter, max_count=max_count)
@_requires_login
def download_saved_posts(self, max_count: Optional[int] = None, fast_update: bool = False,
def download_saved_posts(self, max_count: int = None, fast_update: bool = False,
post_filter: Optional[Callable[[Post], bool]] = None) -> None:
"""Download user's saved pictures.
@@ -1129,12 +1084,18 @@ class Instaloader:
.. versionchanged:: 4.2.9
Require being logged in (as required by Instagram)
"""
yield from SectionIterator(
self.context,
lambda d: d["native_location_data"]["recent"],
lambda m: Post.from_iphone_struct(self.context, m),
f"explore/locations/{location}/",
)
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']
@_requires_login
def download_location(self, location: str,
@@ -1183,8 +1144,8 @@ class Instaloader:
"""Get Posts associated with a #hashtag.
.. deprecated:: 4.4
Use :meth:`Hashtag.get_posts_resumable`."""
return Hashtag.from_name(self.context, hashtag).get_posts_resumable()
Use :meth:`Hashtag.get_posts`."""
return Hashtag.from_name(self.context, hashtag).get_posts()
def download_hashtag(self, hashtag: Union[Hashtag, str],
max_count: Optional[int] = None,
@@ -1220,7 +1181,7 @@ class Instaloader:
self.download_hashtag_profilepic(hashtag)
if posts:
self.context.log("Retrieving pictures with hashtag #{}...".format(hashtag.name))
self.posts_download_loop(hashtag.get_posts_resumable(), target, fast_update, post_filter,
self.posts_download_loop(hashtag.get_all_posts(), target, fast_update, post_filter,
max_count=max_count)
if self.save_metadata:
json_filename = '{0}/{1}'.format(self.dirname_pattern.format(profile=target,
@@ -1242,15 +1203,15 @@ class Instaloader:
posts_takewhile: Optional[Callable[[Post], bool]] = None
if latest_stamps is not None:
last_scraped = latest_stamps.get_last_tagged_timestamp(profile.username)
posts_takewhile = lambda p: p.date_local > last_scraped
posts_takewhile = lambda p: p.date_utc.replace(tzinfo=timezone.utc) > last_scraped
tagged_posts = profile.get_tagged_posts()
self.posts_download_loop(tagged_posts,
target if target
else (Path(_PostPathFormatter.sanitize_path(profile.username, self.sanitize_paths)) /
_PostPathFormatter.sanitize_path(':tagged', self.sanitize_paths)),
else (Path(_PostPathFormatter.sanitize_path(profile.username)) /
_PostPathFormatter.sanitize_path(':tagged')),
fast_update, post_filter, takewhile=posts_takewhile)
if latest_stamps is not None and tagged_posts.first_item is not None:
latest_stamps.set_last_tagged_timestamp(profile.username, tagged_posts.first_item.date_local)
latest_stamps.set_last_tagged_timestamp(profile.username, tagged_posts.first_item.date_local.astimezone())
def download_igtv(self, profile: Profile, fast_update: bool = False,
post_filter: Optional[Callable[[Post], bool]] = None,
@@ -1265,12 +1226,12 @@ class Instaloader:
posts_takewhile: Optional[Callable[[Post], bool]] = None
if latest_stamps is not None:
last_scraped = latest_stamps.get_last_igtv_timestamp(profile.username)
posts_takewhile = lambda p: p.date_local > last_scraped
posts_takewhile = lambda p: p.date_utc.replace(tzinfo=timezone.utc) > last_scraped
igtv_posts = profile.get_igtv_posts()
self.posts_download_loop(igtv_posts, profile.username, fast_update, post_filter,
total_count=profile.igtvcount, owner_profile=profile, takewhile=posts_takewhile)
if latest_stamps is not None and igtv_posts.first_item is not None:
latest_stamps.set_last_igtv_timestamp(profile.username, igtv_posts.first_item.date_local)
latest_stamps.set_last_igtv_timestamp(profile.username, igtv_posts.first_item.date_local.astimezone())
def _get_id_filename(self, profile_name: str) -> str:
if ((format_string_contains_key(self.dirname_pattern, 'profile') or
@@ -1341,10 +1302,6 @@ class Instaloader:
profile_id))
profile_from_id = Profile.from_id(self.context, profile_id)
newname = profile_from_id.username
if profile_name == newname:
self.context.error(
f"Warning: Profile {profile_name} could not be retrieved by its name, but by its ID.")
return profile_from_id
self.context.error("Profile {0} has changed its name to {1}.".format(profile_name, newname))
if latest_stamps is None:
if ((format_string_contains_key(self.dirname_pattern, 'profile') or
@@ -1467,14 +1424,14 @@ class Instaloader:
if latest_stamps is not None:
# pylint:disable=cell-var-from-loop
last_scraped = latest_stamps.get_last_post_timestamp(profile_name)
posts_takewhile = lambda p: p.date_local > last_scraped
posts_takewhile = lambda p: p.date_utc.replace(tzinfo=timezone.utc) > last_scraped
posts_to_download = profile.get_posts()
self.posts_download_loop(posts_to_download, profile_name, fast_update, post_filter,
total_count=profile.mediacount, owner_profile=profile,
takewhile=posts_takewhile)
if latest_stamps is not None and posts_to_download.first_item is not None:
latest_stamps.set_last_post_timestamp(profile_name,
posts_to_download.first_item.date_local)
posts_to_download.first_item.date_local.astimezone())
if stories and profiles:
with self.context.error_catcher("Download stories"):
+45 -9
View File
@@ -1,3 +1,4 @@
import base64
import hashlib
import json
import os
@@ -12,10 +13,13 @@ import urllib.parse
from contextlib import contextmanager
from datetime import datetime, timedelta
from functools import partial
from typing import Any, Callable, Dict, Iterator, List, Optional, Union
from typing import Any, Callable, Dict, Iterator, List, Optional, Union, Tuple
import requests
import requests.utils
from Crypto import Random
from Crypto.Cipher import AES
from nacl.public import PublicKey, SealedBox # type:ignore
from .exceptions import *
@@ -24,16 +28,16 @@ def copy_session(session: requests.Session, request_timeout: Optional[float] = N
"""Duplicates a requests.Session."""
new = requests.Session()
new.cookies = requests.utils.cookiejar_from_dict(requests.utils.dict_from_cookiejar(session.cookies))
new.headers = session.headers.copy() # type: ignore
new.headers = session.headers.copy()
# Override default timeout behavior.
# Need to silence mypy bug for this. See: https://github.com/python/mypy/issues/2427
new.request = partial(new.request, timeout=request_timeout) # type: ignore
new.request = partial(new.request, timeout=request_timeout) # type: ignore
return new
def default_user_agent() -> str:
return 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 ' \
'(KHTML, like Gecko) Chrome/92.0.4515.131 Safari/537.36'
'(KHTML, like Gecko) Chrome/89.0.4389.90 Safari/537.36'
class InstaloaderContext:
@@ -216,9 +220,7 @@ class InstaloaderContext:
session.headers.update({'X-CSRFToken': csrf_token})
# Not using self.get_json() here, because we need to access csrftoken cookie
self.do_sleep()
# Workaround credits to pgrimaud.
# See: https://github.com/pgrimaud/instagram-user-feed/commit/96ad4cf54d1ad331b337f325c73e664999a6d066
enc_password = '#PWD_INSTAGRAM_BROWSER:0:{}:{}'.format(int(datetime.now().timestamp()), passwd)
enc_password = self._get_encrypted_password(passwd)
login = session.post('https://www.instagram.com/accounts/login/ajax/',
data={'enc_password': enc_password, 'username': user}, allow_redirects=True)
try:
@@ -331,8 +333,7 @@ class InstaloaderContext:
while resp.is_redirect:
redirect_url = resp.headers['location']
self.log('\nHTTP redirect from https://{0}/{1} to {2}'.format(host, path, redirect_url))
if (redirect_url.startswith('https://www.instagram.com/accounts/login') or
redirect_url.startswith('https://i.instagram.com/accounts/login')):
if redirect_url.startswith('https://www.instagram.com/accounts/login'):
if not self.is_logged_in:
raise LoginRequiredException("Redirected to login page. Use --login.")
# alternate rate limit exceeded behavior
@@ -563,6 +564,39 @@ class InstaloaderContext:
self._root_rhx_gis = self.get_json('', {}).get('rhx_gis', '')
return self._root_rhx_gis or None
def _get_encryption_data(self) -> Tuple[Optional[str], Optional[str], Optional[str]]:
resp = self._session.get("https://www.instagram.com/accounts/login/")
return (resp.headers.get("ig-set-password-encryption-web-key-id"),
resp.headers.get("ig-set-password-encryption-web-key-version"),
resp.headers.get("ig-set-password-encryption-web-pub-key"))
def _get_encrypted_password(self, password):
# contributed by @twitter-79 in
# https://github.com/instaloader/instaloader/issues/615#issuecomment-630631930
key_id, app_id, public_key = self._get_encryption_data()
timestamp = str(int(datetime.now().timestamp()))
# create a random key of length 32 bytes (for AES 256)
key = Random.get_random_bytes(32)
# create a buffer of length 12 bytes filled with 0
iv = bytearray(12)
aes = AES.new(key, AES.MODE_GCM, nonce=iv)
aes.update(bytearray(timestamp, 'utf-8'))
ciphertext, tag = aes.encrypt_and_digest(bytearray(password, 'utf-8'))
# get a byte array of the given public key
public_key_seal = PublicKey(bytes.fromhex(public_key))
sealed_box = SealedBox(public_key_seal)
sealed = sealed_box.encrypt(key)
enc_password = bytearray()
enc_password += bytearray(
[1, int(key_id), len(sealed) & 255, (len(sealed) >> 8) & 255]) + sealed + tag + ciphertext
return ('#PWD_INSTAGRAM_BROWSER' + ':' + app_id + ':' + timestamp + ':' +
str(base64.b64encode(enc_password), 'utf-8'))
class RateController:
"""
@@ -590,6 +624,7 @@ class RateController:
"""Wait given number of seconds."""
# Not static, to allow for the behavior of this method to depend on context-inherent properties, such as
# whether we are logged in.
# pylint:disable=no-self-use
time.sleep(secs)
def _dump_query_timestamps(self, current_time: float, failed_query_type: str):
@@ -613,6 +648,7 @@ class RateController:
control."""
# Not static, to allow for the count_per_sliding_window to depend on context-inherent properties, such as
# whether we are logged in.
# pylint:disable=no-self-use
return 75 if query_type == 'other' else 200
def _reqs_in_sliding_window(self, query_type: Optional[str], current_time: float, window: float) -> List[float]:
+11 -24
View File
@@ -76,8 +76,7 @@ class NodeIterator(Iterator[T]):
node_wrapper: Callable[[Dict], T],
query_variables: Optional[Dict[str, Any]] = None,
query_referer: Optional[str] = None,
first_data: Optional[Dict[str, Any]] = None,
is_first: Optional[Callable[[T, Optional[T]], bool]] = None):
first_data: Optional[Dict[str, Any]] = None):
self._context = context
self._query_hash = query_hash
self._edge_extractor = edge_extractor
@@ -92,7 +91,6 @@ class NodeIterator(Iterator[T]):
else:
self._data = self._query()
self._first_node: Optional[Dict] = None
self._is_first = is_first
def _query(self, after: Optional[str] = None) -> Dict:
pagination_variables = {'first': NodeIterator._graphql_page_length} # type: Dict[str, Any]
@@ -130,24 +128,19 @@ class NodeIterator(Iterator[T]):
self._page_index, self._total_index = page_index, total_index
raise
item = self._node_wrapper(node)
if self._is_first is not None:
if self._is_first(item, self.first_item):
self._first_node = node
else:
if self._first_node is None:
self._first_node = node
if self._first_node is None:
self._first_node = node
return item
if self._data['page_info']['has_next_page']:
query_response = self._query(self._data['page_info']['end_cursor'])
if self._data['edges'] != query_response['edges'] and len(query_response['edges']) > 0:
page_index, data = self._page_index, self._data
try:
self._page_index = 0
self._data = query_response
except KeyboardInterrupt:
self._page_index, self._data = page_index, data
raise
return self.__next__()
page_index, data = self._page_index, self._data
try:
self._page_index = 0
self._data = query_response
except KeyboardInterrupt:
self._page_index, self._data = page_index, data
raise
return self.__next__()
raise StopIteration()
@property
@@ -174,13 +167,7 @@ class NodeIterator(Iterator[T]):
"""
If this iterator has produced any items, returns the first item produced.
It is possible to override what is considered the first item (for example, to consider the
newest item in case items are not in strict chronological order) by passing a callback
function as the `is_first` parameter when creating the class.
.. versionadded:: 4.8
.. versionchanged:: 4.9.2
What is considered the first item can be overridden.
"""
return self._node_wrapper(self._first_node) if self._first_node is not None else None
-46
View File
@@ -1,46 +0,0 @@
from typing import Any, Callable, Dict, Iterator, Optional, TypeVar
from .instaloadercontext import InstaloaderContext
T = TypeVar('T')
class SectionIterator(Iterator[T]):
"""Iterator for the new 'sections'-style responses.
.. versionadded:: 4.9"""
def __init__(self,
context: InstaloaderContext,
sections_extractor: Callable[[Dict[str, Any]], Dict[str, Any]],
media_wrapper: Callable[[Dict], T],
query_path: str,
first_data: Optional[Dict[str, Any]] = None):
self._context = context
self._sections_extractor = sections_extractor
self._media_wrapper = media_wrapper
self._query_path = query_path
self._data = first_data or self._query()
self._page_index = 0
self._section_index = 0
def __iter__(self):
return self
def _query(self, max_id: Optional[str] = None) -> Dict[str, Any]:
pagination_variables = {"max_id": max_id} if max_id is not None else {}
return self._sections_extractor(
self._context.get_json(self._query_path, params={"__a": 1, "__d": "dis", **pagination_variables})
)
def __next__(self) -> T:
if self._page_index < len(self._data['sections']):
media = self._data['sections'][self._page_index]['layout_content']['medias'][self._section_index]['media']
self._section_index += 1
if self._section_index >= len(self._data['sections'][self._page_index]['layout_content']['medias']):
self._section_index = 0
self._page_index += 1
return self._media_wrapper(media)
if self._data['more_available']:
self._page_index, self._section_index, self._data = 0, 0, self._query(self._data["next_max_id"])
return self.__next__()
raise StopIteration()
+72 -271
View File
@@ -3,18 +3,14 @@ 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 unicodedata import normalize
from typing import Any, Dict, Iterable, Iterator, List, Optional, Tuple, Union
from . import __version__
from .exceptions import *
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."
@@ -39,8 +35,8 @@ PostLocation.id.__doc__ = "ID number of location."
PostLocation.name.__doc__ = "Location name."
PostLocation.slug.__doc__ = "URL friendly variant of location name."
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)."
PostLocation.lat.__doc__ = "Latitude (:class:`float`)."
PostLocation.lng.__doc__ = "Longitude (:class:`float`)."
class Post:
@@ -93,41 +89,6 @@ class Post:
"""Create a post object from a given mediaid"""
return cls.from_shortcode(context, Post.mediaid_to_shortcode(mediaid))
@classmethod
def from_iphone_struct(cls, context: InstaloaderContext, media: Dict[str, Any]):
"""Create a post from a given iphone_struct.
.. versionadded:: 4.9"""
media_types = {
1: "GraphImage",
2: "GraphVideo",
8: "GraphSidecar",
}
fake_node = {
"shortcode": media["code"],
"id": media["pk"],
"__typename": media_types[media["media_type"]],
"is_video": media_types[media["media_type"]] == "GraphVideo",
"date": media["taken_at"],
"caption": media["caption"].get("text") if media.get("caption") is not None else None,
"title": media.get("title"),
"viewer_has_liked": media["has_liked"],
"edge_media_preview_like": {"count": media["like_count"]},
"iphone_struct": media,
}
with suppress(KeyError):
fake_node["display_url"] = media['image_versions2']['candidates'][0]['url']
with suppress(KeyError):
fake_node["video_url"] = media['video_versions'][-1]['url']
fake_node["video_duration"] = media["video_duration"]
fake_node["video_view_count"] = media["view_count"]
with suppress(KeyError):
fake_node["edge_sidecar_to_children"] = {"edges": [{"node": {
"display_url": node['image_versions2']['candidates'][0]['url'],
"is_video": media_types[node["media_type"]] == "GraphVideo",
}} for node in media["carousel_media"]]}
return cls(context, fake_node, Profile.from_iphone_struct(context, media["user"]) if "user" in media else None)
@staticmethod
def shortcode_to_mediaid(code: str) -> int:
if len(code) > 11:
@@ -263,16 +224,17 @@ class Post:
@property
def date_local(self) -> datetime:
"""Timestamp when the post was created (local time zone).
.. versionchanged:: 4.9
Return timezone aware datetime object."""
return datetime.fromtimestamp(self._get_timestamp_date_created()).astimezone()
"""Timestamp when the post was created (local time zone)."""
return datetime.fromtimestamp(self._node["date"]
if "date" in self._node
else self._node["taken_at_timestamp"])
@property
def date_utc(self) -> datetime:
"""Timestamp when the post was created (UTC)."""
return datetime.utcfromtimestamp(self._get_timestamp_date_created())
return datetime.utcfromtimestamp(self._node["date"]
if "date" in self._node
else self._node["taken_at_timestamp"])
@property
def date(self) -> datetime:
@@ -293,7 +255,7 @@ class Post:
url = re.sub(r'([?&])se=\d+&?', r'\1', orig_url).rstrip('&')
return url
except (InstaloaderException, KeyError, IndexError) as err:
self._context.error(f"Unable to fetch high quality image version of {self}: {err}")
self._context.error('{} Unable to fetch high quality image version of {}.'.format(err, self))
return self._node["display_url"] if "display_url" in self._node else self._node["display_src"]
@property
@@ -313,12 +275,6 @@ class Post:
return len(edges)
return 1
def _get_timestamp_date_created(self) -> float:
"""Timestamp when the post was created"""
return (self._node["date"]
if "date" in self._node
else self._node["taken_at_timestamp"])
def get_is_videos(self) -> List[bool]:
"""
Return a list containing the ``is_video`` property for each media in the post.
@@ -357,23 +313,18 @@ class Post:
orig_url = carousel_media[idx]['image_versions2']['candidates'][0]['url']
display_url = re.sub(r'([?&])se=\d+&?', r'\1', orig_url).rstrip('&')
except (InstaloaderException, KeyError, IndexError) as err:
self._context.error(f"Unable to fetch high quality image version of {self}: {err}")
self._context.error('{} Unable to fetch high quality image version of {}.'.format(
err, self))
yield PostSidecarNode(is_video=is_video, display_url=display_url,
video_url=node['video_url'] if is_video else None)
@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 self._node["edge_media_to_caption"]["edges"][0]["node"]["text"]
elif "caption" in self._node:
return _normalize(self._node["caption"])
return self._node["caption"]
return None
@property
@@ -395,7 +346,7 @@ class Post:
# 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)
mention_regex = re.compile(r"(?:^|\W|_)(?:@)(\w(?:(?:\w|(?:\.(?!\.))){0,28}(?:\w))?)")
return re.findall(mention_regex, self.caption.lower())
@property
@@ -405,19 +356,9 @@ class Post:
.. versionadded:: 4.2.6"""
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 (pcaption[:30] + u"\u2026") if len(pcaption) > 31 else pcaption
return _elliptify(self.caption) if self.caption else ''
@property
def accessibility_caption(self) -> Optional[str]:
"""Accessibility caption of the post, if available.
.. versionadded:: 4.9"""
try:
return self._field("accessibility_caption")
except KeyError:
return None
@property
def tagged_users(self) -> List[str]:
"""List of all lowercased users that are tagged in the Post."""
@@ -436,23 +377,20 @@ class Post:
def video_url(self) -> Optional[str]:
"""URL of the video, or None."""
if self.is_video:
version_urls = []
try:
version_urls.append(self._field('video_url'))
except (InstaloaderException, KeyError, IndexError) as err:
self._context.error(f"Warning: Unable to fetch video from graphql of {self}: {err}")
version_urls = [self._field('video_url')]
if self._context.iphone_support and self._context.is_logged_in:
try:
version_urls.extend(version['url'] for version in self._iphone_struct['video_versions'])
except (InstaloaderException, KeyError, IndexError) as err:
self._context.error(f"Unable to fetch high-quality video version of {self}: {err}")
version_urls = list(dict.fromkeys(version_urls))
if len(version_urls) == 0:
return None
if len(version_urls) == 1:
return version_urls[0]
else:
return version_urls[0]
url_candidates: List[Tuple[int, str]] = []
for idx, version_url in enumerate(version_urls):
if any(url_candidate[1] == version_url for url_candidate in url_candidates):
# Skip duplicates
continue
try:
url_candidates.append((
int(self._context.head(version_url, allow_redirects=True).headers.get('Content-Length', 0)),
@@ -558,7 +496,7 @@ class Post:
return []
comment_edges = self._field('edge_media_to_comment', 'edges')
answers_count = sum(edge['node'].get('edge_threaded_comments', {}).get('count', 0) for edge in comment_edges)
answers_count = sum([edge['node'].get('edge_threaded_comments', {}).get('count', 0) for edge in comment_edges])
if self.comments == len(comment_edges) + answers_count:
# If the Post's metadata already contains all parent comments, don't do GraphQL requests to obtain them
@@ -638,18 +576,11 @@ class Post:
location_id = int(loc['id'])
if any(k not in loc for k in ('name', 'slug', 'has_public_page', 'lat', 'lng')):
loc.update(self._context.get_json("explore/locations/{0}/".format(location_id),
params={'__a': 1, '__d': 'dis'})['native_location_data']['location_info'])
params={'__a': 1})['native_location_data']['location_info'])
self._location = PostLocation(location_id, loc['name'], loc['slug'], loc['has_public_page'],
loc.get('lat'), loc.get('lng'))
loc['lat'], loc['lng'])
return self._location
@property
def is_pinned(self) -> bool:
"""True if this Post has been pinned by at least one user.
.. versionadded: 4.9.2"""
return 'pinned_for_users' in self._node and bool(self._node['pinned_for_users'])
class Profile:
"""
@@ -731,20 +662,6 @@ class Profile:
context.profile_id_cache[profile_id] = profile
return profile
@classmethod
def from_iphone_struct(cls, context: InstaloaderContext, media: Dict[str, Any]):
"""Create a profile from a given iphone_struct.
.. versionadded:: 4.9"""
return cls(context, {
"id": media["pk"],
"username": media["username"],
"is_private": media["is_private"],
"full_name": media["full_name"],
"profile_pic_url_hd": media["profile_pic_url"],
"iphone_struct": media,
})
@classmethod
def own_profile(cls, context: InstaloaderContext):
"""Return own profile if logged-in.
@@ -770,19 +687,13 @@ class Profile:
def _obtain_metadata(self):
try:
if not self._has_full_metadata:
metadata = self._context.get_iphone_json(f'api/v1/users/web_profile_info/?username={self.username}',
params={})
if metadata['data']['user'] is None:
raise ProfileNotExistsException('Profile {} does not exist.'.format(self.username))
self._node = metadata['data']['user']
metadata = self._context.get_json('{}/feed/'.format(self.username), params={})
self._node = metadata['entry_data']['ProfilePage'][0]['graphql']['user']
self._has_full_metadata = True
except (QueryReturnedNotFoundException, KeyError) as err:
top_search_results = TopSearchResults(self._context, self.username)
similar_profiles = [profile.username for profile in top_search_results.get_profiles()]
if similar_profiles:
if self.username in similar_profiles:
raise ProfileNotExistsException(
f"Profile {self.username} seems to exist, but could not be loaded.") from err
raise ProfileNotExistsException('Profile {} does not exist.\nThe most similar profile{}: {}.'
.format(self.username,
's are' if len(similar_profiles) > 1 else ' is',
@@ -952,18 +863,11 @@ class Profile:
try:
return self._iphone_struct['hd_profile_pic_url_info']['url']
except (InstaloaderException, KeyError) as err:
self._context.error(f"Unable to fetch high quality profile pic: {err}")
self._context.error('{} Unable to fetch high quality profile pic.'.format(err))
return self._metadata("profile_pic_url_hd")
else:
return self._metadata("profile_pic_url_hd")
@property
def profile_pic_url_no_iphone(self) -> str:
"""Return URL of lower-quality profile picture.
.. versionadded:: 4.9.3"""
return self._metadata("profile_pic_url_hd")
def get_profile_pic_url(self) -> str:
""".. deprecated:: 4.0.3
@@ -983,7 +887,6 @@ class Profile:
{'id': self.userid},
'https://www.instagram.com/{0}/'.format(self.username),
self._metadata('edge_owner_to_timeline_media'),
Profile._make_is_newest_checker()
)
def get_saved_posts(self) -> NodeIterator[Post]:
@@ -1017,7 +920,6 @@ class Profile:
lambda n: Post(self._context, n, self if int(n['owner']['id']) == self.userid else None),
{'id': self.userid},
'https://www.instagram.com/{0}/'.format(self.username),
is_first=Profile._make_is_newest_checker()
)
def get_igtv_posts(self) -> NodeIterator[Post]:
@@ -1035,13 +937,8 @@ class Profile:
{'id': self.userid},
'https://www.instagram.com/{0}/channel/'.format(self.username),
self._metadata('edge_felix_video_timeline'),
Profile._make_is_newest_checker()
)
@staticmethod
def _make_is_newest_checker() -> Callable[[Post, Optional[Post]], bool]:
return lambda post, first: first is None or post.date_local > first.date_local
def get_followers(self) -> NodeIterator['Profile']:
"""
Retrieve list of followers of given profile.
@@ -1147,21 +1044,6 @@ class StoryItem:
def __hash__(self) -> int:
return hash(self.mediaid)
@classmethod
def from_mediaid(cls, context: InstaloaderContext, mediaid: int):
"""Create a StoryItem object from a given mediaid.
.. versionadded:: 4.9
"""
pic_json = context.graphql_query(
'2b0673e0dc4580674a88d426fe00ea90',
{'shortcode': Post.mediaid_to_shortcode(mediaid)}
)
shortcode_media = pic_json['data']['shortcode_media']
if shortcode_media is None:
raise BadResponseException("Fetching StoryItem metadata failed.")
return cls(context, shortcode_media)
@property
def _iphone_struct(self) -> Dict[str, Any]:
if not self._context.iphone_support:
@@ -1169,14 +1051,8 @@ class StoryItem:
if not self._context.is_logged_in:
raise LoginRequiredException("--login required to access iPhone media info endpoint.")
if not self._iphone_struct_:
data = self._context.get_iphone_json(
path='api/v1/feed/reels_media/?reel_ids={}'.format(self.owner_id), params={}
)
self._iphone_struct_ = {}
for item in data['reels'][str(self.owner_id)]['items']:
if item['pk'] == self.mediaid:
self._iphone_struct_ = item
break
data = self._context.get_iphone_json(path='api/v1/media/{}/info/'.format(self.mediaid), params={})
self._iphone_struct_ = data['items'][0]
return self._iphone_struct_
@property
@@ -1199,11 +1075,8 @@ class StoryItem:
@property
def date_local(self) -> datetime:
"""Timestamp when the StoryItem was created (local time zone).
.. versionchanged:: 4.9
Return timezone aware datetime object."""
return datetime.fromtimestamp(self._node['taken_at_timestamp']).astimezone()
"""Timestamp when the StoryItem was created (local time zone)."""
return datetime.fromtimestamp(self._node['taken_at_timestamp'])
@property
def date_utc(self) -> datetime:
@@ -1233,14 +1106,13 @@ class StoryItem:
@property
def url(self) -> str:
"""URL of the picture / video thumbnail of the StoryItem"""
if self.typename in ["GraphStoryImage", "StoryImage"] and \
self._context.iphone_support and self._context.is_logged_in:
if self.typename == "GraphStoryImage" and self._context.iphone_support and self._context.is_logged_in:
try:
orig_url = self._iphone_struct['image_versions2']['candidates'][0]['url']
url = re.sub(r'([?&])se=\d+&?', r'\1', orig_url).rstrip('&')
return url
except (InstaloaderException, KeyError, IndexError) as err:
self._context.error(f"Unable to fetch high quality image version of {self}: {err}")
self._context.error('{} Unable to fetch high quality image version of {}.'.format(err, self))
return self._node['display_resources'][-1]['src']
@property
@@ -1257,23 +1129,20 @@ class StoryItem:
def video_url(self) -> Optional[str]:
"""URL of the video, or None."""
if self.is_video:
version_urls = []
try:
version_urls.append(self._node['video_resources'][-1]['src'])
except (InstaloaderException, KeyError, IndexError) as err:
self._context.error(f"Warning: Unable to fetch video from graphql of {self}: {err}")
version_urls = [self._node['video_resources'][-1]['src']]
if self._context.iphone_support and self._context.is_logged_in:
try:
version_urls.extend(version['url'] for version in self._iphone_struct['video_versions'])
except (InstaloaderException, KeyError, IndexError) as err:
self._context.error(f"Unable to fetch high-quality video version of {self}: {err}")
version_urls = list(dict.fromkeys(version_urls))
if len(version_urls) == 0:
return None
if len(version_urls) == 1:
return version_urls[0]
else:
return version_urls[0]
url_candidates: List[Tuple[int, str]] = []
for idx, version_url in enumerate(version_urls):
if any(url_candidate[1] == version_url for url_candidate in url_candidates):
# Skip duplicates
continue
try:
url_candidates.append((
int(self._context.head(version_url, allow_redirects=True).headers.get('Content-Length', 0)),
@@ -1315,7 +1184,6 @@ class Story:
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]]
def __repr__(self):
return '<Story by {} changed {:%Y-%m-%d_%H-%M-%S_UTC}>'.format(self.owner_username, self.latest_media_utc)
@@ -1386,23 +1254,9 @@ class Story:
"""The story owner's ID."""
return self.owner_profile.userid
def _fetch_iphone_struct(self) -> None:
if self._context.iphone_support and self._context.is_logged_in and not self._iphone_struct_:
data = self._context.get_iphone_json(
path='api/v1/feed/reels_media/?reel_ids={}'.format(self.owner_id), params={}
)
self._iphone_struct_ = data['reels'][str(self.owner_id)]
def get_items(self) -> Iterator[StoryItem]:
"""Retrieve all items from a story."""
self._fetch_iphone_struct()
for item in reversed(self._node['items']):
if self._iphone_struct_ is not None:
for iphone_struct_item in self._iphone_struct_['items']:
if iphone_struct_item['pk'] == int(item['id']):
item['iphone_struct'] = iphone_struct_item
break
yield StoryItem(self._context, item, self.owner_profile)
yield from (StoryItem(self._context, item, self.owner_profile) for item in reversed(self._node['items']))
class Highlight(Story):
@@ -1432,7 +1286,6 @@ class Highlight(Story):
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]]
def __repr__(self):
return '<Highlight by {}: {}>'.format(self.owner_username, self.title)
@@ -1471,13 +1324,6 @@ class Highlight(Story):
"highlight_reel_ids": [str(self.unique_id)],
"precomposed_overlay": False})['data']['reels_media'][0]['items']
def _fetch_iphone_struct(self) -> None:
if self._context.iphone_support and self._context.is_logged_in and not self._iphone_struct_:
data = self._context.get_iphone_json(
path='api/v1/feed/reels_media/?reel_ids=highlight:{}'.format(self.unique_id), params={}
)
self._iphone_struct_ = data['reels']['highlight:{}'.format(self.unique_id)]
@property
def itemcount(self) -> int:
"""Count of items associated with the :class:`Highlight` instance."""
@@ -1488,15 +1334,8 @@ class Highlight(Story):
def get_items(self) -> Iterator[StoryItem]:
"""Retrieve all associated highlight items."""
self._fetch_items()
self._fetch_iphone_struct()
assert self._items is not None
for item in self._items:
if self._iphone_struct_ is not None:
for iphone_struct_item in self._iphone_struct_['items']:
if iphone_struct_item['pk'] == int(item['id']):
item['iphone_struct'] = iphone_struct_item
break
yield StoryItem(self._context, item, self.owner_profile)
yield from (StoryItem(self._context, item, self.owner_profile) for item in self._items)
class Hashtag:
@@ -1514,9 +1353,6 @@ class Hashtag:
L.download_post(post, target="#"+hashtag.name)
Also, this class implements == and is hashable.
.. versionchanged:: 4.9
Removed ``get_related_tags()`` and ``is_top_media_only`` as these features were removed from Instagram.
"""
def __init__(self, context: InstaloaderContext, node: Dict[str, Any]):
assert "name" in node
@@ -1545,21 +1381,19 @@ class Hashtag:
return self._node["name"].lower()
def _query(self, params):
json_response = self._context.get_json("explore/tags/{0}/".format(self.name), params)
return json_response["graphql"]["hashtag"] if "graphql" in json_response else json_response["data"]
return self._context.get_json("explore/tags/{0}/".format(self.name),
params)["graphql"]["hashtag"]
def _obtain_metadata(self):
if not self._has_full_metadata:
self._node = self._query({"__a": 1, "__d": "dis"})
self._node = self._query({"__a": 1})
self._has_full_metadata = True
def _asdict(self):
json_node = self._node.copy()
# remove posts
json_node.pop("edge_hashtag_to_top_posts", None)
json_node.pop("top", None)
json_node.pop("edge_hashtag_to_media", None)
json_node.pop("recent", None)
return json_node
def __repr__(self):
@@ -1595,33 +1429,30 @@ class Hashtag:
return self._metadata("profile_pic_url")
@property
def description(self) -> Optional[str]:
def description(self) -> str:
return self._metadata("description")
@property
def allow_following(self) -> bool:
return bool(self._metadata("allow_following"))
return self._metadata("allow_following")
@property
def is_following(self) -> bool:
try:
return self._metadata("is_following")
except KeyError:
return bool(self._metadata("following"))
return self._metadata("is_following")
@property
def is_top_media_only(self) -> bool:
return self._metadata("is_top_media_only")
def get_related_tags(self) -> Iterator["Hashtag"]:
"""Yields similar hashtags."""
yield from (Hashtag(self._context, edge["node"])
for edge in self._metadata("edge_hashtag_to_related_tags", "edges"))
def get_top_posts(self) -> Iterator[Post]:
"""Yields the top posts of the hashtag."""
try:
yield from (Post(self._context, edge["node"])
for edge in self._metadata("edge_hashtag_to_top_posts", "edges"))
except KeyError:
yield from SectionIterator(
self._context,
lambda d: d["data"]["top"],
lambda m: Post.from_iphone_struct(self._context, m),
f"explore/tags/{self.name}/",
self._metadata("top"),
)
yield from (Post(self._context, edge["node"])
for edge in self._metadata("edge_hashtag_to_top_posts", "edges"))
@property
def mediacount(self) -> int:
@@ -1631,38 +1462,23 @@ class Hashtag:
The number of posts with a certain hashtag may differ from the number of posts that can actually be accessed, as
the hashtag count might include private posts
"""
try:
return self._metadata("edge_hashtag_to_media", "count")
except KeyError:
return self._metadata("media_count")
return self._metadata("edge_hashtag_to_media", "count")
def get_posts(self) -> Iterator[Post]:
"""Yields the recent posts associated with this hashtag.
.. deprecated:: 4.9
Use :meth:`Hashtag.get_posts_resumable` as this method may return incorrect results (:issue:`1457`)"""
try:
self._metadata("edge_hashtag_to_media", "edges")
self._metadata("edge_hashtag_to_media", "page_info")
conn = self._metadata("edge_hashtag_to_media")
"""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"])
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"])
except KeyError:
yield from SectionIterator(
self._context,
lambda d: d["data"]["recent"],
lambda m: Post.from_iphone_struct(self._context, m),
f"explore/tags/{self.name}/",
self._metadata("recent"),
)
def get_all_posts(self) -> Iterator[Post]:
"""Yields all posts, i.e. all most recent posts and the top posts, in almost-chronological order."""
sorted_top_posts = iter(sorted(islice(self.get_top_posts(), 9), key=lambda p: p.date_utc, reverse=True))
other_posts = self.get_posts_resumable()
sorted_top_posts = iter(sorted(self.get_top_posts(), key=lambda p: p.date_utc, reverse=True))
other_posts = self.get_posts()
next_top = next(sorted_top_posts, None)
next_other = next(other_posts, None)
while next_top is not None or next_other is not None:
@@ -1688,20 +1504,6 @@ class Hashtag:
yield next_other
next_other = next(other_posts, None)
def get_posts_resumable(self) -> NodeIterator[Post]:
"""Get the recent posts of the hashtag in a resumable fashion.
:rtype: NodeIterator[Post]
.. versionadded:: 4.9"""
return NodeIterator(
self._context, "9b498c08113f1e09617a1703c22b2f32",
lambda d: d['data']['hashtag']['edge_hashtag_to_media'],
lambda n: Post(self._context, n),
{'tag_name': self.name},
f"https://www.instagram.com/explore/tags/{self.name}/"
)
class TopSearchResults:
"""
@@ -1751,7 +1553,7 @@ class TopSearchResults:
place = location.get('place', {})
slug = place.get('slug')
loc = place.get('location', {})
yield PostLocation(int(loc['pk']), loc['name'], slug, None, loc.get('lat'), loc.get('lng'))
yield PostLocation(int(loc['pk']), loc['name'], slug, None, loc['lat'], loc['lng'])
def get_hashtag_strings(self) -> Iterator[str]:
"""
@@ -1905,7 +1707,6 @@ def load_structure_from_file(context: InstaloaderContext, filename: str) -> Json
if compressed:
fp = lzma.open(filename, 'rt')
else:
# pylint:disable=consider-using-with
fp = open(filename, 'rt')
json_structure = json.load(fp)
fp.close()
+6 -6
View File
@@ -18,10 +18,10 @@ def get_version():
raise SystemExit("Could not find version string.")
if sys.version_info < (3, 8):
sys.exit('Instaloader requires Python >= 3.8.')
if sys.version_info < (3, 6):
sys.exit('Instaloader requires Python >= 3.6.')
requirements = ['requests>=2.4']
requirements = ['requests>=2.4', 'pycryptodome~=3.9', 'PyNaCl~=1.4']
keywords = (['instagram', 'instagram-scraper', 'instagram-client', 'instagram-feed', 'downloader', 'videos', 'photos',
'pictures', 'instagram-user-photos', 'instagram-photos', 'instagram-metadata', 'instagram-downloader',
@@ -43,7 +43,7 @@ setup(
'from Instagram.',
long_description=open(os.path.join(SRC, 'README.rst')).read(),
install_requires=requirements,
python_requires='>=3.8',
python_requires='>=3.6',
entry_points={'console_scripts': ['instaloader=instaloader.__main__:main']},
zip_safe=False,
keywords=keywords,
@@ -54,10 +54,10 @@ setup(
'Intended Audience :: Developers',
'License :: OSI Approved :: MIT License',
'Operating System :: OS Independent',
'Programming Language :: Python :: 3.6',
'Programming Language :: Python :: 3.7',
'Programming Language :: Python :: 3.8',
'Programming Language :: Python :: 3.9',
'Programming Language :: Python :: 3.10',
'Programming Language :: Python :: 3.11',
'Programming Language :: Python :: 3 :: Only',
'Topic :: Internet',
'Topic :: Multimedia :: Graphics'