import random import signal import socket import sys import time import base64 import requests from DrissionPage import ChromiumPage, ChromiumOptions from commons.Logger import logger from oss_upload.oss_upload import AliyunOSSUploader CAPTCHA_TOKEN = "zPzmt1mG1ouCU6GTzsZN2Lmm8pdZypapPcLJTBRETco" CAPTCHA_API_URL = "http://api.jfbym.com/api/YmServer/customApi" SLIDER_OFFSET_FIX = 10 chrome_path = r"C:\Program Files\Google\Chrome\Application\chrome.exe" class YaoShiBangCrawl: def __init__(self, product=None): self.product = product self.driver = None self.account_name = "ysbang_1" self.ossuploader = AliyunOSSUploader() self._register_signal_handler() def _register_signal_handler(self): def handler(signum, frame): logger.info("收到退出信号，正在关闭浏览器...") self._quit_browser() sys.exit(0) signal.signal(signal.SIGINT, handler) if hasattr(signal, "SIGTERM"): signal.signal(signal.SIGTERM, handler) def _quit_browser(self): if self.driver: try: self.driver.quit() except Exception: pass self.driver = None @staticmethod def _get_free_port(): """获取一个当前可用的本地端口，供 Chrome 调试使用。""" with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.bind(("127.0.0.1", 0)) return s.getsockname()[1] def init_browser(self): co = ChromiumOptions().set_browser_path(chrome_path) debug_port = self._get_free_port() co.set_user_data_path(f"./spiders/yaoshibang/{self.account_name}") co.set_local_port(debug_port) co.set_argument(f"--remote-debugging-port={debug_port}") co.set_argument("--remote-debugging-address=127.0.0.1") # co.set_argument("--disable-blink-features=AutomationControlled") co.set_argument("--disable-dev-shm-usage") co.set_argument("--no-first-run") # 避免首次运行弹窗 co.set_argument("--no-default-browser-check") # 避免默认浏览器检查 self.driver = ChromiumPage(co) def _solve_slider_captcha(self): """检测并处理易盾滑块验证码，成功返回 True。""" self.driver.wait.doc_loaded() time.sleep(2) yidun = self.driver.ele("xpath://div[@class='yidun_modal']", timeout=3) if not yidun: return True logger.info("检测到滑块验证码，开始处理") jpg_bytes = yidun.get_screenshot(as_bytes="jpg") distance = self._call_captcha_api(jpg_bytes) if distance is None: logger.error("验证码识别失败") return False logger.info("滑块距离: %s", distance) slider = self.driver.ele( "xpath://div[contains(@class,'yidun_slider--hover')]", timeout=5 ) if not slider: logger.error("未找到滑块元素") return False self._simulate_slider_drag(slider, float(distance) + SLIDER_OFFSET_FIX) time.sleep(3) return True def _call_captcha_api(self, image_bytes): """调用云码平台识别滑块距离，失败返回 None。""" try: b64 = base64.b64encode(image_bytes).decode() resp = requests.post( CAPTCHA_API_URL, json={"token": CAPTCHA_TOKEN, "type": "22222", "image": b64}, headers={"Content-Type": "application/json"}, timeout=15, ).json() logger.info("验证码 API 返回: %s", resp) return resp["data"]["data"] except Exception as e: logger.exception("验证码 API 调用失败: %s", e) return None @staticmethod def _generate_human_track(distance): tracks = [] current = 0 mid = distance * 0.7 t = 0.2 v = 0 move_points = [] while current < mid: a = random.uniform(2, 4) v0 = v v = v0 + a * t move = v0 * t + 0.5 * a * t * t current += move move_points.append(move) while current < distance: a = -random.uniform(0.5, 1.5) v0 = v v = v0 + a * t if v < 0.5: v = 0.5 move = v0 * t + 0.5 * a * t * t current += move move_points.append(move) total_points = len(move_points) for i, move in enumerate(move_points): y_offset = random.randint(-2, 2) if i % random.randint(2, 4) == 0 else 0 if i < total_points * 0.3: duration = random.uniform(0.01, 0.03) elif i > total_points * 0.7: duration = random.uniform(0.03, 0.08) else: duration = random.uniform(0.02, 0.05) if random.random() < 0.05: duration += random.uniform(0.05, 0.1) tracks.append((move, y_offset, duration)) if random.random() < 0.7: tracks.append((-random.randint(1, 3), 0, 0.05)) return tracks def _simulate_slider_drag(self, slider_element, target_distance): self.driver.actions.move_to(slider_element).hold() for offset_x, offset_y, duration in self._generate_human_track(target_distance): self.driver.actions.move(offset_x, offset_y, duration=duration / 1000) self.driver.actions.release() def _is_logged_in(self): title = self.driver.ele( "xpath=//*[contains(text(),'广西好药师大药房连锁有限公司天峨远大药店')]", timeout=5, ) return bool(title) def login(self): logger.info("开始登录药师帮") self.driver.get("https://dian.ysbang.cn/#/login", timeout=15) self.driver.wait.doc_loaded(timeout=10) time.sleep(2) input_name = self.driver.ele("xpath://input[@name='userAccount']", timeout=5) if not input_name: logger.error("未找到账号输入框") return False input_name.input("13097980383") time.sleep(random.uniform(1.5, 2.5)) input_pass = self.driver.ele("xpath://input[@name='password']", timeout=5) if not input_pass: logger.error("未找到密码输入框") return False input_pass.input("a123456") time.sleep(random.uniform(1.5, 2.5)) login_btn = self.driver.ele("xpath://button[text()='登录']", timeout=5) if not login_btn: logger.error("未找到登录按钮") return False login_btn.click() time.sleep(3) for i in range(3): self._solve_slider_captcha() time.sleep(3) if self._is_logged_in(): logger.info("登录成功") cookies_list = self.driver.cookies() cookies_dict = {c['name']: c['value'] for c in cookies_list} print(cookies_dict) return True logger.error("登录后未检测到目标店铺名，登录可能失败") return False def get_snapshot(self, detail_url, item_id): self.driver.get(detail_url, timeout=15) self.driver.wait.doc_loaded(timeout=10) time.sleep(2) self._dismiss_popup_before_screenshot() ele = self.driver.ele("xpath=//div[@class='drug-shopping-wrap']", timeout=8) if not ele: ele = self.driver.ele("xpath=//div[@class='drug-info']", timeout=5) if not ele: logger.warning("未找到详情区域元素，跳过截图 item_id=%s", item_id) return "" jpg_bytes = ele.get_screenshot(as_bytes="jpg") img_url = self.ossuploader.upload_from_bytes(jpg_bytes, str(item_id)) logger.info("截图上传完成: %s", img_url) time.sleep(random.uniform(0.5, 1.5)) return img_url def _dismiss_popup_before_screenshot(self): """截图前关闭或隐藏营销弹窗，避免遮挡。""" close_locs = [ "xpath=//div[contains(@class,'dialog')]//i[contains(@class,'close')]", "xpath=//div[contains(@class,'popup')]//i[contains(@class,'close')]", "xpath=//div[contains(@class,'modal')]//i[contains(@class,'close')]", "xpath=//button[contains(@class,'close')]", "xpath=//span[text()='×']", "xpath=//*[contains(text(),'智能采购')]/ancestor::div[1]//*[contains(@class,'close')]", ] for loc in close_locs: try: btn = self.driver.ele(loc, timeout=0.5) if btn: btn.click() time.sleep(0.2) except Exception: pass try: # 兜底：隐藏常见高层弹窗和遮罩 self.driver.run_js( """ const sels = [ '[class*="modal"]', '[class*="popup"]', '[class*="dialog"]', '[class*="mask"]', '[class*="overlay"]' ]; for (const s of sels) { document.querySelectorAll(s).forEach(el => { const style = getComputedStyle(el); const z = parseInt(style.zIndex || '0', 10); if (z >= 999 && style.display !== 'none') { el.style.display = 'none'; } }); } document.body.style.overflow = 'auto'; """ ) time.sleep(0.2) except Exception: pass def search(self): self.driver.get("https://dian.ysbang.cn/#/home", timeout=15) self.driver.wait.doc_loaded(timeout=10) time.sleep(2) if not self._is_logged_in(): if not self.login(): return False detail_url = "https://dian.ysbang.cn/#/drugInfo?wholesaleid=376456110&isAssemble=true&trafficType=15" self.get_snapshot(detail_url, "376456110") return True def run(self): try: self.init_browser() self.search() except Exception as e: logger.exception("运行异常: %s", e) finally: self._quit_browser() if __name__ == "__main__": YaoShiBangCrawl().run()