yunke/yunke-server/scripts/scrape_via_page.py

#!/usr/bin/env python
"""页面浏览器抓取器 — 当 API 接口被风控时，用 Playwright 浏览器方式抓取数据。

原理：SPA 页面加载时会自己调用 friend_list/es 等接口渲染表格。
我们拦截页面的网络响应，获取原始 JSON —— 效果等同 API 调用，
但因为是浏览器页面上下文（非脚本直调），不触发风控。

用法：
    cd yunke-server
    DB_URL="sqlite+aiosqlite:///./smoke.db" .venv/Scripts/python.exe scripts/scrape_via_page.py --entity friends
    DB_URL="sqlite+aiosqlite:///./smoke.db" .venv/Scripts/python.exe scripts/scrape_via_page.py --entity groups
    DB_URL="sqlite+aiosqlite:///./smoke.db" .venv/Scripts/python.exe scripts/scrape_via_page.py --entity messages
    DB_URL="sqlite+aiosqlite:///./smoke.db" .venv/Scripts/python.exe scripts/scrape_via_page.py --entity accounts
    DB_URL="sqlite+aiosqlite:///./smoke.db" .venv/Scripts/python.exe scripts/scrape_via_page.py --entity wa-accounts
    DB_URL="sqlite+aiosqlite:///./smoke.db" .venv/Scripts/python.exe scripts/scrape_via_page.py --entity wa-friends
    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 读取。
"""
from __future__ import annotations

import argparse
import asyncio
import json
import logging
import os
import re
from typing import Any
from urllib.parse import urlparse

logger = logging.getLogger("scrape_via_page")
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")

YUNKE_BASE = "https://crm2.yunkecn.com"

# API endpoints we intercept (the SPA calls these to render tables)
INTERCEPT_PATHS = {
    "friend_list": "/wechatweb/wxFriends/friend_list/es",  # friends + groups (friendTypeYunke)
    "wxUser_list": "/wechatweb/wxUser/list",               # wechat accounts
    "chatroom": "/wechatweb/chatroom/getChatroomList",      # legacy groups
    "voice_list": "/wechatweb/call/list",                    # voice clips
    "call_list": "/wechatweb/callRecord/queryWeChatVoiceList",  # voice calls
    "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",  # 员工朋友圈
}


def parse_cookies(cookie_str: str, domain: str = "crm2.yunkecn.com") -> list[dict]:
    """Parse a cookie string like 'user=xxx; userToken=yyy' into Playwright cookie dicts."""
    cookies = []
    for pair in cookie_str.split(";"):
        pair = pair.strip()
        if "=" not in pair:
            continue
        name, value = pair.split("=", 1)
        cookies.append({
            "name": name.strip(),
            "value": value.strip(),
            "domain": domain,
            "path": "/",
        })
    return cookies


class PageScraper:
    """Navigates 云客 pages in a browser, intercepts API responses, extracts JSON."""

    def __init__(self, cookie_str: str, headless: bool = True):
        self.cookie_str = cookie_str
        self.headless = headless
        self.browser = None
        self.context = None
        self.page = None
        self._captured: list[dict] = []
        self._capture_path: str | None = None

    async def __aenter__(self):
        from playwright.async_api import async_playwright
        self._pw = await async_playwright().start()
        self.browser = await self._pw.chromium.launch(headless=self.headless)
        self.context = await self.browser.new_context(
            user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/150.0.0.0 Safari/537.36",
            viewport={"width": 1920, "height": 1080},
        )
        # Set cookies
        cookies = parse_cookies(self.cookie_str)
        await self.context.add_cookies(cookies)
        self.page = await self.context.new_page()

        # Intercept API responses
        self.page.on("response", self._on_response)
        return self

    async def __aexit__(self, *args):
        if self.browser:
            await self.browser.close()
        if self._pw:
            await self._pw.stop()

    async def _on_response(self, response):
        """Capture API JSON responses that match our intercept paths."""
        url = response.url
        path = urlparse(url).path
        if self._capture_path and path == self._capture_path:
            try:
                body = await response.json()
                self._captured.append({"url": url, "status": response.status, "body": body})
                logger.info(f"  captured {path}: status={response.status}, keys={list(body.keys()) if isinstance(body, dict) else 'N/A'}")
            except Exception as e:
                logger.warning(f"  capture failed for {path}: {e}")

    def start_capture(self, api_path: str):
        """Start capturing responses for a specific API path."""
        self._capture_path = api_path
        self._captured.clear()

    def get_captured(self) -> list[dict]:
        """Return captured responses and stop capturing."""
        self._capture_path = None
        return self._captured

    async def navigate_and_wait(self, url: str, wait_selector: str = ".ant-table-row", timeout: int = 15000):
        """Navigate to a URL and wait for table rows to appear."""
        await self.page.goto(url, wait_until="domcontentloaded", timeout=timeout)
        try:
            await self.page.wait_for_selector(wait_selector, timeout=timeout)
        except Exception:
            pass  # might be empty page
        await self.page.wait_for_timeout(2000)  # extra settle time

    async def get_total_pages(self) -> int:
        """Extract total page count from Ant Design pagination."""
        try:
            total_text = await self.page.evaluate("""() => {
                const el = document.querySelector('.ant-pagination-total-text');
                return el ? el.textContent : '';
            }""")
            if total_text:
                import re
                m = re.search(r'(\d+)', total_text)
                if m:
                    total = int(m.group(1))
                    page_size = 10  # 云客 default
                    return (total + page_size - 1) // page_size
        except Exception:
            pass
        return 1

    async def click_next_page(self) -> bool:
        """Click the 'next page' button. Returns True if successful."""
        try:
            next_btn = self.page.locator('.ant-pagination-next:not(.ant-pagination-disabled)')
            if await next_btn.count() > 0:
                await next_btn.click()
                await self.page.wait_for_timeout(2000)  # wait for data load
                return True
        except Exception:
            pass
        return False

    async def scrape_entity_pages(
        self,
        list_url: str,
        api_path: str,
        max_pages: int = 100,
    ) -> list[dict]:
        """Navigate to a list page, paginate through all pages, capturing API responses.

        Returns a list of captured JSON response bodies (one per page).
        """
        self.start_capture(api_path)
        await self.navigate_and_wait(list_url)
        await self.page.wait_for_timeout(1000)  # let API call complete

        all_responses = list(self.get_captured())

        total_pages = await self.get_total_pages()
        logger.info(f"  total pages: {min(total_pages, max_pages)}")

        page_num = 1
        while page_num < min(total_pages, max_pages):
            self.start_capture(api_path)
            clicked = await self.click_next_page()
            if not clicked:
                break
            await self.page.wait_for_timeout(1500)
            all_responses.extend(self.get_captured())
            page_num += 1
            if page_num % 10 == 0:
                logger.info(f"  page {page_num}/{min(total_pages, max_pages)}")

        logger.info(f"  captured {len(all_responses)} API responses across {page_num} pages")
        return all_responses

    async def get_employee_links(self) -> list[dict]:
        """Extract employee wxIds + friend-count links from the staff list page."""
        await self.navigate_and_wait(f"{YUNKE_BASE}/cms/wechat/wechatlist")

        employees = await self.page.evaluate("""() => {
            const rows = document.querySelectorAll('.ant-table-row');
            const result = [];
            for (const tr of rows) {
                // Find the friend-count link (href contains wechatdetail/1/)
                const friendLink = tr.querySelector('a[href*="wechatdetail/1/"]');
                const groupLink = tr.querySelector('a[href*="wechatdetail/2/"]');
                if (friendLink) {
                    const href = friendLink.getAttribute('href');
                    const wxId = href.match(/wechatdetail\\/1\\/([^?]+)/)?.[1];
                    const friendCount = parseInt(friendLink.textContent) || 0;
                    const groupHref = groupLink?.getAttribute('href');
                    const groupCount = parseInt(groupLink?.textContent) || 0;
                    result.push({ wxId, friendCount, groupCount, friendHref: href, groupHref });
                }
            }
            return result;
        }""")

        # Paginate through ALL staff list pages
        all_employees = list(employees)
        page = 1
        while await self.click_next_page():
            page += 1
            await self.page.wait_for_timeout(2000)
            more = await self.page.evaluate("""() => {
                const rows = document.querySelectorAll('.ant-table-row');
                const result = [];
                for (const tr of rows) {
                    const friendLink = tr.querySelector('a[href*="wechatdetail/1/"]');
                    const groupLink = tr.querySelector('a[href*="wechatdetail/2/"]');
                    if (friendLink) {
                        const href = friendLink.getAttribute('href');
                        const wxId = href.match(/wechatdetail\\/1\\/([^?]+)/)?.[1];
                        const friendCount = parseInt(friendLink.textContent) || 0;
                        const groupHref = groupLink?.getAttribute('href');
                        const groupCount = parseInt(groupLink?.textContent) || 0;
                        result.push({ wxId, friendCount, groupCount, friendHref: href, groupHref });
                    }
                }
                return result;
            }""")
            all_employees.extend(more)
            logger.info(f"  staff page {page}: +{len(more)} employees (total: {len(all_employees)})")

        # Dedup by wxId
        seen = set()
        deduped = []
        for emp in all_employees:
            if emp["wxId"] not in seen:
                seen.add(emp["wxId"])
                deduped.append(emp)

        logger.info(f"Found {len(deduped)} unique employees with friend links ({page} pages)")
        return deduped


