xinyouhe/yunke
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(sync): 消息全面增量同步(水位+跳过+触底停,4h周期从小时级降到分钟级)

Browse Source 
- 每对话水位=本地max(timestamp);请求带start=水位只拉新消息
- 云客last_chat_time≤水位的对话整跳过(0请求)——实测7046对话3150被跳过(45%)
- 翻页遇≤水位消息即停(已同步区),页数从几十页降到1页
- 微信1:1/WA 1:1/微信群/WA群四个抓取器全接入;CLI --full-history强制全量
- 实测:每对话1页(~3s),3895待访对话约3.5h(含补积压的首次全量);
  稳态后每4h只访有新活动的对话(几十~几百个,15-30min)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
浅念
2026-08-21 12:20:21 +08:00
parent 10baa22acb
commit 9bca5fc08a
1 changed files with 116 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
yunke-server/scripts/scrape_via_page.py
+116 -27
View File
@@ -571,13 +571,47 @@ async def _discover_conversations(channel: str, voice_channel: str) -> list[tupl
return fresh + known
async def scrape_messages(cookie_str: str, max_conversations: int = 100):
async def _conversation_watermarks(channel: str) -> dict[tuple[str, str], int]:
"""每对话的本地最新消息时间(ms)——增量同步水位。"""
from app.db.base import async_engine
from sqlalchemy import text as sa_text
async with async_engine.connect() as conn:
result = await conn.execute(sa_text(
"SELECT owner_account_id, talker_id, max(timestamp) FROM message "
"WHERE channel=:ch AND (room_id IS NULL OR room_id='') "
"AND talker_id IS NOT NULL AND talker_id != '' "
"GROUP BY owner_account_id, talker_id"
), {"ch": channel})
return {(r[0], r[1]): r[2] or 0 for r in result.fetchall()}
async def _customer_activity(channel: str) -> dict[tuple[str, str], int]:
"""客户表 last_chat_time(4h 增量的 friends 实时刷新)——判断对话有无新消息。"""
from app.db.base import async_engine
from sqlalchemy import text as sa_text
async with async_engine.connect() as conn:
result = await conn.execute(sa_text(
"SELECT owner_account_id, friend_id, last_chat_time FROM customer "
"WHERE channel=:ch AND owner_account_id IS NOT NULL AND friend_id IS NOT NULL"
), {"ch": channel})
return {(r[0], r[1]): r[2] or 0 for r in result.fetchall()}
def _page_min_ts(data: dict) -> int:
"""页内最早消息时间(0 当无有效 ts)。"""
ts_list = [m.get("timestamp") for m in (data.get("messages") or []) if isinstance(m, dict) and m.get("timestamp")]
return min(ts_list) if ts_list else 0
async def scrape_messages(cookie_str: str, max_conversations: int = 100, full_history: bool = False):
"""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).
**增量同步(默认)**:每对话水位 = 本地 max(timestamp);请求带
``start=水位`` 只拉新消息,翻页遇到 ≤水位的消息即停(已同步区)。
客户表 last_chat_time(friends 实时刷新)≤ 水位的对话**整跳过**(0 请求)
——活跃对话才访问。``full_history=True`` 强制全量(首次/修复用)。
"""
from sqlalchemy import text as sa_text
@@ -585,8 +619,25 @@ async def scrape_messages(cookie_str: str, max_conversations: int = 100):
from app.storage.repos import upsert_messages
conversations = await _discover_conversations("wechat", "voice_call")
watermarks = await _conversation_watermarks("wechat")
cust_activity = await _customer_activity("wechat")
logger.info(f"Discovered {len(conversations)} conversations, will scrape {min(max_conversations, len(conversations))}")
# 增量模式:云客侧无新活动的对话跳过(0 请求)——速度来源
skipped = 0
if not full_history:
kept = []
for pair in conversations:
wm = watermarks.get(pair, 0)
act = cust_activity.get(pair, 0)
if wm > 0 and act > 0 and act <= wm + 60_000:
skipped += 1
else:
kept.append(pair)
conversations = kept
logger.info(
f"Discovered conversations: {len(conversations)} to visit, {skipped} skipped (no new activity)"
+ (" [FULL HISTORY]" if full_history else " [incremental]")
)
total_fetched = 0
total_upserted = 0
@@ -611,13 +662,15 @@ async def scrape_messages(cookie_str: str, max_conversations: int = 100):
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]) => {
watermark = 0 if full_history else watermarks.get((owner_id, friend_id), 0)
async def _fetch(end_ms: int, _owner=owner_id, _friend=friend_id, _wm=watermark):
return await page.evaluate("""async ([ownerId, friendId, startMs, endMs]) => {
const params = new URLSearchParams();
params.append('wechatId', ownerId);
params.append('friendId', friendId);
params.append('types', '');
params.append('start', '0');
params.append('start', String(startMs));
params.append('end', String(endMs));
const resp = await fetch('/wechatweb/msg/chats', {
method: 'POST',
@@ -625,11 +678,14 @@ async def scrape_messages(cookie_str: str, max_conversations: int = 100):
body: params.toString(),
});
return await resp.json();
}""", [_owner, _friend, end_ms])
}""", [_owner, _friend, _wm, end_ms])
# 翻页取完整历史：hasLast 真 → 用 data.end-1 继续向前翻（页间 sleep 1s）
async for data in paginate_chats(_fetch, now_ms):
pages += 1
# 增量触底停:页内出现 ≤水位 的消息 = 已同步区,不再向前翻
if watermark and _page_min_ts(data) and _page_min_ts(data) <= watermark:
break
# Debug: log first response structure
if i == 0 and pages == 1:
logger.info(f" DEBUG first response keys: {list(data.keys())}")
@@ -764,7 +820,7 @@ def _normalize_group_message(row: dict, owner_id: str, room_id: str) -> dict | N
}
async def scrape_group_messages(cookie_str: str, max_groups: int = 500):
async def scrape_group_messages(cookie_str: str, max_groups: int = 500, full_history: bool = False):
"""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
@@ -787,7 +843,10 @@ async def scrape_group_messages(cookie_str: str, max_groups: int = 500):
))
groups = [(row[0], row[1]) for row in result.fetchall()]
logger.info(f"Discovered {len(groups)} groups, will scrape {min(max_groups, len(groups))}")
room_wms = await _room_watermarks()
logger.info(f"Discovered {len(groups)} groups, will scrape {min(max_groups, len(groups))}"
+ (" [FULL]" if full_history else " [incremental]"))
total_fetched = 0
total_upserted = 0
@@ -814,14 +873,15 @@ async def scrape_group_messages(cookie_str: str, max_groups: int = 500):
pages = 0
try:
now_ms = int(_time.time() * 1000)
watermark = 0 if full_history else room_wms.get(room_id, 0)
async def _fetch(end_ms: int, _owner=owner_id, _room=room_id):
return await page.evaluate("""async ([ownerId, friendId, endMs]) => {
async def _fetch(end_ms: int, _owner=owner_id, _room=room_id, _wm=watermark):
return await page.evaluate("""async ([ownerId, friendId, startMs, endMs]) => {
const params = new URLSearchParams();
params.append('wechatId', ownerId);
params.append('friendId', friendId);
params.append('types', '');
params.append('start', '0');
params.append('start', String(startMs));
params.append('end', String(endMs));
const resp = await fetch('/wechatweb/msg/chats', {
method: 'POST',
@@ -829,10 +889,12 @@ async def scrape_group_messages(cookie_str: str, max_groups: int = 500):
body: params.toString(),
});
return await resp.json();
}""", [_owner, _room, end_ms])
}""", [_owner, _room, _wm, end_ms])
async for data in paginate_chats(_fetch, now_ms, max_pages=200):
pages += 1
if watermark and _page_min_ts(data) and _page_min_ts(data) <= watermark:
break
rows = data.get("messages") or []
for row in rows:
if not isinstance(row, dict):
@@ -1499,7 +1561,7 @@ async def scrape_voice_clips(cookie_str: str, max_employees: int = 200, page_siz
# ---------------------------------------------------------------------------
async def scrape_wa_messages(cookie_str: str, max_conversations: int = 100):
async def scrape_wa_messages(cookie_str: str, max_conversations: int = 100, full_history: bool = False):
"""Scrape WhatsApp chat records via browser fetch POST
/qqweb/WhatsAppUser/chatRecords (JSON body {uid, friendUid, start, end}).
@@ -1515,8 +1577,24 @@ async def scrape_wa_messages(cookie_str: str, max_conversations: int = 100):
from app.storage.repos import upsert_messages
conversations = await _discover_conversations("whatsapp", "wa_voice_call")
watermarks = await _conversation_watermarks("whatsapp")
cust_activity = await _customer_activity("whatsapp")
logger.info(f"Discovered {len(conversations)} WA conversations, will scrape {min(max_conversations, len(conversations))}")
skipped = 0
if not full_history:
kept = []
for pair in conversations:
wm = watermarks.get(pair, 0)
act = cust_activity.get(pair, 0)
if wm > 0 and act > 0 and act <= wm + 60_000:
skipped += 1
else:
kept.append(pair)
conversations = kept
logger.info(
f"Discovered WA conversations: {len(conversations)} to visit, {skipped} skipped (no new activity)"
+ (" [FULL]" if full_history else " [incremental]")
)
total_fetched = 0
total_upserted = 0
@@ -1526,13 +1604,14 @@ async def scrape_wa_messages(cookie_str: str, max_conversations: int = 100):
for i, (uid, friend_uid) in enumerate(conversations[:max_conversations]):
records = []
watermark = 0 if full_history else watermarks.get((uid, friend_uid), 0)
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},
json_body={"uid": uid, "friendUid": friend_uid, "start": watermark, "end": end_ms},
)
data = body.get("data") if isinstance(body, dict) else None
if not isinstance(data, dict):
@@ -1543,6 +1622,9 @@ async def scrape_wa_messages(cookie_str: str, max_conversations: int = 100):
records.append(_normalize_whatsapp_message(row))
if not data.get("hasLast"):
break
# 增量触底停:翻到已同步区
if watermark and _page_min_ts(data) and _page_min_ts(data) <= watermark:
break
end = data.get("end")
if not isinstance(end, (int, float)):
break
@@ -1614,7 +1696,7 @@ def _normalize_wa_group_message(record: dict) -> dict:
return rec
async def scrape_wa_group_messages(cookie_str: str, max_groups: int = 500):
async def scrape_wa_group_messages(cookie_str: str, max_groups: int = 500, full_history: bool = False):
"""WA 群消息:浏览器 fetch chatRecords,friendUid 直接传 @g.us 群 id
(真机确认 2026-08-20,与微信群的 msg/chats+roomId 同套路)。
@@ -1635,7 +1717,10 @@ async def scrape_wa_group_messages(cookie_str: str, max_groups: int = 500):
))
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))}")
room_wms = await _room_watermarks()
logger.info(f"Discovered {len(groups)} WA groups, will scrape {min(max_groups, len(groups))}"
+ (" [FULL]" if full_history else " [incremental]"))
total_fetched = 0
total_upserted = 0
@@ -1656,16 +1741,19 @@ async def scrape_wa_group_messages(cookie_str: str, max_groups: int = 500):
pages = 0
try:
now_ms = int(_time.time() * 1000)
watermark = 0 if full_history else room_wms.get(room_id, 0)
async def _fetch(end_ms: int, _owner=owner_id, _room=room_id):
async def _fetch(end_ms: int, _owner=owner_id, _room=room_id, _wm=watermark):
return await _browser_fetch(
page,
"/qqweb/WhatsAppUser/chatRecords",
json_body={"uid": _owner, "friendUid": _room, "start": 0, "end": end_ms},
json_body={"uid": _owner, "friendUid": _room, "start": _wm, "end": end_ms},
)
async for data in paginate_chats(_fetch, now_ms, max_pages=200):
pages += 1
if watermark and _page_min_ts(data) and _page_min_ts(data) <= watermark:
break
rows = data.get("messages") or []
for row in rows:
if not isinstance(row, dict) or _is_wa_divider(row):
@@ -1794,6 +1882,7 @@ async def main():
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)")
parser.add_argument("--full-history", action="store_true", help="消息类实体强制全量翻历史(默认增量水位)")
args = parser.parse_args()
cookie = os.environ.get("YK_COOKIE", "")
@@ -1812,9 +1901,9 @@ async def main():
elif args.entity == "groups":
await scrape_groups(cookie, args.max_employees)
elif args.entity == "messages":
await scrape_messages(cookie, args.max_conversations)
await scrape_messages(cookie, args.max_conversations, args.full_history)
elif args.entity == "group-messages":
await scrape_group_messages(cookie, args.max_conversations)
await scrape_group_messages(cookie, args.max_conversations, args.full_history)
elif args.entity == "accounts":
await scrape_accounts(cookie, args.max_pages)
elif args.entity == "wa-accounts":
@@ -1828,9 +1917,9 @@ async def main():
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)
await scrape_wa_messages(cookie, args.max_conversations, args.full_history)
elif args.entity == "wa-group-messages":
await scrape_wa_group_messages(cookie, args.max_conversations)
await scrape_wa_group_messages(cookie, args.max_conversations, args.full_history)
elif args.entity == "media":
await scrape_media(cookie, args.batch_size)