From 7b49f94dcb7fbf786e216c76ebddd2958f6be4e1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Sat, 1 Aug 2026 14:13:50 +0800 Subject: [PATCH] feat: media downloader + serve api MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- yunke-server/app/api/main.py | 4 +- yunke-server/app/api/routers/media.py | 54 ++ yunke-server/app/scraper/media.py | 222 +++++++++ yunke-server/app/storage/__init__.py | 2 + yunke-server/app/storage/repos.py | 100 ++++ yunke-server/tests/test_api_media.py | 147 ++++++ yunke-server/tests/test_media_downloader.py | 523 ++++++++++++++++++++ 7 files changed, 1051 insertions(+), 1 deletion(-) create mode 100644 yunke-server/app/api/routers/media.py create mode 100644 yunke-server/app/scraper/media.py create mode 100644 yunke-server/tests/test_api_media.py create mode 100644 yunke-server/tests/test_media_downloader.py diff --git a/yunke-server/app/api/main.py b/yunke-server/app/api/main.py index dcce284..26ad462 100644 --- a/yunke-server/app/api/main.py +++ b/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") diff --git a/yunke-server/app/api/routers/media.py b/yunke-server/app/api/routers/media.py new file mode 100644 index 0000000..b343318 --- /dev/null +++ b/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) diff --git a/yunke-server/app/scraper/media.py b/yunke-server/app/scraper/media.py new file mode 100644 index 0000000..b9432fa --- /dev/null +++ b/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 diff --git a/yunke-server/app/storage/__init__.py b/yunke-server/app/storage/__init__.py index af20576..2a825fa 100644 --- a/yunke-server/app/storage/__init__.py +++ b/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", diff --git a/yunke-server/app/storage/repos.py b/yunke-server/app/storage/repos.py index ef053ba..7a59007 100644 --- a/yunke-server/app/storage/repos.py +++ b/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) diff --git a/yunke-server/tests/test_api_media.py b/yunke-server/tests/test_api_media.py new file mode 100644 index 0000000..9501f73 --- /dev/null +++ b/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" diff --git a/yunke-server/tests/test_media_downloader.py b/yunke-server/tests/test_media_downloader.py new file mode 100644 index 0000000..5a72e10 --- /dev/null +++ b/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()