async def scrape_friends(cookie_str: str, max_employees: int = 100):
    """微信客户:浏览器直发 friend_list/es,**逐员工**(userId 过滤)翻页。

    部门级全量分页排序不稳定(实测:云客声明 8691 vs 到手 distinct 6563,
    8682 行里 2119 行跨页重复挤占页容量漏抓真客户,2026-08-20);逐员工范围
    小漂移可忽略,且与云客后台「按员工看客户列表」口径一致。员工清单来自
    本地 wechat_account(user_id 列)。
    """
    from app.db.base import async_engine, async_session_maker
    from sqlalchemy import text as sa_text
    from app.scraper.wechat import _normalize
    from app.storage.repos import prune_customer_links, upsert_customer

    dept = await _root_dept_id()
    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()][:max_employees]

    import datetime as _dt
    today = _dt.date.today().isoformat()

    total_fetched = 0
    total_upserted = 0
    async with PageScraper(cookie_str) as scraper:
        page = await _open_fetch_page(scraper)

        # ---- 阶段1:逐员工全量(保底完整性;该路径 lastChatTime 有滞后) ----
        for i, wx_id in enumerate(employees):
            # 真机抓包(2026-08-20 云客客户列表页)确认的过滤参数:wechatId(非
            # userId!)+ymd 窗口+friendTypeYunke=1+isDel=0
            base_body = {
                "ymdStart": "0000-00-00",
                "ymdEnd": today,
                "wechatId": wx_id,
                "salesname": "",
                "friendTypeYunke": "1",
                "isDel": 0,
            }
            # total-aware 翻页:该端点分页不稳(中途偶发短页),按声明 total
            # 翻满或连续空页才停;跨页重复行 upsert 幂等无害。
            rows: list[dict] = []
            declared: int | None = None
            empty_streak = 0
            page_num = 1
            while page_num <= 400:
                resp = await _browser_fetch(
                    page, INTERCEPT_PATHS["friend_list"],
                    json_body={**base_body, "pageNum": page_num, "pageSize": 50},
                )
                basic = ((resp.get("data") or {}).get("basicResult") or {})
                declared = basic.get("total") or declared
                page_rows = basic.get("data") or []
                if not page_rows:
                    empty_streak += 1
                    if empty_streak >= 2 or (declared and len(rows) >= declared):
                        break
                else:
                    empty_streak = 0
                    rows.extend(page_rows)
                    if declared and len(rows) >= declared:
                        break
                page_num += 1
                await asyncio.sleep(1)

            records = []
            for row in rows:
                rec = _normalize(row)
                if rec.get("friend_id"):
                    records.append(rec)
            total_fetched += len(records)
            if records:
                await _upsert_all(upsert_customer, records)
                total_upserted += len(records)
            # 对账(2026-09-08):云客 isDel=0 返回当前好友;名单取满(>=声明
            # total)才可信,删掉该账号名下已不在名单里的 link(转移/删好友)。
            # 半份名单宁可不对账,防误删。
            pruned = 0
            if declared is not None and len(rows) >= declared:
                keep_ids = [r.get("friendWxId") for r in rows if r.get("friendWxId")]
                async with async_session_maker() as session:
                    pruned = await prune_customer_links(
                        session, channel="wechat", owner_account_id=wx_id,
                        keep_friend_ids=keep_ids,
                    )
                    await session.commit()
            logger.info(
                f"  [{i+1}/{len(employees)}] {wx_id[:16]}..: declared={declared} "
                f"rows={len(rows)} pruned={pruned} (total fetched {total_fetched})"
            )
            await asyncio.sleep(2 + (i % 3))

        # ---- 阶段2:全局实时刷新(云客列表页的参数集,跨员工按最近沟通实时排序;
        # 阶段1的 lastChatTime 有滞后,这里用实时值覆盖,保证本地排序与云客第一页一致) ----
        global_body = {
            "friendName": "",
            "friendRemark": "",
            "saleName": "",
            "friendPhone": "",
            "realName": "",
            "emotionOrder": "",
            "emotionStatus": "",
            "tagGroupId": "",
            "isDel": 0,
            "isOnJob": 1,
            "showSale": "1",
            "friendTab": "all",
            "mergeSearchType": "saleName",
            "mergeSearchLabel": "",
            "ymdStart": today,
            "ymdEnd": today,
            "friendId": "",
            "departmentId": dept or "",
            "userId": "",
        }
        realtime_rows: list[dict] = []
        for page_num in range(1, 41):  # 前 40 页 x50 = 2000 行,覆盖全部近期活跃客户
            resp = await _browser_fetch(
                page, INTERCEPT_PATHS["friend_list"],
                json_body={**global_body, "pageNum": page_num, "pageSize": 50},
            )
            basic = ((resp.get("data") or {}).get("basicResult") or {})
            page_rows = basic.get("data") or []
            if not page_rows:
                break
            realtime_rows.extend(page_rows)
            if len(page_rows) < 50:
                break
            await asyncio.sleep(1)
        # 只刷 last_chat_time 且只增不减(全局结果里也含旧 ES 副本,后写会
        # 把实时值覆盖回旧值——max 条件天然免疫,2026-08-20 实测踩坑)
        from app.db.base import async_session_maker as _asm
        from sqlalchemy import text as _st

        updated = 0
        async with _asm() as session:
            for row in realtime_rows:
                fid, ts = row.get("friendWxId"), row.get("lastChatTime")
                if not fid or not ts:
                    continue
                r = await session.execute(_st(
                    "UPDATE customer SET last_chat_time = :ts "
                    "WHERE channel='wechat' AND owner_account_id = :o AND friend_id = :f "
                    "AND (last_chat_time IS NULL OR last_chat_time < :ts)"
                ), {"ts": int(ts), "o": row.get("wxId"), "f": fid})
                updated += r.rowcount
            await session.commit()
        logger.info(f"  实时刷新: {len(realtime_rows)} 行(全局实时排序), last_chat_time 更新 {updated} 条")

    logger.info(f"DONE: friends fetched={total_fetched}, upserted={total_upserted}")
    return total_fetched, total_upserted


async def scrape_groups(cookie_str: str, max_employees: int = 100):
    """微信群:浏览器直发 friend_list/es,**逐员工**(wechatId 过滤)翻页。

    与好友共用端点,仅 friendTypeYunke=2;部门级分页不稳(见 scrape_friends
    三连修,2026-08-20),同模式逐员工 + total-aware 翻页。"""
    import datetime as _dt

    from app.db.base import async_engine
    from sqlalchemy import text as sa_text
    from app.scraper.wechat import _normalize_chat_group
    from app.storage.repos import upsert_chat_group

    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()][:max_employees]

    today = _dt.date.today().isoformat()
    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):
            base_body = {
                "ymdStart": "0000-00-00",
                "ymdEnd": today,
                "wechatId": wx_id,
                "salesname": "",
                "friendTypeYunke": "2",
                "isDel": 0,
            }
            rows: list[dict] = []
            declared: int | None = None
            empty_streak = 0
            page_num = 1
            while page_num <= 200:
                resp = await _browser_fetch(
                    page, INTERCEPT_PATHS["friend_list"],
                    json_body={**base_body, "pageNum": page_num, "pageSize": 50},
                )
                basic = ((resp.get("data") or {}).get("basicResult") or {})
                declared = basic.get("total") or declared
                page_rows = basic.get("data") or []
                if not page_rows:
                    empty_streak += 1
                    if empty_streak >= 2 or (declared and len(rows) >= declared):
                        break
                else:
                    empty_streak = 0
                    rows.extend(page_rows)
                    if declared and len(rows) >= declared:
                        break
                page_num += 1
                await asyncio.sleep(1)

            records = []
            for row in rows:
                rec = _normalize_chat_group(row)
                if rec.get("room_id"):
                    records.append(rec)
            total_fetched += len(records)
            if records:
                await _upsert_all(upsert_chat_group, records)
                total_upserted += len(records)
            if (i + 1) % 10 == 0:
                logger.info(f"  [{i+1}/{len(employees)}] employees done, groups total: {total_fetched}")
            await asyncio.sleep(2 + (i % 3))

    logger.info(f"DONE: groups fetched={total_fetched}, upserted={total_upserted}")
    return total_fetched, total_upserted


async def paginate_chats(fetch_page, end_ms: int, max_pages: int = 200, sleep=None):
    """按 hasLast/end 游标翻页 chats 类接口（微信 msg/chats 与 WA chatRecords 同协议）。

    fetch_page(end_ms) -> 响应 dict；逐页 yield data 部分（dict）。
    终止：data 非 dict / hasLast 假 / end 非数值 / 达到 max_pages 页上限（防死循环）。
    """
    if sleep is None:
        sleep = asyncio.sleep
    pages = 0
    while True:
        body = await fetch_page(end_ms)
        pages += 1
        data = body.get("data") if isinstance(body, dict) else None
        if not isinstance(data, dict):
            return
        yield data
        if not data.get("hasLast"):
            return
        end = data.get("end")
        if not isinstance(end, (int, float)):
            return
        end_ms = int(end) - 1
        if pages >= max_pages:
            logger.warning(f"paginate_chats: page cap {max_pages} hit, stop paging")
            return
        await sleep(1)


async def _discover_conversations(channel: str, voice_channel: str) -> list[tuple[str, str]]:
    """Conversation-pair discovery for the message scrapers (wechat + WA).

    Sources (union, dedup):
    * customer table — ALL pairs. The old ``last_chat_time IS NOT NULL`` filter
      skipped customers whose relationship is voice-call-only (NULL last_chat)
      and thus their chat history was never scraped → empty chat pages
      (diagnosed 2026-08-18).
    * voice-call pairs ``(weChatId, talker)`` from phone_call raw_json — peers
      that had calls but have NO customer row at all (deleted friends).

    Order: pairs with no messages yet FIRST (new coverage survives an
    interruption), then the rest by customer last_chat_time DESC.
    """
    import json as _json

    from app.db.base import async_engine
    from sqlalchemy import text as sa_text

    async with async_engine.connect() as conn:
        cust = (
            await conn.execute(sa_text(
                "SELECT owner_account_id, friend_id FROM customer "
                "WHERE channel=:ch AND owner_account_id IS NOT NULL AND friend_id IS NOT NULL "
                "ORDER BY last_chat_time DESC"
            ), {"ch": channel})
        ).fetchall()
        voice = (
            await conn.execute(sa_text(
                "SELECT raw_json FROM phone_call WHERE channel=:vc"
            ), {"vc": voice_channel})
        ).fetchall()
        with_msgs = (
            await conn.execute(sa_text(
                "SELECT DISTINCT owner_account_id, talker_id FROM message "
                "WHERE channel=:ch AND (room_id IS NULL OR room_id='')"
            ), {"ch": channel})
        ).fetchall()

    have = {(r[0], r[1]) for r in with_msgs}
    seen: set[tuple[str, str]] = set()
    fresh: list[tuple[str, str]] = []
    known: list[tuple[str, str]] = []

    def _add(key: tuple[str, str]) -> None:
        if key in seen:
            return
        seen.add(key)
        (fresh if key not in have else known).append(key)

    for owner, friend in cust:
        _add((owner, friend))
    voice_only = 0
    for (rj,) in voice:
        try:
            d = _json.loads(rj)
        except (TypeError, ValueError):
            continue
        owner, talker = d.get("weChatId"), d.get("talker")
        if not owner or not talker or str(talker).endswith("@chatroom"):
            continue
        before = len(seen)
        _add((str(owner), str(talker)))
        if len(seen) > before and (str(owner), str(talker)) not in cust:
            voice_only += 1

    logger.info(
        f"discovery[{channel}]: {len(cust)} customer pairs + {voice_only} voice-only pairs, "
        f"{len(fresh)} unscanned (first) / {len(known)} already have messages"
    )
    return fresh + known


async def _room_watermarks() -> dict[str, int]:
    """每群本地最新消息时间(ms)——群消息增量水位(按 room_id)。"""
    from app.db.base import async_engine
    from sqlalchemy import text as sa_text

    async with async_engine.connect() as conn:
        result = await conn.execute(sa_text(
            "SELECT room_id, max(timestamp) FROM message "
            "WHERE room_id IS NOT NULL AND room_id != '' "
            "GROUP BY room_id"
        ))
        return {r[0]: r[1] or 0 for r in result.fetchall()}


async def _conversation_watermarks(channel: str) -> dict[tuple[str, str], int]:
    """每对话的本地最新消息时间(ms)——增量同步水位。"""
    from app.db.base import async_engine
    from sqlalchemy import text as sa_text

    async with async_engine.connect() as conn:
        result = await conn.execute(sa_text(
            "SELECT owner_account_id, talker_id, max(timestamp) FROM message "
            "WHERE channel=:ch AND (room_id IS NULL OR room_id='') "
            "AND talker_id IS NOT NULL AND talker_id != '' "
            "GROUP BY owner_account_id, talker_id"
        ), {"ch": channel})
        return {(r[0], r[1]): r[2] or 0 for r in result.fetchall()}


async def _customer_activity(channel: str) -> dict[tuple[str, str], int]:
    """客户表 last_chat_time(4h 增量的 friends 实时刷新)——判断对话有无新消息。"""
    from app.db.base import async_engine
    from sqlalchemy import text as sa_text

    async with async_engine.connect() as conn:
        result = await conn.execute(sa_text(
            "SELECT owner_account_id, friend_id, last_chat_time FROM customer "
            "WHERE channel=:ch AND owner_account_id IS NOT NULL AND friend_id IS NOT NULL"
        ), {"ch": channel})
        return {(r[0], r[1]): r[2] or 0 for r in result.fetchall()}


