apunkt/SurfSense
1
0
Fork 0
mirror of https://github.com/MODSetter/SurfSense.git synced 2026-07-16 23:01:06 +02:00
Code Issues Projects Releases Packages Wiki Activity

refactor(instagram): improve anonymous scraping logic and error messaging

Browse source 
Enhanced the Instagram scraper to clarify the requirements for accessing user profiles and hashtags. Updated the error message for blocked access to provide detailed guidance on necessary credentials. Introduced a regex for validating Instagram usernames and refined the discovery function to handle profile queries directly, improving user experience and error handling in anonymous mode.
This commit is contained in:
Anish Sarkar 2026-07-09 20:23:09 +05:30
parent f66ab73044
commit 82f1d0b4e5
2 changed files with 27 additions and 42 deletions
Download patch file Download diff file Expand all files Collapse all files

5
surfsense_backend/app/capabilities/instagram/scrape/executor.py
View file

@ -43,7 +43,10 @@ def build_scrape_executor(scrape_fn: ScrapeFn | None = None) -> Executor:
except InstagramAccessBlockedError as exc: except InstagramAccessBlockedError as exc:
# Anonymous-only scraper; a hard block can't be retried with creds. # Anonymous-only scraper; a hard block can't be retried with creds.
raise ForbiddenError( raise ForbiddenError(
f"Instagram refused anonymous access: {exc}", "Instagram requires a login for this request and SurfSense scrapes "
"anonymously. Provide a profile URL or handle via directUrls; "
"keyword/hashtag search needs an account and is unavailable. "
f"Details: {exc}",
code="INSTAGRAM_ACCESS_BLOCKED", code="INSTAGRAM_ACCESS_BLOCKED",
) from exc ) from exc
emit_progress( emit_progress(

64
surfsense_backend/app/proprietary/platforms/instagram/scraper.py
View file

@ -26,6 +26,7 @@ from __future__ import annotations
import asyncio import asyncio
import logging import logging
import re
from collections.abc import AsyncIterator from collections.abc import AsyncIterator
from contextlib import aclosing from contextlib import aclosing
from datetime import UTC, datetime, timedelta from datetime import UTC, datetime, timedelta
@ -70,6 +71,10 @@ _HASHTAG_PATH = "api/v1/tags/web_info/"
_LOCATION_PATH = "api/v1/locations/web_info/" _LOCATION_PATH = "api/v1/locations/web_info/"
_SEARCH_PATH = "web/search/topsearch/" _SEARCH_PATH = "web/search/topsearch/"
# Instagram usernames: 1-30 chars of letters/digits/period/underscore. Used to
# treat a profile/user discovery query as a direct (anonymous) handle lookup.
_HANDLE_RE = re.compile(r"[A-Za-z0-9._]{1,30}\Z")
def _parse_newer_than(value: str | None) -> datetime | None: def _parse_newer_than(value: str | None) -> datetime | None:
"""Parse ``onlyPostsNewerThan`` (ISO, YYYY-MM-DD, or relative) to UTC. """Parse ``onlyPostsNewerThan`` (ISO, YYYY-MM-DD, or relative) to UTC.
@ -356,48 +361,25 @@ async def _details_flow(
async def _discover( async def _discover(
query: str, *, search_type: str, limit: int query: str, *, search_type: str, limit: int
) -> list[ResolvedUrl]: ) -> list[ResolvedUrl]:
"""Resolve a discovery query into target URLs via topsearch.""" """Resolve a discovery query into targets - anonymously.
data = await fetch_json(_SEARCH_PATH, {"query": query, "context": "blended"})
if not isinstance(data, dict): Instagram's keyword search (``topsearch``) is login-walled, so we never call
return [] it. A profile/user query is resolved as a direct handle lookup against the
out: list[ResolvedUrl] = [] anonymous profile endpoint ("messi" -> instagram.com/messi/). Hashtag/place
keyword discovery has NO anonymous endpoint (topsearch and the tag/location
feeds all require a session), so we surface a clear block instead of a
misleading empty success.
"""
if search_type in ("profile", "user"): if search_type in ("profile", "user"):
for entry in data.get("users", []): handle = query.strip().lstrip("@")
user = entry.get("user", {}) if isinstance(entry, dict) else {} if _HANDLE_RE.match(handle):
name = user.get("username") url = f"https://www.instagram.com/{handle}/"
if not name: return [ResolvedUrl("profile", handle, url)][:limit]
continue return [] # not a handle, and no anonymous fuzzy search -> nothing to do
out.append( raise InstagramAccessBlockedError(
ResolvedUrl("profile", name, f"https://www.instagram.com/{name}/") f"Instagram {search_type} search requires login and is unavailable in "
) "anonymous mode - pass a profile URL or handle via directUrls instead"
elif search_type == "hashtag": )
for entry in data.get("hashtags", []):
tag = entry.get("hashtag", {}) if isinstance(entry, dict) else {}
name = tag.get("name")
if not name:
continue
out.append(
ResolvedUrl(
"hashtag",
name,
f"https://www.instagram.com/explore/tags/{name}/",
)
)
elif search_type == "place":
for entry in data.get("places", []):
place = entry.get("place", {}) if isinstance(entry, dict) else {}
loc = place.get("location", {}) if isinstance(place, dict) else {}
pk = loc.get("pk") or loc.get("id")
if not pk:
continue
out.append(
ResolvedUrl(
"place",
str(pk),
f"https://www.instagram.com/explore/locations/{pk}/",
)
)
return out[:limit]
def _resolve_inputs(input_model: InstagramScrapeInput) -> list[ResolvedUrl]: def _resolve_inputs(input_model: InstagramScrapeInput) -> list[ResolvedUrl]: