feat: media downloader + serve api

M2 Task 2.6 (media downloader) + Task 2.8 (media serve API).

Downloader (app/scraper/media.py):
- download_url(client, url): fetches bytes via the YunkeClient's underlying
  httpx client (carries session cookies; respx-mockable; follows redirects).
- download_message_media(client, msg, store, session): URL-driven fetcher that
  picks the best http(s) URL (file_url -> file_hd_url -> file_th_url), dedups
  by md5 (skips the network when a Media row with local file already exists),
  falls back to sha256(bytes) when no md5, stores via ObjectStore.put_by_hash,
  and upserts a Media row. Skips type-10 system messages (non-media per
  phase0-findings §2) and type-3 opaque-id / text (no http url). Resilient:
  download errors are logged and return None (no raise).
- media_type_from_message / guess_mime_type helpers.

Repo (app/storage/repos.py): upsert_media — idempotent Core upsert into Media by
media_hash. Volatile download artifacts (local_path/size_bytes/mime_type) use
COALESCE(incoming, existing) on conflict so a re-scrape that did not re-download
does not clobber previously stored values. Exported from app.storage.

Serve API (app/api/routers/media.py): GET /api/media/{media_hash} streams the
local file via FileResponse (handles HTTP Range / 206). 404 when the row is
absent, has no local file, or the file is missing on disk. Content-type from the
row's mime_type, falling back to ext-derived guess.

Tests (+19, suite 130 -> 149, all green; ruff clean):
- test_media_downloader.py: image/video download + upsert; md5 dedup skips
  second network hit; sha256 fallback when no md5; type-3/text/system -> None;
  download failure -> None (no raise); upsert_media idempotency +
  null-local_path preservation + 3x-same-key-one-row.
- test_api_media.py: 200 + content-type + bytes; 404 unknown hash; 404 missing
  file on disk; Range -> 206 partial; ext-derived content-type when mime NULL.

Co-Authored-By: Claude <noreply@anthropic.com>

yunke-server/app/api/main.py

@@ -7,6 +7,7 @@ Run locally with::
 Includes:
 * ``GET /api/health``  -> ``{"status": "ok"}`` liveness probe.
 * ``GET /api/customers`` (see :mod:`app.api.routers.customer`).
+* ``GET /api/media/{media_hash}`` (see :mod:`app.api.routers.media`).
 
 CORS is configured permissively (wildcard origin, no credentials) so the local
 frontend can call the API during development. Auth / finer CORS are LATER tasks.
@@ -17,7 +18,7 @@ from __future__ import annotations
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 
-from app.api.routers import customer
+from app.api.routers import customer, media
 
 app = FastAPI(title="yunke data mirror")
 
@@ -29,6 +30,7 @@ app.add_middleware(
 )
 
 app.include_router(customer.router)
+app.include_router(media.router)
 
 
 @app.get("/api/health")

yunke-server/app/api/routers/media.py