def _page_min_ts(data: dict) -> int:
    """页内最早消息时间(0 当无有效 ts)。"""
    ts_list = [m.get("timestamp") for m in (data.get("messages") or []) if isinstance(m, dict) and m.get("timestamp")]
    return min(ts_list) if ts_list else 0


async def scrape_messages(cookie_str: str, max_conversations: int = 100, full_history: bool = False):
    """Scrape wechat messages via browser fetch POST /wechatweb/msg/chats.

    **增量同步(默认)**:每对话水位 = 本地 max(timestamp);请求带
    ``start=水位`` 只拉新消息,翻页遇到 ≤水位的消息即停(已同步区)。
    客户表 last_chat_time(friends 实时刷新)≤ 水位的对话**整跳过**(0 请求)
    ——活跃对话才访问。``full_history=True`` 强制全量(首次/修复用)。
    """
    from sqlalchemy import text as sa_text

    from app.db.base import async_session_maker
    from app.storage.repos import upsert_messages

    conversations = await _discover_conversations("wechat", "voice_call")
    watermarks = await _conversation_watermarks("wechat")
    cust_activity = await _customer_activity("wechat")

    # 增量模式:云客侧无新活动的对话跳过(0 请求)——速度来源
    skipped = 0
    if not full_history:
        kept = []
        for pair in conversations:
            wm = watermarks.get(pair, 0)
            act = cust_activity.get(pair, 0)
            # 注意:不能用 +60s 缓冲——若新消息恰好落在水位后 60s 内,
            # act 永远 ≤ wm+60s → 每轮都跳过 → 该消息**永久漏掉**。
            # act==wm 表示云客最新消息==我们最新消息,精确相等才跳。
            if wm > 0 and act > 0 and act <= wm:
                skipped += 1
            else:
                kept.append(pair)
        conversations = kept
    logger.info(
        f"Discovered conversations: {len(conversations)} to visit, {skipped} skipped (no new activity)"
        + (" [FULL HISTORY]" if full_history else " [incremental]")
    )

    total_fetched = 0
    total_upserted = 0
    total_pages = 0

    async with PageScraper(cookie_str) as scraper:
        # Just load the base URL to establish cookie context — no need for specific page
        # The fetch() calls work from any page (even login redirect) as long as cookies are set
        fetch_page = await scraper.context.new_page()
        await fetch_page.goto(f"{YUNKE_BASE}/", wait_until="domcontentloaded", timeout=15000)
        await fetch_page.wait_for_timeout(2000)

        page = fetch_page  # use the stable page for all evaluate() calls

        for i, (owner_id, friend_id) in enumerate(conversations[:max_conversations]):
            records = []
            pages = 0
            try:
                # Use browser fetch() to call the message API from the page context
                # (the browser sends it with proper cookies/headers → no 风控)
                # NOTE: msg/chats uses FORM-ENCODED + cursor params (start/end, not pageNum)
                import time as _time
                now_ms = int(_time.time() * 1000)

                watermark = 0 if full_history else watermarks.get((owner_id, friend_id), 0)

                async def _fetch(end_ms: int, _owner=owner_id, _friend=friend_id, _wm=watermark):
                    return await page.evaluate("""async ([ownerId, friendId, startMs, endMs]) => {
                        const params = new URLSearchParams();
                        params.append('wechatId', ownerId);
                        params.append('friendId', friendId);
                        params.append('types', '');
                        params.append('start', String(startMs));
                        params.append('end', String(endMs));
                        const resp = await fetch('/wechatweb/msg/chats', {
                            method: 'POST',
                            headers: { 'Content-Type': 'application/x-www-form-urlencoded' },
                            body: params.toString(),
                        });
                        return await resp.json();
                    }""", [_owner, _friend, _wm, end_ms])

                # 翻页取完整历史：hasLast 真 → 用 data.end-1 继续向前翻（页间 sleep 1s）
                async for data in paginate_chats(_fetch, now_ms):
                    pages += 1
                    # 增量触底停:页内出现 ≤水位 的消息 = 已同步区,不再向前翻
                    if watermark and _page_min_ts(data) and _page_min_ts(data) <= watermark:
                        break
                    # Debug: log first response structure
                    if i == 0 and pages == 1:
                        logger.info(f"  DEBUG first response keys: {list(data.keys())}")
                        msgs0 = data.get("messages")
                        logger.info(f"  DEBUG messages type: {type(msgs0)}, len: {len(msgs0) if isinstance(msgs0, list) else 'N/A'}")
                    rows = data.get("messages") or []
                    for row in rows:
                        if not isinstance(row, dict):
                            continue
                        # 跳过 type=-2 日期分隔线（非真实消息，仅显示日期）
                        if str(row.get("type")) == "-2":
                            continue
                        # Normalize to Message shape
                        ts_raw = row.get("createTime") or row.get("timestamp")
                        try:
                            ts = int(str(ts_raw)) if ts_raw else None
                        except (ValueError, TypeError):
                            ts = None
                        records.append({
                            "msg_id": str(row.get("msgId") or row.get("msgSvrId") or row.get("id") or f"{owner_id}:{friend_id}:{i}:{len(records)}"),
                            "channel": "wechat",
                            "owner_account_id": owner_id,
                            # 用发送的 friend_id（与客户表一致）优先，API 返回的 talker 可能脱敏
                            "talker_id": friend_id or row.get("talker") or row.get("oriTalker"),
                            "room_id": None,
                            "is_send": (
                                bool(row["isSend"]) if row.get("isSend") is not None
                                else (bool(row["mine"]) if row.get("mine") is not None else None)
                            ),
                            "type": _normalize_msg_type(row.get("type"), row=row),
                            "type_raw": row.get("type"),
                            "text": row.get("content") or row.get("text") or "",
                            "file_url": row.get("file") or row.get("fileUrl"),
                            # fileTh/fileHd 常在 file 为空时仍带完整 OSS URL——映射进列,媒体下载器才够得着
                            "file_th_url": row.get("fileTh"),
                            "file_hd_url": row.get("fileHd"),
                            "timestamp": ts,
                            "sender_name": row.get("name") or row.get("senderName"),
                            "raw_json": row,
                        })

            except Exception as e:
                logger.warning(f"  [{i+1}] {owner_id[:12]}:{friend_id[:12]} failed: {e}")
                # Recreate the fetch page if context was destroyed.
                # A network blip / yunke outage also breaks this recovery goto —
                # retry patiently (30 × 2min = 1h) so a longer outage doesn't
                # kill the whole run (crashed twice on 2026-08-18 this way,
                # outages ~3h apart, ~1.5h into each run).
                for attempt in range(30):
                    try:
                        try:
                            await fetch_page.close()
                        except Exception:
                            pass
                        fetch_page = await scraper.context.new_page()
                        await fetch_page.goto(f"{YUNKE_BASE}/cms/home/statistics", wait_until="networkidle", timeout=20000)
                        await fetch_page.wait_for_timeout(1000)
                        page = fetch_page
                        break
                    except Exception as rec:
                        logger.warning(f"  recovery {attempt + 1}/30 failed: {rec}")
                        if attempt == 29:
                            raise
                        await asyncio.sleep(120)
                continue

            total_fetched += len(records)
            total_pages += pages

            if records:
                async with async_session_maker() as session:
                    await upsert_messages(session, records)
                    # 时效:用本对话最新消息时间刷新 customer.last_chat_time
                    # (云客列表按最近沟通实时排序;friends 快照每日一次,消息增量每4h)
                    max_ts = max((r.get("timestamp") or 0) for r in records)
                    if max_ts:
                        await session.execute(sa_text(
                            "UPDATE customer SET last_chat_time = :ts "
                            "WHERE channel='wechat' AND owner_account_id = :o AND friend_id = :f "
                            "AND (last_chat_time IS NULL OR last_chat_time < :ts)"
                        ), {"ts": max_ts, "o": owner_id, "f": friend_id})
                    await session.commit()
                total_upserted += len(records)

            if (i + 1) % 10 == 0:
                logger.info(f"  [{i+1}/{min(max_conversations, len(conversations))}] msgs total: {total_fetched}, pages: {total_pages}")

            # Human-like delay
            await asyncio.sleep(2 + (i % 3))

    logger.info(f"DONE: messages fetched={total_fetched}, upserted={total_upserted}, pages={total_pages}")
    return total_fetched, total_upserted


def _normalize_group_message(row: dict, owner_id: str, room_id: str) -> dict | None:
    """msg/chats 群消息行 → Message 形状（friendId=roomId 时端点返回的行）。

    群行与 1:1 行同构（真机确认 2026-08-19），差异：talker=发送者 wxid、
    roomid=群 id、msgId 为 null → 用 owner:room:createTime:talker 合成稳定
    主键，重扫同一条消息得到同一 id（幂等 upsert 前提）。type=-2 日期分隔
    线丢弃。
    """
    if str(row.get("type")) == "-2":
        return None
    ts_raw = row.get("createTime") or row.get("timestamp")
    try:
        ts = int(str(ts_raw)) if ts_raw else None
    except (ValueError, TypeError):
        ts = None
    talker = row.get("talker") or ""
    synth = f"{owner_id}:{room_id}:{ts}:{talker}"
    return {
        "msg_id": str(row.get("msgId") or row.get("msgSvrId") or row.get("id") or synth),
        "channel": "wechat",
        "owner_account_id": owner_id,
        "talker_id": talker or None,
        "room_id": row.get("roomid") or room_id,
        "is_send": (
                                bool(row["isSend"]) if row.get("isSend") is not None
                                else (bool(row["mine"]) if row.get("mine") is not None else None)
                            ),
        "type": _normalize_msg_type(row.get("type"), row=row),
        "type_raw": row.get("type"),
        "text": row.get("content") or row.get("text") or "",
        "file_url": row.get("file") or row.get("fileUrl"),
        # 同 1:1:fileTh/fileHd 的 OSS URL 必须映射进列,媒体下载器才够得着
        "file_th_url": row.get("fileTh"),
        "file_hd_url": row.get("fileHd"),
        "timestamp": ts,
        "sender_name": row.get("name") or row.get("senderName"),
        "raw_json": row,
    }


