浅念
7b49f94dcb
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>
524 lines
18 KiB
Python
524 lines
18 KiB
Python
"""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()
|