@@ -0,0 +1,54 @@
+"""GET /api/media/{media_hash} — stream a cached media blob (Task 2.8).
+
+Looks up the :class:`app.db.models.Media` row keyed by ``media_hash`` and
+streams the local file with the correct ``Content-Type``. HTTP Range requests
+are honored (Starlette's :class:`fastapi.responses.FileResponse` handles
+``Range`` / ``206 Partial Content`` natively). Returns ``404`` when the row is
+absent, has no local file recorded, or the underlying file is missing on disk.
+
+The endpoint reads the absolute ``local_path`` straight off the row (the
+downloader stores it absolute), so it does not need to know the
+:class:`ObjectStore` layout — only that the file exists where the row points.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Annotated
+
+import sqlalchemy as sa
+from fastapi import APIRouter, Depends, HTTPException
+from fastapi.responses import FileResponse
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.api.deps import get_session
+from app.db.models import Media
+from app.scraper.media import guess_mime_type
+
+router = APIRouter(prefix="/api", tags=["media"])
+
+SessionDep = Annotated[AsyncSession, Depends(get_session)]
+
+
+@router.get("/media/{media_hash}")
+async def get_media(media_hash: str, session: SessionDep) -> FileResponse:
+    """Stream the cached media blob for ``media_hash``."""
+    row = (
+        await session.execute(
+            sa.select(Media).where(Media.media_hash == media_hash)
+        )
+    ).scalar_one_or_none()
+    if row is None or not row.local_path:
+        raise HTTPException(status_code=404, detail="media not found")
+
+    path = Path(row.local_path)
+    if not path.is_file():
+        # Row exists but the blob was removed out-of-band — treat as not-found.
+        raise HTTPException(status_code=404, detail="media file missing")
+
+    media_type = (
+        row.mime_type
+        or guess_mime_type(path.suffix)
+        or "application/octet-stream"
+    )
+    return FileResponse(path, media_type=media_type, filename=path.name)

yunke-server/app/scraper/media.py

@@ -0,0 +1,222 @@
+"""Media downloader (M2, Task 2.6).
+
+Fetches WeChat / OSS media blobs referenced by message records into the local
+:class:`app.storage.object_store.ObjectStore` and records them in the
+:class:`app.db.models.Media` table, deduplicated by the source ``md5``.
+
+Phase-0 contract (see ``docs/superpowers/specs/phase0-findings.md`` §2/§3):
+
+* WeChat type ``2`` (image / emoji, ``.webp``) and ``4`` (video, ``.mp4``) carry
+  a DIRECT http(s) URL on ``file`` / ``fileTh`` / ``fileHd`` plus ``md5`` ->
+  downloadable, deduped by ``md5``.
+* WeChat type ``3`` (legacy image) carries an OPAQUE id on ``file`` (not a URL)
+  -> cannot be resolved yet; :func:`download_message_media` returns ``None``.
+* WeChat type ``10`` (system / link / card / invite) carries a
+  ``support.weixin.qq.com`` URL that is a documentation link, NOT a media blob
+  -> explicitly skipped before any network attempt.
+
+The downloader is URL-driven and generic: any http(s) media URL works (the
+recording URLs on ``yunke-pcfile`` are the same shape), so call recordings and
+voice-call recordings can be fetched by the same path once their owning records
+are wired in. That wiring is OUT of scope here (this module only consumes
+normalized message dicts); see ``app.scraper.wechat`` for those.
+
+Resilient by design: a download failure is logged and the function returns
+``None`` rather than raising. A retry queue is a LATER milestone.
+"""
+
+from __future__ import annotations
+
+import logging
+import mimetypes
+from pathlib import PurePosixPath
+from typing import Any
+from urllib.parse import urlparse
+
+import httpx
+import sqlalchemy as sa
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db.models import Media
+from app.session.yunke_client import YunkeClient
+from app.storage.object_store import ObjectStore
+from app.storage.repos import upsert_media
+
+__all__ = [
+    "download_message_media",
+    "download_url",
+    "guess_mime_type",
+    "media_type_from_message",
+]
+
+logger = logging.getLogger(__name__)
+
+# Normalized message ``type`` label -> :attr:`Media.media_type` value. Only the
+# Phase-0-confirmed image / video labels map today; voice / file codes
+# (``6`` / ``9`` / ...) remain ``"unknown"`` at the scraper layer and yield
+# ``None`` here until they are confirmed (phase0-findings.md §2).
+_MEDIA_TYPE_BY_LABEL: dict[str, str] = {
+    "image": "image",
+    "image_legacy": "image",
+    "video": "video",
+}
+
+# Extension -> MIME fallback map for the OSS / call-recording media types we
+# observe. Used when the stdlib ``mimetypes`` database does not recognize the
+# extension (e.g. ``.webp`` / ``.amr`` on some platforms) so content-type
+# derivation does not depend on the system MIME database.
+_EXT_MIME: dict[str, str] = {
+    ".webp": "image/webp",
+    ".jpg": "image/jpeg",
+    ".jpeg": "image/jpeg",
+    ".png": "image/png",
+    ".gif": "image/gif",
+    ".bmp": "image/bmp",
+    ".mp4": "video/mp4",
+    ".mp3": "audio/mpeg",
+    ".m4a": "audio/mp4",
+    ".wav": "audio/wav",
+    ".amr": "audio/amr",
+}
+
+
+def media_type_from_message(msg: dict[str, Any]) -> str | None:
+    """Map a normalized message dict to a :attr:`Media.media_type` label.
+
+    Returns ``"image"`` for ``image`` / ``image_legacy`` messages and ``"video"``
+    for ``video`` messages; ``None`` for everything else (text / system /
+    unknown / not-yet-confirmed voice-or-file codes).
+    """
+    return _MEDIA_TYPE_BY_LABEL.get(msg.get("type"))  # type: ignore[arg-type]
+
+
+def _is_http_url(value: Any) -> bool:
+    """True iff ``value`` is an ``http://`` / ``https://`` string."""
+    return isinstance(value, str) and value.startswith(("http://", "https://"))
+
+
+def _pick_media_url(msg: dict[str, Any]) -> str | None:
+    """Pick the best downloadable http(s) URL from a message dict.
+
+    Preference order: ``file_url`` (original) -> ``file_hd_url`` (high-def) ->
+    ``file_th_url`` (thumbnail). Returns ``None`` when no candidate is an
+    http(s) URL (covers type-``3`` opaque ids and text messages).
+    """
+    for key in ("file_url", "file_hd_url", "file_th_url"):
+        url = msg.get(key)
+        if _is_http_url(url):
+            return url  # type: ignore[return-value]
+    return None
+
+
+def _ext_from_url(url: str) -> str:
+    """Extract the lowercase extension (with leading ``.``) from a URL path.
+
+    Query / fragment are stripped first. Returns ``""`` when the path has no
+    suffix (e.g. an OSS callback URL without an extension).
+    """
+    suffix = PurePosixPath(urlparse(url).path).suffix
+    return suffix.lower()
+
+
+def guess_mime_type(ext: str) -> str | None:
+    """Guess a MIME type from a file extension.
+
+    The built-in :data:`_EXT_MIME` map is consulted first (covers ``.webp`` /
+    ``.mp4`` / ``.amr`` reliably regardless of platform), then the stdlib
+    ``mimetypes`` database. Returns ``None`` when nothing matches.
+    """
+    if not ext:
+        return None
+    key = ext.lower()
+    if key in _EXT_MIME:
+        return _EXT_MIME[key]
+    guessed, _ = mimetypes.guess_type(f"x{key}")
+    return guessed
+
+
+async def download_url(client: YunkeClient, url: str) -> bytes:
+    """Fetch the raw bytes at ``url`` via the client's underlying httpx session.
+
+    The :class:`YunkeClient` httpx client carries the session cookies (harmless
+    for the public OSS media URLs, useful for any same-origin authenticated
+    URLs) and is the transport ``respx`` mocks in the test suite, so media
+    downloads go through the same mocked transport as every other scraper
+    request. Redirects are followed (OSS / avatar URLs commonly redirect).
+    """
+    resp = await client.client.get(url, follow_redirects=True)
+    resp.raise_for_status()
+    return resp.content
+
+
+async def download_message_media(
+    client: YunkeClient,
+    msg: dict[str, Any],
+    store: ObjectStore,
+    session: AsyncSession,
+) -> str | None:
+    """Download one message's media blob and record it in the ``Media`` table.
+
+    Returns the dedup hash (the message ``md5`` when present, else the sha256 of
+    the downloaded bytes), or ``None`` when:
+
+    * the message is a type-``10`` system / link message (non-media; skipped
+      before any network attempt), or
+    * the message carries no downloadable http(s) URL (text, type-``3``
+      opaque-id images, or any message whose ``file``/``fileHd``/``fileTh`` are
+      absent or non-URL), or
+    * the download fails (the error is logged; no exception escapes).
+
+    Dedup: when the message carries an ``md5`` and a :class:`Media` row for that
+    hash already exists with a local file, the download is SKIPPED and the hash
+    is returned without touching the network.
+
+    The blob is written via :meth:`ObjectStore.put_by_hash` (sha256
+    content-addressed on disk), so identical bytes are stored once regardless of
+    how many messages reference them.
+    """
+    # Type-10 system / link messages carry a support URL that is NOT media
+    # (phase0-findings.md §2): skip them explicitly before URL selection so the
+    # downloader never fetches a documentation page as if it were a media blob.
+    if msg.get("type") == "system":
+        return None
+
+    url = _pick_media_url(msg)
+    if url is None:
+        return None
+
+    md5 = msg.get("md5")
+    if md5:
+        existing = (
+            await session.execute(
+                sa.select(Media.local_path).where(Media.media_hash == md5)
+            )
+        ).scalar_one_or_none()
+        if existing:
+            # Already downloaded (local file present); short-circuit the network.
+            return md5  # type: ignore[return-value]
+
+    try:
+        data = await download_url(client, url)
+    except (httpx.HTTPError, OSError) as exc:
+        logger.warning("media download failed for %s: %s", url, exc)
+        return None
+
+    ext = _ext_from_url(url)
+    sha256_hex, path = await store.put_by_hash(data, ext=ext)
+    media_hash: str = md5 if md5 else sha256_hex
+
+    await upsert_media(
+        session,
+        media_hash=media_hash,
+        channel=msg.get("channel") or "wechat",
+        media_type=media_type_from_message(msg),
+        original_url=msg.get("file_url"),
+        original_th_url=msg.get("file_th_url"),
+        original_hd_url=msg.get("file_hd_url"),
+        local_path=str(path),
+        size_bytes=len(data),
+        mime_type=guess_mime_type(ext),
+        raw_json=msg.get("raw_json"),
+    )
+    return media_hash