async def scrape_group_messages(cookie_str: str, max_groups: int = 500, full_history: bool = False):
    """Scrape WeChat GROUP chat messages via browser fetch POST /wechatweb/msg/chats.

    Same endpoint as 1:1 chats — pass the room id as ``friendId`` (confirmed
    live 2026-08-19: rows come back with talker=sender wxid, roomid=the room,
    msgId=null → stable synthetic pk, see _normalize_group_message). Groups
    come from the local chat_group table; full-history pagination via
    paginate_chats (hasLast/end cursor, same as 1:1).
    """
    import time as _time

    from app.db.base import async_engine, async_session_maker
    from app.storage.repos import upsert_messages
    from sqlalchemy import text as sa_text

    async with async_engine.connect() as conn:
        result = await conn.execute(sa_text(
            "SELECT owner_account_id, room_id FROM chat_group "
            "WHERE channel='wechat' AND owner_account_id IS NOT NULL "
            "ORDER BY last_chat_time DESC"
        ))
        groups = [(row[0], row[1]) for row in result.fetchall()]

    room_wms = await _room_watermarks()

    logger.info(f"Discovered {len(groups)} groups, will scrape {min(max_groups, len(groups))}"
                + (" [FULL]" if full_history else " [incremental]"))

    total_fetched = 0
    total_upserted = 0
    total_pages = 0
    scan = groups[:max_groups]
    idx = 0
    relaunches = 0

    # 守护循环：超大群的单页响应可能把渲染进程/整个浏览器撑崩（真机 Target
    # crashed, 2026-08-19）——浏览器级失败时重启浏览器、跳过毒群、从当前
    # 位置继续；页面级失败换页重试（30 × 2min，同 scrape_messages）。
    while idx < len(scan):
        try:
            async with PageScraper(cookie_str) as scraper:
                fetch_page = await scraper.context.new_page()
                await fetch_page.goto(f"{YUNKE_BASE}/", wait_until="domcontentloaded", timeout=15000)
                await fetch_page.wait_for_timeout(2000)
                page = fetch_page

                while idx < len(scan):
                    owner_id, room_id = scan[idx]
                    i = idx
                    records = []
                    pages = 0
                    try:
                        now_ms = int(_time.time() * 1000)
                        watermark = 0 if full_history else room_wms.get(room_id, 0)

                        async def _fetch(end_ms: int, _owner=owner_id, _room=room_id, _wm=watermark):
                            return await page.evaluate("""async ([ownerId, friendId, startMs, endMs]) => {
                                const params = new URLSearchParams();
                                params.append('wechatId', ownerId);
                                params.append('friendId', friendId);
                                params.append('types', '');
                                params.append('start', String(startMs));
                                params.append('end', String(endMs));
                                const resp = await fetch('/wechatweb/msg/chats', {
                                    method: 'POST',
                                    headers: { 'Content-Type': 'application/x-www-form-urlencoded' },
                                    body: params.toString(),
                                });
                                return await resp.json();
                            }""", [_owner, _room, _wm, end_ms])

                        async for data in paginate_chats(_fetch, now_ms, max_pages=200):
                            pages += 1
                            if watermark and _page_min_ts(data) and _page_min_ts(data) <= watermark:
                                break
                            rows = data.get("messages") or []
                            for row in rows:
                                if not isinstance(row, dict):
                                    continue
                                rec = _normalize_group_message(row, owner_id, room_id)
                                if rec is not None:
                                    records.append(rec)

                    except Exception as e:
                        logger.warning(f"  [{i+1}] {owner_id[:12]}:{str(room_id)[:16]} failed: {e}")
                        # 页面级恢复：换新页继续；浏览器已死则抛给外层重启
                        for attempt in range(30):
                            try:
                                try:
                                    await fetch_page.close()
                                except Exception:
                                    pass
                                fetch_page = await scraper.context.new_page()
                                await fetch_page.goto(f"{YUNKE_BASE}/cms/home/statistics", wait_until="networkidle", timeout=20000)
                                await fetch_page.wait_for_timeout(1000)
                                page = fetch_page
                                break
                            except Exception as rec:
                                logger.warning(f"  recovery {attempt + 1}/30 failed: {rec}")
                                if attempt == 29:
                                    raise
                                await asyncio.sleep(120)
                        idx += 1  # 该群放弃（换页成功），继续下一个
                        continue

                    total_fetched += len(records)
                    total_pages += pages

                    if records:
                        async with async_session_maker() as session:
                            await upsert_messages(session, records)
                            await session.commit()
                        total_upserted += len(records)

                    if (i + 1) % 10 == 0:
                        logger.info(f"  [{i+1}/{len(scan)}] msgs total: {total_fetched}, pages: {total_pages}")

                    await asyncio.sleep(2 + (i % 3))
                    idx += 1

        except Exception as fatal:
            relaunches += 1
            if relaunches > 10:
                raise
            logger.warning(f"  browser-level failure, relaunch {relaunches}/10, skip poison group: {fatal}")
            idx += 1  # 跳过导致崩溃的群，避免重启后原地复崩
            await asyncio.sleep(30)

    logger.info(f"DONE: group-messages fetched={total_fetched}, upserted={total_upserted}, pages={total_pages}")
    return total_fetched, total_upserted


def _normalize_msg_type(raw_type, row: dict | None = None) -> str | None:
    """Normalize a raw message type to our canonical type string.

    判别依据 smoke.db 2026-08-25 spike(每类 ≥3 条样本): 云客 API 行里
    ``messageType`` 是经典微信协议码、``type`` 是云客自编码, 两者同在
    raw_json。优先级: messageType 经典码直判 → 49/appmsg 家族行级拆分
    (云客子码/paymsgid/text XML) → 云客码兜底(合成 flag 码时 type 才是
    唯一可靠来源)。

    spike 修正: 云客 type=3 的行 8754/8955 是 messageType='34'(经典语音
    码, file=语音十六进制键) —— 旧 "3"→image_legacy 是误标, 现指向
    voice; messageType='3' 的行(旧 orchestrator 路径写入)才是真图片。
    """
    row = row or {}
    if raw_type is None:
        raw_type = row.get("type")
        if raw_type is None:
            return None
    t = str(raw_type)

    mt_raw = row.get("messageType")
    mt = str(mt_raw) if mt_raw is not None else None

    # --- 1) appmsg 家族(经典码 49): 红包/转账/聊天记录/链接/文件 ---
    if mt == "49" or t == "49":
        return _normalize_appmsg_type(t, row)

    # --- 2) 经典微信协议码直判 ---
    if mt is not None:
        if mt == "50":
            # VOIP 通话: 云客 6=语音通话(2077 行)/7=视频通话(77 行, 频率推断)
            return {"6": "voice_call", "7": "video_call"}.get(t, "voice_call")
        classic = _CLASSIC_MSG_TYPE.get(mt)
        if classic is not None:
            return classic

    # --- 3) 云客自编码兜底(messageType 是合成 flag 码或缺失); 无行信息时
    #     也查经典码表(row-less 调用, 如单测里的 "34"/"42"/"10000") ---
    for candidate in (row.get("type"), t):
        if candidate is None:
            continue
        s = str(candidate)
        if s in _YUNKE_MSG_TYPE:
            return _YUNKE_MSG_TYPE[s]
        if s in _CLASSIC_MSG_TYPE:
            return _CLASSIC_MSG_TYPE[s]
    return "unknown"


# 经典微信协议码(raw_json.messageType)——spike 样本证实
_CLASSIC_MSG_TYPE = {
    "1": "text",        # 473922 行
    "3": "image",       # 50355 行(md5+OSS fileTh URL)
    "34": "voice",      # 云客 type=3 的主体人群(8754 行, 语音十六进制键)
    "43": "video",      # 4835 行
    "47": "emoji",      # 4242 行(tc.qq.com 表情 CDN)
    "42": "namecard",   # 953 行(text=昵称, file=头像)
    "48": "location",   # 651 行(text=地址, file="lat,lng")
    "10000": "system",  # 13820 行(打招呼/已添加提示)
}

# 云客自编码(raw_json.type)——spike 样本证实; 22(1880 行无 text 无字段)/
# 19(仅 5 行)证据不足, 宁留 unknown
_YUNKE_MSG_TYPE = {
    "1": "text",
    "2": "image",        # messageType=3
    "3": "voice",        # messageType=34(旧 image_legacy 为误标)
    "4": "video",        # messageType=43
    "5": "red_packet",   # messageType=436207665, file=wx.gtimg.com/hongbao 封面
    "6": "voice_call",   # messageType=50 VOIP
    "7": "video_call",   # messageType=50 VOIP
    "8": "emoji",        # messageType=47
    "9": "file",         # messageType=49, text=文件名(xlsx/pdf)
    "10": "link",        # messageType=49, 92% 带封面图
    "11": "transfer",    # messageType=419430449, text="发送转账：￥…"
    "13": "namecard",    # messageType=42
    "14": "location",    # messageType=48
    "15": "system",      # messageType=10000/268445456/285222674
    "18": "link",        # messageType=49, 99.8% 带封面图(公众号分享)
    "20": "link",        # messageType=754974769, 视频号分享(封面+话题文本)
    "21": "text",        # messageType=822083633 引用回复(referMsg)
}

# appmsg 家族内云客子码(row.type)——spike 样本: 9/10/18/21(+5/11 同语义码);
# 不复用 _YUNKE_MSG_TYPE 全表, 避免假想组合(如 mt=49+type=6)错判成通话
_APPMSG_SUB_TYPE = {
    "5": "red_packet",
    "9": "file",
    "10": "link",
    "11": "transfer",
    "18": "link",
}

# appmsg text XML 里的 <type> 子码(仅 49 家族无云客子码时的老数据兜底)
_APPMSG_XML_TYPE = {
    "5": "link",
    "6": "file",
    "19": "merged_forward",   # 聊天记录(合并转发), spike 1694+7 行
    "2000": "transfer",
    "2001": "red_packet",
}


def _normalize_appmsg_type(t: str, row: dict) -> str:
    """49/appmsg 家族的行级拆分(红包/转账/聊天记录/链接/文件)。

    云客子码 row.type 优先(spike: 9=file 10/18=link 21+XML=聊天记录);
    老数据缺子码时按 paymsgid(实测全空, 保留兼容)与 text XML 子码拆;
    全缺 → file(现状回退)。未知子码不猜 → unknown。
    """
    text = str(row.get("text") or "")
    sub = row.get("type")
    if sub is not None and str(sub) != "49":
        s = str(sub)
        if s == "21":
            # 云客 21 在 appmsg 家族里=聊天记录 XML(messageType=822083633
            # 的引用回复不走本分支)
            return "merged_forward"
        return _APPMSG_SUB_TYPE.get(s, "unknown")
    if row.get("paymsgid"):
        if ">2001<" in text:
            return "red_packet"
        if ">2000<" in text:
            return "transfer"
        # paymsgid 有值但 XML 无 type → transfer(转账才有 paymsgid 概率高)
        return "transfer"
    if ">2001<" in text:
        return "red_packet"
    if ">2000<" in text:
        return "transfer"
    m = re.search(r"<type>(\d+)</type>", text)
    if m:
        return _APPMSG_XML_TYPE.get(m.group(1), "unknown")
    return "file"


# ---------------------------------------------------------------------------
# Shared helpers for the fetch()-based entities
# ---------------------------------------------------------------------------


async def _open_fetch_page(scraper: PageScraper):
    """Open a stable page in the scraper's browser context for fetch() calls.

    The fetch() calls work from any same-origin page (even a login redirect)
    as long as the context carries the session cookies — same trick the
    message scraper uses.
    """
    page = await scraper.context.new_page()
    await page.goto(f"{YUNKE_BASE}/", wait_until="domcontentloaded", timeout=15000)
    await page.wait_for_timeout(2000)
    return page


async def _browser_fetch(page, path: str, json_body: dict | None = None, form_body: dict | None = None):
    """POST from inside the browser context (proper cookies/referrer => no 风控).

    ``json_body`` sends a JSON body; ``form_body`` sends form-urlencoded.
    Returns the parsed JSON response. 鉴权失败信封(云客 cookie 过期)在此
    统一拦截抛 :class:`CookieExpiredError`(2026-09-03:过期时 WA 路径对
    str data 调 .get 崩 AttributeError、微信路径静默当空数据 55 员工 0 行)。
    """
    if json_body is not None:
        resp = await page.evaluate(
            """async ([path, body]) => {
                const resp = await fetch(path, {
                    method: 'POST',
                    headers: { 'Content-Type': 'application/json' },
                    body: JSON.stringify(body),
                });
                return await resp.json();
            }""",
            [path, json_body],
        )
    else:
        resp = await page.evaluate(
            """async ([path, params]) => {
                const ps = new URLSearchParams();
                for (const [k, v] of Object.entries(params)) ps.append(k, String(v));
                const resp = await fetch(path, {
                    method: 'POST',
                    headers: { 'Content-Type': 'application/x-www-form-urlencoded' },
                    body: ps.toString(),
                });
                return await resp.json();
            }""",
            [path, form_body or {}],
        )
    return _check_auth_envelope(resp)


