diff --git a/yunke-server/app/sync/scheduler.py b/yunke-server/app/sync/scheduler.py index 1d665ed..3a9333f 100644 --- a/yunke-server/app/sync/scheduler.py +++ b/yunke-server/app/sync/scheduler.py @@ -579,21 +579,32 @@ class SyncScheduler: if not bash: logger.error("page sync 'full': bash not found on PATH") return - args: list[str] = [bash, "scripts/full_sync_all.sh"] + commands: list[list[str]] = [[bash, "scripts/full_sync_all.sh"]] elif kind == "messages": - args = [ - sys.executable, - "scripts/scrape_via_page.py", - "--entity", - "messages", - "--max-conversations", - str(self.max_conversations_per_run), + # 先刷客户列表(最近沟通时间是云客列表的实时排序键,friends 快照 + # 滞后会让本地排序与云客第一页完全不同,2026-08-20),再抓消息增量 + commands = [ + [sys.executable, "scripts/scrape_via_page.py", "--entity", "friends"], + [sys.executable, "scripts/scrape_via_page.py", "--entity", "wa-friends"], + [ + sys.executable, + "scripts/scrape_via_page.py", + "--entity", + "messages", + "--max-conversations", + str(self.max_conversations_per_run), + ], ] else: logger.error("page sync: unknown kind %r", kind) return - rc = await self._page_subprocess(args) + rc = 0 + for args in commands: + rc = await self._page_subprocess(args) or rc + if rc != 0: + logger.warning("page sync %s: command failed rc=%s, skipping rest", kind, rc) + break finished_at = dt.datetime.now(dt.UTC) line = json.dumps( { diff --git a/yunke-server/scripts/scrape_via_page.py b/yunke-server/scripts/scrape_via_page.py index b6b86c5..bb8acc0 100644 --- a/yunke-server/scripts/scrape_via_page.py +++ b/yunke-server/scripts/scrape_via_page.py @@ -282,6 +282,8 @@ async def scrape_friends(cookie_str: str, max_employees: int = 100): 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 @@ -334,6 +336,63 @@ async def scrape_friends(cookie_str: str, max_employees: int = 100): ) 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