import asyncio import base64 import json import random import re import time class DoubaoAdapter: def __init__(self, page, cdp): self.page = page self.cdp = cdp self.captured = {} self.response_bodies = {} self._finished_event = asyncio.Event() self.captcha_decision = None self._popup_blocker_injected = False async def setup_intercept(self): self.captured.clear() self.response_bodies.clear() self._finished_event = asyncio.Event() self.captcha_decision = None def on_request(params): req = params.get("request", {}) url = req.get("url", "") if "chat/completion" in url and "doubao.com" in url: self.captured[params.get("requestId")] = { "url": url, "postData": req.get("postData", ""), "timestamp": time.time(), } async def on_finished(params): rid = params.get("requestId") if rid in self.captured: try: result = await self.cdp.send("Network.getResponseBody", {"requestId": rid}) body = result.get("body", "") if result.get("base64Encoded"): body = base64.b64decode(body).decode("utf-8", errors="replace") self.response_bodies[rid] = body self._extract_captcha_decision(body) if "710022004" in body: print(f" [DB] SSE 返回验证码拦截, 不标记完成", flush=True) else: self._finished_event.set() except Exception: pass self.cdp.on("Network.requestWillBeSent", on_request) self.cdp.on("Network.loadingFinished", lambda p: asyncio.create_task(on_finished(p))) def _extract_captcha_decision(self, body): if "710022004" not in body: return for line in body.split("\n"): line = line.strip() if not line.startswith("data:") and not line.startswith("{"): continue raw = line[5:].strip() if line.startswith("data:") else line try: data = json.loads(raw) if data.get("error_code") == 710022004: decision = data.get("decision") if not decision: extra = data.get("extra") if isinstance(extra, str): try: extra = json.loads(extra) except Exception: extra = {} if isinstance(extra, dict): rd = extra.get("decision") if isinstance(rd, str): try: decision = json.loads(rd) except Exception: decision = rd elif isinstance(rd, dict): decision = rd if decision: self.captcha_decision = decision return except Exception: pass async def _inject_popup_blocker(self): """注入 JS + CSS 从源头拦截所有弹窗/覆盖层""" await self.page.add_init_script("""() => { const style = document.createElement('style'); style.textContent = ` [class*="cornerPopup"], [class*="corner-popup"], [class*="guide-"], [class*="Guide"], [class*="onboarding"], [class*="Onboarding"], [class*="tooltip-popup"], [class*="notice-bar"], [class*="upgrade-banner"], [class*="download-banner"], .semi-modal-wrap, .semi-modal-mask, [class*="FloatBanner"], [class*="float-banner"] { display: none !important; visibility: hidden !important; opacity: 0 !important; pointer-events: none !important; } `; (document.head || document.documentElement).appendChild(style); const BLOCKED_CLASSES = /cornerPopup|corner-popup|guide-|Guide|onboarding|Onboarding|tooltip-popup|notice-bar|upgrade-banner|download-banner|FloatBanner|float-banner/; const observer = new MutationObserver(mutations => { for (const m of mutations) { for (const node of m.addedNodes) { if (node.nodeType !== 1) continue; const cls = node.className ? node.className.toString() : ''; if (BLOCKED_CLASSES.test(cls)) { node.remove(); continue; } const cs = getComputedStyle(node); if ((cs.position === 'fixed' || cs.position === 'absolute') && parseInt(cs.zIndex) >= 1000 && node.offsetWidth > window.innerWidth * 0.5 && node.offsetHeight > window.innerHeight * 0.5) { const text = (node.innerText || '').substring(0, 200); if (text.includes('知道了') || text.includes('去设置') || text.includes('下载') || text.includes('升级') || text.includes('引导') || text.includes('水印')) { node.remove(); } } if (cls.includes('semi-modal-wrap') || cls.includes('semi-modal-mask')) { node.remove(); document.body.style.overflow = ''; } } } }); observer.observe(document.documentElement, {childList: true, subtree: true}); const origOverflow = Object.getOwnPropertyDescriptor(CSSStyleDeclaration.prototype, 'overflow'); if (origOverflow && origOverflow.set) { Object.defineProperty(document.body.style, 'overflow', { set(v) { if (v === 'hidden' || v === 'clip') return; origOverflow.set.call(this, v); }, get() { return origOverflow.get.call(this); }, configurable: true, }); } }""") print(" [doubao] 弹窗拦截已注入") async def _physical_click(self, locator): box = await locator.bounding_box() if not box: await locator.click() return x = box['x'] + box['width'] * random.uniform(0.3, 0.7) y = box['y'] + box['height'] * random.uniform(0.3, 0.7) await self.page.mouse.move(x, y, steps=random.randint(5, 12)) await asyncio.sleep(random.uniform(0.05, 0.15)) await self.page.mouse.click(x, y) async def _physical_type(self, locator, text): await self._physical_click(locator) await asyncio.sleep(random.uniform(0.3, 0.6)) for char in text: await self.page.keyboard.type(char, delay=random.randint(40, 120)) if random.random() > 0.9: await asyncio.sleep(random.uniform(0.2, 0.4)) async def new_chat(self): if not self._popup_blocker_injected: await self._inject_popup_blocker() self._popup_blocker_injected = True await self.page.goto("https://www.doubao.com/chat", wait_until="domcontentloaded", timeout=60000) await self.page.wait_for_selector( "textarea[placeholder*='发消息']", state="visible", timeout=15000 ) await asyncio.sleep(random.uniform(1, 2)) async def set_deep_thinking(self, enabled): target = "专家" if enabled else "快速" for attempt in range(3): try: trigger = self.page.locator( 'button[data-slot="dropdown-menu-trigger"]' ).filter(has_text=re.compile(r"快速|专家|办公")).first await trigger.wait_for(state="visible", timeout=15000) current = (await trigger.inner_text() or "").split("\n")[0].strip() if target in current: return await self._physical_click(trigger) await asyncio.sleep(random.uniform(0.5, 1.0)) item = self.page.locator('[role="menuitem"]').filter(has_text=target).first await item.wait_for(state="visible", timeout=8000) await self._physical_click(item) await asyncio.sleep(random.uniform(0.5, 1.0)) return except Exception as e: if attempt < 2: print(f" [WARN] 豆包模式切换重试 {attempt+1}/3: {e}") await asyncio.sleep(2) else: print(f" [WARN] 豆包模式切换失败: {e}") async def send_question(self, question): ta = self.page.locator( "textarea[placeholder*='发消息']:not([aria-hidden='true'])" ).first await ta.wait_for(state="visible", timeout=5000) await self._physical_type(ta, question) await asyncio.sleep(random.uniform(0.3, 0.5)) input_val = await ta.input_value() if len(input_val) < len(question) * 0.5: await ta.fill('') await ta.fill(question) await asyncio.sleep(0.3) try: btn = self.page.locator('button[class*="send-msg-btn"]').first await btn.wait_for(state="visible", timeout=2000) if not await btn.is_disabled(): await self._physical_click(btn) return except Exception: pass await self.page.keyboard.press("Enter") async def wait_done(self, timeout=180): try: await asyncio.wait_for(self._finished_event.wait(), timeout=timeout) except asyncio.TimeoutError: print(f" [WARN] 豆包响应超时 ({timeout}s)") return await asyncio.sleep(1) for _ in range(30): try: streaming = await self.page.evaluate("""() => { var el = document.querySelector('div[class*="md-box-root"][data-streaming]'); return el ? el.getAttribute('data-streaming') : null; }""") if streaming == "false": break except Exception: pass await asyncio.sleep(1) await asyncio.sleep(1) def get_last_response(self): if not self.response_bodies: return None, None last_rid = max(self.response_bodies.keys(), key=lambda r: self.captured.get(r, {}).get("timestamp", 0)) return self.response_bodies[last_rid], self.captured[last_rid].get("postData", "")