feat(scraper): moments 实体(浏览器fetch getFriendsterInfo,秒→ms,hashNext翻页,接入full_sync+4h调度+图片media)

- scrape_via_page:_normalize_moments(createTime秒×1000,缺id丢弃)+
  _iter_moments_pages 翻页内核(fetch回调注入,sleep可注入,hashNext假/
  信封异常/max_pages 上限即止)+ scrape_moments(逐员工遍历wechat_account,
  form小驼峰,首页code≠10000告警cookie失效,换页重试惯例)+CLI注册moments
- media:download_moments_images(image_urls JSON数组展开去重,MomentsPost无
  回链列→用Media.original_url做已下载标记,分片IN防SQLite变量上限,批间commit)
  接入scrape_media第4源
- full_sync_all.sh:9c步moments(media前,同轮图片即可本地化)+汇总加moments计数
- scheduler:4h messages job 末尾追加 moments --max-pages 3 浅扫(全量翻页
  由每日full承担);测试断言7条命令序列
- tests/test_scrape_moments.py:归一化(秒→ms/空安全/坏createTime)+翻页内核
  (两页累积停止/hashNext假单页/max_pages防死循环/丢id行/坏信封)+图片本地化
  (一帖两图/幂等跳过/非http跳过+500容错);全套440测试通过

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

yunke-server/app/scraper/media.py

@@ -37,7 +37,7 @@ import httpx
 import sqlalchemy as sa
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.db.models import Media, Message, PhoneCall, VoiceClip
+from app.db.models import Media, Message, MomentsPost, PhoneCall, VoiceClip
 from app.session.yunke_client import YunkeClient
 from app.storage.object_store import ObjectStore
 from app.storage.repos import upsert_media
