Compare commits

..
34 changed files with 1773 additions and 2283 deletions
-31
View File
@@ -1,31 +0,0 @@
name: "Documentation"
on: [push, pull_request]
jobs:
docs:
name: "Documentation"
runs-on: ubuntu-latest
steps:
- name: "Checkout Instaloader Repository"
uses: actions/checkout@v2
with:
fetch-depth: 0 # needed for building docs
- name: "Setup Python"
uses: actions/setup-python@v2
with:
python-version: 3.12
- name: "Install Dependencies"
run: |
python -m pip install pipenv==2023.12.1
pipenv --python `python --version | grep -Eo '3\.[0-9]+'` sync --dev
- name: "Build Documentation"
run: pipenv --python `python --version | grep -Eo '3\.[0-9]+'` run make -C docs html SPHINXOPTS="-W -n"
- name: "Deploy Documentation"
if: github.event_name == 'push' && github.ref == 'refs/heads/master'
uses: JamesIves/github-pages-deploy-action@v4.2.5
with:
branch: master
repository-name: instaloader/instaloader.github.io
folder: docs/_build/html
token: ${{ secrets.PAGES_DEPLOYMENT_TOKEN }}
+10 -6
View File
@@ -7,21 +7,25 @@ jobs:
name: PyLint and MyPy
runs-on: ubuntu-latest
strategy:
fail-fast: false
matrix:
python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"]
python-version: [3.8, 3.9]
steps:
- name: Checkout Instaloader Repository
uses: actions/checkout@v2
with:
fetch-depth: 0 # needed for building docs
- name: Setup Python
uses: actions/setup-python@v2
with:
python-version: ${{ matrix.python-version }}
- name: Install Dependencies
run: |
python -m pip install pipenv==2023.12.1
pipenv --python `python --version | grep -Eo '3\.[0-9]+'` sync --dev
python -m pip install pipenv==2020.11.15
pipenv sync --dev
- name: PyLint
run: pipenv --python `python --version | grep -Eo '3\.[0-9]+'` run pylint instaloader
run: pipenv run pylint instaloader
- name: MyPy
run: pipenv --python `python --version | grep -Eo '3\.[0-9]+'` run mypy -m instaloader
run: pipenv run mypy -m instaloader
- name: Build Documentation
if: matrix.python-version == '3.8'
run: pipenv run make -C docs html SPHINXOPTS="-W -n"
-38
View File
@@ -1,38 +0,0 @@
name: "Publish Packages"
on:
push:
tags:
- v*
jobs:
publish:
name: "Publish Package"
runs-on: ubuntu-latest
steps:
- name: "Checkout Repository"
uses: actions/checkout@v2
- name: "Setup Python"
uses: actions/setup-python@v2
with:
python-version: 3.12
- name: "Get the tagged version"
id: get_version
env:
GITHUB_REF: ${{ github.ref }}
run: echo ::set-output name=VERSION::${GITHUB_REF/refs\/tags\/v/}
shell: bash
- name: "Install Dependencies"
run: python -m pip install setuptools
- name: "Generate Python Package"
run: python setup.py sdist
- name: "Publish Distribution to PyPI"
uses: pypa/gh-action-pypi-publish@master
with:
password: ${{ secrets.PYPI_PASSWORD }}
- name: "Publish Distribution to AUR"
run: deploy/arch/deploy.sh $VERSION_TAG
env:
VERSION_TAG: v${{ steps.get_version.outputs.VERSION }}
AUR_KEY: ${{ secrets.AUR_KEY }}
AUR_IV: ${{ secrets.AUR_IV }}
+2 -2
View File
@@ -18,7 +18,7 @@ jobs:
stale-issue-message: 'There has been no activity on this question for an extended period of time. This issue will be closed after further 14 days of inactivity.'
stale-issue-label: 'stale'
exempt-issue-labels: 'leave open'
days-before-stale: 21
days-before-stale: 15
days-before-close: -1
remove-stale-when-updated: false
- uses: actions/stale@v1
@@ -30,5 +30,5 @@ jobs:
stale-pr-label: 'stale'
exempt-issue-label: 'leave open'
exempt-pr-label: 'leave open'
days-before-stale: 189
days-before-stale: 135
days-before-close: 14
+2 -2
View File
@@ -10,9 +10,9 @@ jobs:
- name: Checkout Instaloader repository
uses: actions/checkout@v2
- name: Setup Python
uses: actions/setup-python@v2
uses: actions/setup-python@v1
with:
python-version: "3.12"
python-version: 3.9
architecture: x64
- name: Get the tagged version
id: get_version
+405 -467
View File
File diff suppressed because it is too large Load Diff
+35
View File
@@ -0,0 +1,35 @@
dist: xenial
language: python
python:
- "3.6"
- "3.7"
install:
- pipenv sync --dev
script:
- pylint instaloader
- mypy -m instaloader
- make -C docs html SPHINXOPTS="-W -n"
addons:
ssh_known_hosts: aur.archlinux.org
deploy:
- provider: pypi
user: aandergr
on:
tags: true
python: 3.6
password:
secure: KBsATtXNnEzqr0X2ALdyx2zaYj2EUfK8YjP4jEOilBELpJ4evHYXVgRkn9CKXGsD7LKbWmR3gz2rr8NZ6U42pHw0fpQHizz91MpJSKbNY7RZZh4UVWMDh81izeglchWAF+m3Wdpxpo/GBDbrZIbRQFhPgB+kjSzJo3uiCS+yVXtGG+6csJHzd0WkJjsVHz+XSdxyOiLy/qFF3nmY6ZrVrgw6WbX2YLbx9je6DxK3YRrnBqgMgH7Ui2xWWbgJCk3KKpQTMiMKnTa9mdeMJAd26YMCz1gG/X7ck6VNly/6bs5+Zq7/bU0dSolhPBhNthyccYdS0IeGPdPF3YqatblJ4fLGTfHn8oJs3/T11yIW9pvOT3UftEynPOUX+en2Qnb/3WfgwUfVts8i8DAYgEFLGEzbF5XMGy4gwAD/bZ0JW0pmgAG2PZsdOokAt+ydlNj/uxq1GFW2doZbul6k4t03p1mik6itYscvuhKoEm8Yw6wr1P9aDvMR6FFpSu4gDt2XrKmRmWkSXxpNeCfZujTjyAnISxLVQGSY7BG5K0Z2HbDtqqNz9nj4WgNGzZmngiRofCevWR+vPIDUDW7NCX03csfcFS9RRllBO/BrL7NVUV+Qp9dfTQYp+8SqYLWgHtDoXdgyMREjJaAmbx2u/TYTY8bADZ5F3vGUpv2j5hP3NTU=
- provider: pages
skip-cleanup: true
github-token: $GITHUB_TOKEN
local-dir: docs/_build/html
repo: instaloader/instaloader.github.io
target-branch: master
on:
branch: master
python: 3.6
- provider: script
script: deploy/arch/deploy.sh $TRAVIS_TAG
on:
tags: true
python: 3.6
-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
+6 -9
View File
@@ -6,17 +6,14 @@ verify_ssl = true
[dev-packages]
pylint = "*"
mypy = "*"
sphinx = "*"
sphinx = "~=2.3"
pyinstaller = "*"
pefile = "*"
pywin32-ctypes = "*"
psutil = "*"
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
typing-extensions = "*"
[packages]
requests = "*"
browser-cookie3 = "*"
pycryptodome = "*"
pynacl = "*"
Generated
+397 -584
View File
File diff suppressed because it is too large Load Diff
+5 -4
View File
@@ -2,7 +2,11 @@
.. badges-start
|pypi| |pyversion| |license| |aur| |contributors| |downloads|
|travis| |pypi| |pyversion| |license| |aur| |contributors| |downloads|
.. |travis| image:: https://img.shields.io/travis/instaloader/instaloader/master.svg
:alt: Travis-CI Build Status
:target: https://travis-ci.org/instaloader/instaloader
.. |pypi| image:: https://img.shields.io/pypi/v/instaloader.svg
:alt: Instaloader PyPI Project Page
@@ -123,9 +127,6 @@ Supporters
.. current-sponsors-start
| Instaloader is proudly sponsored by
| `@rocketapi-io <https://github.com/rocketapi-io>`__
See `Alex' GitHub Sponsors <https://github.com/sponsors/aandergr>`__ page for
how you can sponsor the development of Instaloader!
+1 -1
View File
@@ -6,7 +6,7 @@ pkgbase = instaloader
arch = any
license = MIT
makedepends = python-setuptools
depends = python>=3.8
depends = python>=3.6
depends = python-requests>=2.4
options = !emptydirs
source = instaloader-{{version}}.tar.gz::https://codeload.github.com/instaloader/instaloader/tar.gz/v{{version}}
+1 -1
View File
@@ -8,7 +8,7 @@ arch=('any')
url="https://instaloader.github.io/"
license=('MIT')
groups=()
depends=('python>=3.8' 'python-requests>=2.4')
depends=('python>=3.6' 'python-requests>=2.4')
makedepends=('python-setuptools')
options=('!emptydirs')
source=($pkgname-$pkgver.tar.gz::https://codeload.github.com/instaloader/instaloader/tar.gz/v$pkgver)
+1 -3
View File
@@ -8,12 +8,10 @@ echo $VERSION | grep -qe "[abc]" && exit 0
cd $(dirname $0)
# decrypt and add ssh key
openssl aes-256-cbc -K $AUR_KEY -iv $AUR_IV -in id_rsa_AUR.enc -out /tmp/AUR_openssh -d
openssl aes-256-cbc -K $encrypted_354637631c28_key -iv $encrypted_354637631c28_iv -in id_rsa_AUR.enc -out /tmp/AUR_openssh -d
eval "$(ssh-agent -s)"
chmod 600 /tmp/AUR_openssh
ssh-add /tmp/AUR_openssh
mkdir -p ~/.ssh
ssh-keyscan -t rsa aur.archlinux.org >> ~/.ssh/known_hosts
# clone and modify AUR repo
git clone --depth 1 ssh://aur@aur.archlinux.org/instaloader.git
Binary file not shown.
+2 -2
View File
@@ -44,8 +44,8 @@ with open('__main__.py', 'w+') as f:
f.writelines(lines)
# install dependencies and invoke PyInstaller
commands = ["pip install pipenv==2023.12.1",
f"pipenv --python {sys.version_info.major}.{sys.version_info.minor} sync --dev",
commands = ["pip install pipenv==2020.11.15",
"pipenv sync --dev",
"pipenv run pyinstaller --log-level=DEBUG instaloader.spec"]
for command in commands:
+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">
+3 -31
View File
@@ -196,25 +196,6 @@ Instaloader to login.
Login name (profile name) for your Instagram account.
.. option:: --load-cookies BROWSER-NAME, -b BROWSER-NAME
Use Instagram cookie in your browser to login.
This feature requires the browser_cookie3 library.
Compatible with :option:`--cookiefile` if you want to load cookies from browser profiles.
Incompatible with :option:`--login` due to potential username mismatch between user input and browser login.
Supported browsers: Chrome, Firefox, Edge, Opera, Safari, Brave.
After loading the cookies run the :option:`--login` option as it is required to download high quality media
and to make full use of Instaloader's features.
.. versionadded:: 4.11
.. option:: --cookiefile COOKIE-FILE, -B COOKIE-FILE
Cookie file path of a browser profile to load cookies from.
.. versionadded:: 4.11
.. option:: --sessionfile SESSIONFILE, -f SESSIONFILE
Path for loading and storing session key file. Defaults to
@@ -258,13 +239,6 @@ How to Download
.. versionadded:: 4.8
.. option:: --sanitize-paths
Force sanitization of paths so that the resulting file and directory names
are valid on both Windows and Unix.
.. versionadded:: 4.9
.. option:: --resume-prefix prefix
For many targets, Instaloader is capable of resuming a previously-aborted
@@ -277,8 +251,7 @@ How to Download
- Profile posts,
- Profile IGTV posts (:option:`--igtv`),
- Profile tagged posts (:option:`--tagged`),
- Saved posts (``:saved``),
- Hashtags.
- Saved posts (``:saved``).
This feature is enabled by default for targets where it is supported;
:option:`--resume-prefix` only changes the name of the iterator files.
@@ -300,7 +273,7 @@ How to Download
.. option:: --user-agent USER_AGENT
User Agent to use for HTTP requests. Per default, Instaloader pretends being
Chrome/120 on Linux.
Chrome/89 on Linux.
.. option:: --max-connection-attempts N
@@ -348,13 +321,12 @@ Miscellaneous Options
Read arguments from file `args.txt`, a shortcut to provide arguments from
file rather than command-line. This provides a convenient way to hide login
info from CLI, and can also be used to simplify management of long arguments.
You can provide more than one file at once, e.g.: ``+args1.txt +args2.txt``.
.. note::
Text file should separate arguments with line breaks.
`args.txt` example::
args.txt example::
--login=MYUSERNAME
--password=MYPASSWORD
+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.kennethreitz.org/en/master/', 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)
+1 -1
View File
@@ -8,7 +8,7 @@ Install Instaloader
To **install Instaloader**,
#. Ensure that you have `Python <https://www.python.org/>`__, at least
version 3.8, and `pip <https://pypi.python.org/pypi/pip>`__
version 3.6, and `pip <https://pypi.python.org/pypi/pip>`__
installed.
#. Then, install Instaloader using::
+1 -1
View File
@@ -30,7 +30,7 @@ Iterator :class:`NodeIterator` and a context manager
It can be serialized and deserialized with :func:`save_structure_to_file`
and :func:`load_structure_from_file`, as well as with :mod:`json` and
:mod:`pickle` thanks to being a :class:`~typing.NamedTuple`.
:mod:`pickle` thanks to being a :func:`~collections.namedtuple`.
``resumable_iteration``
"""""""""""""""""""""""
-8
View File
@@ -71,14 +71,6 @@ TopSearchResults
.. versionadded:: 4.3
TitlePic
""""""""
.. autoclass:: TitlePic
:no-show-inheritance:
.. versionadded:: 4.8
Loading and Saving
""""""""""""""""""
+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
+2 -3
View File
@@ -1,7 +1,7 @@
"""Download pictures (or videos) along with their captions and other metadata from Instagram."""
__version__ = '4.11'
__version__ = '4.8.1'
try:
@@ -18,6 +18,5 @@ 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, TitlePic,
load_structure_from_file, save_structure_to_file,
Profile, Story, StoryItem, TopSearchResults, load_structure_from_file, save_structure_to_file,
load_structure, get_json_structure)
+4 -80
View File
@@ -5,7 +5,6 @@ import datetime
import os
import re
import sys
import textwrap
from argparse import ArgumentParser, ArgumentTypeError, SUPPRESS
from typing import List, Optional
@@ -15,11 +14,6 @@ from . import (AbortDownloadException, BadCredentialsException, Instaloader, Ins
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
def usage_string():
@@ -51,6 +45,7 @@ def filterstr_to_filterfunc(filter_str: str, item_type: type):
class TransformFilterAst(ast.NodeTransformer):
def visit_Name(self, node: ast.Name):
# pylint:disable=no-self-use
if not isinstance(node.ctx, ast.Load):
raise InvalidArgumentException("Invalid filter: Modifying variables ({}) not allowed.".format(node.id))
if node.id == "datetime":
@@ -73,53 +68,6 @@ def filterstr_to_filterfunc(filter_str: str, item_type: type):
return filterfunc
def get_cookies_from_instagram(domain, browser, cookie_file='', cookie_name=''):
supported_browsers = {
"chrome": browser_cookie3.chrome,
"firefox": browser_cookie3.firefox,
"edge": browser_cookie3.edge,
"brave": browser_cookie3.brave,
"opera": browser_cookie3.opera,
"safari": browser_cookie3.safari
}
if browser not in supported_browsers:
print("Loading cookies from the specified browser failed")
print("Supported browsers are Chrome, Firefox, Edge, Brave, Opera and Safari")
return {}
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:
print(f"No cookies found for Instagram in {browser}, Are you logged in succesfully in {browser}?")
if cookie_name:
return cookies.get(cookie_name, {})
else:
return cookies
def import_session(browser, instaloader, cookiefile):
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 SystemExit(f"Not logged in. Are you logged in successfully in {browser}?")
instaloader.context.username = username
print(f"{username} has been successfully logged in.")
next_step_text = (f"Next: Run instaloader --login={username} as it is required to download high quality media "
"and to make full use of instaloader's features.")
print(textwrap.fill(next_step_text))
def _main(instaloader: Instaloader, targetlist: List[str],
username: Optional[str] = None, password: Optional[str] = None,
sessionfile: Optional[str] = None,
@@ -131,9 +79,7 @@ def _main(instaloader: Instaloader, targetlist: List[str],
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) -> None:
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
@@ -148,11 +94,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 SystemExit("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):
@@ -168,10 +109,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: ")
@@ -420,10 +357,6 @@ def main():
'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>"))
@@ -450,9 +383,6 @@ def main():
g_how.add_argument('--resume-prefix', metavar='PREFIX',
help='Prefix for filenames that are used to save the information to resume an interrupted '
'download.')
g_how.add_argument('--sanitize-paths', action='store_true',
help='Sanitize paths so that the resulting file and directory names are valid on both '
'Windows and Unix.')
g_how.add_argument('--no-resume', action='store_true',
help='Do not resume a previously-aborted download iteration, and do not save such information '
'when interrupted.')
@@ -512,9 +442,6 @@ def main():
if args.no_pictures and args.fast_update:
raise SystemExit('--no-pictures and --fast-update cannot be used together.')
if args.login and args.load_cookies:
raise SystemExit('--load-cookies and --login cannot be used together.')
# Determine what to download
download_profile_pic = not args.no_profile_pic or args.profile_pic_only
download_posts = not (args.no_posts or args.stories_only or args.profile_pic_only)
@@ -536,8 +463,7 @@ def main():
slide=args.slide,
fatal_status_codes=args.abort_on,
iphone_support=not args.no_iphone,
title_pattern=args.title_pattern,
sanitize_paths=args.sanitize_paths)
title_pattern=args.title_pattern)
_main(loader,
args.profile,
username=args.login.lower() if args.login is not None else None,
@@ -553,9 +479,7 @@ def main():
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)
storyitem_filter_str=args.storyitem_filter)
loader.close()
except InstaloaderException as err:
raise SystemExit("Fatal error: %s" % err) from err
+67 -133
View File
@@ -22,7 +22,6 @@ from .exceptions import *
from .instaloadercontext import InstaloaderContext, RateController
from .lateststamps import LatestStamps
from .nodeiterator import NodeIterator, resumable_iteration
from .sectioniterator import SectionIterator
from .structures import (Hashtag, Highlight, JsonExportable, Post, PostLocation, Profile, Story, StoryItem,
load_structure_from_file, save_structure_to_file, PostSidecarNode, TitlePic)
@@ -137,38 +136,20 @@ class _ArbitraryItemFormatter(string.Formatter):
class _PostPathFormatter(_ArbitraryItemFormatter):
RESERVED: set = {'CON', 'PRN', 'AUX', 'NUL',
'COM1', 'COM2', 'COM3', 'COM4', 'COM5', 'COM6', 'COM7', 'COM8', 'COM9',
'LPT1', 'LPT2', 'LPT3', 'LPT4', 'LPT5', 'LPT6', 'LPT7', 'LPT8', 'LPT9'}
def __init__(self, item: Any, force_windows_path: bool = False):
super().__init__(item)
self.force_windows_path = force_windows_path
def get_value(self, key, args, kwargs):
ret = super().get_value(key, args, kwargs)
if not isinstance(ret, str):
return ret
return self.sanitize_path(ret, self.force_windows_path)
return self.sanitize_path(ret)
@staticmethod
def sanitize_path(ret: str, force_windows_path: bool = False) -> str:
def sanitize_path(ret: str) -> str:
"""Replaces '/' with similar looking Division Slash and some other illegal filename characters on Windows."""
ret = ret.replace('/', '\u2215')
if ret.startswith('.'):
ret = ret.replace('.', '\u2024', 1)
if force_windows_path or platform.system() == 'Windows':
if platform.system() == 'Windows':
ret = ret.replace(':', '\uff1a').replace('<', '\ufe64').replace('>', '\ufe65').replace('\"', '\uff02')
ret = ret.replace('\\', '\ufe68').replace('|', '\uff5c').replace('?', '\ufe16').replace('*', '\uff0a')
ret = ret.replace('\n', ' ').replace('\r', ' ')
root, ext = os.path.splitext(ret)
if root.upper() in _PostPathFormatter.RESERVED:
root += '_'
if ext == '.':
ext = '\u2024'
ret = root + ext
return ret
@@ -201,7 +182,6 @@ class Instaloader:
:param slide: :option:`--slide`
:param fatal_status_codes: :option:`--abort-on`
:param iphone_support: not :option:`--no-iphone`
:param sanitize_paths: :option:`--sanitize-paths`
.. attribute:: context
@@ -221,8 +201,8 @@ class Instaloader:
download_comments: bool = False,
save_metadata: bool = True,
compress_json: bool = True,
post_metadata_txt_pattern: Optional[str] = None,
storyitem_metadata_txt_pattern: Optional[str] = None,
post_metadata_txt_pattern: str = None,
storyitem_metadata_txt_pattern: str = None,
max_connection_attempts: int = 3,
request_timeout: float = 300.0,
rate_controller: Optional[Callable[[InstaloaderContext], RateController]] = None,
@@ -231,8 +211,7 @@ class Instaloader:
slide: Optional[str] = None,
fatal_status_codes: Optional[List[int]] = None,
iphone_support: bool = True,
title_pattern: Optional[str] = None,
sanitize_paths: bool = False):
title_pattern: Optional[str] = None):
self.context = InstaloaderContext(sleep, quiet, user_agent, max_connection_attempts,
request_timeout, rate_controller, fatal_status_codes,
@@ -249,7 +228,6 @@ class Instaloader:
self.title_pattern = '{date_utc}_UTC_{typename}'
else:
self.title_pattern = '{target}_{date_utc}_UTC_{typename}'
self.sanitize_paths = sanitize_paths
self.download_pictures = download_pictures
self.download_videos = download_videos
self.download_video_thumbnails = download_video_thumbnails
@@ -313,8 +291,7 @@ class Instaloader:
check_resume_bbd=self.check_resume_bbd,
slide=self.slide,
fatal_status_codes=self.context.fatal_status_codes,
iphone_support=self.context.iphone_support,
sanitize_paths=self.sanitize_paths)
iphone_support=self.context.iphone_support)
yield new_loader
self.context.error_log.extend(new_loader.context.error_log)
new_loader.context.error_log = [] # avoid double-printing of errors
@@ -335,25 +312,15 @@ class Instaloader:
filename_suffix: Optional[str] = None, _attempt: int = 1) -> bool:
"""Downloads and saves picture with given url under given directory with given timestamp.
Returns true, if file was actually downloaded, i.e. updated."""
if filename_suffix is not None:
filename += '_' + filename_suffix
urlmatch = re.search('\\.[a-z0-9]*\\?', url)
file_extension = url[-3:] if urlmatch is None else urlmatch.group(0)[1:-1]
nominal_filename = filename + '.' + file_extension
if os.path.isfile(nominal_filename):
self.context.log(nominal_filename + ' exists', end=' ', flush=True)
return False
resp = self.context.get_raw(url)
if 'Content-Type' in resp.headers and resp.headers['Content-Type']:
header_extension = '.' + resp.headers['Content-Type'].split(';')[0].split('/')[-1]
header_extension = header_extension.lower().replace('jpeg', 'jpg')
filename += header_extension
else:
filename = nominal_filename
if filename != nominal_filename and os.path.isfile(filename):
if filename_suffix is not None:
filename += '_' + filename_suffix
filename += '.' + file_extension
if os.path.isfile(filename):
self.context.log(filename + ' exists', end=' ', flush=True)
return False
self.context.write_raw(resp, filename)
self.context.get_and_write_raw(url, filename)
os.utime(filename, (datetime.now().timestamp(), mtime.timestamp()))
return True
@@ -447,7 +414,7 @@ class Instaloader:
"""Updates picture caption / Post metadata info"""
def _elliptify(caption):
pcaption = caption.replace('\n', ' ').strip()
return '[' + ((pcaption[:29] + "\u2026") if len(pcaption) > 31 else pcaption) + ']'
return '[' + ((pcaption[:29] + u"\u2026") if len(pcaption) > 31 else pcaption) + ']'
filename += '.txt'
caption += '\n'
pcaption = _elliptify(caption)
@@ -478,19 +445,17 @@ class Instaloader:
self.context.log(pcaption, end=' ', flush=True)
except UnicodeEncodeError:
self.context.log('txt', end=' ', flush=True)
with open(filename, 'w', encoding='UTF-8') as fio:
fio.write(caption)
with open(filename, 'wb') as text_file:
with BytesIO(bcaption) as bio:
shutil.copyfileobj(cast(IO, bio), text_file)
os.utime(filename, (datetime.now().timestamp(), mtime.timestamp()))
def save_location(self, filename: str, location: PostLocation, mtime: datetime) -> None:
"""Save post location name and Google Maps link."""
filename += '_location.txt'
if location.lat is not None and location.lng is not None:
location_string = (location.name + "\n" +
"https://maps.google.com/maps?q={0},{1}&ll={0},{1}\n".format(location.lat,
location.lng))
else:
location_string = location.name
location_string = (location.name + "\n" +
"https://maps.google.com/maps?q={0},{1}&ll={0},{1}\n".format(location.lat,
location.lng))
with open(filename, 'wb') as text_file:
with BytesIO(location_string.encode()) as bio:
shutil.copyfileobj(cast(IO, bio), text_file)
@@ -524,7 +489,7 @@ class Instaloader:
.. versionadded:: 4.3"""
http_response = self.context.get_raw(url)
date_object: Optional[datetime] = None
date_object = None # type: Optional[datetime]
if 'Last-Modified' in http_response.headers:
date_object = datetime.strptime(http_response.headers["Last-Modified"], '%a, %d %b %Y %H:%M:%S GMT')
date_object = date_object.replace(tzinfo=timezone.utc)
@@ -533,10 +498,9 @@ class Instaloader:
pic_bytes = http_response.content
ig_filename = url.split('/')[-1].split('?')[0]
pic_data = TitlePic(owner_profile, target, name_suffix, ig_filename, date_object)
dirname = _PostPathFormatter(pic_data, self.sanitize_paths).format(self.dirname_pattern, target=target)
filename_template = os.path.join(
dirname,
_PostPathFormatter(pic_data, self.sanitize_paths).format(self.title_pattern, target=target))
dirname = _PostPathFormatter(pic_data).format(self.dirname_pattern, target=target)
filename_template = os.path.join(dirname,
_PostPathFormatter(pic_data).format(self.title_pattern, target=target))
filename = self.__prepare_filename(filename_template, lambda: url) + ".jpg"
content_length = http_response.headers.get('Content-Length', None)
if os.path.isfile(filename) and (not self.context.is_logged_in or
@@ -562,7 +526,7 @@ class Instaloader:
if latest_stamps is None:
self.download_profilepic(profile)
return
profile_pic_basename = profile.profile_pic_url_no_iphone.split('/')[-1].split('?')[0]
profile_pic_basename = profile.profile_pic_url.split('/')[-1].split('?')[0]
saved_basename = latest_stamps.get_profile_pic(profile.username)
if saved_basename == profile_pic_basename:
return
@@ -585,23 +549,6 @@ class Instaloader:
.. versionadded:: 4.4"""
self.download_title_pic(hashtag.profile_pic_url, '#' + hashtag.name, 'profile_pic', None)
@_requires_login
def save_session(self) -> dict:
"""Saves internally stored :class:`requests.Session` object to :class:`dict`.
:raises LoginRequiredException: If called without being logged in.
.. versionadded:: 4.10
"""
return self.context.save_session()
def load_session(self, username: str, session_data: dict) -> None:
"""Internally stores :class:`requests.Session` object from :class:`dict`.
.. versionadded:: 4.10
"""
self.context.load_session(username, session_data)
@_requires_login
def save_session_to_file(self, filename: Optional[str] = None) -> None:
"""Saves internally stored :class:`requests.Session` object.
@@ -678,7 +625,7 @@ class Instaloader:
"""Format filename of a :class:`Post` or :class:`StoryItem` according to ``filename-pattern`` parameter.
.. versionadded:: 4.1"""
return _PostPathFormatter(item, self.sanitize_paths).format(self.filename_pattern, target=target)
return _PostPathFormatter(item).format(self.filename_pattern, target=target)
def download_post(self, post: Post, target: Union[str, Path]) -> bool:
"""
@@ -710,7 +657,7 @@ class Instaloader:
return False
return True
dirname = _PostPathFormatter(post, self.sanitize_paths).format(self.dirname_pattern, target=target)
dirname = _PostPathFormatter(post).format(self.dirname_pattern, target=target)
filename_template = os.path.join(dirname, self.format_filename(post, target=target))
filename = self.__prepare_filename(filename_template, lambda: post.url)
@@ -729,7 +676,7 @@ class Instaloader:
post.get_sidecar_nodes(self.slide_start, self.slide_end),
start=self.slide_start % post.mediacount + 1
):
suffix: Optional[str] = str(edge_number)
suffix = str(edge_number) # type: Optional[str]
if '{filename}' in self.filename_pattern:
suffix = None
if self.download_pictures and (not sidecar_node.is_video or self.download_video_thumbnails):
@@ -861,7 +808,7 @@ class Instaloader:
last_scraped = latest_stamps.get_last_story_timestamp(name)
scraped_timestamp = datetime.now().astimezone()
for item in user_story.get_items():
if latest_stamps is not None and item.date_local <= last_scraped:
if latest_stamps is not None and item.date_utc.replace(tzinfo=timezone.utc) <= last_scraped:
break
if storyitem_filter is not None and not storyitem_filter(item):
self.context.log("<{} skipped>".format(item), flush=True)
@@ -891,22 +838,17 @@ class Instaloader:
return True
date_local = item.date_local
dirname = _PostPathFormatter(item, self.sanitize_paths).format(self.dirname_pattern, target=target)
dirname = _PostPathFormatter(item).format(self.dirname_pattern, target=target)
filename_template = os.path.join(dirname, self.format_filename(item, target=target))
filename = self.__prepare_filename(filename_template, lambda: item.url)
downloaded = False
video_url_fetch_failed = False
if item.is_video and self.download_videos is True:
video_url = item.video_url
if video_url:
filename = self.__prepare_filename(filename_template, lambda: str(video_url))
downloaded |= (not _already_downloaded(filename + ".mp4") and
self.download_pic(filename=filename, url=video_url, mtime=date_local))
else:
video_url_fetch_failed = True
if video_url_fetch_failed or not item.is_video or self.download_video_thumbnails is True:
if not item.is_video or self.download_video_thumbnails is True:
downloaded = (not _already_downloaded(filename + ".jpg") and
self.download_pic(filename=filename, url=item.url, mtime=date_local))
if item.is_video and self.download_videos is True:
filename = self.__prepare_filename(filename_template, lambda: str(item.video_url))
downloaded |= (not _already_downloaded(filename + ".mp4") and
self.download_pic(filename=filename, url=item.video_url, mtime=date_local))
# Save caption if desired
metadata_string = _ArbitraryItemFormatter(item).format(self.storyitem_metadata_txt_pattern).strip()
if metadata_string:
@@ -962,11 +904,10 @@ class Instaloader:
"""
for user_highlight in self.get_highlights(user):
name = user_highlight.owner_username
highlight_target: Union[str, Path] = (filename_target
highlight_target = (filename_target
if filename_target
else (Path(_PostPathFormatter.sanitize_path(name, self.sanitize_paths)) /
_PostPathFormatter.sanitize_path(user_highlight.title,
self.sanitize_paths)))
else (Path(_PostPathFormatter.sanitize_path(name)) /
_PostPathFormatter.sanitize_path(user_highlight.title))) # type: Union[str, Path]
self.context.log("Retrieving highlights \"{}\" from profile {}".format(user_highlight.title, name))
self.download_highlight_cover(user_highlight, highlight_target)
totalcount = user_highlight.itemcount
@@ -991,8 +932,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.
@@ -1004,9 +944,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`.
@@ -1015,14 +952,12 @@ 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)
sanitized_target = target
if isinstance(target, str):
sanitized_target = _PostPathFormatter.sanitize_path(target, self.sanitize_paths)
sanitized_target = _PostPathFormatter.sanitize_path(target)
if takewhile is None:
takewhile = lambda _: True
with resumable_iteration(
@@ -1037,10 +972,7 @@ class Instaloader:
enabled=self.resume_prefix is not None
) as (is_resuming, start_index):
for number, post in enumerate(posts, start=start_index + 1):
should_stop = not takewhile(post)
if should_stop and number <= possibly_pinned:
continue
if (max_count is not None and number > max_count) or should_stop:
if (max_count is not None and number > max_count) or not takewhile(post):
break
if displayed_count is not None:
self.context.log("[{0:{w}d}/{1:{w}d}] ".format(number, displayed_count,
@@ -1072,7 +1004,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:
# disengage fast_update for first post when resuming
if not is_resuming or number > 0:
break
@@ -1103,7 +1035,7 @@ class Instaloader:
'has_stories': False})["data"]
@_requires_login
def download_feed_posts(self, max_count: Optional[int] = None, fast_update: bool = False,
def download_feed_posts(self, max_count: int = None, fast_update: bool = False,
post_filter: Optional[Callable[[Post], bool]] = None) -> None:
"""
Download pictures from the user's feed.
@@ -1124,7 +1056,7 @@ class Instaloader:
self.posts_download_loop(self.get_feed_posts(), ":feed", fast_update, post_filter, max_count=max_count)
@_requires_login
def download_saved_posts(self, max_count: Optional[int] = None, fast_update: bool = False,
def download_saved_posts(self, max_count: int = None, fast_update: bool = False,
post_filter: Optional[Callable[[Post], bool]] = None) -> None:
"""Download user's saved pictures.
@@ -1152,12 +1084,18 @@ class Instaloader:
.. versionchanged:: 4.2.9
Require being logged in (as required by Instagram)
"""
yield from SectionIterator(
self.context,
lambda d: d["native_location_data"]["recent"],
lambda m: Post.from_iphone_struct(self.context, m),
f"explore/locations/{location}/",
)
has_next_page = True
end_cursor = None
while has_next_page:
if end_cursor:
params = {'__a': 1, 'max_id': end_cursor}
else:
params = {'__a': 1}
location_data = self.context.get_json('explore/locations/{0}/'.format(location),
params)['graphql']['location']['edge_location_to_media']
yield from (Post(self.context, edge['node']) for edge in location_data['edges'])
has_next_page = location_data['page_info']['has_next_page']
end_cursor = location_data['page_info']['end_cursor']
@_requires_login
def download_location(self, location: str,
@@ -1206,8 +1144,8 @@ class Instaloader:
"""Get Posts associated with a #hashtag.
.. deprecated:: 4.4
Use :meth:`Hashtag.get_posts_resumable`."""
return Hashtag.from_name(self.context, hashtag).get_posts_resumable()
Use :meth:`Hashtag.get_posts`."""
return Hashtag.from_name(self.context, hashtag).get_posts()
def download_hashtag(self, hashtag: Union[Hashtag, str],
max_count: Optional[int] = None,
@@ -1243,7 +1181,7 @@ class Instaloader:
self.download_hashtag_profilepic(hashtag)
if posts:
self.context.log("Retrieving pictures with hashtag #{}...".format(hashtag.name))
self.posts_download_loop(hashtag.get_posts_resumable(), target, fast_update, post_filter,
self.posts_download_loop(hashtag.get_all_posts(), target, fast_update, post_filter,
max_count=max_count)
if self.save_metadata:
json_filename = '{0}/{1}'.format(self.dirname_pattern.format(profile=target,
@@ -1265,15 +1203,15 @@ class Instaloader:
posts_takewhile: Optional[Callable[[Post], bool]] = None
if latest_stamps is not None:
last_scraped = latest_stamps.get_last_tagged_timestamp(profile.username)
posts_takewhile = lambda p: p.date_local > last_scraped
posts_takewhile = lambda p: p.date_utc.replace(tzinfo=timezone.utc) > last_scraped
tagged_posts = profile.get_tagged_posts()
self.posts_download_loop(tagged_posts,
target if target
else (Path(_PostPathFormatter.sanitize_path(profile.username, self.sanitize_paths)) /
_PostPathFormatter.sanitize_path(':tagged', self.sanitize_paths)),
else (Path(_PostPathFormatter.sanitize_path(profile.username)) /
_PostPathFormatter.sanitize_path(':tagged')),
fast_update, post_filter, takewhile=posts_takewhile)
if latest_stamps is not None and tagged_posts.first_item is not None:
latest_stamps.set_last_tagged_timestamp(profile.username, tagged_posts.first_item.date_local)
latest_stamps.set_last_tagged_timestamp(profile.username, tagged_posts.first_item.date_local.astimezone())
def download_igtv(self, profile: Profile, fast_update: bool = False,
post_filter: Optional[Callable[[Post], bool]] = None,
@@ -1288,12 +1226,12 @@ class Instaloader:
posts_takewhile: Optional[Callable[[Post], bool]] = None
if latest_stamps is not None:
last_scraped = latest_stamps.get_last_igtv_timestamp(profile.username)
posts_takewhile = lambda p: p.date_local > last_scraped
posts_takewhile = lambda p: p.date_utc.replace(tzinfo=timezone.utc) > last_scraped
igtv_posts = profile.get_igtv_posts()
self.posts_download_loop(igtv_posts, profile.username, fast_update, post_filter,
total_count=profile.igtvcount, owner_profile=profile, takewhile=posts_takewhile)
if latest_stamps is not None and igtv_posts.first_item is not None:
latest_stamps.set_last_igtv_timestamp(profile.username, igtv_posts.first_item.date_local)
latest_stamps.set_last_igtv_timestamp(profile.username, igtv_posts.first_item.date_local.astimezone())
def _get_id_filename(self, profile_name: str) -> str:
if ((format_string_contains_key(self.dirname_pattern, 'profile') or
@@ -1364,10 +1302,6 @@ class Instaloader:
profile_id))
profile_from_id = Profile.from_id(self.context, profile_id)
newname = profile_from_id.username
if profile_name == newname:
self.context.error(
f"Warning: Profile {profile_name} could not be retrieved by its name, but by its ID.")
return profile_from_id
self.context.error("Profile {0} has changed its name to {1}.".format(profile_name, newname))
if latest_stamps is None:
if ((format_string_contains_key(self.dirname_pattern, 'profile') or
@@ -1490,14 +1424,14 @@ class Instaloader:
if latest_stamps is not None:
# pylint:disable=cell-var-from-loop
last_scraped = latest_stamps.get_last_post_timestamp(profile_name)
posts_takewhile = lambda p: p.date_local > last_scraped
posts_takewhile = lambda p: p.date_utc.replace(tzinfo=timezone.utc) > last_scraped
posts_to_download = profile.get_posts()
self.posts_download_loop(posts_to_download, profile_name, fast_update, post_filter,
total_count=profile.mediacount, owner_profile=profile,
takewhile=posts_takewhile, possibly_pinned=3)
takewhile=posts_takewhile)
if latest_stamps is not None and posts_to_download.first_item is not None:
latest_stamps.set_last_post_timestamp(profile_name,
posts_to_download.first_item.date_local)
posts_to_download.first_item.date_local.astimezone())
if stories and profiles:
with self.context.error_catcher("Download stories"):
+88 -143
View File
@@ -1,21 +1,25 @@
import base64
import hashlib
import json
import os
import pickle
import random
import re
import shutil
import sys
import textwrap
import time
import urllib.parse
import uuid
from contextlib import contextmanager
from datetime import datetime, timedelta
from functools import partial
from typing import Any, Callable, Dict, Iterator, List, Optional, Union
from typing import Any, Callable, Dict, Iterator, List, Optional, Union, Tuple
import requests
import requests.utils
from Crypto import Random
from Crypto.Cipher import AES
from nacl.public import PublicKey, SealedBox # type:ignore
from .exceptions import *
@@ -24,44 +28,16 @@ def copy_session(session: requests.Session, request_timeout: Optional[float] = N
"""Duplicates a requests.Session."""
new = requests.Session()
new.cookies = requests.utils.cookiejar_from_dict(requests.utils.dict_from_cookiejar(session.cookies))
new.headers = session.headers.copy() # type: ignore
new.headers = session.headers.copy()
# Override default timeout behavior.
# Need to silence mypy bug for this. See: https://github.com/python/mypy/issues/2427
new.request = partial(new.request, timeout=request_timeout) # type: ignore
new.request = partial(new.request, timeout=request_timeout) # type: ignore
return new
def default_user_agent() -> str:
return ('Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 '
'(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36')
def default_iphone_headers() -> Dict[str, Any]:
return {'User-Agent': 'Instagram 273.0.0.16.70 (iPad13,8; iOS 16_3; en_US; en-US; ' \
'scale=2.00; 2048x2732; 452417278) AppleWebKit/420+',
'x-ads-opt-out': '1',
'x-bloks-is-panorama-enabled': 'true',
'x-bloks-version-id': '01507c21540f73e2216b6f62a11a5b5e51aa85491b72475c080da35b1228ddd6',
'x-fb-client-ip': 'True',
'x-fb-connection-type': 'wifi',
'x-fb-http-engine': 'Liger',
'x-fb-server-cluster': 'True',
'x-fb': '1',
'x-ig-abr-connection-speed-kbps': '2',
'x-ig-app-id': '124024574287414',
'x-ig-app-locale': 'en-US',
'x-ig-app-startup-country': 'US',
'x-ig-bandwidth-speed-kbps': '0.000',
'x-ig-capabilities': '36r/F/8=',
'x-ig-connection-speed': '{}kbps'.format(random.randint(1000, 20000)),
'x-ig-connection-type': 'WiFi',
'x-ig-device-locale': 'en-US',
'x-ig-mapped-locale': 'en-US',
'x-ig-timezone-offset': str((datetime.now().astimezone().utcoffset() or timedelta(seconds=0)).seconds),
'x-ig-www-claim': '0',
'x-pigeon-session-id': str(uuid.uuid4()),
'x-tigon-is-retry': 'False',
'x-whatsapp': '0'}
return 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 ' \
'(KHTML, like Gecko) Chrome/89.0.4389.90 Safari/537.36'
class InstaloaderContext:
@@ -89,7 +65,6 @@ class InstaloaderContext:
self.request_timeout = request_timeout
self._session = self.get_anonymous_session()
self.username = None
self.user_id = None
self.sleep = sleep
self.quiet = quiet
self.max_connection_attempts = max_connection_attempts
@@ -97,10 +72,9 @@ class InstaloaderContext:
self._root_rhx_gis = None
self.two_factor_auth_pending = None
self.iphone_support = iphone_support
self.iphone_headers = default_iphone_headers()
# error log, filled with error() and printed at the end of Instaloader.main()
self.error_log: List[str] = []
self.error_log = [] # type: List[str]
self._rate_controller = rate_controller(self) if rate_controller is not None else RateController(self)
@@ -111,26 +85,20 @@ class InstaloaderContext:
self.fatal_status_codes = fatal_status_codes or []
# Cache profile from id (mapping from id to Profile)
self.profile_id_cache: Dict[int, Any] = dict()
self.profile_id_cache = dict() # type: Dict[int, Any]
@contextmanager
def anonymous_copy(self):
session = self._session
username = self.username
user_id = self.user_id
iphone_headers = self.iphone_headers
self._session = self.get_anonymous_session()
self.username = None
self.user_id = None
self.iphone_headers = default_iphone_headers()
try:
yield self
finally:
self._session.close()
self.username = username
self._session = session
self.user_id = user_id
self.iphone_headers = iphone_headers
@property
def is_logged_in(self) -> bool:
@@ -206,34 +174,22 @@ class InstaloaderContext:
session.request = partial(session.request, timeout=self.request_timeout) # type: ignore
return session
def save_session(self):
"""Not meant to be used directly, use :meth:`Instaloader.save_session`."""
return requests.utils.dict_from_cookiejar(self._session.cookies)
def save_session_to_file(self, sessionfile):
"""Not meant to be used directly, use :meth:`Instaloader.save_session_to_file`."""
pickle.dump(requests.utils.dict_from_cookiejar(self._session.cookies), sessionfile)
def 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`."""
def load_session_from_file(self, username, sessionfile):
"""Not meant to be used directly, use :meth:`Instaloader.load_session_from_file`."""
session = requests.Session()
session.cookies = requests.utils.cookiejar_from_dict(sessiondata)
session.cookies = requests.utils.cookiejar_from_dict(pickle.load(sessionfile))
session.headers.update(self._default_http_header())
session.headers.update({'X-CSRFToken': session.cookies.get_dict()['csrftoken']})
# Override default timeout behavior.
# Need to silence mypy bug for this. See: https://github.com/python/mypy/issues/2427
session.request = partial(session.request, timeout=self.request_timeout) # type: ignore
session.request = partial(session.request, timeout=self.request_timeout) # type: ignore
self._session = session
self.username = username
def save_session_to_file(self, sessionfile):
"""Not meant to be used directly, use :meth:`Instaloader.save_session_to_file`."""
pickle.dump(self.save_session(), sessionfile)
def load_session_from_file(self, username, sessionfile):
"""Not meant to be used directly, use :meth:`Instaloader.load_session_from_file`."""
self.load_session(username, pickle.load(sessionfile))
def test_login(self) -> Optional[str]:
"""Not meant to be used directly, use :meth:`Instaloader.test_login`."""
data = self.graphql_query("d6f4427fbe92d846298cf93df0b937d3", {})
@@ -259,23 +215,16 @@ class InstaloaderContext:
# 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/',
enc_password = self._get_encrypted_password(passwd)
login = session.post('https://www.instagram.com/accounts/login/ajax/',
data={'enc_password': enc_password, 'username': user}, allow_redirects=True)
try:
resp_json = login.json()
except json.decoder.JSONDecodeError as err:
raise ConnectionException(
"Login error: JSON decode fail, {} - {}.".format(login.status_code, login.reason)
@@ -318,7 +267,6 @@ class InstaloaderContext:
session.headers.update({'X-CSRFToken': login.cookies['csrftoken']})
self._session = session
self.username = user
self.user_id = resp_json['userId']
def two_factor_login(self, two_factor_code):
"""Second step of login if 2FA is enabled.
@@ -352,8 +300,7 @@ class InstaloaderContext:
time.sleep(min(random.expovariate(0.6), 15.0))
def get_json(self, path: str, params: Dict[str, Any], host: str = 'www.instagram.com',
session: Optional[requests.Session] = None, _attempt=1,
response_headers: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
session: Optional[requests.Session] = None, _attempt=1) -> Dict[str, Any]:
"""JSON request to Instagram.
:param path: URL, relative to the given domain which defaults to www.instagram.com/
@@ -380,29 +327,22 @@ class InstaloaderContext:
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 = ""
if resp.headers['Content-Type'].startswith('application/json'):
body = ': ' + resp.text[:500] + ('…' if len(resp.text) > 501 else '')
raise AbortDownloadException("Query to https://{}/{} responded with \"{} {}\"{}{}".format(
host, path, resp.status_code, resp.reason, redirect, body
raise AbortDownloadException("Query to https://{}/{} responded with \"{} {}\"{}".format(
host, path, resp.status_code, resp.reason, redirect
))
while resp.is_redirect:
redirect_url = resp.headers['location']
self.log('\nHTTP redirect from https://{0}/{1} to {2}'.format(host, path, redirect_url))
if (redirect_url.startswith('https://www.instagram.com/accounts/login') or
redirect_url.startswith('https://i.instagram.com/accounts/login')):
if redirect_url.startswith('https://www.instagram.com/accounts/login'):
if not self.is_logged_in:
raise LoginRequiredException("Redirected to login page. Use --login.")
raise AbortDownloadException("Redirected to login page. You've been logged out, please wait " +
"some time, recreate the session and try again")
# alternate rate limit exceeded behavior
raise TooManyRequestsException("Redirected to login")
if redirect_url.startswith('https://{}/'.format(host)):
resp = sess.get(redirect_url if redirect_url.endswith('/') else redirect_url + '/',
params=params, allow_redirects=False)
else:
break
if response_headers is not None:
response_headers.clear()
response_headers.update(resp.headers)
if resp.status_code == 400:
raise QueryReturnedBadRequestException("400 Bad Request")
if resp.status_code == 404:
@@ -411,6 +351,23 @@ class InstaloaderContext:
raise TooManyRequestsException("429 Too Many Requests")
if resp.status_code != 200:
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:
match = re.search(r'window\._sharedData = (.*);</script>', resp.text)
if match is None:
raise QueryReturnedNotFoundException("Could not find \"window._sharedData\" in html response.")
resp_json = json.loads(match.group(1))
entry_data = resp_json.get('entry_data')
post_or_profile_page = list(entry_data.values())[0] if entry_data is not None else None
if post_or_profile_page is None:
raise ConnectionException("\"window._sharedData\" does not contain required keys.")
# If GraphQL data is missing in `window._sharedData`, search for it in `__additionalDataLoaded`.
if 'graphql' not in post_or_profile_page[0]:
match = re.search(r'window\.__additionalDataLoaded\(.*?({.*"graphql":.*})\);</script>',
resp.text)
if match is not None:
post_or_profile_page[0]['graphql'] = json.loads(match.group(1))['graphql']
return resp_json
else:
resp_json = resp.json()
if 'status' in resp_json and resp_json['status'] != "ok":
@@ -436,8 +393,7 @@ class InstaloaderContext:
self._rate_controller.handle_429('iphone')
if is_other_query:
self._rate_controller.handle_429('other')
return self.get_json(path=path, params=params, host=host, session=sess, _attempt=_attempt + 1,
response_headers=response_headers)
return self.get_json(path=path, params=params, host=host, session=sess, _attempt=_attempt + 1)
except KeyboardInterrupt:
self.error("[skipped by user]", repeat_at_end=False)
raise ConnectionException(error_string) from err
@@ -527,57 +483,11 @@ class InstaloaderContext:
.. versionadded:: 4.2.1"""
with copy_session(self._session, self.request_timeout) as tempsession:
# Set headers to simulate an API request from iPad
tempsession.headers['ig-intended-user-id'] = str(self.user_id)
tempsession.headers['x-pigeon-rawclienttime'] = '{:.6f}'.format(time.time())
# Add headers obtained from previous iPad request
tempsession.headers.update(self.iphone_headers)
# Extract key information from cookies if we haven't got it already from a previous request
header_cookies_mapping = {'x-mid': 'mid',
'ig-u-ds-user-id': 'ds_user_id',
'x-ig-device-id': 'ig_did',
'x-ig-family-device-id': 'ig_did',
'family_device_id': 'ig_did'}
# Map the cookie value to the matching HTTP request header
cookies = tempsession.cookies.get_dict().copy()
for key, value in header_cookies_mapping.items():
if value in cookies:
if key not in tempsession.headers:
tempsession.headers[key] = cookies[value]
else:
# Remove the cookie value if it's already specified as a header
tempsession.cookies.pop(value, None)
# Edge case for ig-u-rur header due to special string encoding in cookie
if 'rur' in cookies:
if 'ig-u-rur' not in tempsession.headers:
tempsession.headers['ig-u-rur'] = cookies['rur'].strip('\"').encode('utf-8') \
.decode('unicode_escape')
else:
tempsession.cookies.pop('rur', None)
# Remove headers specific to Desktop version
for header in ['Host', 'Origin', 'X-Instagram-AJAX', 'X-Requested-With', 'Referer']:
tempsession.headers['User-Agent'] = 'Instagram 146.0.0.27.125 (iPhone12,1; iOS 13_3; en_US; en-US; ' \
'scale=2.00; 1656x3584; 190542906)'
for header in ['Host', 'Origin', 'X-Instagram-AJAX', 'X-Requested-With']:
tempsession.headers.pop(header, None)
# No need for cookies if we have a bearer token
if 'authorization' in tempsession.headers:
tempsession.cookies.clear()
response_headers = dict() # type: Dict[str, Any]
response = self.get_json(path, params, 'i.instagram.com', tempsession, response_headers=response_headers)
# Extract the ig-set-* headers and use them in the next request
for key, value in response_headers.items():
if key.startswith('ig-set-'):
self.iphone_headers[key.replace('ig-set-', '')] = value
elif key.startswith('x-ig-set-'):
self.iphone_headers[key.replace('x-ig-set-', 'x-ig-')] = value
return response
return self.get_json(path, params, 'i.instagram.com', tempsession)
def write_raw(self, resp: Union[bytes, requests.Response], filename: str) -> None:
"""Write raw response data into a file.
@@ -654,6 +564,39 @@ class InstaloaderContext:
self._root_rhx_gis = self.get_json('', {}).get('rhx_gis', '')
return self._root_rhx_gis or None
def _get_encryption_data(self) -> Tuple[Optional[str], Optional[str], Optional[str]]:
resp = self._session.get("https://www.instagram.com/accounts/login/")
return (resp.headers.get("ig-set-password-encryption-web-key-id"),
resp.headers.get("ig-set-password-encryption-web-key-version"),
resp.headers.get("ig-set-password-encryption-web-pub-key"))
def _get_encrypted_password(self, password):
# contributed by @twitter-79 in
# https://github.com/instaloader/instaloader/issues/615#issuecomment-630631930
key_id, app_id, public_key = self._get_encryption_data()
timestamp = str(int(datetime.now().timestamp()))
# create a random key of length 32 bytes (for AES 256)
key = Random.get_random_bytes(32)
# create a buffer of length 12 bytes filled with 0
iv = bytearray(12)
aes = AES.new(key, AES.MODE_GCM, nonce=iv)
aes.update(bytearray(timestamp, 'utf-8'))
ciphertext, tag = aes.encrypt_and_digest(bytearray(password, 'utf-8'))
# get a byte array of the given public key
public_key_seal = PublicKey(bytes.fromhex(public_key))
sealed_box = SealedBox(public_key_seal)
sealed = sealed_box.encrypt(key)
enc_password = bytearray()
enc_password += bytearray(
[1, int(key_id), len(sealed) & 255, (len(sealed) >> 8) & 255]) + sealed + tag + ciphertext
return ('#PWD_INSTAGRAM_BROWSER' + ':' + app_id + ':' + timestamp + ':' +
str(base64.b64encode(enc_password), 'utf-8'))
class RateController:
"""
@@ -673,7 +616,7 @@ class RateController:
def __init__(self, context: InstaloaderContext):
self._context = context
self._query_timestamps: Dict[str, List[float]] = dict()
self._query_timestamps = dict() # type: Dict[str, List[float]]
self._earliest_next_request_time = 0.0
self._iphone_earliest_next_request_time = 0.0
@@ -681,6 +624,7 @@ class RateController:
"""Wait given number of seconds."""
# Not static, to allow for the behavior of this method to depend on context-inherent properties, such as
# whether we are logged in.
# pylint:disable=no-self-use
time.sleep(secs)
def _dump_query_timestamps(self, current_time: float, failed_query_type: str):
@@ -704,6 +648,7 @@ class RateController:
control."""
# Not static, to allow for the count_per_sliding_window to depend on context-inherent properties, such as
# whether we are logged in.
# pylint:disable=no-self-use
return 75 if query_type == 'other' else 200
def _reqs_in_sliding_window(self, query_type: Optional[str], current_time: float, window: float) -> List[float]:
-4
View File
@@ -1,8 +1,6 @@
import configparser
from datetime import datetime, timezone
from typing import Optional
from os.path import dirname
from os import makedirs
class LatestStamps:
@@ -27,8 +25,6 @@ class LatestStamps:
self.data.read(latest_stamps_file)
def _save(self):
if dn := dirname(self.file):
makedirs(dn, exist_ok=True)
with open(self.file, 'w') as f:
self.data.write(f)
+23 -40
View File
@@ -10,15 +10,15 @@ from typing import Any, Callable, Dict, Iterable, Iterator, NamedTuple, Optional
from .exceptions import AbortDownloadException, InvalidArgumentException, QueryReturnedBadRequestException
from .instaloadercontext import InstaloaderContext
class FrozenNodeIterator(NamedTuple):
query_hash: str
query_variables: Dict
query_referer: Optional[str]
context_username: Optional[str]
total_index: int
best_before: Optional[float]
remaining_data: Optional[Dict]
first_node: Optional[Dict]
FrozenNodeIterator = NamedTuple('FrozenNodeIterator',
[('query_hash', str),
('query_variables', Dict),
('query_referer', Optional[str]),
('context_username', Optional[str]),
('total_index', int),
('best_before', Optional[float]),
('remaining_data', Optional[Dict]),
('first_node', Optional[Dict])])
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."""
@@ -76,8 +76,7 @@ class NodeIterator(Iterator[T]):
node_wrapper: Callable[[Dict], T],
query_variables: Optional[Dict[str, Any]] = None,
query_referer: Optional[str] = None,
first_data: Optional[Dict[str, Any]] = None,
is_first: Optional[Callable[[T, Optional[T]], bool]] = None):
first_data: Optional[Dict[str, Any]] = None):
self._context = context
self._query_hash = query_hash
self._edge_extractor = edge_extractor
@@ -92,10 +91,9 @@ class NodeIterator(Iterator[T]):
else:
self._data = self._query()
self._first_node: Optional[Dict] = None
self._is_first = is_first
def _query(self, after: Optional[str] = None) -> Dict:
pagination_variables: Dict[str, Any] = {'first': NodeIterator._graphql_page_length}
pagination_variables = {'first': NodeIterator._graphql_page_length} # type: Dict[str, Any]
if after is not None:
pagination_variables['after'] = after
try:
@@ -130,24 +128,19 @@ class NodeIterator(Iterator[T]):
self._page_index, self._total_index = page_index, total_index
raise
item = self._node_wrapper(node)
if self._is_first is not None:
if self._is_first(item, self.first_item):
self._first_node = node
else:
if self._first_node is None:
self._first_node = node
if self._first_node is None:
self._first_node = node
return item
if self._data.get('page_info', {}).get('has_next_page'):
if self._data['page_info']['has_next_page']:
query_response = self._query(self._data['page_info']['end_cursor'])
if self._data['edges'] != query_response['edges'] and len(query_response['edges']) > 0:
page_index, data = self._page_index, self._data
try:
self._page_index = 0
self._data = query_response
except KeyboardInterrupt:
self._page_index, self._data = page_index, data
raise
return self.__next__()
page_index, data = self._page_index, self._data
try:
self._page_index = 0
self._data = query_response
except KeyboardInterrupt:
self._page_index, self._data = page_index, data
raise
return self.__next__()
raise StopIteration()
@property
@@ -174,20 +167,10 @@ class NodeIterator(Iterator[T]):
"""
If this iterator has produced any items, returns the first item produced.
It is possible to override what is considered the first item (for example, to consider the
newest item in case items are not in strict chronological order) by passing a callback
function as the `is_first` parameter when creating the class.
.. versionadded:: 4.8
.. versionchanged:: 4.9.2
What is considered the first item can be overridden.
"""
return self._node_wrapper(self._first_node) if self._first_node is not None else None
@staticmethod
def page_length() -> int:
return NodeIterator._graphql_page_length
def freeze(self) -> FrozenNodeIterator:
"""Freeze the iterator for later resuming."""
remaining_data = None
@@ -293,7 +276,7 @@ def resumable_iteration(context: InstaloaderContext,
is_resuming = True
start_index = iterator.total_index
context.log("Resuming from {}.".format(resume_file_path))
except (InvalidArgumentException, LZMAError, json.decoder.JSONDecodeError, EOFError) as exc:
except (InvalidArgumentException, LZMAError, json.decoder.JSONDecodeError) as exc:
context.error("Warning: Not resuming from {}: {}".format(resume_file_path, exc))
try:
yield is_resuming, start_index
-46
View File
@@ -1,46 +0,0 @@
from typing import Any, Callable, Dict, Iterator, Optional, TypeVar
from .instaloadercontext import InstaloaderContext
T = TypeVar('T')
class SectionIterator(Iterator[T]):
"""Iterator for the new 'sections'-style responses.
.. versionadded:: 4.9"""
def __init__(self,
context: InstaloaderContext,
sections_extractor: Callable[[Dict[str, Any]], Dict[str, Any]],
media_wrapper: Callable[[Dict], T],
query_path: str,
first_data: Optional[Dict[str, Any]] = None):
self._context = context
self._sections_extractor = sections_extractor
self._media_wrapper = media_wrapper
self._query_path = query_path
self._data = first_data or self._query()
self._page_index = 0
self._section_index = 0
def __iter__(self):
return self
def _query(self, max_id: Optional[str] = None) -> Dict[str, Any]:
pagination_variables = {"max_id": max_id} if max_id is not None else {}
return self._sections_extractor(
self._context.get_json(self._query_path, params={"__a": 1, "__d": "dis", **pagination_variables})
)
def __next__(self) -> T:
if self._page_index < len(self._data['sections']):
media = self._data['sections'][self._page_index]['layout_content']['medias'][self._section_index]['media']
self._section_index += 1
if self._section_index >= len(self._data['sections'][self._page_index]['layout_content']['medias']):
self._section_index = 0
self._page_index += 1
return self._media_wrapper(media)
if self._data['more_available']:
self._page_index, self._section_index, self._data = 0, 0, self._query(self._data["next_max_id"])
return self.__next__()
raise StopIteration()
+103 -565
View File
File diff suppressed because it is too large Load Diff
+6 -11
View File
@@ -18,13 +18,10 @@ def get_version():
raise SystemExit("Could not find version string.")
if sys.version_info < (3, 8):
sys.exit('Instaloader requires Python >= 3.8.')
if sys.version_info < (3, 6):
sys.exit('Instaloader requires Python >= 3.6.')
requirements = ['requests>=2.4']
optional_requirements = {
'browser_cookie3': ['browser_cookie3>=0.19.1'],
}
requirements = ['requests>=2.4', 'pycryptodome~=3.9', 'PyNaCl~=1.4']
keywords = (['instagram', 'instagram-scraper', 'instagram-client', 'instagram-feed', 'downloader', 'videos', 'photos',
'pictures', 'instagram-user-photos', 'instagram-photos', 'instagram-metadata', 'instagram-downloader',
@@ -46,8 +43,7 @@ setup(
'from Instagram.',
long_description=open(os.path.join(SRC, 'README.rst')).read(),
install_requires=requirements,
python_requires='>=3.8',
extras_require=optional_requirements,
python_requires='>=3.6',
entry_points={'console_scripts': ['instaloader=instaloader.__main__:main']},
zip_safe=False,
keywords=keywords,
@@ -58,11 +54,10 @@ setup(
'Intended Audience :: Developers',
'License :: OSI Approved :: MIT License',
'Operating System :: OS Independent',
'Programming Language :: Python :: 3.6',
'Programming Language :: Python :: 3.7',
'Programming Language :: Python :: 3.8',
'Programming Language :: Python :: 3.9',
'Programming Language :: Python :: 3.10',
'Programming Language :: Python :: 3.11',
'Programming Language :: Python :: 3.12',
'Programming Language :: Python :: 3 :: Only',
'Topic :: Internet',
'Topic :: Multimedia :: Graphics'
+1 -1
View File
@@ -23,7 +23,7 @@ PRIVATE_PROFILE_ID = 1706625676
EMPTY_PROFILE = "not_public"
EMPTY_PROFILE_ID = 1928659031
ratecontroller: Optional[instaloader.RateController] = None
ratecontroller = None # type: Optional[instaloader.RateController]
class TestInstaloaderAnonymously(unittest.TestCase):