yunke-server/app/storage/__init__.py

@@ -19,6 +19,7 @@ from app.storage.object_store import ObjectStore
 from app.storage.repos import (
     upsert_customer,
     upsert_customers,
+    upsert_media,
     upsert_messages,
     upsert_staff_from_account,
     upsert_wechat_account,
@@ -28,6 +29,7 @@ __all__ = [
     "ObjectStore",
     "upsert_customer",
     "upsert_customers",
+    "upsert_media",
     "upsert_messages",
     "upsert_staff_from_account",
     "upsert_wechat_account",

yunke-server/app/storage/repos.py

@@ -58,6 +58,7 @@ from app.db.models import (
     Customer,
     Department,
     GroupMember,
+    Media,
     Message,
     PhoneCall,
     Staff,
@@ -69,6 +70,7 @@ __all__ = [
     "upsert_customer",
     "upsert_customers",
     "upsert_department",
+    "upsert_media",
     "upsert_messages",
     "upsert_staff",
     "upsert_staff_from_account",
@@ -474,3 +476,101 @@ async def upsert_voice_call(session: AsyncSession, rec: dict[str, Any]) -> None:
     per-row ``channel`` value disambiguates them.
     """
     await _upsert_one(session, PhoneCall, rec)
+
+
+# ---------------------------------------------------------------------------
+# Media upsert (M2, Task 2.6)
+# ---------------------------------------------------------------------------
+
+# Download artifacts that must NOT be clobbered with NULL when a later upsert
+# (e.g. a re-scrape that did not re-download) omits them. COALESCE'd against the
+# existing row so a genuine new value still wins.
+_MEDIA_PRESERVE_IF_NULL: frozenset[str] = frozenset(
+    {"local_path", "size_bytes", "mime_type"}
+)
+
+
+async def upsert_media(
+    session: AsyncSession,
+    *,
+    media_hash: str,
+    channel: str,
+    media_type: str | None = None,
+    original_url: str | None = None,
+    original_th_url: str | None = None,
+    original_hd_url: str | None = None,
+    local_path: str | None = None,
+    size_bytes: int | None = None,
+    mime_type: str | None = None,
+    duration: int | None = None,
+    raw_json: dict[str, Any] | None = None,
+) -> None:
+    """Idempotently upsert one :class:`Media` row (PK = ``media_hash``).
+
+    Same ``media_hash`` twice -> one row. Non-PK columns are refreshed from the
+    incoming record on conflict, EXCEPT the volatile download artifacts
+    (``local_path`` / ``size_bytes`` / ``mime_type``): those are written with
+    ``COALESCE(incoming, existing)`` so that a re-scrape which did NOT re-download
+    (and therefore passes ``None``) leaves the previously downloaded values
+    intact. A genuine new value (non-``None``) still overrides the old one.
+
+    The ``media_hash`` is the cross-message dedup key: the source ``md5`` when
+    the message carried one, else the sha256 of the downloaded bytes (see
+    :mod:`app.scraper.media`).
+    """
+    values: dict[str, Any] = {
+        "media_hash": media_hash,
+        "channel": channel,
+        "media_type": media_type,
+        "original_url": original_url,
+        "original_th_url": original_th_url,
+        "original_hd_url": original_hd_url,
+        "local_path": local_path,
+        "size_bytes": size_bytes,
+        "mime_type": mime_type,
+        "duration": duration,
+        "raw_json": raw_json,
+    }
+
+    dialect = await _dialect_name(session)
+    pk_cols = _pk_col_names(Media)
+    update_cols = [
+        c.name for c in Media.__table__.columns if c.name not in pk_cols
+    ]
+    media_cols = Media.__table__.c
+
+    def _build_set(other_cols: Any) -> dict[str, Any]:
+        """Map each updated column to its conflict-set expression.
+
+        Volatile download artifacts use ``COALESCE(incoming, existing)``; every
+        other column is refreshed directly from the incoming row.
+        """
+        set_: dict[str, Any] = {}
+        for name in update_cols:
+            if name in _MEDIA_PRESERVE_IF_NULL:
+                set_[name] = sa.func.coalesce(other_cols[name], media_cols[name])
+            else:
+                set_[name] = other_cols[name]
+        return set_
+
+    if dialect == "sqlite":
+        from sqlalchemy.dialects.sqlite import insert as sqlite_insert
+
+        bare = sqlite_insert(Media)
+        set_ = _build_set(bare.excluded)
+        stmt = bare.values(**values).on_conflict_do_update(
+            index_elements=pk_cols, set_=set_
+        )
+    elif dialect == "mysql":
+        from sqlalchemy.dialects.mysql import insert as mysql_insert
+
+        bare = mysql_insert(Media)
+        set_ = _build_set(bare.inserted)
+        stmt = bare.values(**values).on_duplicate_key_update(**set_)
+    else:
+        raise ValueError(
+            f"Unsupported dialect for upsert_media: {dialect!r}. "
+            "Only 'sqlite' (tests) and 'mysql' (prod) are supported."
+        )
+
+    await session.execute(stmt)

yunke-server/tests/test_api_media.py

@@ -0,0 +1,147 @@
+"""Tests for GET /api/media/{media_hash} — media serve API (Task 2.8).
+
+Drives the FastAPI app over an in-process httpx transport (``ASGITransport``) —
+no real network socket. The ``get_session`` dependency is overridden to an
+isolated in-memory async SQLite session (``StaticPool`` so every request sees
+the same seeded schema/rows). Media blobs are written to a real ``tmp_path``
+file and referenced by absolute path from the seeded ``Media`` row.
+
+Coverage:
+* ``GET /api/media/{hash}`` for a seeded row -> 200, correct ``Content-Type``,
+  and the exact bytes streamed back.
+* Unknown hash -> 404.
+* Row exists but the underlying file is missing on disk -> 404.
+* HTTP Range request -> ``206`` with the requested partial bytes (FileResponse
+  handles Range), proving the endpoint supports range / partial-content.
+* A row whose ``mime_type`` is NULL falls back to an ext-derived content type.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+from sqlalchemy.pool import StaticPool
+
+from app.api import deps
+from app.api.main import app
+from app.db.base import Base
+from app.db.models import Media
+
+PNG_BYTES = b"\x89PNG\r\n\x1a\n fake image body"
+
+
+@pytest.fixture
+async def client(tmp_path: Path):
+    """Yield an ``AsyncClient`` against ``app`` with a seeded in-memory DB.
+
+    Seeds one ``Media`` row (``abc123``) pointing at a real PNG-bytes file
+    under ``tmp_path``, plus a second row (``nopng``) whose ``local_path`` is
+    deliberately missing on disk (to exercise the file-missing 404 path), plus
+    a third (``nomime``) whose ``mime_type`` is NULL (to exercise the
+    ext-derived content-type fallback).
+    """
+    engine = create_async_engine(
+        "sqlite+aiosqlite:///:memory:",
+        poolclass=StaticPool,
+        connect_args={"check_same_thread": False},
+    )
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+
+    media_file = tmp_path / "abc123.png"
+    media_file.write_bytes(PNG_BYTES)
+
+    # A second file with no mime_type recorded, to test ext-derived fallback.
+    fallback_file = tmp_path / "fallback.webp"
+    fallback_file.write_bytes(b"WEBPBYTES")
+
+    SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
+    async with SessionLocal() as session:
+        session.add_all(
+            [
+                Media(
+                    media_hash="abc123",
+                    channel="wechat",
+                    media_type="image",
+                    original_url="https://wx.oss.yunkecn.com/x/abc.png",
+                    local_path=str(media_file),
+                    size_bytes=len(PNG_BYTES),
+                    mime_type="image/png",
+                ),
+                Media(
+                    media_hash="nopng",
+                    channel="wechat",
+                    media_type="image",
+                    local_path=str(tmp_path / "does-not-exist.png"),
+                    size_bytes=1,
+                    mime_type="image/png",
+                ),
+                Media(
+                    media_hash="nomime",
+                    channel="wechat",
+                    media_type="image",
+                    local_path=str(fallback_file),
+                    size_bytes=len(b"WEBPBYTES"),
+                    mime_type=None,
+                ),
+            ]
+        )
+        await session.commit()
+
+    async def _override_session():
+        async with SessionLocal() as s:
+            yield s
+
+    app.dependency_overrides[deps.get_session] = _override_session
+    try:
+        async with AsyncClient(
+            transport=ASGITransport(app=app), base_url="http://test"
+        ) as ac:
+            yield ac
+    finally:
+        app.dependency_overrides.clear()
+        await engine.dispose()
+
+
+async def test_get_media_returns_bytes_with_content_type(client: AsyncClient):
+    """Seeded hash -> 200, image/png content-type, exact bytes streamed back."""
+    r = await client.get("/api/media/abc123")
+    assert r.status_code == 200
+    assert r.headers["content-type"].startswith("image/png")
+    assert r.content == PNG_BYTES
+
+
+async def test_get_media_unknown_hash_returns_404(client: AsyncClient):
+    """An unknown media_hash -> 404."""
+    r = await client.get("/api/media/does-not-exist")
+    assert r.status_code == 404
+
+
+async def test_get_media_missing_file_on_disk_returns_404(client: AsyncClient):
+    """Row present but file absent from disk -> 404 (defensive)."""
+    r = await client.get("/api/media/nopng")
+    assert r.status_code == 404
+
+
+async def test_get_media_supports_http_range(client: AsyncClient):
+    """Range request -> 206 Partial Content with the requested byte slice.
+
+    Starlette's ``FileResponse`` honors ``Range``; this pins the behavior so a
+    future swap to a hand-rolled StreamingResponse would have to keep it.
+    """
+    r = await client.get("/api/media/abc123", headers={"Range": "bytes=0-3"})
+    assert r.status_code == 206
+    assert r.content == PNG_BYTES[:4]
+    assert r.headers["content-length"] == "4"
+    assert r.headers["content-range"].startswith("bytes 0-3/")
+
+
+async def test_get_media_falls_back_to_ext_when_mime_null(client: AsyncClient):
+    """mime_type NULL on the row -> content-type guessed from the .webp ext."""
+    r = await client.get("/api/media/nomime")
+    assert r.status_code == 200
+    assert r.headers["content-type"].startswith("image/webp")
+    assert r.content == b"WEBPBYTES"

yunke-server/tests/test_media_downloader.py

@@ -0,0 +1,523 @@
+"""Tests for the media downloader (M2, Task 2.6) and ``upsert_media`` repo.
+
+Covers:
+
+* :func:`app.scraper.media.download_message_media` against ``respx``-mocked OSS
+  URLs (no real network), a tmp_path :class:`ObjectStore`, and an in-memory
+  SQLite session.
+* :func:`app.storage.repos.upsert_media` idempotency, including the rule that a
+  NULL ``local_path``/``size_bytes``/``mime_type`` on a later upsert must NOT
+  clobber the values written by an earlier (downloaded) row.
+
+Phase-0 contract (see ``docs/superpowers/specs/phase0-findings.md`` §2/§3):
+
+* WeChat type ``2`` (image/emoji, ``.webp``) and ``4`` (video, ``.mp4``) carry a
+  DIRECT http(s) URL on ``file``/``fileTh``/``fileHd`` plus ``md5`` -> the
+  downloader fetches + dedups by ``md5``.
+* WeChat type ``3`` (legacy image) carries an OPAQUE id on ``file`` (not a URL)
+  -> the downloader returns ``None`` (cannot resolve yet).
+* Text / type ``10`` system messages carry no downloadable http(s) URL ->
+  ``None``.
+* A download failure is swallowed (returns ``None``); no exception escapes.
+"""
+
+from __future__ import annotations
+
+import hashlib
+
+import httpx
+import respx
+import sqlalchemy as sa
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+from app.db.base import Base
+from app.db.models import Media
+from app.scraper.media import (
+    download_message_media,
+    download_url,
+    media_type_from_message,
+)
+from app.session.yunke_client import YunkeClient
+from app.storage.object_store import ObjectStore
+from app.storage.repos import upsert_media
+
+BASE = "https://crm2.yunkecn.com"
+# A realistic OSS media URL (phase0-findings.md §3): public GET, keyed by md5.
+OSS_WEBP_URL = (
+    "https://wx.oss.yunkecn.com/wechatMsgFile/8m0atd/u1/ossCallback/ea/8f/74/"
+    "ea8f74e7797e63505731e8c51222c554.webp"
+)
+OSS_MP4_URL = (
+    "https://wx.oss.yunkecn.com/wechatMsgFile/8m0atd/u1/ossCallback/01/23/45/"
+    "0123456789abcdef0123456789abcdef.mp4"
+)
+
+USERID = "a" * 32
+COMPANY = "8m0atd"
+
+
+# ---------------------------------------------------------------------------
+# Shared helpers
+# ---------------------------------------------------------------------------
+
+
+def _primed_client() -> YunkeClient:
+    """A YunkeClient already in a logged-in state (login flow skipped)."""
+    return YunkeClient(base_url=BASE, userid=USERID, company=COMPANY)
+
+
+async def _make_engine():
+    """Create an in-memory async SQLite engine with all tables created."""
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    return engine
+
+
+# ---------------------------------------------------------------------------
+# media_type_from_message
+# ---------------------------------------------------------------------------
+
+
+def test_media_type_from_message_maps_known_labels():
+    """image / image_legacy -> image; video -> video; others -> None."""
+    assert media_type_from_message({"type": "image"}) == "image"
+    assert media_type_from_message({"type": "image_legacy"}) == "image"
+    assert media_type_from_message({"type": "video"}) == "video"
+    assert media_type_from_message({"type": "text"}) is None
+    assert media_type_from_message({"type": "system"}) is None
+    assert media_type_from_message({"type": "unknown"}) is None
+    assert media_type_from_message({}) is None
+
+
+# ---------------------------------------------------------------------------
+# download_url
+# ---------------------------------------------------------------------------
+
+
+@respx.mock
+async def test_download_url_returns_bytes_via_client_httpx():
+    """download_url fetches the raw bytes through the YunkeClient's httpx."""
+    respx.get(OSS_WEBP_URL).mock(
+        return_value=httpx.Response(200, content=b"raw-bytes")
+    )
+    client = _primed_client()
+    try:
+        data = await download_url(client, OSS_WEBP_URL)
+        assert data == b"raw-bytes"
+    finally:
+        await client.aclose()
+
+
+@respx.mock
+async def test_download_url_raises_on_http_error():
+    """A non-2xx response raises (caller ``download_message_media`` swallows)."""
+    respx.get(OSS_WEBP_URL).mock(return_value=httpx.Response(404))
+    client = _primed_client()
+    try:
+        try:
+            await download_url(client, OSS_WEBP_URL)
+        except httpx.HTTPStatusError:
+            pass
+        else:  # pragma: no cover - belt and braces
+            raise AssertionError("expected HTTPStatusError on 404")
+    finally:
+        await client.aclose()
+
+
+# ---------------------------------------------------------------------------
+# download_message_media
+# ---------------------------------------------------------------------------
+
+
+@respx.mock
+async def test_download_message_media_image_downloads_stores_upserts(tmp_path):
+    """Type-2 image (file_url + md5): downloads bytes, stores file, upserts Media.
+
+    Returns the md5 hash; file is written under tmp_path at the sha256-content
+    path; the Media row carries media_type=image, mime=image/webp, the OSS url.
+    """
+    store = ObjectStore(tmp_path)
+    engine = await _make_engine()
+    SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
+    msg = {
+        "msg_id": "m1",
+        "channel": "wechat",
+        "type": "image",
+        "type_raw": 2,
+        "file_url": OSS_WEBP_URL,
+        "md5": "ea8f74e7797e63505731e8c51222c554",
+        "raw_json": {"file": OSS_WEBP_URL, "messageType": 2},
+    }
+    blob = b"webp-bytes"
+    respx.get(OSS_WEBP_URL).mock(return_value=httpx.Response(200, content=blob))
+
+    client = _primed_client()
+    try:
+        async with SessionLocal() as session:
+            media_hash = await download_message_media(client, msg, store, session)
+            await session.commit()
+        assert media_hash == "ea8f74e7797e63505731e8c51222c554"
+
+        # File landed on disk at the sha256-content-addressed path with .webp ext.
+        sha = hashlib.sha256(blob).hexdigest()
+        expected_path = tmp_path / sha[:2] / f"{sha}.webp"
+        assert expected_path.is_file()
+        assert expected_path.read_bytes() == blob
+
+        # Media row written with the right shape.
+        async with SessionLocal() as session:
+            row = (
+                await session.execute(
+                    sa.select(Media).where(
+                        Media.media_hash == "ea8f74e7797e63505731e8c51222c554"
+                    )
+                )
+            ).scalar_one()
+            assert row.channel == "wechat"
+            assert row.media_type == "image"
+            assert row.original_url == OSS_WEBP_URL
+            assert row.local_path == str(expected_path)
+            assert row.size_bytes == len(blob)
+            assert row.mime_type == "image/webp"
+    finally:
+        await client.aclose()
+        await engine.dispose()
+
+
+@respx.mock
+async def test_download_message_media_dedup_skips_second_download(tmp_path):
+    """Same md5 twice: second call SKIPS the network and returns the same hash."""
+    store = ObjectStore(tmp_path)
+    engine = await _make_engine()
+    SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
+    msg = {
+        "msg_id": "m1",
+        "channel": "wechat",
+        "type": "image",
+        "type_raw": 2,
+        "file_url": OSS_WEBP_URL,
+        "md5": "ea8f74e7797e63505731e8c51222c554",
+    }
+    route = respx.get(OSS_WEBP_URL).mock(
+        return_value=httpx.Response(200, content=b"webp-bytes")
+    )
+
+    client = _primed_client()
+    try:
+        async with SessionLocal() as session:
+            h1 = await download_message_media(client, msg, store, session)
+            await session.commit()
+        async with SessionLocal() as session:
+            h2 = await download_message_media(client, msg, store, session)
+            await session.commit()
+
+        assert h1 == h2 == "ea8f74e7797e63505731e8c51222c554"
+        # The OSS URL was hit exactly ONCE: the second call short-circuited.
+        assert route.call_count == 1
+    finally:
+        await client.aclose()
+        await engine.dispose()
+
+
+@respx.mock
+async def test_download_message_media_video_mp4(tmp_path):
+    """Type-4 video (file_url + md5): downloads, media_type=video, mime=video/mp4."""
+    store = ObjectStore(tmp_path)
+    engine = await _make_engine()
+    SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
+    msg = {
+        "msg_id": "mv",
+        "channel": "wechat",
+        "type": "video",
+        "type_raw": 4,
+        "file_url": OSS_MP4_URL,
+        "md5": "0123456789abcdef0123456789abcdef",
+    }
+    blob = b"\x00\x00\x00\x18ftypmp42"
+    respx.get(OSS_MP4_URL).mock(return_value=httpx.Response(200, content=blob))
+
+    client = _primed_client()
+    try:
+        async with SessionLocal() as session:
+            h = await download_message_media(client, msg, store, session)
+            await session.commit()
+        assert h == "0123456789abcdef0123456789abcdef"
+
+        async with SessionLocal() as session:
+            row = (
+                await session.execute(
+                    sa.select(Media).where(
+                        Media.media_hash == "0123456789abcdef0123456789abcdef"
+                    )
+                )
+            ).scalar_one()
+            assert row.media_type == "video"
+            assert row.mime_type == "video/mp4"
+            assert row.size_bytes == len(blob)
+            assert row.local_path.endswith(".mp4")
+    finally:
+        await client.aclose()
+        await engine.dispose()
+
+
+@respx.mock
+async def test_download_message_media_falls_back_to_sha256_when_no_md5(tmp_path):
+    """No md5 on the message: dedup key is sha256(bytes); still stored + upserted."""
+    store = ObjectStore(tmp_path)
+    engine = await _make_engine()
+    SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
+    msg = {
+        "msg_id": "m2",
+        "channel": "wechat",
+        "type": "image",
+        "type_raw": 2,
+        "file_url": OSS_WEBP_URL,
+        # no md5
+    }
+    blob = b"no-md5-bytes"
+    respx.get(OSS_WEBP_URL).mock(return_value=httpx.Response(200, content=blob))
+
+    client = _primed_client()
+    try:
+        async with SessionLocal() as session:
+            h = await download_message_media(client, msg, store, session)
+            await session.commit()
+        expected = hashlib.sha256(blob).hexdigest()
+        assert h == expected
+    finally:
+        await client.aclose()
+        await engine.dispose()
+
+
+@respx.mock
+async def test_download_message_media_type3_opaque_id_returns_none(tmp_path):
+    """Type-3 legacy image: file is an opaque id (not a URL) -> None, no download."""
+    store = ObjectStore(tmp_path)
+    engine = await _make_engine()
+    SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
+    msg = {
+        "msg_id": "m3",
+        "channel": "wechat",
+        "type": "image_legacy",
+        "type_raw": 3,
+        "file_url": "07215407292664565911ce5105",  # opaque id, NOT a URL
+        "md5": None,
+    }
+    # No respx route is registered: any attempted GET would raise. The decoder
+    # must return None without touching the network.
+
+    client = _primed_client()
+    try:
+        async with SessionLocal() as session:
+            assert await download_message_media(client, msg, store, session) is None
+        assert respx.calls.call_count == 0
+    finally:
+        await client.aclose()
+        await engine.dispose()
+
+
+@respx.mock
+async def test_download_message_media_text_message_returns_none(tmp_path):
+    """Plain text message has no file url -> None."""
+    store = ObjectStore(tmp_path)
+    engine = await _make_engine()
+    SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
+    msg = {
+        "msg_id": "mt",
+        "channel": "wechat",
+        "type": "text",
+        "type_raw": 1,
+        "text": "hello",
+    }
+    client = _primed_client()
+    try:
+        async with SessionLocal() as session:
+            assert await download_message_media(client, msg, store, session) is None
+        assert respx.calls.call_count == 0
+    finally:
+        await client.aclose()
+        await engine.dispose()
+
+
+@respx.mock
+async def test_download_message_media_system_link_returns_none(tmp_path):
+    """Type-10 system/link: file IS a support.weixin URL but is NOT media.
+
+    Per phase0-findings.md §2 the type-10 ``file`` field is a
+    ``support.weixin.qq.com`` URL — http(s) but a documentation link, not a
+    media blob. The downloader skips system messages explicitly (the normalized
+    ``type == "system"``), so no network call is made and ``None`` is returned.
+    """
+    store = ObjectStore(tmp_path)
+    engine = await _make_engine()
+    SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
+    support_url = "https://support.weixin.qq.com/cgi-bin/msgredirect?page=xyz"
+    msg = {
+        "msg_id": "ms",
+        "channel": "wechat",
+        "type": "system",
+        "type_raw": 10,
+        "file_url": support_url,
+    }
+    client = _primed_client()
+    try:
+        async with SessionLocal() as session:
+            assert await download_message_media(client, msg, store, session) is None
+        # Type-10 is skipped BEFORE any network attempt.
+        assert respx.calls.call_count == 0
+    finally:
+        await client.aclose()
+        await engine.dispose()
+
+
+@respx.mock
+async def test_download_message_media_download_failure_returns_none(tmp_path):
+    """A transport/http error during download -> None, no exception raised."""
+    store = ObjectStore(tmp_path)
+    engine = await _make_engine()
+    SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
+    msg = {
+        "msg_id": "mf",
+        "channel": "wechat",
+        "type": "image",
+        "type_raw": 2,
+        "file_url": OSS_WEBP_URL,
+        "md5": "ea8f74e7797e63505731e8c51222c554",
+    }
+    respx.get(OSS_WEBP_URL).mock(side_effect=httpx.ConnectError("boom"))
+
+    client = _primed_client()
+    try:
+        async with SessionLocal() as session:
+            result = await download_message_media(client, msg, store, session)
+            await session.commit()
+        assert result is None
+        # No Media row written for a failed download.
+        async with SessionLocal() as session:
+            count = (
+                await session.execute(
+                    sa.select(sa.func.count()).select_from(Media)
+                )
+            ).scalar_one()
+            assert count == 0
+    finally:
+        await client.aclose()
+        await engine.dispose()
+
+
+# ---------------------------------------------------------------------------
+# upsert_media repo
+# ---------------------------------------------------------------------------
+
+
+async def test_upsert_media_inserts_then_updates_volatile_fields():
+    """Fresh upsert writes all fields; a second upsert with a new local_path wins."""
+    engine = await _make_engine()
+    SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
+    try:
+        async with SessionLocal() as session:
+            await upsert_media(
+                session,
+                media_hash="h1",
+                channel="wechat",
+                media_type="image",
+                original_url="https://oss/a.webp",
+                local_path="/tmp/a.webp",
+                size_bytes=10,
+                mime_type="image/webp",
+            )
+            await session.commit()
+
+        async with SessionLocal() as session:
+            await upsert_media(
+                session,
+                media_hash="h1",
+                channel="wechat",
+                media_type="image",
+                original_url="https://oss/a.webp",
+                local_path="/tmp/b.webp",
+                size_bytes=20,
+                mime_type="image/webp",
+            )
+            await session.commit()
+
+        async with SessionLocal() as session:
+            row = (
+                await session.execute(
+                    sa.select(Media).where(Media.media_hash == "h1")
+                )
+            ).scalar_one()
+            assert row.local_path == "/tmp/b.webp"
+            assert row.size_bytes == 20
+    finally:
+        await engine.dispose()
+
+
+async def test_upsert_media_does_not_clobber_local_path_with_none():
+    """A later upsert with local_path=None must keep the existing local_path."""
+    engine = await _make_engine()
+    SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
+    try:
+        async with SessionLocal() as session:
+            await upsert_media(
+                session,
+                media_hash="h2",
+                channel="wechat",
+                local_path="/tmp/keep.webp",
+                size_bytes=42,
+                mime_type="image/webp",
+            )
+            await session.commit()
+
+        async with SessionLocal() as session:
+            await upsert_media(
+                session,
+                media_hash="h2",
+                channel="wechat",
+                # No download this time: local_path/size/mime all None.
+            )
+            await session.commit()
+
+        async with SessionLocal() as session:
+            row = (
+                await session.execute(
+                    sa.select(Media).where(Media.media_hash == "h2")
+                )
+            ).scalar_one()
+            # Preserved, not clobbered.
+            assert row.local_path == "/tmp/keep.webp"
+            assert row.size_bytes == 42
+            assert row.mime_type == "image/webp"
+    finally:
+        await engine.dispose()
+
+
+async def test_upsert_media_is_idempotent_one_row():
+    """Three upserts on the same media_hash -> exactly one row."""
+    engine = await _make_engine()
+    SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
+    try:
+        async with SessionLocal() as session:
+            for _ in range(3):
+                await upsert_media(
+                    session,
+                    media_hash="h3",
+                    channel="wechat",
+                    media_type="image",
+                    local_path="/tmp/x.webp",
+                    size_bytes=1,
+                )
+            await session.commit()
+
+        async with SessionLocal() as session:
+            count = (
+                await session.execute(
+                    sa.select(sa.func.count())
+                    .select_from(Media)
+                    .where(Media.media_hash == "h3")
+                )
+            ).scalar_one()
+            assert count == 1
+    finally:
+        await engine.dispose()