feat(patreon): capture media-less/text-only posts (post-only records)

Today the ingest core does `if not media: continue`, so a post with no
downloadable media (a pure-text post — which often holds the ONLY copy of an
external mega/gdrive/pixeldrain link) never upserts a Post. Now the native
ingester emits a post-only sidecar (`_post.json`) for every media-less post,
gated through the seen-ledger via a synthetic `post:<id>` key so the body is
detail-fetched + recorded ONCE (not re-walked every tick); recovery bypasses
the gate. Phase 3 imports these via Importer.upsert_post_record, keyed on
external_post_id so it UPDATES the same Post a media import would create —
never doubles, never clobbers a populated body with an empty one.

- gallery_dl.py: DownloadResult.post_record_paths (default []; gallery-dl path
  unaffected — all constructions are keyword).
- ingest_core.py: media-less branch (optional client/downloader seams via
  getattr; stub clients in tests skip it as before).
- patreon_client.py: post_record_key(post). patreon_downloader.py:
  write_post_record + _write_sidecar_data refactor (shared serializer).
- importer.py: upsert_post_record. download_service.py: phase-3 import loop.
- tests: client/downloader/ingester (gate + recovery)/importer (no-double).

Slice 0b of milestone #64. Refs FC #830.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

tests/test_patreon_client.py

@@ -406,3 +406,16 @@ def test_fetch_post_detail_content_swallows_http_and_transport_errors(monkeypatc
 
     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

@@ -577,3 +577,37 @@ def test_no_enrichment_when_feed_body_present(tmp_path):
     )
     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):
@@ -743,3 +755,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_sidecar_import.py

@@ -8,6 +8,7 @@ from PIL import Image
 from sqlalchemy import func, select
 
 from backend.app.models import (
+    Artist,
     ImageProvenance,
     ImageRecord,
     ImportSettings,
@@ -164,8 +165,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 +182,47 @@ 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