# ==================================================================================== # ybm_crawl.py — 药帮忙平台爬虫 # 遵循 commons1 调度框架，平台专属逻辑保留在类内 # 由 CollectScheduleRunner / run_scheduled_loop 调度执行 # ==================================================================================== # ==================== SECTION 1: IMPORTS ==================== import base64 import json import os import random import re import threading import time import traceback from datetime import datetime from io import BytesIO import oss2 import requests from PIL import Image from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeoutError # ---- commons1 公共能力 ---- from commons.Logger import get_spider_logger from commons.conn_mysql import MySQLPoolOn2 from commons.feishu_webhook import send_text as feishu_send_text, send_error_card as feishu_send_error_card from commons.scheduler import CrawlerScheduler # ---- pipelines (基于 commons) ---- from pipelines.drug_pipelines import DrugPipeline logger = get_spider_logger("ybm") # ==================== SECTION 2: PLATFORM CONFIG ==================== # MySQL（供店铺信息直接读写；商品数据走 DrugPipeline） mysql_pool = MySQLPoolOn2() # 平台标识 PLATFORM_ID = 9 YBM_DEVICE_ID = "18008650300" # API（内部调度系统） TASK_API_BASE = "http://120.24.26.108:8082/api/collect_task" TASK_API_TOKEN = "zhijiayun_crawler_2026" TASK_PULL_URL = TASK_API_BASE + "/pull" TASK_HEARTBEAT_URL = TASK_API_BASE + "/heartbeat" TASK_REPORT_URL = TASK_API_BASE + "/report" # 手动任务（调试用） USE_MANUAL_TASK = False MANUAL_TASK = { "id": 2001, "collect_task_id": 5000, "company_id": 1, "product_name": "三九胃泰颗粒", "product_specs": "10", "product_keyword": "", "product_brand": "999", "sampling_cycle": 1, "sampling_start_time": 1778083200, "sampling_end_time": 1778342399, "collect_equipment_account_id": 0, "collect_region_id": 0, "collect_equipment_id": 2, "collect_round": 1, } # 反爬参数 MIN_CLICK_DELAY = 1.5 MAX_CLICK_DELAY = 3.5 MIN_INPUT_DELAY = 0.1 MAX_INPUT_DELAY = 0.3 MIN_PAGE_DELAY = 2.0 MAX_PAGE_DELAY = 4.0 MIN_KEYWORD_DELAY = 8.0 MAX_KEYWORD_DELAY = 15.0 SCROLL_TARGET_DISTANCE = 400 SCROLL_OFFSET_RANGE = 50 SCROLL_STEP = 50 SCROLL_INTERVAL = 0.05 # Cookie & 登录 COOKIE_FILE_PATH = "ybm_cookies.json" LOGIN_VALIDATE_URL = "https://www.ybm100.com/new/" USERNAME = "18008650300" PASSWORD = "12345678" TARGET_LOGIN_URL = "https://www.ybm100.com/new/login" # 选择器 USERNAME_SELECTOR = "input[placeholder*=请输入账号]" PASSWORD_SELECTOR = "input[placeholder*=请输入密码]" LOGIN_BTN_SELECTOR = "button:has(span:text('登录'))" SEARCH_INPUT_SELECTOR = "input[placeholder*='药品名称/厂家名称']" SEARCH_INPUT_SELECTOR2 = "div.home-search-container-search-head" SEARCH_BTN_SELECTOR = "div.home-search-container-search-head-btn[data-scmd=\"text-搜索\"]" PRODUCT_ITEM_SELECTOR = "div.product-list-item" PRODUCT_TITLE_SELECTOR = "div.product-name" PRODUCT_PRICE_SELECTOR = "div.main-price" PRODUCT_STORE_SELECTOR = 'div.prduct-shop-name div.shop-name' PRODUCT_COMPANY_SELECTOR = "div.product-manufacturer" PRODUCT_VALIDITY_SELECTOR = "div.product-period" # 等待时间 ELEMENT_TIMEOUT = 10000 LOGIN_AFTER_CLICK = 5000 SEARCH_BTN_TIMEOUT = 5000 COLLECT_DELAY = 3000 DETAIL_LOAD_TIMEOUT = 5000 # 浏览器 BROWSER_HEADLESS = False BROWSER_CHANNEL = "chrome" SLOW_MO_MIN = 50 SLOW_MO_MAX = 100 # 百度 OCR request_url_config = "https://aip.baidubce.com/rest/2.0/ocr/v1/business_license" AppKey_config = "tRK2RhyItCSh6BzyT4CNVXQa" AppSecret_config = "TDgKiPo94i2mOM1sDqOuDnlcK1bG66jh" token_url_config = 'https://aip.baidubce.com/oauth/2.0/token' # OSS OSS_ACCESS_KEY_ID = 'LTAI5tDwjfteBvivYN41r8sJ' OSS_ACCESS_KEY_SECRET = 'yowuOGi2nYYnrqGpO3qcz94C4brcPp' OSS_ENDPOINT = "oss-cn-shenzhen.aliyuncs.com" OSS_BUCKET_NAME = "zhijiayun-jiansuo" OSS_PREFIX = "scrape_data/" LOCAL_SCREENSHOT_DIR = "local_screenshots" LOCAL_CROPPED_DIR = "./local_cropped_screenshots" IMAGE_COMPRESS_ENABLE = True IMAGE_COMPRESS_QUALITY = 30 IMAGE_COMPRESS_PNG_LEVEL = 9 # 心跳上报间隔（秒） HEARTBEAT_INTERVAL_SECONDS = 60 # ==================== SECTION 3: OSS / SCREENSHOT HELPERS ==================== def init_local_screenshot_dir(): if not os.path.exists(LOCAL_SCREENSHOT_DIR): os.makedirs(LOCAL_SCREENSHOT_DIR) logger.info(f"本地截图目录已创建: {LOCAL_SCREENSHOT_DIR}") else: logger.debug(f"本地截图目录已存在: {LOCAL_SCREENSHOT_DIR}") def init_oss_bucket(): try: auth = oss2.Auth(OSS_ACCESS_KEY_ID, OSS_ACCESS_KEY_SECRET) bucket = oss2.Bucket(auth, OSS_ENDPOINT, OSS_BUCKET_NAME) bucket.get_bucket_info() logger.info("OSS Bucket 初始化成功") return bucket except Exception as e: logger.error(f"OSS Bucket 初始化失败: {str(e)}") raise def upload_local_screenshot_to_oss(bucket, local_file_path, oss_file_path=None): if not os.path.exists(local_file_path): raise FileNotFoundError(f"本地截图文件不存在: {local_file_path}") if not oss_file_path: local_file_name = os.path.basename(local_file_path) oss_file_path = f"screenshots/{local_file_name}" try: bucket.put_object_from_file(oss_file_path, local_file_path) oss_file_url = f"https://{OSS_BUCKET_NAME}.{OSS_ENDPOINT}/{oss_file_path}" logger.info(f"截图上传 OSS 成功: {oss_file_url}") return oss_file_url except Exception as e: logger.error(f"截图上传 OSS 失败: {str(e)}") raise def crop_local_screenshot(local_file_path, cropped_file_path=None, crop_region=None): if not os.path.exists(local_file_path): raise FileNotFoundError(f"原始截图文件不存在: {local_file_path}") os.makedirs(LOCAL_CROPPED_DIR, exist_ok=True) if not cropped_file_path: file_name = os.path.basename(local_file_path) file_name_no_ext, file_ext = os.path.splitext(file_name) cropped_file_name = f"{file_name_no_ext}_cropped{file_ext}" cropped_file_path = os.path.join(LOCAL_CROPPED_DIR, cropped_file_name) with Image.open(local_file_path) as img: img_width, img_height = img.size if not crop_region: crop_region = (0, 0, int(img_width), int(img_height * 0.3)) c_left, c_upper, c_right, c_lower = crop_region if c_right > img_width or c_lower > img_height or c_left < 0 or c_upper < 0: raise ValueError(f"裁剪区域超出图片范围，图片尺寸=({img_width}, {img_height})，裁剪区域={crop_region}") cropped_img = img.crop(crop_region) file_ext = os.path.splitext(cropped_file_path)[1].lower() try: if IMAGE_COMPRESS_ENABLE: if file_ext in ['.jpg', '.jpeg']: cropped_img.save(cropped_file_path, format='JPEG', quality=IMAGE_COMPRESS_QUALITY, optimize=True, progressive=True) else: cropped_img.save(cropped_file_path) else: cropped_img.save(cropped_file_path, format='JPEG') except Exception: cropped_img.save(cropped_file_path, format='JPEG') try: if os.path.exists(cropped_file_path): os.remove(local_file_path) except OSError: pass return cropped_file_path def screenshot_target_page_to_local_then_oss(target_page, local_file_path=None, oss_file_path=None, full_page=True, crop_region=None): os.makedirs(LOCAL_SCREENSHOT_DIR, exist_ok=True) if not local_file_path: current_time = datetime.now().strftime("%Y%m%d_%H%M%S") local_file_name = f"{current_time}_p{PLATFORM_ID}_target_page.jpg" local_file_path = os.path.join(LOCAL_SCREENSHOT_DIR, local_file_name) logger.info(f"开始页面截图: {local_file_path}") target_page.screenshot(path=local_file_path, full_page=full_page, omit_background=False, timeout=10000) cropped_file_path = crop_local_screenshot(local_file_path=local_file_path, crop_region=crop_region) bucket = init_oss_bucket() oss_file_url = upload_local_screenshot_to_oss(bucket, cropped_file_path, oss_file_path) return cropped_file_path, oss_file_url # ==================== SECTION 4: CITY / PROVINCE MAPPING ==================== CITY_JSON_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "..", "commons", "city.json") PROVINCE_ID_MAP = {} CITY_ID_MAP = {} CITY_TO_PROVINCES_MAP = {} DIRECT_MUNICIPALITIES = {"北京市", "上海市", "天津市", "重庆市"} DIRECT_MUNICIPALITY_BASE_NAMES = {"北京", "上海", "天津", "重庆"} DIRECT_MUNICIPALITY_ALIAS = { "北京": "北京市", "上海": "上海市", "天津": "天津市", "重庆": "重庆市", } def load_city_mapping(): global PROVINCE_ID_MAP, CITY_ID_MAP, CITY_TO_PROVINCES_MAP PROVINCE_ID_MAP.clear() CITY_ID_MAP.clear() CITY_TO_PROVINCES_MAP.clear() if not os.path.exists(CITY_JSON_PATH): logger.error(f"❌ 城市JSON文件不存在：{CITY_JSON_PATH}") return try: with open(CITY_JSON_PATH, "r", encoding="utf-8") as f: data = json.load(f) for province_item in data: p_name = province_item['name'] PROVINCE_ID_MAP[p_name] = province_item['id'] for city_item in province_item.get('sons', []): c_name = city_item['name'] CITY_ID_MAP[(p_name, c_name)] = city_item['id'] CITY_TO_PROVINCES_MAP.setdefault(c_name, set()).add(p_name) logger.info(f"✅ 城市映射加载完成，共 {len(PROVINCE_ID_MAP)} 个省份，{len(CITY_ID_MAP)} 个城市") except Exception as e: logger.error(f"❌ 加载城市JSON失败：{str(e)}") def _clean_province_name(name: str) -> str: return (name or "").replace("省", "").replace("市", "").replace("自治区", "").replace("特别行政区", "").strip() def _clean_city_name(name: str) -> str: return (name or "").replace("市", "").replace("自治州", "").replace("地区", "").replace("盟", "").strip() def normalize_province_city_names(province_name: str, city_name: str): province = (province_name or "").strip() city = (city_name or "").strip() if province and province not in DIRECT_MUNICIPALITIES and province not in PROVINCE_ID_MAP: clean_p = _clean_province_name(province) for standard_name in PROVINCE_ID_MAP.keys(): if clean_p and clean_p == _clean_province_name(standard_name): province = standard_name break if not province and city: matched_provinces = CITY_TO_PROVINCES_MAP.get(city, set()) if not matched_provinces: clean_c = _clean_city_name(city) if clean_c: matched_provinces = { p_name for (p_name, c_name) in CITY_ID_MAP.keys() if _clean_city_name(c_name) == clean_c } if len(matched_provinces) == 1: province = next(iter(matched_provinces)) elif len(matched_provinces) > 1: logger.warning(f"⚠️ 城市名存在跨省重名，无法唯一反推省份: city={city}, candidates={sorted(matched_provinces)}") if province in DIRECT_MUNICIPALITY_BASE_NAMES: province = DIRECT_MUNICIPALITY_ALIAS[province] if province and city and (province, city) not in CITY_ID_MAP: clean_c = _clean_city_name(city) for (p_name, c_name), _ in CITY_ID_MAP.items(): if _clean_province_name(p_name) == _clean_province_name(province) and clean_c and _clean_city_name(c_name) == clean_c: city = c_name break if province in DIRECT_MUNICIPALITIES and not city: city = province return province, city def get_province_city_ids(province_name, city_name): province_name, city_name = normalize_province_city_names(province_name, city_name) province_id = PROVINCE_ID_MAP.get(province_name) if province_name else None if province_name and province_id is None: clean_p = _clean_province_name(province_name) for name, pid in PROVINCE_ID_MAP.items(): if clean_p and clean_p == _clean_province_name(name): province_id = pid province_name = name break if province_id is None: logger.warning(f"⚠️ 未找到省份ID: {province_name}") province_id = 0 elif province_id is None: province_id = 0 if _clean_province_name(province_name) in DIRECT_MUNICIPALITY_BASE_NAMES and not city_name: city_name = f"{_clean_province_name(province_name)}市" city_id = CITY_ID_MAP.get((province_name, city_name)) if province_name and city_name else None if province_name and city_name and city_id is None: clean_c = _clean_city_name(city_name) for (p_name, c_name), cid in CITY_ID_MAP.items(): if p_name == province_name: if clean_c and clean_c == _clean_city_name(c_name): city_id = cid city_name = c_name break if city_id is None: if _clean_province_name(province_name) in DIRECT_MUNICIPALITY_BASE_NAMES and province_id: city_id = province_id else: logger.warning(f"⚠️ 未找到城市ID: {province_name} - {city_name}") city_id = 0 elif city_id is None: city_id = 0 return province_id, city_id def extract_province_city(address): if not address: return "", "" province_pattern = re.compile(r'([^省]+省|.+自治区|北京市|上海市|天津市|重庆市|.+特别行政区)') province_match = province_pattern.search(address) province = province_match.group(1) if province_match else "" address_remain = address.replace(province, "").strip() if province else address.strip() city_pattern = re.compile(r'([^市]+市|.+自治州|.+地区|.+盟|^[^\d区县镇]+)') city_match = city_pattern.search(address_remain) city = city_match.group(1).strip() if city_match else "" if province in ["北京市", "上海市", "天津市", "重庆市"]: city = province if not province and not city: simple_pattern = re.compile(r'^([^\d区县镇]+)') simple_match = simple_pattern.search(address) if simple_match: city = simple_match.group(1).strip() if city and province and city != province and province in city: city = city.replace(province, "").strip() province, city = normalize_province_city_names(province, city) return province.strip(), city.strip() # ==================== SECTION 5: ANTI-CRAWL UTILITIES ==================== def get_current_time(): return datetime.now().strftime('%Y-%m-%d %H:%M:%S') def random_delay(min_seconds, max_seconds): delay = random.uniform(min_seconds, max_seconds) time.sleep(delay) return delay def save_cookies(context, cookie_path=COOKIE_FILE_PATH): try: cookies = context.cookies() with open(cookie_path, "w", encoding="utf-8") as f: json.dump(cookies, f, ensure_ascii=False, indent=2) logger.info(f"Cookie已保存到：{cookie_path}") return True except Exception as e: logger.error(f" 保存Cookie失败：{e}") return False def load_cookies(context, cookie_path=COOKIE_FILE_PATH): if not os.path.exists(cookie_path): logger.warning(f" Cookie文件不存在：{cookie_path}") return False try: with open(cookie_path, "r", encoding="utf-8") as f: cookies = json.load(f) context.add_cookies(cookies) logger.info(f"✅ 已从{cookie_path}加载Cookie") return True except Exception as e: logger.error(f" 加载Cookie失败：{e}") return False def slow_scroll_400px(page, scroll_distance1=400): try: scroll_distance = random.randint(scroll_distance1 - SCROLL_OFFSET_RANGE, scroll_distance1 + SCROLL_OFFSET_RANGE) remaining_distance = scroll_distance total_steps = int(scroll_distance / SCROLL_STEP) logger.info(f"📜 开始慢速滚动（目标距离：{scroll_distance}px，总步数：{total_steps}，总时长约{total_steps*SCROLL_INTERVAL:.2f}秒）") for _ in range(total_steps): step = min(SCROLL_STEP, remaining_distance) page.evaluate(f"window.scrollBy(0, {step});") remaining_distance -= step time.sleep(SCROLL_INTERVAL) if remaining_distance > 0: page.evaluate(f"window.scrollBy(0, {remaining_distance});") time.sleep(SCROLL_INTERVAL) page.wait_for_load_state("networkidle", timeout=8000) random_delay(2.0, 3.0) logger.info(f" 慢速滚动完成，实际滚动距离：{scroll_distance - remaining_distance}px") return True except Exception as e: logger.warning(f" 慢速滚动失败：{e}") return False def type_slow(locator, text: str, min_delay=0.06, max_delay=0.18): for ch in text: locator.type(ch, delay=int(random.uniform(min_delay, max_delay) * 1000)) def kill_masks(page): page.evaluate(r""" () => { const knownSelectors = [ '.v-modal', '.el-overlay', '.el-overlay-dialog', '.el-dialog__wrapper', '.el-message-box__wrapper', '.el-loading-mask', '.el-popup-parent--hidden' ]; for (const sel of knownSelectors) { document.querySelectorAll(sel).forEach(el => el.remove()); } const all = Array.from(document.querySelectorAll('body *')); for (const el of all) { const s = window.getComputedStyle(el); if (!s) continue; const z = parseInt(s.zIndex || '0', 10); if ((s.position === 'fixed' || s.position === 'absolute') && z >= 1000 && s.pointerEvents !== 'none') { const r = el.getBoundingClientRect(); const nearFullScreen = r.width >= window.innerWidth * 0.8 && r.height >= window.innerHeight * 0.8; if (nearFullScreen) { el.style.pointerEvents = 'none'; el.style.display = 'none'; } } } document.documentElement.style.overflow = 'auto'; document.body.style.overflow = 'auto'; document.body.style.position = 'static'; document.body.style.width = 'auto'; document.body.style.paddingRight = '0px'; document.body.classList.remove('el-popup-parent--hidden'); } """) def force_close_popup(page): try: for _ in range(5): btn = page.locator( "//button[normalize-space()='下一步' or normalize-space()='完成' or normalize-space()='我知道了' or normalize-space()='关闭']" ).first if btn.count() > 0 and btn.is_visible(): btn.click(timeout=1500) page.wait_for_timeout(300) continue close_icon = page.locator( "xpath=//*[contains(@class,'close') or contains(@class,'el-icon-close') or name()='svg' or name()='i'][1]" ).first if close_icon.count() > 0 and close_icon.is_visible(): close_icon.click(timeout=1000) page.wait_for_timeout(300) continue break page.evaluate(""" const selectors = [ '.v-modal', '.el-overlay', '.el-overlay-dialog', '.el-dialog__wrapper', '[class*="mask"]', '[class*="overlay"]', '[style*="z-index"]' ]; for (const sel of selectors) { document.querySelectorAll(sel).forEach(el => { const s = window.getComputedStyle(el); if ((s.position === 'fixed' || s.position === 'absolute') && parseInt(s.zIndex || '0', 10) >= 1000) { el.remove(); } }); } """) except Exception: pass def popup_guard(page, tag=""): try: page.wait_for_timeout(300) for _ in range(6): btn = page.locator( "xpath=//button[normalize-space()='下一步' or normalize-space()='完成' or normalize-space()='我知道了' or normalize-space()='关闭']" ).first if btn.count() > 0 and btn.is_visible(): btn.click(timeout=1500) page.wait_for_timeout(250) continue close_btn = page.locator( "css=.el-dialog__headerbtn, .el-message-box__headerbtn, .close, .icon-close, .el-icon-close" ).first if close_btn.count() > 0 and close_btn.is_visible(): close_btn.click(timeout=1200) page.wait_for_timeout(250) continue break page.evaluate(r""" () => { const selectors = [ '.v-modal', '.el-overlay', '.el-overlay-dialog', '.el-dialog__wrapper', '.el-message-box__wrapper', '.el-loading-mask' ]; selectors.forEach(sel => document.querySelectorAll(sel).forEach(e => e.remove())); const all = Array.from(document.querySelectorAll('body *')); for (const el of all) { const s = getComputedStyle(el); const z = parseInt(s.zIndex || '0', 10); if ((s.position === 'fixed' || s.position === 'absolute') && z >= 1000 && s.pointerEvents !== 'none') { const r = el.getBoundingClientRect(); const nearFull = r.width >= innerWidth * 0.8 && r.height >= innerHeight * 0.8; if (nearFull) { el.style.pointerEvents = 'none'; el.style.display = 'none'; } } } document.documentElement.style.overflow = 'auto'; document.body.style.overflow = 'auto'; document.body.classList.remove('el-popup-parent--hidden'); } """) logger.info("杀除弹窗成功") except Exception: pass # ==================== SECTION 6: OCR HELPERS ==================== def compress_image(image_data, max_size=4*1024*1024): try: img = Image.open(BytesIO(image_data)) if img.mode in ('RGBA', 'P'): bg_img = Image.new('RGB', img.size, (255, 255, 255)) bg_img.paste(img, mask=img.split()[-1] if img.mode == 'RGBA' else None) img = bg_img if img.width > 1000: ratio = 1000 / img.width img = img.resize((int(img.width * ratio), int(img.height * ratio)), Image.Resampling.LANCZOS) output = BytesIO() img.save(output, format='JPEG', quality=80) compressed_data = output.getvalue() if len(compressed_data) > max_size: output2 = BytesIO() img.save(output2, format='JPEG', quality=60) compressed_data = output2.getvalue() return compressed_data except Exception as e: logger.debug(f"图片压缩失败：{e}") return image_data def download_image_to_base64(image_url, save_dir="./download_images"): try: if not os.path.exists(save_dir): os.makedirs(save_dir) except Exception as e: print(f"创建保存目录失败：{str(e)}") return None try: headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" } response = requests.get(image_url, headers=headers, timeout=15) response.raise_for_status() compressed_data = compress_image(response.content) image_base64 = base64.b64encode(compressed_data).decode("utf-8") image_data = compressed_data file_name = image_url.split("/")[-1] file_name = file_name.replace("?", "").replace("&", "").replace("=", "") save_path = os.path.join(save_dir, file_name) with open(save_path, "wb") as f: f.write(image_data) print(f"图片已保存到本地：{save_path}") return image_base64 except requests.exceptions.Timeout: print(f"下载图片超时：{image_url}") return None except requests.exceptions.HTTPError as e: code = e.response.status_code if e.response is not None else "?" logger.warning("下载图片 HTTP 错误 url=%s status=%s", image_url, code) return None except Exception as e: print(f"下载图片失败：{str(e)}") return None def get_access_token(): url = f"{token_url_config}?grant_type=client_credentials&client_id={AppKey_config}&client_secret={AppSecret_config}" headers = {'Content-Type': 'application/json', 'Accept': 'application/json'} try: response = requests.request("POST", url, headers=headers, data="") response.raise_for_status() return response.json()['access_token'] except Exception as e: print(f"获取access_token失败：{str(e)}") return None def get_ocr_res(img): try: print(f'开始识别图片:{img}') img_base64 = download_image_to_base64(img) if not img_base64: print("图片下载/转Base64失败，终止OCR识别") return None access_token = get_access_token() if not access_token: print("获取access_token失败，无法调用OCR接口") return None request_url = request_url_config + "?access_token=" + access_token headers = {'content-type': 'application/x-www-form-urlencoded'} response = requests.post(request_url, data={"image": img_base64}, headers=headers) if response: res = response.json() if "error_code" in res: print(f"百度OCR接口错误：{res['error_msg']}（错误码：{res['error_code']}）") return None new_dic = {} for ite in res['words_result'].keys(): new_dic[ite] = res['words_result'][ite]['words'] print('资质数据信息', new_dic) return new_dic else: print("OCR接口返回空响应") return None except requests.exceptions.RequestException as e: print(f"网络错误（图片下载/OCR请求失败）：{str(e)}") return None except KeyError as e: print(f"OCR响应格式异常，缺失字段：{str(e)}") return None except Exception as e: print(f"OCR识别未知错误：{str(e)}") return None # ==================== SECTION 7: SHOP HELPERS ==================== def clean_shop_name(raw_shop_name): if not raw_shop_name: return '' pattern = r'【.*?】|\(.*?\)|\[.*?\]' cleaned = re.sub(pattern, '', raw_shop_name) cleaned = cleaned.strip().replace('\n', '').replace('\r', '') cleaned = re.sub(r'\s+', ' ', cleaned) return cleaned if cleaned else raw_shop_name def shop_is_exists_database(shop): query_sql = """ SELECT province, city, business_license_company, qualification_number, business_license_address FROM retrieve_scrape_shop_info WHERE shop = %s and platform = %s """ try: rows = mysql_pool.select_data(query_sql, (shop, 9)) result = rows[0] if rows else None logger.debug("店铺存在校验 shop=%r result=%s", shop, result) is_exists = bool(result) if is_exists: logger.info(f"【店铺存在校验】店铺已存在 | 店铺名：{repr(shop)} | 结果：存在（True）不要执行采集店铺") else: logger.info(f"【店铺存在校验】店铺不存在 | 店铺名：{repr(shop)} | 结果：不存在（False）") return is_exists, result except Exception as e: logger.error(f"查询店铺失败：{e}") return False, None def insert_shop_info_to_db(shop, contact_address, qualification_number, business_license_company, business_license_address, scrape_date, platform, province, city, create_time, update_time): sql = """ INSERT INTO retrieve_scrape_shop_info ( shop, contact_address, qualification_number, business_license_company, business_license_address, scrape_date, platform, province, city, create_time, update_time ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE contact_address = VALUES(contact_address), qualification_number = VALUES(qualification_number), business_license_company = VALUES(business_license_company), business_license_address = VALUES(business_license_address), scrape_date = VALUES(scrape_date), platform = VALUES(platform), province = VALUES(province), city = VALUES(city), update_time = VALUES(update_time) """ params = (shop, contact_address, qualification_number, business_license_company, business_license_address, scrape_date, platform, province, city, create_time, update_time) conn = None cursor = None try: conn = mysql_pool.get_conn() cursor = conn.cursor() cursor.execute(sql, params) conn.commit() logger.info("店铺信息写入成功 shop=%s company=%s", shop, business_license_company) return True except Exception as e: logger.error("插入店铺失败：%s\n%s", e, traceback.format_exc()) if conn: conn.rollback() return False finally: if cursor: cursor.close() if conn: conn.close() # ==================== SECTION 8: YbmCrawler CLASS ==================== class YbmCrawler: """药帮忙平台爬虫 —— 遵循 commons1 调度框架。 由 CollectScheduleRunner 调用： crawler = YbmCrawler(task_dict) count, is_success = crawler.run() """ def __init__(self, drug_dict=None): self.task_dict = drug_dict or {} self.platform_id = PLATFORM_ID self.username = YBM_DEVICE_ID self.platform_name = '药帮忙' # Pipeline（商品数据走 DrugPipeline，做去重入库） self.pipeline = DrugPipeline("ybm") self.is_success = True # 由 CrawlerScheduler 管理心跳和停止信号 self.scheduler = CrawlerScheduler(YBM_DEVICE_ID, str(PLATFORM_ID), heartbeat_interval=HEARTBEAT_INTERVAL_SECONDS) if self.task_dict: self._extract_task_params() # ---------- task params ---------- def _extract_task_params(self): t = self.task_dict self.task_id = t.get("id", 0) self.collect_task_id = t.get("collect_task_id", 0) self.company_id = t.get("company_id", 0) self.product_name = t.get("product_name", "") self.product_specs = t.get("product_specs", "") self.product_keyword = t.get("product_keyword", "") self.product_brand = t.get("product_brand", "") self.sampling_cycle = t.get("sampling_cycle", 1) self.sampling_start_time = t.get("sampling_start_time", 0) self.sampling_end_time = t.get("sampling_end_time", 0) self.collect_equipment_account_id = t.get("collect_equipment_account_id", 0) self.collect_region_id = t.get("collect_region_id", 0) self.collect_round = t.get("collect_round", 1) self.api_current_page = t.get("current_page", 0) or 0 self.keyword = self.product_brand + self.product_name self.collect_config_info = json.dumps({ "sampling_cycle": self.sampling_cycle, "sampling_start_time": self.sampling_start_time, "sampling_end_time": self.sampling_end_time, }, ensure_ascii=False, default=str) # ---------- API report helpers ---------- def _send_report(self, params: dict, log_msg: str): try: resp = requests.post( TASK_REPORT_URL, json=params, timeout=10, headers={'X-Crawler-Token': TASK_API_TOKEN}, ) if resp.status_code == 200: try: data = resp.json() if data.get("code") == "success": logger.info("回告成功：%s", log_msg) return "success" else: msg = data.get("msg", "") if "已达平台限额" in msg or "限额" in msg: logger.warning("回告接口返回限额错误：code=%s, msg=%s", data.get('code'), msg) return "limit_reached" else: logger.warning("回告接口返回错误：code=%s, msg=%s", data.get('code'), msg) return "error" except ValueError: logger.warning("回告响应非 JSON：%s", resp.text[:200]) return "error" else: logger.warning("回告 HTTP %s", resp.status_code) return "error" except Exception as e: logger.error("回告失败：%s，错误：%s", log_msg, str(e)) return "error" def _report_start(self): payload = { "task_id": self.task_id, "platform": str(self.platform_id), "username": self.username, "current_page": 0, "total_pages": 0, "is_finished": 0, } logger.info("上报开始 task_id=%s keyword=%s", self.task_id, self.keyword) return self._send_report(payload, f"开始上报 task_id={self.task_id}") def _report_progress(self, current_page: int, total_pages: int = 0): payload = { "task_id": self.task_id, "platform": str(self.platform_id), "username": self.username, "current_page": current_page, "total_pages": total_pages, "is_finished": 0, } return self._send_report(payload, f"进度上报 page={current_page}") def _report_end(self, success: bool, real_count: int, page_no: int = 0, exception_type: int = None, need_reassign: int = None, total_pages: int = 0): payload = { "task_id": self.task_id, "platform": str(self.platform_id), "username": self.username, "current_page": page_no, "total_pages": total_pages if total_pages > 0 else page_no, "crawled_count": real_count, "is_finished": 1 if success else 0, } if exception_type is not None: payload["exception_type"] = exception_type if need_reassign is not None: payload["need_reassign"] = need_reassign logger.info("上报结束 task_id=%s keyword=%s payload=%s", self.task_id, self.keyword, payload) self._send_report(payload, f"结束上报 task_id={self.task_id}") # ---------- login ---------- def _is_login(self, page): try: page.goto(LOGIN_VALIDATE_URL, timeout=300000) page.wait_for_load_state("networkidle") if "login" in page.url.lower(): logger.warning(" Cookie失效，需要重新登录") return False logger.info(" Cookie有效，已保持登录状态") return True except Exception as e: logger.error(f" 验证登录状态失败：{e}") return False def _login_operation(self, page): try: page.wait_for_selector(USERNAME_SELECTOR, timeout=ELEMENT_TIMEOUT, state="visible") page.wait_for_timeout(timeout=3000) page.fill(USERNAME_SELECTOR, USERNAME) logger.info(" 已输入登录账号") page.wait_for_selector(PASSWORD_SELECTOR, timeout=ELEMENT_TIMEOUT, state="visible") page.wait_for_timeout(timeout=3000) page.fill(PASSWORD_SELECTOR, PASSWORD) logger.info(" 已输入登录密码") random_delay(1, 2) agree_btn = page.locator('span.el-checkbox__inner') agree_btn.click() page.wait_for_selector(LOGIN_BTN_SELECTOR, timeout=ELEMENT_TIMEOUT) page.wait_for_timeout(timeout=3000) page.click(LOGIN_BTN_SELECTOR) logger.info(" 已点击登录按钮") page.wait_for_timeout(LOGIN_AFTER_CLICK) return True except PlaywrightTimeoutError as e: logger.error(f" 登录失败：元素定位超时 - {str(e)}") return False except Exception as e: logger.error(f" 登录异常：{str(e)}") return False # ---------- search ---------- @staticmethod def _pick_search_input(page): inputs = page.locator(SEARCH_INPUT_SELECTOR) cnt = inputs.count() for i in range(min(cnt, 2)): candidate = inputs.nth(i) try: candidate.wait_for(state="visible", timeout=1500) if candidate.is_enabled(): return candidate except PlaywrightTimeoutError: continue candidate = page.locator(f"{SEARCH_INPUT_SELECTOR}:visible").first candidate.wait_for(state="visible", timeout=ELEMENT_TIMEOUT) return candidate def _search_operation(self, page, keyword, is_first_search=True): try: search_locator = self._pick_search_input(page) search_locator.wait_for(timeout=ELEMENT_TIMEOUT) search_locator.click(force=True) search_locator.fill("") page.keyboard.down("Control") page.keyboard.press("a") page.keyboard.up("Control") page.keyboard.press("Backspace") type_slow(search_locator, keyword, min_delay=0.06, max_delay=0.18) logger.info(f"📝 已输入搜索关键词：{keyword}") btn = page.locator(SEARCH_BTN_SELECTOR) btn.wait_for(state="visible", timeout=SEARCH_BTN_TIMEOUT) page.wait_for_timeout(3000) detail_page = page if is_first_search: try: with page.context.expect_page(timeout=60000) as new_page_info: btn.click() detail_page = new_page_info.value detail_page.wait_for_load_state("networkidle", timeout=20000) except PlaywrightTimeoutError: logger.warning(f"{get_current_time()} 未检测到新标签页") return None, False except Exception as e: logger.warning(f"{get_current_time()} 等待新标签页异常：{e}") return None, False else: btn.click() page.wait_for_load_state("networkidle", timeout=20000) detail_page = page logger.info("✅ 后续搜索：已在原页面完成跳转加载") test_btn = detail_page.locator("div[data-v-c65c36bc].first-time-highlight-message-btn button") if test_btn.count() > 0: test_btn.wait_for(state="attached", timeout=5000) test_btn.click() force_close_popup(detail_page) kill_masks(detail_page) logger.info("✅ 已触发搜索") return detail_page, True except PlaywrightTimeoutError as e: logger.error(f" 搜索失败：元素定位超时 - {str(e)}") return None, False except Exception as e: logger.error(f" 搜索异常：{str(e)}") return None, False # ---------- pagination ---------- @staticmethod def _goto_next_page(page) -> bool: try: next_btn = page.locator("button.btn-next").first next_btn.wait_for(state="attached", timeout=3000) aria_disabled = next_btn.get_attribute("aria-disabled") logger.info(f"下一页按钮 aria-disabled 属性值：{aria_disabled}") if aria_disabled == "true": logger.warning("⚠️ 下一页按钮 aria-disabled=true，已无更多页面") return False page.wait_for_timeout(500) if next_btn.is_visible() and next_btn.is_enabled(): next_btn.click(timeout=5000) else: next_btn.click(force=True, timeout=5000) logger.info("✅ 翻页成功，下一页按钮 aria-disabled=false") return True except PlaywrightTimeoutError: logger.warning("⚠️ 下一页按钮加载超时，判定无更多页面") return False except Exception as e: logger.warning(f"⚠️ 翻页操作异常：{e}，判定无更多页面") return False @staticmethod def _get_total_pages(page): """从分页组件提取总页数，失败返回 None""" try: total_elem = page.locator("div.filter-panel-tags-pagination-number-total").first total_elem.wait_for(state="attached", timeout=3000) if total_elem.count() > 0: text = total_elem.inner_text(timeout=2000).strip() total = int(text) logger.info(f"📊 检测到总页数：{total}") return total except PlaywrightTimeoutError: logger.warning("⚠️ 总页数元素加载超时，降级为盲翻") except (ValueError, TypeError) as e: logger.warning(f"⚠️ 总页数解析失败：{e}，降级为盲翻") except Exception as e: logger.warning(f"⚠️ 获取总页数异常：{e}，降级为盲翻") return None # ---------- shop license collection ---------- @staticmethod def _collect_shop_license(target_page, shop): """采集店铺营业执照信息（OCR识别），返回 dict 或 None""" random_delay(MIN_CLICK_DELAY, MAX_CLICK_DELAY) entershop_btn = target_page.locator('div[data-v-5485589c].shop-info-container-left-info') entershop_btn.wait_for(state="visible", timeout=10000) entershop_btn.scroll_into_view_if_needed() entershop_btn.hover() random_delay(0.2, 0.5) try: with target_page.expect_popup(timeout=15000) as pop: entershop_btn.click() random_delay(0.05, 0.15) shop_page = pop.value except PlaywrightTimeoutError: logger.warning("⚠️ 店铺弹窗未出现") return None shop_page.wait_for_load_state("domcontentloaded") store_url = shop_page.url logger.info(f"📌 获取到店铺链接：{store_url}") random_delay(MIN_CLICK_DELAY, MAX_CLICK_DELAY) shop_license_page = shop_page.locator( '//div[contains(@class, "shop-info-container-right-btns-item") and contains(span, "资质/售后")]') shop_license_page.wait_for(state="attached", timeout=15000) shop_license_page.scroll_into_view_if_needed() shop_license_page.hover() random_delay(0.2, 0.5) shop_license_page.click() random_delay(0.05, 0.15) random_delay(0.05, 0.1) shop_page.wait_for_load_state("networkidle") shop_page.wait_for_load_state("load") shop_license_img = shop_page.locator( '//span[contains(text(), "企业营业执照") or contains(text(), "营业执照（正本）")]/ancestor::div[@class="shop-info-drawer-zz-tab1-list-item"]/img' ).first shop_license_img.wait_for(state="visible", timeout=60000) shop_license_src = None ocr_res = None try: if shop_license_img.count() > 0: shop_license_src = shop_license_img.get_attribute('src') shop_license_src = shop_license_src.strip() if shop_license_src else None ocr_res = get_ocr_res(shop_license_src) except Exception as e: logger.warning(f"提取营业执照图片src失败：{e}") logger.debug("营业执照图片链接：%s", shop_license_src) qualification_number = ocr_res.get('社会信用代码', '') if ocr_res else '' business_license_company = ocr_res.get('单位名称', '') if ocr_res else '' business_license_address = ocr_res.get('地址', '') if ocr_res else '' province, city = extract_province_city(business_license_address) logger.info(f"原始地址：{business_license_address}") logger.info(f"提取的省份：{province} | 城市：{city}") current_time = datetime.now().strftime("%Y-%m-%d") insert_shop_info_to_db( shop=shop, contact_address=store_url, qualification_number=qualification_number, business_license_company=business_license_company, business_license_address=business_license_address, scrape_date=current_time, platform='药帮忙', province=province, city=city, create_time=datetime.now().strftime("%Y-%m-%d %H:%M:%S"), update_time=datetime.now().strftime("%Y-%m-%d %H:%M:%S"), ) return { "shop_page": shop_page, "store_url": store_url, "province": province, "city": city, "business_license_company": business_license_company, "qualification_number": qualification_number, "business_license_address": business_license_address, } # ---------- core collection ---------- def _collect_data(self, store_page): """核心采集循环：遍历翻页 → 逐商品采集 → 走 DrugPipeline 入库""" brand = self.product_brand name = self.product_name keyword = self.keyword collect_result = [] logger.info(f"📊 开始采集「{keyword}」的商品数据") store_page.wait_for_load_state("networkidle") page_no = 1 start_page = self.api_current_page + 1 if self.api_current_page > 0 else 1 # 任务续爬：翻到起始页 if start_page > 1: logger.info(f"⏩ 任务续爬：从第 {start_page} 页开始，正在翻页...") while page_no < start_page: if self.scheduler.end: logger.warning("⛔ 翻页过程中收到停止信号，终止采集") return collect_result, page_no if self._goto_next_page(store_page): page_no += 1 store_page.wait_for_load_state("networkidle") logger.info(f"⏩ 已翻到第 {page_no} 页") else: logger.warning(f"⚠️ 无法翻到第 {start_page} 页（当前仅 {page_no} 页），从当前页开始采集") break total_pages = 0 while True: if self.scheduler.end: logger.warning("⛔ 收到停止信号，终止采集") break logger.info(f"\n📄 「{keyword}」开始采集第 {page_no} 页") list_page_url = store_page.url logger.info(f"📌 已记录商品列表页URL：{list_page_url}") store_page.wait_for_load_state("domcontentloaded") store_page.wait_for_load_state("networkidle") store_page.wait_for_timeout(500) # 首次提取总页数 if total_pages == 0: extracted_total = self._get_total_pages(store_page) if extracted_total is not None: total_pages = extracted_total # 读到总页数立刻回报进度 prog_status = self._report_progress(page_no, total_pages=total_pages) if prog_status == "limit_reached": logger.warning("⛔ 进度上报检测到限额，通知停止") self.scheduler.set_flag(True) total_limit = store_page.locator(PRODUCT_ITEM_SELECTOR).count() logger.info(f"📌 「{keyword}」第{page_no}页 初始商品个数（count）：{total_limit}") collected_count = 0 not_found_keywords = store_page.locator("div.filter-panel-container-empty-text") if not_found_keywords.count() > 0: logger.warning(f"⚠️ 关键词「{keyword}」无匹配商品，直接跳过整个关键词采集") return [], 0 for idx in range(total_limit): if self.scheduler.end: logger.warning("⛔ 收到停止信号，终止当前页采集") break detail_page = None try: item = store_page.locator(PRODUCT_ITEM_SELECTOR).nth(idx) collected_count += 1 store_page.wait_for_load_state("networkidle") delay = random_delay(MIN_CLICK_DELAY, MAX_CLICK_DELAY) logger.info(f"📌 「{keyword}」第{page_no}页 第{collected_count}/{total_limit}个商品 - 等待{delay:.2f}秒后采集（反爬）") # ---- 初始化默认值 ---- title = "" shop = "" expiry_date = "无有效期" manufacture_date = "无生产日期" approval_number = "无批准文号" manufacturer = "未知公司" spec = "未知规格" current_time = datetime.now().strftime("%Y-%m-%d") is_sold_out = 0 business_license_address = '' # 售罄 sold_locator = item.locator('div.product-status') if sold_locator.count() > 0: is_sold_out = 1 logger.warning(f" 「{keyword}」第{page_no}页 第{collected_count}个商品已售罄") # 商品ID (platform_item_id) product_id = '' product_id_elem = item.locator('div.product-card[data-product-id]') if product_id_elem.count() > 0: product_id = product_id_elem.get_attribute("data-product-id") # 标题 product_locator = item.locator(PRODUCT_TITLE_SELECTOR) if product_locator.count() > 0: title = product_locator.inner_text(timeout=3000).strip() logger.info(f"{'='*10}「{keyword}」第{collected_count}个商品 - 列表页标题：{title}{'='*10}") else: logger.warning(f" 「{keyword}」第{collected_count}个商品 - 列表页标题元素未找到") # 品牌/名称筛选 if not (brand in title and name in title): logger.warning(f" 「{keyword}」第{collected_count}个商品 - 标题「{title}」不包含品牌「{brand}」、名称「{name}」，跳过") continue # 价格 price_int = item.locator('//span[@class="price-int"]').text_content().strip() price_decimal_elem = item.locator('//span[@class="price-decimal"]') price_decimal = price_decimal_elem.text_content().strip() if price_decimal_elem.count() > 0 else '' full_price = f"{price_int}{price_decimal}".strip() try: full_price_num = float(full_price) except ValueError: full_price_num = 0.0 logger.warning(f" 「{keyword}」第{collected_count}个商品「{title}」- 列表页价格无法解析 raw={full_price!r}，按 0 处理") logger.info(f"✅ 提取到价格：{full_price_num}") # 公司名 manufacturer_locator = item.locator(PRODUCT_COMPANY_SELECTOR) if manufacturer_locator.count() > 0: manufacturer = manufacturer_locator.inner_text(timeout=3000).strip() logger.info(f"{'='*10}「{keyword}」第{collected_count}个商品 - 列表页公司名：{manufacturer}{'='*10}") else: logger.warning(f" 「{keyword}」第{collected_count}个商品「{title}」- 列表页公司名称元素未找到") # 店铺名 shop_locator = item.locator(PRODUCT_STORE_SELECTOR) if shop_locator.count() > 0: raw_shop = shop_locator.inner_text(timeout=3000).strip() shop = clean_shop_name(raw_shop) logger.info(f"{'='*10}「{keyword}」第{collected_count}个商品 - 列表页店名：{shop}{'='*10}") else: logger.warning(f" 「{keyword}」第{collected_count}个商品「{title}」- 列表页店铺名称元素未找到") # 折扣价 discount_price_locator = item.locator('span[data-v-4cb6cc1f].discount-int').first if discount_price_locator.count() > 0: discount_price_val_origin = discount_price_locator.inner_text(timeout=3000).strip() match = re.search(r'\d+\.?\d*', str(discount_price_val_origin)) discount_price_val = float(match.group()) if match else 0.00 logger.info(f"{'='*10}「{keyword}」第{collected_count}个商品 - 详情页折扣价：{discount_price_val}{'='*10}") else: discount_price_val = full_price_num logger.warning(f" 「{keyword}」第{collected_count}个商品「{title}」- 折扣价元素未找到，使用采购价兜底：{discount_price_val}") # 有效期 expiry_date_locator = item.locator(PRODUCT_VALIDITY_SELECTOR) if expiry_date_locator.count() > 0: expiry_date = expiry_date_locator.inner_text(timeout=3000).strip().replace('-', '') logger.info(f"{'='*10}「{keyword}」第{collected_count}个商品 - 详情页有效期：{expiry_date}{'='*10}") else: logger.warning(f" 「{keyword}」第{collected_count}个商品「{title}」- 有效期元素未找到") # ---- 点击进入详情页 ---- logger.info(f"📌 「{keyword}」第{page_no}页 第{collected_count}个商品「{title}」- 模拟鼠标移动并点击") item.hover() random_delay(0.2, 0.5) item.dispatch_event("mousedown") random_delay(0.05, 0.15) item.dispatch_event("mouseup") random_delay(0.05, 0.1) try: with store_page.context.expect_page(timeout=60000) as p: item.click(delay=random.uniform(0.1, 0.3)) detail_page = p.value except PlaywrightTimeoutError: logger.warning(f" 「{keyword}」第{page_no}页 第{collected_count}个商品「{title}」- 未检测到新标签页，使用当前页采集详情") detail_page = None target_page = detail_page if detail_page else store_page target_page.wait_for_load_state("networkidle", timeout=20000) delay = random_delay(MIN_PAGE_DELAY, MAX_PAGE_DELAY) logger.info(f"📌 「{keyword}」第{page_no}页 第{collected_count}个商品「{title}」- 详情页加载完成，等待{delay:.2f}秒（反爬）") # 详情页链接 product_link = target_page.url logger.info(f"{'='*10}「{keyword}」第{collected_count}个商品 - 详情页链接：{product_link}{'='*10}") # 生产日期 manufacture_date_locator = target_page.locator( '//div[contains(@class, "spec-info-item") and .//div[contains(@class, "spec-info-item-label") and normalize-space(.)="生产日期"]]//div[contains(@class, "spec-info-item-value-text")]') if manufacture_date_locator.count() > 0: manufacture_date = manufacture_date_locator.inner_text(timeout=3000).strip() logger.info(f"{'='*10}「{keyword}」第{collected_count}个商品 - 详情页生产日期：{manufacture_date}{'='*10}") else: logger.warning(f" 「{keyword}」第{collected_count}个商品「{title}」- 生产日期元素未找到") # 批准文号 approval_number_locator = target_page.locator( '//div[contains(@class, "spec-info-item") and .//div[contains(@class, "spec-info-item-label") and normalize-space(.)="批准文号"]]//div[contains(@class, "spec-info-item-value-text")]') if approval_number_locator.count() > 0: approval_number = approval_number_locator.inner_text(timeout=3000).strip() logger.info(f"{'='*10}「{keyword}」第{collected_count}个商品 - 详情页批准文号：{approval_number}{'='*10}") else: logger.warning(f" 「{keyword}」第{collected_count}个商品「{title}」- 批准文号元素未找到") # 规格 spec_locator = target_page.locator( '//div[contains(@class, "spec-info-item") and .//div[contains(@class, "spec-info-item-label") and normalize-space(.)="规格"]]//div[contains(@class, "spec-info-item-value-text")]') if spec_locator.count() > 0: spec = spec_locator.inner_text(timeout=3000).strip() logger.info(f"{'='*10}「{keyword}」第{collected_count}个商品 - 详情页规格：{spec}{'='*10}") else: logger.warning(f" 「{keyword}」第{collected_count}个商品「{title}」- 规格元素数量不足") # 库存 storage = '' storage_locator = target_page.locator('[data-v-51f0e85d].detail-input-num-right-title') if storage_locator.count() > 0: storage = storage_locator.inner_text(timeout=3000).strip() logger.info(f"{'='*10}「{keyword}」第{collected_count}个商品 - 详情页库存：{storage}{'='*10}") # 销量 sell = '' sell_locator = target_page.locator('div.detail-info-content-item-value-price-top-right div[data-v-95163d4a]', has_text='已售') if sell_locator.count() > 0: sell = sell_locator.inner_text(timeout=3000).strip() logger.info(f"{'='*10}「{keyword}」第{collected_count}个商品 - 详情页销量：{sell}{'='*10}") # 快照 oss_url = "" try: local_path, oss_url = screenshot_target_page_to_local_then_oss(target_page=target_page, full_page=True) logger.info("详情页快照已上传 local=%s oss=%s", local_path, oss_url) except Exception as e: logger.warning("详情页快照上传失败：%s", e) # ---- 店铺资质采集 ---- province = "" city = "" business_license_company = "" qualification_number = '' shop_exists, shop_info = shop_is_exists_database(shop) shop_page = None store_url = '' if not shop_exists: logger.info("数据库没有该店铺的营业执照，开始采集店铺资质") license_info = self._collect_shop_license(target_page, shop) if license_info: shop_page = license_info["shop_page"] store_url = license_info["store_url"] province = license_info["province"] city = license_info["city"] business_license_company = license_info["business_license_company"] qualification_number = license_info["qualification_number"] business_license_address = license_info["business_license_address"] else: logger.info("数据库有该店名，在数据库拿取对应字段填充") if shop_info: province = shop_info.get("province", "") or "" city = shop_info.get("city", "") or "" business_license_company = shop_info.get("business_license_company", "") or "" qualification_number = shop_info.get("qualification_number", "") or "" business_license_address = shop_info.get("business_license_address", "") or "" # 关闭店铺页 try: if shop_page and not shop_page.is_closed(): random_delay(4, 8) shop_page.close() logger.info(f"📌 「{keyword}」第{collected_count}个商品 - 已关闭店铺页标签 shop_page") except Exception as e: logger.warning(f"⚠️ 关闭 shop_page 失败：{e}") random_delay(5, 8) # 关闭详情页，切回列表页 if detail_page and not detail_page.is_closed(): detail_page.close() logger.info(f"📌 「{keyword}」第{collected_count}个商品 - 已关闭详情页标签页") store_page.bring_to_front() store_page.mouse.move(random.randint(100, 300), random.randint(200, 400)) random_delay(0.5, 1.0) store_page.wait_for_load_state("networkidle") random_delay(MIN_CLICK_DELAY, MAX_CLICK_DELAY) logger.info(f" 「{keyword}」第{collected_count}个商品「{title}」- 已切回列表页") random_delay(2, 4) # 省市ID province_id, city_id = get_province_city_ids(province, city) # ---- 组装 product dict（兼容 DrugPipeline） ---- now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S") product = { "platform": self.platform_id, "item_id": product_id, "enterprise_id": self.company_id, "product_name": title, "spec": spec, "one_price": discount_price_val, "detail_url": product_link, "shop_name": shop, "anonymous_store_name": "", "shop_url": store_url, "city_name": city, "city_id": city_id, "province_name": province, "province_id": province_id, "shipment_city_name": "", "shipment_city_id": 0, "shipment_province_name": "", "shipment_province_id": 0, "area_info": business_license_address, "factory_name": manufacturer, "scrape_date": current_time, "price": discount_price_val, "sales": sell, "stock_count": storage, "snapshot_url": oss_url, "approval_num": approval_number, "produced_time": manufacture_date, "deadline": expiry_date, "update_time": now_str, "insert_time": now_str, "number": 1, "product_brand": brand, "collect_task_id": self.collect_task_id, "search_name": keyword, "company_name": business_license_company, "collect_config_info": self.collect_config_info, "account_id": self.collect_equipment_account_id, "collect_region_id": self.collect_region_id, "collect_round": self.collect_round, "is_sold_out": is_sold_out, } # 走 DrugPipeline 入库（含去重 + 店铺城市补全） self.pipeline.storge_data(product) collect_result.append(product) logger.info(f" 「{keyword}」第{collected_count}个商品「{title}」采集完成") except Exception as e: logger.exception(f" 「{keyword}」第{collected_count}个商品采集核心异常：{str(e)}") try: if detail_page and not detail_page.is_closed(): detail_page.close() if store_page and not store_page.is_closed(): store_page.bring_to_front() store_page.wait_for_load_state("networkidle") random_delay(MIN_CLICK_DELAY, MAX_CLICK_DELAY) except Exception as e2: logger.error(f" 「{keyword}」第{collected_count}个商品详情采集异常（处理时）：{str(e2)}，原异常：{str(e)}") continue # 每采6个商品滚动一次 if collected_count % 6 == 0 and collected_count > 0 and collected_count != total_limit: logger.info("采满6个往下滑") slow_scroll_400px(store_page) store_page.wait_for_load_state("networkidle") if self.scheduler.end: break delay = random_delay(1.5, 3.0) logger.info(f"⏳ 翻页前随机等待 {delay:.2f}s（反爬）") # 每页结束上报进度 prog_status = self._report_progress(page_no, total_pages=total_pages) if prog_status == "limit_reached": logger.warning("⛔ 进度上报检测到限额，通知停止") self.scheduler.set_flag(True) # 翻页 if self._goto_next_page(store_page): logger.info(f"「{keyword}」还有下一页") page_no += 1 store_page.wait_for_load_state("networkidle") total_limit = store_page.locator(PRODUCT_ITEM_SELECTOR).count() logger.info(f"📌 「{keyword}」第{page_no}页 商品个数更新为：{total_limit}") continue else: logger.info(f" 「{keyword}」已无下一页，关键词采集结束") break long_delay = random_delay(MIN_KEYWORD_DELAY, MAX_KEYWORD_DELAY) logger.info(f" 「{keyword}」采集完成，共{len(collect_result)}条数据（pipeline 实际入库 {self.pipeline.crawl_count} 条），等待{long_delay:.2f}秒后继续下一个关键词（反爬）") return collect_result, page_no # ---------- main entry ---------- def run(self): """蜘蛛主入口，返回 (crawl_count, is_success)。 由 CollectScheduleRunner 调用： count, is_success = YbmCrawler(task_dict).run() """ load_city_mapping() logger.info("\n" + "=" * 50) logger.info("🚀 药帮忙采集程序启动") logger.info(f"⏰ 启动时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") logger.info("=" * 50) with sync_playwright() as p: browser = p.chromium.launch( channel="chrome", headless=True, slow_mo=random.randint(100, 300), args=[ "--disable-blink-features=AutomationControlled", "--enable-automation=false", "--disable-infobars", "--remote-debugging-port=0", "--start-maximized", "--disable-extensions", "--disable-plugins-discovery", "--no-sandbox", "--disable-dev-shm-usage", f"--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{random.randint(110, 120)}.0.0.0 Safari/537.36", ], ) context = browser.new_context( locale="zh-CN", timezone_id="Asia/Shanghai", geolocation={"latitude": 31.230416, "longitude": 121.473701}, permissions=["geolocation"], user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", viewport={"width": 1800, "height": 1000}, java_script_enabled=True, bypass_csp=True, ) page = context.new_page() page.add_init_script(""" Object.defineProperty(navigator, 'webdriver', { get: () => undefined }); Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3] }); Object.defineProperty(navigator, 'mimeTypes', { get: () => [1, 2, 3] }); window.chrome = { runtime: {}, loadTimes: () => ({}) }; delete window.navigator.languages; window.navigator.languages = ['zh-CN', 'zh']; (() => { const originalAddEventListener = EventTarget.prototype.addEventListener; EventTarget.prototype.addEventListener = function(type, listener) { if (type === 'mousemove') { return originalAddEventListener.call(this, type, (e) => { e._automation = undefined; listener(e); }); } return originalAddEventListener.call(this, type, listener); }; })(); """) try: # 登录 load_cookies(context) if not self._is_login(page): page.goto(TARGET_LOGIN_URL) page.wait_for_load_state("networkidle") logger.info("🔑 开始执行登录流程") login_success = self._login_operation(page) if not login_success: logger.error(" 登录失败，程序终止") self.is_success = False self._page_no = 0 return 0, False save_cookies(context) logger.info(" 登录并保存Cookie成功！") # 启动心跳守护线程（CrawlerScheduler） self.scheduler.start() # 搜索 popup_guard(page, "before_search") store_page, search_success = self._search_operation(page, self.keyword, is_first_search=True) if store_page: popup_guard(store_page, "after_search") if store_page is None: logger.warning(f" 「{self.keyword}」搜索页面为空，任务失败") self.is_success = False self._page_no = 0 return 0, False if not search_success: logger.warning(f" 「{self.keyword}」搜索失败，跳过采集") self.is_success = False self._page_no = 0 return 0, False store_page.wait_for_load_state("domcontentloaded") store_page.wait_for_load_state('networkidle') # 上报采集开始，检查是否限额 report_status = self._report_start() if report_status == "limit_reached": logger.warning("⛔ 任务 %s 开始上报时账号已达限额，退出当前任务", self.task_id) self.is_success = False self._page_no = 0 return 0, False # 核心采集 data_list, page_no = self._collect_data(store_page) self._page_no = page_no real_count = len(data_list) self.is_success = True logger.info(f"关键词「{self.keyword}」采集完成，共 {real_count} 条数据（pipeline 实际入库 {self.pipeline.crawl_count} 条），共 {page_no} 页") except Exception as e: logger.exception("程序异常：%s", e) self.is_success = False finally: self.scheduler.stop() # 上报采集结束 try: self._report_end( success=self.is_success, real_count=self.pipeline.crawl_count, page_no=getattr(self, '_page_no', 0), total_pages=0, ) except Exception: pass browser.close() logger.info(" 浏览器已关闭，程序结束") return self.pipeline.crawl_count, self.is_success # ==================== SECTION 9: ENTRY POINT (测试/直接运行) ==================== if __name__ == '__main__': # 直接运行：使用手动任务进行测试 load_city_mapping() if USE_MANUAL_TASK: task = MANUAL_TASK else: # 从内部 API 拉取任务 scheduler = CrawlerScheduler(YBM_DEVICE_ID, str(PLATFORM_ID)) task = scheduler.get_task() or {} if not task: logger.error("拉取任务失败") if not task: logger.error("未获取到任何任务，程序退出") else: crawler = YbmCrawler(task) count, success = crawler.run() logger.info(f"测试完成：crawl_count={count}, is_success={success}") # 发送飞书通知 try: spec_items = [item.strip() for item in re.split(r"[|、，,\n\r]+", task.get("product_specs", "") or "") if item.strip()] display_spec = "、".join(spec_items) notice_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S") if success: feishu_send_text( f"{notice_time} 通知:\n" f"平台: 药帮忙, 药品: {task.get('product_name', '')}, 品规: {display_spec}, 爬取数据: {count}条" ) else: feishu_send_error_card( task_name=task.get("product_name", ""), err_msg=f"task_id={task.get('id')}, company_id={task.get('company_id')}, crawl_count={count}", mention_all=False, ) except Exception as e: logger.warning(f"飞书通知发送失败：{e}")