fix(customers): 客户列表排序实时对齐云客(全局实时body刷新last_chat_time,前5位逐位一致)

- scrape_friends 双阶段:逐员工全量保底 + 全局实时body(friendTab=all/isOnJob/ymd窗)
  刷 last_chat_time(该参数集跨员工按最近沟通实时排序,实测比**赐14:10实时值)
- 阶段2只刷last_chat_time且只增不减(全局结果含旧ES副本,upsert全字段会被后写覆盖)
- 调度4h增量改为 friends→wa-friends→messages(排序键每4h实时化)
- 实测我们top5=云客top5逐位一致(A**杰14:13/比**赐14:10/P**6/A**邱/J**n)

Co-Authored-By: Claude <noreply@anthropic.com>

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 = [
+                # 先刷客户列表(最近沟通时间是云客列表的实时排序键,friends 快照
-                    sys.executable,
+                # 滞后会让本地排序与云客第一页完全不同,2026-08-20),再抓消息增量
-                    "scripts/scrape_via_page.py",
+                commands = [
-                    "--entity",
+                    [sys.executable, "scripts/scrape_via_page.py", "--entity", "friends"],
-                    "messages",
+                    [sys.executable, "scripts/scrape_via_page.py", "--entity", "wa-friends"],
-                    "--max-conversations",
+                    [
-                    str(self.max_conversations_per_run),
+                        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(
                 {

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