From 8a8c91a6a3ba4fc8be341c5caac7e458876d20af Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Thu, 20 Aug 2026 14:46:02 +0800 Subject: [PATCH] =?UTF-8?q?fix(customers):=20=E5=AE=A2=E6=88=B7=E5=88=97?= =?UTF-8?q?=E8=A1=A8=E6=8E=92=E5=BA=8F=E5=AE=9E=E6=97=B6=E5=AF=B9=E9=BD=90?= =?UTF-8?q?=E4=BA=91=E5=AE=A2(=E5=85=A8=E5=B1=80=E5=AE=9E=E6=97=B6body?= =?UTF-8?q?=E5=88=B7=E6=96=B0last=5Fchat=5Ftime,=E5=89=8D5=E4=BD=8D?= =?UTF-8?q?=E9=80=90=E4=BD=8D=E4=B8=80=E8=87=B4)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- yunke-server/app/sync/scheduler.py | 29 ++++++++---- yunke-server/scripts/scrape_via_page.py | 59 +++++++++++++++++++++++++ 2 files changed, 79 insertions(+), 9 deletions(-) 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