From 5ce33232c19eb692881b0a3a0944af015b7bab88 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Tue, 11 Aug 2026 18:24:11 +0800 Subject: [PATCH] =?UTF-8?q?feat(sync):=20=E8=81=8A=E5=A4=A9=E5=9B=BE?= =?UTF-8?q?=E7=89=87=E6=9C=AC=E5=9C=B0=E5=8C=96(message=20media=5Fhash+dow?= =?UTF-8?q?nload=5Fmessage=5Fimages+local-first=20rendering)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Message 模型新增 media_hash 列 (sha256 FK -> Media.media_hash) - Alembic 迁移 e5f6a7b8c9d0_message_media_hash (batch_alter_table, SQLite/MySQL 通用) - app/scraper/media.py:download_message_images 批量下载器，镜像 download_call_recordings - 候选: media_hash IS NULL AND channel IN ('wechat','whatsapp') AND type IN ('image','video','file') AND 任一 file_*_url LIKE 'http%' - image 类型优先 file_th_url (缩略图，小快省)，video/file 优先 file_url - per-row try/except + per-run tried set，失败不挂循环 - sync_message_images orchestrator 入口 + ENTITY_MESSAGE_IMAGES 常量 (提交自身 SyncRun，镜像 call_recordings) - scheduler: _run_message_images + ENTITY_FUNCS['message_images'] + 加入 daily all job (尾部) - MessageOut schema 暴露 media_hash 字段 - MessageBubble.vue local-first: media_hash 优先 /api/media/{hash}，否则回退 OSS file_url/file_th_url/file_hd_url - types.ts: Message 接口添加 media_hash 字段 - 新增 test_message_images_downloader.py (6 tests, 镜像 test_recording_downloader) Co-Authored-By: Claude --- yunke-server/app/api/schemas.py | 5 + yunke-server/app/db/models.py | 8 + yunke-server/app/scraper/media.py | 140 ++++++- yunke-server/app/sync/orchestrator.py | 63 +++ yunke-server/app/sync/scheduler.py | 16 + .../e5f6a7b8c9d0_message_media_hash.py | 35 ++ .../tests/test_message_images_downloader.py | 375 ++++++++++++++++++ yunkes/src/api/types.ts | 7 + yunkes/src/components/MessageBubble.vue | 34 +- 9 files changed, 678 insertions(+), 5 deletions(-) create mode 100644 yunke-server/migrations/versions/e5f6a7b8c9d0_message_media_hash.py create mode 100644 yunke-server/tests/test_message_images_downloader.py diff --git a/yunke-server/app/api/schemas.py b/yunke-server/app/api/schemas.py index 9ee5f71..bb1b50f 100644 --- a/yunke-server/app/api/schemas.py +++ b/yunke-server/app/api/schemas.py @@ -149,6 +149,11 @@ class MessageOut(BaseModel): sender_name: str | None = None sender_head_url: str | None = None timestamp: int | None = None + # sha256 media_hash of the localized media blob (NULL until the + # message_images sync pass downloads it). When present, the frontend + # renders via ``/api/media/{media_hash}`` (local-first); when absent, it + # falls back to the OSS ``file_url`` / ``file_th_url`` / ``file_hd_url``. + media_hash: str | None = None raw_json: dict[str, Any] | None = None diff --git a/yunke-server/app/db/models.py b/yunke-server/app/db/models.py index 681f737..7738c90 100644 --- a/yunke-server/app/db/models.py +++ b/yunke-server/app/db/models.py @@ -377,6 +377,14 @@ class Message(Base): timestamp: Mapped[int | None] = mapped_column(BigInteger, nullable=True) # ms epoch is_del: Mapped[bool | None] = mapped_column(Boolean, nullable=True) + # sha256 media_hash of the localized media blob (FK -> Media.media_hash); + # NULL until sync_message_images downloads it. Populated by the + # message_images sync pass; rendering prefers /api/media/{this} over the + # OSS file_url / file_th_url / file_hd_url URLs. For image messages the + # localized blob is typically the thumbnail (file_th_url) — small + fast + # + sufficient for chat-timeline display. + media_hash: Mapped[str | None] = mapped_column(String(64), nullable=True) + version: Mapped[str | None] = mapped_column(String(64), nullable=True) raw_json: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) fetched_at: Mapped[dt.datetime] = _fetched_at() diff --git a/yunke-server/app/scraper/media.py b/yunke-server/app/scraper/media.py index bea2761..2b76b63 100644 --- a/yunke-server/app/scraper/media.py +++ b/yunke-server/app/scraper/media.py @@ -37,13 +37,14 @@ import httpx import sqlalchemy as sa from sqlalchemy.ext.asyncio import AsyncSession -from app.db.models import Media, PhoneCall, VoiceClip +from app.db.models import Media, Message, PhoneCall, VoiceClip from app.session.yunke_client import YunkeClient from app.storage.object_store import ObjectStore from app.storage.repos import upsert_media __all__ = [ "download_call_recordings", + "download_message_images", "download_message_media", "download_url", "download_voice_clip_audio", @@ -374,3 +375,140 @@ async def download_voice_clip_audio( localized += 1 await session.commit() return attempted, localized + + +def _pick_message_image_url( + *, + type_: str | None, + file_url: str | None, + file_th_url: str | None, + file_hd_url: str | None, +) -> str | None: + """Pick the best downloadable http(s) URL for a message media row. + + For ``image`` messages, prefer the thumbnail (``file_th_url``): thumbnails + are small, fast to fetch, and exactly what the chat timeline renders, so + localizing the thumbnail is sufficient for display and easy on disk + bandwidth. + Falls back to ``file_url`` then ``file_hd_url`` when no http(s) thumbnail is + available. + + For every other media type (``video`` / ``file``), prefer ``file_url`` (the + original), then ``file_hd_url``, then ``file_th_url``. + + Returns ``None`` when no candidate is an http(s) URL (e.g. type-3 opaque ids + or relative filenames). Only ``http://`` / ``https://`` URLs are downloadable. + """ + if type_ == "image": + order = (file_th_url, file_url, file_hd_url) + else: + order = (file_url, file_hd_url, file_th_url) + for url in order: + if _is_http_url(url): + return url # type: ignore[return-value] + return None + + +async def download_message_images( + client: YunkeClient, + store: ObjectStore, + session: AsyncSession, + *, + batch_size: int = 50, +) -> tuple[int, int]: + """Download chat-message media (image / video / file) OSS URLs into the + local object store and link them via ``message.media_hash``. + + Iterates ``message`` rows that carry an http(s) URL on any of + ``file_url`` / ``file_th_url`` / ``file_hd_url``, have ``type`` in + ('image', 'video', 'file'), are on the wechat / whatsapp channel, and do + NOT yet carry a ``media_hash``. For each: pick the best available URL + (image: thumbnail-first; other: original-first — see + :func:`_pick_message_image_url`), fetch bytes (:func:`download_url`), + store content-addressed (:meth:`ObjectStore.put_by_hash`, sha256 dedup), + upsert a ``Media`` row, and write ``media_hash`` back onto the message. + + Returns ``(attempted, localized)``. Resilient: a per-row download/store + failure is logged and counted as attempted-but-not-localized; no exception + escapes. Idempotent within a run: a per-run ``tried`` set excludes already- + attempted msg_ids from later batches so a persistently-failing row is NOT + re-fetched infinitely (it stays ``media_hash IS NULL`` and is retried on + the NEXT run). Rows already carrying ``media_hash`` are skipped; identical + bytes are stored once. + + Mirrors :func:`download_call_recordings` / :func:`download_voice_clip_audio` + exactly. The message ``type`` label filter restricts to media-bearing rows + so text / system / unknown messages are never selected. + """ + attempted = 0 + localized = 0 + tried: set[str] = set() + while True: + stmt = ( + sa.select( + Message.msg_id, + Message.type, + Message.channel, + Message.file_url, + Message.file_th_url, + Message.file_hd_url, + ) + .where(Message.media_hash.is_(None)) + .where(Message.channel.in_(("wechat", "whatsapp"))) + .where(Message.type.in_(("image", "video", "file"))) + .where( + sa.or_( + Message.file_url.like("http%"), + Message.file_th_url.like("http%"), + Message.file_hd_url.like("http%"), + ) + ) + .limit(batch_size) + ) + if tried: + stmt = stmt.where(Message.msg_id.not_in(tried)) + rows = (await session.execute(stmt)).all() + if not rows: + break + for msg_id, type_, channel, file_url, file_th_url, file_hd_url in rows: + attempted += 1 + tried.add(msg_id) + url = _pick_message_image_url( + type_=type_, + file_url=file_url, + file_th_url=file_th_url, + file_hd_url=file_hd_url, + ) + if url is None: + # WHERE clause matched at least one http(s) URL, but the + # preferred slot for this type isn't http(s) — defensive, in + # practice the OR above guarantees _pick returns a value. + continue + try: + data = await download_url(client, url) + except (httpx.HTTPError, OSError) as exc: + logger.warning( + "message media download failed for %s: %s", msg_id, exc + ) + continue + ext = _ext_from_url(url) + sha256_hex, path = await store.put_by_hash(data, ext=ext) + await upsert_media( + session, + media_hash=sha256_hex, + channel=channel or "wechat", + media_type=type_, + original_url=file_url, + original_th_url=file_th_url, + original_hd_url=file_hd_url, + local_path=str(path), + size_bytes=len(data), + mime_type=guess_mime_type(ext), + ) + await session.execute( + sa.update(Message) + .where(Message.msg_id == msg_id) + .values(media_hash=sha256_hex) + ) + localized += 1 + await session.commit() + return attempted, localized diff --git a/yunke-server/app/sync/orchestrator.py b/yunke-server/app/sync/orchestrator.py index 5816b2b..ff9e2a8 100644 --- a/yunke-server/app/sync/orchestrator.py +++ b/yunke-server/app/sync/orchestrator.py @@ -53,6 +53,7 @@ from app.db.models import Department, SyncRun, WechatAccount, WhatsappAccount from app.scraper.call import scrape_phone_calls, scrape_sms from app.scraper.media import ( download_call_recordings, + download_message_images, download_message_media, download_voice_clip_audio, ) @@ -92,11 +93,13 @@ logger = logging.getLogger(__name__) __all__ = [ "ENTITY_CALL_RECORDINGS", + "ENTITY_MESSAGE_IMAGES", "ENTITY_VOICE_CLIP_AUDIO", "SyncResult", "sync_all_wechat", "sync_call_recordings", "sync_departments", + "sync_message_images", "sync_phone_calls", "sync_sms", "sync_voice_clip_audio", @@ -157,6 +160,12 @@ ENTITY_VOICE_CLIP = "voice_clip" # object store (records a SyncRun with this entity_type; no SyncState row # because the voice_clip table itself is the cursor). Mirrors call_recordings. ENTITY_VOICE_CLIP_AUDIO = "voice_clip_audio" +# Message image / media localization pass: not a scrape — reads already-mirrored +# ``message`` rows whose ``file_url`` / ``file_th_url`` / ``file_hd_url`` point +# at OSS and have no ``media_hash`` yet, downloads each into the local object +# store (records a SyncRun with this entity_type; no SyncState row because the +# message table itself is the cursor). Mirrors call_recordings / voice_clip_audio. +ENTITY_MESSAGE_IMAGES = "message_images" CHANNEL_WECHAT = "wechat" CHANNEL_WHATSAPP = "whatsapp" @@ -1152,6 +1161,60 @@ async def sync_voice_clip_audio( ) +async def sync_message_images( + session: AsyncSession, client: YunkeClient, store: ObjectStore +) -> SyncResult: + """Download localized copies of chat-message media for already-mirrored + messages. + + Not a scrape: reads local ``message`` rows whose ``file_url`` / + ``file_th_url`` / ``file_hd_url`` point at OSS, have ``type`` in + ('image', 'video', 'file'), are on the wechat / whatsapp channel, and have + no ``media_hash`` yet; downloads each and links it. Records a ``SyncRun`` + (``entity_type='message_images'``); ``fetched`` = messages needing + download, ``upserted`` = messages localized. + + Mirrors :func:`sync_call_recordings` / :func:`sync_voice_clip_audio` + exactly (same commit semantics): commits its own ``SyncRun`` row because + :func:`download_message_images` commits per batch internally, so the run + bookkeeping must land in the same transaction boundary to stay consistent + with what was actually persisted. + + Failure policy mirrors every other ``sync_*``: a downloader exception is + recorded as a ``status='failed'`` ``SyncRun`` and a failed + :class:`SyncResult` is returned (no re-raise), so a broken image pass does + NOT abort a multi-entity batch. + """ + run = await _start_run(session, ENTITY_MESSAGE_IMAGES) + try: + attempted, localized = await download_message_images( + client, store, session + ) + await _finish_run( + session, run, status="success", fetched=attempted, upserted=localized + ) + await session.commit() + return SyncResult( + entity=ENTITY_MESSAGE_IMAGES, + status="success", + fetched=attempted, + upserted=localized, + ) + except Exception as exc: + logger.exception("sync_message_images failed") + await _finish_run( + session, run, status="failed", fetched=0, upserted=0, error=repr(exc) + ) + await session.commit() + return SyncResult( + entity=ENTITY_MESSAGE_IMAGES, + status="failed", + fetched=0, + upserted=0, + error=repr(exc), + ) + + # --------------------------------------------------------------------------- # Coordinator # --------------------------------------------------------------------------- diff --git a/yunke-server/app/sync/scheduler.py b/yunke-server/app/sync/scheduler.py index 14b6b2a..9dc499f 100644 --- a/yunke-server/app/sync/scheduler.py +++ b/yunke-server/app/sync/scheduler.py @@ -57,6 +57,7 @@ from app.sync.orchestrator import ( sync_all_wechat, sync_call_recordings, sync_departments, + sync_message_images, sync_phone_calls, sync_sms, sync_voice_clip_audio, @@ -218,6 +219,13 @@ async def _run_voice_clip_audio( return [await sync_voice_clip_audio(session, client, store)] +async def _run_message_images( + session: AsyncSession, client: YunkeClient, store: ObjectStore +) -> list[SyncResult]: + """Image / media localization pass for already-mirrored ``message`` rows.""" + return [await sync_message_images(session, client, store)] + + async def _run_messages( session: AsyncSession, client: YunkeClient, store: ObjectStore ) -> list[SyncResult]: @@ -301,6 +309,13 @@ async def _run_all( # in the daily job means no later in-batch entity can be affected by the # mid-batch commit boundary. results.append(await sync_voice_clip_audio(session, client, store)) + # Message image / media localization: download OSS images / videos / files + # for already-mirrored ``message`` rows. Runs AFTER messages (synced + # earlier in this same daily job via _run_all_wechat + _run_wa_messages) + # so freshly-scraped rows are eligible. Commits internally (mirrors + # call_recordings / voice_clip_audio) — also placed at the tail of the + # daily job for the same mid-batch commit-boundary reason. + results.append(await sync_message_images(session, client, store)) return results @@ -324,6 +339,7 @@ ENTITY_FUNCS: dict[ "voice_clips": _run_voice_clips, "call_recordings": _run_call_recordings, "voice_clip_audio": _run_voice_clip_audio, + "message_images": _run_message_images, "messages": _run_messages, "wa_messages": _run_wa_messages, "all_wechat": _run_all_wechat, diff --git a/yunke-server/migrations/versions/e5f6a7b8c9d0_message_media_hash.py b/yunke-server/migrations/versions/e5f6a7b8c9d0_message_media_hash.py new file mode 100644 index 0000000..d36425e --- /dev/null +++ b/yunke-server/migrations/versions/e5f6a7b8c9d0_message_media_hash.py @@ -0,0 +1,35 @@ +"""message media_hash + +Revision ID: e5f6a7b8c9d0 +Revises: c4d5e6f7a8b9 +Create Date: 2026-08-11 00:00:00.000000 + +""" +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + + +# revision identifiers, used by Alembic. +revision: str = 'e5f6a7b8c9d0' +down_revision: str | Sequence[str] | None = 'c4d5e6f7a8b9' +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + """Upgrade schema.""" + # sha256 media_hash of the localized media blob (FK -> Media.media_hash). + # NULL until sync_message_images downloads it. batch_alter_table keeps + # this portable across SQLite (dev/test, table-recreate) and MySQL (prod, + # direct ALTER). Mirrors the voice_clip.media_hash + phone_call.record_media_hash + # migrations. + with op.batch_alter_table('message', schema=None) as batch_op: + batch_op.add_column(sa.Column('media_hash', sa.String(length=64), nullable=True)) + + +def downgrade() -> None: + """Downgrade schema.""" + with op.batch_alter_table('message', schema=None) as batch_op: + batch_op.drop_column('media_hash') diff --git a/yunke-server/tests/test_message_images_downloader.py b/yunke-server/tests/test_message_images_downloader.py new file mode 100644 index 0000000..f2cb0a2 --- /dev/null +++ b/yunke-server/tests/test_message_images_downloader.py @@ -0,0 +1,375 @@ +"""Tests for :func:`app.scraper.media.download_message_images`. + +Covers the message-media batch downloader that localizes chat-message +``file_url`` / ``file_th_url`` / ``file_hd_url`` OSS URLs into the local +object store, writes a ``Media`` row per blob, and links the message back via +``media_hash``. + +Mirrors ``test_recording_downloader.py``: + +* Happy path: one image message -> thumbnail (``file_th_url``) is downloaded, + stored, upserted, linked. +* Thumbnail preference: image with both ``file_url`` and ``file_th_url`` set + -> the thumbnail URL is fetched (not the original). +* Idempotent: a message already carrying ``media_hash`` is excluded from the + candidate set (``attempted == 0``). +* Resilient: a persistent 500 does NOT hang the loop (the per-run ``tried`` + set excludes the failing msg_id from later batches) and returns + ``(attempted=1, localized=0)`` with no exception escaping. +* Filter: a text message with an http URL (e.g. a documentation link) is + NOT selected (only ``type IN ('image','video','file')`` rows are eligible). +""" + +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, Message +from app.scraper.media import download_message_images +from app.session.yunke_client import YunkeClient +from app.storage.object_store import ObjectStore + +BASE = "https://crm2.yunkecn.com" +# Realistic OSS image URL shapes (yunke OSS uses ``wx.oss.yunkecn.com``; the +# downloader is URL-agnostic — any http(s) URL works through the mocked +# transport). Thumbnail + original + hd variants of the same asset. +OSS_THUMB = "https://wx.oss.yunkecn.com/wechat/img/abc_thumb.jpg" +OSS_FULL = "https://wx.oss.yunkecn.com/wechat/img/abc.jpg" +OSS_HD = "https://wx.oss.yunkecn.com/wechat/img/abc_hd.jpg" + +USERID = "a" * 32 +COMPANY = "8m0atd" + + +# --------------------------------------------------------------------------- +# Shared helpers (mirror tests/test_recording_downloader.py) +# --------------------------------------------------------------------------- + + +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 + + +# --------------------------------------------------------------------------- +# download_message_images +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_download_message_images_localizes_and_links(tmp_path): + """One image message with an OSS thumb url: downloaded, stored, linked. + + Returns ``(1, 1)``; the message's ``media_hash`` is the sha256 of the + fetched bytes; a ``Media`` row exists with ``media_type='image'``, + ``mime_type='image/jpeg'``, ``original_url == OSS_FULL``; the file exists + under ``tmp_path`` at the sha256-content-addressed ``.jpg`` path. + """ + store = ObjectStore(tmp_path) + engine = await _make_engine() + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + + blob = b"\xff\xd8\xff\xe0thumb-jpg-payload" + respx.get(OSS_THUMB).mock(return_value=httpx.Response(200, content=blob)) + + client = _primed_client() + try: + async with SessionLocal() as session: + session.add( + Message( + msg_id="m1", + channel="wechat", + type="image", + file_url=OSS_FULL, + file_th_url=OSS_THUMB, + file_hd_url=OSS_HD, + ) + ) + await session.commit() + + async with SessionLocal() as session: + attempted, localized = await download_message_images( + client, store, session + ) + await session.commit() + + assert (attempted, localized) == (1, 1) + + # File landed on disk at the sha256-content-addressed path with .jpg. + expected_sha = hashlib.sha256(blob).hexdigest() + expected_path = tmp_path / expected_sha[:2] / f"{expected_sha}.jpg" + assert expected_path.is_file() + assert expected_path.read_bytes() == blob + + async with SessionLocal() as session: + msg = ( + await session.execute( + sa.select(Message).where(Message.msg_id == "m1") + ) + ).scalar_one() + assert msg.media_hash == expected_sha + + media = ( + await session.execute( + sa.select(Media).where(Media.media_hash == expected_sha) + ) + ).scalar_one() + assert media.media_type == "image" + assert media.mime_type == "image/jpeg" + # All three OSS URLs are preserved on the Media row. + assert media.original_url == OSS_FULL + assert media.original_th_url == OSS_THUMB + assert media.original_hd_url == OSS_HD + assert media.local_path == str(expected_path) + assert media.size_bytes == len(blob) + finally: + await client.aclose() + await engine.dispose() + + +@respx.mock +async def test_image_prefers_thumbnail_url(tmp_path): + """For ``type='image'`` rows, the THUMBNAIL URL is fetched first. + + The row carries all three URLs (file_url / file_th_url / file_hd_url); the + downloader fetches ``file_th_url`` (smaller, sufficient for chat-timeline + display per spec). Verified by asserting only the thumb route was hit. + """ + store = ObjectStore(tmp_path) + engine = await _make_engine() + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + + thumb_blob = b"thumb-bytes" + full_route = respx.get(OSS_FULL).mock(return_value=httpx.Response(200, content=b"full")) + thumb_route = respx.get(OSS_THUMB).mock(return_value=httpx.Response(200, content=thumb_blob)) + + client = _primed_client() + try: + async with SessionLocal() as session: + session.add( + Message( + msg_id="m-pref", + channel="wechat", + type="image", + file_url=OSS_FULL, + file_th_url=OSS_THUMB, + ) + ) + await session.commit() + + async with SessionLocal() as session: + attempted, localized = await download_message_images( + client, store, session + ) + await session.commit() + + assert (attempted, localized) == (1, 1) + assert thumb_route.called + assert not full_route.called + + # media row is keyed by the THUMBNAIL's sha256. + expected_sha = hashlib.sha256(thumb_blob).hexdigest() + async with SessionLocal() as session: + msg = ( + await session.execute( + sa.select(Message).where(Message.msg_id == "m-pref") + ) + ).scalar_one() + assert msg.media_hash == expected_sha + finally: + await client.aclose() + await engine.dispose() + + +@respx.mock +async def test_video_uses_file_url(tmp_path): + """For ``type='video'`` rows, ``file_url`` (the original) is fetched. + + The image-thumbnail preference does NOT apply to video; the original + ``file_url`` is the right asset. + """ + store = ObjectStore(tmp_path) + engine = await _make_engine() + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + + blob = b"\x00\x00\x00 mp4-isom-payload" + video_url = "https://wx.oss.yunkecn.com/wechat/vid/abc.mp4" + respx.get(video_url).mock(return_value=httpx.Response(200, content=blob)) + + client = _primed_client() + try: + async with SessionLocal() as session: + session.add( + Message( + msg_id="m-vid", + channel="wechat", + type="video", + file_url=video_url, + ) + ) + await session.commit() + + async with SessionLocal() as session: + attempted, localized = await download_message_images( + client, store, session + ) + await session.commit() + + assert (attempted, localized) == (1, 1) + expected_sha = hashlib.sha256(blob).hexdigest() + expected_path = tmp_path / expected_sha[:2] / f"{expected_sha}.mp4" + assert expected_path.is_file() + + async with SessionLocal() as session: + msg = ( + await session.execute( + sa.select(Message).where(Message.msg_id == "m-vid") + ) + ).scalar_one() + assert msg.media_hash == expected_sha + finally: + await client.aclose() + await engine.dispose() + + +@respx.mock +async def test_idempotent_skip(tmp_path): + """A message already carrying ``media_hash`` is excluded -> attempted 0. + + The WHERE clause filters ``media_hash IS NULL``, so a previously localized + message is never re-fetched. + """ + store = ObjectStore(tmp_path) + engine = await _make_engine() + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + + client = _primed_client() + try: + async with SessionLocal() as session: + session.add( + Message( + msg_id="m2", + channel="wechat", + type="image", + file_url=OSS_FULL, + file_th_url=OSS_THUMB, + media_hash="already", + ) + ) + await session.commit() + + async with SessionLocal() as session: + attempted, localized = await download_message_images( + client, store, session + ) + await session.commit() + + assert (attempted, localized) == (0, 0) + # No network was touched: the candidate set was empty. + assert respx.calls.call_count == 0 + finally: + await client.aclose() + await engine.dispose() + + +@respx.mock +async def test_resilient_on_failure_and_no_infinite_loop(tmp_path): + """A persistent 500: no exception, loop TERMINATES, ``(1, 0)`` returned. + + This is the key guard for the per-run ``tried`` set: without it, the + failing row would stay ``media_hash IS NULL`` and be re-selected by every + subsequent batch forever, hanging the run. The ``tried`` set excludes + the attempted msg_id from later batches, so the next SELECT returns an + empty page and the loop breaks. + """ + store = ObjectStore(tmp_path) + engine = await _make_engine() + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + + respx.get(OSS_THUMB).mock(return_value=httpx.Response(500)) + + client = _primed_client() + try: + async with SessionLocal() as session: + session.add( + Message( + msg_id="m3", + channel="wechat", + type="image", + file_th_url=OSS_THUMB, + ) + ) + await session.commit() + + async with SessionLocal() as session: + attempted, localized = await download_message_images( + client, store, session + ) + await session.commit() + + assert (attempted, localized) == (1, 0) + + # The failing message is left un-localized (retried on the NEXT run). + async with SessionLocal() as session: + msg = ( + await session.execute( + sa.select(Message).where(Message.msg_id == "m3") + ) + ).scalar_one() + assert msg.media_hash is None + finally: + await client.aclose() + await engine.dispose() + + +@respx.mock +async def test_text_message_with_url_is_not_selected(tmp_path): + """A text message whose text happens to be a URL is NOT selected. + + The candidate filter requires ``type IN ('image','video','file')``; a + text message is skipped even if its columns somehow carry an http URL. + """ + store = ObjectStore(tmp_path) + engine = await _make_engine() + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + + client = _primed_client() + try: + async with SessionLocal() as session: + session.add( + Message( + msg_id="m-text", + channel="wechat", + type="text", + text="see https://wx.oss.yunkecn.com/wechat/img/abc.jpg", + # Even with a file_url, type='text' disqualifies the row. + file_url=OSS_FULL, + ) + ) + await session.commit() + + async with SessionLocal() as session: + attempted, localized = await download_message_images( + client, store, session + ) + await session.commit() + + assert (attempted, localized) == (0, 0) + assert respx.calls.call_count == 0 + finally: + await client.aclose() + await engine.dispose() diff --git a/yunkes/src/api/types.ts b/yunkes/src/api/types.ts index 59a8f7e..b853820 100644 --- a/yunkes/src/api/types.ts +++ b/yunkes/src/api/types.ts @@ -282,6 +282,13 @@ export interface Message { sender_head_url?: string | null /** ms-epoch */ timestamp?: number | null + /** + * sha256 media_hash of the localized media blob (NULL until the + * message_images sync pass downloads it). When present, MessageBubble + * renders via `/api/media/${media_hash}` (local-first); when absent, it + * falls back to the OSS file_url / file_th_url / file_hd_url. + */ + media_hash?: string | null is_del?: boolean | null /** * Verbatim upstream message record (camelCase keys: msgSvrId/keyId, type, diff --git a/yunkes/src/components/MessageBubble.vue b/yunkes/src/components/MessageBubble.vue index 2000ccd..887189f 100644 --- a/yunkes/src/components/MessageBubble.vue +++ b/yunkes/src/components/MessageBubble.vue @@ -59,6 +59,19 @@ const bestMediaUrl = computed(() => { return undefined }) +/** + * Local-first media src — when the backend has localized the media blob + * (`message.media_hash` set by the `message_images` sync pass), render via + * the local `/api/media/{hash}` endpoint instead of hitting yunke OSS. + * Returns `undefined` when no localized copy exists yet; callers fall back + * to `bestMediaUrl` / `imageFullUrl` (the OSS URLs) so a not-yet-localized + * message still renders (degraded — talks to OSS). + */ +const localSrc = computed(() => { + const hash = props.message.media_hash + return hash ? `/api/media/${hash}` : undefined +}) + const IMAGE_EXTS = new Set(['jpg', 'jpeg', 'webp', 'png', 'gif', 'bmp', 'tif', 'tiff']) const VIDEO_EXTS = new Set(['mp4', 'mov', 'webm', 'm4v', 'avi', 'mkv']) const AUDIO_EXTS = new Set(['mp3', 'amr', 'silk', 'm4a', 'aac', 'wav', 'ogg']) @@ -88,11 +101,24 @@ const mediaKind = computed<'image' | 'video' | 'audio' | 'file' | undefined>(() return 'file' }) +/** + * Source for the rendered image / video / audio / file link. Local-first: + * when `media_hash` is present, serve the localized blob via + * `/api/media/{hash}`; otherwise fall back to the best OSS URL on the row + * (degraded — talks to yunke OSS directly). For images, click-to-zoom also + * uses the local URL when available (the localized blob is what we have). + */ +const mediaSrc = computed(() => localSrc.value || bestMediaUrl.value) + /** * Image preview src list — prefer the highest-fidelity http URL available so * click-to-zoom shows the best copy. Falls back to the displayed thumb URL. + * When a localized copy exists, the preview is the local URL (the localized + * blob is what we have on hand — typically the thumbnail, which is + * sufficient for chat-timeline display). */ const imageFullUrl = computed(() => { + if (localSrc.value) return localSrc.value const m = props.message for (const u of [m.file_hd_url, m.file_url, m.file_th_url]) { if (u && HTTP_RE.test(u)) return u @@ -138,7 +164,7 @@ const typeTagLabel = computed(() => {
{