tob/spiders/yaojiujiu/yyjzt_crawl.py

"""药九九搜索爬虫 — 通过 API 接口采集商品价格，缺店铺信息时自动补齐"""
import hashlib
import json
import random
import time

import requests

from commons.Logger import get_spider_logger
from commons.conn_mysql import MySQLPoolOn2
from pipelines.drug_pipelines import DrugPipeline
from pipelines.shop_pipelines import ShopPipeline
from spiders.yaojiujiu.login_yaojiujiu import YaoJiuJiuLogin

logger = get_spider_logger("yaojiujiu")

# ---------- 常量 ----------
PLATFORM_ID = 7
SEARCH_API = "https://web-api.yyjzt.com/search/api/search/item/list"
STORE_API = "https://web-api.yyjzt.com/sale/api/Sale/storeinfo/findById"
PRODUCT_LINK_TPL = (
    "https://www.yyjzt.com/goodDetail?ladderNum"
    "&itemStoreId={item_store_id}"
    "&sourceProdetail=%2Fsearch"
    "&is_store=0"
)
PAGE_SIZE = 60
MAX_PAGES = 30


def compute_one_price(item):
    """
    完全复刻页面 gc-l2-price 展示逻辑:
      有活动(特价/团购) → sortPrice（活动价）
      无活动(普通)     → itemPrice（原价）
    """
    discount = item.get("marketItemStoreDiscountPriceToTrade") or {}
    if discount.get("specialPrice") is not None:
        return float(discount["specialPrice"])
    if discount.get("joinGroupPrice") is not None:
        return float(discount["joinGroupPrice"])
    return float(item.get("itemPrice") or 0)


def generate_sign(token, body_dict, ts):
    """_s 签名: MD5(token + JSON.stringify(body) + timestamp)"""
    body_json = json.dumps(body_dict, separators=(",", ":"), ensure_ascii=False)
    raw = token + body_json + str(ts)
    return hashlib.md5(raw.encode("utf-8")).hexdigest()


