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 media

Cookie 从 .env 的 YK_COOKIE 读取。
"""
from __future__ import annotations

import argparse
import asyncio
import json
import logging
import os
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
}


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
    from sqlalchemy import text as sa_text
    from app.scraper.wechat import _normalize
    from app.storage.repos import 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)
            logger.info(
                f"  [{i+1}/{len(employees)}] {wx_id[:16]}..: declared={declared} "
                f"rows={len(rows)} (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 scrape_messages(cookie_str: str, max_conversations: int = 100):
    """Scrape wechat messages via browser fetch POST /wechatweb/msg/chats.

    Conversation pairs come from _discover_conversations (customer table ∪
    voice-call pairs). Pages backward through full history via data.end /
    data.hasLast (paginate_chats, same cursor strategy as the WA chatRecords
    scraper).
    """
    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")

    logger.info(f"Discovered {len(conversations)} conversations, will scrape {min(max_conversations, len(conversations))}")

    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)

                async def _fetch(end_ms: int, _owner=owner_id, _friend=friend_id):
                    return await page.evaluate("""async ([ownerId, friendId, endMs]) => {
                        const params = new URLSearchParams();
                        params.append('wechatId', ownerId);
                        params.append('friendId', friendId);
                        params.append('types', '');
                        params.append('start', '0');
                        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, end_ms])

                # 翻页取完整历史：hasLast 真 → 用 data.end-1 继续向前翻（页间 sleep 1s）
                async for data in paginate_chats(_fetch, now_ms):
                    pages += 1
                    # 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.get("isSend", 0)) if row.get("isSend") is not None else None,
                            "type": _normalize_msg_type(row.get("type")),
                            "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.get("isSend", 0)) if row.get("isSend") is not None else None,
        "type": _normalize_msg_type(row.get("type")),
        "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):
    """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()]

    logger.info(f"Discovered {len(groups)} groups, will scrape {min(max_groups, len(groups))}")

    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)

                        async def _fetch(end_ms: int, _owner=owner_id, _room=room_id):
                            return await page.evaluate("""async ([ownerId, friendId, endMs]) => {
                                const params = new URLSearchParams();
                                params.append('wechatId', ownerId);
                                params.append('friendId', friendId);
                                params.append('types', '');
                                params.append('start', '0');
                                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, end_ms])

                        async for data in paginate_chats(_fetch, now_ms, max_pages=200):
                            pages += 1
                            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) -> str | None:
    """Normalize a raw message type to our canonical type string.

    微信 type=2 = 图片(新版,fileTh 带完整 OSS URL;真机 A**杰对话确认
    2026-08-20——旧映射漏掉它导致图片被存成 unknown 前端不渲染)。
    """
    if raw_type is None:
        return None
    t = str(raw_type)
    mapping = {
        "1": "text",
        "2": "image",          # 新版图片(fileTh URL)
        "3": "image_legacy",   # 旧版图片(md5/键)
        "34": "voice",
        "43": "video",
        "47": "emoji",
        "49": "file",
    }
    return mapping.get(t, "unknown")


# ---------------------------------------------------------------------------
# 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.
    """
    if json_body is not None:
        return 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],
        )
    return 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 {}],
    )


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 = [_normalize_wxuser_account(row) for row in rows]
    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
    from sqlalchemy import text as sa_text
    from app.scraper.whatsapp import _normalize_whatsapp_friend
    from app.storage.repos import 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)
            logger.info(
                f"  [{i+1}/{len(accounts)}] {wa_uid[:18]}..: declared={declared} rows={len(rows)}"
            )
            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


# ---------------------------------------------------------------------------
# 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):
    """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")

    logger.info(f"Discovered {len(conversations)} WA conversations, will scrape {min(max_conversations, len(conversations))}")

    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 = []
            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": 0, "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
                    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):
    """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()]

    logger.info(f"Discovered {len(groups)} WA groups, will scrape {min(max_groups, len(groups))}")

    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)

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

                        async for data in paginate_chats(_fetch, now_ms, max_pages=200):
                            pages += 1
                            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)

    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_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)
    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)
            "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)
            "media",         # OSS媒体下载 (录音/长语音音频/消息图片)
        ],
        help="抓取实体",
    )
    parser.add_argument("--max-employees", type=int, default=100, help="最多遍历多少员工 (friends/groups/voice-clips)")
    parser.add_argument("--max-conversations", type=int, default=100, help="消息模式：最多遍历多少对话 (messages/wa-messages/group-messages)")
    parser.add_argument("--max-pages", type=int, default=200, help="列表翻页上限 (accounts/wa-accounts/wa-friends/voice-calls)")
    parser.add_argument("--batch-size", type=int, default=50, help="媒体下载批次大小 (media)")
    args = parser.parse_args()

    cookie = os.environ.get("YK_COOKIE", "")
    if not cookie:
        # Try reading from .env
        from app.config import get_settings
        cookie = get_settings().yk_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}")

    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)
    elif args.entity == "group-messages":
        await scrape_group_messages(cookie, args.max_conversations)
    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 == "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)
    elif args.entity == "wa-group-messages":
        await scrape_wa_group_messages(cookie, args.max_conversations)
    elif args.entity == "media":
        await scrape_media(cookie, args.batch_size)


if __name__ == "__main__":
    asyncio.run(main())