Compare commits

..
Author SHA1 Message Date
Alexander Graf 4022e7fae7 attempt to fix 2FA 2023-10-16 08:44:40 +02:00
sakisdog ac28a33b4d Fixed bug that was asking for 2FA code again and again. 2023-10-16 08:28:51 +02:00
47 changed files with 1789 additions and 2622 deletions
-15
View File
@@ -1,15 +0,0 @@
version: 2
updates:
- package-ecosystem: pip
directory: "/"
schedule:
interval: monthly
cooldown:
default-days: 7
- package-ecosystem: github-actions
directory: "/"
schedule:
interval: monthly
cooldown:
default-days: 7
+4 -10
View File
@@ -2,23 +2,17 @@
<!--
Thanks for your willingness to contribute to Instaloader! Please briefly
describe:
describe
- A motivation for this change, e.g.
- Fixes # .
- What problem does the pull request solve?
- More general: What problem does the pull request solve?
- The changes proposed in this pull request
- Keep the PR focused: avoid bundling independent fixes or refactorings.
- The completeness of this change
- Is it just a proof of concept?
- Is the documentation updated (if appropriate)?
- How has it been tested?
- Do you consider it ready to be merged or is it a draft / proof of concept?
- Do you consider it ready to be merged or is it a draft?
- Can we help you at some point?
- Further considerations
- Is it based on master (bugfix), upcoming/v4.X (new feature), or another pull request?
- Does it preserve backward compatibility and existing behavior?
-->
+5 -7
View File
@@ -8,24 +8,22 @@ jobs:
runs-on: ubuntu-latest
steps:
- name: "Checkout Instaloader Repository"
uses: actions/checkout@v7
uses: actions/checkout@v2
with:
fetch-depth: 0 # needed for building docs
persist-credentials: false
- name: "Setup Python"
uses: actions/setup-python@v7
uses: actions/setup-python@v2
with:
python-version: 3.15
cache: 'pipenv'
python-version: 3.8
- name: "Install Dependencies"
run: |
python -m pip install pipenv==2026.8.0
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
uses: JamesIves/github-pages-deploy-action@v4.2.5
with:
branch: master
repository-name: instaloader/instaloader.github.io
+4 -7
View File
@@ -9,20 +9,17 @@ jobs:
strategy:
fail-fast: false
matrix:
python-version: ["3.10", "3.11", "3.12", "3.13", "3.14", "3.15"]
python-version: ["3.8", "3.9", "3.10", "3.11"]
steps:
- name: Checkout Instaloader Repository
uses: actions/checkout@v7
with:
persist-credentials: false
uses: actions/checkout@v2
- name: Setup Python
uses: actions/setup-python@v7
uses: actions/setup-python@v2
with:
python-version: ${{ matrix.python-version }}
cache: 'pipenv'
- name: Install Dependencies
run: |
python -m pip install pipenv==2026.8.0
python -m pip install pipenv==2023.2.4
pipenv --python `python --version | grep -Eo '3\.[0-9]+'` sync --dev
- name: PyLint
run: pipenv --python `python --version | grep -Eo '3\.[0-9]+'` run pylint instaloader
+6 -10
View File
@@ -11,25 +11,21 @@ jobs:
runs-on: ubuntu-latest
steps:
- name: "Checkout Repository"
uses: actions/checkout@v7
with:
persist-credentials: false
uses: actions/checkout@v2
- name: "Setup Python"
uses: actions/setup-python@v7
uses: actions/setup-python@v2
with:
python-version: 3.15
python-version: 3.8
- name: "Get the tagged version"
id: get_version
env:
GITHUB_REF: ${{ github.ref }}
run: echo "VERSION=${GITHUB_REF/refs\/tags\/v/}" >> $GITHUB_OUTPUT
run: echo ::set-output name=VERSION::${GITHUB_REF/refs\/tags\/v/}
shell: bash
- name: "Install Dependencies"
run: python -m pip install -U setuptools wheel build
- name: "Generate Python Package"
run: python -m build
run: python setup.py sdist
- name: "Publish Distribution to PyPI"
uses: pypa/gh-action-pypi-publish@release/v1
uses: pypa/gh-action-pypi-publish@master
with:
password: ${{ secrets.PYPI_PASSWORD }}
- name: "Publish Distribution to AUR"
+5 -5
View File
@@ -2,7 +2,7 @@ name: Mark stale issues and pull requests
on:
schedule:
- cron: "55 7 * * *"
- cron: "10 14 * * *"
jobs:
stale:
@@ -11,7 +11,7 @@ jobs:
runs-on: ubuntu-latest
steps:
- uses: actions/stale@v11
- uses: actions/stale@v3
with:
repo-token: ${{ secrets.GITHUB_TOKEN }}
only-labels: 'question'
@@ -21,14 +21,14 @@ jobs:
days-before-stale: 21
days-before-close: -1
remove-stale-when-updated: false
- uses: actions/stale@v11
- uses: actions/stale@v1
with:
repo-token: ${{ secrets.GITHUB_TOKEN }}
stale-issue-message: 'There has been no activity on this issue for an extended period of time. This issue will be closed after further 14 days of inactivity.'
stale-pr-message: 'There has been no activity on this Pull Request for an extended period of time. This Pull Request will be closed after further 14 days of inactivity.'
stale-issue-label: 'stale'
stale-pr-label: 'stale'
exempt-issue-labels: 'leave open'
exempt-pr-labels: 'leave open'
exempt-issue-label: 'leave open'
exempt-pr-label: 'leave open'
days-before-stale: 189
days-before-close: 14
+18 -11
View File
@@ -8,34 +8,41 @@ jobs:
runs-on: windows-latest
steps:
- name: Checkout Instaloader repository
uses: actions/checkout@v7
with:
persist-credentials: false
uses: actions/checkout@v2
- name: Setup Python
uses: actions/setup-python@v7
uses: actions/setup-python@v2
with:
python-version: 3.15
cache: 'pipenv'
python-version: "3.9"
architecture: x64
- name: Get the tagged version
id: get_version
env:
GITHUB_REF: ${{ github.ref }}
run: echo "VERSION=${GITHUB_REF/refs\/tags\/v/}" >> $GITHUB_OUTPUT
run: echo ::set-output name=VERSION::${GITHUB_REF/refs\/tags\/v/}
shell: bash
- name: Run custom python script for EXE creation
env:
VERSION_TAG: v${{ steps.get_version.outputs.VERSION }}
run: python deploy/windows/create_exe.py
- name: Create draft release and upload EXE as release asset
- name: Create draft release
if: github.event_name == 'push' && startsWith(github.ref, 'refs/tags/v')
id: create_release
uses: softprops/action-gh-release@v3
uses: actions/create-release@v1
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
with:
tag_name: v${{ steps.get_version.outputs.VERSION }}
name: Version ${{ steps.get_version.outputs.VERSION }}
release_name: Version ${{ steps.get_version.outputs.VERSION }}
draft: true
prerelease: contains(steps.get_version.outputs.VERSION, 'a') || contains(steps.get_version.outputs.VERSION, 'b') || contains(steps.get_version.outputs.VERSION, 'c')
files: ./instaloader-v${{ steps.get_version.outputs.VERSION }}-windows-standalone.zip
- name: Upload EXE as release asset
if: github.event_name == 'push' && startsWith(github.ref, 'refs/tags/v')
id: upload-release-asset
uses: actions/upload-release-asset@v1
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
with:
upload_url: ${{ steps.create_release.outputs.upload_url }}
asset_path: ./instaloader-v${{ steps.get_version.outputs.VERSION }}-windows-standalone.zip
asset_name: instaloader-v${{ steps.get_version.outputs.VERSION }}-windows-standalone.zip
asset_content_type: application/zip
+1 -2
View File
@@ -351,7 +351,7 @@ single-line-if-stmt=no
allow-any-import-level=
# Allow explicit reexports by alias from a package __init__.
allow-reexport-from-package=yes
allow-reexport-from-package=no
# Allow wildcard imports from modules that define __all__.
allow-wildcard-with-all=no
@@ -433,7 +433,6 @@ disable=bad-inline-option,
too-many-instance-attributes,
too-many-lines,
too-many-locals,
too-many-positional-arguments,
too-many-public-methods,
too-many-statements,
unnecessary-lambda-assignment,
Binary file not shown.

Before

Width:  |  Height:  |  Size: 2.9 KiB