class YyjztSpider:
    def __init__(self, task_dict=None):
        self.task_dict = task_dict or {}
        self.pipeline = DrugPipeline("yaojiujiu")
        self.shop_pipeline = ShopPipeline("yaojiujiu")
        self.db = MySQLPoolOn2()
        self.page = 0
        self.is_success = True
        self.collect_task_id = None
        self._seen_stores = set()  # 已处理的 storeId，避免重复查库/调接口

        # ---- 从 task 提取参数 ----
        self.task_id = ""
        self.company_id = ""
        self.product_name = ""
        self.product_specs = ""
        self.product_brand = ""
        self.keyword = ""
        self.sampling_cycle = ""
        self.sampling_start_time = ""
        self.sampling_end_time = ""
        self.account_id = ""
        self.collect_region_id = ""
        self.collect_round = "1"

        if self.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.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.account_id = self.task_dict.get("collect_equipment_account_id", "")
        self.collect_region_id = self.task_dict.get("collect_region_id", "")
        self.collect_round = self.task_dict.get("collect_round", "1")
        self.collect_task_id = self.task_dict.get("collect_task_id", "")

        # 登录账号
        self._account_phone = self.task_dict.get("account_phone", "")
        self._account_password = self.task_dict.get("account_password", "")

        # deviceId 用账号 phone 或固定值
        self._device_id = self._account_phone or "yyjzt_device"

        # token 从 accounts_platform 缓存读取，过期后自动重登
        self._token = None
        self._ensure_token()

    # ==================== 进度回告 ====================
    def _report_exception(self, exception_type, remark=""):
        scheduler = self.task_dict.get("scheduler")
        if not scheduler:
            return
        data = {
            "task_id": self.task_id,
            "platform": str(PLATFORM_ID),
            "username": self.task_dict.get("_device_id", ""),
            "current_page": self.page,
            "is_finished": 0,
            "exception_type": exception_type,
            "remark": remark,
        }
        if self.pipeline.crawl_count:
            data["crawled_count"] = self.pipeline.crawl_count
        scheduler.post_report(data)

    def _report_page_progress(self, page, crawled_count, is_finished=0, total_pages=0):
        scheduler = self.task_dict.get("scheduler")
        if not scheduler:
            return False
        data = {
            "task_id": self.task_id,
            "platform": str(PLATFORM_ID),
            "username": self.task_dict.get("_device_id", ""),
            "current_page": page,
            "crawled_count": crawled_count,
            "is_finished": is_finished,
        }
        if total_pages:
            data["total_pages"] = total_pages
        scheduler.post_report(data)
        return scheduler.end

    # ==================== 登录 & Token 管理 ====================
    # 参考 main.py is_login() — 主动检测，不等到 API 报错
    def _is_token_valid(self):
        """用轻量 API 主动检测 token 是否有效（类似 main.py 检测'欢迎登录'文本）"""
        if not self._token:
            return False
        body = {"pageIndex": 1, "pageSize": 1, "platformType": 1,
                "keyword": "999感冒灵", "searchNum": 0, "showSearchTips": 0, "isNeedAggregation": 0}
        try:
            headers = self._signed_headers(body)
            resp = requests.post(SEARCH_API, json=body, headers=headers, timeout=10)
            data = resp.json()
            return data.get("success") and data.get("code") == 200
        except Exception:
            return False

    def _get_token_from_db(self):
        """直接从 DB 读缓存 token"""
        sql = """SELECT `cookie_str` FROM `accounts_platform`
                 WHERE `platform`=%s AND `status`=1 AND `phone`=%s
                 ORDER BY `cookie_timestamp` DESC LIMIT 1"""
        rows = self.db.select_data(sql, (PLATFORM_ID, self._account_phone))
        if not rows:
            return None
        try:
            cd = json.loads(rows[0].get("cookie_str") or "{}")
            return cd.get("Token") or cd.get("token")
        except json.JSONDecodeError:
            return None

    def _ensure_token(self):
        """先读 DB 缓存 → 主动验证是否过期 → 过期则浏览器重新登录"""
        # 1. 读缓存（类似 main.py load_cookies）
        token = self._get_token_from_db()
        if token:
            self._token = token
            # 2. 主动检测是否有效（类似 main.py is_login）
            if self._is_token_valid():
                logger.info("缓存 token 有效: %s...", token[:16])
                return
            else:
                logger.info("缓存 token 已过期")

        # 3. 过期或不存在 → 浏览器登录（类似 main.py login_operation + save_cookies）
        logger.info("启动浏览器登录...")
        login = YaoJiuJiuLogin()
        token = login.run(headless=True)
        if token:
            self._token = token
        else:
            logger.error("登录失败")
            self._token = ""

    def _re_login(self):
        """Token 失效，强制重新登录"""
        logger.info("Token 失效，重新登录...")
        login = YaoJiuJiuLogin()
        token = login.run(headless=True)
        if token:
            self._token = token
            return True
        return False

    # ==================== API 请求头 ====================
    def _base_headers(self):
        return {
            "Origin": "https://www.yyjzt.com",
            "Referer": "https://www.yyjzt.com/",
            "sy_plat": "B2B",
            "token_platform_client_type": "USER",
            "zhcaiToken": self._token,
            "deviceId": self._device_id,
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
        }

    def _signed_headers(self, body):
        ts = int(time.time() * 1000)
        sign = generate_sign(self._token, body, ts)
        headers = self._base_headers()
        headers["Content-Type"] = "application/json"
        headers["_s"] = sign
        headers["_t"] = str(ts)
        return headers

    # ==================== 搜索 API ====================
    def _search_page(self, page_index, retry=True):
        if not self._token:
            raise Exception("无登录 token")
        body = {
            "pageIndex": page_index,
            "pageSize": PAGE_SIZE,
            "platformType": 1,
            "keyword": self.keyword,
            "searchNum": 0,
            "showSearchTips": 0,
            "isNeedAggregation": 0,
        }
        headers = self._signed_headers(body)
        resp = requests.post(SEARCH_API, json=body, headers=headers, timeout=30)
        resp.raise_for_status()
        data = resp.json()

        # 登录失效则重登后重试一次
        if not (data.get("success") and data.get("code") == 200):
            msg = data.get("msg", "")
            if retry and ("登录" in msg or "token" in msg.lower() or "未登录" in msg):
                logger.warning("Token 失效(%s)，尝试重新登录", msg)
                if self._re_login():
                    return self._search_page(page_index, retry=False)
            raise Exception(f"API error: {msg}")

        return data["data"]["itemList"]

    # ==================== 店铺信息补齐 ====================
    def _ensure_shop_info(self, store_id, store_name):
        """确保店铺信息已入库。先查 retrieve_scrape_shop_info，没有则调 API 抓取。"""
        if not store_id or store_id in self._seen_stores:
            return
        self._seen_stores.add(store_id)

        # 查库
        exist = self.db.select_data(
            "SELECT 1 FROM retrieve_scrape_shop_info WHERE platform=%s AND shop=%s LIMIT 1",
            (str(PLATFORM_ID), store_name),
        )
        if exist:
            return

        # 调店铺信息接口
        logger.info("店铺表无 [%s]，抓取店铺信息 storeId=%s", store_name, store_id)
        try:
            resp = requests.get(
                STORE_API,
                params={"id": store_id},
                headers=self._base_headers(),
                timeout=15,
            )
            resp.raise_for_status()
            data = resp.json()
            if not (data.get("success") and data.get("code") == 200):
                logger.warning("店铺接口异常 storeId=%s: %s", store_id, data.get("msg"))
                return
            info = data["data"]
        except Exception as e:
            logger.error("店铺接口请求失败 storeId=%s: %s", store_id, e)
            return

        # 映射到 retrieve_scrape_shop_info 字段
        today = time.strftime("%Y-%m-%d")
        shop_row = {
            "id": PLATFORM_ID,
            "platform": str(PLATFORM_ID),
            "shop": (info.get("storeShortName") or info.get("storeTitle") or store_name),
            "province": info.get("provinceName") or "",
            "city": info.get("cityName") or "",
            "contact_address": info.get("storeAddress") or "",
            "qualification_number": info.get("organizationCode") or "",
            "business_license_company": info.get("storeName") or "",
            "business_license_address": info.get("storeAddress") or "",
            "scrape_date": today,
            "ai_generate": 0,
        }
        try:
            self.shop_pipeline.storge_data(shop_row)
        except Exception as e:
            logger.exception("店铺入库失败 storeId=%s: %s", store_id, e)
        # 店铺接口休息 1-2 秒
        time.sleep(random.uniform(1, 2))

    # ==================== 数据映射 ====================
    def _to_product(self, item):
        now = time.strftime("%Y-%m-%d %H:%M:%S")
        today = time.strftime("%Y-%m-%d")

        store_name = item.get("storeName") or ""
        store_full_name = item.get("storeFullName") or ""
        item_store_id = str(item.get("itemStoreId") or "")

        # 省份/城市/area_info 由 DrugPipeline.get_shop_city() 从 retrieve_scrape_shop_info 表补齐
        return {
            "platform": PLATFORM_ID,
            "item_id": item_store_id,
            "enterprise_id": self.company_id,
            "product_name": item.get("itemStoreName") or "",
            "spec": item.get("specs") or "",
            "one_price": compute_one_price(item),
            "detail_url": PRODUCT_LINK_TPL.format(item_store_id=item_store_id) if item_store_id else "",
            "shop_name": store_name,
            "anonymous_store_name": 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": float(item.get("sortPrice") or 0),
            "sales": item.get("totalCustCount") or "",
            "stock_count": item.get("itemStorage") or item.get("ownStorageNumber") or 0,
            "snapshot_url": "",  # API 模式无快照，快照模式后续用 Playwright 截图
            "approval_num": item.get("approvalNo") or "",
            "produced_time": "",
            "deadline": str(item.get("validateTimeEnd") or ""),
            "update_time": now,
            "insert_time": now,
            "number": int(item.get("middlePackageAmount") or 1),
            "product_brand": item.get("brandName") or self.product_brand,
            "collect_task_id": self.collect_task_id,
            "search_name": self.product_name,
            "company_name": store_full_name,
            "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 _match_item(self, product):
        product_brand = product.get("product_brand", "")
        product_name = product.get("product_name", "")
        product_specs = product.get("spec", "")

        brand_ok = True
        name_ok = True
        specs_ok = True
        if self.product_brand.strip():
            brand_ok = self.product_brand.strip() in str(product_brand)
        if self.product_name.strip():
            name_ok = self.product_name.strip() in str(product_name)
        if self.product_specs.strip():
            specs_ok = self.product_specs.strip() in str(product_specs)
        return brand_ok and name_ok and specs_ok

    def search_data(self):
        if not self.task_dict:
            return

        no_match_count = 0       # 连续不匹配计数
        MAX_NO_MATCH = 20        # 连续超过此数则停止（参考 main.py）
        pages = []               # 实际爬取过的页数

        for page in range(1, MAX_PAGES + 1):
            logger.info("药九九 爬取第 %s 页, 关键词=%s", page, self.keyword)

            try:
                items = self._search_page(page)
            except Exception as e:
                logger.error("第 %s 页请求失败: %s", page, e)
                self.is_success = False
                self._report_exception(5, f"第{page}页请求失败: {e}")
                return

            if not items:
                logger.info("第 %s 页无数据，停止翻页")
                break

            pages.append(page)
            self.page = page

            # 先补齐所有店铺信息
            for item in items:
                store_id = item.get("storeId")
                store_name = item.get("storeName") or ""
                self._ensure_shop_info(store_id, store_name)

            # 逐条入库
            for item in items:
                product = self._to_product(item)
                if not product.get("item_id"):
                    continue

                # 品牌/品名过滤
                if not self._match_item(product):
                    no_match_count += 1
                    logger.warning("不匹配(%s/%s): brand=%s name=%s",
                                   no_match_count, MAX_NO_MATCH,
                                   product.get("product_brand"), product.get("product_name"))
                    if no_match_count >= MAX_NO_MATCH:
                        logger.error("连续 %s 条不匹配，停止翻页", MAX_NO_MATCH)
                        # 回告当前页进度后再退出
                        self._report_page_progress(page, self.pipeline.crawl_count)
                        return
                    continue
                no_match_count = 0  # 匹配成功则重置

                try:
                    self.pipeline.storge_data(product)
                except Exception as e:
                    logger.exception("入库失败: %s", e)

            if self._report_page_progress(page, self.pipeline.crawl_count):
                break

            if len(items) < PAGE_SIZE:
                break

            # 翻页休息 2-3 秒，模拟真人
            time.sleep(random.uniform(2, 3))

        # 用实际爬取页数回告完成
        total_pages = len(pages)
        if total_pages > 0:
            self._report_page_progress(pages[-1], self.pipeline.crawl_count,
                                       is_finished=1, total_pages=total_pages)

    def run(self):
        try:
            self.search_data()
        except Exception as e:
            self.is_success = False
            logger.error("药九九爬虫异常: %s", e, exc_info=True)
            self._report_exception(5, f"未知异常: {e}")

        logger.info("药九九 爬取完成: %s 条, %s 页", self.pipeline.crawl_count, self.page)
        return self.pipeline.crawl_count, self.is_success, self.page