Merge pull request 'Merge dev → main: #830 rich post capture + external-host downloads (+ #768/#789/#739)' (#102) from dev into main

Dockerfile

@@ -18,13 +18,16 @@ ENV PYTHONUNBUFFERED=1 \
 
 # System deps: ffmpeg (transcode + thumbnails, FC-2), unar (archives, FC-2),
 # libpq for psycopg, postgresql-client + zstd for FC-5 backup/restore
-# (pg_dump + tar --zstd), image libs.
+# (pg_dump + tar --zstd), image libs, megatools (mega.nz public-link downloads
+# for off-platform file-host links, #830 — `megatools dl`; Debian-native, no
+# external MEGA apt repo needed).
 RUN apt-get update && apt-get install -y --no-install-recommends \
         ffmpeg \
         unar \
         libpq5 \
         postgresql-client \
         zstd \
+        megatools \
         libjpeg62-turbo \
         libwebp7 \
         libpng16-16 \

alembic/versions/0049_external_link_table.py

@@ -0,0 +1,90 @@
+"""external_link table — off-platform file-host links found in post bodies
+
+Creators host the real files on mega.nz / Google Drive / MediaFire / Dropbox /
+Pixeldrain and link them in the post text. This table records each such link
+(so nothing is silently dropped), and doubles as the dedup + dead-letter ledger
+the download worker (a later slice) walks. `url` keeps the FULL link including
+the `#fragment` — mega.nz's decryption key lives there; truncating it makes the
+file undownloadable.
+
+CHECK whitelists for host + status include the full enum up front (incl. the
+download-worker statuses) so the worker slice needs no constraint migration.
+
+Revision ID: 0049
+Revises: 0048
+Create Date: 2026-06-14
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0049"
+down_revision: Union[str, None] = "0048"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "external_link",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column(
+            "post_id", sa.Integer(),
+            sa.ForeignKey("post.id", ondelete="CASCADE"), nullable=False,
+        ),
+        sa.Column(
+            "artist_id", sa.Integer(),
+            sa.ForeignKey("artist.id", ondelete="SET NULL"), nullable=True,
+        ),
+        sa.Column("host", sa.String(length=16), nullable=False),
+        sa.Column("url", sa.Text(), nullable=False),
+        sa.Column("label", sa.Text(), nullable=True),
+        sa.Column(
+            "status", sa.String(length=16), nullable=False,
+            server_default="pending",
+        ),
+        sa.Column("attempts", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("last_error", sa.Text(), nullable=True),
+        sa.Column(
+            "attachment_id", sa.Integer(),
+            sa.ForeignKey("post_attachment.id", ondelete="SET NULL"),
+            nullable=True,
+        ),
+        sa.Column(
+            "created_at", sa.DateTime(timezone=True), nullable=False,
+            server_default=sa.func.now(),
+        ),
+        sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("duration_seconds", sa.Float(), nullable=True),
+        sa.CheckConstraint(
+            "host IN ('mega','gdrive','mediafire','dropbox','pixeldrain')",
+            name="ck_external_link_host",
+        ),
+        sa.CheckConstraint(
+            "status IN ('pending','downloading','downloaded','failed',"
+            "'skipped','dead')",
+            name="ck_external_link_status",
+        ),
+    )
+    op.create_index(
+        "ix_external_link_post_id", "external_link", ["post_id"],
+    )
+    op.create_index(
+        "ix_external_link_artist_id", "external_link", ["artist_id"],
+    )
+    op.create_index(
+        "ix_external_link_status", "external_link", ["status"],
+    )
+    op.create_index(
+        "uq_external_link_post_url", "external_link", ["post_id", "url"],
+        unique=True,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("uq_external_link_post_url", table_name="external_link")
+    op.drop_index("ix_external_link_status", table_name="external_link")
+    op.drop_index("ix_external_link_artist_id", table_name="external_link")
+    op.drop_index("ix_external_link_post_id", table_name="external_link")
+    op.drop_table("external_link")

alembic/versions/0050_external_link_host_toggles.py

@@ -0,0 +1,38 @@
+"""import_settings: per-host enable toggles for external file-host downloads
+
+Operator levers (#830): disable a single host (e.g. mega.nz when it's
+rate-limiting/banning) without touching the others. The worker reads these via
+getattr and defaults to enabled, so the toggles default TRUE (works out of the
+box, rule #26).
+
+Revision ID: 0050
+Revises: 0049
+Create Date: 2026-06-14
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0050"
+down_revision: Union[str, None] = "0049"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+_HOSTS = ("mega", "gdrive", "mediafire", "dropbox", "pixeldrain")
+
+
+def upgrade() -> None:
+    for host in _HOSTS:
+        op.add_column(
+            "import_settings",
+            sa.Column(
+                f"extdl_{host}_enabled", sa.Boolean(), nullable=False,
+                server_default=sa.true(),
+            ),
+        )
+
+
+def downgrade() -> None:
+    for host in _HOSTS:
+        op.drop_column("import_settings", f"extdl_{host}_enabled")

alembic/versions/0051_image_source_provenance.py

@@ -0,0 +1,38 @@
+"""image_record: source_url + source_filehash (inline-image localization)
+
+#830 Phase 2. To render a post body faithfully we serve LOCAL copies of inline
+images instead of hotlinking the public CDN. The join key between a body
+`<img src=CDN>` and the local file is the CDN's 32-hex filehash (the same
+identity extract_media dedups by). Persist it (indexed) plus the full source
+URL for provenance/debugging. Both NULL for filesystem-imported / pre-existing
+rows — those fall back to hotlinking until re-downloaded.
+
+Revision ID: 0051
+Revises: 0050
+Create Date: 2026-06-14
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0051"
+down_revision: Union[str, None] = "0050"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column("image_record", sa.Column("source_url", sa.Text(), nullable=True))
+    op.add_column(
+        "image_record", sa.Column("source_filehash", sa.String(length=32), nullable=True)
+    )
+    op.create_index(
+        "ix_image_record_source_filehash", "image_record", ["source_filehash"]
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_image_record_source_filehash", table_name="image_record")
+    op.drop_column("image_record", "source_filehash")
+    op.drop_column("image_record", "source_url")

backend/app/api/settings.py

@@ -27,6 +27,20 @@ _EDITABLE_FIELDS = (
     "download_failure_warning_threshold",
     "series_suggest_enabled",
     "series_suggest_threshold",
+    "extdl_mega_enabled",
+    "extdl_gdrive_enabled",
+    "extdl_mediafire_enabled",
+    "extdl_dropbox_enabled",
+    "extdl_pixeldrain_enabled",
+)
+
+# Per-host external-download toggles — all plain booleans, validated uniformly.
+_EXTDL_TOGGLE_FIELDS = (
+    "extdl_mega_enabled",
+    "extdl_gdrive_enabled",
+    "extdl_mediafire_enabled",
+    "extdl_dropbox_enabled",
+    "extdl_pixeldrain_enabled",
 )
 
 
@@ -50,6 +64,11 @@ async def get_import_settings():
         "download_failure_warning_threshold": row.download_failure_warning_threshold,
         "series_suggest_enabled": row.series_suggest_enabled,
         "series_suggest_threshold": row.series_suggest_threshold,
+        "extdl_mega_enabled": row.extdl_mega_enabled,
+        "extdl_gdrive_enabled": row.extdl_gdrive_enabled,
+        "extdl_mediafire_enabled": row.extdl_mediafire_enabled,
+        "extdl_dropbox_enabled": row.extdl_dropbox_enabled,
+        "extdl_pixeldrain_enabled": row.extdl_pixeldrain_enabled,
     })
 
 
@@ -106,6 +125,9 @@ async def update_import_settings():
         return jsonify(
             {"error": "series_suggest_enabled must be a boolean"}
         ), 400
+    for tog in _EXTDL_TOGGLE_FIELDS:
+        if tog in body and not isinstance(body[tog], bool):
+            return jsonify({"error": f"{tog} must be a boolean"}), 400
     if "series_suggest_threshold" in body:
         v = body["series_suggest_threshold"]
         if not isinstance(v, (int, float)) or isinstance(v, bool) or v < 0 or v > 1:

backend/app/celery_app.py

@@ -30,6 +30,7 @@ def make_celery() -> Celery:
             "backend.app.tasks.maintenance",
             "backend.app.tasks.ml",
             "backend.app.tasks.download",
+            "backend.app.tasks.external",
             "backend.app.tasks.backup",
             "backend.app.tasks.admin",
             "backend.app.tasks.library_audit",
@@ -42,6 +43,9 @@ def make_celery() -> Celery:
             "backend.app.tasks.ml.*": {"queue": "ml"},
             "backend.app.tasks.thumbnail.*": {"queue": "thumbnail"},
             "backend.app.tasks.download.*": {"queue": "download"},
+            # External file-host fetches are downloads — same lane (they can run
+            # long, but the download worker already tolerates long backfills).
+            "backend.app.tasks.external.*": {"queue": "download"},
             "backend.app.tasks.scan.*": {"queue": "scan"},
             # `maintenance` is the QUICK lane — recovery sweeps, vacuum, cleanup
             # (concurrency-1 on the scheduler). The long one-shots (DB backups,
@@ -150,6 +154,21 @@ def make_celery() -> Celery:
                 "task": "backend.app.tasks.thumbnail.backfill_thumbnails",
                 "schedule": 86400.0,
             },
+            # External file-host downloads (#830): a steady sweep catches links
+            # the post-download hook missed (worker down, etc.); recovery re-tries
+            # dead links daily; retention prunes long-dead rows.
+            "extdl-sweep": {
+                "task": "backend.app.tasks.external.sweep_external_links",
+                "schedule": 600.0,  # every 10 min
+            },
+            "extdl-recover-daily": {
+                "task": "backend.app.tasks.external.recover_external_links",
+                "schedule": 86400.0,
+            },
+            "extdl-prune-daily": {
+                "task": "backend.app.tasks.external.prune_external_links",
+                "schedule": 86400.0,
+            },
         },
         timezone="UTC",
     )

backend/app/models/__init__.py

@@ -7,6 +7,7 @@ from .backup_run import BackupRun
 from .base import Base
 from .credential import Credential
 from .download_event import DownloadEvent
+from .external_link import ExternalLink
 from .image_prediction import ImagePrediction
 from .image_provenance import ImageProvenance
 from .image_record import ImageRecord
@@ -52,6 +53,7 @@ __all__ = [
     "TagKind",
     "image_tag",
     "DownloadEvent",
+    "ExternalLink",
     "ImportBatch",
     "ImportTask",
     "ImportSettings",

backend/app/models/external_link.py

@@ -0,0 +1,73 @@
+"""ExternalLink — an off-platform file-host link found in a post body.
+
+Creators host the actual files (films, packs) on mega.nz / Google Drive /
+MediaFire / Dropbox / Pixeldrain and drop the link in the post text. This row
+is the record that the link existed (so nothing is silently dropped), the
+dedup + dead-letter ledger for fetching it, and the driver the download worker
+walks. `url` keeps the FULL link including the `#fragment` (mega's decryption
+key) — truncating it makes the file undownloadable.
+
+status lifecycle: pending → downloading → downloaded | failed | dead
+(too many attempts) | skipped (host disabled). `attachment_id` links the
+captured file once a download lands (SET NULL so deleting the attachment
+doesn't delete the link record).
+"""
+
+from datetime import datetime
+
+from sqlalchemy import (
+    DateTime,
+    Float,
+    ForeignKey,
+    Index,
+    Integer,
+    String,
+    Text,
+    func,
+    text,
+)
+from sqlalchemy.orm import Mapped, mapped_column
+
+from .base import Base
+
+# Kept in sync with link_extract.SUPPORTED_HOSTS and the CHECK in migration 0049.
+HOSTS = ("mega", "gdrive", "mediafire", "dropbox", "pixeldrain")
+STATUSES = ("pending", "downloading", "downloaded", "failed", "skipped", "dead")
+
+
+class ExternalLink(Base):
+    __tablename__ = "external_link"
+    __table_args__ = (
+        # One row per (post, url). The full url (incl. #fragment) is the identity
+        # — the same file linked twice in a post collapses to one row.
+        Index("uq_external_link_post_url", "post_id", "url", unique=True),
+        Index("ix_external_link_status", "status"),
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    post_id: Mapped[int] = mapped_column(
+        ForeignKey("post.id", ondelete="CASCADE"), nullable=False, index=True
+    )
+    artist_id: Mapped[int | None] = mapped_column(
+        ForeignKey("artist.id", ondelete="SET NULL"), nullable=True, index=True
+    )
+    host: Mapped[str] = mapped_column(String(16), nullable=False)
+    url: Mapped[str] = mapped_column(Text, nullable=False)
+    label: Mapped[str | None] = mapped_column(Text, nullable=True)
+    status: Mapped[str] = mapped_column(
+        String(16), nullable=False, server_default="pending"
+    )
+    attempts: Mapped[int] = mapped_column(
+        Integer, nullable=False, server_default=text("0")
+    )
+    last_error: Mapped[str | None] = mapped_column(Text, nullable=True)
+    attachment_id: Mapped[int | None] = mapped_column(
+        ForeignKey("post_attachment.id", ondelete="SET NULL"), nullable=True
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+    completed_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )
+    duration_seconds: Mapped[float | None] = mapped_column(Float, nullable=True)

backend/app/models/image_record.py

@@ -49,6 +49,18 @@ class ImageRecord(Base):
     # Thumbnail (populated by FC-2)
     thumbnail_path: Mapped[str | None] = mapped_column(Text, nullable=True)
 
+    # Source provenance for downloaded media (#830 Phase 2). `source_url` is the
+    # CDN/origin URL the file was fetched from (debugging + future re-fetch).
+    # `source_filehash` is the URL's 32-hex CDN identity segment
+    # (utils.paths.filehash_from_url) — the JOIN KEY that maps a post body's
+    # inline `<img src=CDN>` back to this local copy so the rendered body serves
+    # our stored image instead of hotlinking the public source. Indexed for the
+    # render-time lookup. NULL for filesystem-imported / pre-Phase-2 rows.
+    source_url: Mapped[str | None] = mapped_column(Text, nullable=True)
+    source_filehash: Mapped[str | None] = mapped_column(
+        String(32), nullable=True, index=True
+    )
+
     # Origin / provenance pointers
     origin: Mapped[str] = mapped_column(Enum(*ORIGIN_CHOICES, name="origin_enum"), nullable=False)
     primary_post_id: Mapped[int | None] = mapped_column(

backend/app/models/import_settings.py

@@ -73,6 +73,25 @@ class ImportSettings(Base):
         Float, nullable=False, default=0.5,
     )
 
+    # #830 off-platform file-host downloads — per-host enable lever (default on,
+    # rule #26). Column names are extdl_<host>_enabled so the worker reads them
+    # via getattr(settings, f"extdl_{host}_enabled", True).
+    extdl_mega_enabled: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=True, server_default="true",
+    )
+    extdl_gdrive_enabled: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=True, server_default="true",
+    )
+    extdl_mediafire_enabled: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=True, server_default="true",
+    )
+    extdl_dropbox_enabled: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=True, server_default="true",
+    )
+    extdl_pixeldrain_enabled: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=True, server_default="true",
+    )
+
     @classmethod
     async def load(cls, session) -> ImportSettings:
         """The singleton settings row (id=1), via an async session."""

backend/app/services/download_service.py

@@ -380,6 +380,29 @@ class DownloadService:
             else:
                 import_summary["errors"] += 1
 
+        # Post-only records: media-less posts (pure-text) the native ingester
+        # captured so the artist archive is complete. Upsert each (keyed on
+        # external_post_id → updates the same Post a media import would create,
+        # never doubles). No file to clean up; the sidecar stays on disk.
+        for rec_str in getattr(dl_result, "post_record_paths", None) or []:
+            rec_path = Path(rec_str)
+            if not rec_path.exists():  # noqa: ASYNC240
+                continue
+
+            def _upsert(p=rec_path):
+                return self.importer.upsert_post_record(
+                    p, artist=artist, source=source_row,
+                )
+
+            await loop.run_in_executor(None, _upsert)
+
+        # Kick the off-platform file-host downloader for any links this run
+        # recorded (mega/gdrive/…). Global + idempotent (only claims pending/
+        # retryable rows); the beat sweep is the backstop. Lazy import dodges a
+        # task-module import cycle.
+        from ..tasks.external import sweep_external_links
+        sweep_external_links.delay()
+
         ev = (await self.async_session.execute(
             select(DownloadEvent).where(DownloadEvent.id == event_id)
         )).scalar_one()

backend/app/services/external_fetch.py

@@ -0,0 +1,248 @@
+"""Fetchers for off-platform file hosts (mega / gdrive / mediafire / dropbox /
+pixeldrain).
+
+A shared, reusable subsystem: given an external_link URL, fetch the file(s) into
+a destination directory and report the outcome. The download worker (separate
+slice) drives these off the external_link ledger; any in-house downloader can
+call `fetch_external()` directly.
+
+No single tool covers all five hosts, so a small registry maps host → fetch
+function behind one signature:
+
+    fetch_external(host, url, dest_dir, *, timeout, should_stop) -> FetchResult
+
+Backends:
+  - dropbox    : force the direct-download variant (dl=1) + stream GET.
+  - pixeldrain : GET the /api/file/{id} endpoint.
+  - mediafire  : scrape the download page for the direct link + stream GET.
+  - gdrive     : gdown (handles the confirm-token + virus-scan interstitial).
+  - mega       : `megatools dl` subprocess (public link incl. #key); needs the
+                 `megatools` binary in the runtime image (Debian apt package).
+
+Public links work credential-free (rule 26); per-host creds are a later Settings
+concern. Plain-HTTP homelab — no secure-context API. The HTTP / gdown /
+subprocess calls go through module-level seams so unit tests run without
+network, gdown, or MEGAcmd.
+"""
+from __future__ import annotations
+
+import logging
+import os
+import re
+import subprocess
+from collections.abc import Callable
+from dataclasses import dataclass, field
+from pathlib import Path
+from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
+
+import requests
+
+log = logging.getLogger(__name__)
+
+_CHUNK = 1 << 16
+_DEFAULT_TIMEOUT = 600.0
+_USER_AGENT = (
+    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
+    "(KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36"
+)
+# MediaFire's download page embeds the real file URL in a download-button href.
+_MEDIAFIRE_RE = re.compile(
+    r'href="(https://download[^"]+?\.mediafire\.com/[^"]+)"', re.IGNORECASE
+)
+_CD_FILENAME_RE = re.compile(r'filename\*?=(?:UTF-8\'\')?"?([^";]+)"?', re.IGNORECASE)
+
+
+@dataclass
+class FetchResult:
+    files: list[Path] = field(default_factory=list)
+    bytes: int = 0
+    error: str | None = None
+
+    @property
+    def ok(self) -> bool:
+        return self.error is None and bool(self.files)
+
+
+class ExternalFetchError(Exception):
+    """A fetch failed in a way worth recording on the link's last_error."""
+
+
+# -- seams (monkeypatched in tests) ----------------------------------------
+
+def _http_get(url: str, *, timeout: float, headers: dict | None = None,
+              stream: bool = True) -> requests.Response:
+    hdrs = {"User-Agent": _USER_AGENT}
+    if headers:
+        hdrs.update(headers)
+    return requests.get(url, timeout=timeout, headers=hdrs, stream=stream)
+
+
+def _gdown_download(url: str, out_dir: str) -> str | None:
+    """Download a Google Drive url into out_dir via gdown; return the written
+    path. Imported lazily so the dep is optional at module-import time."""
+    # Lazy import: keeps gdown an optional dep that isn't needed to import this
+    # module (e.g. the no-DB unit lane that never exercises a real fetch).
+    import gdown
+    # A trailing sep tells gdown to keep the server-side filename inside out_dir.
+    return gdown.download(url, output=out_dir + os.sep, quiet=True, fuzzy=True)
+
+
+def _run_mega_get(url: str, out_dir: str, *, timeout: float) -> None:
+    """Download a mega.nz public link (key in the #fragment) into out_dir via
+    `megatools dl` (the Debian `megatools` package). Raises ExternalFetchError
+    on non-zero exit."""
+    # Fixed argv (not shell): only `url` is external input, passed positionally,
+    # so there's no shell-injection surface.
+    proc = subprocess.run(
+        ["megatools", "dl", "--path", out_dir, url],
+        capture_output=True, text=True, timeout=timeout, check=False,
+    )
+    if proc.returncode != 0:
+        raise ExternalFetchError(
+            f"megatools dl exit {proc.returncode}: {(proc.stderr or '').strip()[:300]}"
+        )
+
+
+# -- helpers ---------------------------------------------------------------
+
+def _safe_name(name: str, fallback: str) -> str:
+    name = os.path.basename((name or "").strip().strip('"'))
+    # Strip path separators / control chars; never empty.
+    name = re.sub(r'[<>:"/\\|?*\x00-\x1f]', "_", name).strip(". ")
+    return name or fallback
+
+
+def _filename_from(resp: requests.Response, url: str, fallback: str) -> str:
+    cd = resp.headers.get("Content-Disposition", "")
+    m = _CD_FILENAME_RE.search(cd)
+    if m:
+        return _safe_name(m.group(1), fallback)
+    path_name = os.path.basename(urlsplit(url).path)
+    return _safe_name(path_name, fallback)
+
+
+def _stream_to_file(resp: requests.Response, dest: Path,
+                    should_stop: Callable[[], bool]) -> int:
+    """Stream a response body to `dest` (atomic via .part). Returns byte count.
+    Honors should_stop between chunks (partial file removed)."""
+    part = dest.with_name(dest.name + ".part")
+    total = 0
+    try:
+        with part.open("wb") as fh:
+            for chunk in resp.iter_content(chunk_size=_CHUNK):
+                if should_stop():
+                    raise ExternalFetchError("stopped")
+                if chunk:
+                    fh.write(chunk)
+                    total += len(chunk)
+    except BaseException:
+        part.unlink(missing_ok=True)
+        raise
+    os.replace(part, dest)
+    return total
+
+
+def _get_to_dir(url: str, dest_dir: Path, *, timeout: float,
+                should_stop: Callable[[], bool], fallback: str,
+                headers: dict | None = None) -> FetchResult:
+    resp = _http_get(url, timeout=timeout, headers=headers, stream=True)
+    if resp.status_code != 200:
+        return FetchResult(error=f"HTTP {resp.status_code} for {url}")
+    name = _filename_from(resp, url, fallback)
+    dest = dest_dir / name
+    written = _stream_to_file(resp, dest, should_stop)
+    return FetchResult(files=[dest], bytes=written)
+
+
+# -- per-host fetchers -----------------------------------------------------
+
+def _fetch_dropbox(url: str, dest_dir: Path, *, timeout: float,
+                   should_stop: Callable[[], bool]) -> FetchResult:
+    # Force the direct-download variant: dl=1 (Dropbox serves an HTML preview
+    # for dl=0). Rewrite/insert the param rather than string-replace so ?dl=0,
+    # &dl=0, and a missing param all resolve.
+    parts = urlsplit(url)
+    q = dict(parse_qsl(parts.query))
+    q["dl"] = "1"
+    direct = urlunsplit(parts._replace(query=urlencode(q)))
+    return _get_to_dir(direct, dest_dir, timeout=timeout,
+                       should_stop=should_stop, fallback="dropbox-file")
+
+
+def _fetch_pixeldrain(url: str, dest_dir: Path, *, timeout: float,
+                      should_stop: Callable[[], bool]) -> FetchResult:
+    # /u/{id} (and /l/{id}) → the API file endpoint.
+    file_id = urlsplit(url).path.rstrip("/").split("/")[-1]
+    if not file_id:
+        return FetchResult(error=f"no pixeldrain id in {url}")
+    api = f"https://pixeldrain.com/api/file/{file_id}"
+    return _get_to_dir(api, dest_dir, timeout=timeout,
+                       should_stop=should_stop, fallback=f"{file_id}.bin")
+
+
+def _fetch_mediafire(url: str, dest_dir: Path, *, timeout: float,
+                     should_stop: Callable[[], bool]) -> FetchResult:
+    page = _http_get(url, timeout=timeout, stream=False)
+    if page.status_code != 200:
+        return FetchResult(error=f"HTTP {page.status_code} for mediafire page")
+    m = _MEDIAFIRE_RE.search(page.text or "")
+    if not m:
+        return FetchResult(error="mediafire direct link not found on page")
+    return _get_to_dir(m.group(1), dest_dir, timeout=timeout,
+                       should_stop=should_stop, fallback="mediafire-file")
+
+
+def _fetch_gdrive(url: str, dest_dir: Path, *, timeout: float,
+                  should_stop: Callable[[], bool]) -> FetchResult:
+    out = _gdown_download(url, str(dest_dir))
+    if not out:
+        return FetchResult(error="gdown returned no file (quota / private?)")
+    p = Path(out)
+    if not p.exists():
+        return FetchResult(error=f"gdown reported {out} but it is missing")
+    return FetchResult(files=[p], bytes=p.stat().st_size)
+
+
+def _fetch_mega(url: str, dest_dir: Path, *, timeout: float,
+                should_stop: Callable[[], bool]) -> FetchResult:
+    before = set(dest_dir.iterdir()) if dest_dir.exists() else set()
+    _run_mega_get(url, str(dest_dir), timeout=timeout)
+    new = [p for p in dest_dir.iterdir() if p not in before and p.is_file()]
+    if not new:
+        return FetchResult(error="mega-get wrote no new file")
+    return FetchResult(files=new, bytes=sum(p.stat().st_size for p in new))
+
+
+_REGISTRY: dict[str, Callable[..., FetchResult]] = {
+    "dropbox": _fetch_dropbox,
+    "pixeldrain": _fetch_pixeldrain,
+    "mediafire": _fetch_mediafire,
+    "gdrive": _fetch_gdrive,
+    "mega": _fetch_mega,
+}
+
+SUPPORTED_HOSTS = tuple(_REGISTRY)
+
+
+def fetch_external(host: str, url: str, dest_dir: Path, *,
+                   timeout: float = _DEFAULT_TIMEOUT,
+                   should_stop: Callable[[], bool] = lambda: False) -> FetchResult:
+    """Fetch `url` (a `host` link) into `dest_dir`. Returns a FetchResult; never
+    raises — any backend error (transport, non-200, scrape miss, subprocess
+    failure, stop) is captured on `.error` so the worker can record it and move
+    on."""
+    fetcher = _REGISTRY.get(host)
+    if fetcher is None:
+        return FetchResult(error=f"unsupported host {host!r}")
+    dest_dir.mkdir(parents=True, exist_ok=True)
+    try:
+        return fetcher(url, dest_dir, timeout=timeout, should_stop=should_stop)
+    except requests.RequestException as exc:
+        return FetchResult(error=f"transport error: {exc}")
+    except subprocess.TimeoutExpired:
+        return FetchResult(error="timed out")
+    except ExternalFetchError as exc:
+        return FetchResult(error=str(exc))
+    except Exception as exc:  # never let a backend quirk kill the worker
+        log.warning("external fetch (%s) failed for %s: %s", host, url, exc)
+        return FetchResult(error=f"{type(exc).__name__}: {exc}")

backend/app/services/gallery_dl.py

@@ -147,6 +147,12 @@ class DownloadResult:
     files_quarantined: int = 0
     quarantined_paths: list[str] = field(default_factory=list)
     written_paths: list[str] = field(default_factory=list)
+    # Native ingester only: post-only sidecar paths for posts that have NO
+    # downloadable media (pure-text posts), so the importer can still upsert the
+    # Post + its body. Empty on the gallery-dl path. Phase 3 imports these via
+    # Importer.upsert_post_record (keyed on external_post_id → updates, never
+    # doubles, the same Post a media import would create).
+    post_record_paths: list[str] = field(default_factory=list)
     stdout: str = ""
     stderr: str = ""
     return_code: int = 0

backend/app/services/importer.py

@@ -23,6 +23,7 @@ from sqlalchemy.orm import Session
 
 from ..models import (
     Artist,
+    ExternalLink,
     ImageProvenance,
     ImageRecord,
     ImportSettings,
@@ -34,6 +35,7 @@ from ..utils import safe_probe
 from ..utils.paths import (
     derive_subdir,
     derive_top_level_artist,
+    filehash_from_url,
     hash_suffixed_name,
     safe_ext,
 )
@@ -42,6 +44,7 @@ from ..utils.sidecar import find_sidecar, parse_sidecar
 from ..utils.slug import slugify
 from .archive_extractor import extract_archive, is_archive
 from .attachment_store import AttachmentStore
+from .link_extract import extract_external_links
 from .thumbnailer import Thumbnailer
 
 log = logging.getLogger(__name__)
@@ -305,6 +308,32 @@ class Importer:
             ),
         )
 
+    def _sync_external_links(self, post: Post) -> None:
+        """Record off-platform file-host links (mega/gdrive/mediafire/dropbox/
+        pixeldrain) found in the post body, so they're never silently dropped
+        and the download worker can fetch them later. Shared by every platform's
+        import (runs off Post.description). INSERT-MISSING only — an existing row
+        keeps its status/attempts (a re-import must not reset a link already
+        downloaded or dead-lettered). Identity is (post_id, url); the full url
+        incl. #fragment is preserved by the extractor."""
+        links = extract_external_links(post.description)
+        if not links:
+            return
+        self.session.flush()  # ensure post.id is assigned before we reference it
+        existing = set(self.session.execute(
+            select(ExternalLink.url).where(ExternalLink.post_id == post.id)
+        ).scalars().all())
+        for link in links:
+            if link.url in existing:
+                continue
+            self.session.add(ExternalLink(
+                post_id=post.id,
+                artist_id=post.artist_id,
+                host=link.host,
+                url=link.url,
+                label=link.label,
+            ))
+
     def import_one(self, source: Path) -> ImportResult:
         """Dispatch by kind. Media → normal pipeline. Archive → extract
         media members (one Post via the archive-adjacent sidecar) and
@@ -710,6 +739,67 @@ class Importer:
         self.session.commit()
         return ImportResult(status="refreshed", image_id=existing.id)
 
+    def upsert_post_record(
+        self, sidecar: Path, *, artist: Artist | None = None,
+        source: Source | None = None,
+    ) -> bool:
+        """Upsert the Post for a post-ONLY sidecar (a media-less post), so the
+        artist archive includes text posts (their body + external links).
+
+        Reuses `_find_or_create_post` (keyed on external_post_id) so it UPDATES
+        the SAME Post a media import would create — never doubles. Fields are
+        FILLED, never clobbered with empty: parse_sidecar yields None for empty
+        values, and a None field is left untouched (an empty feed body never
+        wipes a populated one). Returns True if a Post was upserted, False if the
+        sidecar was unusable (parse failure / no artist)."""
+        try:
+            data = json.loads(sidecar.read_text("utf-8"))
+            if not isinstance(data, dict):
+                raise ValueError("sidecar JSON is not an object")
+        except Exception as exc:
+            log.warning("post-record sidecar parse failed for %s: %s", sidecar, exc)
+            return False
+
+        sd = parse_sidecar(data)
+
+        if artist is None:
+            name = self._sidecar_artist_name(data)
+            artist = self._upsert_artist(name) if name else None
+        if artist is None:
+            log.warning("post-record sidecar %s has no artist; skipping", sidecar)
+            return False
+
+        if source is not None:
+            src = source
+        else:
+            platform = sd.platform or "unknown"
+            src = self._lookup_source_for_sidecar(
+                artist_id=artist.id, platform=platform,
+            )
+
+        epid = sd.external_post_id or sidecar.stem
+        post = self._find_or_create_post(
+            source_id=src.id if src else None,
+            external_post_id=epid,
+            artist_id=artist.id,
+        )
+        if post.artist_id is None:
+            post.artist_id = artist.id
+        if sd.post_url is not None:
+            post.post_url = sd.post_url
+        if sd.post_title is not None:
+            post.post_title = sd.post_title
+        if sd.post_date is not None:
+            post.post_date = sd.post_date
+        if sd.description is not None:
+            post.description = sd.description
+        if sd.attachment_count is not None:
+            post.attachment_count = sd.attachment_count
+        post.raw_metadata = sd.raw
+        self._sync_external_links(post)
+        self.session.commit()
+        return True
+
     def attach_in_place(
         self,
         path: Path,
@@ -957,6 +1047,14 @@ class Importer:
         if record.artist_id is None:
             record.artist_id = artist.id
 
+        # #830 Phase 2: persist the file's CDN source identity (NULL-only, so a
+        # re-import / enrich-on-duplicate never clobbers it) — the filehash is
+        # the join key that lets post_feed_service remap the body's inline
+        # `<img src=CDN>` to this local copy at render time.
+        if sd.source_url and record.source_filehash is None:
+            record.source_url = sd.source_url
+            record.source_filehash = filehash_from_url(sd.source_url)
+
         if explicit_source is not None:
             src = explicit_source
         else:
@@ -982,6 +1080,7 @@ class Importer:
         if sd.attachment_count is not None:
             post.attachment_count = sd.attachment_count
         post.raw_metadata = sd.raw
+        self._sync_external_links(post)
 
         # Race-safe (image_record_id, post_id) upsert — mirrors the
         # _find_or_create_source/post savepoint pattern. The plain

backend/app/services/ingest_core.py

@@ -118,10 +118,16 @@ class Ingester:
         # tick has no resumable backfill state.
         checkpoint = mode in ("backfill", "recovery")
         ledger_key = self._ledger_key
+        # Optional seams (Patreon native ingester): capture pure-text posts that
+        # have NO downloadable media so the artist archive is complete. Absent on
+        # stub clients/downloaders (unit tests) → media-less posts skipped as before.
+        post_record_key = getattr(self.client, "post_record_key", None)
+        write_post_record = getattr(self.downloader, "write_post_record", None)
         start = time.monotonic()
         last_live = start  # plan #709: last live-progress write timestamp
         log_lines: list[str] = []
         written: list[str] = []
+        post_records: list[str] = []
         quarantined_paths: list[str] = []
         downloaded = 0
         errors = 0
@@ -158,6 +164,7 @@ class Ingester:
                 files_quarantined=quarantined,
                 quarantined_paths=list(quarantined_paths),
                 written_paths=written,
+                post_record_paths=list(post_records),
                 stdout="\n".join(log_lines),
                 stderr="",
                 return_code=return_code,
@@ -220,6 +227,29 @@ class Ingester:
                 # resume_cursor None, so everything counts.
                 if not (resume_cursor and page_cursor == resume_cursor):
                     chunk_new_posts += 1
+                # Capture the post body + external links ONCE per post (gated by
+                # the synthetic post key in the seen-ledger), for EVERY post —
+                # whether or not it has downloadable media. This is what makes a
+                # backfill/recovery re-walk RECAPTURE bodies + links for posts
+                # whose media is already on disk: re-downloading existing media
+                # never fills links the system never had, so the body recapture
+                # has to ride the walk itself. Detail-fetch (for an empty feed
+                # body) happens at most once per post — the gate then spares it on
+                # later walks. bypass_seen (recovery) re-captures unconditionally.
+                if post_record_key and write_post_record:
+                    rk = post_record_key(post)
+                    if rk is not None:
+                        pkey, ppid = rk
+                        already = (
+                            set() if bypass_seen
+                            else self._seen_keys(source_id, [pkey])
+                        )
+                        if pkey not in already:
+                            rec_path = write_post_record(post, artist_slug)
+                            if rec_path is not None:
+                                post_records.append(str(rec_path))
+                                self._mark_seen(source_id, [(pkey, ppid)])
+
                 media = self.client.extract_media(post, included)
                 if not media:
                     continue

backend/app/services/link_extract.py

@@ -0,0 +1,105 @@
+"""Extract off-platform file-host links from a post body.
+
+Pure (no I/O) so it's unit-testable and reusable by EVERY in-house downloader's
+import path — every platform stores its body in `Post.description`, so running
+this there covers them all. Finds links to the supported external file hosts in
+a post's HTML body (both `<a href="...">` anchors and bare URLs in text),
+unwraps a Patreon outbound-redirect wrapper, and preserves the FULL url
+including the `#fragment` (mega.nz puts the decryption key there) and the query
+string — without those a mega download is impossible.
+"""
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from html import unescape
+from urllib.parse import parse_qs, urlsplit
+
+# Supported file-host enum values (kept in sync with the external_link CHECK).
+SUPPORTED_HOSTS = ("mega", "gdrive", "mediafire", "dropbox", "pixeldrain")
+
+# Bare-domain suffix → canonical host label. Matched against the URL netloc
+# (exact or as a dotted suffix, so www./dl. subdomains resolve too).
+_HOST_MAP = {
+    "mega.nz": "mega",
+    "mega.co.nz": "mega",
+    "drive.google.com": "gdrive",
+    "mediafire.com": "mediafire",
+    "dropbox.com": "dropbox",
+    "pixeldrain.com": "pixeldrain",
+}
+
+# `<a href="...">label</a>` — DOTALL so a label spanning tags/newlines is caught.
+_HREF_RE = re.compile(
+    r"""<a\b[^>]*?\bhref=["']([^"']+)["'][^>]*>(.*?)</a>""",
+    re.IGNORECASE | re.DOTALL,
+)
+_TAG_RE = re.compile(r"<[^>]+>")
+# Bare http(s) URL in text. Stops at whitespace, quotes, angle brackets, and
+# closing brackets — but KEEPS `#`, `&`, `?`, `=` so fragments/queries survive.
+_URL_RE = re.compile(r"""https?://[^\s"'<>)\]}]+""", re.IGNORECASE)
+
+
+@dataclass(frozen=True)
+class ExtractedLink:
+    host: str            # one of SUPPORTED_HOSTS
+    url: str             # full url incl. query + #fragment
+    label: str | None    # visible anchor text, when present
+
+
+def _netloc(url: str) -> str:
+    # Lowercase host without credentials or port.
+    return urlsplit(url).netloc.lower().split("@")[-1].split(":")[0]
+
+
+def host_for(url: str) -> str | None:
+    """Canonical host label for a url, or None if it's not a supported host."""
+    netloc = _netloc(url)
+    for suffix, host in _HOST_MAP.items():
+        if netloc == suffix or netloc.endswith("." + suffix):
+            return host
+    return None
+
+
+def _unwrap(url: str) -> str:
+    """Unwrap a Patreon outbound-redirect wrapper to its inner target, so a
+    wrapped mega/gdrive link resolves to the real host. Patreon has used both
+    `www.patreon.com/...?url=<encoded>` and the `l.patreon.com` shim; check the
+    common target-param names. Non-Patreon urls pass through untouched."""
+    netloc = _netloc(url)
+    if not (netloc == "patreon.com" or netloc.endswith(".patreon.com")):
+        return url
+    qs = parse_qs(urlsplit(url).query)
+    for key in ("url", "u", "ext_url", "redirect", "target"):
+        vals = qs.get(key)
+        if vals and vals[0]:
+            return unescape(vals[0]).strip()
+    return url
+
+
+def extract_external_links(html: str | None) -> list[ExtractedLink]:
+    """All supported-host links in `html`, de-duplicated by url (first wins, so
+    an anchor's label is kept over a later bare sighting of the same url)."""
+    if not html:
+        return []
+    found: dict[str, ExtractedLink] = {}
+
+    # 1) Anchors first — they carry a human label ("Mega - Streamable").
+    for raw_href, inner in _HREF_RE.findall(html):
+        url = _unwrap(unescape(raw_href).strip())
+        host = host_for(url)
+        if host is None:
+            continue
+        label = unescape(_TAG_RE.sub("", inner)).strip() or None
+        found.setdefault(url, ExtractedLink(host=host, url=url, label=label))
+
+    # 2) Bare URLs pasted in text (no label). Trailing prose punctuation is
+    #    trimmed; the href values already captured above de-dup away here.
+    for raw in _URL_RE.findall(html):
+        url = _unwrap(unescape(raw).strip().rstrip(".,;"))
+        host = host_for(url)
+        if host is None:
+            continue
+        found.setdefault(url, ExtractedLink(host=host, url=url, label=None))
+
+    return list(found.values())

backend/app/services/patreon_client.py

@@ -39,7 +39,7 @@ from urllib.parse import parse_qs, urlsplit
 
 import requests
 
-from ..utils.paths import safe_ext
+from ..utils.paths import filehash_from_url, safe_ext
 
 log = logging.getLogger(__name__)
 
@@ -90,12 +90,6 @@ _FIELDS_POST = (
 _FIELDS_MEDIA = "id,image_urls,download_url,metadata,file_name"
 _FIELDS_CAMPAIGN = "name,url"
 
-# A CDN download URL embeds a 32-char hex (MD5) path segment; that segment is
-# Patreon's stable per-file identity and is what we dedup + ledger against.
-# Same role gallery-dl's _filehash plays. Match the FIRST 32-hex run anywhere
-# in the URL (path or query); real Patreon CDN URLs carry exactly one.
-_FILEHASH_RE = re.compile(r"([0-9a-fA-F]{32})")
-
 # Inline post `content` is HTML; images are emitted as <img ... src="...">.
 # Pull every src; downstream dedup collapses any that duplicate a gallery item
 # by filehash. Tolerant of attribute ordering and single/double quotes.
@@ -188,10 +182,9 @@ def _load_session(cookies_path: str | Path | None) -> requests.Session:
 
 
 def _filehash(url: str) -> str | None:
-    if not url:
-        return None
-    match = _FILEHASH_RE.search(url)
-    return match.group(1).lower() if match else None
+    # Delegate to the shared extractor (utils.paths) so capture-time persistence
+    # and render-time inline-image matching use the EXACT same identity.
+    return filehash_from_url(url)
 
 
 def _basename_from_url(url: str) -> str:
@@ -526,6 +519,19 @@ class PatreonClient:
             "date": published if isinstance(published, str) else None,
         }
 
+    @staticmethod
+    def post_record_key(post: dict) -> tuple[str, str] | None:
+        """`(ledger_key, post_id)` for a media-less post's seen-ledger entry, or
+        None when the post has no id. The synthetic `post:<id>` key lets the
+        generic core gate text-post capture through the SAME seen-ledger as media
+        — so a text post's body is detail-fetched + recorded ONCE, not re-fetched
+        every tick. Part of the client contract the core uses via getattr."""
+        pid = post.get("id")
+        pid = str(pid) if pid is not None else ""
+        if not pid:
+            return None
+        return (f"post:{pid}", pid)
+
     # -- iteration ---------------------------------------------------------
 
     def iter_posts(
@@ -562,6 +568,49 @@ class PatreonClient:
                 return
             current_cursor = next_cursor
 
+    # -- detail (full body enrichment) -------------------------------------
+
+    def fetch_post_detail_content(self, post_id: str) -> str | None:
+        """Best-effort fetch of a post's full HTML `content` from the per-post
+        DETAIL endpoint (`/api/posts/{id}`).
+
+        The feed/list endpoint (`/api/posts`) frequently returns `content` as
+        null even though we request it — the full body (its formatting, inline
+        `<img>`, and external `<a href>` links) only comes back from the
+        single-post detail resource. The downloader calls this to enrich a post
+        whose feed body was empty before writing the importer sidecar.
+
+        Best-effort BY DESIGN: the media download is the primary job, so a body
+        we can't fetch must never fail the walk. Every failure path (no id,
+        transport error, non-200, non-JSON, missing/empty content) returns None
+        rather than raising — distinct from the loud drift/auth raises on the
+        feed path, which gate real downloads.
+        """
+        if not post_id:
+            return None
+        if self._request_sleep > 0:
+            time.sleep(self._request_sleep)  # pace the API endpoint (plan #703)
+        url = f"{_POSTS_URL}/{post_id}"
+        params = {"fields[post]": "content", "json-api-version": "1.0"}
+        try:
+            resp = self._session.get(url, params=params, timeout=_TIMEOUT_SECONDS)
+        except requests.RequestException as exc:
+            log.warning("Patreon post-detail fetch failed (post %s): %s", post_id, exc)
+            return None
+        if resp.status_code != 200:
+            log.warning(
+                "Patreon post-detail fetch HTTP %s (post %s)", resp.status_code, post_id
+            )
+            return None
+        try:
+            payload = resp.json()
+        except ValueError:
+            return None
+        data = payload.get("data") if isinstance(payload, dict) else None
+        attrs = data.get("attributes") if isinstance(data, dict) else None
+        content = attrs.get("content") if isinstance(attrs, dict) else None
+        return content if isinstance(content, str) and content.strip() else None
+
     # -- verify ------------------------------------------------------------
 
     def verify_auth(self, campaign_id: str) -> tuple[bool | None, str]:

backend/app/services/patreon_downloader.py

@@ -168,10 +168,17 @@ class PatreonDownloader:
         validate: bool = True,
         rate_limit: float = 0.0,
         session: requests.Session | None = None,
+        content_fetcher: Callable[[str], str | None] | None = None,
     ):
         self.images_root = Path(images_root)
         self.cookies_path = str(cookies_path) if cookies_path else None
         self._validate = validate
+        # Best-effort enrichment seam: (post_id) -> full HTML body, or None. The
+        # feed endpoint often omits `content`; the adapter wires this to
+        # PatreonClient.fetch_post_detail_content so the sidecar captures the
+        # real body (formatting + inline <img> + external <a href> links).
+        # None in unit tests / when enrichment isn't wanted.
+        self._content_fetcher = content_fetcher
         # Politeness: seconds to sleep before each actual media download (paces
         # the CDN; honors ImportSettings.download_rate_limit_seconds, the same
         # value gallery-dl used as its between-downloads `sleep`). 0 = no pacing.
@@ -287,7 +294,7 @@ class PatreonDownloader:
                     path=quarantine_dest, error=reason,
                 )
 
-        self._write_sidecar(post, out_path)
+        self._write_sidecar(post, out_path, source_url=media.url)
         return MediaOutcome(media=media, status="downloaded", path=out_path, error=None)
 
     # -- download seams ----------------------------------------------------
@@ -483,7 +490,9 @@ class PatreonDownloader:
 
     # -- sidecar -----------------------------------------------------------
 
-    def _write_sidecar(self, post: dict, media_path: Path) -> Path:
+    def _write_sidecar(
+        self, post: dict, media_path: Path, *, source_url: str | None = None
+    ) -> Path:
         """Write the importer-consumed sidecar next to `media_path`.
 
         Patreon uses base-default sidecar keys (parse_sidecar maps
@@ -491,10 +500,37 @@ class PatreonDownloader:
         content->description, published_at->post_date, url->post_url). Patreon
         registers no derive_post_url, so `url` is trusted as the permalink — we
         pass the post's attributes.url.
+
+        `source_url` is THIS file's CDN URL (#830 Phase 2). The importer persists
+        its filehash on the ImageRecord so the post body's inline `<img src>` can
+        be remapped to the local copy at render time.
         """
+        return self._write_sidecar_data(
+            post, media_path.with_suffix(".json"), source_url=source_url
+        )
+
+    def _write_sidecar_data(
+        self, post: dict, sidecar_path: Path, *, source_url: str | None = None
+    ) -> Path:
+        """Serialize the post's metadata to `sidecar_path`. Shared by the
+        per-media sidecar (next to each file) and the post-only sidecar
+        (`write_post_record`, for media-less posts). `source_url` is set only for
+        the per-media sidecar — a media-less post has no source file."""
         attrs = post.get("attributes") or {}
         title = attrs.get("title")
         content = attrs.get("content")
+        # The feed/list endpoint frequently returns an empty `content`; the full
+        # HTML body (formatting + inline <img> + external <a href> links) only
+        # comes from the per-post detail endpoint. Enrich on first write for this
+        # post and MEMOIZE by mutating the shared `post` dict — so a multi-image
+        # post fetches detail exactly once, and a fully-seen post (no fresh
+        # download → no sidecar write) never pays the extra GET.
+        if (not isinstance(content, str) or not content.strip()) and self._content_fetcher:
+            fetched = self._content_fetcher(str(post.get("id") or ""))
+            if fetched:
+                attrs["content"] = fetched
+                post["attributes"] = attrs
+                content = fetched
         published = attrs.get("published_at")
         url = attrs.get("url")
         data = {
@@ -505,6 +541,20 @@ class PatreonDownloader:
             "published_at": published if isinstance(published, str) else None,
             "url": url if isinstance(url, str) else None,
         }
-        sidecar_path = media_path.with_suffix(".json")
+        if source_url:
+            data["source_url"] = source_url
         sidecar_path.write_text(json.dumps(data, indent=2))
         return sidecar_path
+
+    def write_post_record(self, post: dict, artist_slug: str) -> Path | None:
+        """Write a post-ONLY sidecar (no media file) for a media-less post, so
+        the importer can still upsert the Post + its body — text posts often hold
+        the only copy of an external <a href> link. Named `_post.json`: the
+        leading underscore keeps it from colliding with a media sidecar
+        (`<NN>_<stem>.json`) and from being resolved as some media file's sidecar
+        by find_sidecar. Returns the path, or None when the post has no id."""
+        if not str(post.get("id") or ""):
+            return None
+        post_dir = self.images_root / artist_slug / "patreon" / _post_dir_name(post)
+        post_dir.mkdir(parents=True, exist_ok=True)
+        return self._write_sidecar_data(post, post_dir / "_post.json")

backend/app/services/patreon_ingester.py

@@ -112,7 +112,10 @@ class PatreonIngester(Ingester):
             downloader
             if downloader is not None
             else PatreonDownloader(
-                self.images_root, cookies_path, validate=validate, rate_limit=rate_limit
+                self.images_root, cookies_path, validate=validate, rate_limit=rate_limit,
+                # Enrich empty feed bodies from the per-post detail endpoint, via
+                # the SAME client (shares its cookie session + request pacing).
+                content_fetcher=resolved_client.fetch_post_detail_content,
             )
         )
         super().__init__(

backend/app/services/post_feed_service.py

@@ -11,19 +11,28 @@ attachments from PostAttachment) so the API layer can jsonify directly.
 """
 from __future__ import annotations
 
+from html import unescape
+
 from sqlalchemy import and_, func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from ..models import (
     Artist,
+    ExternalLink,
     ImageProvenance,
     ImageRecord,
     Post,
     PostAttachment,
     Source,
 )
+from ..utils.html_sanitize import (
+    extract_img_srcs,
+    rewrite_img_srcs,
+    sanitize_post_html,
+)
+from ..utils.paths import filehash_from_url
 from ..utils.text import html_to_plain, truncate_at_word
-from .gallery_service import thumbnail_url
+from .gallery_service import image_url, thumbnail_url
 from .pagination import decode_cursor, encode_cursor
 
 DESCRIPTION_LIMIT = 280
@@ -191,8 +200,71 @@ class PostFeedService:
         atts_map = await self._attachments_for([post.id])
         item = self._to_dict(post, artist, source, thumbs_map, atts_map)
         item["description_full"] = html_to_plain(post.description)
+        # Sanitized HTML body for faithful (semantic) rendering in the post view;
+        # detail-only (the feed list stays lightweight plain text). None when the
+        # post has no body. Inline `<img>` sources are remapped to locally-served
+        # copies (#830 Phase 2) so the body never hotlinks the public CDN.
+        item["description_html"] = await self._localize_inline_images(
+            sanitize_post_html(post.description), post.artist_id,
+        )
+        item["external_links"] = await self._external_links_for(post.id)
         return item
 
+    async def _localize_inline_images(
+        self, html: str | None, artist_id: int | None,
+    ) -> str | None:
+        """Rewrite a post body's inline `<img src=CDN>` to locally-served copies.
+
+        The join key is the CDN filehash the downloader persisted on each
+        ImageRecord (source_filehash): for every body image whose filehash maps
+        to a stored image of THIS artist, swap the src to /images/<path>. Images
+        we never captured (or pre-Phase-2 rows with no filehash) are left as-is —
+        they keep hotlinking, which is the prior behavior. Scoped to the post's
+        artist so one creator's body never resolves to another's file."""
+        if not html or artist_id is None:
+            return html
+        srcs = extract_img_srcs(html)
+        if not srcs:
+            return html
+        # filehash -> the raw (as-in-HTML) src strings carrying it. A body can
+        # repeat the same image; keep every raw form so each is substituted.
+        by_hash: dict[str, list[str]] = {}
+        for raw in srcs:
+            fh = filehash_from_url(unescape(raw))
+            if fh:
+                by_hash.setdefault(fh, []).append(raw)
+        if not by_hash:
+            return html
+        rows = (await self.session.execute(
+            select(ImageRecord.source_filehash, ImageRecord.path)
+            .where(
+                ImageRecord.artist_id == artist_id,
+                ImageRecord.source_filehash.in_(list(by_hash)),
+            )
+        )).all()
+        replace: dict[str, str] = {}
+        for fh, path in rows:
+            for raw in by_hash.get(fh, ()):
+                replace[raw] = image_url(path)
+        return rewrite_img_srcs(html, replace)
+
+    async def _external_links_for(self, post_id: int) -> list[dict]:
+        """Off-platform file-host links recorded for a post (detail-only). Each
+        carries its host, full url, label, and download status so the post view
+        can surface them (and, later, a retry/download affordance)."""
+        rows = (await self.session.execute(
+            select(ExternalLink)
+            .where(ExternalLink.post_id == post_id)
+            .order_by(ExternalLink.id.asc())
+        )).scalars().all()
+        return [
+            {
+                "id": e.id, "host": e.host, "url": e.url,
+                "label": e.label, "status": e.status,
+            }
+            for e in rows
+        ]
+
     # --- composition helpers ---------------------------------------------
 
     async def _thumbnails_for(

backend/app/tasks/external.py

@@ -0,0 +1,356 @@
+"""Celery worker for off-platform file-host downloads (external_link ledger).
+
+Walks the external_link table and fetches each pending/retryable link via
+external_fetch, then routes the downloaded file(s) through the existing importer
+(attach_in_place) so an archive becomes ImageRecords and any other file a
+PostAttachment — linked to the SAME post the link came from (via a synthesized
+sidecar). Long-running-task hygiene (rule #89): per-fetch wall-clock timeout,
+attempt tracking + dead-letter, a recovery sweep, and retention of dead rows.
+
+Concurrency: a per-host Redis lock serializes fetches to one-per-host (mega /
+gdrive ban-avoidance), and an atomic status claim (pending/failed →
+downloading) stops two workers grabbing the same link.
+
+Files land in the artist's library tree (so an attached art image stays in
+place, mirroring the gallery-dl download path); a captured archive/attachment is
+copied into its store and the on-disk original removed.
+"""
+from __future__ import annotations
+
+import json
+import logging
+import time
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+
+import redis
+from sqlalchemy import delete, select, update
+
+from ..celery_app import celery
+from ..config import get_config
+from ..models import Artist, ExternalLink, ImportSettings, Post, Source
+from ..services.external_fetch import fetch_external
+from ..services.importer import Importer
+from ..services.thumbnailer import Thumbnailer
+from ._sync_engine import sync_session_factory as _sync_session_factory
+
+log = logging.getLogger(__name__)
+
+IMAGES_ROOT = Path("/images")
+
+# After this many failed attempts a link is dead-lettered (skipped by routine
+# sweeps; an operator recovery still re-attempts). Mirrors the ingester ledger.
+DEAD_LETTER_THRESHOLD = 3
+# Per-fetch wall-clock budget — films/packs are large; the celery time_limit is
+# the backstop above this.
+_FETCH_TIMEOUT = 3000.0
+# Links enqueued per sweep — bounds the burst when a big backfill records many.
+_SWEEP_BATCH = 50
+# Dead rows older than this are pruned (retention).
+_RETENTION_DAYS = 30
+# Per-host serialize lock.
+_LOCK_PREFIX = "fc:extdl_lock:"
+_LOCK_TTL = 3600
+_SERIALIZE_COUNTDOWN = 120
+_MAX_SERIALIZE_WAITS = 30
+
+_redis_client: redis.Redis | None = None
+
+
+def _redis() -> redis.Redis:
+    global _redis_client
+    if _redis_client is None:
+        _redis_client = redis.from_url(get_config().celery_broker_url)
+    return _redis_client
+
+
+def _host_enabled(settings: ImportSettings, host: str) -> bool:
+    """Per-host enable flag — defaults True (rule #26: works out of the box).
+    The Settings UI slice adds real columns; getattr keeps this forward-
+    compatible without a migration in this slice."""
+    return bool(getattr(settings, f"extdl_{host}_enabled", True))
+
+
+def _write_link_sidecar(file: Path, post: Post, platform: str, artist: Artist) -> None:
+    """Sidecar next to a fetched file so the importer links it to `post`
+    (category+id resolve the source+post; matches gallery-dl's sidecar shape)."""
+    data = {
+        "category": platform,
+        "id": post.external_post_id,
+        "url": post.post_url,
+        "title": post.post_title,
+        "artist": artist.name,
+    }
+    Path(str(file) + ".json").write_text(json.dumps(data))
+
+
+@celery.task(
+    name="backend.app.tasks.external.fetch_external_link",
+    bind=True,
+    soft_time_limit=3300,
+    time_limit=3600,
+)
+def fetch_external_link(self, link_id: int, _serialize_waits: int = 0) -> dict:
+    """Fetch one external_link and route its file(s) through the importer."""
+    SessionLocal = _sync_session_factory()
+
+    # Atomic claim: only an actionable (pending/failed) row transitions to
+    # downloading, so a duplicate enqueue (sweep + post-download hook) no-ops.
+    with SessionLocal() as session:
+        claimed = session.execute(
+            update(ExternalLink)
+            .where(
+                ExternalLink.id == link_id,
+                ExternalLink.status.in_(("pending", "failed")),
+            )
+            .values(status="downloading")
+            .returning(ExternalLink.id)
+        ).first()
+        session.commit()
+        if claimed is None:
+            return {"link_id": link_id, "skipped": "not claimable"}
+
+        link = session.get(ExternalLink, link_id)
+        post = session.get(Post, link.post_id) if link.post_id else None
+        if post is None:
+            link.status = "dead"
+            link.last_error = "post missing"
+            link.completed_at = datetime.now(UTC)
+            session.commit()
+            return {"link_id": link_id, "error": "post missing"}
+        artist = session.get(Artist, post.artist_id)
+        source = session.get(Source, post.source_id) if post.source_id else None
+        platform = source.platform if source is not None else "patreon"
+        settings = ImportSettings.load_sync(session)
+
+        if not _host_enabled(settings, link.host):
+            log.info("extdl skip (host disabled): link=%s host=%s", link_id, link.host)
+            link.status = "skipped"
+            link.last_error = f"{link.host} disabled"
+            link.completed_at = datetime.now(UTC)
+            session.commit()
+            return {"link_id": link_id, "skipped": "host disabled"}
+
+        host, url, attempts = link.host, link.url, link.attempts
+        log.info(
+            "extdl start: link=%s host=%s post=%s artist=%s attempt=%d url=%s",
+            link_id, host, post.id, artist.slug, attempts + 1, url,
+        )
+
+    # Per-host serialize: one fetch per host at a time. If busy, requeue.
+    lock = None
+    try:
+        lock = _redis().lock(f"{_LOCK_PREFIX}{host}", timeout=_LOCK_TTL, blocking=False)
+        got = bool(lock.acquire(blocking=False))
+    except redis.RedisError as exc:
+        log.warning("extdl lock unavailable for %s: %s — running uncapped", host, exc)
+        lock, got = None, True
+    if not got:
+        # Release the claim back to pending so a later run can pick it up, then
+        # requeue with a countdown (bounded, like download_source).
+        with SessionLocal() as session:
+            session.execute(
+                update(ExternalLink).where(ExternalLink.id == link_id)
+                .values(status="pending")
+            )
+            session.commit()
+        log.info(
+            "extdl requeue (host %s busy): link=%s wait=%d", host, link_id, _serialize_waits,
+        )
+        if _serialize_waits < _MAX_SERIALIZE_WAITS:
+            fetch_external_link.apply_async(
+                (link_id,), {"_serialize_waits": _serialize_waits + 1},
+                countdown=_SERIALIZE_COUNTDOWN,
+            )
+        else:
+            log.warning("extdl giving up requeue (host %s busy): link=%s", host, link_id)
+        return {"link_id": link_id, "requeued": "host busy"}
+
+    started = time.monotonic()
+    post_dir = IMAGES_ROOT / artist.slug / platform / "external" / str(post.external_post_id)
+    try:
+        result = fetch_external(host, url, post_dir, timeout=_FETCH_TIMEOUT)
+        with SessionLocal() as session:
+            link = session.get(ExternalLink, link_id)
+            if not result.ok:
+                log.warning(
+                    "extdl fetch failed: link=%s host=%s error=%s",
+                    link_id, host, result.error,
+                )
+                _record_failure(link, attempts, result.error or "fetch failed")
+                session.commit()
+                return {"link_id": link_id, "error": result.error}
+
+            log.info(
+                "extdl fetched: link=%s host=%s files=%d bytes=%d",
+                link_id, host, len(result.files), result.bytes,
+            )
+            # Re-load post/artist/source attached to THIS session before handing
+            # them to the importer (the claim block's instances are detached).
+            post = session.get(Post, link.post_id)
+            artist = session.get(Artist, post.artist_id)
+            source = session.get(Source, post.source_id) if post.source_id else None
+            image_ids = _route_files(session, result.files, post, platform, artist, source)
+            link.status = "downloaded"
+            link.last_error = None
+            link.completed_at = datetime.now(UTC)
+            link.duration_seconds = time.monotonic() - started
+            session.commit()
+            log.info(
+                "extdl done: link=%s host=%s post=%s files=%d image(s)=%d dur=%.1fs",
+                link_id, host, post.id, len(result.files), len(image_ids),
+                link.duration_seconds,
+            )
+
+        # Thumbnails + ML for any newly-attached images (mirrors the download
+        # path). Lazy import to dodge a task-module import cycle.
+        if image_ids:
+            from .ml import tag_and_embed
+            from .thumbnail import generate_thumbnail
+            for img_id in image_ids:
+                generate_thumbnail.delay(img_id)
+                tag_and_embed.delay(img_id)
+        return {"link_id": link_id, "files": len(result.files), "images": len(image_ids)}
+    except Exception as exc:  # never leave a link stuck in 'downloading'
+        log.exception("external fetch task failed for link %s", link_id)
+        with SessionLocal() as session:
+            link = session.get(ExternalLink, link_id)
+            if link is not None and link.status == "downloading":
+                _record_failure(link, attempts, f"{type(exc).__name__}: {exc}")
+                session.commit()
+        raise
+    finally:
+        if lock is not None:
+            try:
+                lock.release()
+            except Exception:  # noqa: BLE001 — TTL may have already freed it
+                pass
+
+
+def _record_failure(link: ExternalLink, attempts: int, error: str) -> None:
+    nxt = attempts + 1
+    link.attempts = nxt
+    link.last_error = error[:1000]
+    link.status = "dead" if nxt >= DEAD_LETTER_THRESHOLD else "failed"
+    link.completed_at = datetime.now(UTC)
+    log.warning(
+        "extdl link %s -> %s (attempt %d/%d): %s",
+        link.id, link.status, nxt, DEAD_LETTER_THRESHOLD, error[:200],
+    )
+
+
+def _route_files(session, files, post, platform, artist, source) -> list[int]:
+    """Import each fetched file through the SAME pipeline extracted-zip members
+    use — importer.attach_in_place — so a downloaded archive is extracted to
+    ImageRecords and provenance-linked to the post (via the synthesized sidecar),
+    a non-art file is captured as a PostAttachment, and an art image is attached
+    in place. Returns new image ids so the caller enqueues thumbnail + ML
+    (tagging) for them, exactly as download_service does for downloaded media.
+
+    Result handling mirrors download_service._phase3_persist branch-for-branch
+    (duplicate cleanup, the unextracted-archive warning of #718) and logs every
+    decision — this path is new and the operator expects to iterate on it."""
+    settings = ImportSettings.load_sync(session)
+    importer = Importer(
+        session=session,
+        images_root=IMAGES_ROOT,
+        import_root=IMAGES_ROOT,
+        thumbnailer=Thumbnailer(images_root=IMAGES_ROOT),
+        settings=settings,
+    )
+    image_ids: list[int] = []
+    for f in files:
+        _write_link_sidecar(f, post, platform, artist)
+        result = importer.attach_in_place(f, artist=artist, source=source)
+        session.commit()
+        if result.status in ("imported", "superseded"):
+            ids = list(getattr(result, "member_image_ids", []) or [])
+            if result.image_id is not None and result.image_id not in ids:
+                ids.append(result.image_id)
+            image_ids.extend(ids)
+            log.info(
+                "extdl import: post=%s file=%s status=%s -> %d image(s) "
+                "(provenance-linked, queued for tagging)",
+                post.id, f.name, result.status, len(ids),
+            )
+        elif result.status == "attached":
+            # Non-art / archive captured as a PostAttachment (copied into the
+            # store) — drop the on-disk original. An archive captured WITHOUT
+            # extracting any image carries the reason (the recurring "zip but no
+            # images" symptom, #718) — surface it loudly.
+            if result.error:
+                log.warning(
+                    "extdl archive captured UNEXTRACTED: post=%s file=%s reason=%s",
+                    post.id, f.name, result.error,
+                )
+            else:
+                log.info("extdl attachment captured: post=%s file=%s", post.id, f.name)
+            f.unlink(missing_ok=True)
+        elif result.status == "skipped":
+            reason = result.skip_reason.value if result.skip_reason else None
+            log.info(
+                "extdl skipped: post=%s file=%s reason=%s", post.id, f.name, reason,
+            )
+            if reason in ("duplicate_hash", "duplicate_phash"):
+                f.unlink(missing_ok=True)  # canonical copy already in the library
+        elif result.status == "failed":
+            log.warning(
+                "extdl import FAILED: post=%s file=%s error=%s",
+                post.id, f.name, result.error,
+            )
+            f.unlink(missing_ok=True)
+        else:  # 'refreshed' or any future status — work happened, no new images
+            log.info(
+                "extdl import: post=%s file=%s status=%s", post.id, f.name, result.status,
+            )
+        # The synthesized sidecar has done its job — don't litter the tree.
+        Path(str(f) + ".json").unlink(missing_ok=True)
+    return image_ids
+
+
+@celery.task(name="backend.app.tasks.external.sweep_external_links")
+def sweep_external_links() -> dict:
+    """Enqueue a bounded batch of actionable links (pending, or failed below the
+    dead-letter threshold). Driven by the post-download hook and a beat tick."""
+    SessionLocal = _sync_session_factory()
+    with SessionLocal() as session:
+        ids = session.execute(
+            select(ExternalLink.id).where(
+                ExternalLink.status.in_(("pending", "failed")),
+                ExternalLink.attempts < DEAD_LETTER_THRESHOLD,
+            ).order_by(ExternalLink.id.asc()).limit(_SWEEP_BATCH)
+        ).scalars().all()
+    for link_id in ids:
+        fetch_external_link.delay(link_id)
+    return {"enqueued": len(ids)}
+
+
+@celery.task(name="backend.app.tasks.external.recover_external_links")
+def recover_external_links() -> dict:
+    """Recovery sweep (rule #89): reset dead links back to retryable so a stuck
+    host outage or a since-fixed bug gets another pass, then enqueue. Bounded."""
+    SessionLocal = _sync_session_factory()
+    with SessionLocal() as session:
+        session.execute(
+            update(ExternalLink).where(ExternalLink.status == "dead")
+            .values(status="failed", attempts=0, last_error=None)
+        )
+        session.commit()
+    return sweep_external_links()
+
+
+@celery.task(name="backend.app.tasks.external.prune_external_links")
+def prune_external_links() -> dict:
+    """Retention (rule #89): delete long-dead links so the ledger doesn't grow
+    unbounded. Downloaded links are kept (they're the record of what we have)."""
+    cutoff = datetime.now(UTC) - timedelta(days=_RETENTION_DAYS)
+    SessionLocal = _sync_session_factory()
+    with SessionLocal() as session:
+        res = session.execute(
+            delete(ExternalLink).where(
+                ExternalLink.status == "dead",
+                ExternalLink.created_at < cutoff,
+            )
+        )
+        session.commit()
+    return {"pruned": res.rowcount or 0}

backend/app/utils/html_sanitize.py

@@ -6,12 +6,27 @@ server-side to a tight allowlist. nh3 (Rust/ammonia bindings) is used;
 bleach is deprecated upstream and intentionally not used.
 """
 
+import re
+
 import nh3
 
+# A faithful-but-safe set: enough to reproduce the SEMANTIC look of a scraped
+# post body (headings, lists, quotes, code, inline images, links, line breaks +
+# inline emphasis) without layout-injection wrappers. `div`/`span` are
+# deliberately NOT allowed — their text is preserved by nh3 and they only carry
+# source-site layout/styling we don't want to import (a test pins this).
 ALLOWED_TAGS = {
-    "p", "a", "br", "em", "strong", "b", "i", "ul", "ol", "li", "blockquote",
+    "p", "a", "br", "em", "strong", "b", "i", "u", "s",
+    "ul", "ol", "li", "blockquote",
+    "h1", "h2", "h3", "h4", "h5", "h6", "hr",
+    "pre", "code", "img", "figure", "figcaption",
 }
-ALLOWED_ATTRIBUTES = {"a": {"href", "title"}}
+ALLOWED_ATTRIBUTES = {
+    "a": {"href", "title", "target"},
+    "img": {"src", "alt", "title", "width", "height"},
+}
+# http/https for hotlinked source images today; relative URLs (e.g. a local
+# `/api/...` src once inline images are captured) pass through nh3 untouched.
 ALLOWED_URL_SCHEMES = {"http", "https"}
 
 
@@ -37,3 +52,39 @@ def sanitize_post_html(raw: str | None) -> str | None:
         link_rel="noopener noreferrer",
         clean_content_tags={"script", "style"},
     )
+
+
+# nh3 emits attributes double-quoted, so a single `src="..."` capture is enough
+# to read/rewrite inline-image sources on already-sanitized markup (#830 Phase 2).
+_IMG_SRC_RE = re.compile(r'(<img\b[^>]*?\bsrc=")([^"]*)(")', re.IGNORECASE)
+
+
+def extract_img_srcs(html: str | None) -> list[str]:
+    """The raw `src` values of every <img> in `html`, in document order, deduped
+    (first kept). Returned EXACTLY as they appear (still entity-escaped) so a
+    caller can match them back for verbatim substitution."""
+    if not html:
+        return []
+    out: list[str] = []
+    seen: set[str] = set()
+    for _pre, src, _post in _IMG_SRC_RE.findall(html):
+        if src and src not in seen:
+            seen.add(src)
+            out.append(src)
+    return out
+
+
+def rewrite_img_srcs(html: str | None, replace: dict[str, str]) -> str | None:
+    """Swap each <img src> whose raw (as-in-HTML) value is a key in `replace`
+    with the mapped value; others are left untouched. Used to point a post body's
+    inline images at locally-served copies (#830 Phase 2). Operates only on src
+    values of already-sanitized markup, so the result stays within the allowlist.
+    Returns `html` unchanged when it or `replace` is empty."""
+    if not html or not replace:
+        return html
+
+    def _sub(m: re.Match) -> str:
+        pre, src, post = m.group(1), m.group(2), m.group(3)
+        return f"{pre}{replace.get(src, src)}{post}"
+
+    return _IMG_SRC_RE.sub(_sub, html)

backend/app/utils/paths.py

@@ -1,9 +1,28 @@
 """Filesystem path helpers — destination derivation, hash-suffixed names."""
 
+import re
 from pathlib import Path
 
 _MAX_EXT_LEN = 16
 
+# A Patreon/gallery CDN URL embeds a 32-char hex (MD5) path segment that is the
+# file's stable per-file identity — the same role gallery-dl's `_filehash`
+# plays. It is the join key between a post body `<img src=CDN>` and the local
+# copy we downloaded (extract_media dedups content vs gallery images by it), so
+# this ONE extractor must be used for both capture-time persistence and
+# render-time matching — they cannot be allowed to drift. Match the FIRST 32-hex
+# run anywhere in the URL (path or query); real CDN URLs carry exactly one.
+_FILEHASH_RE = re.compile(r"([0-9a-fA-F]{32})")
+
+
+def filehash_from_url(url: str | None) -> str | None:
+    """The 32-char hex (MD5) CDN identity segment of `url`, lowercased, or None
+    when the URL is empty / carries no such segment."""
+    if not url:
+        return None
+    match = _FILEHASH_RE.search(url)
+    return match.group(1).lower() if match else None
+
 
 def safe_ext(name: str | Path) -> str:
     """Conservatively extract a short, alphanumeric file extension.

backend/app/utils/sidecar.py

@@ -27,6 +27,11 @@ class SidecarData:
     description: str | None
     attachment_count: int | None
     post_date: datetime | None
+    # Per-FILE CDN/origin URL the media was downloaded from (#830 Phase 2). Only
+    # the per-media sidecar carries it; post-only sidecars leave it None. The
+    # importer persists it on the ImageRecord so the body's inline `<img src>`
+    # can be matched (by filehash) to the local copy at render time.
+    source_url: str | None
     raw: dict
 
 
@@ -172,6 +177,11 @@ def parse_sidecar(data: dict) -> SidecarData:
     else:
         post_url = _first_str(data, ("url", "post_url"))
 
+    # Per-file source URL (#830 Phase 2): the native downloader writes it into
+    # each media sidecar. `url` is the post permalink for Patreon (handled
+    # above), so source_url has its own dedicated key and never collides.
+    source_url = _first_str(data, ("source_url",))
+
     return SidecarData(
         platform=platform,
         external_post_id=external_post_id,
@@ -180,5 +190,6 @@ def parse_sidecar(data: dict) -> SidecarData:
         description=description,
         attachment_count=attachment_count,
         post_date=post_date,
+        source_url=source_url,
         raw=data,
     )

frontend/src/components/posts/PostCard.vue

@@ -63,8 +63,16 @@
           Post {{ post.external_post_id }}
         </h3>
 
+        <!-- Faithful (semantic) body render once expanded: backend-sanitized
+             HTML (headings, lists, links, inline images). Collapsed and the
+             no-detail fallback stay plain text. -->
+        <div
+          v-if="hasDescription && descExpanded && descHtml"
+          class="fc-post-card__desc fc-post-card__desc--html"
+          v-html="descHtml"
+        />
         <p
-          v-if="hasDescription" ref="descEl"
+          v-else-if="hasDescription" ref="descEl"
           class="fc-post-card__desc"
           :class="{ 'fc-post-card__desc--clamped': !descExpanded }"
         >{{ descText }}</p>
@@ -79,6 +87,25 @@
           @click="toggleDesc"
         >{{ descExpanded ? 'Show less' : 'Show more' }}</button>
 
+        <!-- Off-platform file-host links (mega/gdrive/…) found in the body.
+             Shown once expanded; clickable now, auto-downloaded by the worker
+             slice. -->
+        <div v-if="descExpanded && externalLinks.length" class="fc-post-card__links">
+          <div class="fc-post-card__links-label">External files</div>
+          <a
+            v-for="lnk in externalLinks" :key="lnk.id"
+            :href="lnk.url" target="_blank" rel="noopener noreferrer"
+            class="fc-post-card__link"
+          >
+            <span class="fc-post-card__link-host">{{ lnk.host }}</span>
+            <span class="fc-post-card__link-text">{{ lnk.label || lnk.url }}</span>
+            <span
+              v-if="lnk.status && lnk.status !== 'pending'"
+              class="fc-post-card__link-status" :data-status="lnk.status"
+            >{{ lnk.status }}</span>
+          </a>
+        </div>
+
         <div v-if="attachments.length" class="fc-post-card__atts">
           <a
             v-for="att in attachments" :key="att.id"
@@ -190,6 +217,11 @@ const descEl = ref(null)
 
 const hasDescription = computed(() => !!props.post.description_plain)
 const fullDescription = computed(() => detail.value?.description_full || null)
+// Backend-sanitized HTML body (detail-only). Drives the faithful render once
+// expanded; falls back to plain text when detail isn't loaded.
+const descHtml = computed(() => detail.value?.description_html || null)
+// Off-platform file-host links (detail-only), surfaced once expanded.
+const externalLinks = computed(() => detail.value?.external_links || [])
 const descText = computed(() =>
   descExpanded.value
     ? (fullDescription.value || props.post.description_plain)
@@ -219,7 +251,9 @@ onBeforeUnmount(() => { if (ro) { ro.disconnect(); ro = null } })
 
 async function toggleDesc () {
   if (!descExpanded.value) {
-    if (props.post.description_truncated && !fullDescription.value && !detail.value) {
+    // Fetch detail on first expand to get the sanitized HTML body (and the full
+    // plain text) — render the formatted body, not just the truncated preview.
+    if (!detail.value) {
       try {
         detail.value = await postsStore.getPostFull(props.post.id)
       } catch { /* render the truncated text rather than nothing */ }
@@ -369,6 +403,58 @@ function formatBytes (n) {
 @container (min-width: 800px) {
   .fc-post-card__desc--clamped { -webkit-line-clamp: 5; }
 }
+/* Faithful HTML body (expanded). Drop the plain-text pre-wrap and style the
+   semantic tags the backend sanitizer allows. :deep() pierces scoped CSS to
+   reach the v-html subtree. */
+.fc-post-card__desc--html { white-space: normal; }
+.fc-post-card__desc--html :deep(p) { margin: 0 0 0.6em; }
+.fc-post-card__desc--html :deep(p:last-child) { margin-bottom: 0; }
+.fc-post-card__desc--html :deep(h1),
+.fc-post-card__desc--html :deep(h2),
+.fc-post-card__desc--html :deep(h3),
+.fc-post-card__desc--html :deep(h4),
+.fc-post-card__desc--html :deep(h5),
+.fc-post-card__desc--html :deep(h6) {
+  margin: 0.8em 0 0.4em;
+  line-height: 1.25;
+  font-weight: 700;
+}
+.fc-post-card__desc--html :deep(h1) { font-size: 1.3rem; }
+.fc-post-card__desc--html :deep(h2) { font-size: 1.2rem; }
+.fc-post-card__desc--html :deep(h3) { font-size: 1.1rem; }
+.fc-post-card__desc--html :deep(h4),
+.fc-post-card__desc--html :deep(h5),
+.fc-post-card__desc--html :deep(h6) { font-size: 1rem; }
+.fc-post-card__desc--html :deep(a) {
+  color: rgb(var(--v-theme-accent));
+  text-decoration: underline;
+  word-break: break-word;
+}
+.fc-post-card__desc--html :deep(img) {
+  max-width: 100%;
+  height: auto;
+  border-radius: 6px;
+  margin: 0.4em 0;
+}
+.fc-post-card__desc--html :deep(ul),
+.fc-post-card__desc--html :deep(ol) { margin: 0 0 0.6em; padding-left: 1.4em; }
+.fc-post-card__desc--html :deep(li) { margin: 0.15em 0; }
+.fc-post-card__desc--html :deep(blockquote) {
+  margin: 0.6em 0;
+  padding-left: 0.8em;
+  border-left: 3px solid rgb(var(--v-theme-on-surface-variant));
+  color: rgb(var(--v-theme-on-surface-variant));
+}
+.fc-post-card__desc--html :deep(pre) {
+  background: rgba(var(--v-theme-on-surface), 0.06);
+  padding: 0.6em; border-radius: 6px; overflow-x: auto;
+}
+.fc-post-card__desc--html :deep(code) { font-family: monospace; font-size: 0.85em; }
+.fc-post-card__desc--html :deep(hr) {
+  border: 0; border-top: 1px solid rgb(var(--v-theme-on-surface-variant));
+  margin: 0.8em 0; opacity: 0.4;
+}
+.fc-post-card__desc--html :deep(figure) { margin: 0.4em 0; }
 .fc-post-card__desc--missing {
   font-style: italic;
   color: rgb(var(--v-theme-on-surface-variant));
@@ -404,4 +490,46 @@ function formatBytes (n) {
 }
 .fc-post-card__att-icon { color: rgb(var(--v-theme-on-surface-variant)); }
 .fc-post-card__att-size { color: rgb(var(--v-theme-on-surface-variant)); }
+
+.fc-post-card__links {
+  display: flex; flex-direction: column; gap: 6px;
+  margin-top: 12px;
+}
+.fc-post-card__links-label {
+  font-size: 0.75rem; font-weight: 600; text-transform: uppercase;
+  letter-spacing: 0.04em;
+  color: rgb(var(--v-theme-on-surface-variant));
+}
+.fc-post-card__link {
+  display: inline-flex; align-items: center; gap: 8px;
+  padding: 6px 12px;
+  border: 1px solid rgb(var(--v-theme-on-surface-variant));
+  border-radius: 8px;
+  color: rgb(var(--v-theme-on-surface));
+  text-decoration: none;
+  font-size: 0.85rem;
+}
+.fc-post-card__link:hover {
+  color: rgb(var(--v-theme-accent));
+  border-color: rgb(var(--v-theme-accent));
+}
+.fc-post-card__link-host {
+  flex: 0 0 auto;
+  font-weight: 700; text-transform: uppercase; font-size: 0.7rem;
+  letter-spacing: 0.04em;
+  padding: 2px 6px; border-radius: 4px;
+  background: rgba(var(--v-theme-accent), 0.16);
+  color: rgb(var(--v-theme-accent));
+}
+.fc-post-card__link-text {
+  overflow: hidden; text-overflow: ellipsis; white-space: nowrap;
+}
+.fc-post-card__link-status {
+  flex: 0 0 auto; margin-left: auto;
+  font-size: 0.7rem; text-transform: uppercase;
+  color: rgb(var(--v-theme-on-surface-variant));
+}
+.fc-post-card__link-status[data-status="downloaded"] {
+  color: rgb(var(--v-theme-accent));
+}
 </style>

frontend/src/components/subscriptions/SettingsTab.vue

@@ -52,6 +52,52 @@
       </v-card-text>
     </v-card>
 
+    <h3 class="text-h6 mb-3 mt-6">External file-host downloads</h3>
+    <v-card variant="outlined">
+      <v-card-text v-if="importStore.settings">
+        <div class="fc-help mb-2">
+          When a creator links files on these hosts in a post, Curator downloads
+          them into the post automatically (tagged + provenance-linked like any
+          other media). Turn a host off if it starts rate-limiting or banning.
+        </div>
+        <v-row>
+          <v-col cols="12" sm="4">
+            <v-switch
+              v-model="dl.extdl_mega_enabled" label="mega.nz"
+              density="compact" hide-details color="accent" @change="saveDownloader"
+            />
+          </v-col>
+          <v-col cols="12" sm="4">
+            <v-switch
+              v-model="dl.extdl_gdrive_enabled" label="Google Drive"
+              density="compact" hide-details color="accent" @change="saveDownloader"
+            />
+          </v-col>
+          <v-col cols="12" sm="4">
+            <v-switch
+              v-model="dl.extdl_pixeldrain_enabled" label="Pixeldrain"
+              density="compact" hide-details color="accent" @change="saveDownloader"
+            />
+          </v-col>
+          <v-col cols="12" sm="4">
+            <v-switch
+              v-model="dl.extdl_mediafire_enabled" label="MediaFire"
+              density="compact" hide-details color="accent" @change="saveDownloader"
+            />
+          </v-col>
+          <v-col cols="12" sm="4">
+            <v-switch
+              v-model="dl.extdl_dropbox_enabled" label="Dropbox"
+              density="compact" hide-details color="accent" @change="saveDownloader"
+            />
+          </v-col>
+        </v-row>
+      </v-card-text>
+      <v-card-text v-else>
+        <v-skeleton-loader type="paragraph" />
+      </v-card-text>
+    </v-card>
+
     <h3 class="text-h6 mb-3 mt-6">Schedule defaults</h3>
     <v-card variant="outlined">
       <v-card-text v-if="importStore.settings">
@@ -151,6 +197,11 @@ const dl = reactive({
   download_schedule_default_seconds: 28800,
   download_event_retention_days: 90,
   download_failure_warning_threshold: 5,
+  extdl_mega_enabled: true,
+  extdl_gdrive_enabled: true,
+  extdl_mediafire_enabled: true,
+  extdl_dropbox_enabled: true,
+  extdl_pixeldrain_enabled: true,
 })
 
 watch(() => importStore.settings, (s) => { if (s) Object.assign(dl, s) }, { immediate: true })

frontend/src/views/SeriesView.vue

@@ -1,35 +1,60 @@
 <template>
   <v-container fluid class="pt-2 pb-6">
-    <v-tabs v-model="tab" density="compact" class="mb-3">
-      <v-tab value="browse">Browse</v-tab>
-      <v-tab value="suggestions">
-        Suggestions
-        <v-badge
-          v-if="sug.suggestions.length" :content="sug.suggestions.length"
-          inline color="accent"
+    <!-- Tabs + the active tab's controls live in one sticky block pinned under
+         the 64px TopNav (operator-asked 2026-06-12), so the axis switcher and
+         search/sort stay reachable on a long grid. The controls can't sit
+         inside v-window (it clips sticky children), so they're hoisted here. -->
+    <div class="fc-series__head">
+      <v-tabs v-model="tab" density="compact">
+        <v-tab value="browse">Browse</v-tab>
+        <v-tab value="suggestions">
+          Suggestions
+          <v-badge
+            v-if="sug.suggestions.length" :content="sug.suggestions.length"
+            inline color="accent"
+          />
+        </v-tab>
+      </v-tabs>
+
+      <div v-if="tab === 'browse'" class="fc-series-browse__controls">
+        <v-text-field
+          v-model="query"
+          density="compact" variant="outlined" hide-details clearable
+          label="Search series" prepend-inner-icon="mdi-magnify"
+          style="max-width: 320px;"
         />
-      </v-tab>
-    </v-tabs>
+        <v-select
+          :model-value="store.sort"
+          :items="sortItems"
+          density="compact" variant="outlined" hide-details
+          label="Sort" style="max-width: 220px;"
+          @update:model-value="store.setSort($event)"
+        />
+      </div>
+
+      <div v-else class="fc-sug__controls">
+        <v-switch
+          :model-value="sug.enabled" color="accent" density="compact"
+          hide-details label="Matching on"
+          @update:model-value="sug.setEnabled($event)"
+        />
+        <v-text-field
+          :model-value="sug.threshold"
+          type="number" min="0" max="1" step="0.05"
+          density="compact" variant="outlined" hide-details
+          label="Threshold" style="max-width: 130px;"
+          @change="sug.setThreshold(Number($event.target.value))"
+        />
+        <v-spacer />
+        <v-btn
+          variant="tonal" prepend-icon="mdi-radar" @click="sug.rescan()"
+        >Rescan</v-btn>
+      </div>
+    </div>
 
     <v-window v-model="tab">
       <!-- ---- Browse ---- -->
       <v-window-item value="browse">
-        <div class="fc-series-browse__controls">
-          <v-text-field
-            v-model="query"
-            density="compact" variant="outlined" hide-details clearable
-            label="Search series" prepend-inner-icon="mdi-magnify"
-            style="max-width: 320px;"
-          />
-          <v-select
-            :model-value="store.sort"
-            :items="sortItems"
-            density="compact" variant="outlined" hide-details
-            label="Sort" style="max-width: 220px;"
-            @update:model-value="store.setSort($event)"
-          />
-        </div>
-
         <v-alert v-if="store.error" type="error" variant="tonal" closable class="my-4">
           Failed to load: {{ store.error }}
         </v-alert>
@@ -138,25 +163,6 @@
 
       <!-- ---- Suggestions ---- -->
       <v-window-item value="suggestions">
-        <div class="fc-sug__controls">
-          <v-switch
-            :model-value="sug.enabled" color="accent" density="compact"
-            hide-details label="Matching on"
-            @update:model-value="sug.setEnabled($event)"
-          />
-          <v-text-field
-            :model-value="sug.threshold"
-            type="number" min="0" max="1" step="0.05"
-            density="compact" variant="outlined" hide-details
-            label="Threshold" style="max-width: 130px;"
-            @change="sug.setThreshold(Number($event.target.value))"
-          />
-          <v-spacer />
-          <v-btn
-            variant="tonal" prepend-icon="mdi-radar" @click="sug.rescan()"
-          >Rescan</v-btn>
-        </div>
-
         <v-alert v-if="sug.error" type="error" variant="tonal" closable class="my-4">
           {{ sug.error }}
         </v-alert>
@@ -287,8 +293,17 @@ onMounted(() => {
 </script>
 
 <style scoped>
+/* Sticky header (tabs + active-tab controls) pinned under the 64px TopNav, so
+   content scrolls cleanly beneath it. Surface bg matches SettingsView. */
+.fc-series__head {
+  position: sticky;
+  top: 64px;
+  z-index: 4;
+  background: rgb(var(--v-theme-surface));
+  padding-bottom: 12px;
+}
 .fc-series-browse__controls {
-  display: flex; flex-wrap: wrap; gap: 12px; margin-bottom: 16px;
+  display: flex; flex-wrap: wrap; gap: 12px; padding-top: 12px;
 }
 .fc-series-browse__empty {
   padding: 48px; text-align: center;
@@ -351,7 +366,7 @@ onMounted(() => {
 
 /* Suggestions tab */
 .fc-sug__controls {
-  display: flex; align-items: center; gap: 16px; margin-bottom: 12px;
+  display: flex; align-items: center; gap: 16px; padding-top: 12px;
 }
 .fc-sug__list { display: flex; flex-direction: column; gap: 8px; }
 .fc-sug__row {

requirements.txt

@@ -38,3 +38,8 @@ nh3>=0.2,<0.3
 # Archive extraction (FC-2d-iii filesystem-import aid)
 rarfile>=4.2,<5
 py7zr>=1,<2
+
+# Google Drive fetcher for off-platform file-host links (external downloads,
+# #830). Handles Drive's confirm-token + virus-scan interstitial. mega.nz uses
+# the `megatools` binary instead (Debian apt pkg in the runtime image, not pip).
+gdown>=5,<6

tests/test_api_settings_downloader.py

@@ -44,3 +44,24 @@ async def test_patch_rejects_non_bool_validate(client):
         "/api/settings/import", json={"download_validate_files": "yes"}
     )
     assert resp.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_extdl_host_toggles_default_on_and_patch(client):
+    body = await (await client.get("/api/settings/import")).get_json()
+    assert body["extdl_mega_enabled"] is True
+    assert body["extdl_pixeldrain_enabled"] is True
+
+    resp = await client.patch(
+        "/api/settings/import", json={"extdl_mega_enabled": False}
+    )
+    assert resp.status_code == 200
+    assert (await resp.get_json())["extdl_mega_enabled"] is False
+
+
+@pytest.mark.asyncio
+async def test_extdl_toggle_rejects_non_bool(client):
+    resp = await client.patch(
+        "/api/settings/import", json={"extdl_gdrive_enabled": "nope"}
+    )
+    assert resp.status_code == 400

tests/test_external_fetch.py

@@ -0,0 +1,141 @@
+"""Unit tests for external_fetch — no network/gdown/MEGAcmd (seams stubbed)."""
+
+from pathlib import Path
+
+import pytest
+import requests
+
+import backend.app.services.external_fetch as ef
+from backend.app.services.external_fetch import fetch_external
+
+
+def _resp(*, status=200, headers=None, chunks=(b"hello",), text=""):
+    class _R:
+        def __init__(self):
+            self.status_code = status
+            self.headers = headers or {}
+            self.text = text
+
+        def iter_content(self, chunk_size=65536):
+            yield from chunks
+
+    return _R()
+
+
+def test_dropbox_forces_dl1_and_downloads(tmp_path, monkeypatch):
+    seen = {}
+
+    def fake_get(url, *, timeout, headers=None, stream=True):
+        seen["url"] = url
+        return _resp(headers={"Content-Disposition": 'attachment; filename="film.zip"'})
+
+    monkeypatch.setattr(ef, "_http_get", fake_get)
+    res = fetch_external(
+        "dropbox", "https://www.dropbox.com/s/abc/film.zip?dl=0", tmp_path
+    )
+    assert res.ok
+    assert "dl=1" in seen["url"]
+    assert res.files[0].name == "film.zip"
+    assert res.files[0].read_bytes() == b"hello"
+
+
+def test_pixeldrain_hits_api_endpoint(tmp_path, monkeypatch):
+    seen = {}
+
+    def fake_get(url, *, timeout, headers=None, stream=True):
+        seen["url"] = url
+        return _resp()
+
+    monkeypatch.setattr(ef, "_http_get", fake_get)
+    res = fetch_external("pixeldrain", "https://pixeldrain.com/u/ems4UomN", tmp_path)
+    assert res.ok
+    assert seen["url"] == "https://pixeldrain.com/api/file/ems4UomN"
+
+
+def test_mediafire_scrapes_then_downloads(tmp_path, monkeypatch):
+    calls = []
+
+    def fake_get(url, *, timeout, headers=None, stream=True):
+        calls.append(url)
+        if "mediafire.com/file" in url:
+            return _resp(
+                text='<a id="downloadButton" '
+                'href="https://download1234.mediafire.com/abc/film.zip">dl</a>'
+            )
+        return _resp(headers={"Content-Disposition": 'filename="film.zip"'})
+
+    monkeypatch.setattr(ef, "_http_get", fake_get)
+    res = fetch_external("mediafire", "https://www.mediafire.com/file/x/film.zip", tmp_path)
+    assert res.ok
+    assert res.files[0].name == "film.zip"
+    assert calls[1].startswith("https://download1234.mediafire.com/")
+
+
+def test_mediafire_missing_link_errors(tmp_path, monkeypatch):
+    monkeypatch.setattr(ef, "_http_get", lambda *a, **k: _resp(text="<html>none</html>"))
+    res = fetch_external("mediafire", "https://www.mediafire.com/file/x", tmp_path)
+    assert not res.ok
+    assert "not found" in res.error
+
+
+def test_gdrive_uses_gdown(tmp_path, monkeypatch):
+    def fake_gdown(url, out_dir):
+        p = Path(out_dir) / "drive.bin"
+        p.write_bytes(b"xyz")
+        return str(p)
+
+    monkeypatch.setattr(ef, "_gdown_download", fake_gdown)
+    res = fetch_external("gdrive", "https://drive.google.com/file/d/ID/view", tmp_path)
+    assert res.ok
+    assert res.files[0].name == "drive.bin"
+    assert res.bytes == 3
+
+
+def test_gdrive_no_file_errors(tmp_path, monkeypatch):
+    monkeypatch.setattr(ef, "_gdown_download", lambda url, out_dir: None)
+    res = fetch_external("gdrive", "https://drive.google.com/file/d/ID/view", tmp_path)
+    assert not res.ok
+    assert "gdown" in res.error
+
+
+def test_mega_collects_new_files(tmp_path, monkeypatch):
+    def fake_mega(url, out_dir, *, timeout):
+        (Path(out_dir) / "film.mp4").write_bytes(b"vid")
+
+    monkeypatch.setattr(ef, "_run_mega_get", fake_mega)
+    res = fetch_external("mega", "https://mega.nz/file/x#k", tmp_path)
+    assert res.ok
+    assert res.files[0].name == "film.mp4"
+
+
+def test_unsupported_host(tmp_path):
+    res = fetch_external("nope", "https://x", tmp_path)
+    assert not res.ok
+    assert "unsupported" in res.error
+
+
+def test_non_200_errors(tmp_path, monkeypatch):
+    monkeypatch.setattr(ef, "_http_get", lambda *a, **k: _resp(status=404))
+    res = fetch_external("pixeldrain", "https://pixeldrain.com/u/x", tmp_path)
+    assert not res.ok
+    assert "404" in res.error
+
+
+def test_transport_error_captured(tmp_path, monkeypatch):
+    def boom(*a, **k):
+        raise requests.ConnectionError("down")
+
+    monkeypatch.setattr(ef, "_http_get", boom)
+    res = fetch_external("dropbox", "https://www.dropbox.com/s/x?dl=0", tmp_path)
+    assert not res.ok
+    assert "transport" in res.error
+
+
+def test_should_stop_aborts_and_cleans_part(tmp_path, monkeypatch):
+    monkeypatch.setattr(ef, "_http_get", lambda *a, **k: _resp(chunks=(b"a", b"b")))
+    res = fetch_external(
+        "pixeldrain", "https://pixeldrain.com/u/x", tmp_path,
+        should_stop=lambda: True,
+    )
+    assert not res.ok
+    assert list(tmp_path.glob("*.part")) == []

tests/test_external_worker.py

@@ -0,0 +1,203 @@
+"""Integration tests for the external-link download worker (tasks/external)."""
+
+import io
+import zipfile
+
+import pytest
+from PIL import Image
+from sqlalchemy import func, select
+
+import backend.app.tasks.external as ext
+from backend.app.models import (
+    Artist,
+    ExternalLink,
+    ImageProvenance,
+    Post,
+    PostAttachment,
+    Source,
+)
+from backend.app.services.external_fetch import FetchResult
+
+pytestmark = pytest.mark.integration
+
+
+def _structured_jpeg_bytes():
+    # A non-uniform image so it survives the import filters (a solid color would
+    # phash-collapse / fail thresholds).
+    im = Image.new("RGB", (256, 256))
+    px = im.load()
+    for y in range(256):
+        for x in range(256):
+            px[x, y] = (x, y, (x + y) % 256)
+    buf = io.BytesIO()
+    im.save(buf, "JPEG")
+    return buf.getvalue()
+
+
+class _FakeLock:
+    def acquire(self, blocking=False):
+        return True
+
+    def release(self):
+        pass
+
+
+class _FakeRedis:
+    def lock(self, name, timeout=None, blocking=False):
+        return _FakeLock()
+
+
+def _seed(db_sync, *, host="pixeldrain", status="pending"):
+    # Key per host so a single test can seed several (artist name is UNIQUE).
+    artist = Artist(name=f"Ext {host}", slug=f"ext-{host}")
+    db_sync.add(artist)
+    db_sync.flush()
+    source = Source(
+        artist_id=artist.id, platform="patreon",
+        url=f"https://patreon.com/{host}", enabled=True, config_overrides={},
+    )
+    db_sync.add(source)
+    db_sync.flush()
+    post = Post(
+        source_id=source.id, artist_id=artist.id, external_post_id=f"EXT-{host}",
+        post_url=f"https://patreon.com/posts/{host}",
+    )
+    db_sync.add(post)
+    db_sync.flush()
+    link = ExternalLink(
+        post_id=post.id, artist_id=artist.id, host=host,
+        url=f"https://{host}.test/file", status=status,
+    )
+    db_sync.add(link)
+    db_sync.commit()
+    return post, link
+
+
+def test_fetch_external_link_downloads_and_attaches(db_sync, tmp_path, monkeypatch):
+    post, link = _seed(db_sync)
+    link_id = link.id
+
+    monkeypatch.setattr(ext, "IMAGES_ROOT", tmp_path)
+    monkeypatch.setattr(ext, "_redis", lambda: _FakeRedis())
+
+    def fake_fetch(host, url, dest_dir, *, timeout, should_stop=lambda: False):
+        dest_dir.mkdir(parents=True, exist_ok=True)
+        f = dest_dir / "film.bin"  # non-art → PostAttachment (no thumb/ML enqueue)
+        f.write_bytes(b"a film pack")
+        return FetchResult(files=[f], bytes=f.stat().st_size)
+
+    monkeypatch.setattr(ext, "fetch_external", fake_fetch)
+
+    out = ext.fetch_external_link(link_id)
+    assert out.get("files") == 1
+
+    db_sync.expire_all()
+    refreshed = db_sync.get(ExternalLink, link_id)
+    assert refreshed.status == "downloaded"
+    assert refreshed.completed_at is not None
+    # The file was captured as a PostAttachment linked to the SAME post.
+    atts = db_sync.execute(
+        select(PostAttachment).where(PostAttachment.post_id == post.id)
+    ).scalars().all()
+    assert len(atts) == 1
+
+
+def test_fetch_external_link_records_failure(db_sync, tmp_path, monkeypatch):
+    _, link = _seed(db_sync)
+    link_id = link.id
+    monkeypatch.setattr(ext, "IMAGES_ROOT", tmp_path)
+    monkeypatch.setattr(ext, "_redis", lambda: _FakeRedis())
+    monkeypatch.setattr(
+        ext, "fetch_external",
+        lambda *a, **k: FetchResult(error="host 503"),
+    )
+
+    ext.fetch_external_link(link_id)
+    db_sync.expire_all()
+    refreshed = db_sync.get(ExternalLink, link_id)
+    assert refreshed.status == "failed"
+    assert refreshed.attempts == 1
+    assert "503" in refreshed.last_error
+
+
+def test_fetch_external_link_dead_letters_at_threshold(db_sync, tmp_path, monkeypatch):
+    _, link = _seed(db_sync)
+    link.attempts = ext.DEAD_LETTER_THRESHOLD - 1
+    db_sync.commit()
+    link_id = link.id
+    monkeypatch.setattr(ext, "IMAGES_ROOT", tmp_path)
+    monkeypatch.setattr(ext, "_redis", lambda: _FakeRedis())
+    monkeypatch.setattr(ext, "fetch_external", lambda *a, **k: FetchResult(error="nope"))
+
+    ext.fetch_external_link(link_id)
+    db_sync.expire_all()
+    assert db_sync.get(ExternalLink, link_id).status == "dead"
+
+
+def test_fetch_external_link_skips_non_claimable(db_sync, tmp_path, monkeypatch):
+    _, link = _seed(db_sync, status="downloaded")
+    link_id = link.id
+    called = []
+    monkeypatch.setattr(ext, "_redis", lambda: _FakeRedis())
+    monkeypatch.setattr(
+        ext, "fetch_external",
+        lambda *a, **k: called.append(1) or FetchResult(),
+    )
+    out = ext.fetch_external_link(link_id)
+    assert out.get("skipped") == "not claimable"
+    assert called == []  # never fetched
+
+
+def test_sweep_enqueues_pending_and_retryable(db_sync, monkeypatch):
+    _, l1 = _seed(db_sync, host="pixeldrain")
+    _, l2 = _seed(db_sync, host="mega", status="failed")
+    # A dead one must NOT be swept.
+    _, l3 = _seed(db_sync, host="dropbox", status="dead")
+
+    enqueued = []
+    monkeypatch.setattr(ext.fetch_external_link, "delay", lambda lid: enqueued.append(lid))
+
+    out = ext.sweep_external_links()
+    assert out["enqueued"] == 2
+    assert set(enqueued) == {l1.id, l2.id}
+
+
+def test_downloaded_archive_gets_provenance_and_tagging(db_sync, tmp_path, monkeypatch):
+    """An archive downloaded by the worker is extracted, provenance-linked to the
+    post, AND queued for tagging + thumbnails — exactly like an extracted zip on
+    the normal download path."""
+    post, link = _seed(db_sync, host="mega")
+    link_id = link.id
+
+    monkeypatch.setattr(ext, "IMAGES_ROOT", tmp_path)
+    monkeypatch.setattr(ext, "_redis", lambda: _FakeRedis())
+
+    def fake_fetch(host, url, dest_dir, *, timeout, should_stop=lambda: False):
+        dest_dir.mkdir(parents=True, exist_ok=True)
+        zpath = dest_dir / "pack.zip"
+        with zipfile.ZipFile(zpath, "w") as z:
+            z.writestr("art.jpg", _structured_jpeg_bytes())
+        return FetchResult(files=[zpath], bytes=zpath.stat().st_size)
+
+    monkeypatch.setattr(ext, "fetch_external", fake_fetch)
+
+    tagged, thumbed = [], []
+    import backend.app.tasks.ml as ml_mod
+    import backend.app.tasks.thumbnail as thumb_mod
+    monkeypatch.setattr(ml_mod.tag_and_embed, "delay", lambda i: tagged.append(i))
+    monkeypatch.setattr(thumb_mod.generate_thumbnail, "delay", lambda i: thumbed.append(i))
+
+    out = ext.fetch_external_link(link_id)
+    assert out["images"] >= 1
+
+    db_sync.expire_all()
+    assert db_sync.get(ExternalLink, link_id).status == "downloaded"
+    # Extracted member is provenance-linked to the SAME post (as extracted zips).
+    prov = db_sync.execute(
+        select(ImageProvenance).where(ImageProvenance.post_id == post.id)
+    ).scalars().all()
+    assert len(prov) >= 1
+    member_ids = {p.image_record_id for p in prov}
+    # Tagging + thumbnails queued for exactly those member images.
+    assert set(tagged) == member_ids
+    assert set(thumbed) == member_ids

tests/test_html_sanitize.py

@@ -1,4 +1,8 @@
-from backend.app.utils.html_sanitize import sanitize_post_html
+from backend.app.utils.html_sanitize import (
+    extract_img_srcs,
+    rewrite_img_srcs,
+    sanitize_post_html,
+)
 
 
 def test_none_and_blank_return_none():
@@ -36,3 +40,56 @@ def test_javascript_href_dropped():
     out = sanitize_post_html('<a href="javascript:alert(1)">x</a>')
     assert "javascript:" not in out
     assert "x" in out  # text preserved, dangerous href removed
+
+
+def test_headings_lists_and_images_survive():
+    out = sanitize_post_html(
+        '<h2>Title</h2><p>x</p>'
+        '<img src="https://cdn.test/a.jpg" alt="art" width="200">'
+        '<ul><li>one</li></ul><blockquote>q</blockquote>'
+    )
+    assert "<h2>" in out
+    assert '<img' in out and 'src="https://cdn.test/a.jpg"' in out and 'alt="art"' in out
+    assert "<li>" in out and "<blockquote>" in out
+
+
+def test_image_javascript_src_dropped():
+    out = sanitize_post_html('<img src="javascript:alert(1)" alt="x">')
+    assert "javascript:" not in out
+
+
+# --- inline-image localization helpers (#830 Phase 2) ----------------------
+
+
+def test_extract_img_srcs_in_order_deduped():
+    html = (
+        '<p>a</p><img src="https://cdn.test/1.jpg">'
+        '<img alt="x" src="https://cdn.test/2.jpg" width="10">'
+        '<img src="https://cdn.test/1.jpg">'
+    )
+    assert extract_img_srcs(html) == [
+        "https://cdn.test/1.jpg",
+        "https://cdn.test/2.jpg",
+    ]
+
+
+def test_extract_img_srcs_empty():
+    assert extract_img_srcs(None) == []
+    assert extract_img_srcs("<p>no images</p>") == []
+
+
+def test_rewrite_img_srcs_swaps_only_mapped():
+    html = (
+        '<img src="https://cdn.test/1.jpg" alt="a">'
+        '<img src="https://cdn.test/2.jpg">'
+    )
+    out = rewrite_img_srcs(html, {"https://cdn.test/1.jpg": "/images/local/1.jpg"})
+    assert 'src="/images/local/1.jpg"' in out
+    assert 'alt="a"' in out  # other attributes preserved
+    assert 'src="https://cdn.test/2.jpg"' in out  # unmapped left alone
+
+
+def test_rewrite_img_srcs_noop_on_empty():
+    html = '<img src="https://cdn.test/1.jpg">'
+    assert rewrite_img_srcs(html, {}) == html
+    assert rewrite_img_srcs(None, {"a": "b"}) is None

tests/test_link_extract.py

@@ -0,0 +1,76 @@
+"""Unit tests for link_extract — pure, no DB/network."""
+
+from backend.app.services.link_extract import extract_external_links, host_for
+
+
+def test_host_classification_covers_all_and_rejects_others():
+    assert host_for("https://mega.nz/file/x#k") == "mega"
+    assert host_for("https://www.dropbox.com/s/x?dl=0") == "dropbox"
+    assert host_for("https://drive.google.com/file/d/ID/view") == "gdrive"
+    assert host_for("https://pixeldrain.com/u/abc") == "pixeldrain"
+    assert host_for("https://www.mediafire.com/file/x") == "mediafire"
+    assert host_for("https://example.com/x") is None
+
+
+def test_anchor_preserves_fragment_and_label():
+    html = ('<p>Watch: '
+            '<a href="https://mega.nz/file/Bux0FRrZ#b2Pn9841">Mega - Streamable</a></p>')
+    links = extract_external_links(html)
+    assert len(links) == 1
+    assert links[0].host == "mega"
+    # The #fragment (mega's decryption key) MUST survive.
+    assert links[0].url == "https://mega.nz/file/Bux0FRrZ#b2Pn9841"
+    assert links[0].label == "Mega - Streamable"
+
+
+def test_extracts_all_five_hosts():
+    html = (
+        '<a href="https://mega.nz/file/a#k">m</a>'
+        '<a href="https://drive.google.com/file/d/ID/view?usp=sharing">g</a>'
+        '<a href="https://www.mediafire.com/file/x/y">mf</a>'
+        '<a href="https://www.dropbox.com/s/x/y?dl=0">db</a>'
+        '<a href="https://pixeldrain.com/u/ems4UomN">px</a>'
+    )
+    hosts = {lnk.host for lnk in extract_external_links(html)}
+    assert hosts == {"mega", "gdrive", "mediafire", "dropbox", "pixeldrain"}
+
+
+def test_bare_url_in_text():
+    html = "Full film: https://pixeldrain.com/u/ems4UomN and more prose."
+    links = extract_external_links(html)
+    assert len(links) == 1
+    assert links[0].host == "pixeldrain"
+    assert links[0].url == "https://pixeldrain.com/u/ems4UomN"  # trailing '.' trimmed
+    assert links[0].label is None
+
+
+def test_unwraps_patreon_redirect():
+    html = ('<a href="https://www.patreon.com/redirect?'
+            'url=https%3A%2F%2Fmega.nz%2Ffile%2Fz%23key">x</a>')
+    links = extract_external_links(html)
+    assert len(links) == 1
+    assert links[0].host == "mega"
+    assert links[0].url == "https://mega.nz/file/z#key"
+
+
+def test_ignores_unsupported_and_dedups_anchor_label_wins():
+    html = (
+        '<a href="https://example.com/x">no</a>'
+        '<a href="https://mega.nz/file/dup#k">a</a>'
+        ' bare https://mega.nz/file/dup#k'
+    )
+    links = extract_external_links(html)
+    assert len(links) == 1            # example.com dropped; mega dup collapsed
+    assert links[0].label == "a"      # anchor (with label) wins over bare
+
+
+def test_amp_escaped_query_is_unescaped():
+    html = '<a href="https://drive.google.com/uc?id=ID&amp;export=download">g</a>'
+    links = extract_external_links(html)
+    assert links[0].url == "https://drive.google.com/uc?id=ID&export=download"
+
+
+def test_empty_inputs():
+    assert extract_external_links(None) == []
+    assert extract_external_links("") == []
+    assert extract_external_links("<p>no links here</p>") == []

tests/test_paths.py

@@ -3,12 +3,31 @@ from pathlib import Path
 from backend.app.utils.paths import (
     derive_subdir,
     derive_top_level_artist,
+    filehash_from_url,
     hash_suffixed_name,
 )
 
 IMPORT = Path("/import")
 
 
+def test_filehash_from_url_extracts_lowercased_md5():
+    url = "https://c10.patreonusercontent.com/4/patreon-media/p/post/1/AbC123DeF456789012345678901234EF/1/img.png?token=x"
+    assert filehash_from_url(url) == "abc123def456789012345678901234ef"
+
+
+def test_filehash_from_url_none_and_no_hash():
+    assert filehash_from_url(None) is None
+    assert filehash_from_url("") is None
+    assert filehash_from_url("https://example.test/a.png") is None
+
+
+def test_filehash_from_url_query_unescaping_irrelevant():
+    # The hash sits in the path before the query, so entity-escaped ampersands
+    # in the query never affect the match (render-time matches escaped srcs).
+    base = "https://cdn.test/p/0123456789abcdef0123456789ABCDEF/x.jpg"
+    assert filehash_from_url(base + "?a=1&b=2") == "0123456789abcdef0123456789abcdef"
+
+
 def test_derive_subdir_nested():
     assert derive_subdir(Path("/import/Alice/sub/x.png"), IMPORT) == "Alice/sub"
 

tests/test_patreon_client.py

@@ -9,6 +9,7 @@ import json
 from pathlib import Path
 
 import pytest
+import requests
 
 from backend.app.services import patreon_client as pc_mod
 from backend.app.services.patreon_client import (
@@ -352,3 +353,69 @@ def test_request_sleep_paces_before_fetch(monkeypatch):
     monkeypatch.setattr(pc_mod.time, "sleep", lambda s: slept.append(s))
     client._fetch("5555", None)
     assert slept == [1.5]
+
+
+# -- fetch_post_detail_content (best-effort full-body enrichment) ----------
+
+
+def test_fetch_post_detail_content_returns_body(monkeypatch):
+    payload = {"data": {"id": "42", "attributes": {"content": "<p>full body</p>"}}}
+    captured: dict = {}
+    client = PatreonClient(cookies_path=None)
+
+    def _get(url, params=None, timeout=None):
+        captured["url"] = url
+        captured["params"] = params
+        return _FakeResp(200, json_data=payload)
+
+    monkeypatch.setattr(client._session, "get", _get)
+    assert client.fetch_post_detail_content("42") == "<p>full body</p>"
+    assert captured["url"].endswith("/api/posts/42")
+    assert captured["params"]["fields[post]"] == "content"
+
+
+def test_fetch_post_detail_content_empty_body_is_none(monkeypatch):
+    payload = {"data": {"attributes": {"content": "   "}}}
+    client = _client_returning(monkeypatch, _FakeResp(200, json_data=payload))
+    assert client.fetch_post_detail_content("42") is None
+
+
+def test_fetch_post_detail_content_blank_id_skips_request(monkeypatch):
+    calls: list = []
+    client = PatreonClient(cookies_path=None)
+    monkeypatch.setattr(
+        client._session, "get",
+        lambda *a, **k: calls.append(1) or _FakeResp(200, json_data={}),
+    )
+    assert client.fetch_post_detail_content("") is None
+    assert calls == []
+
+
+def test_fetch_post_detail_content_swallows_http_and_transport_errors(monkeypatch):
+    # Non-200 → None (best-effort; a missing body must never fail the walk).
+    client = _client_returning(monkeypatch, _FakeResp(500))
+    assert client.fetch_post_detail_content("42") is None
+    # Non-JSON body (HTML challenge) → None, not a raise.
+    client = _client_returning(monkeypatch, _FakeResp(200, raise_json=True))
+    assert client.fetch_post_detail_content("42") is None
+    # Transport error → None.
+    client = PatreonClient(cookies_path=None)
+
+    def _boom(*a, **k):
+        raise requests.ConnectionError("boom")
+
+    monkeypatch.setattr(client._session, "get", _boom)
+    assert client.fetch_post_detail_content("42") is None
+
+
+# -- post_record_key (media-less post seen-ledger gate) --------------------
+
+
+def test_post_record_key_returns_synthetic_key_and_id():
+    assert PatreonClient.post_record_key({"id": "987"}) == ("post:987", "987")
+    assert PatreonClient.post_record_key({"id": 987}) == ("post:987", "987")
+
+
+def test_post_record_key_none_without_id():
+    assert PatreonClient.post_record_key({}) is None
+    assert PatreonClient.post_record_key({"id": None}) is None

tests/test_patreon_downloader.py

@@ -7,6 +7,7 @@ via monkeypatching `_run_ytdlp`. PURE module → no DB, so unmarked (fast lane).
 
 from __future__ import annotations
 
+import json
 from pathlib import Path
 
 import pytest
@@ -174,6 +175,9 @@ def test_sidecar_written_and_findable(tmp_path):
     assert data["id"] == "1001"
     assert data["title"] == "My Post Title"
     assert data["url"] == "https://www.patreon.com/posts/1001"
+    # #830 Phase 2: the per-media sidecar records THIS file's CDN URL so the
+    # importer can persist its filehash for inline-image localization.
+    assert data["source_url"] == "https://cdn.patreon.com/media1.png"
 
 
 def test_skip_seen(tmp_path):
@@ -536,3 +540,77 @@ def test_range_ignored_by_server_refetches_clean(tmp_path, monkeypatch):
     outcomes = dl.download_post(_post(), [_img("a.png")], "artist-x")
     assert outcomes[0].status == "downloaded"
     assert outcomes[0].path.read_bytes() == full  # clean, not full[:400] + full
+
+
+# -- content enrichment (adaptive detail-fetch of an empty feed body) -------
+
+
+def test_enriches_empty_content_via_fetcher(tmp_path):
+    """An empty feed body is backfilled from the detail-fetch seam — once per
+    post (memoized on the shared dict) — and lands in the importer sidecar."""
+    calls: list[str] = []
+
+    def _fetcher(post_id: str) -> str:
+        calls.append(post_id)
+        return "<p>full body</p>"
+
+    dl = PatreonDownloader(
+        images_root=tmp_path, cookies_path=None, validate=False,
+        session=_FakeSession(), content_fetcher=_fetcher,
+    )
+    post = _post()
+    post["attributes"]["content"] = ""  # feed returned an empty body
+    dl.download_post(post, [_img("a.png"), _img("b.png")], "artist-x")
+
+    post_dir = tmp_path / "artist-x" / "patreon" / pd_mod._post_dir_name(post)
+    sc = find_sidecar(post_dir / "01_a.png")
+    assert sc is not None
+    assert json.loads(sc.read_text())["content"] == "<p>full body</p>"
+    # Fetched ONCE for the post, not once per media item.
+    assert calls == ["1001"]
+
+
+def test_no_enrichment_when_feed_body_present(tmp_path):
+    """A non-empty feed body is trusted as-is; the detail endpoint isn't hit."""
+    calls: list[str] = []
+    dl = PatreonDownloader(
+        images_root=tmp_path, cookies_path=None, validate=False,
+        session=_FakeSession(),
+        content_fetcher=lambda pid: calls.append(pid) or "x",
+    )
+    dl.download_post(_post(), [_img("a.png")], "artist-x")  # _post body is non-empty
+    assert calls == []
+
+
+# -- write_post_record (media-less / text-only post capture) ----------------
+
+
+def test_write_post_record_writes_enriched_post_only_sidecar(tmp_path):
+    calls: list[str] = []
+
+    def _fetcher(post_id: str) -> str:
+        calls.append(post_id)
+        return "<p>text post body</p>"
+
+    dl = PatreonDownloader(
+        images_root=tmp_path, cookies_path=None, validate=False,
+        session=_FakeSession(), content_fetcher=_fetcher,
+    )
+    post = _post()
+    post["attributes"]["content"] = ""  # media-less post, empty feed body
+    sc = dl.write_post_record(post, "artist-x")
+
+    assert sc is not None
+    assert sc.name == "_post.json"  # can't collide with a media `NN_*.json`
+    data = json.loads(sc.read_text())
+    assert data["id"] == "1001"
+    assert data["content"] == "<p>text post body</p>"
+    assert calls == ["1001"]
+
+
+def test_write_post_record_none_without_post_id(tmp_path):
+    dl = PatreonDownloader(
+        images_root=tmp_path, cookies_path=None, validate=False,
+        session=_FakeSession(),
+    )
+    assert dl.write_post_record({"id": "", "attributes": {}}, "artist-x") is None

tests/test_patreon_ingester.py

@@ -68,6 +68,11 @@ class _FakeClient:
     def post_meta(self, post):
         return {"title": post.get("id"), "date": None}
 
+    @staticmethod
+    def post_record_key(post):
+        pid = str(post.get("id") or "")
+        return (f"post:{pid}", pid) if pid else None
+
 
 class _FakeDownloader:
     """Stub PatreonDownloader. Honors the injected is_seen (tier-1); media in
@@ -80,6 +85,7 @@ class _FakeDownloader:
         self.quarantine = set(quarantine or ())
         self.error = set(error or ())
         self.download_calls = 0
+        self.post_records = 0
 
     def download_post(self, post, media_items, artist_slug, *, is_seen,
                       should_stop=lambda: False):
@@ -106,6 +112,12 @@ class _FakeDownloader:
                 outcomes.append(MediaOutcome(media=m, status="downloaded", path=p, error=None))
         return outcomes
 
+    def write_post_record(self, post, artist_slug):
+        self.post_records += 1
+        p = self.tmp_path / f"{post.get('id')}__post.json"
+        p.write_text("{}")
+        return p
+
 
 @pytest.fixture
 async def source_id(db):
@@ -161,7 +173,10 @@ async def test_tick_downloads_unseen_and_marks_seen(source_id, sync_engine, tmp_
     # plan #704: structured run_stats carry the real counts.
     assert result.run_stats["downloaded_count"] == 2
     assert result.posts_processed == 1
-    assert _count_ledger(sync_engine, source_id) == 2
+    # 2 media keys + 1 synthetic post key (body/links recaptured per post).
+    assert _count_ledger(sync_engine, source_id) == 3
+    # The post body + links are captured for media posts too (rides the walk).
+    assert len(result.post_record_paths) == 1
 
 
 @pytest.mark.asyncio
@@ -184,8 +199,9 @@ async def test_quarantined_media_surfaced_in_result(source_id, sync_engine, tmp_
     assert result.run_stats["quarantined_count"] == 1
     assert result.run_stats["downloaded_count"] == 1
     assert len(result.written_paths) == 1        # quarantined NOT written
-    # Quarantined media is NOT marked seen (a fixed file may be re-fetched).
-    assert _count_ledger(sync_engine, source_id) == 1
+    # Quarantined media is NOT marked seen (a fixed file may be re-fetched);
+    # m1 + the synthetic post key (body/links captured per post) = 2.
+    assert _count_ledger(sync_engine, source_id) == 2
 
 
 @pytest.mark.asyncio
@@ -538,8 +554,31 @@ async def test_recovery_tier2_disk_still_skips(source_id, sync_engine, tmp_path)
     )
     assert result.files_downloaded == 0
     assert downloader.download_calls == 0
-    # Disk-skip still reconciles the ledger so a later tick skips at tier-1.
-    assert _count_ledger(sync_engine, source_id) == 1
+    # Disk-skip reconciles the media key + the synthetic post key (recovery
+    # recaptures the body/links per post) = 2.
+    assert _count_ledger(sync_engine, source_id) == 2
+
+
+@pytest.mark.asyncio
+async def test_backfill_recaptures_body_for_already_downloaded_post(
+    source_id, sync_engine, tmp_path,
+):
+    """Phase 5 guarantee: a post whose media is ALREADY on disk still gets its
+    body + external links recaptured on a re-walk. Re-downloading existing media
+    can never fill links the system never had — so the recapture rides the walk
+    itself, and a normal backfill is the backfill."""
+    m1 = _media("p1", 1)
+    client = _FakeClient([(None, [("p1", [m1])])])
+    downloader = _FakeDownloader(tmp_path, on_disk={_ledger_key(m1)})
+    ing = _ingester(sync_engine, tmp_path, client, downloader)
+
+    result = ing.run(
+        source_id=source_id, campaign_id="c1", artist_slug="ingest",
+        url="https://patreon.com/ingest", mode="backfill",
+    )
+    assert result.files_downloaded == 0          # media already on disk
+    assert len(result.post_record_paths) == 1    # body/links recaptured anyway
+    assert downloader.post_records == 1
 
 
 # --- dead-letter ledger (plan #705 #7) ------------------------------------
@@ -743,3 +782,55 @@ async def test_verify_patreon_credential_delegates_to_client(monkeypatch):
     ok, msg = await mod.verify_patreon_credential("https://patreon.com/x", None, {})
     assert ok is True
     assert msg == "ok:123"
+
+
+# --- media-less (text-only) post capture ----------------------------------
+
+
+@pytest.mark.asyncio
+async def test_tick_captures_media_less_post_once(source_id, sync_engine, tmp_path):
+    """A post with NO downloadable media still gets a post-only record, gated by
+    the seen-ledger (synthetic post key) so a second walk doesn't re-record it."""
+    client = _FakeClient([(None, [("ptext", [])])])
+    downloader = _FakeDownloader(tmp_path)
+    ing = _ingester(sync_engine, tmp_path, client, downloader)
+
+    result = ing.run(
+        source_id=source_id, campaign_id="c1", artist_slug="ingest",
+        url="https://patreon.com/ingest", mode="tick",
+    )
+    assert result.success is True
+    assert len(result.post_record_paths) == 1
+    assert downloader.post_records == 1
+    # The synthetic `post:ptext` key was marked seen (gates re-capture).
+    assert _count_ledger(sync_engine, source_id) == 1
+
+    # Second walk: already recorded → gated, no re-write, no new ledger row.
+    client2 = _FakeClient([(None, [("ptext", [])])])
+    downloader2 = _FakeDownloader(tmp_path)
+    ing2 = _ingester(sync_engine, tmp_path, client2, downloader2)
+    result2 = ing2.run(
+        source_id=source_id, campaign_id="c1", artist_slug="ingest",
+        url="https://patreon.com/ingest", mode="tick",
+    )
+    assert result2.post_record_paths == []
+    assert downloader2.post_records == 0
+    assert _count_ledger(sync_engine, source_id) == 1
+
+
+@pytest.mark.asyncio
+async def test_recovery_bypasses_gate_and_re_records(source_id, sync_engine, tmp_path):
+    """Recovery bypasses the seen-ledger, so a media-less post is re-recorded
+    (matches recovery semantics for media)."""
+    client = _FakeClient([(None, [("ptext", [])])])
+    ing = _ingester(sync_engine, tmp_path, client, _FakeDownloader(tmp_path))
+    ing.run(source_id=source_id, campaign_id="c1", artist_slug="ingest",
+            url="https://patreon.com/ingest", mode="tick")
+
+    client2 = _FakeClient([(None, [("ptext", [])])])
+    dl2 = _FakeDownloader(tmp_path)
+    ing2 = _ingester(sync_engine, tmp_path, client2, dl2)
+    result = ing2.run(source_id=source_id, campaign_id="c1", artist_slug="ingest",
+                      url="https://patreon.com/ingest", mode="recovery")
+    assert len(result.post_record_paths) == 1
+    assert dl2.post_records == 1

tests/test_post_feed_service.py

@@ -11,6 +11,7 @@ from sqlalchemy import select
 
 from backend.app.models import (
     Artist,
+    ExternalLink,
     ImageProvenance,
     ImageRecord,
     Post,
@@ -419,6 +420,101 @@ async def test_get_post_returns_full_description(db):
     assert len(item["description_full"]) > 280
 
 
+@pytest.mark.asyncio
+async def test_get_post_returns_sanitized_html_body(db):
+    artist = await _seed_artist(db, "alice-html")
+    src = await _seed_source(db, artist.id, "patreon", "https://p/alice-html")
+    p = await _seed_post(
+        db, src.id, external_id="HTMLBODY", post_date=datetime.now(UTC),
+        description='<h2>Heading</h2><p>body</p><script>alert(1)</script>',
+    )
+    await db.commit()
+
+    item = await PostFeedService(db).get_post(p.id)
+    # Detail carries the sanitized HTML body for faithful rendering; <script> is
+    # stripped, semantic tags survive.
+    assert item["description_html"] is not None
+    assert "<h2>" in item["description_html"]
+    assert "<script>" not in item["description_html"]
+
+
+@pytest.mark.asyncio
+async def test_get_post_localizes_inline_images(db):
+    """#830 Phase 2: a body <img src=CDN> whose filehash matches a stored image
+    of this artist is rewritten to the local /images path; an unmatched src is
+    left hotlinking."""
+    artist = await _seed_artist(db, "alice-inline")
+    src = await _seed_source(db, artist.id, "patreon", "https://p/alice-inline")
+    fh = "0123456789abcdef0123456789abcdef"
+    matched = f"https://cdn.test/p/{fh}/img.png"
+    unmatched = "https://cdn.test/p/ffffffffffffffffffffffffffffffff/other.png"
+    p = await _seed_post(
+        db, src.id, external_id="INLINE", post_date=datetime.now(UTC),
+        description=f'<p>x</p><img src="{matched}"><img src="{unmatched}">',
+    )
+    db.add(ImageRecord(
+        path="/images/alice-inline/patreon/post/01_img.png",
+        sha256=f"{p.id:064d}",
+        size_bytes=10, mime="image/png", width=10, height=10,
+        origin="downloaded", primary_post_id=p.id, artist_id=artist.id,
+        source_url=matched, source_filehash=fh,
+    ))
+    await db.commit()
+
+    item = await PostFeedService(db).get_post(p.id)
+    html = item["description_html"]
+    assert 'src="/images/alice-inline/patreon/post/01_img.png"' in html
+    assert matched not in html  # CDN src swapped out
+    assert unmatched in html    # no local copy → still hotlinked
+
+
+@pytest.mark.asyncio
+async def test_get_post_inline_image_scoped_to_artist(db):
+    """A matching filehash owned by a DIFFERENT artist must not leak into this
+    post's body — localization is artist-scoped."""
+    a1 = await _seed_artist(db, "owner-art")
+    a2 = await _seed_artist(db, "other-art")
+    src = await _seed_source(db, a1.id, "patreon", "https://p/owner-art")
+    fh = "abcabcabcabcabcabcabcabcabcabc12"
+    cdn = f"https://cdn.test/p/{fh}/x.png"
+    p = await _seed_post(
+        db, src.id, external_id="SCOPED", post_date=datetime.now(UTC),
+        description=f'<img src="{cdn}">',
+    )
+    # The only image with this filehash belongs to a DIFFERENT artist.
+    db.add(ImageRecord(
+        path="/images/other/x.png", sha256=f"{p.id:064d}",
+        size_bytes=10, mime="image/png", width=10, height=10,
+        origin="downloaded", artist_id=a2.id,
+        source_url=cdn, source_filehash=fh,
+    ))
+    await db.commit()
+
+    item = await PostFeedService(db).get_post(p.id)
+    assert cdn in item["description_html"]  # not localized (wrong artist)
+
+
+@pytest.mark.asyncio
+async def test_get_post_returns_external_links(db):
+    artist = await _seed_artist(db, "alice-extlinks")
+    src = await _seed_source(db, artist.id, "patreon", "https://p/alice-extlinks")
+    p = await _seed_post(
+        db, src.id, external_id="EXTLINKS", post_date=datetime.now(UTC),
+    )
+    db.add(ExternalLink(
+        post_id=p.id, artist_id=artist.id, host="mega",
+        url="https://mega.nz/file/x#k", label="Mega",
+    ))
+    await db.commit()
+
+    item = await PostFeedService(db).get_post(p.id)
+    assert len(item["external_links"]) == 1
+    link = item["external_links"][0]
+    assert link["host"] == "mega"
+    assert link["url"] == "https://mega.nz/file/x#k"
+    assert link["status"] == "pending"
+
+
 @pytest.mark.asyncio
 async def test_get_post_unknown_returns_none(db):
     item = await PostFeedService(db).get_post(99999)

tests/test_sidecar_import.py

@@ -8,6 +8,8 @@ from PIL import Image
 from sqlalchemy import func, select
 
 from backend.app.models import (
+    Artist,
+    ExternalLink,
     ImageProvenance,
     ImageRecord,
     ImportSettings,
@@ -99,6 +101,25 @@ def test_sidecar_creates_provenance(importer, import_layout):
     assert rec.effective_date == post.post_date
 
 
+def test_sidecar_source_url_persists_filehash(importer, import_layout):
+    """#830 Phase 2: a media sidecar's source_url lands on the ImageRecord as
+    source_url + the lowercased CDN filehash (the inline-image join key)."""
+    import_root, _ = import_layout
+    m = import_root / "Alice" / "b.jpg"
+    _split(m, "v")
+    _sidecar(m, {
+        "category": "patreon", "id": 901, "title": "Body post",
+        "source_url": "https://cdn.test/p/0123456789ABCDEF0123456789abcdef/b.jpg",
+    })
+    r = importer.import_one(m)
+    assert r.status == "imported"
+    rec = importer.session.get(ImageRecord, r.image_id)
+    assert rec.source_url == (
+        "https://cdn.test/p/0123456789ABCDEF0123456789abcdef/b.jpg"
+    )
+    assert rec.source_filehash == "0123456789abcdef0123456789abcdef"
+
+
 def test_reimport_same_post_idempotent(importer, import_layout):
     import_root, _ = import_layout
     # Threshold 0: only an exact phash match collapses. Orthogonal splits
@@ -164,8 +185,6 @@ def test_no_artist_anywhere_skips_provenance(importer, import_layout):
 
 
 def test_sidecar_artist_used_when_no_folder_artist(importer, import_layout):
-    from backend.app.models import Artist
-
     import_root, _ = import_layout
     m = import_root / "e.jpg"                 # root → no folder artist
     _split(m, "v")
@@ -183,3 +202,91 @@ def test_sidecar_artist_used_when_no_folder_artist(importer, import_layout):
     assert importer.session.execute(
         select(func.count()).select_from(Source)
     ).scalar_one() == 0
+
+
+def test_upsert_post_record_creates_media_less_post(importer, import_layout):
+    """A post-only sidecar (no media) upserts a Post WITH its body — text posts
+    are captured even when they have nothing to download."""
+    import_root, _ = import_layout
+    artist = Artist(name="Alice", slug="alice")
+    importer.session.add(artist)
+    importer.session.flush()
+    sc = import_root / "Alice" / "_post.json"
+    sc.parent.mkdir(parents=True, exist_ok=True)
+    sc.write_text(json.dumps({
+        "category": "patreon", "id": 777,
+        "url": "https://patreon.com/posts/777", "title": "Text only",
+        "content": "<p>links below: mega</p>",
+        "published_at": "2023-08-01T00:00:00Z",
+    }))
+    assert importer.upsert_post_record(sc, artist=artist) is True
+    post = importer.session.execute(select(Post)).scalar_one()
+    assert post.external_post_id == "777"
+    assert post.post_title == "Text only"
+    assert post.description == "<p>links below: mega</p>"
+    assert post.post_url == "https://patreon.com/posts/777"
+    assert post.post_date is not None
+
+
+def test_upsert_post_record_idempotent_no_double(importer, import_layout):
+    """Re-running upsert_post_record UPDATES the same Post (keyed on
+    external_post_id) — never doubles."""
+    import_root, _ = import_layout
+    artist = Artist(name="Bob", slug="bob")
+    importer.session.add(artist)
+    importer.session.flush()
+    sc = import_root / "Bob" / "_post.json"
+    sc.parent.mkdir(parents=True, exist_ok=True)
+    sc.write_text(json.dumps({
+        "category": "patreon", "id": 888, "title": "T",
+        "content": "<p>body</p>", "url": "https://patreon.com/posts/888",
+    }))
+    assert importer.upsert_post_record(sc, artist=artist) is True
+    assert importer.upsert_post_record(sc, artist=artist) is True
+    assert importer.session.execute(
+        select(func.count()).select_from(Post)
+    ).scalar_one() == 1
+
+
+def test_external_links_recorded_from_body(importer, import_layout):
+    """A mega/gdrive/etc. link in the body is recorded as an external_link row
+    (status pending) so it's never silently dropped."""
+    import_root, _ = import_layout
+    artist = Artist(name="Linker", slug="linker")
+    importer.session.add(artist)
+    importer.session.flush()
+    sc = import_root / "Linker" / "_post.json"
+    sc.parent.mkdir(parents=True, exist_ok=True)
+    sc.write_text(json.dumps({
+        "category": "patreon", "id": 999, "title": "Film",
+        "url": "https://patreon.com/posts/999",
+        "content": '<p>Get it: <a href="https://mega.nz/file/AbC#key123">Mega</a></p>',
+    }))
+    assert importer.upsert_post_record(sc, artist=artist) is True
+    importer.session.expire_all()  # re-read so the server_default status loads
+    links = importer.session.execute(select(ExternalLink)).scalars().all()
+    assert len(links) == 1
+    assert links[0].host == "mega"
+    assert links[0].url == "https://mega.nz/file/AbC#key123"
+    assert links[0].label == "Mega"
+    assert links[0].status == "pending"
+
+
+def test_external_links_not_duplicated_on_reimport(importer, import_layout):
+    """Re-importing the same body keeps ONE external_link row (insert-missing)."""
+    import_root, _ = import_layout
+    artist = Artist(name="Linker2", slug="linker2")
+    importer.session.add(artist)
+    importer.session.flush()
+    sc = import_root / "Linker2" / "_post.json"
+    sc.parent.mkdir(parents=True, exist_ok=True)
+    sc.write_text(json.dumps({
+        "category": "patreon", "id": 1000, "title": "Film",
+        "url": "https://patreon.com/posts/1000",
+        "content": '<a href="https://pixeldrain.com/u/xyz">px</a>',
+    }))
+    assert importer.upsert_post_record(sc, artist=artist) is True
+    assert importer.upsert_post_record(sc, artist=artist) is True
+    assert importer.session.execute(
+        select(func.count()).select_from(ExternalLink)
+    ).scalar_one() == 1

tests/test_sidecar_util.py

@@ -119,6 +119,15 @@ def test_parse_message_used_as_description_fallback():
     assert sd.description == "hello channel"
 
 
+def test_parse_source_url_present_and_absent():
+    """The per-media sidecar carries a `source_url` (#830 Phase 2); a post-only
+    sidecar (no media file) omits it → None."""
+    sd = parse_sidecar({"category": "patreon", "id": "1",
+                        "source_url": "https://cdn.test/abc.png"})
+    assert sd.source_url == "https://cdn.test/abc.png"
+    assert parse_sidecar({"category": "patreon", "id": "1"}).source_url is None
+
+
 def test_parse_date_epoch_and_unparseable_and_naive():
     assert parse_sidecar({"timestamp": 1690857602}).post_date.tzinfo is not None
     assert parse_sidecar({"date": "not-a-date"}).post_date is None