class CookieExpiredError(RuntimeError):
    """云客会话过期(鉴权失败信封)。调用方应中止本轮同步,提示重新登录换 cookie。"""


def _check_auth_envelope(resp):
    """鉴权失败信封 → CookieExpiredError;其余响应原样返回。

    云客过期时的实测信封(code 时而 str 时而 int):
    ``{"code":"-20020","data":"账号认证已过期…","message":"鉴权失败…","success":false}``
    业务错误(如 -90000)不是鉴权问题,照旧放行由调用方语义处理。
    """
    if isinstance(resp, dict) and resp.get("success") is False:
        code = str(resp.get("code", ""))
        if code in {"-20020", "20020"}:
            raise CookieExpiredError(
                f"云客 cookie 已过期(code={code}:{resp.get('message')}),"
                "需重新登录换取后更新 YK_COOKIE"
            )
    return resp


# ── cookie 自动续期(2026-09-03)────────────────────────────────────────── #
# 缓存文件优先于 compose 注入的 YK_COOKIE(自动续期成功时写入,永远比 env 新);
# 实体撞 CookieExpiredError → 浏览器表单重登 → 落缓存 → 整实体重跑一次
# (upsert 幂等 + 水位增量,重跑安全);二次过期直接抛出,绝不循环。


def _cookie_cache_path():
    from pathlib import Path

    override = os.environ.get("YK_COOKIE_CACHE")
    if override:
        return Path(override)
    from app.config import get_settings

    return Path(get_settings().media_dir) / ".yk_cookie_cache"


def _load_cookie() -> str:
    """缓存文件 > YK_COOKIE env > settings(.env)。空串=三者皆无。"""
    try:
        cached = _cookie_cache_path().read_text(encoding="utf-8").strip()
        if cached:
            return cached
    except OSError:
        pass
    cookie = os.environ.get("YK_COOKIE", "")
    if not cookie:
        from app.config import get_settings

        cookie = get_settings().yk_cookie
    return cookie or ""


def _save_cookie_cache(cookie: str) -> None:
    """尽力而为持久化:路径不可写(本地无挂载)只告警不炸。"""
    try:
        p = _cookie_cache_path()
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(cookie, encoding="utf-8")
    except OSError as exc:
        logger.warning(f"cookie 缓存写入失败({exc}),新 cookie 仅本次进程生效")


async def _run_with_auto_relogin(entity, cookie: str, *, relogin=None):
    """跑一个 entity(cookie 参数化);撞过期自动重登换 cookie 重跑一次。"""
    if relogin is None:
        relogin = _relogin
    try:
        return await entity(cookie)
    except CookieExpiredError:
        logger.warning("cookie 过期 → 浏览器表单自动重登…")
        new_cookie = await relogin()
        _save_cookie_cache(new_cookie)
        logger.info("自动续期成功,新 cookie 已缓存;重跑本实体")
        return await entity(new_cookie)


async def _relogin() -> str:
    """浏览器表单登录云客换取新 cookie(2026-09-03 实测一次成功)。

    选择器来自真实页面:tab 文本「密码登录」、placeholder 请输入手机号/
    请输入密码、按钮「登 录」;成功判据 = URL 跳到 /cms/home/* 且
    document.cookie 出现 userToken。
    """
    from app.config import get_settings

    s = get_settings()
    if not s.yunke_username or not s.yunke_password:
        raise CookieExpiredError("自动重登失败:未配置 YUNKE_USERNAME/YUNKE_PASSWORD")
    try:
        async with PageScraper("") as scraper:
            page = await scraper.context.new_page()
            await page.goto(f"{YUNKE_BASE}/cms/auth/login", wait_until="networkidle", timeout=30000)
            await page.click("text=密码登录")
            await page.fill('input[placeholder="请输入手机号"]', s.yunke_username)
            await page.fill('input[placeholder="请输入密码"]', s.yunke_password)
            await page.click('button:has-text("登 录")')
            await page.wait_for_url("**/cms/home/**", timeout=20000)
            cookie = await page.evaluate("() => document.cookie")
    except Exception as exc:
        raise CookieExpiredError(f"自动重登失败:{exc}") from exc
    if "userToken=" not in cookie:
        raise CookieExpiredError("自动重登失败:登录后未见 userToken cookie")
    return cookie


def _rows_flat(body: dict) -> list[dict]:
    """Extract rows from the flat envelope: ``data.data[]`` (wxUser/list,
    queryWeChatVoiceList, call/list)."""
    data = body.get("data") if isinstance(body, dict) else None
    if not isinstance(data, dict):
        return []
    rows = data.get("data")
    return [r for r in rows if isinstance(r, dict)] if isinstance(rows, list) else []


def _rows_basic_result(body: dict) -> list[dict]:
    """friend_list/es 信封: ``data.basicResult.data[]``(好友与群共用)。"""
    data = body.get("data") if isinstance(body, dict) else None
    basic = data.get("basicResult") if isinstance(data, dict) else None
    rows = basic.get("data") if isinstance(basic, dict) else None
    return [r for r in rows if isinstance(r, dict)] if isinstance(rows, list) else []


async def _root_dept_id() -> str:
    """本地 department 表的公司根部门 id(多个抓取端点的必填 body 字段)。"""
    from app.db.base import async_engine
    from sqlalchemy import text as sa_text

    async with async_engine.connect() as conn:
        row = (await conn.execute(sa_text(
            "SELECT dept_id FROM department WHERE parent_id IS NULL LIMIT 1"
        ))).first()
    return row[0] if row else ""


async def _fetch_pages(
    page,
    path: str,
    *,
    base_body: dict,
    parse_rows,
    page_size: int = 50,
    max_pages: int = 500,
    form: bool = False,
    page_key: str = "pageNum",
) -> list[dict]:
    """按页码翻页浏览器直发 POST,直到短页;收集全部 rows 返回。

    2026-08-20 起 list 类实体一律走此路径(导航+拦截被登录墙杀死,0 捕获)。
    ``form=True`` 发 form-urlencoded(Spring @RequestParam,如 wxUser/list);
    ``page_key`` 兼容 WA 的 ``page``。首页响应 code 异常时醒目告警
    cookie 失效(所有 list 端点健康时 code=10000)。
    """
    rows_all: list[dict] = []
    first_row_sig: str | None = None
    page_num = 1
    while page_num <= max_pages:
        body = {**base_body, page_key: page_num, "pageSize": page_size}
        if form:
            resp = await _browser_fetch(page, path, form_body=body)
        else:
            resp = await _browser_fetch(page, path, json_body=body)
        if page_num == 1 and isinstance(resp, dict):
            code = resp.get("code")
            if code is not None and str(code) not in ("10000",):
                logger.warning(
                    f"⚠️ {path} 首页 code={code} ({resp.get('message')}) — "
                    "cookie 可能已过期/失效,请从浏览器重新复制 YK_COOKIE 到 .env!"
                )
        rows = parse_rows(resp)
        # 粘页检测:端点忽略页码参数时第 2 页起首行与第 1 页相同(实测 wxUser/list
        # 的页码字段是 page 而非 pageNum,2026-08-20)——丢弃重复页并停止
        if rows:
            import json as _json
            sig = _json.dumps(rows[0], sort_keys=True, default=str)[:200]
            if page_num == 1:
                first_row_sig = sig
            elif sig == first_row_sig:
                logger.warning(f"  {path} page {page_num} 与首页相同 — 端点忽略页码参数,停止翻页")
                break
        rows_all.extend(rows)
        logger.info(f"  {path.rsplit('/', 1)[-1]} page {page_num}: {len(rows)} rows")
        if len(rows) < page_size:
            break
        page_num += 1
        await asyncio.sleep(1.5)
    return rows_all


def _rows_overviews(body: dict) -> list[dict]:
    """Extract rows from the WA envelope: ``data.overviews[]``
    (businessSalesList / businessFriendList)."""
    data = body.get("data") if isinstance(body, dict) else None
    if not isinstance(data, dict):
        return []
    rows = data.get("overviews")
    return [r for r in rows if isinstance(r, dict)] if isinstance(rows, list) else []


async def _upsert_all(upsert_fn, records: list[dict]) -> None:
    """Batch-upsert records (each fn per record) in one committed session."""
    from app.db.base import async_session_maker

    if not records:
        return
    async with async_session_maker() as session:
        for rec in records:
            await upsert_fn(session, rec)
        await session.commit()


# ---------------------------------------------------------------------------
# WeChat staff accounts (navigation + interception)
# ---------------------------------------------------------------------------


async def scrape_accounts(cookie_str: str, max_pages: int = 100):
    """微信员工账号:浏览器直发 wxUser/list(form-encoded,2026-08-20 起改 fetch)。

    body 必填 isOnJob/departmentId/startdate/enddate(缺则 -90000);时间窗只
    影响每账号统计数字不影响返回集合,取近 7 天。"""
    import datetime as _dt

    from app.scraper.wechat import _normalize_wxuser_account
    from app.storage.repos import upsert_wechat_account

    dept = await _root_dept_id()
    end_date = _dt.datetime.now().date()
    start_date = end_date - _dt.timedelta(days=7)

    async with PageScraper(cookie_str) as scraper:
        page = await _open_fetch_page(scraper)
        rows = await _fetch_pages(
            page, INTERCEPT_PATHS["wxUser_list"],
            base_body={
                "isOnJob": "1",
                "departmentId": dept,
                "startdate": start_date.isoformat(),
                "enddate": end_date.isoformat(),
                "userId": "",
                "salesname": "",
                "weChatNickName": "",
            },
            parse_rows=_rows_flat,
            form=True,
            page_key="page",  # 此端点页码字段是 page(实测 pageNum 被无视)
            max_pages=max_pages,
        )

    records = []
    for idx, row in enumerate(rows):
        rec = _normalize_wxuser_account(row)
        # 记录云客 API 返回顺序(对齐员工列表排列)
        rec["list_order"] = idx
        records.append(rec)
    await _upsert_all(upsert_wechat_account, records)
    total = len(records)
    logger.info(f"DONE: accounts fetched={total}, upserted={total}")
    return total, total


# ---------------------------------------------------------------------------
# WhatsApp staff accounts (browser fetch)
# ---------------------------------------------------------------------------


async def scrape_wa_accounts(cookie_str: str, max_pages: int = 100):
    """WA 员工账号:浏览器直发 businessSalesList(JSON body {pageNum,pageSize})。"""
    from app.scraper.whatsapp import _normalize_whatsapp_account
    from app.storage.repos import upsert_whatsapp_account

    async with PageScraper(cookie_str) as scraper:
        page = await _open_fetch_page(scraper)
        rows = await _fetch_pages(
            page, INTERCEPT_PATHS["wa_user"],
            base_body={}, parse_rows=_rows_overviews,
            max_pages=max_pages,
        )

    records = [_normalize_whatsapp_account(row) for row in rows]
    await _upsert_all(upsert_whatsapp_account, records)
    total = len(records)
    logger.info(f"DONE: wa-accounts fetched={total}, upserted={total}")
    return total, total


