diff --git a/yunke-server/app/scraper/media.py b/yunke-server/app/scraper/media.py index ba83044..4274bc2 100644 --- a/yunke-server/app/scraper/media.py +++ b/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 diff --git a/yunke-server/app/sync/scheduler.py b/yunke-server/app/sync/scheduler.py index ecfa056..8980c2b 100644 --- a/yunke-server/app/sync/scheduler.py +++ b/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):强制全量翻历史,对齐增量模式抓不到的 diff --git a/yunke-server/scripts/full_sync_all.sh b/yunke-server/scripts/full_sync_all.sh index 2f990ab..4f4fbba 100644 --- a/yunke-server/scripts/full_sync_all.sh +++ b/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}') diff --git a/yunke-server/scripts/scrape_via_page.py b/yunke-server/scripts/scrape_via_page.py index 1faa431..b5b3538 100644 --- a/yunke-server/scripts/scrape_via_page.py +++ b/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) diff --git a/yunke-server/tests/test_scrape_moments.py b/yunke-server/tests/test_scrape_moments.py new file mode 100644 index 0000000..4610efe --- /dev/null +++ b/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() diff --git a/yunke-server/tests/test_sync_scheduler.py b/yunke-server/tests/test_sync_scheduler.py index ecb489b..3002153 100644 --- a/yunke-server/tests/test_sync_scheduler.py +++ b/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) - captured["env"] = env + commands.append(list(args)) 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;最后一次执行的命令参数可断言 - args = captured["args"] - assert args[1].endswith("scrape_via_page.py") - i = args.index("--entity") - assert args[i + 1] in ("messages", "wa-messages", "group-messages", "wa-group-messages") - j = args.index("--max-conversations") - assert args[j + 1] == "99999" + # 每条都调 scrape_via_page.py;2026-08-25 末尾追加 moments 浅扫 + assert len(commands) == 7 + for args in commands: + assert args[1].endswith("scrape_via_page.py") + + def entity_of(args: list[str]) -> str: + 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):