-1
View File
@@ -6,5 +6,4 @@ Instaloader is written by
- Alexander Graf (@aandergr)
- André Koch-Kramer (@Thammus)
- Lars Lindqvist (@e5150)
- Eduardo M Kalinowksi (@ekalin)
- ... and many more, see https://github.com/instaloader/instaloader/graphs/contributors
+9 -8
View File
@@ -6,17 +6,18 @@ verify_ssl = true
[dev-packages]
pylint = "*"
mypy = "*"
sphinx = "*"
sphinx = "~=2.3"
pyinstaller = "*"
pefile = "*"
pywin32-ctypes = "*"
psutil = "*"
typing-extensions = "*"
types-requests = "*"
sphinx-autodoc-typehints = "*"
sphinxcontrib-jquery = "*"
tomli = "*" # Needed for Pylint for "python_version < '3.11'"
dill = "*" # Needed for Pylint for "python_version < '3.11'"
pywin32-ctypes = "*" # Needed for pyinstaller on Windows
pefile = "*" # Needed for pyinstaller on Windows
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 = "*"
browser-cookie3 = "*"
Generated
+578 -1205
View File
File diff suppressed because it is too large Load Diff
+3 -21
View File
@@ -53,7 +53,7 @@
::
instaloader [--comments] [--geotags]
[--stories] [--highlights] [--tagged] [--reels] [--igtv]
[--stories] [--highlights] [--tagged] [--igtv]
[--login YOUR-USERNAME] [--fast-update]
profile | "#hashtag" | :stories | :feed | :saved
@@ -124,30 +124,12 @@ Supporters
.. current-sponsors-start
| Instaloader is proudly sponsored by
.. list-table::
:align: center
* - |socialapis|
* - `SocialAPIs <https://socialapis.io/?utm_source=github&utm_medium=sponsor&utm_campaign=instaloader>`__
|
.. list-table::
:align: center
* - `@rocketapi-io <https://github.com/rocketapi-io>`__
|
| `@rocketapi-io <https://github.com/rocketapi-io>`__
| `@socialmethod <https://github.com/socialmethod>`__
See `Alex' GitHub Sponsors <https://github.com/sponsors/aandergr>`__ page for
how you can sponsor the development of Instaloader!
.. |socialapis| image:: ./.sponsor-logos/socialapis.png
:alt: SocialAPIs
:target: https://socialapis.io/?utm_source=github&utm_medium=sponsor&utm_campaign=instaloader
:height: 96px
.. current-sponsors-end
It is a pleasure for us to share our Instaloader to the world, and we are proud
+2 -6
View File
@@ -5,13 +5,9 @@ pkgbase = instaloader
url = https://instaloader.github.io/
arch = any
license = MIT
makedepends = python-build
makedepends = python-installer
makedepends = python-setuptools
makedepends = python-wheel
depends = python
depends = python-requests
optdepends = python-browser-cookie3: Import session cookies from browser
depends = python>=3.8
depends = python-requests>=2.4
options = !emptydirs
source = instaloader-{{version}}.tar.gz::https://codeload.github.com/instaloader/instaloader/tar.gz/v{{version}}
sha512sums = {{hash}}
+5 -15
View File
@@ -8,30 +8,20 @@ arch=('any')
url="https://instaloader.github.io/"
license=('MIT')
groups=()
depends=(
python
python-requests
)
makedepends=(
python-build
python-installer
python-setuptools
python-wheel
)
optdepends=(
'python-browser-cookie3: Import session cookies from browser'
)
depends=('python>=3.8' 'python-requests>=2.4')
makedepends=('python-setuptools')
options=('!emptydirs')
source=($pkgname-$pkgver.tar.gz::https://codeload.github.com/instaloader/instaloader/tar.gz/v$pkgver)
sha512sums=('{{hash}}')
build() {
cd "$srcdir/$pkgname-$pkgver"
python -m build --wheel --no-isolation
python setup.py clean --all
python setup.py build
}
package() {
cd "$srcdir/$pkgname-$pkgver"
install -Dm644 LICENSE "$pkgdir"/usr/share/licenses/$pkgname/LICENSE
python -m installer --destdir="$pkgdir" dist/*.whl
python setup.py install --root "$pkgdir/" --skip-build
}
+1 -1
View File
@@ -44,7 +44,7 @@ with open('__main__.py', 'w+') as f:
f.writelines(lines)
# install dependencies and invoke PyInstaller
commands = ["pip install pipenv==2026.8.0",
commands = ["pip install pipenv==2023.2.4",
f"pipenv --python {sys.version_info.major}.{sys.version_info.minor} sync --dev",
"pipenv run pyinstaller --log-level=DEBUG instaloader.spec"]
-1
View File
@@ -1 +0,0 @@
../.sponsor-logos
+3 -4
View File
@@ -7,10 +7,9 @@ https://instaloader.github.io/
## Build
The documentation is created with [Sphinx](http://www.sphinx-doc.org/). To build it, use (in the top-level directory
of the Instaloader source tree):
The documentation is created with [Sphinx](http://www.sphinx-doc.org/). To build it, use
```
pipenv sync --dev
make -C docs html
pip3 install -r requirements.txt
make html
```
+1 -1
View File
File diff suppressed because one or more lines are too long
+1 -1
View File
@@ -21,7 +21,7 @@
{% block header %}
<nav class="navbar navbar-light navbar-expand bg-light border-bottom">
<a class="navbar-brand" href="{{ pathto(master_doc) }}">
<img src="{{ pathto('_static/logo.png', 1) }}" width="30" height="30" class="d-inline-block align-top" alt="logo">
<img src="{{ pathto('_static/' + logo, 1) }}" width="30" height="30" class="d-inline-block align-top" alt="logo">
<span class="ml-3">Instaloader</span>
</a>
<ul class="navbar-nav mr-auto">
+12 -50
View File
@@ -75,7 +75,7 @@ What to Download
Instaloader supports the following targets:
- ``profile``
Public profile, or private profile with :ref:`login<login>`.
Public profile, or private profile with :option:`--login`.
If an already-downloaded profile has been renamed, Instaloader automatically
finds it by its unique ID and renames the folder accordingly.
@@ -92,41 +92,37 @@ Instaloader supports the following targets:
- :option:`--tagged`
to **download posts where the user is tagged**, and
- :option:`--reels`
to **download Reels videos**.
- :option:`--igtv`
to **download IGTV videos**.
- ``"#hashtag"``
Posts with a certain **hashtag** (the quotes are usually necessary). Requires :ref:`login<login>`.
Posts with a certain **hashtag** (the quotes are usually necessary).
- ``%location id``
Posts tagged with a given location; the location ID is the numerical ID
Instagram labels a location with (e.g.
\https://www.instagram.com/explore/locations/**362629379**/plymouth-naval-memorial/).
Requires :ref:`login<login>`.
Requires :option:`--login`.
.. versionadded:: 4.2
- ``:stories``
The currently-visible **stories** of your followees (requires
:ref:`login<login>`).
:option:`--login`).
- ``:feed``
Your **feed** (requires :ref:`login<login>`).
Your **feed** (requires :option:`--login`).
- ``:saved``
Posts which are marked as **saved** (requires :ref:`login<login>`).
Posts which are marked as **saved** (requires :option:`--login`).
- ``@profile``
All profiles that are followed by ``profile``, i.e. the *followees* of
``profile`` (requires :ref:`login<login>`).
``profile`` (requires :option:`--login`).
- ``-post``
Replace **post** with the post's shortcode to download single post. Must be preceded by ``--`` in
the argument list to not be mistaken as an option flag. For example, to download the post
https://www.instagram.com/p/**B_K4CykAOtf**, run the command::
the argument list to not be mistaken as an option flag::
instaloader -- -B_K4CykAOtf
@@ -144,7 +140,7 @@ downloads the pictures and videos and their captions. You can specify
- :option:`--geotags`
**download geotags** of each post and save them as
Google Maps link (requires :ref:`login<login>`),
Google Maps link (requires :option:`--login`),
For a reference of all supported command line options, see
:ref:`command-line-options`.
@@ -235,7 +231,7 @@ The filter string must be a
where the attributes from :class:`Post` or
:class:`StoryItem` respectively are defined.
The following attributes can be used with both
Id est, the following attributes can be used with both
:option:`--post-filter` and :option:`--storyitem-filter`:
- :attr:`~Post.owner_username` (str), :attr:`~Post.owner_id` (int)
@@ -259,7 +255,7 @@ The following attributes can be used with both
As :option:`--post-filter`, the following attributes can be used additionally:
- :attr:`~Post.viewer_has_liked` (bool)
Whether user (with :ref:`login<login>`) has liked given post. To download the
Whether user (with :option:`--login`) has liked given post. To download the
pictures from your feed that you have liked::
instaloader --login=your_username --post-filter=viewer_has_liked :feed
@@ -311,47 +307,13 @@ the post's caption::
instaloader --post-metadata-txt="{likes} likes." <target>
Note that with this feature, it is possible to easily and quickly extract
Note that with this feature, it is possible to easily and fastly extract
additional metadata of already-downloaded posts, by reimporting their JSON
files. Say, you now also want to export the number of comments the Posts had
when they were downloaded::
instaloader --post-metadata-txt="{likes} likes, {comments} comments." <target>/*.json.xz
.. _exit_codes:
Exit codes
^^^^^^^^^^
Different exit codes are used to indicate different kinds of error:
0
No error, all downloads were successful.
1
A non-fatal error happened. One or more posts, or even one or more
profiles could not be downloaded, but execution was not stopped. The
errors are repeated at the end of the log for easy access.
2
Command-line error. An unrecognized option was passed, or an invalid
combination of options, for example. No interaction with Instagram
was made.
3
Login error. It was not possible to login. Downloads were not
attempted.
4
Fatal download error. Downloads were interrupted and no further
attempts were made. Happens when a response with one of the status
codes in the :option:`--abort-on` option were passed, or when
Instagram logs the user out during downloads.
5
Interrupted by the user. Happens when the user presses Control-C or
sends SIGINT to the process.
.. _instaloader-as-cronjob:
Instaloader as Cronjob
+6 -37
View File
@@ -26,7 +26,7 @@ Targets
^^^^^^^
Specify a list of targets. For each of these, Instaloader creates a folder and
stores all posts along with the pictures' captions there.
stores all posts along with the pictures's captions there.
.. include:: basic-usage.rst
:start-after: targets-start
@@ -61,13 +61,13 @@ What to Download of each Post
Download geotags when available. Geotags are stored as a text file with
the location's name and a Google Maps link. This requires an additional
request to the Instagram server for each picture. Requires :ref:`login<login>`.
request to the Instagram server for each picture. Requires :option:`--login`.
.. option:: --comments, -C
Download and update comments for each post. This requires an additional
request to the Instagram server for each post, which is why it is disabled by
default. Requires :ref:`login<login>`.
default.
.. option:: --no-captions
@@ -116,12 +116,12 @@ What to Download of each Profile
.. option:: --stories, -s
Also download stories of each profile that is downloaded. Requires
:ref:`login<login>`.
:option:`--login`.
.. option:: --highlights
Also download highlights of each profile that is downloaded. Requires
:ref:`login<login>`.
:option:`--login`.
.. versionadded:: 4.1
@@ -131,12 +131,6 @@ What to Download of each Profile
.. versionadded:: 4.1
.. option:: --reels
Also download Reels videos.
.. versionadded:: 4.14
.. option:: --igtv
Also download IGTV videos.
@@ -189,7 +183,6 @@ Which Posts to Download
Do not attempt to download more than COUNT posts. Applies to
``#hashtag``, ``%location_id``, ``:feed``, and ``:saved``.
.. _login:
Login (Download Private Profiles)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
@@ -199,34 +192,10 @@ profiles. To login, pass the :option:`--login` option. Your session cookie (not
password!) will be saved to a local file to be reused next time you want
Instaloader to login.
Instead of :option:`--login`, it is possible to use
:option:`--load-cookies` to import a session from a browser.
.. option:: --login YOUR-USERNAME, -l YOUR-USERNAME
Login name (profile name) for your Instagram account.
.. option:: --load-cookies BROWSER-NAME, -b BROWSER-NAME
Load Instagram session cookies from a browser instead of using Instaloader's own login mechanism.
This method requires the `browser_cookie3 <https://github.com/borisbabic/browser_cookie3>`__ library.
By default, cookies are loaded from the browser's main profile, but you can specify a custom cookie file using :option:`--cookiefile`.
Since :option:`--load-cookies` and :option:`--login` are alternative ways to acquire a session, they cannot be used together when initially acquiring a session.
Supported browsers: Arc, Brave, Chrome, Chromium, Edge, Firefox, LibreWolf, Opera, Opera_GX, Safari, and Vivaldi.
Once logged in via browser cookies, Instaloader saves the session, allowing you to use :option:`--login` in subsequent runs.
.. versionadded:: 4.11
.. option:: --cookiefile COOKIE-FILE, -B COOKIE-FILE
Specifies a cookie file path for :option:`--load-cookies` to load cookies from a particular browser profile.
This option must be used together with :option:`--load-cookies` and does not work on its own.
.. versionadded:: 4.11
.. option:: --sessionfile SESSIONFILE, -f SESSIONFILE
Path for loading and storing session key file. Defaults to
@@ -312,7 +281,7 @@ How to Download
.. option:: --user-agent USER_AGENT
User Agent to use for HTTP requests. Per default, Instaloader pretends being
Chrome/142 on Linux.
Chrome/92 on Linux.
.. option:: --max-connection-attempts N
+1 -1
View File
@@ -2,7 +2,7 @@ import instaloader
L = instaloader.Instaloader()
posts = instaloader.Hashtag.from_name(L.context, 'urbanphotography').get_posts_resumable()
posts = instaloader.Hashtag.from_name(L.context, 'urbanphotography').get_posts()
users = set()
@@ -14,13 +14,8 @@ def get_cookiefile():
default_cookiefile = {
"Windows": "~/AppData/Roaming/Mozilla/Firefox/Profiles/*/cookies.sqlite",
"Darwin": "~/Library/Application Support/Firefox/Profiles/*/cookies.sqlite",
}.get(system(), None)
if default_cookiefile:
cookiefiles = glob(expanduser(default_cookiefile))
else:
# Check Firefox 147+ XDG path, then legacy
cookiefiles = glob(expanduser("~/.config/mozilla/firefox/*/cookies.sqlite")) or \
glob(expanduser("~/.mozilla/firefox/*/cookies.sqlite"))
}.get(system(), "~/.mozilla/firefox/*/cookies.sqlite")
cookiefiles = glob(expanduser(default_cookiefile))
if not cookiefiles:
raise SystemExit("No Firefox cookies.sqlite file found. Use -c COOKIEFILE.")
return cookiefiles[0]
@@ -3,7 +3,7 @@ import instaloader
L = instaloader.Instaloader()
posts = instaloader.Hashtag.from_name(L.context, "urbanphotography").get_posts_resumable()
posts = instaloader.Hashtag.from_name(L.context, "urbanphotography").get_posts()
SINCE = datetime(2020, 5, 10) # further from today, inclusive
UNTIL = datetime(2020, 5, 11) # closer to today, not inclusive
+354 -63
View File
@@ -1,84 +1,56 @@
# Configuration file for the Sphinx documentation builder.
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# For the full list of built-in configuration values, see the documentation:
# https://www.sphinx-doc.org/en/master/usage/configuration.html
# Instaloader documentation build configuration file, created by
# sphinx-quickstart on Fri Aug 25 15:38:26 2017.
#
# This file is execfile()d with the current directory set to its
# containing dir.
#
# Note that not all possible configuration values are present in this
# autogenerated file.
#
# All configuration values have a default; values that are commented out
# serve to show the default.
# If extensions (or modules to document with autodoc) are in another directory,
# add these directories to sys.path here. If the directory is relative to the
# documentation root, use os.path.abspath to make it absolute, like shown here.
#
import os
import platform
import subprocess
import sys
import typing
sys.path.insert(0, os.path.abspath('..'))
import docs.sphinx_autodoc_typehints as sphinx_autodoc_typehints
# -- Project information -----------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#project-information
# -- General configuration ------------------------------------------------
project = 'Instaloader'
copyright = '2023, Alexander Graf and Andre Koch-Kramer'
author = 'Alexander Graf and Andre Koch-Kramer'
# -- General configuration ---------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration
# If your documentation needs a minimal Sphinx version, state it here.
#
# needs_sphinx = '1.0'
# Add any Sphinx extension module names here, as strings. They can be
# extensions coming with Sphinx (named 'sphinx.ext.*') or your custom
# ones.
extensions = [
"sphinx.ext.autodoc",
"sphinx.ext.extlinks",
"sphinx.ext.githubpages",
"sphinx.ext.intersphinx",
"sphinxcontrib.jquery",
'sphinx.ext.autodoc',
'sphinx.ext.githubpages',
'sphinx.ext.intersphinx',
'sphinx.ext.extlinks'
]
# autodoc options
autodoc_default_flags = ['show-inheritance', 'members', 'undoc-members']
autodoc_default_options = {
'show-inheritance': True,
'members': True,
'undoc-members': True,
}
autodoc_member_order = 'bysource'
# extlinks options
intersphinx_mapping = {'python': ('https://docs.python.org/3', None),
'requests': ('https://requests.readthedocs.io/en/latest/', None)}
extlinks = {
'issue': (
'https://github.com/instaloader/instaloader/issues/%s',
'Issue #%s'
),
'example': (
'https://raw.githubusercontent.com/instaloader/instaloader/master/docs/codesnippets/%s',
'Example %s'
),
}
# intersphinx options
intersphinx_mapping = {
'python': ('https://docs.python.org/3', None),
'requests': ('https://requests.readthedocs.io/en/latest/', None),
}
# other options
templates_path = ['_templates']
exclude_patterns = ['_build', 'Thumbs.db', '.DS_Store']
# ignore TypeVar as it is currently unsupported by Sphinx
nitpick_ignore = [('py:class', 'instaloader.nodeiterator.T')]
# -- Options for HTML output -------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#options-for-html-output
pygments_style = 'fruity'
html_theme = 'basic'
html_static_path = ['_static']
html_logo = "logo.png"
html_css_files = ['instaloaderdoc.css']
html_js_files = ['instaloaderdoc.js']
# get release number
nitpick_ignore = [('py:class', 'typing.Tuple')]
current_release = subprocess.check_output(["git", "describe", "--abbrev=0"]).decode("ascii")[1:-1]
date_format = "%e %b %Y" if platform.system() != "Windows" else "%d %b %Y"
@@ -86,3 +58,322 @@ current_release_date = subprocess.check_output(
["git", "log", "-1", "--tags", "--format=%ad", "--date=format:"+date_format]).decode("ascii")[:-1]
html_context = {'current_release': current_release, 'current_release_date': current_release_date}
extlinks = {
'issue': (
'https://github.com/instaloader/instaloader/issues/%s',
'Issue #'
),
'example': (
'https://raw.githubusercontent.com/instaloader/instaloader/master/docs/codesnippets/%s',
'Example '
),
}
# Add any paths that contain templates here, relative to this directory.
templates_path = ['_templates']
# The suffix(es) of source filenames.
# You can specify multiple suffix as a list of string:
#
# source_suffix = ['.rst', '.md']
source_suffix = '.rst'
# The encoding of source files.
#
# source_encoding = 'utf-8-sig'
# The master toctree document.
master_doc = 'index'
# General information about the project.
project = 'Instaloader'
copyright = '2017-2020, Alexander Graf and Andre Koch-Kramer'
author = 'Alexander Graf and Andre Koch-Kramer'
# The version info for the project you're documenting, acts as replacement for
# |version| and |release|, also used in various other places throughout the
# built documents.
#
# The short X.Y version.
#version = '3.0'
# The full version, including alpha/beta/rc tags.
#release = '3.0rc0'
# The language for content autogenerated by Sphinx. Refer to documentation
# for a list of supported languages.
#
# This is also used if you do content translation via gettext catalogs.
# Usually you set "language" from the command line for these cases.
language = "en"
# There are two options for replacing |today|: either, you set today to some
# non-false value, then it is used:
#
# today = ''
#
# Else, today_fmt is used as the format for a strftime call.
#
# today_fmt = '%B %d, %Y'
# List of patterns, relative to source directory, that match files and
# directories to ignore when looking for source files.
# This patterns also effect to html_static_path and html_extra_path
exclude_patterns = ['_build', 'Thumbs.db', '.DS_Store']
# The reST default role (used for this markup: `text`) to use for all
# documents.
#
# default_role = None
# If true, '()' will be appended to :func: etc. cross-reference text.
#
# add_function_parentheses = True
# If true, the current module name will be prepended to all description
# unit titles (such as .. function::).
#
add_module_names = False
# If true, sectionauthor and moduleauthor directives will be shown in the
# output. They are ignored by default.
#
# show_authors = False
# The name of the Pygments (syntax highlighting) style to use.
pygments_style = 'fruity'
# A list of ignored prefixes for module index sorting.
# modindex_common_prefix = []
# If true, keep warnings as "system message" paragraphs in the built documents.
# keep_warnings = False
# If true, `todo` and `todoList` produce output, else they produce nothing.
todo_include_todos = False
# -- Options for HTML output ----------------------------------------------
# The theme to use for HTML and HTML Help pages. See the documentation for
# a list of builtin themes.
#
html_theme = 'basic'
# Add any paths that contain custom themes here, relative to this directory.
# html_theme_path = []
# The name for this set of Sphinx documents.
# "<project> v<release> documentation" by default.
#
# html_title = 'Instaloader v3.0rc0'
# A shorter title for the navigation bar. Default is the same as html_title.
#
# html_short_title = None
# The name of an image file (relative to this directory) to place at the top
# of the sidebar.
#
html_logo = "logo.png"
# The name of an image file (relative to this directory) to use as a favicon of
# the docs. This file should be a Windows icon file (.ico) being 16x16 or 32x32
# pixels large.
#
html_favicon = "favicon.ico"
# Add any paths that contain custom static files (such as style sheets) here,
# relative to this directory. They are copied after the builtin static files,
# so a file named "default.css" will overwrite the builtin "default.css".
html_static_path = ['_static']
# Add any extra paths that contain custom files (such as robots.txt or
# .htaccess) here, relative to this directory. These files are copied
# directly to the root of the documentation.
#
# html_extra_path = []
# If not None, a 'Last updated on:' timestamp is inserted at every page
# bottom, using the given strftime format.
# The empty string is equivalent to '%b %d, %Y'.
#
# html_last_updated_fmt = None
# If true, SmartyPants will be used to convert quotes and dashes to
# typographically correct entities.
#
# html_use_smartypants = True
# Custom sidebar templates, maps document names to template names.
#
#html_sidebars = {'**': ["relations.html", "links.html"] }
html_sidebars = {}
# Additional templates that should be rendered to pages, maps page names to
# template names.
#
# html_additional_pages = {}
# If false, no module index is generated.
#
html_domain_indices = False
# If false, no index is generated.
#
html_use_index = True
# If true, the index is split into individual pages for each letter.
#
# html_split_index = False
# If true, links to the reST sources are added to the pages.
#
html_show_sourcelink = False
# If true, "Created using Sphinx" is shown in the HTML footer. Default is True.
#
html_show_sphinx = False
# If true, "(C) Copyright ..." is shown in the HTML footer. Default is True.
#
html_show_copyright = False
# If true, an OpenSearch description file will be output, and all pages will
# contain a <link> tag referring to it. The value of this option must be the
# base URL from which the finished HTML is served.
#
# html_use_opensearch = ''
# This is the file name suffix for HTML files (e.g. ".xhtml").
# html_file_suffix = None
# Language to be used for generating the HTML full-text search index.
# Sphinx supports the following languages:
# 'da', 'de', 'en', 'es', 'fi', 'fr', 'h', 'it', 'ja'
# 'nl', 'no', 'pt', 'ro', 'r', 'sv', 'tr', 'zh'
#
# html_search_language = 'en'
# A dictionary with options for the search language support, empty by default.
# 'ja' uses this config value.
# 'zh' user can custom change `jieba` dictionary path.
#
# html_search_options = {'type': 'default'}
# The name of a javascript file (relative to the configuration directory) that
# implements a search results scorer. If empty, the default will be used.
#
# html_search_scorer = 'scorer.js'
# Output file base name for HTML help builder.
htmlhelp_basename = 'Instaloaderdoc'
# -- Options for LaTeX output ---------------------------------------------
latex_elements = {
# The paper size ('letterpaper' or 'a4paper').
#
# 'papersize': 'letterpaper',
# The font size ('10pt', '11pt' or '12pt').
#
# 'pointsize': '10pt',
# Additional stuff for the LaTeX preamble.
#
# 'preamble': '',
# Latex figure (float) alignment
#
# 'figure_align': 'htbp',
}
# Grouping the document tree into LaTeX files. List of tuples
# (source start file, target name, title,
# author, documentclass [howto, manual, or own class]).
latex_documents = [
(master_doc, 'Instaloader.tex', 'Instaloader Documentation',
'Alexander Graf, André Koch-Kramer', 'manual'),
]
# The name of an image file (relative to this directory) to place at the top of
# the title page.
#
# latex_logo = None
# For "manual" documents, if this is true, then toplevel headings are parts,
# not chapters.
#
# latex_use_parts = False
# If true, show page references after internal links.
#
# latex_show_pagerefs = False
# If true, show URL addresses after external links.
#
# latex_show_urls = False
# Documents to append as an appendix to all manuals.
#
# latex_appendices = []
# It false, will not define \strong, \code, itleref, \crossref ... but only
# \sphinxstrong, ..., \sphinxtitleref, ... To help avoid clash with user added
# packages.
#
# latex_keep_old_macro_names = True
# If false, no module index is generated.
#
# latex_domain_indices = True
# -- Options for manual page output ---------------------------------------
# One entry per manual page. List of tuples
# (source start file, name, description, authors, manual section).
man_pages = [
(master_doc, 'instaloader', 'Instaloader Documentation',
[author], 1)
]
# If true, show URL addresses after external links.
#
# man_show_urls = False
# -- Options for Texinfo output -------------------------------------------
# Grouping the document tree into Texinfo files. List of tuples
# (source start file, target name, title, author,
# dir menu entry, description, category)
texinfo_documents = [
(master_doc, 'Instaloader', 'Instaloader Documentation',
author, 'Instaloader', 'One line description of project.',
'Miscellaneous'),
]
# Documents to append as an appendix to all manuals.
#
# texinfo_appendices = []
# If false, no module index is generated.
#
# texinfo_domain_indices = True
# How to display URL addresses: 'footnote', 'no', or 'inline'.
#
# texinfo_show_urls = 'footnote'
# If true, do not generate a @detailmenu in the "Top" node's menu.
#
# texinfo_no_detailmenu = False
def setup(app):
typing.TYPE_CHECKING = True
app.add_stylesheet("instaloaderdoc.css")
app.connect('autodoc-process-signature', sphinx_autodoc_typehints.process_signature)
app.connect('autodoc-process-docstring', sphinx_autodoc_typehints.process_docstring)
-8
View File
@@ -74,14 +74,6 @@ Improvements of the Instaloader source or its documentation can be proposed as a
- a bug fix that does require thorough testing before being released to a
final version.
- Keep Pull Requests focused: address a single problem or feature in a
self-contained change. Avoid bundling multiple independent fixes, refactorings,
or architectural changes into one pull request, as this makes it hard to
review, validate, or selectively merge them.
- Be mindful of backward compatibility: changes should preserve existing intended
behavior whenever possible.
- All Pull Requests are analyzed by `Pylint <https://www.pylint.org/>`__ for
error and syntax checking of the source and
`Mypy <https://github.com/python/mypy>`__ for type checking. You can run them
+1 -1
View File
@@ -48,7 +48,7 @@ See :ref:`install` for more options on how to install Instaloader.
::
instaloader [--comments] [--geotags]
[--stories] [--highlights] [--tagged] [--reels] [--igtv]
[--stories] [--highlights] [--tagged] [--igtv]
[--login YOUR-USERNAME] [--fast-update]
profile | "#hashtag" | %location_id |
:stories | :feed | :saved
+1 -11
View File
@@ -2,7 +2,7 @@ Exceptions
^^^^^^^^^^
.. module:: instaloader
:no-index:
:noindex:
.. highlight:: python
@@ -27,26 +27,16 @@ Exceptions
.. autoexception:: LoginRequiredException
.. autoexception:: LoginException
.. versionadded:: 4.12
.. autoexception:: TwoFactorAuthRequiredException
.. versionadded:: 4.2
.. versionchanged:: 4.12
Inherits LoginException
.. autoexception:: InvalidArgumentException
.. autoexception:: BadResponseException
.. autoexception:: BadCredentialsException
.. versionchanged:: 4.12
Inherits LoginException
.. autoexception:: PostChangedException
.. autoexception:: QueryReturnedNotFoundException
+1 -1
View File
@@ -2,7 +2,7 @@
^^^^^^^^^^^^^^^^^^^^^^^^^^^^
.. module:: instaloader
:no-index:
:noindex:
.. highlight:: python
+1 -1
View File
@@ -2,7 +2,7 @@
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
.. module:: instaloader
:no-index:
:noindex:
.. highlight:: python
+1 -1
View File
@@ -2,7 +2,7 @@ Resumable Iterations
^^^^^^^^^^^^^^^^^^^^
.. module:: instaloader
:no-index:
:noindex:
.. highlight:: python
+1 -9
View File
@@ -2,7 +2,7 @@ Instagram Structures
^^^^^^^^^^^^^^^^^^^^
.. module:: instaloader
:no-index:
:noindex:
.. highlight:: python
@@ -71,14 +71,6 @@ TopSearchResults
.. versionadded:: 4.3
TitlePic
""""""""
.. autoclass:: TitlePic
:no-show-inheritance:
.. versionadded:: 4.8
Loading and Saving
""""""""""""""""""
+2
View File
@@ -0,0 +1,2 @@
requests
sphinx~=2.3
+251
View File
@@ -0,0 +1,251 @@
# sphinx_autodoc_typehints from Nov 20 2018 (version 1.5.2).
# https://github.com/agronholm/sphinx-autodoc-typehints
# (slightly modified to fix class links, maybe related to agronholm/sphinx-autodoc-typehints#38)
#
# This is the MIT license: http://www.opensource.org/licenses/mit-license.php
#
# Copyright (c) Alex Grönholm
#
# Permission is hereby granted, free of charge, to any person obtaining a copy of this
# software and associated documentation files (the "Software"), to deal in the Software
# without restriction, including without limitation the rights to use, copy, modify, merge,
# publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons
# to whom the Software is furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all copies or
# substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED,
# INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR
# PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE
# FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
# OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
# DEALINGS IN THE SOFTWARE.
#
import inspect
from typing import get_type_hints, TypeVar, Any, AnyStr, Generic, Union
from sphinx.util import logging
from sphinx.util.inspect import Signature
try:
from inspect import unwrap
except ImportError:
def unwrap(func, *, stop=None):
"""This is the inspect.unwrap() method copied from Python 3.5's standard library."""
if stop is None:
def _is_wrapper(f):
return hasattr(f, '__wrapped__')
else:
def _is_wrapper(f):
return hasattr(f, '__wrapped__') and not stop(f)
f = func # remember the original func for error reporting
memo = {id(f)} # Memoise by id to tolerate non-hashable objects
while _is_wrapper(func):
func = func.__wrapped__
id_func = id(func)
if id_func in memo:
raise ValueError('wrapper loop when unwrapping {!r}'.format(f))
memo.add(id_func)
return func
logger = logging.getLogger(__name__)
def format_annotation(annotation):
if inspect.isclass(annotation) and annotation.__module__ == 'builtins':
if annotation.__qualname__ == 'NoneType':
return '``None``'
else:
return ':py:class:`{}`'.format(annotation.__qualname__)
annotation_cls = annotation if inspect.isclass(annotation) else type(annotation)
class_name = None
if annotation_cls.__module__ == 'typing':
params = None
prefix = ':py:class:'
module = 'typing'
extra = ''
if inspect.isclass(getattr(annotation, '__origin__', None)):
annotation_cls = annotation.__origin__
try:
if Generic in annotation_cls.mro():
module = annotation_cls.__module__
except TypeError:
pass # annotation_cls was either the "type" object or typing.Type
if annotation is Any:
return ':py:data:`~typing.Any`'
elif annotation is AnyStr:
return ':py:data:`~typing.AnyStr`'
elif isinstance(annotation, TypeVar):
return '\\%r' % annotation
elif (annotation is Union or getattr(annotation, '__origin__', None) is Union or
hasattr(annotation, '__union_params__')):
prefix = ':py:data:'
class_name = 'Union'
if hasattr(annotation, '__union_params__'):
params = annotation.__union_params__
elif hasattr(annotation, '__args__'):
params = annotation.__args__
if params and len(params) == 2 and (hasattr(params[1], '__qualname__') and
params[1].__qualname__ == 'NoneType'):
class_name = 'Optional'
params = (params[0],)
elif annotation_cls.__qualname__ == 'Tuple' and hasattr(annotation, '__tuple_params__'):
params = annotation.__tuple_params__
if annotation.__tuple_use_ellipsis__:
params += (Ellipsis,)
elif annotation_cls.__qualname__ == 'Callable':
prefix = ':py:data:'
arg_annotations = result_annotation = None
if hasattr(annotation, '__result__'):
arg_annotations = annotation.__args__
result_annotation = annotation.__result__
elif getattr(annotation, '__args__', None):
arg_annotations = annotation.__args__[:-1]
result_annotation = annotation.__args__[-1]
if arg_annotations in (Ellipsis, (Ellipsis,)):
params = [Ellipsis, result_annotation]
elif arg_annotations is not None:
params = [
'\\[{}]'.format(
', '.join(format_annotation(param) for param in arg_annotations)),
result_annotation
]
elif hasattr(annotation, 'type_var'):
# Type alias
class_name = annotation.name
params = (annotation.type_var,)
elif getattr(annotation, '__args__', None) is not None:
params = annotation.__args__
elif hasattr(annotation, '__parameters__'):
params = annotation.__parameters__
if params:
extra = '\\[{}]'.format(', '.join(format_annotation(param) for param in params))
if not class_name:
class_name = annotation_cls.__qualname__.title()
return '{}`~{}.{}`{}'.format(prefix, module, class_name, extra)
elif annotation is Ellipsis:
return '...'
elif inspect.isclass(annotation) or inspect.isclass(getattr(annotation, '__origin__', None)):
if not inspect.isclass(annotation):
annotation_cls = annotation.__origin__
extra = ''
if Generic in annotation_cls.mro():
params = (getattr(annotation, '__parameters__', None) or
getattr(annotation, '__args__', None))
extra = '\\[{}]'.format(', '.join(format_annotation(param) for param in params))
module = annotation.__module__.split('.')[0] # hack to 'fix' class linking for Instaloader project
return ':py:class:`~{}.{}`{}'.format(module, annotation_cls.__qualname__,
extra)
return str(annotation)
def process_signature(app, what: str, name: str, obj, options, signature, return_annotation):
if not callable(obj):
return
if what in ('class', 'exception'):
obj = getattr(obj, '__init__', getattr(obj, '__new__', None))
if not getattr(obj, '__annotations__', None):
return
obj = unwrap(obj)
signature = Signature(obj)
parameters = [
param.replace(annotation=inspect.Parameter.empty)
for param in signature.signature.parameters.values()
]
if parameters:
if what in ('class', 'exception'):
del parameters[0]
elif what == 'method':
outer = inspect.getmodule(obj)
for clsname in obj.__qualname__.split('.')[:-1]:
outer = getattr(outer, clsname)
method_name = obj.__name__
if method_name.startswith("__") and not method_name.endswith("__"):
# If the method starts with double underscore (dunder)
# Python applies mangling so we need to prepend the class name.
# This doesn't happen if it always ends with double underscore.
class_name = obj.__qualname__.split('.')[-2]
method_name = "_{c}{m}".format(c=class_name, m=method_name)
method_object = outer.__dict__[method_name]
if not isinstance(method_object, (classmethod, staticmethod)):
del parameters[0]
signature.signature = signature.signature.replace(
parameters=parameters,
return_annotation=inspect.Signature.empty)
return signature.format_args().replace('\\', '\\\\'), None
def process_docstring(app, what, name, obj, options, lines):
if isinstance(obj, property):
obj = obj.fget
if callable(obj):
if what in ('class', 'exception'):
obj = getattr(obj, '__init__')
obj = unwrap(obj)
try:
type_hints = get_type_hints(obj)
except (AttributeError, TypeError):
# Introspecting a slot wrapper will raise TypeError
return
except NameError as exc:
logger.warning('Cannot resolve forward reference in type annotations of "%s": %s',
name, exc)
type_hints = obj.__annotations__
for argname, annotation in type_hints.items():
if argname.endswith('_'):
argname = '{}\\_'.format(argname[:-1])
formatted_annotation = format_annotation(annotation)
if argname == 'return':
if what in ('class', 'exception'):
# Don't add return type None from __init__()
continue
insert_index = len(lines)
for i, line in enumerate(lines):
if line.startswith(':rtype:'):
insert_index = None
break
elif line.startswith(':return:') or line.startswith(':returns:'):
insert_index = i
if insert_index is not None:
if insert_index == len(lines):
# Ensure that :rtype: doesn't get joined with a paragraph of text, which
# prevents it being interpreted.
lines.append('')
insert_index += 1
lines.insert(insert_index, ':rtype: {}'.format(formatted_annotation))
else:
searchfor = ':param {}:'.format(argname)
for i, line in enumerate(lines):
if line.startswith(searchfor):
lines.insert(i, ':type {}: {}'.format(argname, formatted_annotation))
break
+1 -1
View File
@@ -33,7 +33,7 @@ happen under normal conditions, consider adjusting the
There have been observations that services, that in their nature offer
promiscuous IP addresses, such as cloud, VPN and public proxy services, might be
subject to significantly stricter limits for anonymous access. However,
:ref:`logged-in accesses<login>` do not seem to be affected.
logged-in accesses (see :option:`--login`) do not seem to be affected.
Instaloader allows to adjust the rate controlling behavior by overriding
:class:`instaloader.RateController`.
+8 -24
View File
@@ -1,7 +1,7 @@
"""Download pictures (or videos) along with their captions and other metadata from Instagram."""
__version__ = '4.15.4'
__version__ = '4.10'
try:
@@ -13,26 +13,10 @@ else:
win_unicode_console.enable()
from .exceptions import *
from .instaloader import Instaloader as Instaloader
from .instaloadercontext import (InstaloaderContext as InstaloaderContext,
RateController as RateController)
from .lateststamps import LatestStamps as LatestStamps
from .nodeiterator import (NodeIterator as NodeIterator,
FrozenNodeIterator as FrozenNodeIterator,
resumable_iteration as resumable_iteration)
from .structures import (Hashtag as Hashtag,
Highlight as Highlight,
Post as Post,
PostSidecarNode as PostSidecarNode,
PostComment as PostComment,
PostCommentAnswer as PostCommentAnswer,
PostLocation as PostLocation,
Profile as Profile,
Story as Story,
StoryItem as StoryItem,
TopSearchResults as TopSearchResults,
TitlePic as TitlePic,
load_structure_from_file as load_structure_from_file,
save_structure_to_file as save_structure_to_file,
load_structure as load_structure,
get_json_structure as get_json_structure)
from .instaloader import Instaloader
from .instaloadercontext import InstaloaderContext, RateController
from .lateststamps import LatestStamps
from .nodeiterator import NodeIterator, FrozenNodeIterator, resumable_iteration
from .structures import (Hashtag, Highlight, Post, PostSidecarNode, PostComment, PostCommentAnswer, PostLocation,
Profile, Story, StoryItem, TopSearchResults, load_structure_from_file, save_structure_to_file,
load_structure, get_json_structure)
+54 -178
View File
@@ -6,38 +6,23 @@ import os
import re
import sys
from argparse import ArgumentParser, ArgumentTypeError, SUPPRESS
from enum import IntEnum
from typing import List, Optional
from . import (AbortDownloadException, BadCredentialsException, Instaloader, InstaloaderException,
InvalidArgumentException, LoginException, Post, Profile, ProfileNotExistsException, StoryItem,
InvalidArgumentException, Post, Profile, ProfileNotExistsException, StoryItem,
TwoFactorAuthRequiredException, __version__, load_structure_from_file)
from .instaloader import (get_default_session_filename, get_default_stamps_filename)
from .instaloadercontext import default_user_agent
from .lateststamps import LatestStamps
try:
import browser_cookie3
bc3_library = True
except ImportError:
bc3_library = False
class ExitCode(IntEnum):
SUCCESS = 0
NON_FATAL_ERROR = 1
INIT_FAILURE = 2
LOGIN_FAILURE = 3
DOWNLOAD_ABORTED = 4
USER_ABORTED = 5
UNEXPECTED_ERROR = 99
def usage_string():
# NOTE: duplicated in README.rst and docs/index.rst
argv0 = os.path.basename(sys.argv[0])
argv0 = "instaloader" if argv0 == "__main__.py" else argv0
return """
{0} [--comments] [--geotags]
{2:{1}} [--stories] [--highlights] [--tagged] [--reels] [--igtv]
{2:{1}} [--stories] [--highlights] [--tagged] [--igtv]
{2:{1}} [--login YOUR-USERNAME] [--fast-update]
{2:{1}} profile | "#hashtag" | %%location_id | :stories | :feed | :saved
{0} --help""".format(argv0, len(argv0), '')
@@ -82,53 +67,6 @@ def filterstr_to_filterfunc(filter_str: str, item_type: type):
return filterfunc
def get_cookies_from_instagram(domain: str, browser: str, cookie_file: Optional[str] = None):
supported_browsers = {
"brave": browser_cookie3.brave,
"chrome": browser_cookie3.chrome,
"chromium": browser_cookie3.chromium,
"edge": browser_cookie3.edge,
"firefox": browser_cookie3.firefox,
"librewolf": browser_cookie3.librewolf,
"opera": browser_cookie3.opera,
"opera_gx": browser_cookie3.opera_gx,
"safari": browser_cookie3.safari,
"vivaldi": browser_cookie3.vivaldi,
}
if browser not in supported_browsers:
raise InvalidArgumentException("Loading cookies from the specified browser failed\n"
"Supported browsers are Brave, Chrome, Chromium, Edge, Firefox, LibreWolf, "
"Opera, Opera_GX, Safari and Vivaldi")
cookies = {}
browser_cookies = list(supported_browsers[browser](cookie_file=cookie_file))
for cookie in browser_cookies:
if domain in cookie.domain:
cookies[cookie.name] = cookie.value
if cookies:
print(f"Cookies loaded successfully from {browser}")
else:
raise LoginException(f"No cookies found for Instagram in {browser}, "
f"Are you logged in successfully in {browser}?")
return cookies
def import_session(browser: str, instaloader: Instaloader, cookiefile: Optional[str]):
cookie = get_cookies_from_instagram('instagram', browser, cookiefile)
if cookie is not None:
instaloader.context.update_cookies(cookie)
username = instaloader.test_login()
if not username:
raise LoginException(f"Not logged in. Are you logged in successfully in {browser}?")
instaloader.context.username = username
print(f"{username} has been successfully logged in.")
print(f"Next time use --login={username} to reuse the same session.")
def _main(instaloader: Instaloader, targetlist: List[str],
username: Optional[str] = None, password: Optional[str] = None,
sessionfile: Optional[str] = None,
@@ -136,14 +74,11 @@ def _main(instaloader: Instaloader, targetlist: List[str],
download_stories: bool = False,
download_highlights: bool = False,
download_tagged: bool = False,
download_reels: bool = False,
download_igtv: bool = False,
fast_update: bool = False,
latest_stamps_file: Optional[str] = None,
max_count: Optional[int] = None, post_filter_str: Optional[str] = None,
storyitem_filter_str: Optional[str] = None,
browser: Optional[str] = None,
cookiefile: Optional[str] = None) -> ExitCode:
storyitem_filter_str: Optional[str] = None) -> None:
"""Download set of profiles, hashtags etc. and handle logging in and session files if desired."""
# Parse and generate filter function
post_filter = None
@@ -158,11 +93,6 @@ def _main(instaloader: Instaloader, targetlist: List[str],
if latest_stamps_file is not None:
latest_stamps = LatestStamps(latest_stamps_file)
instaloader.context.log(f"Using latest stamps from {latest_stamps_file}.")
# load cookies if browser is not None
if browser and bc3_library:
import_session(browser.lower(), instaloader, cookiefile)
elif browser and not bc3_library:
raise InvalidArgumentException("browser_cookie3 library is needed to load cookies from browsers")
# Login, if desired
if username is not None:
if not re.match(r"^[A-Za-z0-9._]+$", username):
@@ -178,10 +108,6 @@ def _main(instaloader: Instaloader, targetlist: List[str],
try:
instaloader.login(username, password)
except TwoFactorAuthRequiredException:
# https://github.com/instaloader/instaloader/issues/1217
instaloader.context.error("Warning: There have been reports of 2FA currently not working. "
"Consider importing session cookies from your browser with "
"--load-cookies.")
while True:
try:
code = input("Enter 2FA verification code: ")
@@ -191,19 +117,14 @@ def _main(instaloader: Instaloader, targetlist: List[str],
print(err, file=sys.stderr)
pass
else:
try:
instaloader.interactive_login(username)
except KeyboardInterrupt:
print("\nInterrupted by user.", file=sys.stderr)
return ExitCode.USER_ABORTED
instaloader.interactive_login(username)
instaloader.context.log("Logged in as %s." % username)
# since 4.2.9 login is required for geotags
if instaloader.download_geotags and not instaloader.context.is_logged_in:
instaloader.context.error("Warning: Login is required to download geotags of posts.")
instaloader.context.error("Warning: Use --login to download geotags of posts.")
# Try block for KeyboardInterrupt (save session on ^C)
profiles = set()
anonymous_retry_profiles = set()
exit_code = ExitCode.SUCCESS
try:
# Generate set of profiles, already downloading non-profile targets
for target in targetlist:
@@ -256,13 +177,11 @@ def _main(instaloader: Instaloader, targetlist: List[str],
instaloader.download_saved_posts(fast_update=fast_update, max_count=max_count,
post_filter=post_filter)
elif re.match(r"^[A-Za-z0-9._]+$", target):
download_profile_content = download_posts or download_tagged or download_reels or download_igtv
try:
profile = instaloader.check_profile_id(target, latest_stamps)
if instaloader.context.is_logged_in and profile.has_blocked_viewer:
if download_profile_pic or (
download_profile_content and not profile.is_private
):
if download_profile_pic or ((download_posts or download_tagged or download_igtv)
and not profile.is_private):
raise ProfileNotExistsException("{} blocked you; But we download her anonymously."
.format(target))
else:
@@ -272,7 +191,8 @@ def _main(instaloader: Instaloader, targetlist: List[str],
except ProfileNotExistsException as err:
# Not only our profile.has_blocked_viewer condition raises ProfileNotExistsException,
# check_profile_id() also does, since access to blocked profile may be responded with 404.
if instaloader.context.is_logged_in and (download_profile_pic or download_profile_content):
if instaloader.context.is_logged_in and (download_profile_pic or download_posts or
download_tagged or download_igtv):
instaloader.context.log(err)
instaloader.context.log("Trying again anonymously, helps in case you are just blocked.")
with instaloader.anonymous_copy() as anonymous_loader:
@@ -295,42 +215,23 @@ def _main(instaloader: Instaloader, targetlist: List[str],
' '.join([p.username for p in profiles])))
if instaloader.context.iphone_support and profiles and (download_profile_pic or download_posts) and \
not instaloader.context.is_logged_in:
instaloader.context.log("Hint: Login to download higher-quality versions of pictures.")
instaloader.download_profiles(
profiles,
download_profile_pic,
download_posts,
download_tagged,
download_igtv,
download_highlights,
download_stories,
fast_update,
post_filter,
storyitem_filter,
latest_stamps=latest_stamps,
reels=download_reels,
)
instaloader.context.log("Hint: Use --login to download higher-quality versions of pictures.")
instaloader.download_profiles(profiles,
download_profile_pic, download_posts, download_tagged, download_igtv,
download_highlights, download_stories,
fast_update, post_filter, storyitem_filter, latest_stamps=latest_stamps)
if anonymous_retry_profiles:
instaloader.context.log("Downloading anonymously: {}"
.format(' '.join([p.username for p in anonymous_retry_profiles])))
with instaloader.anonymous_copy() as anonymous_loader:
anonymous_loader.download_profiles(
anonymous_retry_profiles,
download_profile_pic,
download_posts,
download_tagged,
download_igtv,
fast_update=fast_update,
post_filter=post_filter,
latest_stamps=latest_stamps,
reels=download_reels
)
anonymous_loader.download_profiles(anonymous_retry_profiles,
download_profile_pic, download_posts, download_tagged, download_igtv,
fast_update=fast_update, post_filter=post_filter,
latest_stamps=latest_stamps)
except KeyboardInterrupt:
print("\nInterrupted by user.", file=sys.stderr)
exit_code = ExitCode.USER_ABORTED
except AbortDownloadException as exc:
print("\nDownload aborted: {}.".format(exc), file=sys.stderr)
exit_code = ExitCode.DOWNLOAD_ABORTED
# Save session if it is useful
if instaloader.context.is_logged_in:
instaloader.save_session_to_file(sessionfile)
@@ -342,8 +243,6 @@ def _main(instaloader: Instaloader, targetlist: List[str],
else:
# Instaloader did not do anything
instaloader.context.log("usage:" + usage_string())
exit_code = ExitCode.INIT_FAILURE
return exit_code
def main():
@@ -359,17 +258,17 @@ def main():
help="Download profile. If an already-downloaded profile has been renamed, Instaloader "
"automatically finds it by its unique ID and renames the folder likewise.")
g_targets.add_argument('_at_profile', nargs='*', metavar="@profile",
help="Download all followees of profile. Requires login. "
help="Download all followees of profile. Requires --login. "
"Consider using :feed rather than @yourself.")
g_targets.add_argument('_hashtag', nargs='*', metavar='"#hashtag"', help="Download #hashtag.")
g_targets.add_argument('_location', nargs='*', metavar='%location_id',
help="Download %%location_id. Requires login.")
help="Download %%location_id. Requires --login.")
g_targets.add_argument('_feed', nargs='*', metavar=":feed",
help="Download pictures from your feed. Requires login.")
help="Download pictures from your feed. Requires --login.")
g_targets.add_argument('_stories', nargs='*', metavar=":stories",
help="Download the stories of your followees. Requires login.")
help="Download the stories of your followees. Requires --login.")
g_targets.add_argument('_saved', nargs='*', metavar=":saved",
help="Download the posts that you marked as saved. Requires login.")
help="Download the posts that you marked as saved. Requires --login.")
g_targets.add_argument('_singlepost', nargs='*', metavar="-- -shortcode",
help="Download the post with the given shortcode")
g_targets.add_argument('_json', nargs='*', metavar="filename.json[.xz]",
@@ -400,11 +299,11 @@ def main():
help='Download geotags when available. Geotags are stored as a '
'text file with the location\'s name and a Google Maps link. '
'This requires an additional request to the Instagram '
'server for each picture. Requires login.')
'server for each picture. Requires --login.')
g_post.add_argument('-C', '--comments', action='store_true',
help='Download and update comments for each post. '
'This requires an additional request to the Instagram '
'server for each post, which is why it is disabled by default. Requires login.')
'server for each post, which is why it is disabled by default.')
g_post.add_argument('--no-captions', action='store_true',
help='Do not create txt files.')
g_post.add_argument('--post-metadata-txt', action='append',
@@ -418,15 +317,13 @@ def main():
g_post.add_argument('--no-compress-json', action='store_true',
help='Do not xz compress JSON files, rather create pretty formatted JSONs.')
g_prof.add_argument('-s', '--stories', action='store_true',
help='Also download stories of each profile that is downloaded. Requires login.')
help='Also download stories of each profile that is downloaded. Requires --login.')
g_prof.add_argument('--stories-only', action='store_true',
help=SUPPRESS)
g_prof.add_argument('--highlights', action='store_true',
help='Also download highlights of each profile that is downloaded. Requires login.')
help='Also download highlights of each profile that is downloaded. Requires --login.')
g_prof.add_argument('--tagged', action='store_true',
help='Also download posts where each profile is tagged.')
g_prof.add_argument('--reels', action='store_true',
help='Also download Reels videos.')
g_prof.add_argument('--igtv', action='store_true',
help='Also download IGTV videos.')
@@ -456,14 +353,9 @@ def main():
'Instaloader can login to Instagram. This allows downloading private profiles. '
'To login, pass the --login option. Your session cookie (not your password!) '
'will be saved to a local file to be reused next time you want Instaloader '
'to login. Instead of --login, the --load-cookies option can be used to '
'import a session from a browser.')
'to login.')
g_login.add_argument('-l', '--login', metavar='YOUR-USERNAME',
help='Login name (profile name) for your Instagram account.')
g_login.add_argument('-b', '--load-cookies', metavar='BROWSER-NAME',
help='Browser name to load cookies from Instagram')
g_login.add_argument('-B', '--cookiefile', metavar='COOKIE-FILE',
help='Cookie file of a profile to load cookies')
g_login.add_argument('-f', '--sessionfile',
help='Path for loading and storing session key file. '
'Defaults to ' + get_default_session_filename("<login_name>"))
@@ -524,15 +416,15 @@ def main():
args = parser.parse_args()
try:
if (args.login is None and args.load_cookies is None) and (args.stories or args.stories_only):
print("Login is required to download stories.", file=sys.stderr)
if args.login is None and (args.stories or args.stories_only):
print("--login=USERNAME required to download stories.", file=sys.stderr)
args.stories = False
if args.stories_only:
raise InvalidArgumentException()
raise SystemExit(1)
if ':feed-all' in args.profile or ':feed-liked' in args.profile:
raise InvalidArgumentException(":feed-all and :feed-liked were removed. Use :feed as target and "
"eventually --post-filter=viewer_has_liked.")
raise SystemExit(":feed-all and :feed-liked were removed. Use :feed as target and "
"eventually --post-filter=viewer_has_liked.")
post_metadata_txt_pattern = '\n'.join(args.post_metadata_txt) if args.post_metadata_txt else None
storyitem_metadata_txt_pattern = '\n'.join(args.storyitem_metadata_txt) if args.storyitem_metadata_txt else None
@@ -542,18 +434,15 @@ def main():
post_metadata_txt_pattern = ''
storyitem_metadata_txt_pattern = ''
else:
raise InvalidArgumentException("--no-captions and --post-metadata-txt or --storyitem-metadata-txt "
"given; That contradicts.")
raise SystemExit("--no-captions and --post-metadata-txt or --storyitem-metadata-txt given; "
"That contradicts.")
if args.no_resume and args.resume_prefix:
raise InvalidArgumentException("--no-resume and --resume-prefix given; That contradicts.")
raise SystemExit("--no-resume and --resume-prefix given; That contradicts.")
resume_prefix = (args.resume_prefix if args.resume_prefix else 'iterator') if not args.no_resume else None
if args.no_pictures and args.fast_update:
raise InvalidArgumentException('--no-pictures and --fast-update cannot be used together.')
if args.login and args.load_cookies:
raise InvalidArgumentException('--load-cookies and --login cannot be used together.')
raise SystemExit('--no-pictures and --fast-update cannot be used together.')
# Determine what to download
download_profile_pic = not args.no_profile_pic or args.profile_pic_only
@@ -578,38 +467,25 @@ def main():
iphone_support=not args.no_iphone,
title_pattern=args.title_pattern,
sanitize_paths=args.sanitize_paths)
exit_code = _main(loader,
args.profile,
username=args.login.lower() if args.login is not None else None,
password=args.password,
sessionfile=args.sessionfile,
download_profile_pic=download_profile_pic,
download_posts=download_posts,
download_stories=download_stories,
download_highlights=args.highlights,
download_tagged=args.tagged,
download_reels=args.reels,
download_igtv=args.igtv,
fast_update=args.fast_update,
latest_stamps_file=args.latest_stamps,
max_count=int(args.count) if args.count is not None else None,
post_filter_str=args.post_filter,
storyitem_filter_str=args.storyitem_filter,
browser=args.load_cookies,
cookiefile=args.cookiefile)
_main(loader,
args.profile,
username=args.login.lower() if args.login is not None else None,
password=args.password,
sessionfile=args.sessionfile,
download_profile_pic=download_profile_pic,
download_posts=download_posts,
download_stories=download_stories,
download_highlights=args.highlights,
download_tagged=args.tagged,
download_igtv=args.igtv,
fast_update=args.fast_update,
latest_stamps_file=args.latest_stamps,
max_count=int(args.count) if args.count is not None else None,
post_filter_str=args.post_filter,
storyitem_filter_str=args.storyitem_filter)
loader.close()
if loader.has_stored_errors:
exit_code = ExitCode.NON_FATAL_ERROR
except InvalidArgumentException as err:
print(err, file=sys.stderr)
exit_code = ExitCode.INIT_FAILURE
except LoginException as err:
print(err, file=sys.stderr)
exit_code = ExitCode.LOGIN_FAILURE
except InstaloaderException as err:
print("Fatal error: %s" % err)
exit_code = ExitCode.UNEXPECTED_ERROR
sys.exit(exit_code)
raise SystemExit("Fatal error: %s" % err) from err
if __name__ == "__main__":
+2 -6
View File
@@ -33,11 +33,7 @@ class LoginRequiredException(InstaloaderException):
pass
class LoginException(InstaloaderException):
pass
class TwoFactorAuthRequiredException(LoginException):
class TwoFactorAuthRequiredException(InstaloaderException):
pass
@@ -49,7 +45,7 @@ class BadResponseException(InstaloaderException):
pass
class BadCredentialsException(LoginException):
class BadCredentialsException(InstaloaderException):
pass
+24 -106
View File
@@ -77,13 +77,13 @@ def _requires_login(func: Callable) -> Callable:
@wraps(func)
def call(instaloader, *args, **kwargs):
if not instaloader.context.is_logged_in:
raise LoginRequiredException("Login required.")
raise LoginRequiredException("--login=USERNAME required.")
return func(instaloader, *args, **kwargs)
return call
def _retry_on_connection_error(func: Callable) -> Callable:
"""Decorator to retry the function max_connection_attempts number of times.
"""Decorator to retry the function max_connection_attemps number of times.
Herewith-decorated functions need an ``_attempt`` keyword argument.
@@ -330,9 +330,6 @@ class Instaloader:
def __exit__(self, *args):
self.close()
def __repr__(self):
return f"<Instaloader context={self.context}>"
@_retry_on_connection_error
def download_pic(self, filename: str, url: str, mtime: datetime,
filename_suffix: Optional[str] = None, _attempt: int = 1) -> bool:
@@ -646,16 +643,11 @@ class Instaloader:
def login(self, user: str, passwd: str) -> None:
"""Log in to instagram with given username and password and internally store session object.
:raises InvalidArgumentException: If the provided username does not exist.
:raises BadCredentialsException: If the provided password is wrong.
:raises ConnectionException: If connection to Instagram failed.
:raises TwoFactorAuthRequiredException: First step of 2FA login done, now call
:meth:`Instaloader.two_factor_login`.
:raises LoginException: An error happened during login (for example, an invalid response was received).
Or if the provided username does not exist.
.. versionchanged:: 4.12
Raises LoginException instead of ConnectionException when an error happens.
Raises LoginException instead of InvalidArgumentException when the username does not exist.
"""
:meth:`Instaloader.two_factor_login`."""
self.context.login(user, passwd)
def two_factor_login(self, two_factor_code) -> None:
@@ -725,7 +717,7 @@ class Instaloader:
# Download the image(s) / video thumbnail and videos within sidecars if desired
downloaded = True
if post.typename == 'GraphSidecar':
if (self.download_pictures or self.download_videos) and post.mediacount > 0:
if self.download_pictures or self.download_videos:
if not _all_already_downloaded(
filename_template, enumerate(
(post.get_is_videos()[i]
@@ -740,20 +732,19 @@ class Instaloader:
suffix: Optional[str] = str(edge_number)
if '{filename}' in self.filename_pattern:
suffix = None
video_url = sidecar_node.video_url
if self.download_pictures and (video_url is None or self.download_video_thumbnails):
if self.download_pictures and (not sidecar_node.is_video or self.download_video_thumbnails):
# pylint:disable=cell-var-from-loop
sidecar_filename = self.__prepare_filename(filename_template,
lambda: sidecar_node.display_url)
# Download sidecar picture or video thumbnail (--no-pictures implies --no-video-thumbnails)
downloaded &= self.download_pic(filename=sidecar_filename, url=sidecar_node.display_url,
mtime=post.date_local, filename_suffix=suffix)
if video_url is not None and self.download_videos:
if sidecar_node.is_video and self.download_videos:
# pylint:disable=cell-var-from-loop
sidecar_filename = self.__prepare_filename(filename_template,
lambda: video_url)
lambda: sidecar_node.video_url)
# Download sidecar video if desired
downloaded &= self.download_pic(filename=sidecar_filename, url=video_url,
downloaded &= self.download_pic(filename=sidecar_filename, url=sidecar_node.video_url,
mtime=post.date_local, filename_suffix=suffix)
else:
downloaded = False
@@ -870,9 +861,8 @@ 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:
if item.date_local <= last_scraped:
break
if latest_stamps is not None and item.date_local <= last_scraped:
break
if storyitem_filter is not None and not storyitem_filter(item):
self.context.log("<{} skipped>".format(item), flush=True)
continue
@@ -963,11 +953,6 @@ class Instaloader:
.. versionchanged:: 4.3
Also downloads and saves the Highlight's cover pictures.
.. versionchanged:: 4.16
Add progress output for highlight retrieval, displaying a counter of the current
highlight number and the total number of highlights being processed. This provides
feedback similar to the per-item download counter already present in this function.
:param user: ID or Profile of the user whose highlights should get downloaded.
:param fast_update: If true, abort when first already-downloaded picture is encountered
:param filename_target: Replacement for {target} in dirname_pattern and filename_pattern
@@ -975,20 +960,14 @@ class Instaloader:
:param storyitem_filter: function(storyitem), which returns True if given StoryItem should be downloaded
:raises LoginRequiredException: If called without being logged in.
"""
user_highlights = list(self.get_highlights(user))
hl_size = len(user_highlights)
for hl_number, user_highlight in enumerate(user_highlights, start=1):
for user_highlight in self.get_highlights(user):
name = user_highlight.owner_username
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)))
self.context.log(
"[{0:{w}d}/{1:{w}d}] Retrieving highlights \"{2}\" from profile {3}".format(
hl_number, hl_size, user_highlight.title, name, w=len(str(hl_size))
)
)
self.context.log("Retrieving highlights \"{}\" from profile {}".format(user_highlight.title, name))
self.download_highlight_cover(user_highlight, highlight_target)
totalcount = user_highlight.itemcount
count = 1
@@ -1012,8 +991,7 @@ class Instaloader:
max_count: Optional[int] = None,
total_count: Optional[int] = None,
owner_profile: Optional[Profile] = None,
takewhile: Optional[Callable[[Post], bool]] = None,
possibly_pinned: int = 0) -> None:
takewhile: Optional[Callable[[Post], bool]] = None) -> None:
"""
Download the Posts returned by given Post Iterator.
@@ -1025,9 +1003,6 @@ class Instaloader:
.. versionchanged:: 4.8
Add `takewhile` parameter.
.. versionchanged:: 4.10.3
Add `possibly_pinned` parameter.
:param posts: Post Iterator to loop through.
:param target: Target name.
:param fast_update: :option:`--fast-update`.
@@ -1036,8 +1011,6 @@ class Instaloader:
:param total_count: Total number of posts returned by given iterator.
:param owner_profile: Associated profile, if any.
:param takewhile: Expression evaluated for each post. Once it returns false, downloading stops.
:param possibly_pinned: Number of posts that might be pinned. These posts do not cause download
to stop even if they've already been downloaded.
"""
displayed_count = (max_count if total_count is None or max_count is not None and max_count < total_count
else total_count)
@@ -1059,7 +1032,7 @@ class Instaloader:
) as (is_resuming, start_index):
for number, post in enumerate(posts, start=start_index + 1):
should_stop = not takewhile(post)
if should_stop and number <= possibly_pinned:
if should_stop and post.is_pinned:
continue
if (max_count is not None and number > max_count) or should_stop:
break
@@ -1093,7 +1066,7 @@ class Instaloader:
except PostChangedException:
post_changed = True
continue
if fast_update and not downloaded and not post_changed and number > possibly_pinned:
if fast_update and not downloaded and not post_changed and not post.is_pinned:
# disengage fast_update for first post when resuming
if not is_resuming or number > 0:
break
@@ -1230,7 +1203,6 @@ class Instaloader:
Use :meth:`Hashtag.get_posts_resumable`."""
return Hashtag.from_name(self.context, hashtag).get_posts_resumable()
@_requires_login
def download_hashtag(self, hashtag: Union[Hashtag, str],
max_count: Optional[int] = None,
post_filter: Optional[Callable[[Post], bool]] = None,
@@ -1297,32 +1269,6 @@ class Instaloader:
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)
def download_reels(self, profile: Profile, fast_update: bool = False,
post_filter: Optional[Callable[[Post], bool]] = None,
latest_stamps: Optional[LatestStamps] = None) -> None:
"""Download reels videos of a profile.
.. versionadded:: 4.14.0
"""
self.context.log("Retrieving reels videos for profile {}.".format(profile.username))
posts_takewhile: Optional[Callable[[Post], bool]] = None
if latest_stamps is not None:
last_scraped = latest_stamps.get_last_reels_timestamp(profile.username)
posts_takewhile = lambda p: p.date_local > last_scraped
reels = profile.get_reels()
self.posts_download_loop(
reels,
profile.username,
fast_update,
post_filter,
owner_profile=profile,
takewhile=posts_takewhile,
possibly_pinned=3,
)
if latest_stamps is not None and reels.first_item is not None:
latest_stamps.set_last_reels_timestamp(profile.username, reels.first_item.date_local)
def download_igtv(self, profile: Profile, fast_update: bool = False,
post_filter: Optional[Callable[[Post], bool]] = None,
latest_stamps: Optional[LatestStamps] = None) -> None:
@@ -1452,9 +1398,7 @@ class Instaloader:
post_filter: Optional[Callable[[Post], bool]] = None,
storyitem_filter: Optional[Callable[[Post], bool]] = None,
raise_errors: bool = False,
latest_stamps: Optional[LatestStamps] = None,
max_count: Optional[int] = None,
reels: bool = False):
latest_stamps: Optional[LatestStamps] = None):
"""High-level method to download set of profiles.
:param profiles: Set of profiles to download.
@@ -1471,8 +1415,6 @@ class Instaloader:
Whether :exc:`LoginRequiredException` and :exc:`PrivateProfileNotFollowedException` should be raised or
catched and printed with :meth:`InstaloaderContext.error_catcher`.
:param latest_stamps: :option:`--latest-stamps`.
:param max_count: Maximum count of posts to download.
:param reels: :option:`--reels`.
.. versionadded:: 4.1
@@ -1481,12 +1423,6 @@ class Instaloader:
.. versionchanged:: 4.8
Add `latest_stamps` parameter.
.. versionchanged:: 4.13
Add `max_count` parameter.
.. versionchanged:: 4.14
Add `reels` parameter.
"""
@contextmanager
@@ -1518,7 +1454,7 @@ class Instaloader:
if tagged or igtv or highlights or posts:
if (not self.context.is_logged_in and
profile.is_private):
raise LoginRequiredException("Login required.")
raise LoginRequiredException("--login=USERNAME required.")
if (self.context.username != profile.username and
profile.is_private and
not profile.followed_by_viewer):
@@ -1530,12 +1466,6 @@ class Instaloader:
self.download_tagged(profile, fast_update=fast_update, post_filter=post_filter,
latest_stamps=latest_stamps)
# Download reels, if requested
if reels:
with self.context.error_catcher('Download reels of {}'.format(profile_name)):
self.download_reels(profile, fast_update=fast_update, post_filter=post_filter,
latest_stamps=latest_stamps)
# Download IGTV, if requested
if igtv:
with self.context.error_catcher('Download IGTV of {}'.format(profile_name)):
@@ -1558,7 +1488,7 @@ class Instaloader:
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, possibly_pinned=3, max_count=max_count)
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)
@@ -1646,16 +1576,11 @@ class Instaloader:
def interactive_login(self, username: str) -> None:
"""Logs in and internally stores session, asking user for password interactively.
:raises InvalidArgumentException: when in quiet mode.
:raises LoginException: If the provided username does not exist.
:raises ConnectionException: If connection to Instagram failed.
.. versionchanged:: 4.12
Raises InvalidArgumentException instead of LoginRequiredException when in quiet mode.
Raises LoginException instead of InvalidArgumentException when the username does not exist.
"""
:raises LoginRequiredException: when in quiet mode.
:raises InvalidArgumentException: If the provided username does not exist.
:raises ConnectionException: If connection to Instagram failed."""
if self.context.quiet:
raise InvalidArgumentException("Quiet mode requires given password or valid session file.")
raise LoginRequiredException("Quiet mode requires given password or valid session file.")
try:
password = None
while password is None:
@@ -1674,10 +1599,3 @@ class Instaloader:
except BadCredentialsException as err:
print(err, file=sys.stderr)
pass
@property
def has_stored_errors(self) -> bool:
"""Returns whether any error has been reported and stored to be repeated at program termination.
.. versionadded: 4.12"""
return self.context.has_stored_errors
+97 -214
View File
@@ -1,3 +1,4 @@
import hashlib
import json
import os
import pickle
@@ -9,7 +10,7 @@ import textwrap
import time
import urllib.parse
import uuid
from contextlib import contextmanager, suppress
from contextlib import contextmanager
from datetime import datetime, timedelta
from functools import partial
from typing import Any, Callable, Dict, Iterator, List, Optional, Union
@@ -31,30 +32,17 @@ def copy_session(session: requests.Session, request_timeout: Optional[float] = N
return new
def _embedded_query_data(obj: Any) -> Iterator[Dict[str, Any]]:
"""Yield the data of each GraphQL query result in JSON that a page of www.instagram.com embeds."""
if isinstance(obj, dict):
result = obj['__bbox'].get('result') if isinstance(obj.get('__bbox'), dict) else None
if isinstance(result, dict) and isinstance(result.get('data'), dict):
yield result['data']
for value in obj.values():
yield from _embedded_query_data(value)
elif isinstance(obj, list):
for value in obj:
yield from _embedded_query_data(value)
def default_user_agent() -> str:
return ('Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 '
'(KHTML, like Gecko) Chrome/142.0.0.0 Safari/537.36')
return 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 ' \
'(KHTML, like Gecko) Chrome/92.0.4515.131 Safari/537.36'
def default_iphone_headers() -> Dict[str, Any]:
return {'User-Agent': 'Instagram 361.0.0.35.82 (iPad13,8; iOS 18_0; en_US; en-US; ' \
'scale=2.00; 2048x2732; 674117118) AppleWebKit/420+',
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': '16b7bd25c6c06886d57c4d455265669345a2d96625385b8ee30026ac2dc5ed97',
'x-bloks-version-id': '01507c21540f73e2216b6f62a11a5b5e51aa85491b72475c080da35b1228ddd6',
'x-fb-client-ip': 'True',
'x-fb-connection-type': 'wifi',
'x-fb-http-engine': 'Liger',
@@ -107,6 +95,7 @@ class InstaloaderContext:
self.quiet = quiet
self.max_connection_attempts = max_connection_attempts
self._graphql_page_length = 50
self._root_rhx_gis = None
self.two_factor_auth_pending = None
self.iphone_support = iphone_support
self.iphone_headers = default_iphone_headers()
@@ -116,7 +105,7 @@ class InstaloaderContext:
self._rate_controller = rate_controller(self) if rate_controller is not None else RateController(self)
# Can be set to True for testing, disables suppression of InstaloaderContext._error_catcher
# Can be set to True for testing, disables supression of InstaloaderContext._error_catcher
self.raise_all_errors = False
# HTTP status codes that should cause an AbortDownloadException
@@ -163,16 +152,6 @@ class InstaloaderContext:
if repeat_at_end:
self.error_log.append(msg)
def __repr__(self):
return f"<InstaloaderContext is_logged_in={self.is_logged_in}>"
@property
def has_stored_errors(self) -> bool:
"""Returns whether any error has been reported and stored to be repeated at program termination.
.. versionadded: 4.12"""
return bool(self.error_log)
def close(self):
"""Print error log and close session"""
if self.error_log and not self.quiet:
@@ -208,8 +187,7 @@ class InstaloaderContext:
'Referer': 'https://www.instagram.com/',
'User-Agent': self.user_agent,
'X-Instagram-AJAX': '1',
'X-Requested-With': 'XMLHttpRequest',
'x-ig-app-id': '936619743392459'}
'X-Requested-With': 'XMLHttpRequest'}
if empty_session_only:
del header['Host']
del header['Origin']
@@ -233,10 +211,6 @@ class InstaloaderContext:
"""Not meant to be used directly, use :meth:`Instaloader.save_session`."""
return requests.utils.dict_from_cookiejar(self._session.cookies)
def update_cookies(self, cookie):
""".. versionadded:: 4.11"""
self._session.cookies.update(cookie)
def load_session(self, username, sessiondata):
"""Not meant to be used directly, use :meth:`Instaloader.load_session`."""
session = requests.Session()
@@ -259,26 +233,17 @@ class InstaloaderContext:
def test_login(self) -> Optional[str]:
"""Not meant to be used directly, use :meth:`Instaloader.test_login`."""
try:
data = self.graphql_query("d6f4427fbe92d846298cf93df0b937d3", {})
return data["data"]["user"]["username"] if data["data"]["user"] is not None else None
except (AbortDownloadException, ConnectionException) as err:
self.error(f"Error when checking if logged in: {err}")
return None
data = self.graphql_query("d6f4427fbe92d846298cf93df0b937d3", {})
return data["data"]["user"]["username"] if data["data"]["user"] is not None else None
def login(self, user, passwd):
"""Not meant to be used directly, use :meth:`Instaloader.login`.
:raises InvalidArgumentException: If the provided username does not exist.
:raises BadCredentialsException: If the provided password is wrong.
:raises ConnectionException: If connection to Instagram failed.
:raises TwoFactorAuthRequiredException: First step of 2FA login done, now call
:meth:`Instaloader.two_factor_login`.
:raises LoginException: An error happened during login (for example, and invalid response).
Or if the provided username does not exist.
.. versionchanged:: 4.12
Raises LoginException instead of ConnectionException when an error happens.
Raises LoginException instead of InvalidArgumentException when the username does not exist.
"""
:meth:`Instaloader.two_factor_login`."""
# pylint:disable=import-outside-toplevel
import http.client
# pylint:disable=protected-access
@@ -288,28 +253,25 @@ class InstaloaderContext:
'ig_vw': '1920', 'ig_cb': '1', 'csrftoken': '',
's_network': '', 'ds_user_id': ''})
session.headers.update(self._default_http_header())
# Getting mid cookie value
session.get("https://i.instagram.com/api/v1/web/login_page/")
# 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
# Make a request to Instagram's root URL, which will set the session's csrftoken cookie
# Not using self.get_json() here, because we need to access the cookie
session.get('https://www.instagram.com/')
# Add session's csrftoken cookie to session headers
csrf_token = session.cookies.get_dict()['csrftoken']
csrf_json = self.get_json('accounts/login/', {}, session=session)
csrf_token = csrf_json['config']['csrf_token']
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)
login = session.post('https://www.instagram.com/api/v1/web/accounts/login/ajax/',
login = session.post('https://www.instagram.com/accounts/login/ajax/',
data={'enc_password': enc_password, 'username': user}, allow_redirects=True)
try:
resp_json = login.json()
except json.decoder.JSONDecodeError as err:
raise LoginException(
raise ConnectionException(
"Login error: JSON decode fail, {} - {}.".format(login.status_code, login.reason)
) from err
if resp_json.get('two_factor_required'):
@@ -321,34 +283,34 @@ class InstaloaderContext:
resp_json['two_factor_info']['two_factor_identifier'])
raise TwoFactorAuthRequiredException("Login error: two-factor authentication required.")
if resp_json.get('checkpoint_url'):
raise LoginException(
f"Login: Checkpoint required. Point your browser to {resp_json.get('checkpoint_url')} - "
f"follow the instructions, then retry."
)
raise ConnectionException("Login: Checkpoint required. Point your browser to "
"https://www.instagram.com{} - "
"follow the instructions, then retry.".format(resp_json.get('checkpoint_url')))
if resp_json['status'] != 'ok':
if 'message' in resp_json:
raise LoginException("Login error: \"{}\" status, message \"{}\".".format(resp_json['status'],
resp_json['message']))
raise ConnectionException("Login error: \"{}\" status, message \"{}\".".format(resp_json['status'],
resp_json['message']))
else:
raise LoginException("Login error: \"{}\" status.".format(resp_json['status']))
raise ConnectionException("Login error: \"{}\" status.".format(resp_json['status']))
if 'authenticated' not in resp_json:
# Issue #472
if 'message' in resp_json:
raise LoginException("Login error: Unexpected response, \"{}\".".format(resp_json['message']))
raise ConnectionException("Login error: Unexpected response, \"{}\".".format(resp_json['message']))
else:
raise LoginException("Login error: Unexpected response, this might indicate a blocked IP.")
raise ConnectionException("Login error: Unexpected response, this might indicate a blocked IP.")
if not resp_json['authenticated']:
if resp_json['user']:
# '{"authenticated": false, "user": true, "status": "ok"}'
raise BadCredentialsException('Login error: Wrong password.')
else:
# '{"authenticated": false, "user": false, "status": "ok"}'
# Raise LoginException rather than BadCredentialException, because BadCredentialException
# Raise InvalidArgumentException rather than BadCredentialException, because BadCredentialException
# triggers re-asking of password in Instaloader.interactive_login(), which makes no sense if the
# username is invalid.
raise LoginException('Login error: User {} does not exist.'.format(user))
raise InvalidArgumentException('Login error: User {} does not exist.'.format(user))
# '{"authenticated": true, "user": true, "userId": ..., "oneTapPrompt": false, "status": "ok"}'
session.headers.update({'X-CSRFToken': login.cookies['csrftoken']})
if 'csrftoken' in login.cookies:
session.headers.update({'X-CSRFToken': login.cookies['csrftoken']})
self._session = session
self.username = user
self.user_id = resp_json['userId']
@@ -365,8 +327,15 @@ class InstaloaderContext:
raise InvalidArgumentException("No two-factor authentication pending.")
(session, user, two_factor_id) = self.two_factor_auth_pending
login = session.post('https://www.instagram.com/accounts/login/ajax/two_factor/',
data={'username': user, 'verificationCode': two_factor_code, 'identifier': two_factor_id},
login = session.post('https://www.instagram.com/api/v1/web/accounts/login/ajax/two_factor/',
data={
'identifier': two_factor_id,
'queryParams': '{"next":"/"}',
'trust_signal': 'true',
'username': user,
'verification_method': '1',
'verificationCode': two_factor_code,
},
allow_redirects=True)
resp_json = login.json()
if resp_json['status'] != 'ok':
@@ -374,7 +343,8 @@ class InstaloaderContext:
raise BadCredentialsException("2FA error: {}".format(resp_json['message']))
else:
raise BadCredentialsException("2FA error: \"{}\" status.".format(resp_json['status']))
session.headers.update({'X-CSRFToken': login.cookies['csrftoken']})
if 'csrftoken' in login.cookies:
session.headers.update({'X-CSRFToken': login.cookies['csrftoken']})
self._session = session
self.username = user
self.two_factor_auth_pending = None
@@ -384,64 +354,33 @@ class InstaloaderContext:
if self.sleep:
time.sleep(min(random.expovariate(0.6), 15.0))
@staticmethod
def _response_error(resp: requests.Response) -> str:
extra_from_json: Optional[str] = None
with suppress(json.decoder.JSONDecodeError):
resp_json = resp.json()
if "status" in resp_json:
extra_from_json = (
f"\"{resp_json['status']}\" status, message \"{resp_json['message']}\""
if "message" in resp_json
else f"\"{resp_json['status']}\" status"
)
return (
f"{resp.status_code} {resp.reason}"
f"{f' - {extra_from_json}' if extra_from_json is not None else ''}"
f" when accessing {resp.url}"
)
def get_json(self, path: str, params: Dict[str, Any], host: str = 'www.instagram.com',
session: Optional[requests.Session] = None, _attempt=1,
response_headers: Optional[Dict[str, Any]] = None,
use_post: bool = False) -> Dict[str, Any]:
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/
:param params: request parameters
:param params: GET parameters
:param host: Domain part of the URL from where to download the requested JSON; defaults to www.instagram.com
:param session: Session to use, or None to use self.session
:param use_post: Use POST instead of GET to make the request
:return: Decoded response dictionary
:raises AbortDownloadException: When the server responds with
'feedback_required'/'checkpoint_required'/'challenge_required'
:raises QueryReturnedBadRequestException: When the server responds with a 400 (and not
'feedback_required'/'checkpoint_required'/'challenge_required').
:raises QueryReturnedBadRequestException: When the server responds with a 400.
:raises QueryReturnedNotFoundException: When the server responds with a 404.
:raises ConnectionException: When query repeatedly failed.
.. versionchanged:: 4.13
Added `use_post` parameter.
"""
is_graphql_query = 'query_hash' in params and 'graphql/query' in path
is_doc_id_query = 'doc_id' in params and 'graphql/query' in path
is_iphone_query = host == 'i.instagram.com'
is_other_query = not is_graphql_query and not is_doc_id_query and host == "www.instagram.com"
is_other_query = not is_graphql_query and host == "www.instagram.com"
sess = session if session else self._session
try:
self.do_sleep()
if is_graphql_query:
self._rate_controller.wait_before_query(params['query_hash'])
if is_doc_id_query:
self._rate_controller.wait_before_query(params['doc_id'])
if is_iphone_query:
self._rate_controller.wait_before_query('iphone')
if is_other_query:
self._rate_controller.wait_before_query('other')
if use_post:
resp = sess.post('https://{0}/{1}'.format(host, path), data=params, allow_redirects=False)
else:
resp = sess.get('https://{0}/{1}'.format(host, path), params=params, allow_redirects=False)
resp = sess.get('https://{0}/{1}'.format(host, path), params=params, allow_redirects=False)
if resp.status_code in self.fatal_status_codes:
redirect = " redirect to {}".format(resp.headers['location']) if 'location' in resp.headers else ""
body = ""
@@ -456,7 +395,7 @@ class InstaloaderContext:
if (redirect_url.startswith('https://www.instagram.com/accounts/login') or
redirect_url.startswith('https://i.instagram.com/accounts/login')):
if not self.is_logged_in:
raise LoginRequiredException("Redirected to login page. Use --login or --load-cookies.")
raise LoginRequiredException("Redirected to login page. Use --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)):
@@ -468,26 +407,31 @@ class InstaloaderContext:
response_headers.clear()
response_headers.update(resp.headers)
if resp.status_code == 400:
with suppress(json.decoder.JSONDecodeError):
if resp.json().get("message") in [
"feedback_required",
"checkpoint_required",
"challenge_required",
]:
# Raise AbortDownloadException in case of substantial Instagram
# requirements to stop producing more requests
raise AbortDownloadException(self._response_error(resp))
raise QueryReturnedBadRequestException(self._response_error(resp))
raise QueryReturnedBadRequestException("400 Bad Request")
if resp.status_code == 404:
raise QueryReturnedNotFoundException(self._response_error(resp))
raise QueryReturnedNotFoundException("404 Not Found")
if resp.status_code == 429:
raise TooManyRequestsException(self._response_error(resp))
raise TooManyRequestsException("429 Too Many Requests")
if resp.status_code != 200:
raise ConnectionException(self._response_error(resp))
raise ConnectionException("HTTP error code {}.".format(resp.status_code))
is_html_query = not is_graphql_query and not "__a" in params and host == "www.instagram.com"
if is_html_query:
# Extract JSON from HTML response
match = re.search('(?<={"raw":").*?(?<!\\\\)(?=")', resp.text)
if match is None:
raise QueryReturnedNotFoundException("Could not find JSON data in html response.")
# Unescape escaped JSON string
unescaped_string = match.group(0).encode("utf-8").decode("unicode_escape")
resp_json = json.loads(unescaped_string)
return resp_json
else:
resp_json = resp.json()
if 'status' in resp_json and resp_json['status'] != "ok":
raise ConnectionException(self._response_error(resp))
if 'message' in resp_json:
raise ConnectionException("Returned \"{}\" status, message \"{}\".".format(resp_json['status'],
resp_json['message']))
else:
raise ConnectionException("Returned \"{}\" status.".format(resp_json['status']))
return resp_json
except (ConnectionException, json.decoder.JSONDecodeError, requests.exceptions.RequestException) as err:
error_string = "JSON Query to {}: {}".format(path, err)
@@ -501,8 +445,6 @@ class InstaloaderContext:
if isinstance(err, TooManyRequestsException):
if is_graphql_query:
self._rate_controller.handle_429(params['query_hash'])
if is_doc_id_query:
self._rate_controller.handle_429(params['doc_id'])
if is_iphone_query:
self._rate_controller.handle_429('iphone')
if is_other_query:
@@ -514,17 +456,15 @@ class InstaloaderContext:
raise ConnectionException(error_string) from err
def graphql_query(self, query_hash: str, variables: Dict[str, Any],
referer: Optional[str] = None) -> Dict[str, Any]:
referer: Optional[str] = None, rhx_gis: Optional[str] = None) -> Dict[str, Any]:
"""
Do a GraphQL Query.
:param query_hash: Query identifying hash.
:param variables: Variables for the Query.
:param referer: HTTP Referer, or None.
:param rhx_gis: 'rhx_gis' variable as somewhere returned by Instagram, needed to 'sign' request
:return: The server's response dictionary.
.. versionchanged:: 4.13.1
Removed the `rhx_gis` parameter.
"""
with copy_session(self._session, self.request_timeout) as tmpsession:
tmpsession.headers.update(self._default_http_header(empty_session_only=True))
@@ -538,6 +478,12 @@ class InstaloaderContext:
variables_json = json.dumps(variables, separators=(',', ':'))
if rhx_gis:
#self.log("rhx_gis {} query_hash {}".format(rhx_gis, query_hash))
values = "{}:{}".format(rhx_gis, variables_json)
x_instagram_gis = hashlib.md5(values.encode()).hexdigest()
tmpsession.headers['x-instagram-gis'] = x_instagram_gis
resp_json = self.get_json('graphql/query',
params={'query_hash': query_hash,
'variables': variables_json},
@@ -546,52 +492,10 @@ class InstaloaderContext:
self.error("GraphQL response did not contain a \"status\" field.")
return resp_json
def doc_id_graphql_query(self, doc_id: str, variables: Dict[str, Any],
referer: Optional[str] = None) -> Dict[str, Any]:
"""
Do a doc_id-based GraphQL Query using method POST.
.. versionadded:: 4.13
:param doc_id: doc_id for the query.
:param variables: Variables for the Query.
:param referer: HTTP Referer, or None.
:return: The server's response dictionary.
"""
csrf = next((c.value for c in self._session.cookies
if c.name == 'csrftoken' and c.value), None)
if not csrf:
self._session.get('https://www.instagram.com/', timeout=self.request_timeout)
csrf = next((c.value for c in self._session.cookies
if c.name == 'csrftoken' and c.value), '')
with copy_session(self._session, self.request_timeout) as tmpsession:
tmpsession.headers.update(self._default_http_header(empty_session_only=True))
del tmpsession.headers['Connection']
del tmpsession.headers['Content-Length']
tmpsession.headers['authority'] = 'www.instagram.com'
tmpsession.headers['scheme'] = 'https'
tmpsession.headers['accept'] = '*/*'
tmpsession.headers['x-csrftoken'] = csrf
if referer is not None:
tmpsession.headers['referer'] = urllib.parse.quote(referer)
variables_json = json.dumps(variables, separators=(',', ':'))
resp_json = self.get_json('graphql/query',
params={'variables': variables_json,
'doc_id': doc_id,
'server_timestamps': 'true'},
session=tmpsession,
use_post=True)
if 'status' not in resp_json:
self.error("GraphQL response did not contain a \"status\" field.")
return resp_json
def graphql_node_list(self, query_hash: str, query_variables: Dict[str, Any],
query_referer: Optional[str],
edge_extractor: Callable[[Dict[str, Any]], Dict[str, Any]],
_rhx_gis: Optional[str] = None,
rhx_gis: Optional[str] = None,
first_data: Optional[Dict[str, Any]] = None) -> Iterator[Dict[str, Any]]:
"""
Retrieve a list of GraphQL nodes.
@@ -603,7 +507,7 @@ class InstaloaderContext:
def _query():
query_variables['first'] = self._graphql_page_length
try:
return edge_extractor(self.graphql_query(query_hash, query_variables, query_referer))
return edge_extractor(self.graphql_query(query_hash, query_variables, query_referer, rhx_gis))
except QueryReturnedBadRequestException:
new_page_length = int(self._graphql_page_length / 2)
if new_page_length >= 12:
@@ -688,38 +592,6 @@ class InstaloaderContext:
return response
def get_page_data(self, path: str) -> List[Dict[str, Any]]:
"""Load a page of ``www.instagram.com`` as a logged-out visitor and return the data it embeds.
Instagram renders public profile and post pages with the results of their GraphQL queries
embedded in the HTML, even when it refuses anonymous API requests for the same data.
:param path: URL, relative to ``www.instagram.com/``
:return: The ``data`` object of each query result that the page embeds, in page order.
:raises QueryReturnedNotFoundException: When the server responds with a 404.
:raises ConnectionException: When the request failed.
"""
self.do_sleep()
self._rate_controller.wait_before_query('other')
url = 'https://www.instagram.com/{0}'.format(path)
# Instagram only embeds the data when the request looks like a browser's page navigation.
headers = {'Accept': 'text/html', 'Sec-Fetch-Dest': 'document', 'Sec-Fetch-Mode': 'navigate',
'Sec-Fetch-Site': 'none'}
try:
with self.get_anonymous_session() as anonymous_session:
resp = anonymous_session.get(url, headers=headers, allow_redirects=False)
except requests.exceptions.RequestException as err:
raise ConnectionException("Page request to {}: {}".format(url, err)) from err
if resp.status_code == 404:
raise QueryReturnedNotFoundException(self._response_error(resp))
if resp.status_code != 200:
raise ConnectionException(self._response_error(resp))
data: List[Dict[str, Any]] = []
for script in re.finditer(r'<script type="application/json"[^>]*>(.*?)</script>', resp.text, re.DOTALL):
with suppress(json.decoder.JSONDecodeError):
data.extend(_embedded_query_data(json.loads(script.group(1))))
return data
def write_raw(self, resp: Union[bytes, requests.Response], filename: str) -> None:
"""Write raw response data into a file.
@@ -748,11 +620,11 @@ class InstaloaderContext:
else:
if resp.status_code == 403:
# suspected invalid URL signature
raise QueryReturnedForbiddenException(self._response_error(resp))
raise QueryReturnedForbiddenException("403 when accessing {}.".format(url))
if resp.status_code == 404:
# 404 not worth retrying.
raise QueryReturnedNotFoundException(self._response_error(resp))
raise ConnectionException(self._response_error(resp))
raise QueryReturnedNotFoundException("404 when accessing {}.".format(url))
raise ConnectionException("HTTP error code {}.".format(resp.status_code))
def get_and_write_raw(self, url: str, filename: str) -> None:
"""Downloads and writes anonymously-requested raw data into a file.
@@ -778,11 +650,22 @@ class InstaloaderContext:
else:
if resp.status_code == 403:
# suspected invalid URL signature
raise QueryReturnedForbiddenException(self._response_error(resp))
raise QueryReturnedForbiddenException("403 when accessing {}.".format(url))
if resp.status_code == 404:
# 404 not worth retrying.
raise QueryReturnedNotFoundException(self._response_error(resp))
raise ConnectionException(self._response_error(resp))
raise QueryReturnedNotFoundException("404 when accessing {}.".format(url))
raise ConnectionException("HTTP error code {}.".format(resp.status_code))
@property
def root_rhx_gis(self) -> Optional[str]:
"""rhx_gis string returned in the / query."""
if self.is_logged_in:
# At the moment, rhx_gis seems to be required for anonymous requests only. By returning None when logged
# in, we can save the root_rhx_gis lookup query.
return None
if self._root_rhx_gis is None:
self._root_rhx_gis = self.get_json('', {}).get('rhx_gis', '')
return self._root_rhx_gis or None
class RateController:
-9
View File
@@ -18,7 +18,6 @@ class LatestStamps:
POST_TIMESTAMP = 'post-timestamp'
TAGGED_TIMESTAMP = 'tagged-timestamp'
IGTV_TIMESTAMP = 'igtv-timestamp'
REELS_TIMESTAMP = 'reels-timestamp'
STORY_TIMESTAMP = 'story-timestamp'
ISO_FORMAT = '%Y-%m-%dT%H:%M:%S.%f%z'
@@ -88,14 +87,6 @@ class LatestStamps:
"""Sets timestamp of last download of a profile's tagged posts."""
self._set_timestamp(profile_name, self.TAGGED_TIMESTAMP, timestamp)
def get_last_reels_timestamp(self, profile_name: str) -> datetime:
"""Returns timestamp of last download of a profile's reels posts."""
return self._get_timestamp(profile_name, self.REELS_TIMESTAMP)
def set_last_reels_timestamp(self, profile_name: str, timestamp: datetime):
"""Sets timestamp of last download of a profile's reels posts."""
self._set_timestamp(profile_name, self.REELS_TIMESTAMP, timestamp)
def get_last_igtv_timestamp(self, profile_name: str) -> datetime:
"""Returns timestamp of last download of a profile's igtv posts."""
return self._get_timestamp(profile_name, self.IGTV_TIMESTAMP)
+23 -48
View File
@@ -7,11 +7,11 @@ from datetime import datetime, timedelta
from lzma import LZMAError
from typing import Any, Callable, Dict, Iterable, Iterator, NamedTuple, Optional, Tuple, TypeVar
from .exceptions import InvalidArgumentException
from .exceptions import AbortDownloadException, InvalidArgumentException, QueryReturnedBadRequestException
from .instaloadercontext import InstaloaderContext
class FrozenNodeIterator(NamedTuple):
query_hash: Optional[str]
query_hash: str
query_variables: Dict
query_referer: Optional[str]
context_username: Optional[str]
@@ -19,7 +19,6 @@ class FrozenNodeIterator(NamedTuple):
best_before: Optional[float]
remaining_data: Optional[Dict]
first_node: Optional[Dict]
doc_id: Optional[str]
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."""
@@ -29,7 +28,6 @@ FrozenNodeIterator.best_before.__doc__ = """Date when parts of the stored nodes
FrozenNodeIterator.remaining_data.__doc__ = \
"""The already-retrieved, yet-unprocessed ``edges`` and the ``page_info`` at time of freezing."""
FrozenNodeIterator.first_node.__doc__ = """Node data of the first item, if an item has been produced."""
FrozenNodeIterator.doc_id.__doc__ = """The GraphQL ``doc_id`` parameter."""
T = TypeVar('T')
@@ -66,27 +64,22 @@ class NodeIterator(Iterator[T]):
NodeIterators are matching if and only if they have the same magic.
See also :func:`resumable_iteration` for a high-level context manager that handles a resumable iteration.
.. versionchanged: 4.13
Included support for `doc_id`-based queries (using POST method).
"""
_graphql_page_length = 12
_graphql_page_length = 50
_shelf_life = timedelta(days=29)
def __init__(self,
context: InstaloaderContext,
query_hash: Optional[str],
query_hash: str,
edge_extractor: Callable[[Dict[str, Any]], Dict[str, Any]],
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,
doc_id: Optional[str] = None):
is_first: Optional[Callable[[T, Optional[T]], bool]] = None):
self._context = context
self._query_hash = query_hash
self._doc_id = doc_id
self._edge_extractor = edge_extractor
self._node_wrapper = node_wrapper
self._query_variables = query_variables if query_variables is not None else {}
@@ -102,38 +95,26 @@ class NodeIterator(Iterator[T]):
self._is_first = is_first
def _query(self, after: Optional[str] = None) -> Dict:
if self._doc_id is not None:
return self._query_doc_id(self._doc_id, after)
else:
assert self._query_hash is not None
return self._query_query_hash(self._query_hash, after)
def _query_doc_id(self, doc_id: str, after: Optional[str] = None) -> Dict:
pagination_variables: Dict[str, Any] = {'__relay_internal__pv__PolarisFeedShareMenurelayprovider': False}
if after is not None:
pagination_variables['after'] = after
pagination_variables['before'] = None
pagination_variables['first'] = 12
pagination_variables['last'] = None
data = self._edge_extractor(
self._context.doc_id_graphql_query(
doc_id, {**self._query_variables, **pagination_variables}, self._query_referer
)
)
self._best_before = datetime.now() + NodeIterator._shelf_life
return data
def _query_query_hash(self, query_hash: str, after: Optional[str] = None) -> Dict:
pagination_variables: Dict[str, Any] = {'first': NodeIterator._graphql_page_length}
if after is not None:
pagination_variables['after'] = after
data = self._edge_extractor(
self._context.graphql_query(
query_hash, {**self._query_variables, **pagination_variables}, self._query_referer
try:
data = self._edge_extractor(
self._context.graphql_query(
self._query_hash, {**self._query_variables, **pagination_variables}, self._query_referer
)
)
)
self._best_before = datetime.now() + NodeIterator._shelf_life
return data
self._best_before = datetime.now() + NodeIterator._shelf_life
return data
except QueryReturnedBadRequestException:
new_page_length = int(NodeIterator._graphql_page_length / 2)
if new_page_length >= 12:
NodeIterator._graphql_page_length = new_page_length
self._context.error("HTTP Error 400 (Bad Request) on GraphQL Query. Retrying with shorter page length.",
repeat_at_end=False)
return self._query(after)
else:
raise
def __iter__(self):
return self
@@ -203,10 +184,6 @@ class NodeIterator(Iterator[T]):
"""
return self._node_wrapper(self._first_node) if self._first_node is not None else None
@staticmethod
def page_length() -> int:
return NodeIterator._graphql_page_length
def freeze(self) -> FrozenNodeIterator:
"""Freeze the iterator for later resuming."""
remaining_data = None
@@ -222,7 +199,6 @@ class NodeIterator(Iterator[T]):
best_before=self._best_before.timestamp() if self._best_before else None,
remaining_data=remaining_data,
first_node=self._first_node,
doc_id=self._doc_id,
)
def thaw(self, frozen: FrozenNodeIterator) -> None:
@@ -240,8 +216,7 @@ class NodeIterator(Iterator[T]):
if (self._query_hash != frozen.query_hash or
self._query_variables != frozen.query_variables or
self._query_referer != frozen.query_referer or
self._context.username != frozen.context_username or
self._doc_id != frozen.doc_id):
self._context.username != frozen.context_username):
raise InvalidArgumentException("Mismatching resume information.")
if not frozen.best_before:
raise InvalidArgumentException("\"best before\" date missing.")
@@ -318,7 +293,7 @@ def resumable_iteration(context: InstaloaderContext,
context.error("Warning: Not resuming from {}: {}".format(resume_file_path, exc))
try:
yield is_resuming, start_index
except (Exception, KeyboardInterrupt):
except (KeyboardInterrupt, AbortDownloadException):
if os.path.dirname(resume_file_path):
os.makedirs(os.path.dirname(resume_file_path), exist_ok=True)
save(iterator.freeze(), resume_file_path)
+79 -498
View File
@@ -20,7 +20,7 @@ class PostSidecarNode(NamedTuple):
"""Item of a Sidecar Post."""
is_video: bool
display_url: str
video_url: Optional[str]
video_url: str
PostSidecarNode.is_video.__doc__ = "Whether this node is a video."
@@ -43,91 +43,19 @@ PostCommentAnswer.owner.__doc__ = "Owner :class:`Profile` of the comment."
PostCommentAnswer.likes_count.__doc__ = "Number of likes on comment."
class PostComment:
def __init__(self, context: 'InstaloaderContext', node: Dict[str, Any],
answers: Iterator['PostCommentAnswer'], post: 'Post'):
self._context = context
self._node = node
self._answers = answers
self._post = post
class PostComment(NamedTuple):
id: int
created_at_utc: datetime
text: str
owner: 'Profile'
likes_count: int
answers: Iterator[PostCommentAnswer]
@classmethod
def from_iphone_struct(
cls,
context: "InstaloaderContext",
media: Dict[str, Any],
answers: Iterator["PostCommentAnswer"],
post: "Post",
):
return cls(
context=context,
node={
"id": int(media["pk"]),
"created_at": media["created_at"],
"text": media["text"],
"edge_liked_by": {
"count": media["comment_like_count"],
},
"iphone_struct": media,
},
answers=answers,
post=post,
)
@property
def id(self) -> int:
""" ID number of comment. """
return self._node['id']
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."
@property
def created_at_utc(self) -> datetime:
""" :class:`~datetime.datetime` when comment was created (UTC). """
return datetime.utcfromtimestamp(self._node['created_at'])
@property
def text(self):
""" Comment text. """
return self._node['text']
@property
def owner(self) -> "Profile":
""" Owner :class:`Profile` of the comment. """
if "iphone_struct" in self._node:
return Profile.from_iphone_struct(
self._context, self._node["iphone_struct"]["user"]
)
return Profile(self._context, self._node["owner"])
@property
def likes_count(self):
""" Number of likes on comment. """
return self._node.get('edge_liked_by', {}).get('count', 0)
@property
def answers(self) -> Iterator['PostCommentAnswer']:
""" Iterator which yields all :class:`PostCommentAnswer` for the comment. """
return self._answers
@property
def likes(self) -> Iterable['Profile']:
"""
Iterate over all likes of a comment. A :class:`Profile` instance of each like is yielded.
.. versionadded:: 4.11
"""
if self.likes_count != 0:
return NodeIterator(
self._context,
'5f0b1f6281e72053cbc07909c8d154ae',
lambda d: d['data']['comment']['edge_liked_by'],
lambda n: Profile(self._context, n),
{'comment_id': self.id},
'https://www.instagram.com/p/{0}/'.format(self._post.shortcode),
)
return []
def __repr__(self):
return f'<PostComment {self.id} of {self._post.shortcode}>'
class PostLocation(NamedTuple):
id: int
@@ -152,7 +80,7 @@ _hashtag_regex = re.compile(r"(?:#)((?:\w){1,150})")
# 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\n]|_)(?:@)(\w(?:(?:\w|(?:\.(?!\.))){0,28}(?:\w))?)", re.ASCII)
def _optional_normalize(string: Optional[str]) -> Optional[str]:
@@ -180,7 +108,7 @@ class Post:
This class unifies access to the properties associated with a post. It implements == and is
hashable.
:param context: :attr:`Instaloader.context` used for additional queries if necessary..
:param context: :attr:`Instaloader.context` used for additional queries if neccessary..
:param node: Node structure, as returned by Instagram.
:param owner_profile: The Profile of the owner, if already known at creation.
"""
@@ -232,126 +160,21 @@ class Post:
"title": media.get("title"),
"viewer_has_liked": media["has_liked"],
"edge_media_preview_like": {"count": media["like_count"]},
"accessibility_caption": media.get("accessibility_caption"),
"comments": media.get("comment_count"),
"iphone_struct": media,
}
with suppress(KeyError):
fake_node["display_url"] = media['image_versions2']['candidates'][0]['url']
with suppress(KeyError, TypeError):
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, TypeError):
fake_node["edge_sidecar_to_children"] = {"edges": [{"node":
Post._convert_iphone_carousel(node, media_types)}
for node in media["carousel_media"]]}
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 _convert_iphone_carousel(iphone_node: Dict[str, Any], media_types: Dict[int, str]) -> Dict[str, Any]:
fake_node = {
"display_url": iphone_node["image_versions2"]["candidates"][0]["url"],
"is_video": media_types[iphone_node["media_type"]] == "GraphVideo",
}
if "video_versions" in iphone_node and iphone_node["video_versions"] is not None:
fake_node["video_url"] = iphone_node["video_versions"][0]["url"]
return fake_node
@staticmethod
def _fetch_play_count_from_clips(context: 'InstaloaderContext', user_id: str,
shortcode: str) -> Optional[int]:
"""Fetch play_count for a reel via the clips connection endpoint as a fallback."""
try:
resp = context.doc_id_graphql_query(
"27234427476213202",
{"data": {"include_feed_video": True, "page_size": 12,
"target_user_id": str(user_id)}},
)
edges = ((resp.get("data") or {})
.get("xdt_api__v1__clips__user__connection_v2") or {})
for edge in edges.get("edges") or []:
media = (edge.get("node") or {}).get("media") or {}
if media.get("code") == shortcode:
return media.get("play_count")
except (ConnectionException, BadResponseException):
pass
return None
@staticmethod
def _normalize_post_data(media: Dict[str, Any], context: 'InstaloaderContext') -> Dict[str, Any]:
"""Normalize a Polaris media item to a legacy-compatible node, preserving the source structure."""
media_types = {1: "GraphImage", 2: "GraphVideo", 8: "GraphSidecar"}
media_type = media.get("media_type")
typename = None
if isinstance(media_type, int):
typename = media_types.get(media_type)
if not typename:
raise BadResponseException(f"Unknown media_type in metadata: {media_type}.")
pic_json: Dict[str, Any] = media.copy()
pic_json["shortcode"] = media["code"]
pic_json["id"] = media["pk"]
pic_json["__typename"] = typename
pic_json["is_video"] = media_type == 2
pic_json["taken_at_timestamp"] = media["taken_at"]
pic_json["owner"] = {
"id": media["user"]["pk"],
"username": media["user"].get("username", ""),
"full_name": media["user"].get("full_name", ""),
}
candidates = (media.get("image_versions2") or {}).get("candidates") or []
pic_json["display_url"] = candidates[0]["url"] if candidates else None
video_versions = media.get("video_versions") or []
pic_json["video_url"] = video_versions[0]["url"] if video_versions else None
pic_json["video_duration"] = media.get("video_duration")
pic_json["video_view_count"] = media.get("view_count")
pic_json["video_play_count"] = media.get("play_count")
if media_type == 2 and pic_json["video_view_count"] is None:
pic_json["video_play_count"] = Post._fetch_play_count_from_clips(
context, media["user"]["pk"], media["code"]
)
caption = media.get("caption")
caption_text = caption.get("text") if isinstance(caption, dict) else None
pic_json["edge_media_to_caption"] = (
{"edges": [{"node": {"text": caption_text}}]} if caption_text is not None
else {"edges": []}
)
pic_json["edge_media_preview_like"] = {"count": media.get("like_count") or 0}
pic_json["edge_media_to_parent_comment"] = {
"count": media.get("comment_count") or 0,
"edges": [],
}
if media.get("has_liked") is not None:
pic_json["viewer_has_liked"] = media["has_liked"]
carousel = media.get("carousel_media") or []
if carousel:
carousel_nodes = []
for item in carousel:
item_type = item.get("media_type", 1)
node: Dict[str, Any] = {
"shortcode": item.get("code", ""),
"__typename": media_types.get(item_type, "GraphImage"),
"is_video": item_type == 2,
}
item_candidates = (item.get("image_versions2") or {}).get("candidates") or []
node["display_url"] = item_candidates[0]["url"] if item_candidates else ""
item_videos = item.get("video_versions") or []
node["video_url"] = item_videos[0]["url"] if item_videos else None
if item.get("accessibility_caption") is not None:
node["accessibility_caption"] = item["accessibility_caption"]
carousel_nodes.append({"node": node})
pic_json["edge_sidecar_to_children"] = {"edges": carousel_nodes}
tagged = (media.get("usertags") or {}).get("in") or []
if tagged:
pic_json["edge_media_to_tagged_user"] = {
"edges": [
{"node": {"user": {"username": t["user"]["username"].lower()}}}
for t in tagged
if (t.get("user") or {}).get("username")
]
}
return pic_json
@staticmethod
def shortcode_to_mediaid(code: str) -> int:
if len(code) > 11:
@@ -413,18 +236,13 @@ class Post:
def _obtain_metadata(self):
if not self._full_metadata_dict:
resp = self._context.doc_id_graphql_query(
"27128499623469141",
{
"shortcode": self.shortcode,
"__relay_internal__pv__PolarisAIGMMediaWebLabelEnabledrelayprovider": False,
},
pic_json = self._context.graphql_query(
'2b0673e0dc4580674a88d426fe00ea90',
{'shortcode': self.shortcode}
)
web_info = (resp.get("data") or {}).get("xdt_api__v1__media__shortcode__web_info") or {}
items = web_info.get("items")
if not items:
self._full_metadata_dict = pic_json['data']['shortcode_media']
if self._full_metadata_dict is None:
raise BadResponseException("Fetching Post metadata failed.")
self._full_metadata_dict = Post._normalize_post_data(items[0], self._context)
if self.shortcode != self._full_metadata_dict['shortcode']:
self._node.update(self._full_metadata_dict)
raise PostChangedException
@@ -440,7 +258,7 @@ class Post:
if not self._context.iphone_support:
raise IPhoneSupportDisabledException("iPhone support is disabled.")
if not self._context.is_logged_in:
raise LoginRequiredException("Login required to access iPhone media info endpoint.")
raise LoginRequiredException("--login required to access iPhone media info endpoint.")
if not self._iphone_struct_:
data = self._context.get_iphone_json(path='api/v1/media/{}/info/'.format(self.mediaid), params={})
self._iphone_struct_ = data['items'][0]
@@ -601,14 +419,14 @@ class Post:
@property
def caption_hashtags(self) -> List[str]:
"""List of all lowercased hashtags (without preceding #) that occur in the Post's caption."""
"""List of all lowercased hashtags (without preceeding #) that occur in the Post's caption."""
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 Post's caption, without preceding @."""
"""List of all lowercased profiles that are mentioned in the Post's caption, without preceeding @."""
if not self.caption:
return []
return _mention_regex.findall(self.caption.lower())
@@ -691,15 +509,6 @@ class Post:
return self._field('video_view_count')
return None
@property
def video_play_count(self) -> Optional[int]:
"""Play count of the video, or None.
.. versionadded:: 4.14.3"""
if self.is_video:
return self._field('video_play_count')
return None
@property
def video_duration(self) -> Optional[float]:
"""Duration of the video in seconds, or None.
@@ -736,77 +545,16 @@ class Post:
except KeyError:
return self._field('edge_media_to_comment', 'count')
def _get_comments_via_iphone_endpoint(self) -> Iterable[PostComment]:
"""
Iterate over all comments of the post via an iPhone endpoint.
.. versionadded:: 4.10.3
fallback for :issue:`2125`.
"""
def _query(min_id=None):
pagination_params = {"min_id": min_id} if min_id is not None else {}
return self._context.get_iphone_json(
f"api/v1/media/{self.mediaid}/comments/",
{
"can_support_threading": "true",
"permalink_enabled": "false",
**pagination_params,
},
)
def _answers(comment_node):
def _answer(child_comment):
return PostCommentAnswer(
id=int(child_comment["pk"]),
created_at_utc=datetime.utcfromtimestamp(child_comment["created_at"]),
text=child_comment["text"],
owner=Profile.from_iphone_struct(self._context, child_comment["user"]),
likes_count=child_comment["comment_like_count"],
)
child_comment_count = comment_node["child_comment_count"]
if child_comment_count == 0:
return
preview_child_comments = comment_node["preview_child_comments"]
if child_comment_count == len(preview_child_comments):
yield from (
_answer(child_comment) for child_comment in preview_child_comments
)
return
pk = comment_node["pk"]
answers_json = self._context.get_iphone_json(
f"api/v1/media/{self.mediaid}/comments/{pk}/child_comments/",
{"max_id": ""},
)
yield from (
_answer(child_comment) for child_comment in answers_json["child_comments"]
)
def _paginated_comments(comments_json):
for comment_node in comments_json.get("comments", []):
yield PostComment.from_iphone_struct(
self._context, comment_node, _answers(comment_node), self
)
next_min_id = comments_json.get("next_min_id")
if next_min_id:
yield from _paginated_comments(_query(next_min_id))
return _paginated_comments(_query())
def get_comments(self) -> Iterable[PostComment]:
"""Iterate over all comments of the post.
r"""Iterate over all comments of the post.
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`])
id (int), owner (:class:`Profile`) and answers (:class:`~typing.Iterator`\ [:class:`PostCommentAnswer`])
if available.
.. versionchanged:: 4.7
Change return type to ``Iterable``.
"""
if not self._context.is_logged_in:
raise LoginRequiredException("Login required to access comments of a post.")
def _postcommentanswer(node):
return PostCommentAnswer(id=int(node['id']),
created_at_utc=datetime.utcfromtimestamp(node['created_at']),
@@ -836,28 +584,18 @@ class Post:
)
def _postcomment(node):
return PostComment(context=self._context, node=node,
answers=_postcommentanswers(node), post=self)
return PostComment(*_postcommentanswer(node),
answers=_postcommentanswers(node))
if self.comments == 0:
# Avoid doing additional requests if there are no comments
return []
try:
comment_edges = self._field("edge_media_to_parent_comment", "edges")
except KeyError:
comment_edges = self._field("edge_media_to_comment", "edges")
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)
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
return [_postcomment(comment['node']) for comment in comment_edges]
if self.comments > NodeIterator.page_length():
# comments pagination via our graphql query does not work reliably anymore (issue #2125), fallback to an
# iphone endpoint if needed.
return self._get_comments_via_iphone_endpoint()
return NodeIterator(
self._context,
'97b41c52301f77ce508f55e66d17620e',
@@ -875,7 +613,7 @@ class Post:
Require being logged in (as required by Instagram).
"""
if not self._context.is_logged_in:
raise LoginRequiredException("Login required to access likes of a post.")
raise LoginRequiredException("--login required to access likes of a post.")
if self.likes == 0:
# Avoid doing additional requests if there are no comments
return
@@ -940,11 +678,7 @@ class Post:
@property
def is_pinned(self) -> bool:
"""
.. deprecated: 4.10.3
This information is not returned by IG anymore
Used to return True if this Post has been pinned by at least one user, now likely returns always false.
"""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'])
@@ -978,8 +712,7 @@ class Profile:
Also, this class implements == and is hashable.
"""
def __init__(self, context: InstaloaderContext, node: Dict[str, Any]):
assert "username" in node
assert "id" in node or "pk" in node
assert 'username' in node
self._context = context
self._has_public_story: Optional[bool] = None
self._node = node
@@ -999,30 +732,10 @@ class Profile:
:param username: Username
:raises: :class:`ProfileNotExistsException`
"""
# Resolve the profile through its public web page, which Instagram renders with the
# profile embedded. The web_profile_info endpoint previously used here is refused
# with HTTP 400 for some professional profiles, and with HTTP 401 or 429 for some
# sessions, both anonymously and when logged in.
try:
page = context.get_page_data("{0}/".format(username.lower()))
except QueryReturnedNotFoundException:
page = []
# The page embeds two queries for the profile: one for its details and one for its
# posts.
user: Dict[str, Any] = {}
for data in page:
user.update(data.get("xig_user_by_username") or {})
if not user.get("pk"):
raise ProfileNotExistsException("Profile {} does not exist.".format(username))
user.pop("polaris_ordered_timeline_connection", None)
# The page's "id" is not the user ID, which is "pk".
node = dict(user, id=user["pk"])
if context.is_logged_in:
# The page shows the profile as a logged-out visitor sees it. Leave the missing
# fields, notably the viewer's relationship to the profile, such as
# followed_by_viewer, to the GraphQL profile query in _obtain_metadata().
return cls(context, node)
return cls(context, cls._normalize_profile_data(node))
# pylint:disable=protected-access
profile = cls(context, {'username': username.lower()})
profile._obtain_metadata() # to raise ProfileNotExistsException now in case username is invalid
return profile
@classmethod
def from_id(cls, context: InstaloaderContext, profile_id: int):
@@ -1035,16 +748,16 @@ class Profile:
"""
if profile_id in context.profile_id_cache:
return context.profile_id_cache[profile_id]
# Resolve the current username from the user id, then load the full profile.
# The GraphQL user query previously used here started responding with HTTP 400.
try:
user = context.get_json(
"api/v1/users/{0}/info/".format(profile_id), params={}
).get('user')
except QueryReturnedNotFoundException:
user = None
if user and user.get('username'):
profile = cls.from_username(context, user['username'])
data = context.graphql_query('7c16654f22c819fb63d1183034a5162f',
{'user_id': str(profile_id),
'include_chaining': False,
'include_reel': True,
'include_suggested_users': False,
'include_logged_out_extras': False,
'include_highlight_reels': False},
rhx_gis=context.root_rhx_gis)['data']['user']
if data:
profile = cls(context, data['reel']['owner'])
else:
raise ProfileNotExistsException("No profile found, the user may have blocked you (ID: " +
str(profile_id) + ").")
@@ -1073,7 +786,7 @@ class Profile:
.. versionadded:: 4.5.2"""
if not context.is_logged_in:
raise LoginRequiredException("Login required to access own profile.")
raise LoginRequiredException("--login required to access own profile.")
return cls(context, context.graphql_query("d6f4427fbe92d846298cf93df0b937d3", {})["data"]["user"])
def _asdict(self):
@@ -1090,41 +803,11 @@ class Profile:
def _obtain_metadata(self):
try:
if not self._has_full_metadata:
if not self._context.is_logged_in:
# Anonymous access: the web_profile_info endpoint returns the full
# node in the legacy format and still works, unlike the GraphQL
# profile query.
try:
data = self._context.get_json(
"api/v1/users/web_profile_info/",
params={"username": self.username},
).get('data')
except QueryReturnedNotFoundException as err:
raise ProfileNotExistsException(
'Profile {} does not exist.'.format(self.username)) from err
user_data = data.get('user') if data else None
if user_data is None:
raise ProfileNotExistsException('Profile {} does not exist.'.format(self.username))
self._node = user_data
self._has_full_metadata = True
return
user_id = self._node.get('id') or self._node.get('pk')
variables = {
"id": str(user_id),
"render_surface": "PROFILE",
"__relay_internal__pv__PolarisCannesGuardianExperienceEnabledrelayprovider": True,
"__relay_internal__pv__PolarisCASB976ProfileEnabledrelayprovider": False,
"__relay_internal__pv__PolarisRepostsConsumptionEnabledrelayprovider": False,
"__relay_internal__pv__PolarisWebSchoolsEnabledrelayprovider": False,
"enable_integrity_filters": True,
}
data = self._context.doc_id_graphql_query('27937681195819736', variables)
if data is None:
raise QueryReturnedNotFoundException('GraphQL query returned None')
user_data = data.get('data', {}).get('user')
if user_data is None:
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 = self._normalize_profile_data(user_data)
self._node = metadata['data']['user']
self._has_full_metadata = True
except (QueryReturnedNotFoundException, KeyError) as err:
top_search_results = TopSearchResults(self._context, self.username)
@@ -1139,45 +822,6 @@ class Profile:
', '.join(similar_profiles[0:5]))) from err
raise ProfileNotExistsException('Profile {} does not exist.'.format(self.username)) from err
@staticmethod
def _normalize_profile_data(user_data: Dict[str, Any]) -> Dict[str, Any]:
"""Normalize PolarisProfilePageContentQuery response to match legacy format."""
normalized = user_data.copy()
if 'id' not in normalized and 'pk' in normalized:
normalized['id'] = normalized['pk']
if 'edge_owner_to_timeline_media' not in normalized and 'media_count' in normalized:
normalized['edge_owner_to_timeline_media'] = {'count': normalized['media_count']}
if 'edge_felix_video_timeline' not in normalized:
normalized['edge_felix_video_timeline'] = {'count': 0}
if 'edge_followed_by' not in normalized and 'follower_count' in normalized:
normalized['edge_followed_by'] = {'count': normalized['follower_count']}
if 'edge_follow' not in normalized and 'following_count' in normalized:
normalized['edge_follow'] = {'count': normalized['following_count']}
if 'is_business_account' not in normalized and 'is_business' in normalized:
normalized['is_business_account'] = normalized['is_business']
if 'business_category_name' not in normalized and 'category' in normalized:
normalized['business_category_name'] = normalized['category']
friendship = normalized.get('friendship_status', {}) or {}
if 'followed_by_viewer' not in normalized:
normalized['followed_by_viewer'] = friendship.get('following', False)
if 'follows_viewer' not in normalized:
normalized['follows_viewer'] = friendship.get('followed_by', False)
if 'blocked_by_viewer' not in normalized:
normalized['blocked_by_viewer'] = friendship.get('blocking', False)
if 'has_blocked_viewer' not in normalized:
normalized['has_blocked_viewer'] = False
if 'has_requested_viewer' not in normalized:
normalized['has_requested_viewer'] = friendship.get('incoming_request', False)
if 'requested_by_viewer' not in normalized:
normalized['requested_by_viewer'] = friendship.get('outgoing_request', False)
if 'profile_pic_url_hd' not in normalized:
hd_info = normalized.get('hd_profile_pic_url_info')
if hd_info and 'url' in hd_info:
normalized['profile_pic_url_hd'] = hd_info['url']
elif 'profile_pic_url' in normalized:
normalized['profile_pic_url_hd'] = normalized['profile_pic_url']
return normalized
def _metadata(self, *keys) -> Any:
try:
d = self._node
@@ -1196,7 +840,7 @@ class Profile:
if not self._context.iphone_support:
raise IPhoneSupportDisabledException("iPhone support is disabled.")
if not self._context.is_logged_in:
raise LoginRequiredException("Login required to access iPhone profile info endpoint.")
raise LoginRequiredException("--login required to access iPhone profile info endpoint.")
if not self._iphone_struct_:
data = self._context.get_iphone_json(path='api/v1/users/{}/info/'.format(self.userid), params={})
self._iphone_struct_ = data['user']
@@ -1268,7 +912,7 @@ class Profile:
@property
def biography_hashtags(self) -> List[str]:
"""
List of all lowercased hashtags (without preceding #) that occur in the Profile's biography.
List of all lowercased hashtags (without preceeding #) that occur in the Profile's biography.
.. versionadded:: 4.10
"""
@@ -1279,7 +923,7 @@ class Profile:
@property
def biography_mentions(self) -> List[str]:
"""
List of all lowercased profiles that are mentioned in the Profile's biography, without preceding @.
List of all lowercased profiles that are mentioned in the Profile's biography, without preceeding @.
.. versionadded:: 4.10
"""
@@ -1386,52 +1030,15 @@ class Profile:
:rtype: NodeIterator[Post]"""
self._obtain_metadata()
if not self._context.is_logged_in:
return NodeIterator(
context=self._context,
edge_extractor=lambda d: d["data"]["user"]["edge_owner_to_timeline_media"],
node_wrapper=lambda n: Post(self._context, n, self),
query_variables={
"data": {
"count": 12,
"include_relationship_info": True,
"latest_besties_reel_media": True,
"latest_reel_media": True,
},
"id": self.userid,
},
query_referer="https://www.instagram.com/{0}/".format(self.username),
is_first=Profile._make_is_newest_checker(),
doc_id="7950326061742207",
query_hash=None,
first_data=self._metadata("edge_owner_to_timeline_media"),
)
return NodeIterator(
context=self._context,
edge_extractor=lambda d: d["data"]["xdt_api__v1__feed__user_timeline_graphql_connection"],
node_wrapper=lambda n: Post.from_iphone_struct(self._context, n),
# The query and variables that the web app sends for a profile's posts tab
# (PolarisProfilePostsTabContentQuery_connection). Instagram answers the
# previous query, 7898261790222653, with "execution error".
query_variables={
"data": {
"count": 12,
"include_reel_media_seen_timestamp": True,
"include_relationship_info": True,
"latest_besties_reel_media": True,
"latest_reel_media": True,
},
"first": 12,
"include_multi_captions": True,
"username": self.username,
"__relay_internal__pv__PolarisMultiCaptionCarouselEnabledrelayprovider": True,
"__relay_internal__pv__PolarisShortDramaEnabledrelayprovider": False,
"__relay_internal__pv__PolarisReelsRecoDebugOverlayEnabledrelayprovider": False,
},
query_referer="https://www.instagram.com/{0}/".format(self.username),
is_first=Profile._make_is_newest_checker(),
doc_id="28975909992013618",
query_hash=None,
self._context,
'003056d32c2554def87228bc3fd9668a',
lambda d: d['data']['user']['edge_owner_to_timeline_media'],
lambda n: Post(self._context, n, self),
{'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]:
@@ -1440,7 +1047,7 @@ class Profile:
:rtype: NodeIterator[Post]"""
if self.username != self._context.username:
raise LoginRequiredException(f"Login as {self.username} required to get that profile's saved posts.")
raise LoginRequiredException("--login={} required to get that profile's saved posts.".format(self.username))
return NodeIterator(
self._context,
@@ -1468,30 +1075,6 @@ class Profile:
is_first=Profile._make_is_newest_checker()
)
def get_reels(self) -> NodeIterator[Post]:
"""Retrieve all reels from a profile.
:rtype: NodeIterator[Post]
.. versionadded:: 4.14.0
"""
self._obtain_metadata()
return NodeIterator(
context = self._context,
edge_extractor = lambda d: d['data']['xdt_api__v1__clips__user__connection_v2'],
# Reels post info is incomplete relative to regular posts so we create a Post from the shortcode
# and fetch the additional metadata with an additional API request per Reel
node_wrapper = lambda n: Post.from_shortcode(context=self._context, shortcode=n["media"]["code"]),
query_variables = {'data': {
'page_size': 12, 'include_feed_video': True, "target_user_id": str(self.userid)}},
query_referer = 'https://www.instagram.com/{0}/'.format(self.username),
is_first = Profile._make_is_newest_checker(),
# fb_api_req_friendly_name=PolarisProfileReelsTabContentQuery_connection
doc_id = '7845543455542541',
query_hash = None,
)
def get_igtv_posts(self) -> NodeIterator[Post]:
"""Retrieve all IGTV posts.
@@ -1524,7 +1107,7 @@ class Profile:
.. versionadded:: 4.10
"""
if not self._context.is_logged_in:
raise LoginRequiredException("Login required to get a profile's followers.")
raise LoginRequiredException("--login required to get a profile's followers.")
self._obtain_metadata()
return NodeIterator(
self._context,
@@ -1543,7 +1126,7 @@ class Profile:
:rtype: NodeIterator[Profile]
"""
if not self._context.is_logged_in:
raise LoginRequiredException("Login required to get a profile's followers.")
raise LoginRequiredException("--login required to get a profile's followers.")
self._obtain_metadata()
return NodeIterator(
self._context,
@@ -1562,7 +1145,7 @@ class Profile:
:rtype: NodeIterator[Profile]
"""
if not self._context.is_logged_in:
raise LoginRequiredException("Login required to get a profile's followees.")
raise LoginRequiredException("--login required to get a profile's followees.")
self._obtain_metadata()
return NodeIterator(
self._context,
@@ -1581,7 +1164,7 @@ class Profile:
.. versionadded:: 4.4
"""
if not self._context.is_logged_in:
raise LoginRequiredException("Login required to get a profile's similar accounts.")
raise LoginRequiredException("--login required to get a profile's similar accounts.")
self._obtain_metadata()
yield from (Profile(self._context, edge["node"]) for edge in
self._context.graphql_query("ad99dd9d3646cc3c0dda65debcd266a7",
@@ -1660,7 +1243,7 @@ class StoryItem:
if not self._context.iphone_support:
raise IPhoneSupportDisabledException("iPhone support is disabled.")
if not self._context.is_logged_in:
raise LoginRequiredException("Login required to access iPhone media info endpoint.")
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={}
@@ -1757,7 +1340,7 @@ class StoryItem:
@property
def caption_hashtags(self) -> List[str]:
"""
List of all lowercased hashtags (without preceding #) that occur in the StoryItem's caption.
List of all lowercased hashtags (without preceeding #) that occur in the StoryItem's caption.
.. versionadded:: 4.10
"""
@@ -1768,7 +1351,7 @@ class StoryItem:
@property
def caption_mentions(self) -> List[str]:
"""
List of all lowercased profiles that are mentioned in the StoryItem's caption, without preceding @.
List of all lowercased profiles that are mentioned in the StoryItem's caption, without preceeding @.
.. versionadded:: 4.10
"""
@@ -2067,11 +1650,11 @@ class Hashtag:
@classmethod
def from_name(cls, context: InstaloaderContext, name: str):
"""
Create a Hashtag instance from a given hashtag name, without preceding '#'. Raises an Exception if there is no
Create a Hashtag instance from a given hashtag name, without preceeding '#'. Raises an Exception if there is no
hashtag with the given name.
:param context: :attr:`Instaloader.context`
:param name: Hashtag, without preceding '#'
:param name: Hashtag, without preceeding '#'
:raises: :class:`QueryReturnedNotFoundException`
"""
# pylint:disable=protected-access
@@ -2081,13 +1664,11 @@ class Hashtag:
@property
def name(self):
"""Hashtag name lowercased, without preceding '#'"""
"""Hashtag name lowercased, without preceeding '#'"""
return self._node["name"].lower()
def _query(self, params):
json_response = self._context.get_iphone_json(
"api/v1/tags/web_info/", {**params, "tag_name": self.name}
)
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"]
def _obtain_metadata(self):
@@ -2410,7 +1991,7 @@ def load_structure(context: InstaloaderContext, json_structure: dict) -> JsonExp
"""Loads a :class:`Post`, :class:`Profile`, :class:`StoryItem`, :class:`Hashtag` or :class:`FrozenNodeIterator` from
a json structure.
:param context: :attr:`Instaloader.context` linked to the new object, used for additional queries if necessary.
:param context: :attr:`Instaloader.context` linked to the new object, used for additional queries if neccessary.
:param json_structure: Instaloader JSON structure
.. versionadded:: 4.8
@@ -2440,7 +2021,7 @@ def load_structure_from_file(context: InstaloaderContext, filename: str) -> Json
"""Loads a :class:`Post`, :class:`Profile`, :class:`StoryItem`, :class:`Hashtag` or :class:`FrozenNodeIterator` from
a '.json' or '.json.xz' file that has been saved by :func:`save_structure_to_file`.
:param context: :attr:`Instaloader.context` linked to the new object, used for additional queries if necessary.
:param context: :attr:`Instaloader.context` linked to the new object, used for additional queries if neccessary.
:param filename: Filename, ends in '.json' or '.json.xz'
"""
compressed = filename.endswith('.xz')
+6 -12
View File
@@ -18,13 +18,10 @@ def get_version():
raise SystemExit("Could not find version string.")
if sys.version_info < (3, 10):
sys.exit('Instaloader requires Python >= 3.10.')
if sys.version_info < (3, 8):
sys.exit('Instaloader requires Python >= 3.8.')
requirements = ['requests>=2.25']
optional_requirements = {
'browser_cookie3': ['browser_cookie3>=0.19.1'],
}
requirements = ['requests>=2.4']
keywords = (['instagram', 'instagram-scraper', 'instagram-client', 'instagram-feed', 'downloader', 'videos', 'photos',
'pictures', 'instagram-user-photos', 'instagram-photos', 'instagram-metadata', 'instagram-downloader',
@@ -46,8 +43,7 @@ setup(
'from Instagram.',
long_description=open(os.path.join(SRC, 'README.rst')).read(),
install_requires=requirements,
python_requires='>=3.10',
extras_require=optional_requirements,
python_requires='>=3.8',
entry_points={'console_scripts': ['instaloader=instaloader.__main__:main']},
zip_safe=False,
keywords=keywords,
@@ -58,12 +54,10 @@ setup(
'Intended Audience :: Developers',
'License :: OSI Approved :: MIT License',
'Operating System :: OS Independent',
'Programming Language :: Python :: 3.8',
'Programming Language :: Python :: 3.9',
'Programming Language :: Python :: 3.10',
'Programming Language :: Python :: 3.11',
'Programming Language :: Python :: 3.12',
'Programming Language :: Python :: 3.13',
'Programming Language :: Python :: 3.14',
'Programming Language :: Python :: 3.15',
'Programming Language :: Python :: 3 :: Only',
'Topic :: Internet',
'Topic :: Multimedia :: Graphics'
-4
View File
@@ -1,4 +0,0 @@
# Instaloader Tests
Integration tests for Instaloader features that require Instagram access are maintained here:
https://github.com/instaloader/instaloader-tests
+214
View File
@@ -0,0 +1,214 @@
"""Unit Tests for Instaloader"""
import os
import shutil
import tempfile
import unittest
from itertools import islice
from typing import Optional
import instaloader
PROFILE_WITH_HIGHLIGHTS = 325732271
PUBLIC_PROFILE = "selenagomez"
PUBLIC_PROFILE_ID = 460563723
PUBLIC_PROFILE_WITH_IGTV = "natgeo"
HASHTAG = "kitten"
LOCATION = "362629379"
OWN_USERNAME = "aandergr"
NORMAL_MAX_COUNT = 2
PAGING_MAX_COUNT = 15
PRIVATE_PROFILE = "aandergr"
PRIVATE_PROFILE_ID = 1706625676
EMPTY_PROFILE = "not_public"
EMPTY_PROFILE_ID = 1928659031
ratecontroller: Optional[instaloader.RateController] = None
class TestInstaloaderAnonymously(unittest.TestCase):
def setUp(self):
self.dir = tempfile.mkdtemp()
print("Testing in {}".format(self.dir))
os.chdir(self.dir)
self.L = instaloader.Instaloader(download_geotags=True,
download_comments=True,
save_metadata=True)
self.L.context.raise_all_errors = True
if ratecontroller is not None:
# pylint:disable=protected-access
ratecontroller._context = self.L.context
self.L.context._rate_controller = ratecontroller
def tearDown(self):
# pylint:disable=global-statement,protected-access
global ratecontroller
ratecontroller = self.L.context._rate_controller
self.L.close()
os.chdir('/')
print("Removing {}".format(self.dir))
shutil.rmtree(self.dir)
def post_paging_test(self, iterator):
previous_post = None
for post in islice(iterator, PAGING_MAX_COUNT):
print(post)
if previous_post:
self.assertTrue(post.date_utc < previous_post.date_utc)
previous_post = post
@unittest.skip('')
def test_public_profile_download(self):
profiles = {self.L.check_profile_id(PUBLIC_PROFILE)}
self.L.download_profiles(profiles, profile_pic=False, fast_update=True, raise_errors=True)
self.L.download_profiles(profiles, profile_pic=False, fast_update=True, raise_errors=True)
def test_public_profile_paging(self):
self.post_paging_test(instaloader.Profile.from_username(self.L.context, PUBLIC_PROFILE).get_posts())
def test_profile_pic_download(self):
self.L.download_profiles({self.L.check_profile_id(PUBLIC_PROFILE)}, posts=False, raise_errors=True)
def test_hashtag_download(self):
self.L.download_hashtag(HASHTAG, NORMAL_MAX_COUNT)
def test_hashtag_paging(self):
for count, post in enumerate(instaloader.Hashtag.from_name(self.L.context, HASHTAG).get_all_posts()):
print(post)
if count == PAGING_MAX_COUNT:
break
def test_get_id_by_username(self):
self.assertEqual(PUBLIC_PROFILE_ID,
instaloader.Profile.from_username(self.L.context, PUBLIC_PROFILE).userid)
def test_get_username_by_id_private(self):
self.assertEqual(PRIVATE_PROFILE.lower(),
instaloader.Profile.from_id(self.L.context, PRIVATE_PROFILE_ID).username)
def test_get_username_by_id_public(self):
self.assertEqual(PUBLIC_PROFILE.lower(),
instaloader.Profile.from_id(self.L.context, PUBLIC_PROFILE_ID).username)
def test_get_username_by_id_empty(self):
self.assertEqual(EMPTY_PROFILE.lower(),
instaloader.Profile.from_id(self.L.context, EMPTY_PROFILE_ID).username)
def test_get_username_by_name_empty(self):
self.assertEqual(EMPTY_PROFILE_ID,
instaloader.Profile.from_username(self.L.context, EMPTY_PROFILE).userid)
def test_post_from_mediaid(self):
for post in instaloader.Profile.from_username(self.L.context, PUBLIC_PROFILE).get_posts():
post2 = instaloader.Post.from_mediaid(self.L.context, post.mediaid)
self.assertEqual(post, post2)
break
def test_public_profile_tagged_paging(self):
for post in islice(instaloader.Profile.from_username(self.L.context, PUBLIC_PROFILE).get_tagged_posts(),
PAGING_MAX_COUNT):
print(post)
def test_public_profile_igtv(self):
for post in islice(instaloader.Profile.from_username(self.L.context, PUBLIC_PROFILE_WITH_IGTV).get_igtv_posts(),
PAGING_MAX_COUNT):
print(post)
class TestInstaloaderLoggedIn(TestInstaloaderAnonymously):
def setUp(self):
super().setUp()
self.L.load_session_from_file(OWN_USERNAME)
@unittest.skip('')
def test_stories_download(self):
self.L.download_stories()
@unittest.skip('')
def test_private_profile_download(self):
self.L.download_profiles({self.L.check_profile_id(PRIVATE_PROFILE)}, stories=True, raise_errors=True)
def test_stories_paging(self):
for user_story in self.L.get_stories():
print("profile {}.".format(user_story.owner_username))
for item in user_story.get_items():
print(item)
def test_highlights_paging(self):
for user_highlight in self.L.get_highlights(PROFILE_WITH_HIGHLIGHTS):
print("Retrieving {} highlights \"{}\" from profile {}".format(user_highlight.itemcount,
user_highlight.title,
user_highlight.owner_username))
for item in user_highlight.get_items():
print(item)
def test_private_profile_paging(self):
self.post_paging_test(instaloader.Profile.from_username(self.L.context, PRIVATE_PROFILE).get_posts())
def test_feed_download(self):
self.L.download_feed_posts(NORMAL_MAX_COUNT)
def test_feed_paging(self):
for count, post in enumerate(self.L.get_feed_posts()):
print(post)
if count == PAGING_MAX_COUNT:
break
def test_saved_download(self):
self.L.download_saved_posts(NORMAL_MAX_COUNT)
def test_saved_paging(self):
for count, post in enumerate(instaloader.Profile.from_username(self.L.context, OWN_USERNAME).get_saved_posts()):
print(post)
if count == PAGING_MAX_COUNT:
break
def test_test_login(self):
self.assertEqual(OWN_USERNAME, self.L.test_login())
def test_followees_and_stories(self):
profile = instaloader.Profile.from_username(self.L.context, OWN_USERNAME)
followees = set(islice(profile.get_followees(), PAGING_MAX_COUNT))
self.L.download_profiles(followees, profile_pic=False, stories=True, posts=False, raise_errors=True)
def test_get_followees(self):
profile = instaloader.Profile.from_username(self.L.context, OWN_USERNAME)
for f in profile.get_followees():
print(f.username)
def test_get_followers(self):
profile = instaloader.Profile.from_username(self.L.context, OWN_USERNAME)
for f in profile.get_followers():
print(f.username)
def test_get_similar_accounts(self):
profile = instaloader.Profile.from_username(self.L.context, OWN_USERNAME)
for f in profile.get_similar_accounts():
print(f.username)
def test_get_likes(self):
for post in instaloader.Profile.from_username(self.L.context, OWN_USERNAME).get_posts():
for like in post.get_likes():
print(like.username)
break
def test_explore_paging(self):
for count, post in enumerate(self.L.get_explore_posts()):
print(post)
if count == PAGING_MAX_COUNT:
break
def test_location_download(self):
self.L.download_location(LOCATION, NORMAL_MAX_COUNT)
def test_location_paging(self):
for count, post in enumerate(self.L.get_location_posts(LOCATION)):
print(post)
if count == PAGING_MAX_COUNT:
break
if __name__ == '__main__':
unittest.main()