@@ -46,6 +46,7 @@ __all__ = [
     "download_call_recordings",
     "download_message_images",
     "download_message_media",
+    "download_moments_images",
     "download_url",
     "download_voice_clip_audio",
     "guess_mime_type",
@@ -535,3 +536,88 @@ async def download_message_images(
             localized += 1
         await session.commit()
     return attempted, localized
+
+
+async def download_moments_images(
+    client: YunkeClient,
+    store: ObjectStore,
+    session: AsyncSession,
+    *,
+    batch_size: int = 50,
+) -> tuple[int, int]:
+    """Download 员工朋友圈 images (``moments_post.image_urls`` OSS 直链) into
+    the local object store, one ``Media`` row per blob.
+
+    朋友圈与消息图片的关键差异(2026-08-25 接入时确认):``image_urls`` 是
+    JSON **数组**(一帖多图),且 :class:`MomentsPost` 没有 ``media_hash``
+    回链列 —— 已下载标记改用 ``Media.original_url``:URL 已有对应 Media 行
+    即视为已本地化并跳过(不重下载)。无回链也就无法"标记"帖子行,故候选
+    集一次物化(全表展开 + 去重),再按 ``original_url`` 剔除已下载的,分批
+    下载(批间 commit,部分进度可保留)。
+
+    Returns ``(attempted, localized)``. Resilient: a per-URL download failure is
+    logged and counted as attempted-but-not-localized; no exception escapes.
+    Failed URLs keep no Media row → retried on the NEXT run. Identical bytes are
+    stored once (:meth:`ObjectStore.put_by_hash`, sha256 dedup).
+
+    Known edge: two DISTINCT URLs hosting byte-identical content share one
+    ``media_hash`` (content-addressed PK), and the upsert refreshes
+    ``original_url`` to the latest — so the loser URL is re-fetched once per run
+    (bounded, never grows; identical bytes at two OSS paths is rare).
+    """
+    rows = (
+        await session.execute(
+            sa.select(MomentsPost.id, MomentsPost.image_urls).where(
+                MomentsPost.image_urls.is_not(None)
+            )
+        )
+    ).all()
+
+    # 一帖多图 → 展开去重(同图多帖只下载一次)
+    urls: list[str] = []
+    seen: set[str] = set()
+    for _post_id, image_urls in rows:
+        for url in image_urls or []:
+            if _is_http_url(url) and url not in seen:
+                seen.add(url)
+                urls.append(url)
+    if not urls:
+        return 0, 0
+
+    # 剔除已本地化的 URL。分片 IN —— 单次塞全部 URL 会超 SQLite 变量上限
+    # (同 download_message_images 的 tried 分片惯例,2026-08-20)。
+    existing: set[str] = set()
+    for i in range(0, len(urls), 500):
+        res = await session.execute(
+            sa.select(Media.original_url).where(
+                Media.original_url.in_(urls[i : i + 500])
+            )
+        )
+        existing.update(u for (u,) in res.fetchall() if u)
+    pending = [u for u in urls if u not in existing]
+
+    attempted = 0
+    localized = 0
+    for i in range(0, len(pending), batch_size):
+        for url in pending[i : i + batch_size]:
+            attempted += 1
+            try:
+                data = await download_url(client, url)
+            except (httpx.HTTPError, OSError) as exc:
+                logger.warning("moments image download failed for %s: %s", url, 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="wechat",  # 朋友圈是微信独有实体(无 WA 朋友圈)
+                media_type="image",
+                original_url=url,
+                local_path=str(path),
+                size_bytes=len(data),
+                mime_type=guess_mime_type(ext),
+            )
+            localized += 1
+        await session.commit()
+    return attempted, localized

yunke-server/app/sync/scheduler.py

@@ -578,7 +578,8 @@ class SyncScheduler:
 
         * ``full``     → ``bash scripts/full_sync_all.sh``(页面方式全量序列)
         * ``messages`` → ``python scripts/scrape_via_page.py --entity messages
-          --max-conversations N``(增量:未扫对话优先,其余按最近沟通)
+          --max-conversations N``(增量:未扫对话优先,其余按最近沟通),末尾
+          追加 ``--entity moments --max-pages 3``(朋友圈浅扫,每员工最多 3 页)
 
         子进程隔离——浏览器崩溃/断连不影响调度进程;非零退出只记 JSON 日志
         不抛异常(保住下一个周期)。asyncio.Lock 保证 full/messages 互斥。
@@ -594,7 +595,9 @@ class SyncScheduler:
             elif kind == "messages":
                 # 先刷客户列表(最近沟通时间是云客列表的实时排序键,friends 快照
                 # 滞后会让本地排序与云客第一页完全不同,2026-08-20),再抓全部四类
-                # 消息增量(2026-08-21 全接入:水位增量后单对话 1 页,可承受)
+                # 消息增量(2026-08-21 全接入:水位增量后单对话 1 页,可承受);
+                # 末尾追加员工朋友圈浅扫(2026-08-25,max-pages 3 即每员工最多
+                # 3 页,新帖通常落第一页;全量翻页由每日 full 走 full_sync_all.sh)
                 commands = [
                     [sys.executable, "scripts/scrape_via_page.py", "--entity", "friends"],
                     [sys.executable, "scripts/scrape_via_page.py", "--entity", "wa-friends"],
@@ -608,6 +611,15 @@ class SyncScheduler:
                         "99999",
                     ]
                     for e in ("messages", "wa-messages", "group-messages", "wa-group-messages")
+                ] + [
+                    [
+                        sys.executable,
+                        "scripts/scrape_via_page.py",
+                        "--entity",
+                        "moments",
+                        "--max-pages",
+                        "3",
+                    ]
                 ]
             elif kind == "full-history":
                 # 每周全量对齐(周日 05:00):强制全量翻历史,对齐增量模式抓不到的

yunke-server/scripts/full_sync_all.sh

@@ -41,6 +41,8 @@ echo "$(date) --- 9/10 WA messages ---"
 $PY scripts/scrape_via_page.py --entity wa-messages --max-conversations 99999
 echo "$(date) --- 9b/10 WA group messages ---"
 $PY scripts/scrape_via_page.py --entity wa-group-messages --max-conversations 99999
+echo "$(date) --- 9c/10 Moments (员工朋友圈) ---"
+$PY scripts/scrape_via_page.py --entity moments
 echo "$(date) --- 10/10 Media download (audio+images) ---"
 $PY scripts/scrape_via_page.py --entity media
 
@@ -65,6 +67,7 @@ for label, sql in [
     ('whatsapp messages', \"SELECT COUNT(*) FROM message WHERE channel='whatsapp'\"),
     ('voice calls', \"SELECT COUNT(*) FROM phone_call WHERE channel IN ('voice_call','wa_voice_call')\"),
     ('voice clips', 'SELECT COUNT(*) FROM voice_clip'),
+    ('moments posts', 'SELECT COUNT(*) FROM moments_post'),
     ('media files', 'SELECT COUNT(*) FROM media'),
 ]:
     print(f'  {label:22s} {c.execute(sql).fetchone()[0]:>6}')

yunke-server/scripts/scrape_via_page.py

@@ -16,6 +16,7 @@
     DB_URL="sqlite+aiosqlite:///./smoke.db" .venv/Scripts/python.exe scripts/scrape_via_page.py --entity voice-calls
     DB_URL="sqlite+aiosqlite:///./smoke.db" .venv/Scripts/python.exe scripts/scrape_via_page.py --entity voice-clips
     DB_URL="sqlite+aiosqlite:///./smoke.db" .venv/Scripts/python.exe scripts/scrape_via_page.py --entity wa-messages
+    DB_URL="sqlite+aiosqlite:///./smoke.db" .venv/Scripts/python.exe scripts/scrape_via_page.py --entity moments
     DB_URL="sqlite+aiosqlite:///./smoke.db" .venv/Scripts/python.exe scripts/scrape_via_page.py --entity media
 
 Cookie 从 .env 的 YK_COOKIE 读取。
@@ -27,6 +28,7 @@ import asyncio
 import json
 import logging
 import os
+from typing import Any
 from urllib.parse import urlparse
 
 logger = logging.getLogger("scrape_via_page")
@@ -44,6 +46,7 @@ INTERCEPT_PATHS = {
     "wa_friend": "/qqweb/WhatsAppUser/businessFriendList",  # WA friends + groups
     "wa_user": "/qqweb/WhatsAppUser/businessSalesList",     # WA accounts
     "msg_chats": "/wechatweb/msg/chats",                    # chat messages
+    "friendster": "/wechatweb/wechatFriendster/getFriendsterInfo",  # 员工朋友圈
 }
 
 
@@ -1377,6 +1380,153 @@ async def scrape_wa_groups(cookie_str: str, max_pages: int = 200):
     return total_fetched, total_upserted
 
 
+# ---------------------------------------------------------------------------
+# 员工朋友圈 Moments (browser fetch, per employee)
+# ---------------------------------------------------------------------------
+
+
+def _normalize_moments(record: dict[str, Any]) -> dict[str, Any]:
+    """云客朋友圈行(getFriendsterInfo data.data[]) → moments_post 记录。
+
+    真机抓包(2026-08-25):行的 ``createTime`` 是**秒**字符串(与消息接口的
+    ms 不同),统一 ×1000 成 ms epoch;缺 id 行返回 id=None 由调用方丢弃。
+    """
+    ct = record.get("createTime")
+    try:
+        ct_ms = int(ct) * 1000 if ct is not None else None
+    except (TypeError, ValueError):
+        ct_ms = None
+    return {
+        "id": record.get("id"),
+        "wechat_id": record.get("wechatId"),
+        "sns_id": record.get("snsId"),
+        "author_name": record.get("friendsterName"),
+        "author_wxid": record.get("friendsterWXId"),
+        "author_avatar": record.get("friendsterWXIdImg"),
+        "content": record.get("content"),
+        "type_raw": str(record.get("type")) if record.get("type") is not None else None,
+        "create_time": ct_ms,
+        "praise_json": record.get("praise") or [],
+        "comment_json": record.get("comment") or [],
+        "image_urls": record.get("imgUrlsList") or [],
+        "raw_json": record,
+    }
+
+
+async def _iter_moments_pages(
+    fetch,
+    wechat_id: str,
+    page_size: int = 100,
+    max_pages: int = 50,
+    sleep=None,
+):
+    """getFriendsterInfo 翻页内核(fetch 回调注入,测试不启浏览器)。
+
+    ``fetch(body) -> 响应 dict``;body 由本内核按真机抓包构造(form 小驼峰:
+    ``startTime``/``endTime`` 空串、``pageNum`` 从 1、``pageSize``、
+    ``wechatId``)。逐页 ``data.data[]`` → :func:`_normalize_moments`(丢
+    id=None 行)累积;``data.hashNext`` 假 / 信封异常 / 达 ``max_pages`` 页
+    上限即止(4h 调度浅扫 max-pages 3 靠此上限)。页间 sleep 1s。
+    """
+    if sleep is None:
+        sleep = asyncio.sleep
+    records: list[dict] = []
+    page_num = 1
+    while page_num <= max_pages:
+        body = {
+            "startTime": "",
+            "endTime": "",
+            "pageNum": page_num,
+            "pageSize": page_size,
+            "wechatId": wechat_id,
+        }
+        resp = await fetch(body)
+        data = resp.get("data") if isinstance(resp, dict) else None
+        rows = data.get("data") if isinstance(data, dict) else None
+        if not isinstance(rows, list):
+            break
+        for row in rows:
+            if not isinstance(row, dict):
+                continue
+            rec = _normalize_moments(row)
+            if rec.get("id") is not None:
+                records.append(rec)
+        if not data.get("hashNext"):
+            break
+        page_num += 1
+        await sleep(1)
+    return records
+
+
+async def scrape_moments(cookie_str: str, max_pages: int = 50, page_size: int = 100):
+    """员工朋友圈:浏览器直发 getFriendsterInfo(form-encoded),逐员工翻页。
+
+    结构镜像 scrape_wa_groups;差异在**逐员工**遍历(端点按 wechatId 过滤,
+    微信渠道独有实体,wechat_account 表全量)。翻页开关是 ``data.hashNext``
+    (不是短页判断);批量 upsert 走 :func:`upsert_moments_posts`(收整列表)。
+    返回总计 dict。
+    """
+    from sqlalchemy import text as sa_text
+
+    from app.db.base import async_engine, async_session_maker
+    from app.storage.repos import upsert_moments_posts
+
+    async with async_engine.connect() as conn:
+        result = await conn.execute(sa_text(
+            "SELECT wx_id FROM wechat_account WHERE wx_id IS NOT NULL AND wx_id != ''"
+        ))
+        employees = [row[0] for row in result.fetchall()]
+
+    total_fetched = 0
+    total_upserted = 0
+    async with PageScraper(cookie_str) as scraper:
+        page = await _open_fetch_page(scraper)
+        for i, wx_id in enumerate(employees):
+            records: list[dict] = []
+
+            async def _fetch(body: dict):
+                resp = await _browser_fetch(
+                    page, INTERCEPT_PATHS["friendster"], form_body=body
+                )
+                # cookie 失效告警惯例(同 _fetch_pages 首页 code 校验):
+                # 健康 list 端点 code=10000
+                if body["pageNum"] == 1 and isinstance(resp, dict):
+                    code = resp.get("code")
+                    if code is not None and str(code) not in ("10000",):
+                        logger.warning(
+                            f"⚠️ {INTERCEPT_PATHS['friendster']} code={code} "
+                            f"({resp.get('message')}) — cookie 可能已过期/失效,"
+                            "请从浏览器重新复制 YK_COOKIE 到 .env!"
+                        )
+                return resp
+
+            try:
+                records = await _iter_moments_pages(
+                    _fetch, wx_id, page_size=page_size, max_pages=max_pages
+                )
+            except Exception as e:
+                logger.warning(f"  [{i+1}/{len(employees)}] {wx_id[:16]}.. failed: {e}")
+                # 换页重试惯例(同 scrape_voice_clips):该员工放弃,继续下一个
+                try:
+                    await page.close()
+                except Exception:
+                    pass
+                page = await _open_fetch_page(scraper)
+                continue
+
+            total_fetched += len(records)
+            if records:
+                async with async_session_maker() as session:
+                    await upsert_moments_posts(session, records)
+                    await session.commit()
+                total_upserted += len(records)
+            logger.info(f"DONE moments {wx_id} +{len(records)}")
+            await asyncio.sleep(2 + (i % 3))
+
+    logger.info(f"DONE: moments fetched={total_fetched}, upserted={total_upserted}")
+    return {"fetched": total_fetched, "upserted": total_upserted}
+
+
 # ---------------------------------------------------------------------------
 # Voice call records (navigation + interception, both WeChat + WA pages)
 # ---------------------------------------------------------------------------
@@ -1841,6 +1991,7 @@ async def scrape_media(cookie_str: str, batch_size: int = 50):
     * call recordings (phone_call.record_file OSS URLs)
     * voice clip audio (voice_clip.oss_file_name OSS URLs)
     * message images/videos (message.file_url / file_th_url / file_hd_url)
+    * moments images (moments_post.image_urls OSS URLs, 一帖多图)
 
     These are plain OSS CDN downloads (not CRM API calls), so httpx direct
     fetching is safe. The YunkeClient is only used as the httpx transport.
@@ -1850,6 +2001,7 @@ async def scrape_media(cookie_str: str, batch_size: int = 50):
     from app.scraper.media import (
         download_call_recordings,
         download_message_images,
+        download_moments_images,
         download_voice_clip_audio,
     )
     from app.session.yunke_client import YunkeClient
@@ -1872,6 +2024,7 @@ async def scrape_media(cookie_str: str, batch_size: int = 50):
             totals["call_recordings"] = await download_call_recordings(client, store, session, batch_size=batch_size)
             totals["voice_clip_audio"] = await download_voice_clip_audio(client, store, session, batch_size=batch_size)
             totals["message_images"] = await download_message_images(client, store, session, batch_size=batch_size)
+            totals["moments_images"] = await download_moments_images(client, store, session, batch_size=batch_size)
     finally:
         await client.aclose()
 
@@ -1899,13 +2052,14 @@ async def main():
             "voice-clips",   # 长语音 (browser fetch call/list per employee)
             "wa-messages",   # WA消息 (browser fetch chatRecords per conversation)
             "wa-group-messages", # WA群消息 (chatRecords with friendUid=@g.us)
+            "moments",       # 员工朋友圈 (browser fetch getFriendsterInfo)
             "media",         # OSS媒体下载 (录音/长语音音频/消息图片)
         ],
         help="抓取实体",
     )
     parser.add_argument("--max-employees", type=int, default=100, help="最多遍历多少员工 (friends/groups/voice-clips)")
     parser.add_argument("--max-conversations", type=int, default=100, help="消息模式：最多遍历多少对话 (messages/wa-messages/group-messages)")
-    parser.add_argument("--max-pages", type=int, default=200, help="列表翻页上限 (accounts/wa-accounts/wa-friends/voice-calls)")
+    parser.add_argument("--max-pages", type=int, default=200, help="列表翻页上限 (accounts/wa-accounts/wa-friends/voice-calls/moments)")
     parser.add_argument("--batch-size", type=int, default=50, help="媒体下载批次大小 (media)")
     parser.add_argument("--full-history", action="store_true", help="消息类实体强制全量翻历史(默认增量水位)")
     args = parser.parse_args()
@@ -1945,6 +2099,8 @@ async def main():
         await scrape_wa_messages(cookie, args.max_conversations, args.full_history)
     elif args.entity == "wa-group-messages":
         await scrape_wa_group_messages(cookie, args.max_conversations, args.full_history)
+    elif args.entity == "moments":
+        await scrape_moments(cookie, args.max_pages)
     elif args.entity == "media":
         await scrape_media(cookie, args.batch_size)
 

yunke-server/tests/test_scrape_moments.py

@@ -0,0 +1,266 @@
+"""scrape_moments 归一化 + 翻页内核 + 图片本地化测试（不启浏览器）。
+
+云客接口事实（2026-08-25 真机抓包，已验证）：
+``POST /wechatweb/wechatFriendster/getFriendsterInfo``，form 小驼峰字段
+``startTime``/``endTime``(空串)/``pageNum``(从1)/``pageSize``/``wechatId``；
+信封 ``data.data[]`` + ``data.hashNext``；行的 ``createTime`` 是**秒**
+（与消息接口的 ms 不同，归一化 ×1000）。
+"""
+
+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, MomentsPost
+from app.scraper.media import download_moments_images
+from app.session.yunke_client import YunkeClient
+from app.storage.object_store import ObjectStore
+from scripts.scrape_via_page import _iter_moments_pages, _normalize_moments
+
+RAW = {  # 2026-08-25 真机抓包样本（截取关键字段）
+    "id": "69e082a3c3524a7fe3ac53b4", "wechatId": "wxid_wo97u1u2x11w22",
+    "snsId": "-3546042415332912522", "friendsterName": "Yoyo-HESU",
+    "friendsterWXId": "wxid_wo97u1u2x11w22",
+    "friendsterWXIdImg": "https://wx.qlogo.cn/mmhead/ver_1/vsHn.../0",
+    "content": "Welcome to HESU", "type": "1", "createTime": "1776302051",  # 秒！
+    "praise": [{"name": "E**U", "praiseId": "wxid***r822", "praiseImg": "https://wx.qlogo.cn/p/0"}],
+    "comment": [],
+    "imgUrlsList": ["https://wx.oss.yunkecn.com/friendstr/8m0atd/16/b1/ea/xx.jpg"],
+}
+
+
+# ---------------------------------------------------------------------------
+# _normalize_moments
+# ---------------------------------------------------------------------------
+
+
+def test_normalize_moments_converts_seconds_to_ms():
+    rec = _normalize_moments(RAW)
+    assert rec["id"] == "69e082a3c3524a7fe3ac53b4"
+    assert rec["create_time"] == 1776302051 * 1000  # 秒→ms
+    assert rec["author_name"] == "Yoyo-HESU"
+    assert rec["praise_json"][0]["name"] == "E**U"
+    assert rec["image_urls"] == RAW["imgUrlsList"]
+    assert rec["raw_json"]["snsId"] == "-3546042415332912522"
+
+
+def test_normalize_moments_null_safe():
+    rec = _normalize_moments({})
+    assert rec["id"] is None  # 缺 id 行由调用方丢弃
+    assert rec["create_time"] is None
+    assert rec["praise_json"] == []
+    assert rec["image_urls"] == []
+
+
+def test_normalize_moments_bad_create_time_is_none():
+    rec = _normalize_moments({**RAW, "createTime": "not-a-number"})
+    assert rec["create_time"] is None
+    assert rec["id"] == RAW["id"]  # 其余字段不受影响
+
+
+# ---------------------------------------------------------------------------
+# _iter_moments_pages（翻页内核，fetch 回调注入）
+# ---------------------------------------------------------------------------
+
+
+def _page(rows: list[dict], hash_next: bool) -> dict:
+    return {"code": "10000", "data": {"hashNext": hash_next, "data": rows}}
+
+
+async def _noop_sleep(_seconds: float) -> None:
+    return None
+
+
+async def test_iter_moments_pages_accumulates_two_pages_and_stops():
+    """第一页 hashNext=True → 继续；第二页 hashNext=False → 停（恰好 2 次请求）。"""
+    bodies: list[dict] = []
+
+    async def fetch(body: dict) -> dict:
+        bodies.append(body)
+        return _page([RAW] if body["pageNum"] == 1 else [{**RAW, "id": "second"}], body["pageNum"] == 1)
+
+    records = await _iter_moments_pages(
+        fetch, "wxid_wo97u1u2x11w22", page_size=100, max_pages=50, sleep=_noop_sleep
+    )
+
+    assert [r["id"] for r in records] == [RAW["id"], "second"]
+    assert records[0]["create_time"] == 1776302051 * 1000
+    # 恰好两次请求：hashNext=False 即止
+    assert len(bodies) == 2
+    # form 字段与真机抓包一致（小驼峰，pageNum 从 1 起）
+    assert bodies[0] == {
+        "startTime": "", "endTime": "", "pageNum": 1, "pageSize": 100,
+        "wechatId": "wxid_wo97u1u2x11w22",
+    }
+    assert bodies[1]["pageNum"] == 2
+
+
+async def test_iter_moments_pages_stops_when_hash_next_falsy():
+    """单页 hashNext=False → 只请求一次。"""
+    calls: list[int] = []
+
+    async def fetch(body: dict) -> dict:
+        calls.append(body["pageNum"])
+        return _page([RAW], hash_next=False)
+
+    records = await _iter_moments_pages(fetch, "wx", sleep=_noop_sleep)
+
+    assert len(records) == 1
+    assert calls == [1]
+
+
+async def test_iter_moments_pages_respects_max_pages_cap():
+    """hashNext 永真 → max_pages 页上限防死循环（4h 调度浅扫 max-pages 3 靠它）。"""
+    calls: list[int] = []
+
+    async def fetch(body: dict) -> dict:
+        calls.append(body["pageNum"])
+        return _page([{**RAW, "id": f"id-{body['pageNum']}"}], hash_next=True)
+
+    records = await _iter_moments_pages(fetch, "wx", max_pages=3, sleep=_noop_sleep)
+
+    assert len(calls) == 3
+    assert [r["id"] for r in records] == ["id-1", "id-2", "id-3"]
+
+
+async def test_iter_moments_pages_drops_null_id_rows_and_bad_envelope():
+    """缺 id 行丢弃；data 非 dict / data.data 非列表 → 停止不抛。"""
+    async def fetch(body: dict) -> dict:
+        if body["pageNum"] == 1:
+            return _page([RAW, {"content": "no id"}], hash_next=True)
+        return _page([], hash_next=False)
+
+    records = await _iter_moments_pages(fetch, "wx", sleep=_noop_sleep)
+    assert [r["id"] for r in records] == [RAW["id"]]
+
+    async def fetch_bad(_body: dict) -> dict:
+        return {"code": "10000", "data": None}
+
+    assert await _iter_moments_pages(fetch_bad, "wx", sleep=_noop_sleep) == []
+
+
+# ---------------------------------------------------------------------------
+# download_moments_images（朋友圈图片本地化）
+# ---------------------------------------------------------------------------
+
+BASE = "https://crm2.yunkecn.com"
+OSS_A = "https://wx.oss.yunkecn.com/friendstr/8m0atd/16/b1/ea/a.jpg"
+OSS_B = "https://wx.oss.yunkecn.com/friendstr/8m0atd/16/b1/ea/b.jpg"
+
+
+async def _make_engine():
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    return engine
+
+
+def _primed_client() -> YunkeClient:
+    return YunkeClient(base_url=BASE, userid="a" * 32, company="8m0atd")
+
+
+@respx.mock
+async def test_download_moments_images_localizes_all_urls(tmp_path):
+    """一帖两图(image_urls JSON 数组)全部下载入 media 表,media_type='image'。"""
+    store = ObjectStore(tmp_path)
+    engine = await _make_engine()
+    SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
+
+    blob_a, blob_b = b"\xff\xd8\xff\xe0jpg-a", b"\xff\xd8\xff\xe0jpg-b"
+    respx.get(OSS_A).mock(return_value=httpx.Response(200, content=blob_a))
+    respx.get(OSS_B).mock(return_value=httpx.Response(200, content=blob_b))
+
+    client = _primed_client()
+    try:
+        async with SessionLocal() as session:
+            session.add(MomentsPost(
+                id="p1", wechat_id="wx1", image_urls=[OSS_A, OSS_B],
+            ))
+            await session.commit()
+
+        async with SessionLocal() as session:
+            attempted, localized = await download_moments_images(client, store, session)
+            await session.commit()
+
+        assert (attempted, localized) == (2, 2)
+        for blob, url in ((blob_a, OSS_A), (blob_b, OSS_B)):
+            sha = hashlib.sha256(blob).hexdigest()
+            assert (tmp_path / sha[:2] / f"{sha}.jpg").is_file()
+            async with SessionLocal() as session:
+                media = (await session.execute(
+                    sa.select(Media).where(Media.media_hash == sha)
+                )).scalar_one()
+                assert media.media_type == "image"
+                assert media.channel == "wechat"
+                assert media.original_url == url
+                assert media.mime_type == "image/jpeg"
+    finally:
+        await client.aclose()
+        await engine.dispose()
+
+
+@respx.mock
+async def test_download_moments_images_idempotent_via_original_url(tmp_path):
+    """MomentsPost 无 media_hash 回链列 → 已有同 original_url 的 Media 行即跳过
+    (第二轮 attempted=0,不触网)。"""
+    store = ObjectStore(tmp_path)
+    engine = await _make_engine()
+    SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
+
+    respx.get(OSS_A).mock(return_value=httpx.Response(200, content=b"blob"))
+
+    client = _primed_client()
+    try:
+        async with SessionLocal() as session:
+            session.add(MomentsPost(
+                id="p1", wechat_id="wx1", image_urls=[OSS_A],
+            ))
+            await session.commit()
+
+        async with SessionLocal() as session:
+            first = await download_moments_images(client, store, session)
+            await session.commit()
+        assert first == (1, 1)
+
+        async with SessionLocal() as session:
+            second = await download_moments_images(client, store, session)
+            await session.commit()
+        assert second == (0, 0)
+    finally:
+        await client.aclose()
+        await engine.dispose()
+
+
+@respx.mock
+async def test_download_moments_images_skips_non_http_and_survives_failure(tmp_path):
+    """非 http 条目跳过;下载 500 只计数不抛(下轮重试)。"""
+    store = ObjectStore(tmp_path)
+    engine = await _make_engine()
+    SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
+
+    respx.get(OSS_A).mock(return_value=httpx.Response(500))
+
+    client = _primed_client()
+    try:
+        async with SessionLocal() as session:
+            session.add(MomentsPost(
+                id="p1", wechat_id="wx1", image_urls=[OSS_A, "not-a-url"],
+            ))
+            await session.commit()
+
+        async with SessionLocal() as session:
+            attempted, localized = await download_moments_images(client, store, session)
+            await session.commit()
+
+        assert (attempted, localized) == (1, 0)
+        async with SessionLocal() as session:
+            assert (await session.execute(sa.select(Media))).scalars().all() == []
+    finally:
+        await client.aclose()
+        await engine.dispose()

yunke-server/tests/test_sync_scheduler.py

@@ -559,13 +559,12 @@ async def test_safe_page_run_full_invokes_bash_full_sync(engine, monkeypatch):
 
 
 async def test_safe_page_run_messages_builds_scrape_command(engine, monkeypatch):
-    """messages job = python scripts/scrape_via_page.py --entity messages。"""
+    """messages job = friends→wa-friends→四类消息→moments 浅扫的命令序列。"""
     sched = SyncScheduler(engine=engine, max_conversations_per_run=123)
-    captured: dict = {}
+    commands: list[list[str]] = []
 
     async def fake_exec(*args, cwd=None, env=None):
-        captured["args"] = list(args)
+        commands.append(list(args))
-        captured["env"] = env
         return _FakeProc(0)
 
     monkeypatch.setattr(scheduler_module.asyncio, "create_subprocess_exec", fake_exec)
@@ -573,13 +572,25 @@ async def test_safe_page_run_messages_builds_scrape_command(engine, monkeypatch)
     await sched._safe_page_run("messages")
 
     # 2026-08-21: messages job 是 friends→wa-friends→四类消息的命令序列,
-    # 每条都调 scrape_via_page.py;最后一次执行的命令参数可断言
+    # 每条都调 scrape_via_page.py;2026-08-25 末尾追加 moments 浅扫
-    args = captured["args"]
+    assert len(commands) == 7
-    assert args[1].endswith("scrape_via_page.py")
+    for args in commands:
-    i = args.index("--entity")
+        assert args[1].endswith("scrape_via_page.py")
-    assert args[i + 1] in ("messages", "wa-messages", "group-messages", "wa-group-messages")
+
-    j = args.index("--max-conversations")
+    def entity_of(args: list[str]) -> str:
-    assert args[j + 1] == "99999"
+        return args[args.index("--entity") + 1]
+
+    assert entity_of(commands[0]) == "friends"
+    assert entity_of(commands[1]) == "wa-friends"
+    for args in commands[2:6]:
+        assert entity_of(args) in (
+            "messages", "wa-messages", "group-messages", "wa-group-messages",
+        )
+        assert args[args.index("--max-conversations") + 1] == "99999"
+    # moments 浅扫在序列末尾:每员工最多 3 页
+    moments_cmd = commands[-1]
+    assert entity_of(moments_cmd) == "moments"
+    assert moments_cmd[moments_cmd.index("--max-pages") + 1] == "3"
 
 
 async def test_safe_page_run_nonzero_rc_is_logged_not_raised(engine, monkeypatch, capsys):