# ---------------------------------------------------------------------------
# WhatsApp customers (browser fetch)
# ---------------------------------------------------------------------------


async def scrape_wa_friends(cookie_str: str, max_pages: int = 2000):
    """WA 客户:浏览器直发 businessFriendList type=0,**逐员工**(uid 过滤)翻页。

    全局查询有 ~1 万行硬上限(实测云客声明 11931 vs 全局翻页只到 10000,
    2026-08-20);逐员工 ``uid`` 过滤(员工 WA 号)范围小无上限,页码字段是
    ``page``,total-aware 翻页(短页继续,满声明数或连续空页停)。"""
    from app.db.base import async_engine, async_session_maker
    from sqlalchemy import text as sa_text
    from app.scraper.whatsapp import _normalize_whatsapp_friend
    from app.storage.repos import prune_customer_links, upsert_customer

    dept = await _root_dept_id()
    async with async_engine.connect() as conn:
        result = await conn.execute(sa_text(
            "SELECT uid FROM whatsapp_account WHERE uid IS NOT NULL AND uid != ''"
        ))
        accounts = [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, wa_uid in enumerate(accounts):
            base_body = {
                "searchInfo": "",
                "friendSearchInfo": "",
                "departmentId": dept,
                "type": "0",
                "uid": wa_uid,
            }
            rows: list[dict] = []
            declared: int | None = None
            empty_streak = 0
            page_num = 1
            while page_num <= max_pages:
                resp = await _browser_fetch(
                    page, INTERCEPT_PATHS["wa_friend"],
                    json_body={**base_body, "page": page_num, "pageSize": 50},
                )
                data = resp.get("data") or {}
                declared = data.get("totalCount") or declared
                page_rows = data.get("overviews") or []
                if not page_rows:
                    empty_streak += 1
                    if empty_streak >= 2 or (declared and len(rows) >= declared):
                        break
                else:
                    empty_streak = 0
                    rows.extend(page_rows)
                    if declared and len(rows) >= declared:
                        break
                page_num += 1
                await asyncio.sleep(1)

            records = []
            for row in rows:
                rec = _normalize_whatsapp_friend(row)
                if rec.get("friend_id"):
                    records.append(rec)
            total_fetched += len(records)
            if records:
                await _upsert_all(upsert_customer, records)
                total_upserted += len(records)
            # 对账(2026-09-08,同微信 friends):名单取满才删该账号名下
            # 不在当前列表里的 link。
            pruned = 0
            if declared is not None and len(rows) >= declared:
                keep_ids = [r.get("friendUid") for r in rows if r.get("friendUid")]
                async with async_session_maker() as session:
                    pruned = await prune_customer_links(
                        session, channel="whatsapp", owner_account_id=wa_uid,
                        keep_friend_ids=keep_ids,
                    )
                    await session.commit()
            logger.info(
                f"  [{i+1}/{len(accounts)}] {wa_uid[:18]}..: declared={declared} "
                f"rows={len(rows)} pruned={pruned}"
            )
            await asyncio.sleep(2 + (i % 3))

    logger.info(f"DONE: wa-friends fetched={total_fetched}, upserted={total_upserted}")
    return total_fetched, total_upserted


def _normalize_wa_group_row(row: dict) -> dict | None:
    """businessFriendList type=1 行 → ChatGroup 形状（真机确认 2026-08-20）。

    type=1 返回的是好友字段名结构：friendUid=群 id（@g.us 后缀）、nickName=
    群名、uid=员工 WA 号、salesNickName=员工 WA 昵称——与微信群的
    friendWxId/friendWxNickname 完全不同，故不能复用 _normalize_chat_group。
    非 @g.us 行（异常混入）返回 None。
    """
    friend_uid = str(row.get("friendUid") or "")
    if not friend_uid.endswith("@g.us"):
        return None
    return {
        "room_id": friend_uid,
        "channel": "whatsapp",
        "room_name": row.get("nickName"),
        "owner_account_id": row.get("uid"),
        "owner_account_nickname": row.get("salesNickName"),
        "owner_user_id": row.get("userId"),
        "owner_user_phone": row.get("salePhone"),
        "message_count": row.get("messageGroupCount"),
        "raw_json": row,
    }


async def scrape_wa_groups(cookie_str: str, max_pages: int = 200):
    """Scrape WhatsApp GROUPS via browser fetch POST businessFriendList type=1.

    Same endpoint as WA friends with ``type:"1"`` — live-verified 2026-08-20:
    rows carry friendUid ending ``@g.us`` (group id), nickName = group name.
    Envelope: data.overviews; upsert to chat_group (channel='whatsapp').
    """
    from app.db.base import async_engine
    from sqlalchemy import text as sa_text
    from app.storage.repos import upsert_chat_group

    async with async_engine.connect() as conn:
        row = (await conn.execute(sa_text(
            "SELECT dept_id FROM department WHERE parent_id IS NULL LIMIT 1"
        ))).first()
    department_id = row[0] if row else ""
    if not department_id:
        logger.warning("wa-groups: no root department in DB — abort")
        return 0, 0

    page_size = 50
    total_fetched = 0
    total_upserted = 0
    async with PageScraper(cookie_str) as scraper:
        page = await _open_fetch_page(scraper)
        page_num = 1
        while page_num <= max_pages:
            body = await _browser_fetch(
                page,
                INTERCEPT_PATHS["wa_friend"],
                json_body={
                    "searchInfo": "",
                    "friendSearchInfo": "",
                    "page": page_num,
                    "pageSize": page_size,
                    "departmentId": department_id,
                    "type": "1",
                },
            )
            rows = _rows_overviews(body)
            records = [rec for r in rows if (rec := _normalize_wa_group_row(r)) is not None]
            total_fetched += len(records)
            await _upsert_all(upsert_chat_group, records)
            total_upserted += len(records)
            logger.info(f"  wa-groups page {page_num}: {len(rows)} rows, {len(records)} groups")
            if len(rows) < page_size:
                break
            page_num += 1
            await asyncio.sleep(1.5)

    logger.info(f"DONE: wa-groups fetched={total_fetched}, upserted={total_upserted}")
    return total_fetched, total_upserted


# ---------------------------------------------------------------------------
# 微信群统计 getChatroomList(2026-09-09 真机探明,云客同款数据源)
#
# 云客「沟通→微信号管理→微信群统计」页(/cms/wechat/wechatgroup)数据源:
# POST /wechatweb/chatroom/getChatroomList(form-urlencoded)
#   departmentId + startTime/endTime('YYYY-MM-DD HH:MM:SS' 时间窗) + 分页;
# 信封 data.overviews[] + totalCount(1,854 行,员工×群维度)。
# 行含:真 memberCount(群成员数量——friend_list/es 的 friendNum 其实是
# 员工好友数,旧映射存反了)、remark(群备注)、员工微信(wechatId=微信号
# alias/wechatNickName/头像)、员工信息(userName/userPhone)、以及**窗内**
# addMembers/lossMembers(新增/退出成员数,任意窗实时算)。
#
# 本地无成员进出事件流 → add/loss 只能按天积累:每夜抓「当日窗」落
# chat_group_member_stats 日行,页面任意日窗=按天求和(历史无数据)。
# ---------------------------------------------------------------------------

CHATROOM_LIST_PATH = "/wechatweb/chatroom/getChatroomList"


def _normalize_chatroom_stat_row(row: dict, owner_wxid: str) -> dict:
    """getChatroomList 行 → ChatGroup 补列记录。

    只放该端点有的字段——未出现的列不参与 upsert 的 UPDATE 子句,
    friend_list/es 抓的群主信息等旧值得以保留。
    """
    rec = {
        "room_id": row.get("chatRoomId"),
        "channel": "wechat",
        "owner_account_id": owner_wxid,
        "room_name": row.get("chatRoomName"),
        # 真群成员数(修正:旧源 friendNum 是「员工好友数」被误存此列)
        "member_count": row.get("memberCount"),
        "head_image_url": row.get("headImageUrl"),
        "owner_account_nickname": row.get("wechatNickName"),
        "owner_user_id": row.get("userId"),
        "owner_user_phone": row.get("userPhone"),
        "message_count": row.get("messageCount"),
        "own_message_count": row.get("ownMessageCount"),
        "add_members": row.get("addMembers"),  # 本窗快照(日窗=当日进出)
        "loss_members": row.get("lossMembers"),
        "last_chat_time": row.get("lastChatTime"),
        "update_time": row.get("updateTime"),
        "raw_json": row,
    }
    # 群备注:getChatroomList 行通常无 remark 键(云客页面也显示 -);有才写,
    # 无则不放该键——upsert 不更新未出现的列,friend_list/es 来源的
    # friendWxRemark 备注得以保留
    if row.get("remark"):
        rec["group_remark"] = row["remark"]
    return rec


async def _chatroom_rows(page, dept: str, start_str: str, end_str: str, max_pages: int = 100):
    """翻完 getChatroomList 一个窗,返回全部行(list)。"""
    rows_all: list[dict] = []
    page_num = 1
    while page_num <= max_pages:
        body = await _browser_fetch(
            page, CHATROOM_LIST_PATH,
            form_body={
                "departmentId": dept, "startTime": start_str, "endTime": end_str,
                "pageNum": page_num, "pageSize": 50,
            },
        )
        data = body.get("data") or {}
        rows = data.get("overviews") or []
        rows_all.extend(rows)
        total = data.get("totalCount") or 0
        logger.info(f"  chatroom-list page {page_num}: {len(rows)} rows (total {total})")
        if not rows or page_num * 50 >= (total or 0):
            break
        page_num += 1
        await asyncio.sleep(1.2)
    return rows_all


async def _owner_alias_map(conn):
    """wechat_account 的 alias/nick_name → wx_id 映射。

    getChatroomList 行的 wechatId 是员工**微信号**(alias,如 Nancyjade)
    而非 wxId,owner 归属需经此解析;miss 时回退存原值(不影响其余列)。
    """
    from sqlalchemy import text as sa_text

    result = await conn.execute(sa_text(
        "SELECT alias, nick_name, wx_id FROM wechat_account "
        "WHERE wx_id IS NOT NULL AND wx_id != ''"
    ))
    mapping: dict[str, str] = {}
    for alias, nick, wx in result.fetchall():
        if alias:
            mapping[str(alias)] = wx
        if nick:
            mapping.setdefault(str(nick), wx)
    return mapping


async def scrape_chatroom_stats(cookie_str: str, day: str | None = None):
    """getChatroomList 抓取(云客微信群统计同款)。

    * ``day=None``:全窗(2020 起)——chat_group 档案补列基线(真成员数/
      群备注/员工信息列;add/loss 存全窗快照);
    * ``day='YYYY-MM-DD'``:单日窗——落 chat_group_member_stats 当日行
      (页面任意日窗的新增/退出=按天求和)。
    """
    import datetime as _dt

    from app.db.base import async_engine
    from app.storage.repos import upsert_chat_group, upsert_chat_group_member_stats

    dept = await _root_dept_id()
    async with async_engine.connect() as conn:
        alias_map = await _owner_alias_map(conn)

    if day:
        start_str, end_str = f"{day} 00:00:00", f"{day} 23:59:59"
        day_key = int(day.replace("-", ""))
    else:
        start_str = "2020-01-01 00:00:00"
        end_str = f"{_dt.date.today()} 23:59:59"
        day_key = None

    group_recs: list[dict] = []
    stats_recs: list[dict] = []
    async with PageScraper(cookie_str) as scraper:
        page = await _open_fetch_page(scraper)
        for row in await _chatroom_rows(page, dept, start_str, end_str):
            rid = row.get("chatRoomId")
            if not rid:
                continue
            owner_wxid = (
                alias_map.get(str(row.get("wechatId") or ""))
                or alias_map.get(str(row.get("wechatNickName") or ""))
                or str(row.get("wechatId") or "")
            )
            group_recs.append(_normalize_chatroom_stat_row(row, owner_wxid))
            if day_key is not None:
                stats_recs.append({
                    "channel": "wechat",
                    "owner_account_id": owner_wxid,
                    "room_id": rid,
                    "day": day_key,
                    "added": row.get("addMembers") or 0,
                    "lost": row.get("lossMembers") or 0,
                })

    if group_recs:
        await _upsert_all(upsert_chat_group, group_recs)
    if stats_recs:
        await _upsert_all(upsert_chat_group_member_stats, stats_recs)
    logger.info(
        f"DONE: chatroom-stats day={day or '全窗'} groups={len(group_recs)} "
        f"stats_rows={len(stats_recs)}"
    )
    return len(group_recs), len(stats_recs)


# ---------------------------------------------------------------------------
# 员工朋友圈 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
        # 每页一行进度(镜像 _fetch_pages 的 "<endpoint> page N: X rows")
        logger.info(f"  friendster page {page_num}: {len(rows)} rows")
        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)
