MT/mt1/spider.py

# mt_spider/spider.py
import re, time, random, datetime, json, logging, requests, base64, cv2, uiautomator2 as u2, subprocess
from monitor import SpiderMonitor
from config import Config
from db import get_mysql

# ------------------ 装饰器 ------------------
def safe_method(func):
    import functools
    @functools.wraps(func)
    def wrapper(self, *args, **kwargs):
        return self.safe_exec(func, self, *args, **kwargs)
    return wrapper

# ------------------ 主类 ------------------
class MT:
    def __init__(self, key: str):
        self.package_name = Config.PACKAGE_NAME
        self.search_key = key
        self.unrelated_data = 0
        self.monitor = None
        self.d = None

    # ------------------ 通用 ------------------
    def safe_exec(self, func, *args, **kwargs):
        while self.monitor and self.monitor.pausing.is_set():
            time.sleep(1)
        return func(*args, **kwargs)

    @staticmethod
    def get_sleep_time():
        return random.randint(5, 8)

    @staticmethod
    def get_current_date():
        return datetime.datetime.now().strftime('%Y-%m-%d')

    def stop_all(self):
        logging.warning("收到停止信号，准备退出")
        if self.monitor:
            self.monitor.stop()

    # ------------------ 设备/APP ------------------
    def connect_devices(self, device_id):
        """
        连接设备
        :return:
        """
        # try:
        #     self.d = u2.connect_usb(device_id)
        #     # 设置隐形等待时间
        #     # self.d.implicitly_wait(5)
        #     self.restart_uiautomator_services(device_id)
        #     print(f'连接到设备:{device_id}')
        # except Exception as e:
        #     print(f'{device_id} 连接错误: {e}')
        #     raise Exception(e)


        self.d = u2.connect_usb(device_id)
        print(f'连接到设备:{device_id}')
        subprocess.run(
            f'adb -s {device_id} shell /data/local/tmp/atx-agent server -d'.split(),
            capture_output=True
        )
        time.sleep(3)

    def restart_app(self):
        self.d.app_stop(self.package_name)
        time.sleep(2)
        self.d.app_start(self.package_name)
        time.sleep(5)

    # ------------------ 页面操作 ------------------
    def enter_target_page(self):
        self.d.xpath('//*[@content-desc="看病买药"]').click()
        self.d.xpath('//*[@resource-id="com.sankuai.meituan:id/vf_search_carousel_text"]').click()
        self.d.xpath('//*[@text="搜索"]').click()
        self.d.send_keys(self.search_key, clear=True)
        self.d.xpath('//*[@text="搜索"]').click()

    # ------------------ 数据抓取 ------------------
    @safe_method
    def get_title(self):
        if "999" in self.search_key:
            self.search_key = self.search_key.replace("999", "")
        title = self.d.xpath(f'//*[contains(@text, "{self.search_key}")]').text

        # title = self.d.xpath('//*[@resource-id="com.sankuai.meituan: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.widget.ScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[3]/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.view.ViewGroup[1]/android.widget.FrameLayout[1]/android.widget.TextView').text

        print(f'获取到药品标题:{title}')

        match = re.search(r'(\[[^\]]+\])(.+?)(\d+.*)', title)
        if match:
            drugs_name = match.group(1).strip() + match.group(2).strip()
            specifications = match.group(3).strip()
            print("药品名:", drugs_name)
            print("规格:", specifications)
            print('完整药名:', drugs_name + specifications)
            return drugs_name, specifications
        else:
            print("没有匹配到预期格式")
            return None, None

        # match = re.search(r'(\[[^\]]+\])(.+?)(\d+.*)', title or "")
        # return (match.group(1) + match.group(2), match.group(3)) if match else (None, None)
    
    @safe_method
    def swipe_up(self):
        """
        上滑
        :return:
        """
        screen_width = self.d.info['displayWidth']
        screen_height = self.d.info['displayHeight']
        duration_rate = random.uniform(0, 0.3)
        self.d.swipe(screen_width // 2, screen_height - 100, screen_width // 2, 100, duration=duration_rate)
        no = random.uniform(0, 1)
        if no > 0.85:
            # 有的时候卡着 再稍微往上滑一点点
            self.d.swipe_ext("up", 0.1)
            time.sleep(self.get_sleep_time())

    @safe_method
    def swipe_back(self, no):
        """
        返回
        :param no: 回退次数
        :return:
        """
        for idx in range(no):
            self.d.press('back')
            time.sleep(self.get_sleep_time())

    @safe_method
    def drug_price(self):
        """
        获取药品价格
        :return:
        """
        try:
            price_str = self.d.xpath('//*[starts-with(@text,"¥")]').text
            price = float(re.search('[\d\.]+', price_str).group())
            print(f'获取到价格:{price}')
            return price
        except Exception as e:
            print(f'提取价格出错-->{e}')
            return None

        # txt = self.d.xpath('//*[starts-with(@text,"¥")]').text
        # return float(re.search(r'[\d.]+', txt).group()) if txt else None

    @safe_method
    def get_shop_name(self):
        try:
            return self.d.xpath('//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').text
        except:
            try:
                return self.d.xpath('//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').text
            except Exception as e:
                logging.error('获取店铺名出错: %s', e)
                return None

    @safe_method
    def get_shop_address(self):
        try:
            shop_address = self.d.xpath('//*[@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').text
            print(f'获取到店铺地址:{shop_address}')
            return shop_address
        except:
            return None

    @safe_method
    def get_qualification_number(self):
        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
            return qualification_number_str.strip('资质编号：').strip()
        except:
            return None

    # ------------------ OCR ------------------
    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}
            # access_token = get_access_token()
            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 screenshot_the_business_license(self, qualification_number: str):
        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]

        if qualification_number:
            cropped_screenshot_path = 'D:\\work\\dfwy_spider\\drug_data\\mt\\screenshot\\' + qualification_number + '.png'
        else:
            cropped_screenshot_path = 'cropped_screenshot.png'

        cv2.imwrite(cropped_screenshot_path, cropped_img)
        return cropped_screenshot_path

    # ------------------ 说明书 ------------------
    @safe_method
    def get_instructions_data(self):
        self.d.xpath('//*[@text="说明"]').click()
        time.sleep(1)
        self.d.xpath('//*[@text="查看详细说明"]').click()
        time.sleep(1)
        self.d.xpath('//*[@text="加载更多"]').click_exists()
        loop_page = 5
        new_list = []
        for i in range(loop_page):
            self.d.xpath('//*[@text="加载更多"]').click_exists()
            time.sleep(0.2)
            if i == 0:
                self.d.swipe(200, 1000, 200, 300, 0.4)
            else:
                self.d.swipe(200, 1000, 200, 62)
            time.sleep(0.2)
            if self.d.xpath('//*[@text="加载更多"]').exists:
                self.d.xpath('//*[@text="加载更多"]').click()
                time.sleep(0.2)
            all_tt = 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[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[3]/android.widget.ScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup').all()

            for idx in range(1, len(all_tt) + 1):
                all_tt1 = self.d.xpath(
                    f'//*[@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[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[3]/android.widget.ScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[{idx}]//android.widget.TextView').all()
                print(f'当前说明书列表数据:{all_tt1}')
                for tt in all_tt1:
                    if tt.text and tt.text != '展开全文':
                        new_list.append(tt.text)
            if i == 0:
                height = 938
            else:
                drug_box = 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[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[3]/android.widget.ScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]').info
                bounds = drug_box['bounds']
                height = bounds['bottom'] - bounds['top']
            if height < 938:
                # print('说明书翻页到底部')
                break

         # 展开全文
        new_list = [item for item in new_list if item != '展开全文']
        print(f'当前说明书列表数据:{new_list}')

        res_data = {
            "有效期":   (new_list[new_list.index("有效期") + 1])   if "有效期"   in new_list and new_list.index("有效期") + 1   < len(new_list) else "",
            "生产单位": (new_list[new_list.index("生产单位") + 1]) if "生产单位" in new_list and new_list.index("生产单位") + 1 < len(new_list) else "",
            "批准文号": (new_list[new_list.index("批准文号") + 1]) if "批准文号" in new_list and new_list.index("批准文号") + 1 < len(new_list) else ""
        }

        print(f'当前说明书字典数据:{res_data}')
        return res_data       

    # ------------------ 店铺资质 ------------------
    def enter_shop(self):
        self.d.xpath('//*[@text="店铺"]').click()
        time.sleep(self.get_sleep_time())

    def enter_shoper(self):
        self.d.xpath('//*[@text="商家"]').click()
        time.sleep(self.get_sleep_time())

    def scan_shoper_license(self):
        self.d.xpath('//*[@text="查看商家资质"]').click()
        time.sleep(self.get_sleep_time())

    @safe_method
    def get_license_info_ex(self):
        self.enter_shop()
        self.enter_shoper()
        contact_address = self.get_shop_address()
        self.scan_shoper_license()
        qualification_number = self.get_qualification_number()

    
        if qualification_number:
            business_license_company = ''
            business_license_address = ''
            self.d.click(0.603, 0.27)
            time.sleep(self.get_sleep_time())
            img_path = self.screenshot_the_business_license(qualification_number)
            print(f'cropped_screenshot_path:{img_path}')
            ocr_res = self.get_ocr_res(img_path)
            print(f'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 data_is_exists(self, data):
        try:
            conn = get_mysql()
            cur = conn.cursor()
            query_sql = """
                SELECT * FROM {} 
                WHERE product = %s 
                AND min_price = %s 
                AND shop = %s 
                AND scrape_date = %s 
                AND platform = %s
            """.format(Config.DB_TABLE)

            cur.execute(query_sql, (
                data['product'],
                data['min_price'],
                data['shop'],
                data['scrape_date'],
                data['platform']
            ))

            result = cur.fetchone()

            return bool(result)  # 如果存在返回True，否则False
        except Exception as e:
            print(f"MySQL 错误: {str(e)}")
            logging.error('检查商品存在性失败: %s', e)
            return False

    def shop_is_exists_database(self, shop):
        try:
            conn = get_mysql()
            cur = conn.cursor()
            query_sql = """
                SELECT * FROM {} 
                WHERE  shop = %s 
            """.format(Config.DB_SHOP_TABLE)
            cur.execute(query_sql, (
                shop
            ))
            result = cur.fetchone()
            return bool(result) # 如果存在返回True，否则False
        except Exception as e:
            print(f"MySQL 错误: {str(e)}")
            logging.error('检查店铺存在性失败: %s', e)
            return False

    def save_to_database(self, data):
        print(f'保存数据到数据库:{data}')

        try:
            conn = get_mysql()
            cur = conn.cursor()
            add_sql = f"""
            INSERT INTO {Config.DB_TABLE} 
            (product, min_price, manufacture_date, expiry_date, shop, business_license_company, province, city, manufacturer, specification, approval_number, product_link, scrape_date, scrape_province, availability, credit_code, platform) 
            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
            """
            cur.execute(add_sql, (
                data['product'], data['min_price'], data['manufacture_date'], data['expiry_date'], data['shop'],
                data['business_license_company'], data['province'], data['city'], data['manufacturer'],
                data['specification'], data['approval_number'], data['product_link'], data['scrape_date'],
                data['scrape_province'], data['availability'], data['credit_code'], data['platform']
            ))
            conn.commit() # 提交数据
            print(f"存入数据库成功")
            logging.info('商品数据已入库')
        except Exception as e:
            logging.error('写入商品数据失败: %s', e)

    def save_shop_info_to_database(self, data):
        print(f'保存店铺数据到数据库:{data}')
        try:
            conn = get_mysql()
            cur = conn.cursor()
            add_sql = f"""
            INSERT INTO {Config.DB_TABLE} 
            (shop, contact_address, qualification_number, business_license_company, business_license_address, scrape_date, platform) 
            VALUES (%s, %s, %s, %s, %s, %s, %s)
            """
            cur.execute(add_sql, (
                data['shop'], data['contact_address'], data['qualification_number'],
                data['business_license_company'], data['business_license_address'],
                data['scrape_date'], data['platform']
            ))
            conn.commit() # 提交数据
            print(f'存入店铺信息到数据库成功')
            logging.info('店铺数据已入库')
        except Exception as e:
            logging.error('写入店铺数据失败: %s', e)

    

        # ------------------ 采集一条完整商品数据 ------------------
    @safe_method
    def integrate_data(self):
        """
        采集一条完整商品 + 店铺信息并入库
        """
        logger = logging.getLogger()
        logger.info('开始采集当前商品详情')

        # 1. 商品名 + 规格
        title_info = self.get_title()
        if not title_info:
            logger.warning('未获取到标题，跳过')
            self.swipe_back(1)
            return
        product, specifications = title_info
        if self.search_key not in product.replace(' ', ''):
            logger.info('无关商品，跳过')
            self.unrelated_data += 1
            self.swipe_back(1)
            return

        # 2. 价格
        min_price = self.drug_price()
        if min_price is None:
            logger.warning('未获取到价格，跳过')
            self.swipe_back(1)
            return

        # 3. 自营判断
        if self.d.xpath('//*[@text="自营"]').exists:
            shop = "美团自营大药房（快递电商）"
            scrape_date = self.get_current_date()
            dup_data = {
                'product': product, 'min_price': min_price,
                'shop': shop, 'scrape_date': scrape_date, 'platform': '美团'
            }
            if self.data_is_exists(dup_data):
                logger.info('自营商品已存在，跳过')
                self.swipe_back(1)
                return
        else:
            # 4. 非自营：找进店
            for i in range(3):
                if self.d.xpath('//*[@text="进店"]').exists:
                    print('开始获取店铺名1')
                    break
                self.d.swipe_ext('up', 0.2)
                time.sleep(1)
                if self.d.xpath('//*[@text="进店"]').exists:
                    print('开始获取店铺名2')
                    break

            shop = self.get_shop_name()
            scrape_date = self.get_current_date()
            dup_data = {
                'product': product, 'min_price': min_price,
                'shop': shop, 'scrape_date': scrape_date, 'platform': '美团'
            }
            if self.data_is_exists(dup_data):
                logger.info('商品已存在，跳过')
                self.swipe_back(1)
                return
            if not shop or '自营' in shop:
                logger.info('店铺为自营或空，跳过')
                self.swipe_back(1)
                return

            # 5. 采集店铺资质（仅新店铺）
            if self.d.xpath('//*[@text="进店"]').exists and '美团官方' not in shop and not self.shop_is_exists_database(shop):
                lic = self.get_license_info_ex()
                save_shop_data = {
                    'shop': shop,
                    'contact_address': lic['contact_address'],
                    'qualification_number': lic['qualification_number'],
                    'business_license_company': lic['business_license_company'],
                    'business_license_address': lic['business_license_address'],
                    'scrape_date': scrape_date,
                    'platform': '美团'
                }
                self.save_shop_info_to_database(save_shop_data)
                self.swipe_back(2)  # 返回两次：资质页 -> 店铺 -> 列表

        # 6. 说明书信息
        if not self.has_instructions():
            logger.info('无说明书，跳过')
            self.swipe_back(1)
            return

        instructions = self.get_instructions_data()
        expiry_date = instructions.get('有效期', '').strip('。')
        manufacturer = instructions.get('生产单位', '').strip('。')
        approval_number = instructions.get('批准文号', '').strip('。')

        # 7. 组装入库数据
        save_data = {
            'product': product,
            'min_price': min_price,
            'manufacture_date': '',
            'expiry_date': expiry_date,
            'shop': shop,
            'business_license_company': '',
            'province': '',
            'city': '',
            'manufacturer': manufacturer,
            'specification': specifications,
            'approval_number': approval_number,
            'product_link': '',
            'scrape_date': scrape_date,
            'scrape_province': '广东',
            'availability': '',
            'credit_code': '',
            'platform': '美团'
        }
        self.save_to_database(save_data)
        logger.info('商品数据已入库：%s', product)
        self.unrelated_data = 0
        self.swipe_back(1)


    # ------------------ 主流程 ------------------
    def main(self, device_id, retry_count=0):
        MAX_RETRY = 3
        logger = logging.getLogger()
        spider_no = 0

        self.connect_devices(device_id)
        time.sleep(self.get_sleep_time())

        self.monitor = SpiderMonitor(self)
        self.monitor.start()

        try:
            self.restart_app()
            self.enter_target_page()

            for idx in range(300):
                logger.info('========== 第 %s 页 ==========', idx + 1)

                if spider_no > 30:
                    logger.info('已采集 30 条，休息 120 秒')
                    time.sleep(120)
                    spider_no = 0

                if self.monitor.verification_count >= self.monitor.MAX_VERIFICATION_RETRY:
                    logger.warning('验证码重试超限，等待人工处理')
                    self.d.toast('请处理验证码后点击继续', 30)
                    self.monitor.verification_count = 0
                    continue

                drug_lis = self.safe_exec(
                    self.d.xpath('//android.support.v7.widget.RecyclerView/android.widget.FrameLayout').all
                )
                lis_len = len(drug_lis)
                logger.info('当前页面共有 %s 个商品', lis_len)

                for drug_one in drug_lis:
                    bounds = drug_one.info['bounds']
                    top, bottom = bounds['top'], bounds['bottom']
                    if not (304 <= top and bottom <= 1559):
                        continue

                    self.safe_exec(drug_one.click)
                    time.sleep(2)

                    try:
                        self.integrate_data()
                        spider_no += 1
                    except Exception as e:
                        logger.exception('采集详情异常: %s', e)
                        self.swipe_back(1)
                        continue

                    if self.safe_exec(self.distinct_target):
                        logger.debug('已返回列表页')
                    else:
                        if self.d.xpath('//*[@text="搜索"]').exists:
                            logger.warning('已回到搜索页，重新开始流程')
                            if retry_count < MAX_RETRY:
                                self.monitor.stop()
                                self.monitor.join()
                                return self.main(device_id, retry_count + 1)
                            else:
                                logger.error('超过最大重试次数，终止')
                                return
                        else:
                            logger.error('无法恢复页面，终止')
                            return

                    time.sleep(self.get_sleep_time())

                if self.d.xpath('//*[@text="已经到底啦"]').exists:
                    logger.info('已到底')
                    break

                self.d.drag(300, 1400, 300, 400, 1)
                time.sleep(self.get_sleep_time())

        finally:
            self.monitor.stop()
            self.monitor.join()