diff --git a/yunke-server/scripts/scrape_via_page.py b/yunke-server/scripts/scrape_via_page.py index 12c59ac..bde5efb 100644 --- a/yunke-server/scripts/scrape_via_page.py +++ b/yunke-server/scripts/scrape_via_page.py @@ -256,108 +256,68 @@ class PageScraper: async def scrape_friends(cookie_str: str, max_employees: int = 100): - """Scrape wechat customers (friends) via page navigation.""" - from app.db.base import async_session_maker + """微信客户:浏览器直发 friend_list/es 部门级全量翻页(2026-08-20 起, + 旧导航+拦截被登录墙杀死 0 捕获)。记录自带 owner(wxId),无需逐员工遍历; + ``max_employees`` 参数保留兼容 CLI,已不使用。""" + from app.scraper.wechat import _normalize from app.storage.repos import upsert_customer - from app.scraper.wechat import _normalize # reuse the existing normalizer - total_fetched = 0 - total_upserted = 0 + dept = await _root_dept_id() + base_body = {"departmentId": dept} if dept else {} async with PageScraper(cookie_str) as scraper: - employees = await scraper.get_employee_links() + page = await _open_fetch_page(scraper) + rows = await _fetch_pages( + page, INTERCEPT_PATHS["friend_list"], + base_body=base_body, parse_rows=_rows_basic_result, + ) - for i, emp in enumerate(employees[:max_employees]): - if emp["friendCount"] == 0: - continue - wx_id = emp["wxId"] - url = f"{YUNKE_BASE}/cms/wechat/wechatlist/wechatdetail/1/{wx_id}?ymdStart=0000-00-00&ymdEnd=2099-12-31" - logger.info(f"[{i+1}/{len(employees)}] {wx_id}: {emp['friendCount']} friends") + records = [] + for row in rows: + rec = _normalize(row) + if rec.get("friend_id"): + records.append(rec) - responses = await scraper.scrape_entity_pages( - list_url=url, - api_path=INTERCEPT_PATHS["friend_list"], - max_pages=200, - ) - - # Extract friend records from captured responses - records = [] - for resp in responses: - body = resp.get("body", {}) - data = body.get("data", {}) - rows = data.get("basicResult", {}).get("data", []) or data.get("data", []) - for row in rows: - if isinstance(row, dict): - normalized = _normalize(row) - normalized["channel"] = "wechat" - normalized["owner_account_id"] = wx_id - records.append(normalized) - - total_fetched += len(records) - - # Batch upsert - if records: - async with async_session_maker() as session: - for rec in records: - await upsert_customer(session, rec) - await session.commit() - total_upserted += len(records) - logger.info(f" upserted {len(records)} customers (total: {total_upserted})") - - # Human-like delay between employees - await asyncio.sleep(3 + (i % 3)) # 3-5s random-ish + total_upserted = 0 + for i in range(0, len(records), 500): + await _upsert_all(upsert_customer, records[i:i + 500]) + total_upserted += len(records[i:i + 500]) + total_fetched = len(records) 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): - """Scrape wechat groups via page navigation.""" - from app.db.base import async_session_maker - from app.storage.repos import upsert_chat_group + """微信群:浏览器直发 friend_list/es + friendTypeYunke=2 部门级全量翻页 + (与好友共用端点,仅 body 开关不同;2026-08-20 起改 fetch)。""" from app.scraper.wechat import _normalize_chat_group + from app.storage.repos import upsert_chat_group - total_fetched = 0 - total_upserted = 0 + dept = await _root_dept_id() + base_body: dict = {"friendTypeYunke": "2"} + if dept: + base_body["departmentId"] = dept async with PageScraper(cookie_str) as scraper: - employees = await scraper.get_employee_links() + page = await _open_fetch_page(scraper) + rows = await _fetch_pages( + page, INTERCEPT_PATHS["friend_list"], + base_body=base_body, parse_rows=_rows_basic_result, + ) - for i, emp in enumerate(employees[:max_employees]): - if emp.get("groupCount", 0) == 0 or not emp.get("groupHref"): - continue - wx_id = emp["wxId"] - url = f"{YUNKE_BASE}{emp['groupHref']}" - logger.info(f"[{i+1}/{len(employees)}] {wx_id}: {emp['groupCount']} groups") + records = [] + for row in rows: + rec = _normalize_chat_group(row) + if rec.get("room_id"): + records.append(rec) - responses = await scraper.scrape_entity_pages( - list_url=url, - api_path=INTERCEPT_PATHS["friend_list"], - max_pages=100, - ) - - records = [] - for resp in responses: - body = resp.get("body", {}) - data = body.get("data", {}) - rows = data.get("basicResult", {}).get("data", []) or data.get("data", []) - for row in rows: - if isinstance(row, dict): - normalized = _normalize_chat_group(row) - normalized["channel"] = "wechat" - records.append(normalized) - - total_fetched += len(records) - if records: - async with async_session_maker() as session: - for rec in records: - await upsert_chat_group(session, rec) - await session.commit() - total_upserted += len(records) - logger.info(f" upserted {len(records)} groups (total: {total_upserted})") - - await asyncio.sleep(3 + (i % 3)) + total_upserted = 0 + for i in range(0, len(records), 500): + await _upsert_all(upsert_chat_group, records[i:i + 500]) + total_upserted += len(records[i:i + 500]) + total_fetched = len(records) logger.info(f"DONE: groups fetched={total_fetched}, upserted={total_upserted}") return total_fetched, total_upserted @@ -835,6 +795,80 @@ def _rows_flat(body: dict) -> list[dict]: 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).""" @@ -863,91 +897,109 @@ async def _upsert_all(upsert_fn, records: list[dict]) -> None: async def scrape_accounts(cookie_str: str, max_pages: int = 100): - """Scrape WeChat staff accounts: navigate /cms/wechat/wechatlist, - intercept /wechatweb/wxUser/list, upsert to wechat_account (+staff).""" + """微信员工账号:浏览器直发 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 - total_fetched = 0 + 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: - responses = await scraper.scrape_entity_pages( - list_url=f"{YUNKE_BASE}/cms/wechat/wechatlist", - api_path=INTERCEPT_PATHS["wxUser_list"], + 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 resp in responses: - for row in _rows_flat(resp.get("body", {})): - records.append(_normalize_wxuser_account(row)) - total_fetched = len(records) - await _upsert_all(upsert_wechat_account, records) - logger.info(f"DONE: accounts fetched={total_fetched}, upserted={total_fetched}") - return total_fetched, total_fetched + 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 (navigation + interception) +# WhatsApp staff accounts (browser fetch) # --------------------------------------------------------------------------- async def scrape_wa_accounts(cookie_str: str, max_pages: int = 100): - """Scrape WA staff accounts: navigate /cms/wechat/waBusiness/userlist, - intercept /qqweb/WhatsAppUser/businessSalesList, upsert to whatsapp_account.""" + """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: - responses = await scraper.scrape_entity_pages( - list_url=f"{YUNKE_BASE}/cms/wechat/waBusiness/userlist", - api_path=INTERCEPT_PATHS["wa_user"], + 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 = [] - for resp in responses: - for row in _rows_overviews(resp.get("body", {})): - records.append(_normalize_whatsapp_account(row)) - total = len(records) - await _upsert_all(upsert_whatsapp_account, records) + 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 (navigation + interception) +# WhatsApp customers (browser fetch) # --------------------------------------------------------------------------- async def scrape_wa_friends(cookie_str: str, max_pages: int = 2000): - """Scrape WA customers: navigate /cms/wechat/waBusiness/friendlist, - intercept /qqweb/WhatsAppUser/businessFriendList, upsert to customer - (channel='whatsapp'). Live total ~11k rows => high default max_pages.""" + """WA 客户:浏览器直发 businessFriendList type=0(与 WA 群共用端点, + 2026-08-20 起改 fetch;注意此端点页码字段是 ``page`` 非 ``pageNum``)。""" from app.scraper.whatsapp import _normalize_whatsapp_friend from app.storage.repos import upsert_customer - total_fetched = 0 - total_upserted = 0 - async with PageScraper(cookie_str) as scraper: - responses = await scraper.scrape_entity_pages( - list_url=f"{YUNKE_BASE}/cms/wechat/waBusiness/friendlist", - api_path=INTERCEPT_PATHS["wa_friend"], - max_pages=max_pages, - ) - # Upsert in chunks so one giant page-walk doesn't build a 11k-record list - # before the first commit. - chunk: list[dict] = [] - for resp in responses: - records = [_normalize_whatsapp_friend(row) for row in _rows_overviews(resp.get("body", {}))] - total_fetched += len(records) - chunk.extend(records) - if len(chunk) >= 500: - await _upsert_all(upsert_customer, chunk) - total_upserted += len(chunk) - chunk = [] - await _upsert_all(upsert_customer, chunk) - total_upserted += len(chunk) + dept = await _root_dept_id() + base_body: dict = { + "searchInfo": "", + "friendSearchInfo": "", + "departmentId": dept, + "type": "0", + } + async with PageScraper(cookie_str) as scraper: + page = await _open_fetch_page(scraper) + rows = await _fetch_pages( + page, INTERCEPT_PATHS["wa_friend"], + base_body=base_body, parse_rows=_rows_overviews, + page_key="page", max_pages=max_pages, + ) + + records = [] + for row in rows: + rec = _normalize_whatsapp_friend(row) + if rec.get("friend_id"): + records.append(rec) + + total_upserted = 0 + for i in range(0, len(records), 500): + await _upsert_all(upsert_customer, records[i:i + 500]) + total_upserted += len(records[i:i + 500]) + + total_fetched = len(records) logger.info(f"DONE: wa-friends fetched={total_fetched}, upserted={total_upserted}") return total_fetched, total_upserted