# ---------------------------------------------------------------------------


async def scrape_voice_calls(cookie_str: str, max_pages: int = 200):
    """Scrape voice call records from BOTH channels, upsert to phone_call.

    * WeChat: browser-fetch queryWeChatVoiceList (callSource=0) with TWO
      passes isRoom=0(好友)/1(群). The voice-call-list page only sends
      isRoom=0 on load — group voice calls live behind the 微信群 tab, so
      pure navigation+interception never captured them (root cause of the
      empty 群 scope tab). Body must include callSource+departmentId+isRoom+
      startTime+endTime or the API returns 200 with empty data.
    * WA: browser-fetch the same unified endpoint with callSource=4
      (no group concept; interception stopped capturing on 2026-08-18).
    """
    from app.db.base import async_engine
    from sqlalchemy import text as sa_text
    from app.scraper.wechat import _normalize_voice_call
    from app.scraper.whatsapp import _normalize_wa_voice_call
    from app.storage.repos import upsert_voice_call

    # queryWeChatVoiceList requires departmentId — take the company-root dept
    async with async_engine.connect() as conn:
        row = (await conn.execute(sa_text(
            "SELECT dept_id FROM department WHERE parent_id IS NULL LIMIT 1"
        ))).first()
    department_id = row[0] if row else ""
    if not department_id:
        logger.warning("voice-calls: no root department in DB — skipping wechat fetch passes")

    page_size = 100
    total_fetched = 0
    total_upserted = 0
    async with PageScraper(cookie_str) as scraper:
        # --- WeChat: browser fetch, isRoom 0(好友)/1(群) two passes ---
        if department_id:
            page = await _open_fetch_page(scraper)
            for is_room, label in (("0", "好友"), ("1", "群")):
                records = []
                page_num = 1
                while page_num <= max_pages:
                    body = await _browser_fetch(
                        page,
                        INTERCEPT_PATHS["call_list"],
                        json_body={
                            "pageNum": page_num,
                            "pageSize": page_size,
                            "callSource": "0",
                            "isRoom": is_room,
                            "startTime": "2000-01-01 00:00:00",
                            "endTime": "2099-12-31 23:59:59",
                            "sort": "",
                            "ids": [],
                            "userId": "",
                            "departmentId": department_id,
                        },
                    )
                    rows = _rows_flat(body)
                    for row in rows:
                        rec = _normalize_voice_call(row)
                        if rec.get("call_id") is not None:
                            records.append(rec)
                    if len(rows) < page_size:
                        break
                    page_num += 1
                    await asyncio.sleep(1)
                total_fetched += len(records)
                await _upsert_all(upsert_voice_call, records)
                total_upserted += len(records)
                logger.info(f"  wechat isRoom={is_room}({label}): {len(records)} calls, {page_num} page(s)")
                await asyncio.sleep(2)

        # --- WA: browser fetch (callSource=4, no group concept; same unified endpoint) ---
        # Interception on the wavoicecalllist page used to work but captured 0
        # responses on 2026-08-18 (page no longer fires the API on load) — direct
        # browser fetch with the full body is the reliable path, same as WeChat.
        if department_id:
            records = []
            page_num = 1
            while page_num <= max_pages:
                body = await _browser_fetch(
                    page,
                    INTERCEPT_PATHS["call_list"],
                    json_body={
                        "pageNum": page_num,
                        "pageSize": page_size,
                        "callSource": "4",
                        "isRoom": "0",
                        "startTime": "2000-01-01 00:00:00",
                        "endTime": "2099-12-31 23:59:59",
                        "sort": "",
                        "ids": [],
                        "userId": "",
                        "departmentId": department_id,
                    },
                )
                rows = _rows_flat(body)
                for row in rows:
                    rec = _normalize_wa_voice_call(row)
                    if rec.get("call_id") is not None:
                        records.append(rec)
                if len(rows) < page_size:
                    break
                page_num += 1
                await asyncio.sleep(1)
            total_fetched += len(records)
            await _upsert_all(upsert_voice_call, records)
            total_upserted += len(records)
            logger.info(f"  wa(callSource=4): {len(records)} calls, {page_num} page(s)")

    logger.info(f"DONE: voice-calls fetched={total_fetched}, upserted={total_upserted}")
    return total_fetched, total_upserted


# ---------------------------------------------------------------------------
# Long voice clips (browser fetch, per employee)
# ---------------------------------------------------------------------------


async def scrape_voice_clips(cookie_str: str, max_employees: int = 200, page_size: int = 50):
    """Scrape long voice clips via browser fetch POST /wechatweb/call/list.

    The endpoint is per-employee: employee user_ids come from the LOCAL
    wechat_account / whatsapp_account tables (channel resolved the same way).
    JSON body {pageNum, pageSize, userId} (真机确认 2026-08-07 spec §2).
    """
    from app.db.base import async_engine
    from app.scraper.voice_clip import _normalize_voice_clip
    from app.storage.repos import upsert_voice_clip
    from sqlalchemy import text as sa_text

    # Resolve employee user_id -> channel from local staff-account tables.
    # WeChat wins when an id appears in both tables.
    user_channel: dict[str, str] = {}
    async with async_engine.connect() as conn:
        result = await conn.execute(sa_text(
            "SELECT DISTINCT user_id FROM whatsapp_account "
            "WHERE user_id IS NOT NULL AND user_id != ''"
        ))
        for (uid,) in result.fetchall():
            user_channel[uid] = "whatsapp"
        result = await conn.execute(sa_text(
            "SELECT DISTINCT user_id FROM wechat_account "
            "WHERE user_id IS NOT NULL AND user_id != ''"
        ))
        for (uid,) in result.fetchall():
            user_channel[uid] = "wechat"

    employees = sorted(user_channel.items())
    logger.info(f"Found {len(employees)} employees; scraping {min(max_employees, len(employees))}")

    total_fetched = 0
    total_upserted = 0

    async with PageScraper(cookie_str) as scraper:
        page = await _open_fetch_page(scraper)

        for i, (user_id, channel) in enumerate(employees[:max_employees]):
            records = []
            try:
                page_num = 1
                while True:
                    body = await _browser_fetch(
                        page,
                        INTERCEPT_PATHS["voice_list"],
                        json_body={"pageNum": page_num, "pageSize": page_size, "userId": user_id},
                    )
                    rows = _rows_flat(body if isinstance(body, dict) else {})
                    if not rows:
                        break
                    records.extend(_normalize_voice_clip(row, channel=channel) for row in rows)
                    data = body.get("data") or {}
                    total = data.get("total")
                    if isinstance(total, (int, float)) and page_num * page_size >= int(total):
                        break
                    page_num += 1
                    await asyncio.sleep(1)
            except Exception as e:
                logger.warning(f"  [{i+1}] user {user_id} failed: {e}")
                try:
                    await page.close()
                except Exception:
                    pass
                page = await _open_fetch_page(scraper)
                continue

            total_fetched += len(records)
            if records:
                await _upsert_all(upsert_voice_clip, records)
                total_upserted += len(records)

            if (i + 1) % 10 == 0 or not records:
                logger.info(f"  [{i+1}/{min(max_employees, len(employees))}] {user_id}({channel}): {len(records)} clips (total: {total_fetched})")
            await asyncio.sleep(2 + (i % 3))

    logger.info(f"DONE: voice-clips fetched={total_fetched}, upserted={total_upserted}")
    return total_fetched, total_upserted


# ---------------------------------------------------------------------------
# WhatsApp messages (browser fetch, per conversation, backward cursor)
# ---------------------------------------------------------------------------


async def scrape_wa_messages(cookie_str: str, max_conversations: int = 100, full_history: bool = False):
    """Scrape WhatsApp chat records via browser fetch POST
    /qqweb/WhatsAppUser/chatRecords (JSON body {uid, friendUid, start, end}).

    Conversation pairs come from _discover_conversations (customer table ∪
    voice-call pairs). Pages backward via data.end / data.hasLast
    (same cursor strategy as the WeChat msg/chats scraper).
    """
    import time as _time
    from sqlalchemy import text as sa_text

    from app.db.base import async_session_maker
    from app.scraper.whatsapp import _is_wa_divider, _normalize_whatsapp_message
    from app.storage.repos import upsert_messages

    conversations = await _discover_conversations("whatsapp", "wa_voice_call")
    watermarks = await _conversation_watermarks("whatsapp")
    cust_activity = await _customer_activity("whatsapp")

    skipped = 0
    if not full_history:
        kept = []
        for pair in conversations:
            wm = watermarks.get(pair, 0)
            act = cust_activity.get(pair, 0)
            # 注意:不能用 +60s 缓冲——若新消息恰好落在水位后 60s 内,
            # act 永远 ≤ wm+60s → 每轮都跳过 → 该消息**永久漏掉**。
            # act==wm 表示云客最新消息==我们最新消息,精确相等才跳。
            if wm > 0 and act > 0 and act <= wm:
                skipped += 1
            else:
                kept.append(pair)
        conversations = kept
    logger.info(
        f"Discovered WA conversations: {len(conversations)} to visit, {skipped} skipped (no new activity)"
        + (" [FULL]" if full_history else " [incremental]")
    )

    total_fetched = 0
    total_upserted = 0

    async with PageScraper(cookie_str) as scraper:
        page = await _open_fetch_page(scraper)

        for i, (uid, friend_uid) in enumerate(conversations[:max_conversations]):
            records = []
            watermark = 0 if full_history else watermarks.get((uid, friend_uid), 0)
            try:
                end_ms = int(_time.time() * 1000)
                while True:
                    body = await _browser_fetch(
                        page,
                        "/qqweb/WhatsAppUser/chatRecords",
                        json_body={"uid": uid, "friendUid": friend_uid, "start": watermark, "end": end_ms},
                    )
                    data = body.get("data") if isinstance(body, dict) else None
                    if not isinstance(data, dict):
                        break
                    msgs = data.get("messages") or []
                    for row in msgs:
                        if isinstance(row, dict) and not _is_wa_divider(row):
                            records.append(_normalize_whatsapp_message(row))
                    if not data.get("hasLast"):
                        break
                    # 增量触底停:翻到已同步区
                    if watermark and _page_min_ts(data) and _page_min_ts(data) <= watermark:
                        break
                    end = data.get("end")
                    if not isinstance(end, (int, float)):
                        break
                    end_ms = int(end) - 1
                    await asyncio.sleep(1)
            except Exception as e:
                logger.warning(f"  [{i+1}] {str(uid)[:20]}:{str(friend_uid)[:20]} failed: {e}")
                # Recreate the fetch page; retry patiently on outages (same
                # resilience as the WeChat message scraper — 30 × 2min).
                for attempt in range(30):
                    try:
                        try:
                            await page.close()
                        except Exception:
                            pass
                        page = await _open_fetch_page(scraper)
                        break
                    except Exception as rec:
                        logger.warning(f"  recovery {attempt + 1}/30 failed: {rec}")
                        if attempt == 29:
                            raise
                        await asyncio.sleep(120)
                continue

            total_fetched += len(records)
            if records:
                # upsert_messages 收整列表(逐条版 _upsert_all 会把 dict 迭代成
                # key 字符串 → TypeError, 2026-08-19 WA 阶段首条即崩的根因)
                async with async_session_maker() as session:
                    await upsert_messages(session, records)
                    # 时效:刷新 customer.last_chat_time(同微信 1:1,云客按最近沟通实时排序)
                    max_ts = max((r.get("timestamp") or 0) for r in records)
                    if max_ts:
                        await session.execute(sa_text(
                            "UPDATE customer SET last_chat_time = :ts "
                            "WHERE channel='whatsapp' AND owner_account_id = :o AND friend_id = :f "
                            "AND (last_chat_time IS NULL OR last_chat_time < :ts)"
                        ), {"ts": max_ts, "o": uid, "f": friend_uid})
                    await session.commit()
                total_upserted += len(records)

            if (i + 1) % 10 == 0:
                logger.info(f"  [{i+1}/{min(max_conversations, len(conversations))}] msgs total: {total_fetched}")
            await asyncio.sleep(2 + (i % 3))

    logger.info(f"DONE: wa-messages fetched={total_fetched}, upserted={total_upserted}")
    return total_fetched, total_upserted


