diff --git a/surfsense_backend/app/scrapers/__init__.py b/surfsense_backend/app/proprietary/scrapers/__init__.py similarity index 100% rename from surfsense_backend/app/scrapers/__init__.py rename to surfsense_backend/app/proprietary/scrapers/__init__.py diff --git a/surfsense_backend/app/proprietary/scrapers/youtube/README.md b/surfsense_backend/app/proprietary/scrapers/youtube/README.md new file mode 100644 index 000000000..ca1105cbb --- /dev/null +++ b/surfsense_backend/app/proprietary/scrapers/youtube/README.md @@ -0,0 +1,303 @@ +# YouTube Scraper + +A platform-native YouTube scraper that is a **drop-in clone of the Apify +"YouTube Scraper" and "YouTube Comments Scraper" actors** — same input surface, +same output item shape. It talks to YouTube's internal **InnerTube** API plus +the public watch/channel HTML, egresses through a residential proxy, and streams +Apify-shaped dicts. + +No API keys, no Apify account, no headless browser on the happy path. + +--- + +## Quick start + +```python +from app.proprietary.scrapers.youtube import ( + YouTubeScrapeInput, scrape_youtube, + YouTubeCommentsInput, scrape_comments, +) + +# Videos — by search query and/or direct URLs (video/channel/playlist/hashtag/search) +videos = await scrape_youtube( + YouTubeScrapeInput(searchQueries=["surfsense"], maxResults=50) +) +videos = await scrape_youtube( + YouTubeScrapeInput(startUrls=[{"url": "https://www.youtube.com/@SomeChannel"}], + maxResults=20, downloadSubtitles=True) +) + +# Comments — one output item per top-level comment AND per reply +comments = await scrape_comments( + YouTubeCommentsInput( + startUrls=[{"url": "https://www.youtube.com/watch?v=VIDEO_ID"}], + maxComments=200, sortCommentsBy="TOP_COMMENTS", + ) +) +``` + +Both have a streaming twin — `iter_youtube()` / `iter_comments()` — that yields +items as they arrive (unbounded, continuation-paged). `scrape_*` is just a +collector with an optional `limit` guard. + +The HTTP surface lives in `app/routes/youtube_routes.py`. + +--- + +## Module map + + +| File | Responsibility | +| ------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | +| `__init__.py` | Public exports (entry points + schemas). | +| `schemas.py` | Pydantic input/output models mirroring the Apify camelCase spec. `extra="allow"` on outputs keeps the contract open. | +| `scraper.py` | Video orchestrator. Resolves URLs → per-flow async generators (`_video_flow`, `_search_flow`, `_channel_flow`, `_playlist_flow`), runs them through the `fan_out` worker pool. | +| `comments.py` | Comments orchestrator. Watch page → comments-section continuation → `/next` paging, with concurrent per-thread reply fetching. | +| `innertube.py` | **The network seam.** Proxy-only fetch (`fetch_html`, `post_innertube`), reusable sticky-IP sessions, reactive IP rotation, `StealthyFetcher` fallback, and the InnerTube payload builder. | +| `parsers.py` | Pure, I/O-free JSON/HTML traversal + normalization (`find_all`/`find_first`/`dig`, `parse_video_page`, `parse_search_response`, comment/continuation token extractors, `parse_count`, …). | +| `url_resolver.py` | Classify a URL into `video` / `channel` / `playlist` / `hashtag` / `search` and extract its id. | +| `search_filters.py` | Encode Apify search filters into YouTube's `sp=` base64 protobuf (sort/date/type/length/feature flags), composable. | +| `subtitles.py` | Subtitle download via `youtube-transcript-api`, shaped to Apify `subtitles[]`. | + + +Everything in `parsers.py` is deterministic and unit-tested offline; everything +that touches the network is funneled through `innertube.py`. + +--- + +## How it fetches (the important part) + +All network I/O goes through `**fetch_html**` (GET watch/channel pages) and +`**post_innertube**` (POST InnerTube `browse`/`search`/`next`). Design rules: + +1. **Proxy-only egress.** Every request goes through the residential proxy + (`app/utils/proxy.get_proxy_url`). We never connect directly — a direct hit + would expose and risk-block the server IP. +2. **Session reuse = sticky IP.** Within one flow (a continuation chain, or the + jobs a worker pulls), a single keep-alive `FetcherSession` is reused. This + roughly **halves warm latency** (~2.1s → ~1.0s) because only the first + request pays the TCP+TLS handshake, and it pins one sticky exit IP instead of + drawing a new (often slow) residential node per request. +3. **Reactive IP rotation.** A sticky IP is kept until it's actually blocked. On + `403`/`429` or a connection error, the session rotates to a fresh IP and + retries, up to `_MAX_ROTATIONS` (3). A probe of 120 sequential requests on + one IP saw zero blocks, so rotation is reactive, not proactive. +4. **Browser fallback.** If all proxy attempts fail on an HTML page, `fetch_html` + falls back to `StealthyFetcher` (headless, `solve_cloudflare=True`) in a + worker thread. Optional — needs patchright browsers installed. Age-gated + content requires login and is **not** bypassable. + +The active session is bound to the current async task via a `ContextVar` +(`_current_session`), so parsers and orchestrators never thread a session +argument through every call — each concurrent flow transparently uses its own +session/IP. + +### InnerTube payloads + +`build_innertube_payload(...)` builds the `WEB` client `context` payload +(I/O-free, unit-testable). Some endpoints reject a keyless POST; `scraper._post` +retries once with the public web key (`INNERTUBE_PUBLIC_API_KEY`) when the +keyless call returns nothing. `hl=` on a `/next` call returns the +creator-localized title/description (the translation flow). + +--- + +## Concurrency model + +Independent jobs — each `startUrl`, each `searchQuery`, each comment video — run +concurrently through `**fan_out**`, a warm **worker pool** (`_FANOUT_CONCURRENCY = 16`): + +- Each worker opens **one** proxy session and reuses it across the sequential +jobs it pulls, so only the first job per worker pays the handshake. +- **A bad job yields nothing rather than aborting the batch** (per-job +try/except). One dead URL / comments-disabled video never kills the run. +- Results stream out as each job finishes; **within** a flow, continuation +paging stays sequential. +- If the consumer stops early (collector hits its `limit`), workers are +cancelled and **awaited** so every session's `finally` closes — no leaked +keep-alive connections. + +Comment reply threads for a page are fetched **concurrently** on the same +multiplexed session (`asyncio.gather`), capped at the remaining budget. + +--- + +## Data flow + +- **Video by URL** → fetch watch HTML → `parse_video_page` (reads +`ytInitialData` + `ytInitialPlayerResponse`) → optional subtitles + translation. +- **Search** → InnerTube `/search` (+ `sp=` filter protobuf) → paginate via +continuation tokens up to `maxResults`. +- **Channel** → fetch the videos-tab seed once (reused for channel-wide metadata + - the About panel via `/browse`), then page `videos` / `shorts` / `streams` + tabs, each capped independently (`maxResults` / `maxResultsShorts` / + `maxResultStreams`). `sortVideosBy` uses the sort chips; `oldestPostDate` cuts + off newest-first. +- **Playlist** → `/browse` `VL`, paged via the continuation token → resolve + each video via the video flow. +- **Hashtag** → the dedicated hashtag page (`/hashtag/`), whose feed is + `videoRenderer` lockups (parsed like search) — not a `#tag` search. +- **Comments** → watch HTML seeds the comments-section token → `/next` returns +comment entities + per-thread reply tokens + the page token. `maxComments` +counts **every** emitted item (comments + replies). + +### `commentsCount` + +For the **comments** scraper, the authoritative total is read from the +comments-section header (`commentsHeaderRenderer.countText`), not the watch-page +HTML where it's lazy-loaded/absent. **Known gap:** the **video** scraper's +`VideoItem.commentsCount` still comes from search/watch HTML and is often `null` +— it would need an extra `/next` call to backfill (intentionally not done to +keep the video path cheap). + +--- + +## API spec + +Mirrors the Apify "YouTube Scraper" and "YouTube Comments Scraper" actors +(camelCase, `extra="allow"`). Inputs use Pydantic defaults; **every field is +additive** — unknown inputs are accepted, unsourced outputs come back as +`None`/`[]` — so parity grows without breaking consumers. `schemas.py` is the +source of truth. + +### Video scraper — input (`YouTubeScrapeInput`) + + +| Field | Type / values | Default | Notes | +| ------------------------------------------------------------------------------------------------------- | -------------------------------------------- | ------- | ---------------------------------------------------------------------------------- | +| `searchQueries` | `string[]` | `[]` | Discovery by query. Ignored when `startUrls` is set. | +| `startUrls` | `[{ "url": string }]` | `[]` | Direct URLs: video, channel, playlist, hashtag, search. Overrides `searchQueries`. | +| `maxResults` | `int ≥ 0` | `0` | Cap of regular videos **per query and per channel**. `0` = fetch none. | +| `maxResultsShorts` | `int ≥ 0` | `0` | Cap of Shorts per channel. | +| `maxResultStreams` | `int ≥ 0` | `0` | Cap of live/streams per channel. | +| `downloadSubtitles` | `bool` | `false` | Populate `subtitles[]`. | +| `subtitlesLanguage` | `string` | `"en"` | Also drives the translation flow when non-`en` (see `translatedTitle`). | +| `subtitlesFormat` | `srt` | `vtt` | `xml` | `plaintext` | `"srt"` | | +| `preferAutoGeneratedSubtitles` | `bool` | `false` | | +| `saveSubsToKVS` | `bool` | `false` | Accepted for parity; no-op (Apify key-value-store concept). | +| `sortingOrder` | `relevance` | `rating` | `date` | `views` | `null` | Search only. | +| `dateFilter` | `hour` | `today` | `week` | `month` | `year` | `null` | Search only. | +| `videoType` | `video` | `movie` | `null` | Search only. | +| `lengthFilter` | `under4` | `between420` | `plus20` | `null` | Search only (<4min / 4–20min / >20min). | +| `isHD` `hasSubtitles` `hasCC` `is3D` `isLive` `isBought` `is4K` `is360` `hasLocation` `isHDR` `isVR180` | `bool` | `null` | Search feature filters (encoded into `sp=`). | +| `oldestPostDate` | `string` (date) | `null` | Channel cutoff; day-accurate (relative times). | +| `sortVideosBy` | `NEWEST` | `POPULAR` | `OLDEST` | `null` | Channel videos tab sort chip. | + + +### Video scraper — output (`VideoItem`) + + +| Field | Type | Populated? | +| -------------------------------------------------------------- | ----------------------------------- | ------------------------------------------------------------------- | +| `title` `id` `url` `viewCount` `date` `duration` | str/int | yes | +| `type` | `video` | `shorts` | `stream` | yes | +| `thumbnailUrl` | str | yes | +| `input` `fromYTUrl` `order` | str/int | yes (provenance: source query/URL, origin URL, index) | +| `text` | str | yes (description) | +| `descriptionLinks` | `[{ url, text }]` | yes | +| `hashtags` | `string[]` | yes | +| `likes` `commentsCount` `commentsTurnedOff` | int/bool | partial (often `null` on the video path — see `commentsCount` note) | +| `location` | str | when present | +| `collaborators` | `[{ name, username, url }]` | when present | +| `translatedTitle` `translatedText` | str | when `subtitlesLanguage != "en"` | +| `subtitles` | `[{ srtUrl, type, language, srt }]` | when `downloadSubtitles` | +| `isMembersOnly` `isPaidContent` | bool | yes (default `false`) | +| `isMonetized` `isAgeRestricted` | bool | best-effort (`null` when unknown) | +| `channelName` `channelUrl` `channelUsername` `channelId` | str | yes | +| `numberOfSubscribers` `channelTotalVideos` `channelTotalViews` | int | channel/deep fields | +| `channelDescription` `channelLocation` `channelJoinedDate` | str | channel About panel | +| `isChannelVerified` `channelBannerUrl` `channelAvatarUrl` | bool/str | channel fields | + + +### Comments scraper — input (`YouTubeCommentsInput`) + + +| Field | Type / values | Default | Notes | +| ------------------- | ------------------------------- | ---------------- | ------------------------------------------------------------------- | +| `startUrls` | `[{ "url": string }]` | `[]` | Video URLs only (non-video URLs skipped). | +| `maxComments` | `int ≥ 1` | `1` | Counts **every** emitted item (top-level comments **and** replies). | +| `sortCommentsBy` | `TOP_COMMENTS` | `NEWEST_FIRST` | `"NEWEST_FIRST"` | | +| `oldestCommentDate` | `string` (date) | `null` | Forces newest-first and stops at the cutoff. | + + +### Comments scraper — output (`CommentItem`) + + +| Field | Type | Notes | +| ---------------------------------------- | ------------------- | --------------------------------------------- | +| `cid` | str | Comment id. | +| `comment` | str | Text. | +| `author` | str | | +| `type` | `comment` | `reply` | | +| `replyToCid` | str | Parent `cid` (replies only). | +| `replyCount` | int | Replies under a top-level comment. | +| `voteCount` | int | Likes. | +| `authorIsChannelOwner` `hasCreatorHeart` | bool | | +| `publishedTimeText` | str | Relative time ("2 days ago"). | +| `videoId` `pageUrl` `title` | str | Source video. | +| `commentsCount` | int | Authoritative total from the comments header. | + + +--- + +## Configuration + +- **Proxy** — required for real runs; configured via `app/utils/proxy.py` +(residential rotating gateway env vars). With no proxy configured the fetchers +fall back to one-shot direct `AsyncFetcher` calls (fine for local tests, not +for production). +- **Concurrency** — `scraper._FANOUT_CONCURRENCY` (16). The gateway handled 64 +parallel flows with zero failures in a ramp probe, so this leaves headroom. +- **Rotation** — `innertube._BLOCK_STATUSES` (`403`, `429`) and +`_MAX_ROTATIONS` (3). + +--- + +## Testing + +- **Offline unit tests** (no network) — run these on every change: + ```bash + cd surfsense_backend + .venv/Scripts/python.exe -m pytest tests/unit/scrapers/youtube/ + ``` + - `test_parsers.py` — parser/normalization + filter-protobuf + URL-resolver + cases against hand-built and (if present) captured real fixtures. + - `test_fetch_resilience.py` — deterministic rotate-on-block (`429`/error → + rotate → `200`, exhaustion, no-rotate on `404`, stealthy fallback) and the + `fan_out` no-session-leak-on-early-stop guarantee, all with stubbed sessions. +- **Live functional harness** — `scripts/e2e_youtube_scraper.py` (needs live +network + optional proxy creds). Exercises video/search/channel/comments/ +location/collaborators/translation end to end, and **regenerates the offline +fixtures** into `tests/unit/scrapers/youtube/fixtures/`: + ```bash + .venv/Scripts/python.exe scripts/e2e_youtube_scraper.py + ``` + +--- + +## Extending it + +- **Add an output field** → populate it in the relevant `parsers.py` function +and add it to `schemas.py`. Because outputs are `extra="allow"`, forgetting the +schema line won't drop the value, but declaring it documents the contract. +- **Add a URL kind** → extend `url_resolver.resolve_url` + add a `_*_flow` in +`scraper.py` and a branch in `_dispatch`. +- **Add a search filter** → add the field to `YouTubeScrapeInput` and encode it +in `search_filters.build_search_params` (verify byte-for-byte against a real +YouTube `sp=` token in the unit test). + +### Known ceilings (grep `ponytail:` in the source for the live list) + +- Hashtag scraping returns a single feed page (~20-35 videos); YouTube exposes +no continuation for the hashtag feed through this path. Upgrade path for more +depth: fall back to the `#tag` search route. +- Playlist video ids are paged sequentially (each continuation depends on the +last), then the per-video watch-page fetches run concurrently via `fan_out` +(~150 videos ≈ 70s). Because resolution is fanned out, items stream back in +completion order, not playlist order — sort by the `order` field to restore it. +- `oldestPostDate` / `oldestCommentDate` cutoffs are day-accurate at best +(channel/list pages only expose coarse relative times like "2 years ago"). +- Keyless-vs-keyed InnerTube retry does one extra request on the keyed path +instead of remembering which worked. +- Video-path `commentsCount` (see above). + diff --git a/surfsense_backend/app/scrapers/youtube/__init__.py b/surfsense_backend/app/proprietary/scrapers/youtube/__init__.py similarity index 100% rename from surfsense_backend/app/scrapers/youtube/__init__.py rename to surfsense_backend/app/proprietary/scrapers/youtube/__init__.py diff --git a/surfsense_backend/app/scrapers/youtube/comments.py b/surfsense_backend/app/proprietary/scrapers/youtube/comments.py similarity index 100% rename from surfsense_backend/app/scrapers/youtube/comments.py rename to surfsense_backend/app/proprietary/scrapers/youtube/comments.py diff --git a/surfsense_backend/app/scrapers/youtube/innertube.py b/surfsense_backend/app/proprietary/scrapers/youtube/innertube.py similarity index 100% rename from surfsense_backend/app/scrapers/youtube/innertube.py rename to surfsense_backend/app/proprietary/scrapers/youtube/innertube.py diff --git a/surfsense_backend/app/scrapers/youtube/parsers.py b/surfsense_backend/app/proprietary/scrapers/youtube/parsers.py similarity index 98% rename from surfsense_backend/app/scrapers/youtube/parsers.py rename to surfsense_backend/app/proprietary/scrapers/youtube/parsers.py index 3e6701de3..e07a2b869 100644 --- a/surfsense_backend/app/scrapers/youtube/parsers.py +++ b/surfsense_backend/app/proprietary/scrapers/youtube/parsers.py @@ -590,12 +590,14 @@ def _continuation_token(data: dict) -> str | None: return tokens[-1] if tokens else None -def parse_playlist_video_ids(data: dict) -> list[str]: - """Ordered, de-duped video ids from a playlist ``/browse`` response. +def parse_playlist_video_ids(data: dict) -> tuple[list[str], str | None]: + """Ordered, de-duped video ids + paging token from a playlist ``/browse``. Playlist entries are ``lockupViewModel`` (the old ``playlistVideoRenderer`` is retired); ``contentId`` holds the id. The 11-char guard keeps only videos - and drops any playlist/channel lockup (e.g. the sidebar self-lockup). + and drops any playlist/channel lockup (e.g. the sidebar self-lockup). The + token pages long playlists; callers must still guard against an empty page, + since a short playlist can emit a spurious (non-paging) continuation. """ seen: set[str] = set() ids: list[str] = [] @@ -604,7 +606,7 @@ def parse_playlist_video_ids(data: dict) -> list[str]: if vid and len(vid) == 11 and vid not in seen: seen.add(vid) ids.append(vid) - return ids + return ids, _continuation_token(data) def parse_channel_videos(data: dict) -> tuple[list[dict[str, Any]], str | None]: diff --git a/surfsense_backend/app/scrapers/youtube/schemas.py b/surfsense_backend/app/proprietary/scrapers/youtube/schemas.py similarity index 100% rename from surfsense_backend/app/scrapers/youtube/schemas.py rename to surfsense_backend/app/proprietary/scrapers/youtube/schemas.py diff --git a/surfsense_backend/app/scrapers/youtube/scraper.py b/surfsense_backend/app/proprietary/scrapers/youtube/scraper.py similarity index 83% rename from surfsense_backend/app/scrapers/youtube/scraper.py rename to surfsense_backend/app/proprietary/scrapers/youtube/scraper.py index d5eb2aa3c..c72343c20 100644 --- a/surfsense_backend/app/scrapers/youtube/scraper.py +++ b/surfsense_backend/app/proprietary/scrapers/youtube/scraper.py @@ -395,24 +395,91 @@ async def _playlist_flow( data = await _post( INNERTUBE_BROWSE_URL, build_innertube_payload(browse_id=f"VL{playlist_id}") ) - if not data: - return - # ponytail: reads the first browse page only (~100 items). Upgrade path: - # follow the continuation token for playlists longer than one page. - video_ids = parse_playlist_video_ids(data) + # Phase 1: page the playlist for video ids (cheap browse calls, sequential + # because each continuation depends on the last). + seen: set[str] = set() + ordered_ids: list[str] = [] + while data and len(ordered_ids) < limit: + ids, token = parse_playlist_video_ids(data) + # A short playlist emits a spurious continuation whose page is empty; + # stopping on "no new ids" ends both real exhaustion and that loop. + new_ids = [v for v in ids if v not in seen] + if not new_ids: + break + for vid in new_ids: + seen.add(vid) + ordered_ids.append(vid) + if len(ordered_ids) >= limit: + break + if not token: + break + data = await _post( + INNERTUBE_BROWSE_URL, build_innertube_payload(continuation_token=token) + ) - for order, vid in enumerate(video_ids): - if order >= limit: - return - async for item in _video_flow( + # Phase 2: resolve the videos concurrently — the per-video watch-page fetch + # is the bottleneck, so fan them out (each carries its playlist position in + # ``order``; fan_out emits as they finish, not in playlist order). + # ponytail: nested fan_out — when many playlist URLs run at once this can + # stack pools (outer × inner) of proxy sessions. Fine for the common + # single/few-playlist case; cap inner concurrency if bulk-playlist runs trip it. + jobs = [ + _video_flow( vid, input_model=input_model, source_input=source_input, from_url=source_input, - order=order, + order=i, content_type="video", - ): - yield item + ) + for i, vid in enumerate(ordered_ids) + ] + async for item in fan_out(jobs): + yield item + + +async def _hashtag_flow( + tag: str, + *, + input_model: YouTubeScrapeInput, + source_input: str, +) -> AsyncIterator[dict[str, Any]]: + """Scrape the dedicated hashtag feed (not a #tag search). + + The hashtag page embeds its feed as ``videoRenderer`` lockups (reused via + ``parse_search_response``). ponytail: YouTube exposes no continuation for the + hashtag feed through this path, so it is a single page (~20-35 videos); the + paging loop is kept for the day a token appears. Upgrade path for more depth: + fall back to the ``#tag`` search route. + """ + limit = input_model.maxResults + if limit <= 0: + return + url = f"https://www.youtube.com/hashtag/{quote(tag)}" + html = await fetch_html(url) + if not html: + return + data = extract_yt_initial_data(html) + order = 0 + while data: + items, token = parse_search_response(data) + for it in items: + if order >= limit: + return + yield await _finalize( + it, + input_model=input_model, + source_input=source_input, + from_url=url, + order=order, + content_type="video", + ) + order += 1 + if not token or order >= limit: + return + data = await _post( + INNERTUBE_BROWSE_URL, build_innertube_payload(continuation_token=token) + ) async def _dispatch( @@ -439,12 +506,14 @@ async def _dispatch( resolved.value, input_model=input_model, source_input=resolved.url ): yield item - elif resolved.kind in ("search", "hashtag"): - # ponytail: hashtag pages have their own layout; MVP routes the tag - # through search (query "#tag"). Upgrade path: dedicated hashtag browse. - query = resolved.value if resolved.kind == "search" else f"#{resolved.value}" + elif resolved.kind == "hashtag": + async for item in _hashtag_flow( + resolved.value, input_model=input_model, source_input=resolved.url + ): + yield item + elif resolved.kind == "search": async for item in _search_flow( - query, input_model=input_model, source_input=resolved.url + resolved.value, input_model=input_model, source_input=resolved.url ): yield item diff --git a/surfsense_backend/app/scrapers/youtube/search_filters.py b/surfsense_backend/app/proprietary/scrapers/youtube/search_filters.py similarity index 100% rename from surfsense_backend/app/scrapers/youtube/search_filters.py rename to surfsense_backend/app/proprietary/scrapers/youtube/search_filters.py diff --git a/surfsense_backend/app/scrapers/youtube/subtitles.py b/surfsense_backend/app/proprietary/scrapers/youtube/subtitles.py similarity index 100% rename from surfsense_backend/app/scrapers/youtube/subtitles.py rename to surfsense_backend/app/proprietary/scrapers/youtube/subtitles.py diff --git a/surfsense_backend/app/scrapers/youtube/url_resolver.py b/surfsense_backend/app/proprietary/scrapers/youtube/url_resolver.py similarity index 100% rename from surfsense_backend/app/scrapers/youtube/url_resolver.py rename to surfsense_backend/app/proprietary/scrapers/youtube/url_resolver.py diff --git a/surfsense_backend/app/routes/youtube_routes.py b/surfsense_backend/app/routes/youtube_routes.py index 55190792a..22d8c6ace 100644 --- a/surfsense_backend/app/routes/youtube_routes.py +++ b/surfsense_backend/app/routes/youtube_routes.py @@ -9,7 +9,7 @@ from fastapi import APIRouter, Depends, HTTPException, Query from scrapling.fetchers import AsyncFetcher from app.auth.context import AuthContext -from app.scrapers.youtube import ( +from app.proprietary.scrapers.youtube import ( YouTubeCommentsInput, YouTubeScrapeInput, scrape_comments, diff --git a/surfsense_backend/app/scrapers/youtube/README.md b/surfsense_backend/app/scrapers/youtube/README.md deleted file mode 100644 index 0b4bef63e..000000000 --- a/surfsense_backend/app/scrapers/youtube/README.md +++ /dev/null @@ -1,287 +0,0 @@ -# YouTube Scraper - -A platform-native YouTube scraper that is a **drop-in clone of the Apify -"YouTube Scraper" and "YouTube Comments Scraper" actors** — same input surface, -same output item shape. It talks to YouTube's internal **InnerTube** API plus -the public watch/channel HTML, egresses through a residential proxy, and streams -Apify-shaped dicts. - -No API keys, no Apify account, no headless browser on the happy path. - ---- - -## Quick start - -```python -from app.scrapers.youtube import ( - YouTubeScrapeInput, scrape_youtube, - YouTubeCommentsInput, scrape_comments, -) - -# Videos — by search query and/or direct URLs (video/channel/playlist/hashtag/search) -videos = await scrape_youtube( - YouTubeScrapeInput(searchQueries=["surfsense"], maxResults=50) -) -videos = await scrape_youtube( - YouTubeScrapeInput(startUrls=[{"url": "https://www.youtube.com/@SomeChannel"}], - maxResults=20, downloadSubtitles=True) -) - -# Comments — one output item per top-level comment AND per reply -comments = await scrape_comments( - YouTubeCommentsInput( - startUrls=[{"url": "https://www.youtube.com/watch?v=VIDEO_ID"}], - maxComments=200, sortCommentsBy="TOP_COMMENTS", - ) -) -``` - -Both have a streaming twin — `iter_youtube()` / `iter_comments()` — that yields -items as they arrive (unbounded, continuation-paged). `scrape_*` is just a -collector with an optional `limit` guard. - -The HTTP surface lives in `app/routes/youtube_routes.py`. - ---- - -## Module map - -| File | Responsibility | -|------|----------------| -| `__init__.py` | Public exports (entry points + schemas). | -| `schemas.py` | Pydantic input/output models mirroring the Apify camelCase spec. `extra="allow"` on outputs keeps the contract open. | -| `scraper.py` | Video orchestrator. Resolves URLs → per-flow async generators (`_video_flow`, `_search_flow`, `_channel_flow`, `_playlist_flow`), runs them through the `fan_out` worker pool. | -| `comments.py` | Comments orchestrator. Watch page → comments-section continuation → `/next` paging, with concurrent per-thread reply fetching. | -| `innertube.py` | **The network seam.** Proxy-only fetch (`fetch_html`, `post_innertube`), reusable sticky-IP sessions, reactive IP rotation, `StealthyFetcher` fallback, and the InnerTube payload builder. | -| `parsers.py` | Pure, I/O-free JSON/HTML traversal + normalization (`find_all`/`find_first`/`dig`, `parse_video_page`, `parse_search_response`, comment/continuation token extractors, `parse_count`, …). | -| `url_resolver.py` | Classify a URL into `video` / `channel` / `playlist` / `hashtag` / `search` and extract its id. | -| `search_filters.py` | Encode Apify search filters into YouTube's `sp=` base64 protobuf (sort/date/type/length/feature flags), composable. | -| `subtitles.py` | Subtitle download via `youtube-transcript-api`, shaped to Apify `subtitles[]`. | - -Everything in `parsers.py` is deterministic and unit-tested offline; everything -that touches the network is funneled through `innertube.py`. - ---- - -## How it fetches (the important part) - -All network I/O goes through **`fetch_html`** (GET watch/channel pages) and -**`post_innertube`** (POST InnerTube `browse`/`search`/`next`). Design rules: - -1. **Proxy-only egress.** Every request goes through the residential proxy - (`app/utils/proxy.get_proxy_url`). We never connect directly — a direct hit - would expose and risk-block the server IP. -2. **Session reuse = sticky IP.** Within one flow (a continuation chain, or the - jobs a worker pulls), a single keep-alive `FetcherSession` is reused. This - roughly **halves warm latency** (~2.1s → ~1.0s) because only the first - request pays the TCP+TLS handshake, and it pins one sticky exit IP instead of - drawing a new (often slow) residential node per request. -3. **Reactive IP rotation.** A sticky IP is kept until it's actually blocked. On - `403`/`429` or a connection error, the session rotates to a fresh IP and - retries, up to `_MAX_ROTATIONS` (3). A probe of 120 sequential requests on - one IP saw zero blocks, so rotation is reactive, not proactive. -4. **Browser fallback.** If all proxy attempts fail on an HTML page, `fetch_html` - falls back to `StealthyFetcher` (headless, `solve_cloudflare=True`) in a - worker thread. Optional — needs patchright browsers installed. Age-gated - content requires login and is **not** bypassable. - -The active session is bound to the current async task via a `ContextVar` -(`_current_session`), so parsers and orchestrators never thread a session -argument through every call — each concurrent flow transparently uses its own -session/IP. - -### InnerTube payloads - -`build_innertube_payload(...)` builds the `WEB` client `context` payload -(I/O-free, unit-testable). Some endpoints reject a keyless POST; `scraper._post` -retries once with the public web key (`INNERTUBE_PUBLIC_API_KEY`) when the -keyless call returns nothing. `hl=` on a `/next` call returns the -creator-localized title/description (the translation flow). - ---- - -## Concurrency model - -Independent jobs — each `startUrl`, each `searchQuery`, each comment video — run -concurrently through **`fan_out`**, a warm **worker pool** (`_FANOUT_CONCURRENCY -= 16`): - -- Each worker opens **one** proxy session and reuses it across the sequential - jobs it pulls, so only the first job per worker pays the handshake. -- **A bad job yields nothing rather than aborting the batch** (per-job - try/except). One dead URL / comments-disabled video never kills the run. -- Results stream out as each job finishes; **within** a flow, continuation - paging stays sequential. -- If the consumer stops early (collector hits its `limit`), workers are - cancelled and **awaited** so every session's `finally` closes — no leaked - keep-alive connections. - -Comment reply threads for a page are fetched **concurrently** on the same -multiplexed session (`asyncio.gather`), capped at the remaining budget. - ---- - -## Data flow - -- **Video by URL** → fetch watch HTML → `parse_video_page` (reads - `ytInitialData` + `ytInitialPlayerResponse`) → optional subtitles + translation. -- **Search** → InnerTube `/search` (+ `sp=` filter protobuf) → paginate via - continuation tokens up to `maxResults`. -- **Channel** → fetch the videos-tab seed once (reused for channel-wide metadata - + the About panel via `/browse`), then page `videos` / `shorts` / `streams` - tabs, each capped independently (`maxResults` / `maxResultsShorts` / - `maxResultStreams`). `sortVideosBy` uses the sort chips; `oldestPostDate` cuts - off newest-first. -- **Playlist** → `/browse` `VL` → resolve each video via the video flow. -- **Comments** → watch HTML seeds the comments-section token → `/next` returns - comment entities + per-thread reply tokens + the page token. `maxComments` - counts **every** emitted item (comments + replies). - -### `commentsCount` - -For the **comments** scraper, the authoritative total is read from the -comments-section header (`commentsHeaderRenderer.countText`), not the watch-page -HTML where it's lazy-loaded/absent. **Known gap:** the **video** scraper's -`VideoItem.commentsCount` still comes from search/watch HTML and is often `null` -— it would need an extra `/next` call to backfill (intentionally not done to -keep the video path cheap). - ---- - -## API spec - -Mirrors the Apify "YouTube Scraper" and "YouTube Comments Scraper" actors -(camelCase, `extra="allow"`). Inputs use Pydantic defaults; **every field is -additive** — unknown inputs are accepted, unsourced outputs come back as -`None`/`[]` — so parity grows without breaking consumers. `schemas.py` is the -source of truth. - -### Video scraper — input (`YouTubeScrapeInput`) - -| Field | Type / values | Default | Notes | -|-------|---------------|---------|-------| -| `searchQueries` | `string[]` | `[]` | Discovery by query. Ignored when `startUrls` is set. | -| `startUrls` | `[{ "url": string }]` | `[]` | Direct URLs: video, channel, playlist, hashtag, search. Overrides `searchQueries`. | -| `maxResults` | `int ≥ 0` | `0` | Cap of regular videos **per query and per channel**. `0` = fetch none. | -| `maxResultsShorts` | `int ≥ 0` | `0` | Cap of Shorts per channel. | -| `maxResultStreams` | `int ≥ 0` | `0` | Cap of live/streams per channel. | -| `downloadSubtitles` | `bool` | `false` | Populate `subtitles[]`. | -| `subtitlesLanguage` | `string` | `"en"` | Also drives the translation flow when non-`en` (see `translatedTitle`). | -| `subtitlesFormat` | `srt` \| `vtt` \| `xml` \| `plaintext` | `"srt"` | | -| `preferAutoGeneratedSubtitles` | `bool` | `false` | | -| `saveSubsToKVS` | `bool` | `false` | Accepted for parity; no-op (Apify key-value-store concept). | -| `sortingOrder` | `relevance` \| `rating` \| `date` \| `views` | `null` | Search only. | -| `dateFilter` | `hour` \| `today` \| `week` \| `month` \| `year` | `null` | Search only. | -| `videoType` | `video` \| `movie` | `null` | Search only. | -| `lengthFilter` | `under4` \| `between420` \| `plus20` | `null` | Search only (<4min / 4–20min / >20min). | -| `isHD` `hasSubtitles` `hasCC` `is3D` `isLive` `isBought` `is4K` `is360` `hasLocation` `isHDR` `isVR180` | `bool` | `null` | Search feature filters (encoded into `sp=`). | -| `oldestPostDate` | `string` (date) | `null` | Channel cutoff; day-accurate (relative times). | -| `sortVideosBy` | `NEWEST` \| `POPULAR` \| `OLDEST` | `null` | Channel videos tab sort chip. | - -### Video scraper — output (`VideoItem`) - -| Field | Type | Populated? | -|-------|------|-----------| -| `title` `id` `url` `viewCount` `date` `duration` | str/int | yes | -| `type` | `video` \| `shorts` \| `stream` | yes | -| `thumbnailUrl` | str | yes | -| `input` `fromYTUrl` `order` | str/int | yes (provenance: source query/URL, origin URL, index) | -| `text` | str | yes (description) | -| `descriptionLinks` | `[{ url, text }]` | yes | -| `hashtags` | `string[]` | yes | -| `likes` `commentsCount` `commentsTurnedOff` | int/bool | partial (often `null` on the video path — see `commentsCount` note) | -| `location` | str | when present | -| `collaborators` | `[{ name, username, url }]` | when present | -| `translatedTitle` `translatedText` | str | when `subtitlesLanguage != "en"` | -| `subtitles` | `[{ srtUrl, type, language, srt }]` | when `downloadSubtitles` | -| `isMembersOnly` `isPaidContent` | bool | yes (default `false`) | -| `isMonetized` `isAgeRestricted` | bool | best-effort (`null` when unknown) | -| `channelName` `channelUrl` `channelUsername` `channelId` | str | yes | -| `numberOfSubscribers` `channelTotalVideos` `channelTotalViews` | int | channel/deep fields | -| `channelDescription` `channelLocation` `channelJoinedDate` | str | channel About panel | -| `isChannelVerified` `channelBannerUrl` `channelAvatarUrl` | bool/str | channel fields | - -### Comments scraper — input (`YouTubeCommentsInput`) - -| Field | Type / values | Default | Notes | -|-------|---------------|---------|-------| -| `startUrls` | `[{ "url": string }]` | `[]` | Video URLs only (non-video URLs skipped). | -| `maxComments` | `int ≥ 1` | `1` | Counts **every** emitted item (top-level comments **and** replies). | -| `sortCommentsBy` | `TOP_COMMENTS` \| `NEWEST_FIRST` | `"NEWEST_FIRST"` | | -| `oldestCommentDate` | `string` (date) | `null` | Forces newest-first and stops at the cutoff. | - -### Comments scraper — output (`CommentItem`) - -| Field | Type | Notes | -|-------|------|-------| -| `cid` | str | Comment id. | -| `comment` | str | Text. | -| `author` | str | | -| `type` | `comment` \| `reply` | | -| `replyToCid` | str | Parent `cid` (replies only). | -| `replyCount` | int | Replies under a top-level comment. | -| `voteCount` | int | Likes. | -| `authorIsChannelOwner` `hasCreatorHeart` | bool | | -| `publishedTimeText` | str | Relative time ("2 days ago"). | -| `videoId` `pageUrl` `title` | str | Source video. | -| `commentsCount` | int | Authoritative total from the comments header. | - ---- - -## Configuration - -- **Proxy** — required for real runs; configured via `app/utils/proxy.py` - (residential rotating gateway env vars). With no proxy configured the fetchers - fall back to one-shot direct `AsyncFetcher` calls (fine for local tests, not - for production). -- **Concurrency** — `scraper._FANOUT_CONCURRENCY` (16). The gateway handled 64 - parallel flows with zero failures in a ramp probe, so this leaves headroom. -- **Rotation** — `innertube._BLOCK_STATUSES` (`403`, `429`) and - `_MAX_ROTATIONS` (3). - ---- - -## Testing - -- **Offline unit tests** (no network) — run these on every change: - ```bash - cd surfsense_backend - .venv/Scripts/python.exe -m pytest tests/unit/scrapers/youtube/ - ``` - - `test_parsers.py` — parser/normalization + filter-protobuf + URL-resolver - cases against hand-built and (if present) captured real fixtures. - - `test_fetch_resilience.py` — deterministic rotate-on-block (`429`/error → - rotate → `200`, exhaustion, no-rotate on `404`, stealthy fallback) and the - `fan_out` no-session-leak-on-early-stop guarantee, all with stubbed sessions. -- **Live functional harness** — `scripts/e2e_youtube_scraper.py` (needs live - network + optional proxy creds). Exercises video/search/channel/comments/ - location/collaborators/translation end to end, and **regenerates the offline - fixtures** into `tests/unit/scrapers/youtube/fixtures/`: - ```bash - .venv/Scripts/python.exe scripts/e2e_youtube_scraper.py - ``` - ---- - -## Extending it - -- **Add an output field** → populate it in the relevant `parsers.py` function - and add it to `schemas.py`. Because outputs are `extra="allow"`, forgetting the - schema line won't drop the value, but declaring it documents the contract. -- **Add a URL kind** → extend `url_resolver.resolve_url` + add a `_*_flow` in - `scraper.py` and a branch in `_dispatch`. -- **Add a search filter** → add the field to `YouTubeScrapeInput` and encode it - in `search_filters.build_search_params` (verify byte-for-byte against a real - YouTube `sp=` token in the unit test). - -### Known ceilings (grep `ponytail:` in the source for the live list) - -- Hashtag pages are routed through search (`#tag` query), not a dedicated - hashtag browse. -- `oldestPostDate` / `oldestCommentDate` cutoffs are day-accurate at best - (channel/list pages only expose coarse relative times like "2 years ago"). -- Keyless-vs-keyed InnerTube retry does one extra request on the keyed path - instead of remembering which worked. -- Video-path `commentsCount` (see above). -- Playlist scraping reads only the first `/browse` page (~100 videos); paging - the continuation token for longer playlists is not yet wired. diff --git a/surfsense_backend/scripts/e2e_youtube_scraper.py b/surfsense_backend/scripts/e2e_youtube_scraper.py index 78f2511da..92ca9635a 100644 --- a/surfsense_backend/scripts/e2e_youtube_scraper.py +++ b/surfsense_backend/scripts/e2e_youtube_scraper.py @@ -1,4 +1,4 @@ -"""Manual functional e2e for the YouTube scraper (app/scrapers/youtube). +"""Manual functional e2e for the YouTube scraper (app/proprietary/scrapers/youtube). Run from the backend directory: cd surfsense_backend @@ -31,20 +31,20 @@ for _candidate in (_BACKEND_ROOT / ".env", _BACKEND_ROOT.parent / ".env"): load_dotenv(_candidate) break -from app.scrapers.youtube import ( # noqa: E402 +from app.proprietary.scrapers.youtube import ( # noqa: E402 YouTubeCommentsInput, YouTubeScrapeInput, scrape_comments, scrape_youtube, ) -from app.scrapers.youtube.innertube import ( # noqa: E402 +from app.proprietary.scrapers.youtube.innertube import ( # noqa: E402 INNERTUBE_PUBLIC_API_KEY, INNERTUBE_SEARCH_URL, build_innertube_payload, fetch_html, post_innertube, ) -from app.scrapers.youtube.parsers import ( # noqa: E402 +from app.proprietary.scrapers.youtube.parsers import ( # noqa: E402 extract_yt_initial_data, extract_yt_initial_player_response, ) diff --git a/surfsense_backend/tests/unit/scrapers/youtube/test_fetch_resilience.py b/surfsense_backend/tests/unit/scrapers/youtube/test_fetch_resilience.py index 9676388ca..068c37b00 100644 --- a/surfsense_backend/tests/unit/scrapers/youtube/test_fetch_resilience.py +++ b/surfsense_backend/tests/unit/scrapers/youtube/test_fetch_resilience.py @@ -11,8 +11,8 @@ from collections.abc import AsyncIterator import pytest -from app.scrapers.youtube import innertube, scraper -from app.scrapers.youtube.innertube import ( +from app.proprietary.scrapers.youtube import innertube, scraper +from app.proprietary.scrapers.youtube.innertube import ( INNERTUBE_SEARCH_URL, _current_session, fetch_html, diff --git a/surfsense_backend/tests/unit/scrapers/youtube/test_parsers.py b/surfsense_backend/tests/unit/scrapers/youtube/test_parsers.py index 70ddf5823..28c624f62 100644 --- a/surfsense_backend/tests/unit/scrapers/youtube/test_parsers.py +++ b/surfsense_backend/tests/unit/scrapers/youtube/test_parsers.py @@ -12,7 +12,7 @@ from pathlib import Path import pytest -from app.scrapers.youtube.parsers import ( +from app.proprietary.scrapers.youtube.parsers import ( channel_about_tokens, comment_next_token, comment_reply_tokens, @@ -35,9 +35,9 @@ from app.scrapers.youtube.parsers import ( parse_video_page, seconds_to_duration, ) -from app.scrapers.youtube.schemas import YouTubeScrapeInput -from app.scrapers.youtube.search_filters import build_search_params -from app.scrapers.youtube.url_resolver import resolve_url +from app.proprietary.scrapers.youtube.schemas import YouTubeScrapeInput +from app.proprietary.scrapers.youtube.search_filters import build_search_params +from app.proprietary.scrapers.youtube.url_resolver import resolve_url pytestmark = pytest.mark.unit @@ -429,8 +429,13 @@ def test_parse_playlist_video_ids(): }, # a playlist self-lockup (non-video, longer id) that must be ignored "sidebar": {"lockupViewModel": {"contentId": "PLZHQObOWTQDPD3MizzM2xVFitgF8hE_ab"}}, + "continuationItemRenderer": { + "continuationEndpoint": {"continuationCommand": {"token": "PAGE2"}} + }, } - assert parse_playlist_video_ids(data) == ["fNk_zzaMoSs", "k7RM-ot2NWY"] + ids, token = parse_playlist_video_ids(data) + assert ids == ["fNk_zzaMoSs", "k7RM-ot2NWY"] + assert token == "PAGE2" # --- search filter (sp=) encoder --------------------------------------------