import sys import sys import logging import threading from concurrent.futures import ThreadPoolExecutor import pymysql import requests import base64 import cv2 import uiautomator2 as u2 import time import subprocess import re import random import datetime import json from aip import AipOcr import threading from collections import deque import numpy as np import secrets import os import oss2 import urllib.parse from config import Config from logger import setup_logger import logging from PIL import Image from pathlib import Path from PIL import Image, ImageDraw, ImageFont import http.client _DEFAULT_PATH = Path(__file__).with_name("city.json") import city_name_to_id setup_logger("mt_spider") # 初始化日志 def get_access_token(): AppKey = "tRK2RhyItCSh6BzyT4CNVXQa" AppSrcret = "TDgKiPo94i2mOM1sDqOuDnlcK1bG66jh" token_url = 'https://aip.baidubce.com/oauth/2.0/token' url = f"{token_url}?grant_type=client_credentials&client_id={AppKey}&client_secret={AppSrcret}" payload = "" headers = { 'Content-Type': 'application/json', 'Accept': 'application/json' } response = requests.request("POST", url, headers=headers, data=payload) try: return response.json()['access_token'] except: return None def get_mysql(): # return pymysql.connect( # host='39.108.116.125', # "localhost", # 修改后的主机 # port=3306, # 3306, # 添加端口号 # user='drug_retrieve', # 'root', # 修改后的用户名 # password='Pem287cwM58jNpe2', # 修改后的密码 # db='drug_retrieve', # # charset='utf8mb4' # ) return pymysql.connect( host='120.24.49.2', # 修改后的主机 port=3306, # 添加端口号 user='drug_retrieve', # 修改后的用户名 password='ksCt3xm6chzdkafj', # 修改后的密码 db='drug_retrieve', # 修改后的数据库名 charset='utf8mb4' ) SCHEDULER_INTERVAL_SECONDS = 600 PLATFORM_MT = 4 TASK_STATUS_PENDING = 1 DEVICE_STATUS_IDLE = 0 dispatch_lock = threading.Lock() running_task_ids = set() running_device_ids = set() worker_threads = {} scheduler_stop_event = threading.Event() scheduler_timer = None def parse_optional_int(value, default=None): if value in (None, ""): return default try: return int(value) except (TypeError, ValueError): return default def parse_spec_list(value): if value is None: return [] if isinstance(value, (list, tuple)): return [str(item).strip() for item in value if str(item).strip()] text = str(value).strip() if not text: return [] parts = re.split(r"[，,/\s]+", text) return [part.strip() for part in parts if part.strip()] def fetch_pending_tasks(): conn = None try: conn = get_mysql() with conn.cursor() as cursor: sql = """ SELECT * FROM retrieve_collect_task_allocate WHERE platform = %s AND status = %s ORDER BY id ASC """ cursor.execute(sql, (PLATFORM_MT, TASK_STATUS_PENDING)) return cursor.fetchall() except Exception as e: logging.exception(f"读取美团待执行任务失败: {e}") return [] finally: if conn: conn.close() def fetch_idle_device_by_equipment_id(equipment_id): conn = None try: conn = get_mysql() with conn.cursor() as cursor: sql = """ SELECT * FROM retrieve_collect_equipment WHERE name LIKE %s AND id = %s AND status = %s LIMIT 1 """ cursor.execute(sql, ('%mt%', equipment_id, DEVICE_STATUS_IDLE)) return cursor.fetchone() except Exception as e: logging.exception(f"读取美团空闲设备失败 equipment_id={equipment_id}: {e}") return None finally: if conn: conn.close() def build_mt_task_payload(task_row, device_row): title_key = (task_row[5] if len(task_row) > 5 and task_row[5] is not None else "").strip() spec_list = parse_spec_list(task_row[6] if len(task_row) > 6 else None) brand = (task_row[7] if len(task_row) > 7 and task_row[7] is not None else "").strip() search_key = f"{brand}{title_key}".strip() or title_key start_page = parse_optional_int(task_row[9] if len(task_row) > 9 else None, None) end_page = parse_optional_int(task_row[10] if len(task_row) > 10 else None, None) page_range = [] if start_page and end_page: page_range = [start_page, end_page] return { "task_id": task_row[0], "equipment_id": task_row[2], "enterprise_id": task_row[3] if len(task_row) > 3 else None, "platform": task_row[4], "title_key": title_key, "spec_list": spec_list, "brand": brand, "search_key": search_key, "sort": "升序", "collect_range": [], "page_range": page_range, "start_page": start_page, "end_page": end_page, "device_id": device_row[2], "workflow_retry_limit": { "start_app": 3, "open_product_list_page": 3, "collect_single_product": 3, }, "workflow_error_action": { "start_app": "start_app", "open_product_list_page": "start_app", "collect_single_product": "back_to_list_page", }, "task_row": task_row, } def fetch_runnable_mt_task_payloads(): tasks = fetch_pending_tasks() if not tasks: logging.info("当前没有美团待执行任务") return [] payloads = [] reserved_equipment_ids = set() for task_row in tasks: task_id = task_row[0] equipment_id = task_row[2] with dispatch_lock: if task_id in running_task_ids: continue if equipment_id in reserved_equipment_ids: continue device_row = fetch_idle_device_by_equipment_id(equipment_id) if not device_row: logging.info(f"美团任务 {task_id} 对应设备 {equipment_id} 当前不空闲，跳过本轮") continue device_id = device_row[2] with dispatch_lock: if device_id in running_device_ids: logging.info(f"美团设备 {device_id} 已在本进程执行任务，跳过任务 {task_id}") continue running_task_ids.add(task_id) running_device_ids.add(device_id) reserved_equipment_ids.add(equipment_id) payloads.append(build_mt_task_payload(task_row, device_row)) return payloads def cleanup_finished_workers(): dead_threads = [] with dispatch_lock: for device_id, thread in worker_threads.items(): if not thread.is_alive(): dead_threads.append(device_id) for device_id in dead_threads: worker_threads.pop(device_id, None) def run_mt_task_worker(task_payload): task_id = task_payload["task_id"] device_id = task_payload["device_id"] mt = None try: logging.info(f"[美团任务 {task_id}] 开始执行，设备: {device_id}") mt = MT( task_payload["search_key"], task_payload["title_key"], task_payload["spec_list"], task_payload["brand"], task_payload.get("sort"), task_payload.get("collect_range"), task_payload.get("page_range"), task_payload.get("workflow_retry_limit"), task_payload.get("workflow_error_action"), ) mt.main(device_id) logging.info(f"[美团任务 {task_id}] 执行完成，设备: {device_id}") except Exception as e: logging.exception(f"[美团任务 {task_id}] 执行异常，设备: {device_id}，错误: {e}") finally: if mt and hasattr(mt, 'close'): mt.close() with dispatch_lock: running_task_ids.discard(task_id) running_device_ids.discard(device_id) worker_threads.pop(device_id, None) def dispatch_pending_tasks(): cleanup_finished_workers() task_payloads = fetch_runnable_mt_task_payloads() if not task_payloads: return for task_payload in task_payloads: device_id = task_payload["device_id"] try: thread = threading.Thread( target=run_mt_task_worker, args=(task_payload,), daemon=True, name=f"mt-{device_id}", ) with dispatch_lock: worker_threads[device_id] = thread thread.start() logging.info(f"[美团任务 {task_payload['task_id']}] 已分发到设备 {device_id}") except Exception: with dispatch_lock: running_task_ids.discard(task_payload["task_id"]) running_device_ids.discard(device_id) worker_threads.pop(device_id, None) raise def schedule_dispatch(delay_seconds=SCHEDULER_INTERVAL_SECONDS): global scheduler_timer if scheduler_stop_event.is_set(): return scheduler_timer = threading.Timer(delay_seconds, scheduled_dispatch_job) scheduler_timer.daemon = False scheduler_timer.name = "mt-scheduler" scheduler_timer.start() def scheduled_dispatch_job(): try: dispatch_pending_tasks() except Exception as e: logging.exception(f"美团定时调度异常: {e}") finally: schedule_dispatch(SCHEDULER_INTERVAL_SECONDS) class SpiderMonitor(threading.Thread): """全局弹窗监控线程（增强版）""" def __init__(self, spider_instance): super().__init__(daemon=True) self.spider = spider_instance self.running = True self.pausing = threading.Event() # 主线程同步事件 self.last_verification_time = 0 self.verification_count = 0 self.MAX_VERIFICATION_RETRY = 10 self.recent_clicks = deque(maxlen=10) # 防重复点击 self.logger = logging.getLogger("SpiderMonitor") # 可配置化弹窗规则 self.popup_rules = { "simple": [ ('//*[@text="确定"]', "点击确定"), ('//*[@text="允许"]', "点击允许"), ('//*[@text="关闭"]', "点击关闭"), ('//*[@resource-id="com.sankuai.meituan:id/close"]', "关闭按钮"), ('//*[@resource-id="com.sankuai.meituan:id/address_center_location_close"]', "关闭按钮"), ('//*[@resource-id="com.sankuai.meituan:id/location_close"]', "关闭按钮"), ('//*[@resource-id="com.sankuai.meituan:id/btn_close"]', "关闭按钮"), ], "verification": [ '//*[contains(@text, "验证")]', '//*[contains(@text, "滑块")]', '//*[contains(@text, "依次点击")]', '//*[contains(@text, "请点击")]', '//*[contains(@text, "拖动滑块刚")]', # 这个需要拖动滑块至最右边，然后再截图 '//*[contains(@text, "请输入图片中的内容")]', '//*[contains(@text, "用最短线连接")]', '//*[contains(@text, "请按语序依次点击")]', '//*[contains(@text, "请向右滑动滑块")]', '//*[contains(@text, "请拖动下方滑块完成拼图")]', '//*[contains(@resource-id, "captcha")]' ] } def run(self): while self.running: try: handled = self.check_and_handle_popup() time.sleep(2 if handled else 1) except Exception as e: self.logger.exception("监控线程异常: %s", e) time.sleep(1) def _is_recent_click(self, xpath): """防止重复点击同一个弹窗""" key = f"{xpath}_{int(time.time())}" if key in self.recent_clicks: return True self.recent_clicks.append(key) return False def check_and_handle_popup(self): d = self.spider.d # 1. 处理简单弹窗 for xpath, desc in self.popup_rules["simple"]: if d.xpath(xpath).exists and not self._is_recent_click(xpath): self.logger.info("检测到弹窗: %s", desc) d.xpath(xpath).click() return True # 2. 处理验证码弹窗 for xpath in self.popup_rules["verification"]: if d.xpath(xpath).exists: now = time.time() if now - self.last_verification_time < 30: return False # 30秒内不重复触发 self.last_verification_time = now self.verification_count += 1 self.logger.warning("验证码弹窗触发，等待人工处理...") if self.verification_count > self.MAX_VERIFICATION_RETRY: self.logger.error("验证码重试超限，终止任务") self.spider.stop_all() return True self.pausing.set() # 通知主线程暂停 # d.toast.show("需要人工处理验证码", 120) while True: if not d.xpath(xpath).exists: self.logger.info("验证码已处理") # d.toast.show("验证完成", 2) self.pausing.clear() # 放行主线程 return True time.sleep(2) self.logger.warning("验证码超时，重启APP") self.spider.restart_app() return True # 3. 处理广告弹窗（点击右上角） if d.xpath('//*[contains(@text, "广告")]').exists: w, h = d.info['displayWidth'], d.info['displayHeight'] d.click(w - 50, 50) self.logger.info("关闭广告弹窗") return True return False def stop(self): self.running = False class MTScreenshot: def __init__(self, d, oss_config, search_key, title_key, scroll_times=4, compress_quality=7, resize_ratio=0.8): # 接收外部已连接好的u2设备实例 self.d = d self.search_key = search_key # 添加这行 self.title_key = title_key # 启动全局弹窗监控 self.monitor = SpiderMonitor(self) self.monitor.start() self.loggerMT = logging.getLogger() # 日志初始化 self.logger = self._init_logger() # OSS配置与初始化（核心配置，无冗余） self.oss_config = oss_config self.oss_bucket = self._init_oss_bucket() # 截图核心参数 self.scroll_times = scroll_times self.compress_quality = compress_quality self.resize_ratio = resize_ratio def _init_logger(self): # 极简日志配置，仅保留必要输出 logger = logging.getLogger("mt_screenshot") logger.setLevel(logging.INFO) logger.handlers.clear() handler = logging.StreamHandler() handler.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")) logger.addHandler(handler) return logger def _init_oss_bucket(self): # 仅做OSS配置校验和Bucket连接，无额外功能 if not all([self.oss_config.get("access_key_id"), self.oss_config.get("access_key_secret"), self.oss_config.get("endpoint"), self.oss_config.get("bucket_name")]): self.logger.warning("OSS配置不完整，无法上传") return None try: auth = oss2.Auth(self.oss_config["access_key_id"], self.oss_config["access_key_secret"]) bucket = oss2.Bucket(auth, self.oss_config["endpoint"], self.oss_config["bucket_name"]) bucket.get_bucket_info() # 验证连接 self.logger.info("OSS Bucket连接成功") return bucket except Exception as e: self.logger.error(f"OSS Bucket连接失败: {e}") return None def _upload_to_oss(self, local_path): # 极简上传逻辑，仅返回OSS URL或None if not self.oss_bucket or not os.path.exists(local_path): return None file_name = os.path.basename(local_path) safe_name = re.sub(r'[^\w\.\-]', '_', file_name) oss_key = f"{self.oss_config.get('oss_prefix', 'scrape_data/')}{safe_name}" try: oss2.resumable_upload(self.oss_bucket, oss_key, local_path) # 生成并返回完整OSS URL oss_file_url = f"https://{self.oss_config['bucket_name']}.{self.oss_config['endpoint']}/{urllib.parse.quote(oss_key, safe='/')}" self.logger.info(f"OSS上传成功: {oss_file_url}") return oss_file_url except Exception as e: self.logger.error(f"OSS上传失败: {e}") return None def safe_exec(self, func, *args, **kwargs): """ 万能安全壳：执行 func 前检查验证码， 若监控线程已置位 pausing，则一直阻塞直到放行。 """ while self.monitor.pausing.is_set(): time.sleep(1) # 执行真正逻辑 return func(*args, **kwargs) def _get_title(self): # try: def _inner(): print(f'获取商品title时的搜索关键字:{self.title_key}') # 初始化 drugs_name = '' specifications = '' title = '' # 循环的获取title为了有时间来处理人机验证 for m in range(1, 6000): if self.d.xpath(f'//*[contains(@text, "{self.title_key}")]').exists: title = self.safe_exec( lambda: self.d.xpath(f'//*[contains(@text, "{self.title_key}")]').text ) self.loggerMT.info(f"第{m}次获取title成功") print(f"第{m}次获取title成功") break else: time.sleep(1) # return drugs_name, specifications title = title[1:] if title.startswith('0') else title print(f'获取到药品标题:{title}') match = re.match(r'(\[[^\]]+\])(.*?)\s*((?:\d+\S*|\(.+))$', title) if match: drugs_name = title specifications = match.group(3).strip() print("药品名:", drugs_name) print("规格:", specifications) # print('完整药名:', drugs_name + specifications) return drugs_name # , specifications else: drugs_name = title specifications = '' return drugs_name # 用 safe_exec 包装内部逻辑，确保验证码阻塞 return self.safe_exec(_inner) def _merge_screenshots(self, screens): # 仅拼接截图，无额外功能 if len(screens) == 1: return screens[0].convert('RGB') rgb_screens = [s.convert('RGB') for s in screens] total_width = rgb_screens[0].width total_height = sum(s.height for s in rgb_screens) merged_img = Image.new('RGB', (total_width, total_height)) y_offset = 0 for img in rgb_screens: merged_img.paste(img, (0, y_offset)) y_offset += img.height return merged_img def get_oss_url(self): """核心方法：截图+临时本地保存+上传OSS+上传成功删本地文件+返回OSS URL，可直接赋值给oss_file""" local_file_path = None try: # 1. 提取标题 title = self._get_title() self.logger.info(f"获取标题: {title[:20]}..." if title else "未获取到标题") # 2. 生成本地文件路径 timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") safe_title = re.sub(r'[\\/*?:"<>|]', '_', title) local_dir = "../scrape_data" os.makedirs(local_dir, exist_ok=True) local_file_path = os.path.join(local_dir, f"{timestamp}_{safe_title}.jpg") # 3. 滚动截图 screen_list = [self.d.screenshot()] w, h = self.d.window_size() for i in range(self.scroll_times): # 可能滑动距离太短，截不到店名。原本是0.8 # self.d.swipe(w // 2, h * 0.9, w // 2, h * 0.1, duration=random.uniform(0.6, 1.2)) self.d.swipe(w // 2, h * 0.85, w // 2, h * 0.15, # 滑动到15% duration=random.uniform(0.8, 1.5)) time.sleep(random.uniform(2.0, 4.0)) screen_list.append(self.d.screenshot()) if self.d(textContains='商家服务').exists: # 看情况是否需要补滑 break # 4. 拼接+压缩+保存 merged_img = self._merge_screenshots(screen_list) if 0.1 < self.resize_ratio < 1.0: new_size = (int(merged_img.width * self.resize_ratio), int(merged_img.height * self.resize_ratio)) resample_mode = Image.Resampling.LANCZOS if hasattr(Image, 'Resampling') else Image.LANCZOS merged_img = merged_img.resize(new_size, resample_mode) # 临时保存到本地 merged_img.save(local_file_path, format='JPEG', quality=self.compress_quality) merged_img.close() # 释放长图句柄 self.logger.info(f"临时本地保存: {local_file_path}") # 5. 上传OSS oss_url = self._upload_to_oss(local_file_path) # 6. 核心：OSS上传成功后，删除本地临时文件 if oss_url is not None: try: self.logger.info(f"✅ OSS上传成功，已删除本地临时文件: {local_file_path}") except Exception as e: self.logger.warning(f"⚠️ OSS上传成功，但删除本地文件失败: {e}") return oss_url except Exception as e: self.logger.error(f"截图/上传失败: {e}") return None class MT: def __init__( self, key, title_key, spec_list, brand, sort=None, collect_range=None, page_range=None, workflow_retry_limit=None, workflow_error_action=None, ): self.package_name = Config.PACKAGE_NAME self.access_token = get_access_token() self.city2province = self.get_city_info() self.APP_ID = '116857964' self.API_KEY = '1gAzACJOAr7BeILKqkqPOETh' self.SECRET_KEY = 'ZNArANb9GwJYgLKg4EfYhukKBfPdl1n3' self.client = AipOcr(self.APP_ID, self.API_KEY, self.SECRET_KEY) self.city_to_name = city_name_to_id.build_city_name_to_id(_DEFAULT_PATH) self.table_name = "retrieve_scrape_data" self.shop_table_name = "mt_shop_info_middle" self.loggerMT = logging.getLogger() self.page = 0 self.search_key = key self.title_key = title_key self.spec_list = spec_list self.brand = brand self.sort = sort self.collect_range = self.normalize_collect_range(collect_range) self.page_range = self.normalize_page_range(page_range) self.sort_key = 0 self.unrelated_data = 0 self.shop_data_num = 0 self.max_unrelated_data = 15 self.collection_cursor = {"page_no": 1, "item_index": 0} self.workflow_retry_limit = workflow_retry_limit or { "start_app": 3, "open_product_list_page": 3, "collect_single_product": 3, } self.workflow_error_action = workflow_error_action or { "start_app": "start_app", "open_product_list_page": "start_app", "collect_single_product": "back_to_list_page", } @staticmethod def normalize_collect_range(collect_range): if not collect_range: return None start = None end = None if isinstance(collect_range, dict): start = collect_range.get("start") end = collect_range.get("end") elif isinstance(collect_range, (list, tuple)) and len(collect_range) >= 2: start, end = collect_range[0], collect_range[1] elif isinstance(collect_range, str): matched = re.match(r"^\s*(\d+(?:\.\d+)?)\s*[-,~]\s*(\d+(?:\.\d+)?)\s*$", collect_range) if matched: start, end = matched.group(1), matched.group(2) try: start = float(start) end = float(end) except (TypeError, ValueError): return None if start < 0 or end < 0: return None if start > end: start, end = end, start return {"start": start, "end": end} @staticmethod def normalize_page_range(page_range): if not page_range: return None start = None end = None if isinstance(page_range, dict): start = page_range.get("start") end = page_range.get("end") elif isinstance(page_range, (list, tuple)) and len(page_range) >= 2: start, end = page_range[0], page_range[1] elif isinstance(page_range, str): matched = re.match(r"^\s*[\[\(]?\s*(\d+)\s*[,，\-~]\s*(\d+)\s*[\]\)]?\s*$", page_range) if matched: start, end = matched.group(1), matched.group(2) try: start = int(float(start)) end = int(float(end)) except (TypeError, ValueError): return None if start <= 0 or end <= 0: return None if start > end: start, end = end, start return {"start": start, "end": end} def stop_app(self): self.d.app_stop(self.package_name) time.sleep(1) def start_app(self): self.d.app_start(self.package_name) time.sleep(1) def restart_app(self): self.stop_app() self.start_app() def li_or_lo(self, key): if key == "升序": self.sort_key += 1 self.d.xpath('//*[@text="价格"]').click() n = self.d.xpath('//*[@text="总价低到高"]') if n.exists: n.click() time.sleep(1) if key == "降序": self.sort_key += 1 self.d.xpath('//*[@text="价格"]').click() time.sleep(2) self.d.xpath('//*[@text="价格"]').click() def wr_re(self, mod, device_id, sort=None, page=None): file_path = f'./ycwj/{device_id}_{self.title_key}.txt' if mod == "写": try: data = { "page": page if page else "", "sort": sort if sort else "", } os.makedirs(os.path.dirname(file_path), exist_ok=True) with open(file_path, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) print(f"进度保存成功:{sort},{page}页") except Exception as e: print("保存进度失败") elif mod == "读": try: if not os.path.exists(file_path): return None with open(file_path, 'r', encoding='utf-8') as f: data = json.load(f) print(self.sort) if self.sort and self.sort_key == 0: self.li_or_lo(self.sort) i = 0 while True: if i == data['page']: self.page = data['page'] print("当前页", self.page) break else: i += 1 self.d.drag(300, 1400, 300, 400, 1) return data except Exception as e: print(f"读取进度失败") return None return None # 任何一个spec满足都算有效 def is_link_spec_useful(self, product_title): if len(self.spec_list) == 0: return True for spec in self.spec_list: if spec in product_title: return True return False # TODO 继续优化这里的判断逻辑，可以考虑搭配config的修改 def is_link_useful(self, product_title): if self.title_key != "" and self.title_key not in product_title: print(f"当前商品名称:{product_title} 不包含{self.title_key}关键字") return False if self.brand != "" and self.brand not in product_title: print(f"当前商品名称:{product_title} 不包含{self.brand}品牌") return False if not self.is_link_spec_useful(product_title): print(f"当前商品名称:{product_title} 不包含{self.spec_list}品规") return False return True @staticmethod def get_sleep_time(): # return random.randint(5, 8) # return random.randint(1, 2) return 1 @staticmethod def get_current_date(): return datetime.datetime.now().strftime('%Y/%m/%d') @staticmethod def get_city_info(): """ 获取所有的省市数据 :return: """ file_path = '../kailin_city.json' with open(file_path, 'r', encoding='utf-8') as f: data = json.load(f) province = {province_one["id"]: province_one for province_one in data['province']} city2province = dict() city = data['city'] for city_one in city: name = city_one['name'] pid = city_one['pid'] if len(str(pid)) > 2: pid = int(re.match('^\d{2}', str(pid)).group()) city2province[name] = province[pid]['name'] return city2province def get_shop_name_from_current_page(self): """ 仅从当前商品详情页读取店铺名，不做任何页面跳转。 """ shop_name = self.get_first_text_by_xpaths([ '//android.widget.ScrollView/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[last()]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[3]/android.widget.FrameLayout[1]/android.widget.TextView', '//android.widget.ScrollView/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[last()-1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[3]/android.widget.FrameLayout[1]/android.widget.TextView', ]) if shop_name: print(f'获取到店铺名:{shop_name}') return shop_name def get_shop_name(self): """ 获取店铺名 :return: """ shop_name = self.get_shop_name_from_current_page() if shop_name: return shop_name try: # 点击店铺进入后获取店铺名称 print("点击店铺进入后获取店铺名称") self.enter_shop() shop_xpath = '//*[@resource-id="com.sankuai.meituan:id/layout_header_view"]/android.widget.FrameLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[1]//android.widget.FrameLayout[2]/android.widget.FrameLayout[1]/android.widget.TextView' if self.d.xpath(shop_xpath).exists: shop_name = self.d.xpath(shop_xpath).text self.swipe_back(1) return shop_name shop_name = '' return shop_name except Exception as e: print(f'获取店铺名出错:{e}') return '' def get_qualification_number(self): """ 获取资质编号 :return: """ try: qualification_number_str = self.d.xpath( '//*[@resource-id="com.sankuai.meituan:id/mil_container"]/android.webkit.WebView[1]/android.webkit.WebView[1]/android.view.View[1]/android.view.View[1]/android.widget.TextView[2]').text qualification_number = qualification_number_str.strip('资质编号：').strip() return qualification_number except: return None def get_shop_address(self): try: shop_address_xpaths = [ '//*[@resource-id="com.sankuai.meituan:id/wm_sc_drug_shop_content_mrn_container_id_2"]/android.widget.FrameLayout[1]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.widget.ScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.widget.TextView', '//*[@resource-id="com.sankuai.meituan:id/wm_sc_drug_shop_content_mrn_container_id_2"]/android.widget.FrameLayout[1]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.widget.ScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.widget.TextView', ] shop_address = self.get_first_text_by_xpaths(shop_address_xpaths) print(f'111-获取到店铺地址:{shop_address}') if '发货时间' in shop_address: print(f'店铺地址包含发货时间，再次获取店铺地址') shop_address = self._read_xpath_text(shop_address_xpaths[1]) if shop_address: print(f'222-获取到店铺地址:{shop_address}') else: print(f'222-xpath2获取店铺地址失败') print(f'333-获取到店铺地址:{shop_address}') return shop_address except: print(f'获取店铺地址出错-get_shop_address') return None def execute_db_write(self, sql, params, action_desc, max_retries=5): for attempt in range(max_retries): conn = None try: conn = get_mysql() with conn.cursor() as cur: cur.execute(sql, params) conn.commit() print(f"{action_desc}成功") return True except Exception as e: print(f'{action_desc}异常 (尝试 {attempt + 1}/{max_retries}): {e}') if conn: conn.rollback() if attempt == max_retries - 1: print(f"{action_desc}失败，达到最大重试次数") return False time.sleep(2) finally: if conn: conn.close() def query_exists(self, sql, params, error_desc): conn = None try: conn = get_mysql() with conn.cursor() as cur: cur.execute(sql, params) return bool(cur.fetchone()) except Exception as e: print(f"{error_desc}错误: {str(e)}") return None finally: if conn: conn.close() def save_to_database(self, data): print(f'保存数据到数据库:{data}') add_sql = f""" INSERT INTO {self.table_name} ( enterprise_id, platform_id, platform_item_id, province_id, city_id, province_name, city_name, area_info, product_name, product_specs, one_box_price, manufacture_date, expiry_date, manufacturer, approval_number, is_sold_out, online_posting_count, continuous_listing_count, link_url, store_name, store_url, shipment_province_id, shipment_province_name, shipment_city_id, shipment_city_name, company_name, qualification_number, scrape_date, min_price, number, sales, inventory, snapshot_url, insert_time ) 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, %s, %s, %s, %s ) """ params = ( data['enterprise_id'], data['platform_id'], data['platform_item_id'], data['province_id'], data['city_id'], data['province_name'], data['city_name'], data['area_info'], data['product_name'], data['product_specs'], data['one_box_price'], data['manufacture_date'], data['expiry_date'], data['manufacturer'], data['approval_number'], data['is_sold_out'], data['online_posting_count'], data['continuous_listing_count'], data['link_url'], data['store_name'], data['store_url'], data['shipment_province_id'], data['shipment_province_name'], data['shipment_city_id'], data['shipment_city_name'], data['company_name'], data['qualification_number'], data['scrape_date'], data['min_price'], data['number'], data['sales'], data['inventory'], data['snapshot_url'], data['insert_time'], ) return self.execute_db_write(add_sql, params, "保存商品数据到数据库") def save_shop_info_to_database(self, data): print(f'保存店铺数据到数据库:{data}') add_sql = f""" INSERT INTO {self.shop_table_name} (shop, contact_address, qualification_number, business_license_company, business_license_address, scrape_date, platform) VALUES (%s, %s, %s, %s, %s, %s, %s) """ params = ( data['shop'], data['contact_address'], data['qualification_number'], data['business_license_company'], data['business_license_address'], data['scrape_date'], data['platform'], ) return self.execute_db_write(add_sql, params, "保存店铺数据到数据库") def swipe_back(self, no): """ 返回 :param no: 回退次数 :return: """ for idx in range(no): self.d.press('back') time.sleep(self.get_sleep_time()) def drug_price(self): """ 获取药品价格 :return: """ try: price_str = self.d.xpath('//*[starts-with(@text,"¥")]').text price = float(re.search(r'[\d\.]+', price_str).group()) print(f'获取到价格:{price}') return price except Exception as e: print(f'提取价格出错-->{e}') return None def drug_sale_num(self): """ 获取药品销量 :return: """ try: sales_element = self.d.xpath('//*[starts-with(@text,"已售")]') if sales_element.exists: sales_num_str = self.d.xpath('//*[starts-with(@text,"已售")]').text sales_num_str = sales_num_str.replace("已售", "").strip() # price = float(re.search(r'[\d\.]+', price_str).group()) print(f'获取到已售数量:{sales_num_str}') return sales_num_str return None except Exception as e: print(f'提取已售数量出错-->{e}') return None def restart_uiautomator_services(self, device_id): """ 重启atx的uiautomator 服务 :param device_id: :return: """ stop_uiautomator_services = f'adb -s {device_id} shell /data/local/tmp/atx-agent server -d --stop' start_uiautomator_services = f'adb -s {device_id} shell /data/local/tmp/atx-agent server -d' subprocess.run(stop_uiautomator_services, capture_output=True, text=True, shell=True) time.sleep(self.get_sleep_time()) subprocess.run(start_uiautomator_services, capture_output=True, text=True, shell=True) time.sleep(self.get_sleep_time()) def reconnect_device(self): """重启 atx-agent 并重新连接设备""" try: # 停止 atx-agent subprocess.run(["adb", "-s", self.device_id, "shell", "/data/local/tmp/atx-agent", "server", "-d", "--stop"], capture_output=True, timeout=5) time.sleep(1) # 启动 atx-agent subprocess.run(["adb", "-s", self.device_id, "shell", "/data/local/tmp/atx-agent", "server", "-d"], capture_output=True, timeout=5) time.sleep(2) # 重新连接 uiautomator2 self.d = u2.connect_usb(self.device_id) self.restart_uiautomator_services(self.device_id) self.loggerMT.info("设备重连成功") return True except Exception as e: self.loggerMT.error(f"设备重连失败: {e}") return False def connect_devices(self, device_id): """ 连接设备 :return: """ try: self.device_id = device_id self.d = u2.connect_usb(device_id) self.restart_uiautomator_services(device_id) self.oss_config = { "access_key_id": 'LTAI5tDwjfteBvivYN41r8sJ', "access_key_secret": 'yowuOGi2nYYnrqGpO3qcz94C4brcPp', "endpoint": "oss-cn-shenzhen.aliyuncs.com", # 例：oss-cn-beijing.aliyuncs.com "bucket_name": "zhijiayun-jiansuo", "oss_prefix": "scrape_data/" # OSS中存放截图的前缀（虚拟文件夹） } print(f'连接到设备:{device_id}') self.loggerMT.info(f'连接到设备:{device_id}') except Exception as e: print(f'{device_id} 连接错误: {e}') self.loggerMT.info(f'{device_id} 连接错误: {e}') raise Exception(e) def get_ocr_res(self, img): try: # img地址 print(f'开始识别图片:{img}') request_url = "https://aip.baidubce.com/rest/2.0/ocr/v1/business_license" f = open(img, 'rb') img = base64.b64encode(f.read()) params = {"image": img} request_url = request_url + "?access_token=" + self.access_token headers = {'content-type': 'application/x-www-form-urlencoded'} response = requests.post(request_url, data=params, headers=headers) if response: res = response.json() 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: return None except: return None def remove_watermark(self, img_path): """ 图片去水印(将水印部分变成白色背景)并将数据转化为二进制数据 :param img_path: 图片路径 :return: 二进制图片数据 """ img = cv2.imdecode(np.fromfile(img_path, dtype=np.uint8), -1) endswith = os.path.splitext(img_path)[1] new = np.clip(1.4057577998008846 * img - 38.33089999653017, 0, 255).astype(np.uint8) _, img_binary = cv2.imencode(endswith, new) return img_binary def get_ocr_res_image(self, img): try: image = self.remove_watermark(img) # image_file = open(img,'wb') # image_file.write(image) # res_image = self.client.basicAccurate(image) # 高精度 res_image = self.client.basicGeneral(image) data = res_image.get('words_result', '') print(f'百度api返回结果:{data}') return data except: return None def screenshot_the_business_license(self, qualification_number): screenshot_path = 'screenshot1.png' self.d.screenshot(screenshot_path) img = cv2.imread(screenshot_path) # 指定裁剪区域 (left, top, right, bottom) left = 0 top = 480 right = 720 bottom = 1420 cropped_img = img[top:bottom, left:right] # 创建目录 SCREENSHOT_DIR = Path('screenshot') # 注意这里的变化和py文件同一级目录即可 SCREENSHOT_DIR.mkdir(parents=True, exist_ok=True) if qualification_number: # cropped_screenshot_path = 'D:\\work\\dfwy_spider\\drug_data\\mt\\screenshot\\' + qualification_number + '.png' cropped_screenshot_path = SCREENSHOT_DIR / f'{qualification_number}.png' else: cropped_screenshot_path = 'cropped_screenshot.png' cv2.imwrite(cropped_screenshot_path, cropped_img) return cropped_screenshot_path def screenshot_instruction(self): # 获取当前时间 current_time = datetime.datetime.now() # 格式化为时分秒 time_str = current_time.strftime("%H-%M-%S") # 生成随机的 8 位字符串 random_str = secrets.token_hex(4) # 生成 4 个字节的随机字符串，转换为 8 位十六进制字符串 print(time_str) screenshot_path = 'instructionscreenshot1-' + time_str + '-' + random_str + '.png' self.d.screenshot(screenshot_path) return screenshot_path def extract_specification(self, text): """提取药品规格信息""" # 方法1：简单去除到期信息 pattern = r'^[^【]+' match = re.search(pattern, text) if match: return match.group(0).strip() return text # 获取商品title def get_title(self): def _inner(): print(f'获取商品title时的搜索关键字:{self.title_key}') # 初始化 drugs_name = '' specifications = '' title = '' # 循环的获取title为了有时间来处理人机验证 for m in range(1, 6000): if self.d.xpath(f'//*[contains(@text, "{self.title_key}")]').exists: title = self.safe_exec( lambda: self.d.xpath(f'//*[contains(@text, "{self.title_key}")]').text ) print(f"第{m}次获取title成功") break else: time.sleep(3) # return drugs_name, specifications title = title[1:] if title.startswith('0') else title print(f'获取到药品标题:{title}') # match = re.match(r'(\[[^\]]+\])(.*?)\s*((?:\d+\S*|\(.+))$', title) match = re.match(r'^(?:0?)?(?:\[([^\]]+)\])?\s*(.*?)\s*(\d+[^\s]+)$', title) if match: # drugs_name = match.group(1).strip() + match.group(2).strip() drugs_name = title specifications = match.group(3).strip() print("药品名:", drugs_name) print("规格:", specifications) # 如果品规中包含到期则需要再次的正则处理 if '到期' in specifications: specifications = self.extract_specification(specifications) # print('完整药名:', drugs_name + specifications) return drugs_name, specifications else: print("没有匹配到预期格式") drugs_name = title specifications = '' return drugs_name, specifications # 用 safe_exec 包装内部逻辑，确保验证码阻塞 return self.safe_exec(_inner) def enter_shop(self): """ 进店，方便提取资质环境 :return: """ # self.d.xpath('//*[@text="进店"]').click() self.d.xpath('//*[@text="店铺"]').click() time.sleep(self.get_sleep_time()) def enter_shoper(self): """ 进入商家 :return: """ is_shoper_exists = 0 for i in range(10): if self.d.xpath('//*[@text="商家"]').exists: print(f'第{i}次商家存在') is_shoper_exists = 1 break else: print(f'第{i}次商家不存在') time.sleep(self.get_sleep_time()) if is_shoper_exists == 1: self.d.xpath('//*[@text="商家"]').click() time.sleep(self.get_sleep_time()) return True else: return False # 点击查看商家资质 def scan_shoper_license(self): exist_shoper = 0 for i in range(10): if self.d.xpath('//*[@text="查看商家资质"]').exists: print(f'第{i}次查看商家资质存在') exist_shoper = 1 break else: print(f'第{i}次查看商家资质不存在') time.sleep(self.get_sleep_time()) if exist_shoper == 1: self.d.xpath('//*[@text="查看商家资质"]').click() time.sleep(self.get_sleep_time()) else: self.swipe_back(1) # 验证商品的信息是否在数据库中已存在 def data_is_exists(self, data): """ 检查指定数据是否已存在于数据库表中（仅检查存在性） 参数: data: 包含查询条件的字典，键为列名，值为条件值 返回: True: 数据存在 False: 数据不存在 None: 检查过程中出错 """ required_keys = ['product', 'min_price', 'shop', 'scrape_date', 'platform'] if not all(key in data for key in required_keys): missing = [key for key in required_keys if key not in data] logging.error(f"缺少必要字段: {', '.join(missing)}") return None query_sql = f""" SELECT 1 FROM {self.table_name} WHERE product_name = %s AND min_price = %s AND store_name = %s AND scrape_date = %s AND platform_id = %s LIMIT 1 """ params = ( data['product'], data['min_price'], data['shop'], data['scrape_date'], data['platform'] ) return self.query_exists(query_sql, params, "商品查重") # 验证店铺信息是否在数据库中已存在 def shop_is_exists_database(self, shop): query_sql = f""" SELECT 1 FROM {self.shop_table_name} WHERE shop = %s LIMIT 1 """ return self.query_exists(query_sql, (shop,), "店铺查重") def wait_if_verifying(self, monitor, timeout=120): """验证码处理期间阻塞主线程""" start = time.time() while monitor.pausing.is_set() and time.time() - start < timeout: time.sleep(1) def wait_for_ready(self, monitor, timeout=86400): """进入每一页前都先等验证码""" start = time.time() while monitor.pausing.is_set() and time.time() - start < timeout: time.sleep(1) # 额外保险：如果验证码突然在这一秒才弹，再主动扫一次 monitor.check_and_handle_popup() def safe_list(self, xpath, monitor): """线程安全地拿商品列表""" self.wait_for_ready(monitor) return self.d.xpath(xpath).all() def safe_exec(self, func, *args, **kwargs): """ 万能安全壳：执行 func 前检查验证码， 若监控线程已置位 pausing，则一直阻塞直到放行。 """ while self.monitor.pausing.is_set(): time.sleep(1) max_retries = 3 for attempt in range(max_retries): try: return func(*args, **kwargs) except http.client.RemoteDisconnected as e: self.loggerMT.error(f"连接断开 (尝试 {attempt + 1}/{max_retries}): {e}") if attempt == max_retries - 1: raise # 最后一次失败，向上抛出 # 尝试重连 if self.reconnect_device(): self.loggerMT.info("重连成功，准备重试...") time.sleep(2) # 等待设备稳定 continue else: self.loggerMT.error("重连失败，无法继续") raise except Exception as e: # 其他异常直接抛出 raise # 执行真正逻辑 return func(*args, **kwargs) def get_next_data(self, data, target): for i, item in enumerate(data): if item['words'] == target: if i + 1 < len(data): return data[i + 1]['words'] return None def delete_instruction_screenshot(self, screenshot_path): # 删除截图文件 try: os.remove(screenshot_path) print(f"截图文件已删除：{screenshot_path}") except FileNotFoundError: print(f"文件未找到，无法删除：{screenshot_path}") except Exception as e: print(f"删除文件时出错：{e}") def get_instructions_data(self): """ 确定有说明书之后，提取所有的说明书数据 :return: """ self.d.xpath('//*[@text="说明"]').click() # time.sleep(random.randint(3, 5)) time.sleep(0.5) if self.d.xpath('//*[@text="查看详细说明"]').exists: self.d.xpath('//*[@text="查看详细说明"]').click() else: view_all_xpath = self.find_xpath_with_swipes( ['//*[@text="查看全部"]'], swipe_direction='down', swipe_scale=0.3, max_swipes=8, found_desc='查看全部' ) if view_all_xpath: self.d.xpath(view_all_xpath).click() else: res_data = { "有效期": '', "生产单位": '', "批准文号": '' } self.loggerMT.info('获取到的说明书信息为空。') return res_data time.sleep(0.5) for ii in range(8): if self.d.xpath('//*[@text="加载更多"]').exists: self.d.xpath('//*[@text="加载更多"]').click() time.sleep(0.2) break else: self.d.swipe(200, 1000, 200, 300, 0.3) # self.d.swipe_ext("up", scale=0.3) for iii in range(10): if self.d.xpath('//*[@text="生产单位"]').exists and self.d.xpath('//*[@text="批准文号"]').exists: break else: self.d.swipe(200, 1300, 200, 300, 0.3) # self.d.swipe_ext("up", scale=0.3) instruction_path = self.screenshot_instruction() print(f"instruction_path= {instruction_path}") time.sleep(2) ocr_res = self.get_ocr_res_image(instruction_path) # print(f'ocr_res:{ocr_res}') if ocr_res: # 获取有效期的下一个数据 validity = self.get_next_data(ocr_res, '有效期') # 获取批准文号的下一个数据 approval_number = self.get_next_data(ocr_res, '批准文号') # 获取生产单位的下一个数据 manufacturer = self.get_next_data(ocr_res, '生产单位') else: validity = '' approval_number = '' manufacturer = '' res_data = { "有效期": validity, "生产单位": manufacturer, "批准文号": approval_number } print(f"res_data={res_data}") time.sleep(1) self.delete_instruction_screenshot(instruction_path) return res_data def has_instructions(self): """ 是否有说明书 :return: """ # 没有说明书的无法采集具体数据 time.sleep(self.get_sleep_time()) return bool(self.find_xpath_with_swipes( ['//*[@text="说明"]'], swipe_direction='down', swipe_scale=0.3, max_swipes=8, found_desc='说明' )) def has_shop(self): """ 是否有进店按钮 :return: """ # self.d.swipe_ext('up', 0.1) time.sleep(self.get_sleep_time()) is_has_enter_shop = self.d.xpath('//*[@text="进店"]').exists return is_has_enter_shop # 获取商品对应的店铺信息 def get_license_info_ex(self): # self.enter_shop() self.safe_exec(self.enter_shop) # self.enter_shoper() result = self.safe_exec(self.enter_shoper) if result == False: license_info_data = {'contact_address': '', 'qualification_number': '', 'business_license_company': '', 'business_license_address': ''} return license_info_data for i in range(10): if self.d.xpath('//*[@text="查看商家资质"]').exists: print(f"第{i}次有商家资质") break else: print(f"第{i}次没有商家资质") time.sleep(self.get_sleep_time()) # 获取地址 # contact_address = self.get_shop_address() contact_address = self.safe_exec(self.get_shop_address) # time.sleep(50000) ### # self.scan_shoper_license() self.safe_exec(self.scan_shoper_license) # 获取资质编码 # qualification_number = self.get_qualification_number() qualification_number = self.safe_exec(self.get_qualification_number) # qualification_number 不为None继续下一步 if qualification_number: # 营业执照公司名称 business_license_company = '' # 营业执照地址 business_license_address = '' self.d.click(0.603, 0.27) time.sleep(self.get_sleep_time()) cropped_screenshot_path = self.screenshot_the_business_license(qualification_number) print(f'cropped_screenshot_path:{cropped_screenshot_path}') # if qualification_number: # cropped_screenshot_path = 'D:\\work\\dfwy_spider\\drug_data\\mt\\screenshot\\' + qualification_number + '.png' # else: # cropped_screenshot_path = 'cropped_screenshot.png' # ocr_res = self.get_ocr_res('cropped_screenshot.png') ocr_res = self.get_ocr_res(cropped_screenshot_path) print(f'ocr_res:{ocr_res}') # 获取ocr_res 中的地址、单位名称 if ocr_res: if '单位名称' in ocr_res.keys(): business_license_company = ocr_res['单位名称'] if '地址' in ocr_res.keys(): business_license_address = ocr_res['地址'] license_info_data = {'contact_address': contact_address, 'qualification_number': qualification_number, 'business_license_company': business_license_company, 'business_license_address': business_license_address} else: license_info_data = {'contact_address': contact_address, 'qualification_number': '', 'business_license_company': '', 'business_license_address': ''} return license_info_data def distinct_target(self): result = False position_xpath = '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[3]' position_xpath2 = '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[2]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[3]' is_position = self.d.xpath(position_xpath).exists is_position2 = self.d.xpath(position_xpath2).exists xpath = '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.widget.ScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.support.v7.widget.RecyclerView[1]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.widget.HorizontalScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[last()]' xpath2 = '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.widget.ScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.support.v7.widget.RecyclerView[1]/android.widget.FrameLayout[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.widget.HorizontalScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[last()]' xpath3 = '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[2]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.support.v7.widget.RecyclerView[1]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.widget.HorizontalScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[last()]' xpath4 = '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[2]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.support.v7.widget.RecyclerView[1]/android.widget.FrameLayout[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.widget.HorizontalScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[last()]' is_position5 = self.d.xpath(xpath).exists is_position6 = self.d.xpath(xpath2).exists is_position7 = self.d.xpath(xpath3).exists is_position8 = self.d.xpath(xpath4).exists # print(f"is_position = {is_position}") # print(f"is_position2 = {is_position2}") if is_position or is_position2 or is_position5 or is_position6 or is_position7 or is_position8: result = True if result == False: print("---检测没有回到列表页---") else: print("---检测回到了列表页---") return result # return is_position def enter_target_page(self): self.d.xpath('//*[@content-desc="看病买药"]').click() time.sleep(self.get_sleep_time()) self.d.xpath('//*[@resource-id="com.sankuai.meituan:id/vf_search_carousel_text"]').click() time.sleep(self.get_sleep_time()) self.d.xpath( '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]').click() time.sleep(1) self.d.send_keys(self.search_key, clear=True) time.sleep(1) self.d.xpath('//*[@text="搜索"]').click() time.sleep(1) self.click_express_send() time.sleep(1) self.wr_re("读", self.device_id) def click_express_send(self): # xpath= '//*[@resource-id="com.sankuai.meituan:id/container"]//android.widget.HorizontalScrollView[last()]' slide_xpaths = [ '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.widget.ScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.support.v7.widget.RecyclerView[1]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.widget.HorizontalScrollView[1]', '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.widget.ScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.support.v7.widget.RecyclerView[1]/android.widget.FrameLayout[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.widget.HorizontalScrollView[1]', '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[2]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.support.v7.widget.RecyclerView[1]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.widget.HorizontalScrollView[1]', '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[2]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.support.v7.widget.RecyclerView[1]/android.widget.FrameLayout[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.widget.HorizontalScrollView[1]', ] for i in range(1, 3): matched_slide_xpath = self.get_first_existing_xpath(slide_xpaths) if not matched_slide_xpath: time.sleep(self.get_sleep_time()) continue bounds = self.d.xpath(matched_slide_xpath).info['bounds'] top = bounds['top'] bottom = bounds['bottom'] print(f'top={top}') print(f'bottom={bottom}') y = (top + bottom) // 2 print(f'y={y}') self.loggerMT.info(f'开始滑动{i}') self.d.swipe(500, y, 100, y, 0.5) time.sleep(self.get_sleep_time()) break express_send_xpaths = [ '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.widget.ScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.support.v7.widget.RecyclerView[1]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.widget.HorizontalScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[last()]', '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.widget.ScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.support.v7.widget.RecyclerView[1]/android.widget.FrameLayout[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.widget.HorizontalScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[last()]', '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[2]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.support.v7.widget.RecyclerView[1]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.widget.HorizontalScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[last()]', '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[2]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.support.v7.widget.RecyclerView[1]/android.widget.FrameLayout[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.widget.HorizontalScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[last()]', '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[2]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/androidx.recyclerview.widget.RecyclerView[1]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[3]/android.view.ViewGroup[1]', ] self.click_candidate_xpaths( express_send_xpaths, action_desc="点击快递送", max_retries=5, sleep_after=self.get_sleep_time(), ) def get_clipboard(self): time.sleep(1) self.loggerMT.info(f"Clipboard content:{self.d.clipboard}") # 打印调试信息 clipboard_content = self.d.clipboard if clipboard_content is None: return '' return clipboard_content.strip() # return self.d.clipboard.strip() def clear_clipboard(self): self.d.set_clipboard("", "text/plain") def get_product_link(self): product_link = '' dots_xpaths = [ '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[3]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.widget.ImageView[1]', '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[3]/android.view.ViewGroup[2]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.widget.ImageView[1]', '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.widget.ImageView[1]' ] max_retry = 5 # 最多尝试次数 for idx in range(1, max_retry + 1): if product_link: # 已经拿到则退出 break clicked_xpath = self.click_candidate_xpaths( dots_xpaths, action_desc=f'{idx}-进入分享点点点', max_retries=1, sleep_after=0.2, ) if not clicked_xpath: print(f'{idx}-分享入口点击失败') time.sleep(self.get_sleep_time()) continue self.loggerMT.info(f'{idx}-click_exists') self.d.xpath('//*[@text="分享商品"]').click_exists() time.sleep(0.2) link_xpath = '//*[@text="复制链接"]' if self.d.xpath(link_xpath).exists: self.loggerMT.info(f'{idx}-link_xpath click') self.d.xpath(link_xpath).click() time.sleep(1) product_link = self.get_clipboard() time.sleep(0.5) print(f'{idx}-商品链接:{product_link}') self.loggerMT.info(f'{idx}-商品链接:{product_link}') break print(f'{idx}-商品链接:{product_link}') self.loggerMT.info(f'{idx}-商品链接:{product_link}') product_link = '' if not product_link and idx < max_retry: time.sleep(0.5) # 最后一次不需要再等待 return product_link def run_parallel_tasks(self, task_map): """ 并行执行相互独立的只读任务。 任务本身不能包含点击、滑动、返回等会改变页面状态的操作。 """ if not task_map: return {} results = {} with ThreadPoolExecutor(max_workers=len(task_map)) as executor: future_map = { task_name: executor.submit(self.safe_exec, task_func) for task_name, task_func in task_map.items() } for task_name, future in future_map.items(): try: results[task_name] = future.result() except Exception as e: print(f'并行采集任务 {task_name} 执行失败: {e}') results[task_name] = None return results def get_available_xpaths(self, xpaths): check_tasks = { f'xpath_{idx}': (lambda xp=xp: self.d.xpath(xp).exists) for idx, xp in enumerate(xpaths) } exists_results = self.run_parallel_tasks(check_tasks) return [ xpath for idx, xpath in enumerate(xpaths) if exists_results.get(f'xpath_{idx}') ] def get_first_existing_xpath(self, xpaths): available_xpaths = self.get_available_xpaths(xpaths) if not available_xpaths: return None return available_xpaths[0] def get_first_text_by_xpaths(self, xpaths): text_tasks = { f'xpath_{idx}': (lambda xp=xp: self._read_xpath_text(xp)) for idx, xp in enumerate(xpaths) } text_results = self.run_parallel_tasks(text_tasks) for idx, _ in enumerate(xpaths): text = text_results.get(f'xpath_{idx}') if text: return text return '' def get_first_texts_by_xpath_groups(self, xpath_groups): tasks = {} group_keys = {} for group_name, xpaths in xpath_groups.items(): group_keys[group_name] = [] for idx, xpath in enumerate(xpaths): task_name = f'{group_name}_{idx}' group_keys[group_name].append(task_name) tasks[task_name] = (lambda xp=xpath: self._read_xpath_text(xp)) text_results = self.run_parallel_tasks(tasks) grouped_results = {} for group_name, task_names in group_keys.items(): grouped_results[group_name] = '' for task_name in task_names: text = text_results.get(task_name) if text: grouped_results[group_name] = text break return grouped_results def _read_xpath_text(self, xpath): selector = self.d.xpath(xpath) if not selector.exists: return '' try: text = selector.text return text.strip() if isinstance(text, str) else text except Exception: return '' def click_candidate_xpaths(self, xpaths, action_desc, max_retries=1, sleep_after=0): for attempt in range(1, max_retries + 1): available_xpaths = self.get_available_xpaths(xpaths) if not available_xpaths: print(f'{action_desc}失败，第{attempt}次没有匹配到可点击的xpath') time.sleep(self.get_sleep_time()) continue rotate_offset = (attempt - 1) % len(available_xpaths) candidate_xpaths = available_xpaths[rotate_offset:] + available_xpaths[:rotate_offset] for xpath in candidate_xpaths: try: self.safe_exec(lambda xp=xpath: self.d.xpath(xp).click()) print(f'{action_desc}成功') if sleep_after: time.sleep(sleep_after) return xpath except Exception as e: print(f'{action_desc}点击异常: {e}') time.sleep(self.get_sleep_time()) return None def find_xpath_with_swipes(self, xpaths, swipe_direction='down', swipe_scale=0.3, max_swipes=8, found_desc=''): for idx in range(max_swipes): matched_xpath = self.get_first_existing_xpath(xpaths) if matched_xpath: if found_desc: print(f'第{idx}次找到{found_desc}') return matched_xpath self.d.swipe_ext(swipe_direction, swipe_scale) time.sleep(1) matched_xpath = self.get_first_existing_xpath(xpaths) if matched_xpath and found_desc: print(f'第{max_swipes}次找到{found_desc}') return matched_xpath def build_dup_data(self, product, min_price, shop, scrape_date): return { 'product': product, 'min_price': min_price, 'shop': shop, 'scrape_date': scrape_date, 'platform': '4', } def integrate_data(self): """ 整合数据 :return: """ # title_info = self.get_title() # 药品，规格 # title_info = self.safe_exec(self.get_title) # 药品，规格 product, specifications = self.safe_exec(self.get_title) # 药品，规格 if not product: self.swipe_back(1) return page_data = self.run_parallel_tasks({ "min_price": self.drug_price, "sales_num": self.drug_sale_num, "is_self_operated": lambda: self.d.xpath('//*[@text="自营"]').exists, }) min_price = page_data.get("min_price") # 最低价格 sales_num = page_data.get("sales_num") # 销售数量 product_link = '' if page_data.get("is_self_operated"): shop = "美团自营大药房（快递电商）" # 爬取日期 scrape_date = self.get_current_date() dup_data = self.build_dup_data(product, min_price, shop, scrape_date) print(f'当前数据:{dup_data}') if self.data_is_exists(dup_data): print('存在相同数据不入库') self.back_to_list_page() return else: self.find_xpath_with_swipes( ['//*[@text="进店"]'], swipe_direction='up', swipe_scale=0.3, max_swipes=8, found_desc='进店' ) shop_page_data = self.run_parallel_tasks({ "shop": self.get_shop_name_from_current_page, "is_has_enter_shop": self.has_shop, }) shop = shop_page_data.get("shop") if not shop: shop = self.get_shop_name() # 爬取日期 scrape_date = self.get_current_date() dup_data = self.build_dup_data(product, min_price, shop, scrape_date) print(f'当前数据:{dup_data}') if not shop: print('未获取到店铺名：开始回退') self.back_to_list_page() return if '自营' in shop: self.back_to_list_page() return db_check_results = self.run_parallel_tasks({ "dup_exists": lambda: self.data_is_exists(dup_data), "shop_exists": lambda: self.shop_is_exists_database(shop), }) if db_check_results.get("dup_exists"): print('存在相同数据不入库') self.back_to_list_page() return # 获取店铺信息开始 is_has_enter_shop = bool(shop_page_data.get("is_has_enter_shop")) # 需要判断shop是否已经在数据库中存在，如果存在，则不再进入店铺，直接进入下一个商品 shop_is_exists = bool(db_check_results.get("shop_exists")) # 存在进店 并且店铺的名称不包含美团官方的字样 print(f"已采集{self.shop_data_num}家店铺数据") if is_has_enter_shop and '美团官方' not in shop and '美团自营' not in shop and not shop_is_exists and self.shop_data_num < 500: license_info = self.safe_exec(self.get_license_info_ex) contact_address = license_info['contact_address'] qualification_number = license_info['qualification_number'] business_license_company = license_info['business_license_company'] business_license_address = license_info['business_license_address'] save_shop_data = { 'shop': shop, 'contact_address': contact_address, 'qualification_number': qualification_number, 'scrape_date': scrape_date, 'business_license_company': business_license_company, 'business_license_address': business_license_address, 'platform': '4' } self.save_shop_info_to_database(save_shop_data) self.shop_data_num += 1 self.swipe_back(2) else: print('不采集店铺信息') # 获取店铺信息结束 # 商品链接 product_link = self.get_product_link() print(f'获取到product_link: {product_link}') time.sleep(self.get_sleep_time()) # 生产日期为空 manufacture_date = '' expiry_date = '' manufacturer = '' approval_number = '' # 暂时不获取说明书信息 start is_has_instructions = self.safe_exec(self.has_instructions) # 说明书等信息 if is_has_instructions: print('开始获取说明书信息') instructions_info = self.safe_exec(self.get_instructions_data) if instructions_info['有效期'] is not None: expiry_date = instructions_info['有效期'].strip('。') if instructions_info['生产单位'] is not None: manufacturer = instructions_info['生产单位'].strip('。') if instructions_info['批准文号'] is not None: approval_number = instructions_info['批准文号'].strip('。') else: expiry_date = "未知" manufacturer = None approval_number = None province_id = 0 city_id = 0 city = '' province = '' if province in self.city_to_name: province_id = self.city_to_name[province] if city in self.city_to_name: city_id = self.city_to_name[city] save_data = { 'enterprise_id': 3, 'platform_id': 4, 'platform_item_id': '', 'province_id': province_id, 'city_id': city_id, 'province_name': '', 'city_name': '', 'area_info': "", 'product_name': product, 'product_specs': specifications, 'one_box_price': 0.00, 'manufacture_date': manufacture_date, 'expiry_date': expiry_date, 'manufacturer': manufacturer, 'approval_number': approval_number, 'is_sold_out': 0, 'online_posting_count': 1, 'continuous_listing_count': 1, 'link_url': product_link, 'store_name': shop, 'store_url': '', 'shipment_province_id': 0, 'shipment_province_name': "", 'shipment_city_id': 0, 'shipment_city_name': "", 'company_name': "", 'qualification_number': "", 'scrape_date': scrape_date, 'min_price': min_price, 'number': 0, 'sales': sales_num, 'inventory': "", 'snapshot_url': "", 'insert_time': time.strftime('%Y-%m-%d %H:%M:%S'), 'update_time': time.strftime('%Y-%m-%d %H:%M:%S'), } self.save_to_database(save_data) def back_to_list_page(self): for i in range(5): if self.distinct_target(): return True print(f'第{i}次尝试退回到列表页') self.swipe_back(1) time.sleep(self.get_sleep_time()) print('页面出错，没有退回到列表页') return False def reset_collection_cursor(self): self.collection_cursor["page_no"] = 1 self.collection_cursor["item_index"] = 0 def get_current_page_no(self): return self.page + self.collection_cursor["page_no"] def jump_to_page(self, target_page): current_page = self.get_current_page_no() if target_page <= current_page: return while current_page < target_page: if self.d.xpath('//*[@text="已经到底啦"]').exists: print(f"列表实际页数不足，当前停留在第{current_page}页，无法跳转到第{target_page}页") return print(f"跳过第{current_page}页，前往第{current_page + 1}页") self.d.drag(300, 1400, 300, 400, 1) time.sleep(1) self.collection_cursor["page_no"] += 1 self.collection_cursor["item_index"] = 0 current_page = self.get_current_page_no() def move_to_page_range_start(self): if not self.page_range: return start_page = self.page_range["start"] current_page = self.get_current_page_no() if current_page < start_page: self.jump_to_page(start_page) def start_collection_app(self): self.sort_key = 0 self.restart_app() def open_product_list_page(self): self.safe_exec(self.enter_target_page) self.reset_collection_cursor() if self.sort and self.sort_key == 0: self.li_or_lo(self.sort) self.move_to_page_range_start() def handle_workflow_error(self, step_name): action = self.workflow_error_action.get(step_name) if action == "back_to_list_page": if not self.back_to_list_page(): raise RuntimeError("退回列表页失败") return "collect_single_product" if action == "open_product_list_page": return "open_product_list_page" if action == "start_app": return "start_app" raise RuntimeError(f"未配置步骤 {step_name} 的错误处理动作") def get_list_items(self): for _ in range(10): items = self.safe_exec( self.d.xpath('//android.support.v7.widget.RecyclerView/android.widget.FrameLayout').all ) if items: return items time.sleep(1) raise RuntimeError("列表页商品加载失败") def move_to_next_list_page(self): current_page = self.get_current_page_no() if self.page_range and current_page >= self.page_range["end"]: self.wr_re("写", self.device_id, self.sort, current_page) print(f'已完成第{current_page}页采集，达到结束页{self.page_range["end"]}，停止采集') return False if self.d.xpath('//*[@text="已经到底啦"]').exists: return False self.wr_re("写", self.device_id, self.sort, current_page) print(f'当前第{current_page}页采集完成，开始滑动到下一页') self.d.drag(300, 1400, 300, 400, 1) time.sleep(1) self.collection_cursor["page_no"] += 1 self.collection_cursor["item_index"] = 0 return True def _collect_list_item(self, drug_idx, drug_one): bounds = drug_one.info['bounds'] top = bounds['top'] bottom = bounds['bottom'] print(f'当前商品bottom:{bottom}') print(f'当前商品top:{top}') if not (304 <= top and bottom <= 1475): return "skip" print(f"这页的第几个商品:{drug_idx}") item_text_data = self.get_first_texts_by_xpath_groups({ "product_title": [ f'//android.support.v7.widget.RecyclerView/android.widget.FrameLayout[{drug_idx}]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.widget.FrameLayout[1]/android.widget.TextView', f'//android.support.v7.widget.RecyclerView/android.widget.FrameLayout[{drug_idx}]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.widget.FrameLayout[1]/android.widget.TextView', ], "price_str": [ f'//android.support.v7.widget.RecyclerView/android.widget.FrameLayout[{drug_idx}]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.widget.FrameLayout[1]/android.widget.TextView', f'//android.support.v7.widget.RecyclerView/android.widget.FrameLayout[{drug_idx}]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.widget.FrameLayout[1]/android.widget.TextView', f'//android.support.v7.widget.RecyclerView/android.widget.FrameLayout[{drug_idx}]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.widget.FrameLayout[1]/android.widget.TextView', f'//android.support.v7.widget.RecyclerView/android.widget.FrameLayout[{drug_idx}]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.widget.FrameLayout[1]/android.widget.TextView', ], "shop_name": [ f'//android.support.v7.widget.RecyclerView/android.widget.FrameLayout[{drug_idx}]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[2]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.widget.FrameLayout[last()]/android.widget.TextView[1]', f'//android.support.v7.widget.RecyclerView/android.widget.FrameLayout[{drug_idx}]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[2]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.widget.FrameLayout[last()]/android.widget.TextView[1]', ], }) product_title = item_text_data.get("product_title", "") if not product_title: print("列表当前商品名称不存在") self.unrelated_data += 1 return "continue" product_title = product_title[1:] if product_title.startswith('0') else product_title print(f"列表当前商品名称:{product_title}") if not self.is_link_useful(product_title): print(f"is_link_useful 没通过:{product_title}") self.unrelated_data += 1 return "continue" self.unrelated_data = 0 price = '' price_str = item_text_data.get("price_str", "") print(f"列表当前商品价格:{price_str}") if price_str: price = float(re.search(r'[\d\.]+', price_str).group()) if price != '' and self.collect_range: range_start = self.collect_range["start"] range_end = self.collect_range["end"] if not (range_start <= price <= range_end): print(f"price {price} not in range {range_start}-{range_end}, skip") return "continue" shop_name = item_text_data.get("shop_name", "") print(f"列表当前商品店铺名称:{shop_name}") if price == '' or shop_name == '': print("列表当前商品价格或店铺名称不存在") return "continue" scrape_date = self.get_current_date() dup_data = { 'product': product_title, 'min_price': price, 'shop': shop_name, 'scrape_date': scrape_date, 'platform': '4', } if self.data_is_exists(dup_data): print('列表存在相同数据不入库') return "continue" self.safe_exec(drug_one.click) print('点击目标药品完毕') time.sleep(1) self.safe_exec(self.integrate_data) print('integrate_data结束') time.sleep(1) return "collected" def collect_single_product(self): if self.monitor.verification_count >= self.monitor.MAX_VERIFICATION_RETRY: raise RuntimeError("验证码触发过多，暂停程序") if self.page_range: self.move_to_page_range_start() current_page = self.get_current_page_no() if current_page > self.page_range["end"]: print(f"当前已在第{current_page}页，超过结束页{self.page_range['end']}，停止采集") return False items = self.get_list_items() print(f'当前第{self.get_current_page_no()}页，共有{len(items)}个商品') while self.collection_cursor["item_index"] < len(items): item_index = self.collection_cursor["item_index"] self.collection_cursor["item_index"] += 1 result = self._collect_list_item(item_index + 1, items[item_index]) if result in {"continue", "collected"}: self.back_to_list_page() return True if not self.move_to_next_list_page(): print('已经到达列表页最底部') return False return True def execute_workflow_step(self, step_name): if step_name == "start_app": self.safe_exec(self.start_collection_app) return "open_product_list_page" if step_name == "open_product_list_page": self.safe_exec(self.open_product_list_page) return "collect_single_product" if step_name == "collect_single_product": has_next = self.safe_exec(self.collect_single_product) if not has_next: return None print('目前连续无关数据量: ', self.unrelated_data) if self.unrelated_data > self.max_unrelated_data: print(f"连续超过{self.max_unrelated_data}个不达标的数据则停止采集") return None return "collect_single_product" raise RuntimeError(f"未知流程步骤: {step_name}") def main(self, device_id): self.device_id = device_id self.connect_devices(device_id) time.sleep(self.get_sleep_time()) self.monitor = SpiderMonitor(self) self.monitor.start() current_step = "start_app" step_failures = {step: 0 for step in self.workflow_retry_limit} try: while current_step: try: next_step = self.execute_workflow_step(current_step) step_failures[current_step] = 0 current_step = next_step except Exception as e: print(f'{current_step} 执行异常: {e}') time.sleep(5) step_failures[current_step] += 1 if step_failures[current_step] > self.workflow_retry_limit[current_step]: raise current_step = self.handle_workflow_error(current_step) finally: self.monitor.stop() self.monitor.join() device_list = { "U8ONIJJJS4CELVD6": [ { "search_key": "喇叭牌正露丸 100粒", "title_key": "正露丸", "spec_list": [""], "brand": "喇叭牌", "sort": "升序", "collect_range": [], "page_range": [], "workflow_retry_limit": { "start_app": 3, "open_product_list_page": 3, "collect_single_product": 3, }, "workflow_error_action": { "start_app": "start_app", "open_product_list_page": "start_app", "collect_single_product": "back_to_list_page", }, }, ], } def run_device(device_id): """单个设备的采集任务（运行于独立线程）""" if device_id not in device_list: logging.error(f"设备id没有配置: {device_id}") return tasks = device_list[device_id] logging.info(f"[设备 {device_id}] 开始执行，共 {len(tasks)} 个任务") for task in tasks: cycle_no = 0 while True: cycle_no += 1 mt = None logging.info(f'[设备 {device_id}] ========== {task["search_key"]} 第 {cycle_no} 轮采集开始 ==========') try: # 注意：MT类需要从外部导入或定义，此处假设已存在 mt = MT( task["search_key"], task["title_key"], task["spec_list"], task["brand"], task.get("sort"), task.get("collect_range"), task.get("page_range"), task.get("workflow_retry_limit"), task.get("workflow_error_action"), ) mt.main(device_id) logging.info(f'[设备 {device_id}] 关键字 {task["search_key"]} 本轮采集完成') break # 成功则跳出重试循环 except Exception as e: logging.exception(f'[设备 {device_id}] 关键字 {task["search_key"]} 采集异常：{e}') # 发生异常后继续循环重试 finally: if mt and hasattr(mt, 'close'): mt.close() logging.info(f"[设备 {device_id}] 所有任务执行完毕") def main(): # 配置日志格式，便于区分不同线程的输出 logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(threadName)s] %(levelname)s: %(message)s' ) if len(sys.argv) < 2: logging.info(f"美团数据库调度器启动，轮询间隔 {SCHEDULER_INTERVAL_SECONDS} 秒") dispatch_pending_tasks() schedule_dispatch(SCHEDULER_INTERVAL_SECONDS) return device_ids = sys.argv[1:] invalid_ids = [did for did in device_ids if did not in device_list] if invalid_ids: logging.error(f"以下设备ID未配置: {invalid_ids}") return logging.info(f"将运行指定的设备: {device_ids}") if not device_ids: logging.warning("没有可运行的设备") return with ThreadPoolExecutor(max_workers=len(device_ids)) as executor: futures = [executor.submit(run_device, did) for did in device_ids] for future in futures: future.result() if __name__ == '__main__': main()