def _normalize_wa_group_message(record: dict) -> dict:
    """chatRecords 群消息行 → Message 形状（friendUid=@g.us,真机确认 2026-08-20）。

    与 WA 1:1 行同构,差异:``chatRoomTalker`` = 群内发送者(@lid)、friendUid =
    群 id(@g.us);keyId 通常有值(与微信群 msgId=null 不同),缺失时用
    owner:room:createTime:chatRoomTalker 合成稳定主键。
    """
    from app.scraper.whatsapp import _normalize_whatsapp_message

    rec = _normalize_whatsapp_message(record)
    rec["room_id"] = record.get("friendUid")
    rec["talker_id"] = record.get("chatRoomTalker")
    # 群行的 file 常带直连 OSS URL(yk-whatsapp.oss),优先于 1:1 用的 mediaName 不透明名
    rec["file_url"] = record.get("file") or record.get("mediaName")
    if not rec.get("msg_id"):
        rec["msg_id"] = (
            f"{record.get('uid')}:{record.get('friendUid')}:"
            f"{record.get('createTime')}:{record.get('chatRoomTalker')}"
        )
    return rec


async def scrape_wa_group_messages(cookie_str: str, max_groups: int = 500, full_history: bool = False):
    """WA 群消息:浏览器 fetch chatRecords,friendUid 直接传 @g.us 群 id
    (真机确认 2026-08-20,与微信群的 msg/chats+roomId 同套路)。

    群来自本地 chat_group(channel='whatsapp');翻页复用 paginate_chats
    (hasLast/end 游标);守护循环同微信群抓取器(毒群重启跳过)。
    """
    import time as _time

    from app.db.base import async_engine, async_session_maker
    from app.scraper.whatsapp import _is_wa_divider
    from app.storage.repos import upsert_messages
    from sqlalchemy import text as sa_text

    async with async_engine.connect() as conn:
        result = await conn.execute(sa_text(
            "SELECT owner_account_id, room_id FROM chat_group "
            "WHERE channel='whatsapp' AND owner_account_id IS NOT NULL"
        ))
        groups = [(row[0], row[1]) for row in result.fetchall()]

    room_wms = await _room_watermarks()

    logger.info(f"Discovered {len(groups)} WA groups, will scrape {min(max_groups, len(groups))}"
                + (" [FULL]" if full_history else " [incremental]"))

    total_fetched = 0
    total_upserted = 0
    total_pages = 0
    scan = groups[:max_groups]
    idx = 0
    relaunches = 0

    while idx < len(scan):
        try:
            async with PageScraper(cookie_str) as scraper:
                page = await _open_fetch_page(scraper)

                while idx < len(scan):
                    owner_id, room_id = scan[idx]
                    i = idx
                    records = []
                    pages = 0
                    try:
                        now_ms = int(_time.time() * 1000)
                        watermark = 0 if full_history else room_wms.get(room_id, 0)

                        async def _fetch(end_ms: int, _owner=owner_id, _room=room_id, _wm=watermark):
                            return await _browser_fetch(
                                page,
                                "/qqweb/WhatsAppUser/chatRecords",
                                json_body={"uid": _owner, "friendUid": _room, "start": _wm, "end": end_ms},
                            )

                        async for data in paginate_chats(_fetch, now_ms, max_pages=200):
                            pages += 1
                            if watermark and _page_min_ts(data) and _page_min_ts(data) <= watermark:
                                break
                            rows = data.get("messages") or []
                            for row in rows:
                                if not isinstance(row, dict) or _is_wa_divider(row):
                                    continue
                                records.append(_normalize_wa_group_message(row))

                    except Exception as e:
                        logger.warning(f"  [{i+1}] {str(owner_id)[:18]}:{str(room_id)[:18]} failed: {e}")
                        for attempt in range(30):
                            try:
                                try:
                                    await page.close()
                                except Exception:
                                    pass
                                page = await _open_fetch_page(scraper)
                                break
                            except Exception as rec:
                                logger.warning(f"  recovery {attempt + 1}/30 failed: {rec}")
                                if attempt == 29:
                                    raise
                                await asyncio.sleep(120)
                        idx += 1
                        continue

                    total_fetched += len(records)
                    total_pages += pages

                    if records:
                        async with async_session_maker() as session:
                            await upsert_messages(session, records)
                            await session.commit()
                        total_upserted += len(records)

                    if (i + 1) % 10 == 0:
                        logger.info(f"  [{i+1}/{len(scan)}] msgs total: {total_fetched}, pages: {total_pages}")

                    await asyncio.sleep(2 + (i % 3))
                    idx += 1

        except Exception as fatal:
            relaunches += 1
            if relaunches > 10:
                raise
            logger.warning(f"  browser-level failure, relaunch {relaunches}/10, skip poison group: {fatal}")
            idx += 1
            await asyncio.sleep(30)

    logger.info(f"DONE: wa-group-messages fetched={total_fetched}, upserted={total_upserted}, pages={total_pages}")
    return total_fetched, total_upserted


# ---------------------------------------------------------------------------
# Media downloads (OSS CDN — NOT API calls, safe from 风控)
# ---------------------------------------------------------------------------


async def scrape_media(cookie_str: str, batch_size: int = 50):
    """Download media artifacts into the local object store:

    * 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.
    """
    from app.config import get_settings
    from app.db.base import async_session_maker
    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
    from app.storage.object_store import ObjectStore

    settings = get_settings()
    cookie_dict = {c["name"]: c["value"] for c in parse_cookies(cookie_str)}
    # Session-mode client purely as an httpx transport for OSS downloads —
    # userid/company only matter for login(), which we never call here.
    client = YunkeClient(
        base_url=YUNKE_BASE,
        userid=settings.yk_userid or "oss",
        company=settings.yk_company or "oss",
        cookies=cookie_dict,
    )
    store = ObjectStore()
    totals: dict[str, tuple[int, int]] = {}
    try:
        async with async_session_maker() as session:
            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()

    for name, (attempted, localized) in totals.items():
        logger.info(f"  {name}: attempted={attempted}, localized={localized}")
    logger.info(f"DONE: media {totals}")
    return totals


async def main():
    parser = argparse.ArgumentParser(description="页面浏览器抓取器")
    parser.add_argument(
        "--entity",
        required=True,
        choices=[
            "friends",       # 微信客户 (per-employee friend pages)
            "groups",        # 微信群 (per-employee group pages)
            "messages",      # 微信消息 (browser fetch msg/chats)
            "group-messages", # 微信群消息 (msg/chats with friendId=roomId)
            "accounts",      # 微信员工账号 (wechatlist page interception)
            "wa-accounts",   # WA员工账号 (waBusiness/userlist interception)
            "wa-friends",    # WA客户 (waBusiness/friendlist interception)
            "wa-groups",     # WA群 (browser fetch businessFriendList type=1)
            "chatroom-stats", # 微信群统计 (getChatroomList 带窗,云客同款)
            "voice-calls",   # 语音通话记录 (voice-call-list + wavoicecalllist)
            "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/moments)")
    parser.add_argument("--batch-size", type=int, default=50, help="媒体下载批次大小 (media)")
    parser.add_argument("--full-history", action="store_true", help="消息类实体强制全量翻历史(默认增量水位)")
    parser.add_argument("--day", type=str, default=None, help="chatroom-stats:单日窗 YYYY-MM-DD(落按天进出表);缺省=全窗档案基线")
    args = parser.parse_args()

    cookie = _load_cookie()
    if not cookie:
        logger.error("YK_COOKIE not set in .env or environment")
        return

    logger.info(f"Starting page-based scrape: entity={args.entity}")

    async def _run(c: str) -> None:
        await _dispatch_entity(args, c)

    # cookie 过期自动续期:重登→落缓存→整实体重跑(2026-09-03)
    await _run_with_auto_relogin(_run, cookie)


async def _dispatch_entity(args, cookie: str) -> None:
    if args.entity == "friends":
        await scrape_friends(cookie, args.max_employees)
    elif args.entity == "groups":
        await scrape_groups(cookie, args.max_employees)
    elif args.entity == "messages":
        await scrape_messages(cookie, args.max_conversations, args.full_history)
    elif args.entity == "group-messages":
        await scrape_group_messages(cookie, args.max_conversations, args.full_history)
    elif args.entity == "accounts":
        await scrape_accounts(cookie, args.max_pages)
    elif args.entity == "wa-accounts":
        await scrape_wa_accounts(cookie, args.max_pages)
    elif args.entity == "wa-friends":
        await scrape_wa_friends(cookie, args.max_pages)
    elif args.entity == "wa-groups":
        await scrape_wa_groups(cookie, args.max_pages)
    elif args.entity == "chatroom-stats":
        await scrape_chatroom_stats(cookie, args.day)
    elif args.entity == "voice-calls":
        await scrape_voice_calls(cookie, args.max_pages)
    elif args.entity == "voice-clips":
        await scrape_voice_clips(cookie, args.max_employees)
    elif args.entity == "wa-messages":
        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)


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except CookieExpiredError as exc:
        # 一句话可读 + 专用退出码,调度器/脚本日志第一眼就能定位;不再输出天书堆栈
        print(f"[cookie-expired] {exc}", file=sys.stderr)
        sys.exit(3)