feat: Discord on the native core ingester — client, downloader, ledgers, wiring (milestone 428)

Discord was the last focus platform still on gallery-dl. This adds the native path, mirrored from gallery-dl 1.32.13's discord extractor: - discord_client: API v10 with the user token and gallery-dl's request profile (dated Firefox UA, Referer). Walks a server, category, forum, channel or thread in gallery-dl's order and pages each channel newest-first. Files are attachments, then embeds, then forwards, numbered across the message. The resume cursor is <channel>:<before>. Text-only messages are not posts, since gallery-dl never made them. - discord_downloader: gallery-dl's on-disk layout, cleaned the way it cleans names on Linux (only `/` and control characters change), so existing files are skipped_disk rather than fetched again. Sidecars carry identity only. The message record keeps gallery-dl's keys, so parse_sidecar, derive_post_url and the drop grouping read it unchanged. - The ledger keys on the attachment id (or a hash of an embed's URL path), not the file's position, which an edit can renumber. Migration 0111. - DiscordIngester: token auth, body canary off (files-only drops are normal). Registered as native, verified by token, and serialised per-platform, since every source shares one user token. - ingest_core: optional `skip_feed` client seam (#4413). A tick's early-out on a multi-channel source now ends the quiet channel, not the whole walk. Clients without the seam behave as before. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01LVjrnpQjRgHdvq95rASoiR
2026-09-24 19:42:00 -04:00
parent 058fa85606
commit 84e5448941
15 changed files with 1579 additions and 15 deletions
@@ -0,0 +1,544 @@
+"""Native Discord read client — the Discord counterpart to subscribestar_client.
+
+Mirrors gallery-dl 1.32.13's `extractor/discord.py` (rule 130: gallery-dl is the
+known-working base), adapted to the native core's client contract
+(`ingest_core` module docstring): `iter_posts` / `extract_media`, plus the
+post-first `post_record_key` and the `post_meta` date the revisit window reads.
+
+What is mirrored exactly, because drift in any of it changes what we fetch or
+where it lands on disk:
+
+  - API v10, `Authorization: <user token>` (a USER token, not a bot token).
+  - gallery-dl's request profile: its date-derived Firefox User-Agent,
+    `Accept: */*`, `Accept-Language`, `Referer: https://discord.com/`.
+  - `GET /channels/{id}/messages?limit=100&before=<last id>`, newest first,
+    stopping on a short page. Message types {0, 19, 21} only.
+  - The walk: a text/news channel's own messages then its threads, a forum's
+    threads, a category's children, a server's text/news/forum channels.
+  - Files: attachments, then embeds of type image/gifv/video (FC configures
+    `embeds: all`, which for files is the same three plus rich/link embeds that
+    carry an image), then forwarded `message_snapshots`, numbered from 1 across
+    the lot — the `num` in `{date}_{message_id}_{num}_{filename}`.
+  - Text: `content`, rich-embed author/title/description/fields/footer, poll.
+
+Two deliberate departures, both about the walk order, neither about content:
+
+  - Threads are walked newest-CREATED first (by id), not by last-message time.
+    A backfill resumes from a checkpointed channel; last-message order shifts
+    between chunks whenever someone posts, which can move an unwalked thread
+    above the resume point and skip it. Creation order only ever grows at the
+    front, where the next tick finds it.
+  - A 403 on a thread or a nested channel skips that feed instead of failing
+    the walk. gallery-dl skips only nested channels; one private thread the
+    token cannot read would otherwise stop every channel after it.
+
+FC runs on a plain-HTTP homelab; nothing here uses a secure-context Web API.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import logging
+import re
+import time
+from collections.abc import Iterator
+from dataclasses import dataclass
+from datetime import date
+from urllib.parse import unquote
+
+import requests
+
+from .native_ingest_common import (
+    NativeAuthError,
+    NativeDriftError,
+    NativeIngestError,
+    retry_after_seconds,
+)
+
+log = logging.getLogger(__name__)
+
+API_ROOT = "https://discord.com/api/v10"
+_ROOT = "https://discord.com"
+
+_TIMEOUT_SECONDS = 60.0
+_MESSAGES_BATCH = 100
+_THREADS_BATCH = 25
+# gallery-dl retries a 429 up to its default 4 retries, waiting
+# `request_interval_429` (60s) between them. Discord's Retry-After is exact, so
+# it is honoured when present; 60s is the fallback and the cap.
+_MAX_429_RETRIES = 4
+_429_WAIT_SECONDS = 60.0
+
+# https://discord.com/developers/docs/resources/message#message-object-message-types
+# DEFAULT, REPLY, CHAT_INPUT_COMMAND — the ones that carry user content.
+MESSAGE_TYPES = frozenset({0, 19, 21})
+# https://discord.com/developers/docs/resources/channel#channel-object-channel-types
+_TEXT = frozenset({0, 5})               # text, announcement: messages + threads
+_DIRECT = frozenset({1, 3, 10, 11, 12})  # DMs and threads: messages only
+_FORUM = frozenset({15, 16})            # forum, media: threads only
+_CATEGORY = 4
+_SERVER_WALK = _TEXT | _FORUM
+_EMBED_TYPES = frozenset({"image", "gifv", "video"})
+
+_URL_RE = re.compile(
+    r"^(?:https?://)?(?:www\.|ptb\.|canary\.)?discord(?:app)?\.com/channels/"
+    r"(?P<server>@me|\d+)(?:/(?:\d+/threads/)?(?P<channel>\d+))?(?P<rest>/.*)?/?$"
+)
+
+
+class DiscordAPIError(NativeIngestError):
+    """Base for native Discord client failures."""
+
+
+class DiscordAuthError(DiscordAPIError, NativeAuthError):
+    """401 (the token is invalid or expired) or a 403 on the channel the
+    source names. The fix is a new token, not a new client."""
+
+
+class DiscordDriftError(DiscordAPIError, NativeDriftError):
+    """A response did not have the shape the walk depends on."""
+
+
+def firefox_user_agent(today: date | None = None) -> str:
+    """gallery-dl's default User-Agent: a Firefox whose version advances every
+    four weeks (`util._ff_ver`, "147 on 2026-01-13"). Computed the same way so
+    the profile keeps matching the gallery-dl this replaced."""
+    ver = ((today or date.today()).toordinal() - 735_513) // 28
+    return (
+        f"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:{ver}.0) "
+        f"Gecko/20100101 Firefox/{ver}.0"
+    )
+
+
+def nameext_from_url(url: str) -> tuple[str, str]:
+    """gallery-dl's `text.nameext_from_url`: the URL's last path segment,
+    unquoted, split at the last dot when the extension is at most 16 chars
+    (lowercased); otherwise the whole name and no extension."""
+    filename = unquote(url.partition("?")[0].rpartition("/")[2])
+    name, _, ext = filename.rpartition(".")
+    if name and len(ext) <= 16:
+        return name, ext.lower()
+    return filename, ""
+
+
+def parse_source_url(url: str) -> tuple[str | None, str | None]:
+    """`(server_id, channel_id)` from a Discord channel/server URL. `server_id`
+    is None for a DM (`@me`); `channel_id` is None for a whole server. Raises
+    DiscordAPIError for anything else, including a link to a single message —
+    a message is not something a source can subscribe to."""
+    m = _URL_RE.match((url or "").strip())
+    if not m or (m.group("rest") or "").strip("/"):
+        raise DiscordAPIError(
+            f"Not a Discord channel or server link: {url!r} "
+            "(expected https://discord.com/channels/<server>[/<channel>])"
+        )
+    server = m.group("server")
+    channel = m.group("channel")
+    if server == "@me":
+        if not channel:
+            raise DiscordAPIError(f"A DM link needs a channel id: {url!r}")
+        return None, channel
+    return server, channel
+
+
+def message_text(message: dict) -> str:
+    """gallery-dl's `extract_message_text`: the body plus the text of rich
+    embeds and polls, newline-joined, empties dropped."""
+    parts = [message.get("content") or ""]
+    for embed in message.get("embeds") or []:
+        if embed.get("type") != "rich":
+            continue
+        parts.append((embed.get("author") or {}).get("name") or "")
+        parts.append(embed.get("title") or "")
+        parts.append(embed.get("description") or "")
+        for fld in embed.get("fields") or []:
+            parts.append(fld.get("name") or "")
+            parts.append(fld.get("value") or "")
+        parts.append((embed.get("footer") or {}).get("text") or "")
+    poll = message.get("poll")
+    if poll:
+        parts.append(((poll.get("question") or {}).get("text")) or "")
+        for answer in poll.get("answers") or []:
+            parts.append(((answer.get("poll_media") or {}).get("text")) or "")
+    return "\n".join(p for p in parts if p)
+
+
+@dataclass
+class MediaItem:
+    """One file of a Discord message. `filename`/`extension` are gallery-dl's
+    split of the URL; `num` is its 1-based position across the message's files,
+    which is what names it on disk.
+
+    `media_id` is what the seen-ledger keys on, and it is deliberately NOT
+    `num`: an edit that removes a file renumbers the ones after it, and a
+    positional key would then call a different file seen. It is the
+    attachment's id, or for an embed (which has none) a hash of its URL path —
+    the query string is a signature that changes on every fetch. `filehash` is
+    always None; nothing in a signed CDN URL is a content hash."""
+
+    url: str
+    filename: str
+    extension: str
+    kind: str
+    post_id: str
+    num: int
+    media_id: str
+    filehash: str | None = None
+
+
+class DiscordClient:
+    """Synchronous Discord API v10 read client for one user token."""
+
+    def __init__(
+        self,
+        token: str | None,
+        *,
+        request_sleep: float = 0.0,
+        max_retries: int = _MAX_429_RETRIES,
+        session: requests.Session | None = None,
+    ):
+        self._session = session or requests.Session()
+        self._session.headers.update({
+            "User-Agent": firefox_user_agent(),
+            "Accept": "*/*",
+            "Accept-Language": "en-US,en;q=0.5",
+            "Referer": _ROOT + "/",
+        })
+        if token:
+            self._session.headers["Authorization"] = token
+        self._token = token
+        self._request_sleep = request_sleep or 0.0
+        self._max_retries = max_retries
+        self._server: dict = {}
+        self._channels: dict[str, dict] = {}
+        self._skip_feed = False
+
+    # -- request -----------------------------------------------------------
+
+    def _get(self, endpoint: str, params: dict | None = None):
+        if not self._token:
+            raise DiscordAuthError("No Discord token is configured for this source")
+        if self._request_sleep > 0:
+            time.sleep(self._request_sleep)
+        url = API_ROOT + endpoint
+        attempt = 0
+        while True:
+            try:
+                resp = self._session.get(url, params=params, timeout=_TIMEOUT_SECONDS)
+            except requests.RequestException as exc:
+                raise DiscordAPIError(f"Discord request failed ({endpoint}): {exc}") from exc
+            if resp.status_code == 429 and attempt < self._max_retries:
+                attempt += 1
+                delay = retry_after_seconds(
+                    resp, attempt, base=_429_WAIT_SECONDS, cap=_429_WAIT_SECONDS,
+                )
+                log.warning(
+                    "Discord 429 (%s) — waiting %.1fs (retry %d/%d)",
+                    endpoint, delay, attempt, self._max_retries,
+                )
+                time.sleep(delay)
+                continue
+            break
+        if resp.status_code == 401:
+            raise DiscordAuthError(
+                "Discord rejected the token (HTTP 401) — it is invalid or has "
+                "expired; copy a fresh one from the browser",
+                status_code=401,
+            )
+        if resp.status_code != 200:
+            raise DiscordAPIError(
+                f"Discord returned HTTP {resp.status_code} ({endpoint})",
+                status_code=resp.status_code,
+                retry_after=_retry_after(resp),
+            )
+        try:
+            return resp.json()
+        except ValueError as exc:
+            raise DiscordDriftError(
+                f"Discord returned non-JSON for {endpoint} ({len(resp.content)} bytes)"
+            ) from exc
+
+    # -- metadata (gallery-dl parse_server / parse_channel) -----------------
+
+    def _load_server(self, server_id: str) -> None:
+        server = self._get(f"/guilds/{server_id}")
+        if not isinstance(server, dict) or "id" not in server:
+            raise DiscordDriftError(f"Discord server {server_id} came back without an id")
+        self._server = {
+            "server": server.get("name") or "",
+            "server_id": str(server["id"]),
+            "owner_id": server.get("owner_id"),
+        }
+        channels = self._get(f"/guilds/{server_id}/channels")
+        if not isinstance(channels, list):
+            raise DiscordDriftError(f"Discord server {server_id} channel list is not a list")
+        # Categories first, so every child can name its parent.
+        for channel in sorted(channels, key=lambda ch: ch.get("type") != _CATEGORY):
+            self._parse_channel(channel)
+
+    def _parse_channel(self, channel: dict) -> dict:
+        parent_id = channel.get("parent_id")
+        meta = {
+            "channel": channel.get("name") or "",
+            "channel_id": str(channel.get("id")),
+            "channel_type": channel.get("type"),
+            "channel_topic": channel.get("topic") or "",
+            "parent_id": parent_id,
+            "is_thread": "thread_metadata" in channel,
+        }
+        parent = self._channels.get(parent_id) if parent_id else None
+        if parent:
+            meta["parent"] = parent["channel"]
+            meta["parent_type"] = parent["channel_type"]
+        if meta["channel_type"] in {1, 3}:
+            recipients = channel.get("recipients") or []
+            meta["channel"] = "DMs"
+            meta["recipients"] = [u.get("username") for u in recipients]
+            meta["recipients_id"] = [u.get("id") for u in recipients]
+        self._channels[meta["channel_id"]] = meta
+        return meta
+
+    def _channel_meta(self, channel_id: str) -> dict:
+        if channel_id not in self._channels:
+            self._parse_channel(self._get(f"/channels/{channel_id}"))
+        return self._channels[channel_id]
+
+    def _threads(self, channel_id: str) -> list[dict]:
+        """Every thread of a channel or forum, newest-created first (see the
+        module docstring for why not last-message order)."""
+        threads: list[dict] = []
+        offset = 0
+        while True:
+            data = self._get(f"/channels/{channel_id}/threads/search", {
+                "sort_by": "last_message_time",
+                "sort_order": "desc",
+                "limit": _THREADS_BATCH,
+                "offset": offset,
+            })
+            batch = (data.get("threads") or []) if isinstance(data, dict) else []
+            threads.extend(batch)
+            if len(batch) < _THREADS_BATCH:
+                break
+            offset += len(batch)
+        threads.sort(key=lambda t: int(t.get("id") or 0), reverse=True)
+        return threads
+
+    # -- the walk ------------------------------------------------------------
+
+    def _feeds(self, channel_id: str, *, safe: bool) -> Iterator[tuple[str, bool]]:
+        """`(channel_id, safe)` for every message feed under `channel_id`, in
+        gallery-dl's order. `safe` feeds are skipped on a 403."""
+        try:
+            ctype = self._channel_meta(channel_id)["channel_type"]
+        except DiscordAPIError as exc:
+            if exc.status_code != 403:
+                raise
+            if not safe:
+                raise DiscordAuthError(
+                    f"The Discord token cannot see channel {channel_id} (HTTP 403)",
+                    status_code=403,
+                ) from exc
+            log.info("Discord: no access to channel %s — skipped", channel_id)
+            return
+        if ctype in _TEXT or ctype in _DIRECT:
+            yield channel_id, safe
+        if ctype in _TEXT or ctype in _FORUM:
+            try:
+                threads = self._threads(channel_id)
+            except DiscordAPIError as exc:
+                if exc.status_code != 403:
+                    raise
+                log.info("Discord: cannot list threads of %s — skipped", channel_id)
+                threads = []
+            for thread in threads:
+                yield self._parse_channel(thread)["channel_id"], True
+        elif ctype == _CATEGORY:
+            for child in list(self._channels.values()):
+                if child.get("parent_id") == channel_id:
+                    yield from self._feeds(child["channel_id"], safe=True)
+        elif ctype not in _DIRECT and not safe:
+            raise DiscordAPIError(
+                f"Discord channel {channel_id} is of type {ctype}, which has no messages"
+            )
+
+    def _source_feeds(self, url: str) -> Iterator[tuple[str, bool]]:
+        server_id, channel_id = parse_source_url(url)
+        self._server, self._channels = {}, {}
+        if server_id is not None:
+            self._load_server(server_id)
+        if channel_id is not None:
+            yield from self._feeds(channel_id, safe=False)
+            return
+        for meta in list(self._channels.values()):
+            if meta["channel_type"] in _SERVER_WALK:
+                yield from self._feeds(meta["channel_id"], safe=True)
+
+    def skip_feed(self) -> None:
+        """Optional core seam (#4413): end the current channel and go on to the
+        next one. A tick's early-out means THIS channel has nothing new, not
+        that the server has nothing new."""
+        self._skip_feed = True
+
+    def iter_posts(
+        self, campaign_id: str, cursor: str | None = None
+    ) -> Iterator[tuple[dict, dict, str | None]]:
+        """Yield `(message, channel_meta, page_cursor)` for every content
+        message the source reaches, channel by channel, each newest first.
+
+        `campaign_id` is the source URL. The cursor is `<channel_id>:<before>`
+        — the channel and the `before` id that fetched the page (empty for a
+        channel's first page) — so a backfill resumes inside the right channel
+        and re-fetches the page it was cut in. A cursor naming a channel the
+        walk no longer reaches (a deleted thread) restarts from the top rather
+        than walking nothing.
+        """
+        resume_channel, _, resume_before = (cursor or "").partition(":")
+        resuming = bool(resume_channel)
+        feeds = list(self._source_feeds(campaign_id)) if resuming else None
+        if feeds is not None and resume_channel not in {cid for cid, _ in feeds}:
+            log.warning(
+                "Discord: resume channel %s is no longer in %s — restarting",
+                resume_channel, campaign_id,
+            )
+            resuming = False
+        for channel_id, safe in feeds if feeds is not None else self._source_feeds(campaign_id):
+            before = None
+            if resuming:
+                if channel_id != resume_channel:
+                    continue
+                resuming = False
+                before = resume_before or None
+            yield from self._iter_channel(channel_id, before, safe=safe)
+
+    def _iter_channel(
+        self, channel_id: str, before: str | None, *, safe: bool
+    ) -> Iterator[tuple[dict, dict, str | None]]:
+        self._skip_feed = False
+        meta = {**self._server, **self._channels.get(channel_id, {})}
+        while True:
+            page_cursor = f"{channel_id}:{before or ''}"
+            try:
+                messages = self._get(
+                    f"/channels/{channel_id}/messages",
+                    {"limit": _MESSAGES_BATCH, "before": before},
+                )
+            except DiscordAPIError as exc:
+                if exc.status_code != 403:
+                    raise
+                if not safe:
+                    raise DiscordAuthError(
+                        f"The Discord token cannot read channel {channel_id} (HTTP 403)",
+                        status_code=403,
+                    ) from exc
+                log.info("Discord: no access to messages of %s — skipped", channel_id)
+                return
+            if not isinstance(messages, list):
+                raise DiscordDriftError(
+                    f"Discord messages of {channel_id} came back as "
+                    f"{type(messages).__name__}, not a list"
+                )
+            for message in messages:
+                if message.get("type") not in MESSAGE_TYPES:
+                    continue
+                message["_meta"] = meta
+                yield message, meta, page_cursor
+                if self._skip_feed:
+                    return
+            if len(messages) < _MESSAGES_BATCH:
+                return
+            before = str(messages[-1]["id"])
+
+    # -- per-message -------------------------------------------------------
+
+    @staticmethod
+    def extract_media(post: dict, included: dict | None = None) -> list[MediaItem]:
+        """gallery-dl's file list for one message: attachments, then the first
+        of video/image/thumbnail `proxy_url` of each file-bearing embed, then
+        the same for every forwarded snapshot; numbered from 1 across them."""
+        mid = str(post.get("id") or "")
+        snapshots = [post] + [
+            (s or {}).get("message") or {}
+            for s in post.get("message_snapshots") or []
+            if ((s or {}).get("message") or {}).get("type", 0) in MESSAGE_TYPES
+        ]
+        found: list[tuple[str, str, str | None]] = []
+        for snap in snapshots:
+            for att in snap.get("attachments") or []:
+                if att.get("url"):
+                    aid = att.get("id")
+                    found.append((att["url"], "attachment", str(aid) if aid else None))
+            for embed in snap.get("embeds") or []:
+                if embed.get("type") not in _EMBED_TYPES:
+                    continue
+                for fld in ("video", "image", "thumbnail"):
+                    url = (embed.get(fld) or {}).get("proxy_url")
+                    if url:
+                        found.append((url, "embed", None))
+                        break
+        items = []
+        for num, (url, kind, fid) in enumerate(found, start=1):
+            name, ext = nameext_from_url(url)
+            if fid is None:
+                path = url.partition("?")[0].encode()
+                fid = "u" + hashlib.sha1(path, usedforsecurity=False).hexdigest()[:32]
+            items.append(MediaItem(
+                url=url, filename=name, extension=ext, kind=kind, post_id=mid,
+                num=num, media_id=fid,
+            ))
+        return items
+
+    @staticmethod
+    def post_meta(post: dict) -> dict:
+        """No title (Discord has none); `date` is the message timestamp, ISO
+        with an offset — what the core's revisit window reads."""
+        return {"title": None, "date": post.get("timestamp")}
+
+    @classmethod
+    def post_record_key(cls, post: dict) -> tuple[str, str] | None:
+        """`(message:<id>, <id>)` — gates the message record through the seen
+        ledger, like `post:<id>` on the other platforms.
+
+        None for a message with no files. gallery-dl wrote a sidecar only
+        beside a file, so a text-only chat line never became a post, and the
+        drop grouping (discord_grouping) is built on that: a channel's chatter
+        recorded as posts would bury the drops it exists to surface."""
+        mid = post.get("id")
+        mid = str(mid) if mid is not None else ""
+        if not mid or not cls.extract_media(post):
+            return None
+        return (f"message:{mid}", mid)
+
+    # -- verify ------------------------------------------------------------
+
+    def verify_auth(self, url: str) -> tuple[bool | None, str]:
+        """Is the token valid, and can it see what the source names?"""
+        try:
+            server_id, channel_id = parse_source_url(url)
+        except DiscordAPIError as exc:
+            return None, str(exc)
+        try:
+            me = self._get("/users/@me")
+            if channel_id is not None:
+                self._get(f"/channels/{channel_id}")
+            elif server_id is not None:
+                self._get(f"/guilds/{server_id}")
+        except DiscordAuthError as exc:
+            return False, f"Discord rejected the token — {exc}"
+        except DiscordAPIError as exc:
+            if exc.status_code in (403, 404):
+                return False, (
+                    "The token is valid, but its account cannot see "
+                    f"{'this channel' if channel_id else 'this server'} "
+                    f"(HTTP {exc.status_code})"
+                )
+            return None, f"Couldn't verify (network/HTTP issue): {exc}"
+        who = (me or {}).get("username") if isinstance(me, dict) else None
+        return True, f"Token valid{f' ({who})' if who else ''} — the source is readable."
+
+
+def _retry_after(resp: requests.Response) -> float | None:
+    hdr = resp.headers.get("Retry-After")
+    try:
+        return float(hdr) if hdr else None
+    except (TypeError, ValueError):
+        return None