2.0/spiders/yaojiujiu/yyjzt_snapshot.py

"""药九九快照模式 — 一个浏览器从头到尾：检查登录 → 搜索 → 监听API → 匹配DOM → 截卡片"""
import json
import random
import time

import requests
from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeoutError

from commons.Logger import get_spider_logger
from commons.conn_mysql import MySQLPoolOn2
from pipelines.drug_pipelines import DrugPipeline
from oss_upload.oss_upload import AliyunOSSUploader
from spiders.yaojiujiu.yyjzt_crawl import (
    PLATFORM_ID, PAGE_SIZE, MAX_PAGES, SEARCH_API,
    PRODUCT_LINK_TPL, compute_one_price, generate_sign,
)

logger = get_spider_logger("yaojiujiu_snapshot")

import urllib.parse
SEARCH_PAGE_URL = "https://www.yyjzt.com/search?keyword={keyword}"
LOGIN_URL = "https://www.yyjzt.com/login"
HOME_URL = "https://www.yyjzt.com"

# 参考 main.py 的反爬/弹窗处理
USERNAME_SELECTOR = "input[name='username']"
PASSWORD_SELECTOR = "input[name='password']"
LOGIN_BTN_SELECTOR = "button:has-text('登录')"


class YyjztSnapshotSpider:
    def __init__(self, task_dict=None):
        self.task_dict = task_dict or {}
        self.pipeline = DrugPipeline("yaojiujiu_snapshot")
        self.db = MySQLPoolOn2()
        self.is_success = True

        # 任务参数
        self.task_id = ""
        self.company_id = ""
        self.product_name = ""
        self.product_specs = ""
        self.product_brand = ""
        self.keyword = ""
        self.collect_task_id = ""
        self.collect_round = "1"
        self.account_id = ""
        self.collect_region_id = ""
        self.sampling_cycle = ""
        self.sampling_start_time = ""
        self.sampling_end_time = ""

        # 账号
        self._phone = ""
        self._password = ""

        if task_dict:
            self._parse_task()

    def _parse_task(self):
        self.task_id = self.task_dict.get("id", "")
        self.company_id = self.task_dict.get("company_id", "")
        self.product_name = self.task_dict.get("product_name", "")
        self.product_specs = self.task_dict.get("product_specs", "")
        self.product_brand = self.task_dict.get("product_brand", "")
        self.keyword = (
            self.task_dict.get("product_keyword", "")
            or (self.task_dict.get("product_brand", "").strip() + " " + self.product_name.strip() + " " + self.product_specs.strip()).strip()
        ).strip()
        self.collect_task_id = self.task_dict.get("collect_task_id", "")
        self.collect_round = self.task_dict.get("collect_round", "1")
        self.account_id = self.task_dict.get("collect_equipment_account_id", "")
        self.collect_region_id = self.task_dict.get("collect_region_id", "")
        self.sampling_cycle = self.task_dict.get("sampling_cycle", "")
        self.sampling_start_time = self.task_dict.get("sampling_start_time", "")
        self.sampling_end_time = self.task_dict.get("sampling_end_time", "")
        self._phone = self.task_dict.get("account_phone", "")
        self._password = self.task_dict.get("account_password", "")

    # ============ 浏览器启动 + 登录 ============
    def _get_account(self):
        sql = """SELECT `phone`, `password`, `cookie_str` FROM `accounts_platform`
                 WHERE `platform`=%s AND `status`=1
                 ORDER BY `cookie_timestamp` ASC LIMIT 1"""
        rows = self.db.select_data(sql, (PLATFORM_ID,))
        return rows[0] if rows else None

    def _is_login(self, page):
        """参考 main.py is_login()"""
        try:
            page.goto(HOME_URL, timeout=15000)
            page.wait_for_load_state("networkidle")
            time.sleep(2)
            if "login" in page.url.lower():
                return False
            if page.locator(":has-text('欢迎登录')").count() > 0:
                return False
            return True
        except Exception:
            return False

    def _login_operation(self, page, phone, password):
        """参考 main.py login_operation()"""
        try:
            page.goto(LOGIN_URL, timeout=15000)
            page.wait_for_load_state("networkidle")
            time.sleep(2)

            # 切到密码登录 tab
            pwd_tab = page.locator("#tab-first")
            if pwd_tab.count() > 0:
                pwd_tab.click()
                page.wait_for_timeout(500)

            # 填手机号
            page.fill(USERNAME_SELECTOR, "")
            for ch in str(phone):
                page.locator(USERNAME_SELECTOR).type(ch, delay=random.randint(60, 180))
            logger.info("已输入账号")

            # 填密码
            page.fill(PASSWORD_SELECTOR, "")
            for ch in str(password):
                page.locator(PASSWORD_SELECTOR).type(ch, delay=random.randint(60, 180))
            logger.info("已输入密码")

            # 点登录
            page.click(LOGIN_BTN_SELECTOR)
            logger.info("已点击登录")
            time.sleep(3)

            try:
                page.wait_for_load_state("networkidle", timeout=15000)
            except Exception:
                pass

            # 检查是否登录成功
            if "login" in page.url.lower():
                err = page.locator(".el-message--error, [class*='error']").first
                if err.count() > 0:
                    logger.error("登录失败: %s", err.inner_text(timeout=3000))
                return False
            return True
        except PlaywrightTimeoutError as e:
            logger.error("登录超时: %s", e)
            return False

    def _kill_popups(self, page):
        """参考 main.py: 先 JS 暴力清遮罩，再点按钮，三层防线不崩"""
        # 第一层：JS 先暴力移除遮罩 + 滑块验证 + 恢复滚动（main.py kill_masks）
        page.evaluate("""
        () => {
          // 精准删已知遮罩 + 滑块验证
          const selectors = [
            '.v-modal', '.el-overlay', '.el-overlay-dialog',
            '.el-dialog__wrapper', '.el-message-box__wrapper', '.el-loading-mask',
            '.safety-slider-box', '[class*="safety"]', '[class*="slider"]',
            '[class*="captcha"]', '.el-popup-parent--hidden'
          ];
          selectors.forEach(s => {
            document.querySelectorAll(s).forEach(e => {
              try { e.remove(); } catch(_) {}
            });
          });

          // 泛化兜底：近似全屏 + 高 z-index 的层强制隐藏
          const all = Array.from(document.querySelectorAll('body *'));
          for (const el of all) {
            try {
              const s = getComputedStyle(el);
              const z = parseInt(s.zIndex || '0', 10);
              if ((s.position === 'fixed' || s.position === 'absolute')
                  && z >= 1000 && s.pointerEvents !== 'none') {
                const r = el.getBoundingClientRect();
                const nearFull = r.width >= innerWidth * 0.7 && r.height >= innerHeight * 0.7;
                if (nearFull) {
                  el.style.pointerEvents = 'none';
                  el.style.display = 'none';
                }
              }
            } catch(_) {}
          }

          // 恢复滚动
          document.documentElement.style.overflow = 'auto';
          document.body.style.overflow = 'auto';
          document.body.style.position = 'static';
          document.body.style.width = 'auto';
          document.body.classList.remove('el-popup-parent--hidden');
        }
        """)
        page.wait_for_timeout(300)

        # 第二层：连续点引导/关闭按钮（兼容多步引导），try 包住不崩
        for _ in range(6):
            try:
                btn = page.locator(
                    "xpath=//button[normalize-space()='下一步' or normalize-space()='完成' "
                    "or normalize-space()='我知道了' or normalize-space()='关闭']"
                ).first
                if btn.count() > 0 and btn.is_visible():
                    btn.click(timeout=1500)
                    page.wait_for_timeout(300)
                    continue
            except Exception:
                pass
            try:
                close_icon = page.locator(
                    ".el-dialog__headerbtn, .el-message-box__headerbtn, "
                    ".close, .icon-close, .el-icon-close"
                ).first
                if close_icon.count() > 0 and close_icon.is_visible():
                    close_icon.click(timeout=1200, force=True)
                    page.wait_for_timeout(300)
                    continue
            except Exception:
                pass
            break

    # ============ API 搜索 ============
    def _api_search_all(self):
        """API 搜全量，返回 [(item_dict, page), ...]"""
        all_items = []
        for p in range(1, MAX_PAGES + 1):
            body = {"pageIndex": p, "pageSize": PAGE_SIZE, "platformType": 1,
                    "keyword": self.keyword, "searchNum": 0,
                    "showSearchTips": 0, "isNeedAggregation": 0}
            ts = int(time.time() * 1000)
            # 用 browser 当前 cookie 里的 token
            headers = {
                "Content-Type": "application/json", "Origin": "https://www.yyjzt.com",
                "sy_plat": "B2B", "token_platform_client_type": "USER",
                "zhcaiToken": self._token, "deviceId": self._phone,
                "_s": generate_sign(self._token, body, ts), "_t": str(ts),
            }
            try:
                resp = requests.post(SEARCH_API, json=body, headers=headers, timeout=30)
                data = resp.json()
                if data.get("success") and data.get("code") == 200:
                    items = data["data"]["itemList"]
                    for item in items:
                        all_items.append((item, p))
                    total_count = int(data["data"].get("totalCount") or 0)
                    if total_count > 0 and len(all_items) >= total_count:
                        break
                    if len(items) == 0:
                        break
                else:
                    logger.error("API page %s: %s", p, data.get("msg"))
                    break
            except Exception as e:
                logger.error("API page %s: %s", p, e)
                break
            time.sleep(random.uniform(1, 2))
        logger.info("API 共 %s 条", len(all_items))
        return all_items

    # ============ DOM 匹配 ============
    def _extract_dom_cards(self, page):
        """逐张卡片滚到可视区，触发懒加载后再取 innerText"""
        cards = []
        card_els = page.locator(".goods_card")
        cnt = card_els.count()
        if cnt == 0:
            logger.warning("页面无 .goods_card 元素")
            return cards

        # 先滚回顶部
        page.evaluate("window.scrollTo(0, 0)")
        page.wait_for_timeout(300)

        # 预估每张卡高度，用于判断是否需要继续滚
        first_card_height = 0
        try:
            box = card_els.nth(0).bounding_box()
            if box:
                first_card_height = box["height"]
        except Exception:
            pass
        # 每屏大约能显示多少张卡（假设卡高约 260px，视口 900px → ~3排 = 随实际布局调整）
        cards_per_screen = max(2, int(900 / max(first_card_height, 200))) if first_card_height else 3

        for i in range(cnt):
            try:
                card = card_els.nth(i)
                # 滚到这张卡中间位置，确保它进入可视区
                card.scroll_into_view_if_needed()
                page.wait_for_timeout(300)  # 给懒加载时间

                all_text = card.inner_text(timeout=2000).strip()
                if all_text:
                    cards.append({"all_text": all_text, "locator": card, "index": i})
                else:
                    # 偶尔 inner_text 第一次拿不到（渲染慢），重试一次
                    page.wait_for_timeout(500)
                    all_text = card.inner_text(timeout=2000).strip()
                    if all_text:
                        cards.append({"all_text": all_text, "locator": card, "index": i})
                    else:
                        logger.warning("card[%s] 文本为空，可能的结构异常", i)
            except Exception as e:
                logger.warning("card[%s] 提取失败: %s", i, e)

            # 每屏结束时额外等一下，防止滚太快
            if (i + 1) % cards_per_screen == 0:
                page.wait_for_timeout(400)

        logger.info("逐卡提取: %s/%s 个卡片有文本", len(cards), cnt)
        return cards

    @staticmethod
    def _price_variants(value):
        """同一个价格的不同字符串形态：15.0 → {"15.00","15.0","15"}"""
        vs = set()
        vs.add(f"{value:.2f}")
        vs.add(str(value))
        vs.add(f"{value:.2f}".rstrip("0").rstrip("."))
        vs.add(f"{value:.1f}".rstrip("0").rstrip("."))
        return vs

    def _match(self, api_item, dom_cards, used):
        api_name = api_item.get("itemStoreName", "")
        api_store = api_item.get("storeName", "")
        api_specs = api_item.get("specs", "")
        api_price = compute_one_price(api_item)
        # sortPrice 是页面实际展示价（折后/拼团后），compute_one_price 是计算价，
        # DOM 可能显示其中任意一个，两者都纳入匹配池
        api_sort_price = float(api_item.get("sortPrice") or 0)

        # 价格变体：compute_one_price + sortPrice 的所有字符串形态
        price_pool = self._price_variants(api_price)
        if api_sort_price and abs(api_sort_price - api_price) > 0.001:
            price_pool |= self._price_variants(api_sort_price)

        fail_reasons = []
        for c in dom_cards:
            if c["index"] in used:
                continue
            txt = c["all_text"]
            name_ok = bool(api_name) and (api_name in txt)
            store_ok = bool(api_store) and (api_store in txt)
            price_ok = any(p in txt for p in price_pool)
            # specs 辅助消歧：同名+同店+同价时，用规格区分
            specs_ok = (not api_specs) or (api_specs in txt)
            if name_ok and store_ok and price_ok and specs_ok:
                return c, ""
            fail_reasons.append(
                f"  card[{c['index']}]: name={name_ok} store={store_ok} price={price_ok} specs={specs_ok} "
                f"card_text={txt[:80]}"
            )
        # 没匹配上，返回原因
        detail = (
            f"API: name={api_name!r} store={api_store!r} price={api_price}"
            f" sortPrice={api_sort_price} specs={api_specs!r} pool={price_pool}\n"
            + "\n".join(fail_reasons[:5])
        )
        return None, detail

    # ============ 翻页 ============
    def _goto_next_page(self, page):
        """点击下一页按钮（参考 main.py goto_next_page），成功返回 True"""
        candidates = [
            ".el-pagination button.btn-next:not(.is-disabled)",
            ".el-pagination__next:not(.is-disabled)",
            "button:has-text('下一页'):not([disabled])",
            "a:has-text('下一页')",
        ]
        for sel in candidates:
            btn = page.locator(sel).first
            if btn.count() > 0:
                try:
                    page.evaluate("window.scrollTo(0, 0);")
                    btn.click(timeout=5000)
                    page.wait_for_load_state("networkidle")
                    return True
                except Exception:
                    continue
        return False

    # ============ 截图+入库 ============
    def _process_page(self, page, items, page_no):
        self._kill_popups(page)
        if page_no == 1:
            # 第一页：URL 打开搜索
            url = SEARCH_PAGE_URL.format(keyword=urllib.parse.quote(self.keyword))
            page.goto(url, timeout=20000)
        else:
            # 后续页：点击"下一页"按钮（URL 翻页不生效，Vue 前端必须点按钮）
            if not self._goto_next_page(page):
                logger.warning("第 %s 页翻页失败，停止", page_no)
                return False
        page.wait_for_load_state("networkidle")
        time.sleep(random.uniform(1, 2))
        self._kill_popups(page)  # 页面加载完再清一次，防止延迟弹窗

        # 截个整页图方便排查
        page.screenshot(path=f"./snapshots/full_page_{page_no}.png", full_page=True)

        dom_cards = self._extract_dom_cards(page)
        logger.info("页%s: DOM=%s API=%s", page_no, len(dom_cards), len(items))

        import os
        os.makedirs("./snapshots", exist_ok=True)
        os.makedirs("./snapshots/unmatched", exist_ok=True)
        os.makedirs("./snapshots/debug", exist_ok=True)

        # 先把所有 DOM 卡片截小图存 debug，方便排查
        for c in dom_cards[:5]:  # 只截前5个看结构
            try:
                jpg = c["locator"].screenshot(type="jpeg", quality=60)
                fname = f"debug_card_{page_no}_{c['index']}.jpg"
                with open(os.path.join("./snapshots/debug", fname), "wb") as f:
                    f.write(jpg)
            except Exception:
                pass

        used = set()
        matched = 0
        unmatched_log = []
        for item, _ in items:
            if self.product_brand and self.product_brand.strip() not in str(item.get("brandName") or ""):
                continue
            if self.product_name and self.product_name.strip() not in str(item.get("itemStoreName") or ""):
                continue
            if self.product_specs and self.product_specs.strip() not in str(item.get("specs") or ""):
                continue
            card, fail_detail = self._match(item, dom_cards, used)
            if not card:
                unmatched_log.append(fail_detail)
                continue
            used.add(card["index"])
            try:
                card["locator"].scroll_into_view_if_needed()
            except Exception:
                pass
            time.sleep(0.3)

            # 截图
            try:
                jpg = card["locator"].screenshot(type="jpeg", quality=80)
            except Exception as e:
                logger.warning("截图失败 item=%s: %s", item.get("itemStoreId"), e)
                continue

            # 上传 OSS
            fname = f"yjj_{item.get('itemStoreId','?')}_{page_no}_{card['index']}"
            try:
                snapshot_url = self.uploader.upload_from_bytes(jpg, fname)
            except Exception as e:
                logger.warning("OSS上传失败: %s, 存本地", e)
                local_path = os.path.join("./snapshots", fname + ".jpg")
                with open(local_path, "wb") as f:
                    f.write(jpg)
                snapshot_url = local_path

            # 入库
            product = self._to_product(item, snapshot_url)
            try:
                self.pipeline.storge_data(product)
                matched += 1
            except Exception as e:
                logger.exception("入库失败: %s", e)

            # 同时写文件
            self._all_results.append(product)

        # 未匹配的截全卡片图存 unmatched，同时写日志
        skipped = len(items) - matched
        if unmatched_log:
            log_path = os.path.join("./snapshots", f"unmatched_page{page_no}.log")
            with open(log_path, "w", encoding="utf-8") as lf:
                lf.write(f"=== 页{page_no} 未匹配 {skipped} 条 ===\n\n")
                lf.write("\n---\n".join(unmatched_log))
            logger.info("未匹配日志: %s", log_path)
            # 截几个未匹配的 DOM 卡片
            for i, c in enumerate(dom_cards):
                if c["index"] not in used and i < 10:
                    try:
                        jpg = c["locator"].screenshot(type="jpeg", quality=70)
                        fname = f"unmatched_{page_no}_{c['index']}.jpg"
                        with open(os.path.join("./snapshots/unmatched", fname), "wb") as f:
                            f.write(jpg)
                    except Exception:
                        pass

        if skipped:
            logger.info("页%s: 匹配入库 %s 条, 未匹配跳过 %s 条", page_no, matched, skipped)

    def _to_product(self, item, snapshot_url=""):
        now = time.strftime("%Y-%m-%d %H:%M:%S")
        today = time.strftime("%Y-%m-%d")
        sid = str(item.get("itemStoreId") or "")
        # min_price = gc-l2-price = 页面实际显示价
        min_price = compute_one_price(item)
        return {
            "platform": PLATFORM_ID, "item_id": sid,
            "enterprise_id": self.company_id,
            "product_name": item.get("itemStoreName") or "",
            "spec": item.get("specs") or "",
            "one_price": 0,   # 不存此字段，填0
            "detail_url": PRODUCT_LINK_TPL.format(item_store_id=sid) if sid else "",
            "shop_name": item.get("storeName") or "",
            "anonymous_store_name": "",   # 不存此字段
            "shop_url": "", "city_name": "", "city_id": "",
            "province_name": "", "province_id": "", "area_info": "",
            "factory_name": item.get("manufacturer") or item.get("manufactureName") or "",
            "scrape_date": today,
            "price": min_price,   # → DB min_price = gc-l2-price
            "sales": item.get("totalCustCount") or "",
            "stock_count": item.get("itemStorage") or item.get("ownStorageNumber") or 0,
            "snapshot_url": snapshot_url,
            "approval_num": item.get("approvalNo") or "",
            "produced_time": "",
            "deadline": str(item.get("validateTimeEnd") or ""),
            "update_time": now, "insert_time": now,
            "number": 1,   # 固定为1
            "product_brand": item.get("brandName") or self.product_brand,
            "collect_task_id": self.collect_task_id,
            "search_name": self.product_name,
            "company_name": item.get("storeFullName") or "",
            "collect_config_info": json.dumps({
                "sampling_cycle": self.sampling_cycle,
                "sampling_start_time": self.sampling_start_time,
                "sampling_end_time": self.sampling_end_time,
            }, ensure_ascii=False),
            "account_id": self.account_id,
            "collect_region_id": self.collect_region_id,
            "collect_round": self.collect_round,
            "is_sold_out": 1 if item.get("isSoldOut") else 0,
        }

    # ============ 主流程 ============
    def run(self):
        self.uploader = AliyunOSSUploader()
        self._all_results = []

        # 1. 拿账号
        acc = self._get_account()
        if not acc:
            logger.error("无可用账号")
            return 0, False
        phone = acc["phone"]
        password = acc["password"]

        # 2. 一个浏览器干到底（headless + Linux 兼容）
        with sync_playwright() as p:
            browser = p.chromium.launch(
                headless=True, channel="chrome",
                args=[
                    "--disable-blink-features=AutomationControlled",
                    "--no-sandbox",
                    "--disable-setuid-sandbox",
                    "--disable-dev-shm-usage",
                    "--disable-gpu",
                ],
            )
            context = browser.new_context(
                locale="zh-CN", timezone_id="Asia/Shanghai",
                viewport={"width": 1600, "height": 900},
            )
            page = context.new_page()

            # 3. 加载 DB 缓存的 cookie（类似 main.py load_cookies）
            db_cookie_str = acc.get("cookie_str") or ""
            if db_cookie_str:
                try:
                    db_cookies = json.loads(db_cookie_str)
                    cookie_list = [
                        {"name": k, "value": str(v), "domain": ".yyjzt.com", "path": "/"}
                        for k, v in db_cookies.items() if v and k not in ("Token",)
                    ]
                    if cookie_list:
                        page.goto(HOME_URL, timeout=10000)  # 先访问域名
                        context.add_cookies(cookie_list)
                        logger.info("已加载 DB 缓存 %s 个 cookie", len(cookie_list))
                except (json.JSONDecodeError, Exception) as e:
                    logger.warning("加载缓存 cookie 失败: %s", e)

            # 4. 检查登录态 → 没登录当场登（同一个浏览器）
            if self._is_login(page):
                logger.info("Cookie 有效，已登录")
            else:
                logger.info("未登录，开始登录...")
                if not self._login_operation(page, phone, password):
                    logger.error("登录失败")
                    browser.close()
                    return 0, False
                logger.info("登录成功")
                # 登录后清弹窗
                self._kill_popups(page)
                # 保存 cookie 到 DB
                cookies = context.cookies()
                cookie_map = {c["name"]: c["value"] for c in cookies}
                token = cookie_map.get("yjj-token", "").replace("ssr_", "")
                cookie_map["Token"] = token
                now_ts = int(time.time())
                self.db.execute(
                    "UPDATE accounts_platform SET cookie_str=%s, cookie_timestamp=%s WHERE phone=%s AND platform=%s",
                    (json.dumps(cookie_map, ensure_ascii=False), now_ts, str(phone), PLATFORM_ID),
                )
                logger.info("已保存 %s 个 cookie 到 DB", len(cookie_map))

            # 4. 从 browser cookie 提取 token 给 API 用
            cookies = context.cookies()
            for c in cookies:
                if c["name"] == "yjj-token":
                    self._token = c["value"].replace("ssr_", "")
                    break

            # 5. API 搜全量数据
            all_items = self._api_search_all()
            if not all_items:
                browser.close()
                return 0, False

            # 6. 按页码分组，逐页截图
            from collections import defaultdict
            pages = defaultdict(list)
            for item, page_no in all_items:
                pages[page_no].append((item, page_no))

            for page_no in sorted(pages.keys()):
                if self._process_page(page, pages[page_no], page_no) is False:
                    break
                time.sleep(random.uniform(2, 3))

            # 写结果文件
            # out_path = f"./snapshots/result_{int(time.time())}.json"
            # with open(out_path, "w", encoding="utf-8") as f:
            #     json.dump(self._all_results, f, ensure_ascii=False, indent=2, default=str)
            # logger.info("结果文件: %s (%s 条)", out_path, len(self._all_results))

            browser.close()

        logger.info("快照完成: %s 条", self.pipeline.crawl_count)
        return self.pipeline.crawl_count, self.is_success, len(pages)


if __name__ == "__main__":
    task = {
        "id": 999, "company_id": "1",
        "product_name": "蒲地蓝消炎片", "product_specs": "",
        "product_brand": "999", "product_keyword": "999 蒲地蓝消炎片",
        "current_page": 0, "start_page": 1, "end_page": 0,
        "scheduler": None, "_device_id": "test", "_platform": "7",
        "collect_task_id": "999", "snapshot_collect_status": 0,
        "collect_round": 1,
        "account_phone": "", "account_password": "", "account_name": "",
        "sampling_cycle": "", "sampling_start_time": "", "sampling_end_time": "",
        "collect_equipment_account_id": "", "collect_region_id": "",
    }
    s = YyjztSnapshotSpider(task)
    print(s.run())