[instagram] add 'user-cache' option (#8978 #9025)

This commit is contained in:
Mike Fährmann
2026-02-10 11:54:39 +01:00
parent ace8c50278
commit a8376f2804
3 changed files with 56 additions and 36 deletions

View File

@@ -3865,6 +3865,22 @@ Description
Download video previews. Download video previews.
extractor.instagram.user-cache
------------------------------
Type
* ``bool``
* ``string``
Default
``"disk"``
Description
Selects how to cache user profile data.
``true`` | ``"disk"``
Cache user data on disk
``false`` | ``"memory"``
Cache user data in memory
extractor.instagram.videos extractor.instagram.videos
-------------------------- --------------------------
Type Type

View File

@@ -474,6 +474,7 @@
"order-files": "asc", "order-files": "asc",
"order-posts": "asc", "order-posts": "asc",
"previews" : false, "previews" : false,
"user-cache" : "disk",
"videos" : true, "videos" : true,
"warn-images": true, "warn-images": true,
"warn-videos": true, "warn-videos": true,

View File

@@ -11,7 +11,7 @@
from .common import Extractor, Message, Dispatch from .common import Extractor, Message, Dispatch
from .. import text, util, exception from .. import text, util, exception
from ..cache import cache from ..cache import cache, memcache
import itertools import itertools
import binascii import binascii
@@ -805,6 +805,12 @@ class InstagramRestAPI():
def __init__(self, extractor): def __init__(self, extractor):
self.extractor = extractor self.extractor = extractor
_cache = self.extractor.config("user-cache", True)
_cache = memcache if not _cache or _cache == "memory" else cache
self.user_by_id = _cache(36500*86400, 0)(self.user_by_id)
self.user_by_name = _cache(36500*86400, 0)(self.user_by_name)
self.user_by_search = _cache(36500*86400, 0)(self.user_by_search)
def guide(self, guide_id): def guide(self, guide_id):
endpoint = "/v1/guides/web_info/" endpoint = "/v1/guides/web_info/"
params = {"guide_id": guide_id} params = {"guide_id": guide_id}
@@ -869,23 +875,43 @@ class InstagramRestAPI():
} }
return self._pagination_sections(endpoint, data) return self._pagination_sections(endpoint, data)
@cache(maxage=36500*86400, keyarg=1)
def user_by_id(self, user_id): def user_by_id(self, user_id):
endpoint = f"/v1/users/{user_id}/info/" endpoint = f"/v1/users/{user_id}/info/"
return self._call(endpoint)["user"] return self._call(endpoint)["user"]
def user_by_name(self, username):
endpoint = "/v1/users/web_profile_info/"
params = {"username": username}
try:
return self._call(
endpoint, params=params, notfound="user")["data"]["user"]
except KeyError:
raise exception.NotFoundError("user")
def user_by_search(self, username):
url = "https://www.instagram.com/web/search/topsearch/"
params = {"query": username}
name = username.lower()
for result in self._call(url, params=params)["users"]:
user = result["user"]
if user["username"].lower() == name:
return user
def user_by_screen_name(self, screen_name): def user_by_screen_name(self, screen_name):
user = user_by_search(self, screen_name) if user := self.user_by_search(screen_name):
if user is None: return user
user_by_search.invalidate(screen_name)
self.extractor.log.warning( self.user_by_search.invalidate(screen_name)
"Failed to find profile '%s' via search. " self.extractor.log.warning(
"Trying 'web_profile_info' fallback", screen_name) "Failed to find profile '%s' via search. "
user = user_by_name(self, screen_name) "Trying 'web_profile_info' fallback", screen_name)
if user is None:
user_by_name.invalidate(screen_name) if user := self.user_by_name(screen_name):
raise exception.NotFoundError("user") return user
return user
self.user_by_name.invalidate(screen_name)
raise exception.NotFoundError("user")
def user_id(self, screen_name, check_private=True): def user_id(self, screen_name, check_private=True):
if screen_name.startswith("id:"): if screen_name.startswith("id:"):
@@ -1159,29 +1185,6 @@ def _login_impl(extr, username, password):
return {} return {}
@cache(maxage=36500*86400, keyarg=1)
def user_by_name(self, screen_name):
endpoint = "/v1/users/web_profile_info/"
params = {"username": screen_name}
try:
return self._call(
endpoint, params=params, notfound="user")["data"]["user"]
except KeyError:
raise exception.NotFoundError("user")
@cache(maxage=36500*86400, keyarg=1)
def user_by_search(self, screen_name):
url = "https://www.instagram.com/web/search/topsearch/"
params = {"query": screen_name}
name = screen_name.lower()
for result in self._call(url, params=params)["users"]:
user = result["user"]
if user["username"].lower() == name:
return user
def id_from_shortcode(shortcode): def id_from_shortcode(shortcode):
return util.bdecode(shortcode, _ALPHABET) return util.bdecode(shortcode, _ALPHABET)