|
|
@@ -0,0 +1,1755 @@
|
|
|
+from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeoutError
|
|
|
+from logger_config import logger
|
|
|
+from datetime import datetime
|
|
|
+import random
|
|
|
+import csv
|
|
|
+import os
|
|
|
+import time
|
|
|
+import json
|
|
|
+import sys
|
|
|
+from config import *
|
|
|
+from conn_mysql import MySQLPoolOnline
|
|
|
+import re
|
|
|
+import uuid
|
|
|
+import requests
|
|
|
+import base64
|
|
|
+from io import BytesIO
|
|
|
+from PIL import Image
|
|
|
+import traceback
|
|
|
+import oss2
|
|
|
+from scheduler import CrawlerScheduler
|
|
|
+
|
|
|
+# ==================== 调度器配置 ====================
|
|
|
+PLATFORM_ID = "7" # 药九九平台编码
|
|
|
+
|
|
|
+# ==================== 从数据库获取可用账号 ====================
|
|
|
+def get_account_from_db():
|
|
|
+ """从 accounts_platform 表获取一个可用的药九九账号（platform=7, status=1）"""
|
|
|
+ sql = """SELECT `name`, `phone`, `password` FROM `accounts_platform`
|
|
|
+ WHERE `platform`=7 AND `status`=1
|
|
|
+ ORDER BY `cookie_timestamp` ASC LIMIT 1"""
|
|
|
+ rows = mysql_pool.select_data(sql)
|
|
|
+ if not rows:
|
|
|
+ logger.error("无可用药九九账号（platform=7, status=1）")
|
|
|
+ return None
|
|
|
+ return rows[0]
|
|
|
+
|
|
|
+# ==================== 回告辅助函数 ====================
|
|
|
+def _build_report_data(task_dict, **kwargs):
|
|
|
+ return {
|
|
|
+ "task_id": task_dict.get("id"),
|
|
|
+ "platform": task_dict.get("_platform", PLATFORM_ID),
|
|
|
+ "username": task_dict.get("_device_id", "unknown"),
|
|
|
+ **kwargs,
|
|
|
+ }
|
|
|
+
|
|
|
+def _report_page_progress(scheduler, task_dict, page, crawled_count, is_finished=0, total_pages=0):
|
|
|
+ """逐页回告，返回 True=code=error 需停止"""
|
|
|
+ if not scheduler:
|
|
|
+ return False
|
|
|
+ data = _build_report_data(task_dict,
|
|
|
+ current_page=page, crawled_count=crawled_count, is_finished=is_finished)
|
|
|
+ if total_pages:
|
|
|
+ data["total_pages"] = total_pages
|
|
|
+ scheduler.post_report(data)
|
|
|
+ return scheduler.end
|
|
|
+
|
|
|
+def _report_exception(scheduler, task_dict, page, exception_type, remark="", crawled_count=0):
|
|
|
+ """异常回告"""
|
|
|
+ if not scheduler:
|
|
|
+ return
|
|
|
+ data = _build_report_data(task_dict,
|
|
|
+ current_page=page, is_finished=0,
|
|
|
+ exception_type=exception_type, remark=remark)
|
|
|
+ if page:
|
|
|
+ data["total_pages"] = page
|
|
|
+ if crawled_count:
|
|
|
+ data["crawled_count"] = crawled_count
|
|
|
+ scheduler.post_report(data)
|
|
|
+# from faker import Faker
|
|
|
+
|
|
|
+# 代理IP池
|
|
|
+PROXY_POOL_URL = ""
|
|
|
+PROXY_VALIDATION_URL = "" # 用于验证代理有效性的URL
|
|
|
+PROXY_TIMEOUT = 10 # 代理验证超时时间(秒)
|
|
|
+
|
|
|
+mysql_pool = MySQLPoolOnline()
|
|
|
+finished_task_ids = set()
|
|
|
+
|
|
|
+
|
|
|
+def _chromium_launch_args(chrome_major=None):
|
|
|
+ """Playwright Chromium 启动参数（反自动化 + 随机 UA 主版本）。"""
|
|
|
+ major = chrome_major if chrome_major is not None else random.randint(110, 120)
|
|
|
+ return [
|
|
|
+ "--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 "
|
|
|
+ f"(KHTML, like Gecko) Chrome/{major}.0.0.0 Safari/537.36",
|
|
|
+ ]
|
|
|
+
|
|
|
+
|
|
|
+def get_random_proxy():
|
|
|
+ """从代理池获取随机代理IP"""
|
|
|
+ if not PROXY_POOL_URL.strip():
|
|
|
+ return None
|
|
|
+ try:
|
|
|
+ response = requests.get(PROXY_POOL_URL, timeout=10)
|
|
|
+ if response.status_code == 200:
|
|
|
+ proxy = response.text.strip()
|
|
|
+ if validate_proxy(proxy):
|
|
|
+ logger.info(f"获取到有效代理: {proxy}")
|
|
|
+ return proxy
|
|
|
+ logger.warning(f"代理无效: {proxy}")
|
|
|
+ except Exception as e:
|
|
|
+ logger.error(f"获取代理失败: {str(e)}")
|
|
|
+ return None
|
|
|
+
|
|
|
+def validate_proxy(proxy):
|
|
|
+ """验证代理IP有效性"""
|
|
|
+ if not PROXY_VALIDATION_URL.strip():
|
|
|
+ return False
|
|
|
+ try:
|
|
|
+ proxies = {
|
|
|
+ "http": f"http://{proxy}",
|
|
|
+ "https": f"https://{proxy}"
|
|
|
+ }
|
|
|
+ response = requests.get(
|
|
|
+ PROXY_VALIDATION_URL,
|
|
|
+ proxies=proxies,
|
|
|
+ timeout=PROXY_TIMEOUT
|
|
|
+ )
|
|
|
+ return response.status_code == 200
|
|
|
+ except Exception:
|
|
|
+ return False
|
|
|
+
|
|
|
+def init_browser_with_proxy(playwright):
|
|
|
+ proxy = get_random_proxy()
|
|
|
+ proxy_config = None
|
|
|
+ if proxy:
|
|
|
+ proxy_server, proxy_port = proxy.split(":")
|
|
|
+ proxy_config = {
|
|
|
+ "server": f"http://{proxy_server}:{proxy_port}",
|
|
|
+ # "username": "your_proxy_username",
|
|
|
+ # "password": "your_proxy_password"
|
|
|
+ }
|
|
|
+ logger.info(f"使用代理: {proxy_server}:{proxy_port}")
|
|
|
+ else:
|
|
|
+ logger.warning("未获取到有效代理，将使用本地IP")
|
|
|
+
|
|
|
+ # 启动浏览器（保留原有反爬配置）
|
|
|
+ return playwright.chromium.launch(
|
|
|
+ headless=False, # 非无头模式
|
|
|
+ channel="chrome", # 使用Chrome内核
|
|
|
+ slow_mo=random.randint(100, 300), # 随机操作延迟
|
|
|
+ proxy=proxy_config, # 代理配置（None则不使用代理）
|
|
|
+ args=_chromium_launch_args(),
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+# ==================== 2. 反爬工具函数 ====================
|
|
|
+def random_delay(min_seconds, max_seconds):
|
|
|
+ """生成随机延迟（核心反爬：避免固定间隔）"""
|
|
|
+ delay = random.uniform(min_seconds, max_seconds)
|
|
|
+ time.sleep(delay)
|
|
|
+ return delay
|
|
|
+
|
|
|
+
|
|
|
+def simulate_human_typing(page, locator, text):
|
|
|
+ """模拟真人打字（逐个字符输入，带随机间隔）"""
|
|
|
+ try:
|
|
|
+ locator.click()
|
|
|
+ locator.clear()
|
|
|
+ for char in text:
|
|
|
+ locator.type(char, delay=random.uniform(MIN_INPUT_DELAY, MAX_INPUT_DELAY))
|
|
|
+ random_delay(0.05, 0.1) # 字符间额外小延迟
|
|
|
+ logger.info(f" 模拟真人输入完成：{text}")
|
|
|
+ except Exception as e:
|
|
|
+ logger.error(f"模拟打字失败：{e}")
|
|
|
+ locator.fill(text) # 兜底：直接填充
|
|
|
+
|
|
|
+
|
|
|
+def save_cookies(context, cookie_path=COOKIE_FILE_PATH):
|
|
|
+ """保存Cookie到本地JSON文件"""
|
|
|
+ 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):
|
|
|
+ """从本地JSON文件加载Cookie到浏览器上下文"""
|
|
|
+ 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 is_login(page):
|
|
|
+ """验证是否已登录（核心：检测登录态）"""
|
|
|
+ try:
|
|
|
+ # 访问需要登录的页面
|
|
|
+ page.goto(LOGIN_VALIDATE_URL, timeout=ELEMENT_TIMEOUT)
|
|
|
+ time.sleep(5)
|
|
|
+ page.wait_for_load_state("networkidle")
|
|
|
+
|
|
|
+ # 检测是否跳转到登录页（URL包含login则未登录）
|
|
|
+ if "login" in page.url.lower():
|
|
|
+ logger.warning(" Cookie失效，需要重新登录")
|
|
|
+ return False
|
|
|
+
|
|
|
+ # 可选：检测登录后的专属元素（比如用户名、个人中心等）
|
|
|
+ # if page.locator("用户中心选择器").count() > 0:
|
|
|
+ # return True
|
|
|
+ logger.info(" Cookie有效，已保持登录状态")
|
|
|
+ return True
|
|
|
+ except Exception as e:
|
|
|
+ logger.error(f" 验证登录状态失败：{e}")
|
|
|
+ return False
|
|
|
+
|
|
|
+# ==================== 滚动函数重构（核心修改） ====================
|
|
|
+def slow_scroll_400px(page,scroll_distance1=400):
|
|
|
+ """
|
|
|
+ 慢速滚动400px±50px（模拟真人滑动）
|
|
|
+ :param page: 页面对象
|
|
|
+ :return: 滚动是否成功
|
|
|
+ """
|
|
|
+ try:
|
|
|
+ # 生成400±50px的随机滚动距离
|
|
|
+ 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}秒）"
|
|
|
+ )
|
|
|
+
|
|
|
+ # 渐进式滚动（每步50px，间隔0.05秒）
|
|
|
+ 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
|
|
|
+
|
|
|
+# CSV配置
|
|
|
+CSV_FILE_PATH = f"yjj_collect_data_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv" # CSV保存路径
|
|
|
+CSV_HEADERS = [
|
|
|
+ "商品标题", "商品采购价格", "商品折扣价格", "规格", "盒数",
|
|
|
+ "店铺名称", "公司名称",
|
|
|
+ "有效日期", "生产日期", "批准文号", "采集时间"
|
|
|
+] #表头
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+# ==================== 登录函数 ====================
|
|
|
+def login_operation(page, username, password):
|
|
|
+ """登录操作函数"""
|
|
|
+ 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(" 已输入登录密码")
|
|
|
+
|
|
|
+ # 点击登录按钮
|
|
|
+ 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
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+def kill_masks(page):
|
|
|
+ """
|
|
|
+ 强制清理残留遮罩层/覆盖层，并恢复 body 可滚动、可点击状态
|
|
|
+ """
|
|
|
+ page.evaluate(r"""
|
|
|
+ () => {
|
|
|
+ const removed = [];
|
|
|
+ const hidden = [];
|
|
|
+
|
|
|
+ // 1) 先处理已知常见遮罩
|
|
|
+ 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 => {
|
|
|
+ // v-modal / overlay 直接 remove 最省事
|
|
|
+ removed.push(sel);
|
|
|
+ el.remove();
|
|
|
+ });
|
|
|
+ }
|
|
|
+
|
|
|
+ // 2) 再做一次“泛化兜底”：全屏 fixed/absolute + 高 z-index 的覆盖层
|
|
|
+ // 注意：不要误删页面正常的固定导航，所以加上“近似全屏”的判断
|
|
|
+ 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);
|
|
|
+ const pos = s.position;
|
|
|
+ const pe = s.pointerEvents;
|
|
|
+
|
|
|
+ if ((pos === 'fixed' || pos === 'absolute') && z >= 1000 && pe !== 'none') {
|
|
|
+ const r = el.getBoundingClientRect();
|
|
|
+ const nearFullScreen =
|
|
|
+ r.width >= window.innerWidth * 0.8 &&
|
|
|
+ r.height >= window.innerHeight * 0.8 &&
|
|
|
+ r.left <= window.innerWidth * 0.1 &&
|
|
|
+ r.top <= window.innerHeight * 0.1;
|
|
|
+
|
|
|
+ // 常见遮罩是半透明背景色，或者透明但拦截点击
|
|
|
+ const bg = s.backgroundColor || '';
|
|
|
+ const looksLikeMask =
|
|
|
+ nearFullScreen && (bg.includes('rgba') || bg.includes('rgb') || s.opacity !== '1');
|
|
|
+
|
|
|
+ if (nearFullScreen) {
|
|
|
+ // 不管透明不透明，只要近似全屏且高 z-index，就先让它不拦截点击
|
|
|
+ el.style.pointerEvents = 'none';
|
|
|
+ el.style.display = 'none';
|
|
|
+ hidden.push(el.tagName + '.' + (el.className || ''));
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ // 3) 恢复 body / html 的滚动与交互（很多弹窗会锁滚动）
|
|
|
+ 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';
|
|
|
+
|
|
|
+ // 4) 去掉 Element-UI 常见的锁定 class
|
|
|
+ document.body.classList.remove('el-popup-parent--hidden');
|
|
|
+
|
|
|
+ return { removed, hiddenCount: hidden.length, hidden };
|
|
|
+ }
|
|
|
+ """)
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+def force_close_popup(page):
|
|
|
+ """关闭新手引导/遮罩（多步：下一步/完成/我知道了），并兜底移除遮罩层"""
|
|
|
+ try:
|
|
|
+ # 1) 尝试连续点“下一步/完成/我知道了/关闭”
|
|
|
+ for _ in range(5): # 最多点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
|
|
|
+
|
|
|
+ # 有些引导是右上角 X（如果存在就点）
|
|
|
+ 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
|
|
|
+
|
|
|
+ # 2) 兜底：移除常见遮罩层（element-ui / 通用 mask/overlay）
|
|
|
+ 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);
|
|
|
+ // 只移除“覆盖层”倾向的元素：fixed/absolute 且 z-index 很高
|
|
|
+ if ((s.position === 'fixed' || s.position === 'absolute') && parseInt(s.zIndex || '0', 10) >= 1000) {
|
|
|
+ el.remove();
|
|
|
+ }
|
|
|
+ });
|
|
|
+ }
|
|
|
+ """)
|
|
|
+ except Exception:
|
|
|
+ pass
|
|
|
+
|
|
|
+
|
|
|
+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
|
|
|
+
|
|
|
+ # 兜底：直接找任意可见的（避免命中 hidden 模板）
|
|
|
+ candidate = page.locator(f"{SEARCH_INPUT_SELECTOR}:visible").first
|
|
|
+ candidate.wait_for(state="visible", timeout=ELEMENT_TIMEOUT)
|
|
|
+ return candidate
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+def type_slow(locator, text, 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 search_operation(page, keyword):
|
|
|
+ """搜索框填充+提交搜索"""
|
|
|
+ try:
|
|
|
+
|
|
|
+ search_url = f"https://www.yyjzt.com/search?keyword={keyword}"
|
|
|
+ page.goto(search_url)
|
|
|
+ page.wait_for_timeout(600)
|
|
|
+ try:
|
|
|
+ page.wait_for_load_state("networkidle", timeout=10000)
|
|
|
+ except Exception:
|
|
|
+ pass
|
|
|
+ force_close_popup(page)
|
|
|
+ kill_masks(page)
|
|
|
+ logger.info("✅ 已触发搜索")
|
|
|
+ return True
|
|
|
+
|
|
|
+ except PlaywrightTimeoutError as e:
|
|
|
+ logger.error(f" 搜索失败：元素定位超时 - {str(e)}")
|
|
|
+ return False
|
|
|
+ except Exception as e:
|
|
|
+ logger.error(f" 搜索异常：{str(e)}")
|
|
|
+ return False
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+#翻下一页
|
|
|
+def goto_next_page(page):
|
|
|
+ """
|
|
|
+ 尝试翻到下一页；成功返回True，没下一页/翻页失败返回False
|
|
|
+ 适配常见 ElementUI: .el-pagination .btn-next / .el-pagination__next
|
|
|
+ """
|
|
|
+ # 多写几个候选，哪个能用就用哪个
|
|
|
+ candidates = [
|
|
|
+ ".el-pagination button.btn-next:not(.is-disabled)",
|
|
|
+ ".el-pagination__next:not(.is-disabled)",
|
|
|
+ "button:has-text('下一页'):not([disabled])",
|
|
|
+ "a:has-text('下一页')",
|
|
|
+ ]
|
|
|
+
|
|
|
+
|
|
|
+ next_btn = None
|
|
|
+ for sel in candidates:
|
|
|
+ loc = page.locator(sel).first
|
|
|
+ if loc.count() > 0:
|
|
|
+ next_btn = loc
|
|
|
+ break
|
|
|
+
|
|
|
+ if not next_btn:
|
|
|
+ return False
|
|
|
+
|
|
|
+ # 用“当前页第一个商品标题”做翻页完成的判据（比只等networkidle更稳）
|
|
|
+ first_title = page.locator(PRODUCT_TITLE_SELECTOR).first
|
|
|
+ before = ""
|
|
|
+ try:
|
|
|
+ if first_title.count() > 0:
|
|
|
+ before = first_title.inner_text(timeout=2000).strip()
|
|
|
+ except:
|
|
|
+ pass
|
|
|
+
|
|
|
+ try:
|
|
|
+ page.evaluate("window.scrollTo(0, 0);")
|
|
|
+ next_btn.click(timeout=5000)
|
|
|
+ page.wait_for_load_state("networkidle")
|
|
|
+
|
|
|
+ # 等列表发生变化（标题变了 / 或者至少第一个标题重新出现）
|
|
|
+ if before:
|
|
|
+ page.wait_for_function(
|
|
|
+ """(sel, oldText) => {
|
|
|
+ const el = document.querySelector(sel);
|
|
|
+ return el && el.innerText && el.innerText.trim() !== oldText;
|
|
|
+ }""",
|
|
|
+ arg=(PRODUCT_TITLE_SELECTOR, before),
|
|
|
+ timeout=5000
|
|
|
+ )
|
|
|
+ else:
|
|
|
+ first_title.wait_for(timeout=1000)
|
|
|
+
|
|
|
+ return True
|
|
|
+ except Exception as e:
|
|
|
+ logger.warning(f" 翻页失败：{e}")
|
|
|
+ return False
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+def popup_guard(page, tag=""):
|
|
|
+ """
|
|
|
+ 全局弹窗/遮罩守卫：多步引导 + 关闭按钮 + 遮罩清理 + 恢复滚动
|
|
|
+ tag 仅用于日志区分调用位置
|
|
|
+ """
|
|
|
+ try:
|
|
|
+ # 给弹窗一点出现时间
|
|
|
+ page.wait_for_timeout(300)
|
|
|
+
|
|
|
+ # 1) 连续点“下一步/完成/我知道了/关闭”
|
|
|
+ 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
|
|
|
+
|
|
|
+ # 2) 常见的 close icon
|
|
|
+ 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
|
|
|
+
|
|
|
+ # 3) 清遮罩 + 恢复滚动/交互
|
|
|
+ page.evaluate(r"""
|
|
|
+ () => {
|
|
|
+ // 第一步：精准清理已知的遮罩/弹窗类名（Element UI框架常用）
|
|
|
+ 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()));
|
|
|
+
|
|
|
+ // 泛化兜底：近似全屏 + 高 z-index 的层直接屏蔽
|
|
|
+ const all = Array.from(document.querySelectorAll('body *'));
|
|
|
+ for (const el of all) {
|
|
|
+ const s = getComputedStyle(el); // 获取元素的实际样式（含CSS生效的样式）
|
|
|
+ const z = parseInt(s.zIndex || '0', 10); // 取元素的层级（z-index），默认0
|
|
|
+ // 条件1：元素是固定/绝对定位（弹窗/遮罩常见定位方式）+ 层级≥1000（高优先级遮挡）+ 能拦截鼠标事件
|
|
|
+ if ((s.position === 'fixed' || s.position === 'absolute') && z >= 1000 && s.pointerEvents !== 'none') {
|
|
|
+ const r = el.getBoundingClientRect(); // 获取元素的尺寸和位置
|
|
|
+ // 条件2：元素宽度/高度≥屏幕80%（近似全屏遮罩）
|
|
|
+ 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'; // html标签恢复滚动
|
|
|
+ document.body.style.overflow = 'auto'; // body标签恢复滚动
|
|
|
+ document.body.classList.remove('el-popup-parent--hidden'); // 移除Element UI的滚动禁用类
|
|
|
+ }
|
|
|
+ """)
|
|
|
+
|
|
|
+ logger.info("杀除弹窗成功")
|
|
|
+ except Exception:
|
|
|
+ pass
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+def open_detail_page(list_page, item, keyword, idx, *, timeout=15000):
|
|
|
+ """
|
|
|
+ 点击商品进入详情页，兼容：
|
|
|
+ 1) 新开 tab（返回 detail_page != list_page, opened_new_tab=True）
|
|
|
+ 2) 同 tab 跳转（detail_page == list_page, opened_new_tab=False）
|
|
|
+ """
|
|
|
+ ctx = list_page.context
|
|
|
+ list_url = list_page.url
|
|
|
+ detail_page = None
|
|
|
+ opened_new_tab = False
|
|
|
+
|
|
|
+ try:
|
|
|
+ # 期望新开 tab（很多站点会这样）
|
|
|
+ with ctx.expect_page(timeout=timeout) as p:
|
|
|
+ item.click(delay=random.uniform(0.1, 0.3))
|
|
|
+ detail_page = p.value
|
|
|
+ opened_new_tab = True
|
|
|
+ logger.info(f" 「{keyword}」第{idx}个商品 - 新开标签页进入详情")
|
|
|
+ except PlaywrightTimeoutError:
|
|
|
+ # 兜底：没新开 tab，大概率是同页跳转/弹层
|
|
|
+ detail_page = list_page
|
|
|
+ opened_new_tab = False
|
|
|
+ logger.info(f" 「{keyword}」第{idx}个商品 - 未新开标签页，按同页进入详情处理")
|
|
|
+
|
|
|
+ return detail_page, opened_new_tab, list_url
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+def return_to_list(list_page, detail_page, opened_new_tab, list_url, keyword, idx):
|
|
|
+ """
|
|
|
+ 从详情页返回列表页：
|
|
|
+ - 新 tab：关闭 tab，然后 bring_to_front 切回
|
|
|
+ - 同 tab：尽量 go_back 回到 list_url；如果没跳转而是弹层，尝试 ESC
|
|
|
+ """
|
|
|
+ # 如果浏览器/页面已经被关了，直接退出，避免二次异常
|
|
|
+ if list_page is None or list_page.is_closed():
|
|
|
+ logger.warning(f" 「{keyword}」第{idx}个商品 - 列表页已关闭，无法切回")
|
|
|
+ return
|
|
|
+
|
|
|
+ if opened_new_tab:
|
|
|
+ # 只关“新开的详情 tab”，绝不关 list_page
|
|
|
+ try:
|
|
|
+ if detail_page and (detail_page is not list_page) and (not detail_page.is_closed()):
|
|
|
+ detail_page.close()
|
|
|
+ logger.info(f"📌 「{keyword}」第{idx}个商品 - 已关闭详情页标签页")
|
|
|
+ except Exception as e:
|
|
|
+ logger.warning(f" 「{keyword}」第{idx}个商品 - 关闭详情页失败：{e}")
|
|
|
+
|
|
|
+ # 切回列表页
|
|
|
+ try:
|
|
|
+ list_page.bring_to_front()
|
|
|
+ list_page.mouse.move(random.randint(100, 300), random.randint(200, 400))
|
|
|
+ random_delay(0.3, 0.8)
|
|
|
+ list_page.wait_for_load_state("networkidle")
|
|
|
+ logger.info(f" 「{keyword}」第{idx}个商品 - 已切回列表页（新tab模式）")
|
|
|
+ except Exception as e:
|
|
|
+ logger.warning(f" 「{keyword}」第{idx}个商品 - 切回列表页失败：{e}")
|
|
|
+ return
|
|
|
+
|
|
|
+ # 同 tab：detail_page == list_page
|
|
|
+ try:
|
|
|
+ # 1) 如果 URL 变了，说明确实跳转了 → go_back 回去
|
|
|
+ if list_page.url != list_url:
|
|
|
+ for _ in range(3): # 最多退 3 次，防止死循环
|
|
|
+ list_page.go_back(timeout=15000)
|
|
|
+ list_page.wait_for_load_state("domcontentloaded", timeout=15000)
|
|
|
+ random_delay(0.2, 0.5)
|
|
|
+ if list_page.url == list_url:
|
|
|
+ break
|
|
|
+ logger.info(f" 「{keyword}」第{idx}个商品 - 已返回列表页（同tab跳转模式）")
|
|
|
+ else:
|
|
|
+ # 2) URL 没变：可能是弹层详情 → 尝试 ESC 关闭弹层
|
|
|
+ list_page.keyboard.press("Escape")
|
|
|
+ random_delay(0.2, 0.5)
|
|
|
+ logger.info(f" 「{keyword}」第{idx}个商品 - 已尝试关闭弹层并留在列表页（同tab弹层模式）")
|
|
|
+
|
|
|
+ list_page.bring_to_front()
|
|
|
+ list_page.wait_for_load_state("networkidle")
|
|
|
+ except Exception as e:
|
|
|
+ logger.warning(f" 「{keyword}」第{idx}个商品 - 同tab返回列表页失败：{e}")
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+#判断店名是否已经在数据库
|
|
|
+def shop_is_exists_database(shop):
|
|
|
+ query_sql = """
|
|
|
+ SELECT province, city, business_license_company, qualification_number
|
|
|
+ FROM retrieve_yjj_shop_info_middle
|
|
|
+ WHERE shop = %s
|
|
|
+ """
|
|
|
+ try:
|
|
|
+ rows = mysql_pool.select_data(query_sql, (shop,))
|
|
|
+ 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):
|
|
|
+ """
|
|
|
+ 把字段插入到yjj_shop_info_middle表
|
|
|
+ :param 各参数: 你要插入的字段值（空字符串也可）
|
|
|
+ :return: bool - 插入成功返回True，失败返回False
|
|
|
+ """
|
|
|
+ sql = """
|
|
|
+ INSERT INTO retrieve_yjj_shop_info_middle (
|
|
|
+ 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,
|
|
|
+ )
|
|
|
+
|
|
|
+ try:
|
|
|
+ n = mysql_pool.execute(sql, params)
|
|
|
+ if n > 0:
|
|
|
+ logger.info("店铺信息已入库 | shop=%s", shop)
|
|
|
+ return True
|
|
|
+ logger.error("店铺信息插入失败：受影响行数为 0 | shop=%s", shop)
|
|
|
+ return False
|
|
|
+ except Exception as e:
|
|
|
+ logger.error("店铺信息插入失败：%s", e)
|
|
|
+ return False
|
|
|
+
|
|
|
+
|
|
|
+def insert_single_to_mysql(single_data):
|
|
|
+ """
|
|
|
+ 逐条插入单条数据到MySQL数据库
|
|
|
+ :param single_data: 单条商品数据元组
|
|
|
+ :return: 插入是否成功
|
|
|
+ """
|
|
|
+ insert_sql = """
|
|
|
+ INSERT INTO retrieve_scrape_data
|
|
|
+ (enterprise_id,product_name, min_price, manufacture_date,
|
|
|
+ expiry_date, store_name, company_name, province_name, city_name, manufacturer, product_specs, approval_number, link_url,
|
|
|
+ scrape_date, is_sold_out, qualification_number, platform_id, number, sales,inventory,snapshot_url,search_name,product_brand,collect_config_info,task_id,insert_time,update_time,collect_equipment_account_id,collect_region_id,collect_round) VALUES (%s, %s, %s, %s,%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,%s, %s, %s);
|
|
|
+ """
|
|
|
+
|
|
|
+ values = (
|
|
|
+ single_data["company_id"],
|
|
|
+ single_data["product"],
|
|
|
+ single_data["min_price"],
|
|
|
+ single_data["manufacture_date"],
|
|
|
+ single_data["expiry_date"],
|
|
|
+ single_data["shop"],
|
|
|
+ single_data["business_license_company"],
|
|
|
+ single_data["province"],
|
|
|
+ single_data["city"],
|
|
|
+ single_data["manufacturer"],
|
|
|
+ single_data["specification"],
|
|
|
+ single_data["approval_number"],
|
|
|
+ single_data["product_link"],
|
|
|
+ single_data["scrape_date"],
|
|
|
+ single_data["is_sold_out"],
|
|
|
+ single_data["credit_code"],
|
|
|
+ 7,
|
|
|
+ single_data["number"],
|
|
|
+ single_data["sales"],
|
|
|
+ single_data["inventory"],
|
|
|
+ single_data["snapshot_url"],
|
|
|
+ single_data["search_name"],
|
|
|
+ single_data["product_brand"],
|
|
|
+ single_data["collect_config_info"],
|
|
|
+ single_data["task_id"],
|
|
|
+ single_data["insert_time"],
|
|
|
+ single_data["update_time"],
|
|
|
+ single_data["collect_equipment_account_id"],
|
|
|
+ single_data["collect_region_id"],
|
|
|
+ single_data["collect_round"],
|
|
|
+ )
|
|
|
+
|
|
|
+ try:
|
|
|
+ n = mysql_pool.execute(insert_sql, values)
|
|
|
+ if n > 0:
|
|
|
+ logger.info("单条采集数据已入库 | product=%s", single_data.get("product", "")[:80])
|
|
|
+ return True
|
|
|
+ logger.error("单条数据插入失败：受影响行数为 0")
|
|
|
+ return False
|
|
|
+ except Exception as e:
|
|
|
+ logger.error("单条数据插入失败：%s", e)
|
|
|
+ return False
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+def check_dup_in_biz_db(product_link, discount_price_val, scrape_date,keyword_dict):
|
|
|
+ """直接查询业务表是否存在该商品链接+价格"""
|
|
|
+ log_context = (
|
|
|
+ f"【去重校验】商品链接：{product_link.strip()} | 价格：{discount_price_val} "
|
|
|
+ f"采集日期：{scrape_date.strip()}"
|
|
|
+ )
|
|
|
+ collect_round = keyword_dict["collect_round"]
|
|
|
+ collect_equipment_account_id = keyword_dict["collect_equipment_account_id"]
|
|
|
+ collect_region_id = keyword_dict["collect_region_id"]
|
|
|
+ sql = """
|
|
|
+ SELECT 1 FROM retrieve_scrape_data
|
|
|
+ WHERE link_url = %s AND min_price = %s AND scrape_date = %s AND platform_id = %s
|
|
|
+ AND collect_round = %s AND collect_equipment_account_id = %s AND collect_region_id = %s
|
|
|
+ LIMIT 1
|
|
|
+ """
|
|
|
+ params = (
|
|
|
+ product_link.strip(),
|
|
|
+ discount_price_val,
|
|
|
+ scrape_date.strip(),
|
|
|
+ 7,
|
|
|
+ collect_round,
|
|
|
+ collect_equipment_account_id,
|
|
|
+ collect_region_id,
|
|
|
+ )
|
|
|
+ try:
|
|
|
+ rows = mysql_pool.select_data(sql, params)
|
|
|
+ is_dup = bool(rows)
|
|
|
+
|
|
|
+ if is_dup:
|
|
|
+ logger.warning(f"{log_context} - 表中已存在重复记录，跳过本次采集")
|
|
|
+ else:
|
|
|
+ logger.info(f"{log_context} - 表中无重复记录，正常采集")
|
|
|
+
|
|
|
+ return is_dup
|
|
|
+
|
|
|
+ except Exception as e:
|
|
|
+ logger.error(f"查询业务表去重失败：{str(e)}")
|
|
|
+ return False
|
|
|
+
|
|
|
+
|
|
|
+# 压缩图片函数
|
|
|
+def compress_image(image_data, max_size=4*1024*1024): # 4MB上限
|
|
|
+ try:
|
|
|
+ img = Image.open(BytesIO(image_data))
|
|
|
+
|
|
|
+ # 将RGBA模式转为RGB（兼容JPEG）
|
|
|
+ if img.mode in ('RGBA', 'P'): # P是PNG的调色板模式，也需转换
|
|
|
+ # 新建白色背景的RGB图片，把透明图贴上去（避免透明区域变黑）
|
|
|
+ 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
|
|
|
+
|
|
|
+ # 缩小分辨率（按比例缩到宽≤1000px）
|
|
|
+ if img.width > 1000:
|
|
|
+ ratio = 1000 / img.width
|
|
|
+ new_size = (int(img.width*ratio), int(img.height*ratio))
|
|
|
+ img = img.resize(new_size, 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"):
|
|
|
+ """下载网络图片，返回图片二进制数据（BytesIO）"""
|
|
|
+ try:
|
|
|
+ if not os.path.exists(save_dir):
|
|
|
+ os.makedirs(save_dir) # 创建多级目录（比如a/b/c）
|
|
|
+ print(f"创建本地保存目录：{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
|
|
|
+
|
|
|
+ # 步骤3：提取图片文件名（从URL中截取，避免重复）
|
|
|
+ # 示例URL：https://xxx.com/123.jpg → 文件名：123.jpg
|
|
|
+ file_name = image_url.split("/")[-1]
|
|
|
+ # 处理特殊字符（避免文件名非法）
|
|
|
+ file_name = file_name.replace("?", "").replace("&", "").replace("=", "")
|
|
|
+ save_path = os.path.join(save_dir, file_name) # 完整保存路径
|
|
|
+
|
|
|
+ # 步骤4：保存图片到本地
|
|
|
+ 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_ocr_res(img):
|
|
|
+ try:
|
|
|
+ #img地址
|
|
|
+ print(f'开始识别图片:{img}')
|
|
|
+ request_url = request_url_config
|
|
|
+
|
|
|
+ img_base64 = download_image_to_base64(img)
|
|
|
+ if not img_base64:
|
|
|
+ print("图片下载/转Base64失败，终止OCR识别")
|
|
|
+ return None
|
|
|
+
|
|
|
+ # 获取access_token
|
|
|
+ access_token = get_access_token()
|
|
|
+ if not access_token:
|
|
|
+ print("获取access_token失败，无法调用OCR接口")
|
|
|
+ return None
|
|
|
+ params = {"image": img_base64}
|
|
|
+ request_url = request_url + "?access_token=" + access_token
|
|
|
+ headers = {'content-type': 'application/x-www-form-urlencoded'}
|
|
|
+ response = requests.post(request_url, data=params, headers=headers)
|
|
|
+
|
|
|
+ if response:
|
|
|
+ res = response.json()
|
|
|
+ # 检查OCR返回是否有错误
|
|
|
+ if "error_code" in res:
|
|
|
+ print(f"百度OCR接口错误：{res['error_msg']}（错误码：{res['error_code']}）")
|
|
|
+ return None
|
|
|
+ # 解析识别结果
|
|
|
+ new_dic = dict()
|
|
|
+ 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
|
|
|
+
|
|
|
+def get_access_token():
|
|
|
+ AppKey = AppKey_config
|
|
|
+ AppSrcret = AppSecret_config
|
|
|
+ token_url =token_url_config
|
|
|
+ url = f"{token_url}?grant_type=client_credentials&client_id={AppKey}&client_secret={AppSrcret}"
|
|
|
+
|
|
|
+ payload = ""
|
|
|
+ headers = {
|
|
|
+ 'Content-Type': 'application/json',
|
|
|
+ 'Accept': 'application/json'
|
|
|
+ }
|
|
|
+ try:
|
|
|
+ response = requests.request("POST", url, headers=headers, data=payload)
|
|
|
+ response.raise_for_status() # 触发HTTP错误
|
|
|
+ return response.json()['access_token']
|
|
|
+ except Exception as e:
|
|
|
+ print(f"获取access_token失败：{str(e)}")
|
|
|
+ return None
|
|
|
+
|
|
|
+def extract_province_city(address):
|
|
|
+ """
|
|
|
+ 从地址中提取省份和城市
|
|
|
+ :param address: 营业执照地址（如"福建省福州市马尾区"）
|
|
|
+ :return: (province, city) - 提取到的省份/城市，提取失败返回空字符串
|
|
|
+ """
|
|
|
+ if not address: # 地址为空，直接返回空
|
|
|
+ return "", ""
|
|
|
+
|
|
|
+ # 正则1：匹配省份（兼容省/自治区/直辖市/特别行政区）
|
|
|
+ province_pattern = re.compile(r'([^省]+省|.+自治区|北京市|上海市|天津市|重庆市|.+特别行政区)')
|
|
|
+ province_match = province_pattern.search(address)
|
|
|
+ province = province_match.group(1) if province_match else ""
|
|
|
+
|
|
|
+ # 正则2：匹配城市（兼容市/自治州/地区/盟，且排除省份已匹配的部分）
|
|
|
+ # 先去掉已匹配的省份，再匹配城市
|
|
|
+ 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 ""
|
|
|
+
|
|
|
+ # 兼容直辖市（如"北京市朝阳区"→city=北京市）
|
|
|
+ if province in ["北京市", "上海市", "天津市", "重庆市"]:
|
|
|
+ city = province
|
|
|
+
|
|
|
+ # 兼容地址不规范的情况（如"福建福州马尾区"，无"省"/"市"字）
|
|
|
+ if not province and not city:
|
|
|
+ # 匹配前两个地名（如"福建福州"→province=福建，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 in city:
|
|
|
+ city = city.replace(province, "").strip()
|
|
|
+
|
|
|
+ return province.strip(), city.strip()
|
|
|
+
|
|
|
+
|
|
|
+#采集数据核心
|
|
|
+def collect_data(page, keyword, keyword_dict, scheduler=None):
|
|
|
+ """
|
|
|
+ 1) 先获取当前页商品个数（count）
|
|
|
+ 2) 按循环次数采集；每循环15次滚动一次 slow_scroll_1200px
|
|
|
+ 3) 当前页循环完 -> goto_next_page；有下一页继续；无下一页结束该关键词
|
|
|
+ """
|
|
|
+ collect_result = []
|
|
|
+ # seen = set()
|
|
|
+ # ========== 初始化异常关键词存储 ==========
|
|
|
+ error_keywords = []
|
|
|
+ kw = '' # 单个异常关键词变量
|
|
|
+
|
|
|
+ logger.info(f"📊 开始采集「{keyword}」的商品数据")
|
|
|
+ page.wait_for_load_state("networkidle")
|
|
|
+ #没有找到商品就跳过这个商品
|
|
|
+
|
|
|
+
|
|
|
+ page_no = keyword_dict.get("start_page", 1) # 续传起始页
|
|
|
+ while True:
|
|
|
+
|
|
|
+ logger.info(f"\n📄 「{keyword}」开始采集第 {page_no} 页")
|
|
|
+
|
|
|
+ # 记录列表页URL（可用于你后续兜底）
|
|
|
+ list_page_url = page.url
|
|
|
+ logger.info(f"📌 已记录商品列表页URL：{list_page_url}")
|
|
|
+
|
|
|
+ # ✅ 先获取当前页商品个数
|
|
|
+ page.wait_for_load_state("networkidle")
|
|
|
+ total_limit = page.locator(PRODUCT_ITEM_SELECTOR).count()
|
|
|
+ logger.info(f"📌 「{keyword}」第{page_no}页 初始商品个数（count）：{total_limit}")
|
|
|
+
|
|
|
+ #获取该商品的总个数
|
|
|
+ total_goods_nums_elem = page.locator("div.sr-page_turner-pagination-total")
|
|
|
+ if total_goods_nums_elem.count() > 0:
|
|
|
+ total_goods_nums = total_goods_nums_elem.inner_text().strip()
|
|
|
+ logger.info(f"📌 「{keyword} 商品个数（count）：{total_goods_nums}")
|
|
|
+ else:
|
|
|
+ logger.info(f"📌 「{keyword} 商品个数（count）：不超过60个")
|
|
|
+ # 重置当前页的采集计数
|
|
|
+ collected_count = 0
|
|
|
+
|
|
|
+
|
|
|
+ # ========= 初始化无匹配计数器（记录标题不包含核心关键词的次数） =========
|
|
|
+ no_match_count = 0 # 无匹配次数初始化为0
|
|
|
+ MAX_NO_MATCH = 20 # 最大无匹配次数阈值
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+ for idx in range(total_limit):
|
|
|
+ detail_page = None
|
|
|
+ try:
|
|
|
+ item = page.locator(PRODUCT_ITEM_SELECTOR).nth(idx)
|
|
|
+ collected_count += 1 # 实际采集计数（用于日志）
|
|
|
+ # ========= 反爬随机延迟（保留你的原逻辑也行） =========
|
|
|
+ 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}秒后采集（反爬）")
|
|
|
+
|
|
|
+
|
|
|
+ # 1. 初始化所有字段默认值
|
|
|
+ title = "无标题"
|
|
|
+ price = "0.00"
|
|
|
+ shop = "无店名"
|
|
|
+ expiry_date = "无有效期"
|
|
|
+ manufacture_date = "无生产日期"
|
|
|
+ approval_number = "无批准文号"
|
|
|
+ manufacturer = "未知公司"
|
|
|
+ # discount_price = "0.00"
|
|
|
+ spec = "未知规格"
|
|
|
+ num = 1 # ✅ 默认 1
|
|
|
+ platform = '药九九'
|
|
|
+ current_time = datetime.now().strftime("%Y-%m-%d")
|
|
|
+ is_sold_out = 0
|
|
|
+
|
|
|
+
|
|
|
+ # ========= 售罄不跳过 =========
|
|
|
+ sold_locator = item.locator('div.gc-l1-cirle_tip')
|
|
|
+ if sold_locator.count() > 0:
|
|
|
+ is_sold_out = 1
|
|
|
+ logger.warning(f" 「{keyword}」第{page_no}页 第{collected_count}个商品已售罄")
|
|
|
+
|
|
|
+ # 提取商品标题（处理空值）
|
|
|
+ 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}个商品 - 列表页标题元素未找到，使用默认值：{title}")
|
|
|
+
|
|
|
+
|
|
|
+ if no_match_count >= MAX_NO_MATCH:
|
|
|
+ logger.error(f"❌ 关键词「{keyword}」无匹配商品次数已达{MAX_NO_MATCH}次，直接终止当前关键词采集，进入下一个关键词")
|
|
|
+ return []
|
|
|
+
|
|
|
+ # 关键词不在标题中，跳过当前商品
|
|
|
+ product_brand = keyword_dict.get("product_brand")
|
|
|
+ if product_brand not in title:
|
|
|
+ no_match_count += 1
|
|
|
+ logger.warning(f" 「{keyword}」第{collected_count}个商品 - 标题「{title}」不包含品牌「{product_brand}」（无匹配次数：{no_match_count}/{MAX_NO_MATCH}），跳过本次循环")
|
|
|
+ continue
|
|
|
+
|
|
|
+ product_name = keyword_dict.get("product_name")
|
|
|
+ if product_name not in title:
|
|
|
+ no_match_count += 1
|
|
|
+ logger.warning(f" 「{keyword}」第{collected_count}个商品 - 标题「{title}」不包含核心关键词「{product_name}」（无匹配次数：{no_match_count}/{MAX_NO_MATCH}），跳过本次循环")
|
|
|
+ continue
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+ # 提取价格（带缺失日志）
|
|
|
+ price_locator = item.locator(PRODUCT_PRICE_SELECTOR).nth(0)
|
|
|
+ if price_locator.count() > 0:
|
|
|
+ price = price_locator.inner_text(timeout=3000).strip()
|
|
|
+ logger.info(f"{'='*10}「{keyword}」第{collected_count}个商品 - 列表页采购价格：{price}{'='*10}")
|
|
|
+ else:
|
|
|
+ price = "0.00" # 初始化默认值，避免后续报错
|
|
|
+ logger.warning(f" 「{keyword}」第{collected_count}个商品「{title}」- 列表页采购价格元素未找到，使用默认值：{price}")
|
|
|
+
|
|
|
+ # 5. 提取公司名称（带缺失日志）
|
|
|
+ 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}」- 列表页公司名称元素未找到，使用默认值：{manufacturer}")
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+ #提取店铺名称
|
|
|
+ shop_locator = item.locator(PRODUCT_STORE_SELECTOR)
|
|
|
+ if shop_locator.count() > 0:
|
|
|
+ shop = shop_locator.inner_text(timeout=3000).strip()
|
|
|
+ logger.info(f"{'='*10}「{keyword}」第{collected_count}个商品 - 列表页店名：{shop}{'='*10}")
|
|
|
+ else:
|
|
|
+
|
|
|
+ logger.warning(f" 「{keyword}」第{collected_count}个商品「{title}」- 列表页店铺名称元素未找到，使用默认值：{shop}")
|
|
|
+
|
|
|
+ #提取折扣价
|
|
|
+ discount_price_val_origin = ""
|
|
|
+ discount_price = ""
|
|
|
+ discount_price_locator = item.locator('span.gc-l2-discount_price').first
|
|
|
+ if discount_price_locator.count() > 0:
|
|
|
+ discount_price = discount_price_locator.inner_text(timeout=3000).strip()
|
|
|
+ discount_price_val_origin = discount_price
|
|
|
+ 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:
|
|
|
+ #如果没有拿原价替换
|
|
|
+ price = float(price.replace("￥", "").replace(",", "")) if price.replace("￥", "").replace(",", "").replace(".", "") else "0.00"
|
|
|
+ discount_price_val = float(price)
|
|
|
+ logger.warning(f" 「{keyword}」第{collected_count}个商品「{title}」- 折扣价元素未找到，使用采购价兜底：{discount_price_val}")
|
|
|
+
|
|
|
+ merged_price = f"{price}{discount_price_val_origin}" if discount_price_val_origin else price
|
|
|
+
|
|
|
+ # ========= 模拟点击商品进入详情页 =========
|
|
|
+ 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 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 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}秒（反爬）"
|
|
|
+ )
|
|
|
+ # 反爬：检测详情页反爬验证
|
|
|
+ # check_anti_crawl(page)
|
|
|
+ # ========== 采集详情页的专属信息（有效期/生产日期/批准文号） ==========
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+ #获取商品详情页链接
|
|
|
+ product_link = target_page.url
|
|
|
+ logger.info(f"{'='*10}「{keyword}」第{collected_count}个商品 - 详情页链接：{product_link}{'='*10}")
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+ #如果有需要，还可能要加兜底。
|
|
|
+
|
|
|
+ # ========= ✅ 去重逻辑，拿商品链接和折扣价和有效期和采集日期 =========
|
|
|
+ if check_dup_in_biz_db(product_link, discount_price_val, current_time,keyword_dict):
|
|
|
+ logger.warning(f" 「{keyword}」第{page_no}页 第{collected_count}个商品（重复）：{title}，跳过")
|
|
|
+ # ========== 关闭新标签页，切回列表页 ==========
|
|
|
+ if detail_page and not detail_page.is_closed():
|
|
|
+ detail_page.close() # 关闭详情页标签
|
|
|
+ logger.info(f"📌 「{keyword}」第{collected_count}个商品 - 已关闭详情页标签页")
|
|
|
+ # 切回原列表页（第一个标签页）
|
|
|
+ page.bring_to_front() # 激活列表页
|
|
|
+ page.mouse.move(random.randint(100, 300), random.randint(200, 400)) # 随机移动鼠标
|
|
|
+ random_delay(0.5, 1.0) # 增加切换后延迟
|
|
|
+ page.wait_for_load_state("networkidle")
|
|
|
+ random_delay(MIN_CLICK_DELAY, MAX_CLICK_DELAY)
|
|
|
+ logger.info(f" 「{keyword}」第{collected_count}个商品「{title}」- 已切回列表页")
|
|
|
+ if collected_count % 5 == 0 and collected_count > 0:
|
|
|
+ logger.info("采满5个往下滑")
|
|
|
+ slow_scroll_400px(page)
|
|
|
+ page.wait_for_load_state("networkidle")
|
|
|
+ continue
|
|
|
+
|
|
|
+ # 提取有效期（处理空值）
|
|
|
+ expiry_date_locator = target_page.locator("//span[contains(text(), '有效期')]/following-sibling::span[contains(@class, 'gdb-desc-value4')]")
|
|
|
+ if expiry_date_locator.count() > 0:
|
|
|
+ expiry_date = expiry_date_locator.inner_text(timeout=3000).strip().replace('-', '') #.replace('近效期','')
|
|
|
+ logger.info(f"{'='*10}「{keyword}」第{collected_count}个商品 - 详情页有效期：{expiry_date}{'='*10}")
|
|
|
+ else:
|
|
|
+ # 修复：替换未定义的i为collected_count
|
|
|
+ logger.warning(f" 「{keyword}」第{collected_count}个商品「{title}」- 有效期元素未找到，使用默认值：{expiry_date}")
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+ # 提取生产日期（修复完成）
|
|
|
+ manufacture_date_locator = target_page.locator("//span[@class='gdb-desc-label' and text()='生产日期']/following-sibling::span[1]")
|
|
|
+ if manufacture_date_locator.count() > 0:
|
|
|
+ manufacture_date = manufacture_date_locator.inner_text(timeout=3000).strip().replace('-', "")
|
|
|
+ logger.info(f"{'='*10}「{keyword}」第{collected_count}个商品 - 详情页生产日期：{manufacture_date}{'='*10}")
|
|
|
+ else:
|
|
|
+ # 修复：替换未定义的i为collected_count
|
|
|
+ logger.warning(f" 「{keyword}」第{collected_count}个商品「{title}」- 生产日期元素未找到，使用默认值：{manufacture_date}")
|
|
|
+
|
|
|
+
|
|
|
+ # 提取批准文号
|
|
|
+ approval_number_locator = target_page.locator("//span[contains(text(), '国药准字')]").first
|
|
|
+ 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:
|
|
|
+ # 修复：替换未定义的i为collected_count
|
|
|
+ logger.warning(f" 「{keyword}」第{collected_count}个商品「{title}」- 批准文号元素未找到，使用默认值：{approval_number}")
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+ #提取规格
|
|
|
+ spec_locator = target_page.locator('span.gddd-params_text_line_1[title]')
|
|
|
+ if spec_locator.count() > 0:
|
|
|
+ spec = spec_locator.nth(2).inner_text(timeout=3000).strip()
|
|
|
+ logger.info(f"{'='*10}「{keyword}」第{collected_count}个商品 - 详情页规格：{spec}{'='*10}")
|
|
|
+ else:
|
|
|
+ # 修复：替换未定义的i为collected_count，补充规格数量不足的提示
|
|
|
+ logger.warning(f" 「{keyword}」第{collected_count}个商品「{title}」- 规格元素数量不足，使用默认值：{spec}")
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+ #提取库存
|
|
|
+ storage = ''
|
|
|
+ storage_locator = target_page.locator("span.gdb-desc-value7")
|
|
|
+ if storage_locator.count() > 0:
|
|
|
+ storage = storage_locator.inner_text(timeout=3000).strip()
|
|
|
+ logger.info(f"{'='*10}「{keyword}」第{collected_count}个商品 - 详情页库存：{storage}{'='*10}")
|
|
|
+ else:
|
|
|
+ logger.warning(f" 「{keyword}」第{collected_count}个商品「{title}」- 库存元素数量不足，使用默认值：{storage}")
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+ #提取销量
|
|
|
+ sell = ''
|
|
|
+ sell_locator = target_page.locator('.has-join-group span.packUnit-class')
|
|
|
+ if sell_locator.count() > 0:
|
|
|
+ sell = sell_locator.inner_text(timeout=3000).strip()
|
|
|
+ logger.info(f"{'='*10}「{keyword}」第{collected_count}个商品 - 详情页销量：{sell}{'='*10}")
|
|
|
+ else:
|
|
|
+ logger.warning(f" 「{keyword}」第{collected_count}个商品「{title}」- 没有销量元素，使用默认值：{sell}")
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+ 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)
|
|
|
+
|
|
|
+ shop_exists, shop_info = shop_is_exists_database(shop)
|
|
|
+ #店铺名不是药品预约中心且店铺名不在数据库就要点击
|
|
|
+ if shop != "药品预约中心" and not shop_exists :
|
|
|
+ logger.info("店铺名不是药品预约中心且数据库没有该公司的营业执照")
|
|
|
+ # 获取营业执照图片 li[data-v-4f79abe8].nth(2)
|
|
|
+ # 进入店铺
|
|
|
+ random_delay(MIN_CLICK_DELAY, MAX_CLICK_DELAY)
|
|
|
+ entershop_btn = target_page.locator('[data-v-c5790f48].btn-text')
|
|
|
+ # 增强：先等待进入店铺按钮可见
|
|
|
+ entershop_btn.wait_for(state="visible", timeout=10000)
|
|
|
+ entershop_btn.scroll_into_view_if_needed() # 确保按钮在视口内
|
|
|
+ entershop_btn.hover() # 先悬停
|
|
|
+ random_delay(0.2, 0.5) # 悬停后延迟
|
|
|
+ entershop_btn.click()
|
|
|
+ # entershop_btn.dispatch_event("mousedown")
|
|
|
+ random_delay(0.05, 0.15) # 鼠标按下后延迟
|
|
|
+ # entershop_btn.dispatch_event("mouseup")
|
|
|
+ random_delay(0.05, 0.1) # 鼠标松开后延迟
|
|
|
+ target_page.wait_for_load_state("domcontentloaded") # 等DOM加载（比networkidle更适合页面内切换）
|
|
|
+
|
|
|
+
|
|
|
+ #点击店铺资质
|
|
|
+ random_delay(MIN_CLICK_DELAY, MAX_CLICK_DELAY)
|
|
|
+ shop_license_page = target_page.locator('li:has-text("店铺资质")')
|
|
|
+ shop_license_page.wait_for(state="visible", timeout=10000) # 等待元素加载完成
|
|
|
+ shop_license_page.hover() # 先悬停
|
|
|
+ random_delay(0.2, 0.5) # 悬停后延迟
|
|
|
+ # shop_license_page.dispatch_event("mousedown")
|
|
|
+ shop_license_page.click()
|
|
|
+ random_delay(0.05, 0.15) # 鼠标按下后延迟
|
|
|
+ # shop_license_page.dispatch_event("mouseup")
|
|
|
+ random_delay(0.05, 0.1) # 鼠标松开后延迟
|
|
|
+ target_page.wait_for_load_state("networkidle")
|
|
|
+ slow_scroll_400px(target_page, scroll_distance1=700)
|
|
|
+
|
|
|
+
|
|
|
+ #获取营业执照图片
|
|
|
+ target_page.wait_for_load_state("load")
|
|
|
+ ocr_res = None
|
|
|
+ shop_license_div = target_page.locator('div.shop-licensesImg').nth(0)
|
|
|
+ shop_license_div.wait_for(state="visible", timeout=60000)
|
|
|
+ shop_license_img = shop_license_div.locator('img')
|
|
|
+ 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)
|
|
|
+ # print(f'ocr_res:{ocr_res}')
|
|
|
+ else:
|
|
|
+ shop_license_src = None
|
|
|
+ except Exception as e:
|
|
|
+ # 捕获定位/提取失败的异常，避免程序崩溃
|
|
|
+ logger.warning(f"提取营业执照图片src失败：{e}")
|
|
|
+ shop_license_src = None
|
|
|
+ print("营业执照图片链接：", shop_license_src)
|
|
|
+ # input("..")
|
|
|
+
|
|
|
+ contact_address = ''
|
|
|
+ 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 ''
|
|
|
+ # scrape_date = ''
|
|
|
+
|
|
|
+ # 调用提取函数，获取省份和城市
|
|
|
+ province, city = extract_province_city(business_license_address)
|
|
|
+ logger.info(f"原始地址：{business_license_address}")
|
|
|
+ logger.info(f"提取的省份：{province} | 城市：{city}")
|
|
|
+ insert_result = insert_shop_info_to_db(
|
|
|
+ shop=shop,
|
|
|
+ contact_address=contact_address,
|
|
|
+ qualification_number=qualification_number,
|
|
|
+ business_license_company=business_license_company,
|
|
|
+ business_license_address=business_license_address,
|
|
|
+ scrape_date=current_time,
|
|
|
+ platform=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")
|
|
|
+ )
|
|
|
+ elif shop == "药品预约中心":
|
|
|
+ province, city = "未知", "未知"
|
|
|
+ business_license_company, qualification_number = "", ""
|
|
|
+ if shop_info:
|
|
|
+ province = shop_info["province"]
|
|
|
+ city = shop_info["city"]
|
|
|
+ business_license_company = shop_info["business_license_company"]
|
|
|
+ qualification_number = shop_info["qualification_number"]
|
|
|
+ else:
|
|
|
+ logger.info("数据库有该店名，在数据库拿取对应字段填充yjj_drug_middle表")
|
|
|
+ province, city = "", ""
|
|
|
+ business_license_company, qualification_number = "", ""
|
|
|
+ if shop_info:
|
|
|
+ province = shop_info["province"]
|
|
|
+ city = shop_info["city"]
|
|
|
+ business_license_company = shop_info["business_license_company"]
|
|
|
+ qualification_number = shop_info["qualification_number"]
|
|
|
+
|
|
|
+ # ========== 关闭新标签页，切回列表页 ==========
|
|
|
+ if detail_page and not detail_page.is_closed():
|
|
|
+ detail_page.close() # 关闭详情页标签
|
|
|
+ logger.info(f"📌 「{keyword}」第{collected_count}个商品 - 已关闭详情页标签页")
|
|
|
+ # 切回原列表页（第一个标签页）
|
|
|
+ page.bring_to_front() # 激活列表页
|
|
|
+ page.mouse.move(random.randint(100, 300), random.randint(200, 400)) # 随机移动鼠标
|
|
|
+ random_delay(0.5, 1.0) # 增加切换后延迟
|
|
|
+ page.wait_for_load_state("networkidle")
|
|
|
+ random_delay(MIN_CLICK_DELAY, MAX_CLICK_DELAY)
|
|
|
+ logger.info(f" 「{keyword}」第{collected_count}个商品「{title}」- 已切回列表页")
|
|
|
+
|
|
|
+ # credit_code = ""
|
|
|
+ availability = ""
|
|
|
+
|
|
|
+ # 组装单条数据（仅新增生产日期/批准文号字段，原有字段顺序/逻辑不变）
|
|
|
+ # 构造单条数据元组（适配MySQL字段）
|
|
|
+ single_data = {
|
|
|
+ # 核心商品信息
|
|
|
+ "product": title, # 商品名称
|
|
|
+ "my_good_price": merged_price, # 自定义价格（可与min_price相同或单独提取）
|
|
|
+ "min_price": discount_price_val, # 最低价格
|
|
|
+ "manufacture_date": manufacture_date, # 生产日期
|
|
|
+ "expiry_date": expiry_date, # 有效期
|
|
|
+ "shop": shop, # 店铺名
|
|
|
+ "business_license_company": business_license_company, # 营业执照主体（公司名称）
|
|
|
+ "province": province, # 省份
|
|
|
+ "city": city, # 城市
|
|
|
+ "manufacturer": manufacturer, # 生产厂家
|
|
|
+ "specification": spec, # 规格
|
|
|
+ "approval_number": approval_number, # 批准文号
|
|
|
+ "product_link": product_link, # 商品链接
|
|
|
+ "scrape_date": current_time, # 采集日期
|
|
|
+ "scrape_province": "", # 采集省份（可留空或根据IP获取）
|
|
|
+ "availability": availability, # 库存状态
|
|
|
+ "credit_code": qualification_number, # 统一信用代码（如有可补充提取）
|
|
|
+ "platform": platform, # 平台名称（固定或动态获取）
|
|
|
+ "search_key": keyword, # 搜索关键词
|
|
|
+ "number": num, # 数量（盒数）
|
|
|
+ "is_sold_out": is_sold_out, # 售罄标记（0/1）
|
|
|
+ "sales": sell, #销量
|
|
|
+ "inventory": storage, #库存
|
|
|
+ "snapshot_url": oss_url, #快照链接
|
|
|
+ "update_time": time.strftime("%Y-%m-%d %H:%M:%S"), # 更新时间
|
|
|
+ "insert_time": time.strftime("%Y-%m-%d %H:%M:%S"), # 创建时间
|
|
|
+ "task_id": keyword_dict["id"], # 任务id
|
|
|
+ "company_id": keyword_dict["company_id"], # 平台id
|
|
|
+ "product_brand": keyword_dict["product_brand"], # 创建时间
|
|
|
+ "search_name": keyword_dict["product_name"], # 创建时间
|
|
|
+ "collect_config_info": json.dumps(
|
|
|
+ {"sampling_cycle": keyword_dict["sampling_cycle"], "sampling_start_time":keyword_dict["sampling_start_time"],
|
|
|
+ "sampling_end_time":keyword_dict["sampling_end_time"]}),
|
|
|
+ "collect_equipment_account_id": keyword_dict["collect_equipment_account_id"],
|
|
|
+ "collect_region_id": keyword_dict["collect_region_id"],
|
|
|
+ "collect_round": keyword_dict["collect_round"],
|
|
|
+
|
|
|
+ }
|
|
|
+ # 调用逐条插入函数
|
|
|
+ insert_single_to_mysql(single_data)
|
|
|
+ collect_result.append(single_data)
|
|
|
+ logger.info(f" 「{keyword}」第{collected_count}个商品「{title}」采集完成")
|
|
|
+
|
|
|
+ except Exception as e:
|
|
|
+ kw = keyword
|
|
|
+ if kw not in error_keywords:
|
|
|
+ error_keywords.append(kw)
|
|
|
+ _report_exception(scheduler, keyword_dict, page_no, 5, f"商品采集异常: {e}", crawled_count=len(collect_result))
|
|
|
+ logger.exception(f" 「{keyword}」第{collected_count}个商品采集核心异常：{str(e)}")
|
|
|
+
|
|
|
+ try:
|
|
|
+ if detail_page and not detail_page.is_closed():
|
|
|
+ detail_page.close()
|
|
|
+ logger.info(f"📌 「{keyword}」第{collected_count}个商品 - 异常时关闭详情页标签页")
|
|
|
+ if page and not page.is_closed():
|
|
|
+ page.bring_to_front() # 切回列表页
|
|
|
+ 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
|
|
|
+
|
|
|
+
|
|
|
+ # ✅ 每15次滚动一次（修复：用collected_count，且排除0的情况）
|
|
|
+ if collected_count % 5 == 0 and collected_count > 0 and collected_count != total_limit:
|
|
|
+ logger.info("采满5个往下滑")
|
|
|
+ slow_scroll_400px(page,)
|
|
|
+ page.wait_for_load_state("networkidle")
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+ # ====== 当前页采集完毕，逐页回告 ======
|
|
|
+ if _report_page_progress(scheduler, keyword_dict, page_no, len(collect_result)) == True:
|
|
|
+ logger.error("逐页回告返回error，停止采集")
|
|
|
+ break
|
|
|
+
|
|
|
+ delay = random_delay(1.5, 3.0)
|
|
|
+ logger.info(f"⏳ 翻页前随机等待 {delay:.2f}s（反爬）")
|
|
|
+
|
|
|
+ if goto_next_page(page):
|
|
|
+ page_no += 1
|
|
|
+ continue
|
|
|
+ else:
|
|
|
+ logger.info(f" 「{keyword}」已无下一页，关键词采集结束")
|
|
|
+ break
|
|
|
+ # 关键词采集完成回告
|
|
|
+ _report_page_progress(scheduler, keyword_dict, page_no, len(collect_result),
|
|
|
+ is_finished=1, total_pages=page_no)
|
|
|
+ long_delay = random_delay(MIN_KEYWORD_DELAY, MAX_KEYWORD_DELAY)
|
|
|
+ logger.info(f" 「{keyword}」采集完成，共{len(collect_result)}条数据，等待{long_delay:.2f}秒后继续下一个关键词（反爬）")
|
|
|
+ result_data = {
|
|
|
+ "real_count": len(collect_result),
|
|
|
+ "end_page": page_no,
|
|
|
+ "start_page": 1,
|
|
|
+ }
|
|
|
+ return result_data
|
|
|
+
|
|
|
+
|
|
|
+# ==================== 主函数（登录+调度器拉任务） ====================
|
|
|
+def main():
|
|
|
+ logger.info("\n" + "="*50)
|
|
|
+ logger.info("🚀 药九九采集程序启动")
|
|
|
+ logger.info(f"⏰ 启动时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
|
|
+ logger.info("="*50)
|
|
|
+
|
|
|
+ # 从数据库获取可用账号
|
|
|
+ account = get_account_from_db()
|
|
|
+ if not account:
|
|
|
+ logger.error("无可用药九九账号，退出本轮")
|
|
|
+ return
|
|
|
+ device_id = account["phone"] # phone 作为 DEVICE_ID，回告接口的 username 也是这个
|
|
|
+ username = account["phone"]
|
|
|
+ password = account["password"]
|
|
|
+ logger.info("获取到账号: %s, phone=%s", account["name"], device_id)
|
|
|
+
|
|
|
+ # 创建调度器，启动心跳（先上报心跳再拉任务）
|
|
|
+ scheduler = CrawlerScheduler(DEVICE_ID=username, platform=PLATFORM_ID)
|
|
|
+ scheduler.start()
|
|
|
+ time.sleep(2)
|
|
|
+ logger.info("调度器已启动，心跳已就绪")
|
|
|
+
|
|
|
+ with sync_playwright() as p:
|
|
|
+ browser = p.chromium.launch(
|
|
|
+ headless=True,
|
|
|
+ channel="chrome",
|
|
|
+ slow_mo=random.randint(100, 300),
|
|
|
+ args=_chromium_launch_args(),
|
|
|
+ )
|
|
|
+ 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": 1600, "height": 1400},
|
|
|
+ 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);
|
|
|
+ };
|
|
|
+ })();
|
|
|
+""")
|
|
|
+
|
|
|
+ keyword_dict = None
|
|
|
+ try:
|
|
|
+ # ========== Cookie复用 + 登录 ==========
|
|
|
+ load_cookies(context)
|
|
|
+ if not is_login(page):
|
|
|
+ page.goto(TARGET_LOGIN_URL)
|
|
|
+ time.sleep(5)
|
|
|
+ page.wait_for_load_state("networkidle")
|
|
|
+ logger.info("🔑 开始执行登录流程，账号=%s", username)
|
|
|
+ login_success = login_operation(page, username, password)
|
|
|
+ if not login_success:
|
|
|
+ logger.error("登录失败，程序终止")
|
|
|
+ _report_exception(scheduler, {}, 0, 1, "登录失败")
|
|
|
+ return
|
|
|
+ save_cookies(context)
|
|
|
+ logger.info("登录并保存Cookie成功！")
|
|
|
+
|
|
|
+ # ========== 循环拉取任务 ==========
|
|
|
+ while True:
|
|
|
+ task = scheduler.get_task()
|
|
|
+ if not task:
|
|
|
+ logger.info("暂无任务，本轮结束")
|
|
|
+ break
|
|
|
+
|
|
|
+ # 构建 keyword_dict（兼容原有字段）
|
|
|
+ current_page = task.get("current_page", 0)
|
|
|
+ start_page = current_page if current_page > 0 else 1
|
|
|
+ keyword_dict = {
|
|
|
+ "id": task.get("id"),
|
|
|
+ "company_id": task.get("company_id"),
|
|
|
+ "product_name": task.get("product_name", ""),
|
|
|
+ "product_specs": task.get("product_specs", ""),
|
|
|
+ "product_brand": task.get("product_brand", ""),
|
|
|
+ "product_keyword": task.get("product_keyword", ""),
|
|
|
+ "current_page": current_page,
|
|
|
+ "start_offset": task.get("start_offset", 0),
|
|
|
+ "start_page": start_page,
|
|
|
+ "end_page": 0,
|
|
|
+ "_platform": PLATFORM_ID,
|
|
|
+ "_device_id": device_id,
|
|
|
+ "collect_task_id": str(task.get("id", "")),
|
|
|
+ "snapshot_collect_status": 1,
|
|
|
+ "sampling_cycle": "",
|
|
|
+ "sampling_start_time": "",
|
|
|
+ "sampling_end_time": "",
|
|
|
+ "collect_equipment_account_id": 0,
|
|
|
+ "collect_region_id": 0,
|
|
|
+ "collect_equipment_id": 0,
|
|
|
+ "collect_round": 0,
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ keyword = f"{keyword_dict['product_brand']} {keyword_dict['product_name']} {keyword_dict['product_specs']}".strip()
|
|
|
+ logger.info(f"\n{'='*40}\n🔍 开始处理：{keyword}\n{'='*40}")
|
|
|
+
|
|
|
+ # 执行搜索
|
|
|
+ popup_guard(page, "before_search")
|
|
|
+ search_success = search_operation(page, keyword)
|
|
|
+ popup_guard(page, "after_search")
|
|
|
+
|
|
|
+ if not search_success:
|
|
|
+ logger.warning(f"「{keyword}」搜索失败，跳过")
|
|
|
+ _report_exception(scheduler, keyword_dict, 0, 4, "搜索失败")
|
|
|
+ continue
|
|
|
+
|
|
|
+ page.wait_for_load_state("domcontentloaded")
|
|
|
+ page.wait_for_load_state('networkidle')
|
|
|
+
|
|
|
+ # 采集数据（内部含逐页回告）
|
|
|
+ collect_data(page, keyword, keyword_dict, scheduler=scheduler)
|
|
|
+
|
|
|
+ finished_task_ids.add(keyword_dict["id"])
|
|
|
+ logger.info(f"任务 {keyword_dict['id']} 完成")
|
|
|
+
|
|
|
+ # 任务间隔
|
|
|
+ time.sleep(random.randint(180, 300))
|
|
|
+
|
|
|
+ except Exception as e:
|
|
|
+ if keyword_dict is not None:
|
|
|
+ _report_exception(scheduler, keyword_dict, 0, 5, f"程序异常: {e}")
|
|
|
+ logger.exception("程序异常：%s", e)
|
|
|
+ finally:
|
|
|
+ browser.close()
|
|
|
+ logger.info("浏览器已关闭，程序结束")
|
|
|
+
|
|
|
+# ==================== 程序入口 ====================
|
|
|
+if __name__ == '__main__':
|
|
|
+ while True:
|
|
|
+ try:
|
|
|
+ print(f"【任务开始】时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
|
|
+ main()
|
|
|
+ print(f"【任务结束】时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
|
|
+ except Exception as e:
|
|
|
+ print(f"【异常】: {e}")
|
|
|
+ time.sleep(random.randint(180, 300))
|
|
|
+
|
