tob/spiders/yaoshibang/ysbang_crawl.py

import base64
import hashlib
import json
import random
import re
import secrets
import string
import time
import token
import zlib
from datetime import datetime, timedelta
import requests
from Crypto.Cipher import AES
from openpyxl.worksheet import page

from commons.Logger import get_spider_logger
from pipelines.drug_pipelines import DrugPipeline
from area_info.city_name_to_id import get_city
from commons.conn_mysql import MySQLPoolOn2
from spiders.yaoshibang.login_yaoshibang import YaoShiBangLogin

logger = get_spider_logger("yaoshibang")


class YsbSpider:
    def __init__(self, drug_dict=None):
        self.url = "https://dian.ysbang.cn/wholesale-drug/sales/getWholesaleList/v4270"
        self.headers = self.build_headers()
        self.start_date = (datetime.now() - timedelta(minutes=500)).strftime("%Y-%m-%d %H:%M")
        self.platform = 5
        self.page = 0
        self.approval_num = ""
        self.task_dict = drug_dict or {}
        self.collect_task_id = None
        self.token = None
        self.account_name = "x"
        self.pipeline = DrugPipeline("yaoshibang")
        if self.task_dict:
            self.get_product_data()
        self.is_success = True
        self.db_online = MySQLPoolOn2()
        self.is_product_count = 0

    def get_token(self, _retry_login=False):
        sql_account = f""" select `name`,`cookie_str` from `accounts_platform` where `platform`=5 and `status`=1 and `equipment_id`=3 order by `cookie_timestamp` asc limit 1 """
        account_list = self.db_online.select_data(sql_account)
        if not account_list:
            logger.error("无可用爬取账号")
            return None
        account_dict = account_list[0]
        self.account_name = account_dict["name"]
        cookie_str = account_dict.get("cookie_str") or ""
        if not cookie_str:
            if _retry_login:
                logger.error("账号 %s 登录后 cookie 仍为空", self.account_name)
                return None
            logger.warning("账号 %s cookie_str 为空，尝试登录", self.account_name)
            YaoShiBangLogin().run()
            time.sleep(5)
            return self.get_token(_retry_login=True)
        try:
            cookie_dict = json.loads(cookie_str)
        except json.JSONDecodeError:
            logger.exception("账号 %s cookie_str 不是合法 JSON", self.account_name)
            return None
        token = cookie_dict.get("Token") or cookie_dict.get("token")
        if not token:
            logger.error("账号 %s cookie 中无 Token 字段: %s", self.account_name, list(cookie_dict.keys()))
            return None
        self.token = token
        logger.info("已刷新 token，账号=%s", self.account_name)
        return self.token

    def get_product_data(self):
        self.task_id = self.task_dict["id"]
        self.company_id = self.task_dict["company_id"]
        self.product = self.task_dict["product_name"]
        self.product_desc = self.task_dict.get("product_specs", "")
        self.brand = self.task_dict.get("product_brand", "")
        self.product_keyword = self.task_dict.get("product_keyword", "")
        self.collect_task_id = self.task_dict.get("collect_task_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.collect_equipment_id = self.task_dict.get("collect_equipment_id", "")
        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)

    def _report_exception(self, exception_type, remark=""):
        """异常回告，上报 exception_type (1-6)"""
        scheduler = self.task_dict.get("scheduler")
        if scheduler:
            data = {
                "task_id": self.task_id,
                "platform": self.task_dict.get("_platform", str(self.platform)),
                "username": self.task_dict.get("_device_id", ""),
                "current_page": self.page,
                "is_finished": 0,
                "exception_type": exception_type,
                "remark": remark,
            }
            if self.page:
                data["total_pages"] = self.page
            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):
        """逐页回告进度，返回 True=code=error 需停止"""
        scheduler = self.task_dict.get("scheduler")
        if not scheduler:
            return False
        data = {
            "task_id": self.task_id,
            "platform": self.task_dict.get("_platform", str(self.platform)),
            "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

    def pkcs7_unpad(self, data):
        if not data:
            raise ValueError("Empty data for PKCS7 unpad")
        pad_len = data[-1]
        if pad_len < 1 or pad_len > 16:
            raise ValueError("Invalid PKCS7 padding length")
        if data[-pad_len:] != bytes([pad_len]) * pad_len:
            raise ValueError("Invalid PKCS7 padding bytes")
        return data[:-pad_len]

    def derive_key(self):
        base = "BhCLxFfFhd12K4qRGPfy"
        md5_hex = hashlib.md5(base.encode("utf-8")).hexdigest()
        return md5_hex[:16].upper().encode("utf-8")

    def decrypt_payload(self, cipher_text_b64):
        key = self.derive_key()
        cipher_bytes = base64.b64decode(cipher_text_b64)
        cipher = AES.new(key, AES.MODE_ECB)
        decrypted = cipher.decrypt(cipher_bytes)
        unpadded = self.pkcs7_unpad(decrypted)
        json_bytes = zlib.decompress(unpadded, zlib.MAX_WBITS | 16)
        return json.loads(json_bytes.decode("utf-8"))

    def gen_pair(self, ex1_len=9, o_raw_len=16):
        alphabet = string.ascii_lowercase + string.digits
        ex1 = "".join(secrets.choice(alphabet) for _ in range(ex1_len))
        o = base64.b64encode(secrets.token_bytes(o_raw_len)).decode("ascii")
        return {"ex1": ex1, "o": o}

    def build_headers(self):
        return {
            "Accept": "*/*",
            "Accept-Language": "zh-CN,zh;q=0.9",
            "Connection": "keep-alive",
            "Content-Type": "application/json",
            "Origin": "https://dian.ysbang.cn",
            "Referer": "https://dian.ysbang.cn/",
            "Sec-Fetch-Dest": "empty",
            "Sec-Fetch-Mode": "cors",
            "Sec-Fetch-Site": "same-origin",
            "User-Agent": (
                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
                "(KHTML, like Gecko) Chrome/146.0.0.0 Safari/537.36"
            ),
            "sec-ch-ua": '"Chromium";v="146", "Not-A.Brand";v="24", "Google Chrome";v="146"',
            "sec-ch-ua-mobile": "?0",
            "sec-ch-ua-platform": '"Windows"',
        }

    def build_base_payload(self):

        keyword = self.product
        if self.brand:
            keyword = self.brand + " " + self.product
        if self.product_desc:
            keyword = keyword + self.product_desc

        date_str = time.strftime("%Y-%m-%d %H:%M")
        json_data = {
            'platform': 'pc',
            'version': '6.1.10',
            'ua': 'Chrome148',
            'ex': f'{date_str} https://dian.ysbang.cn os=Windows 10 indexContent 05-25 11:20:53 05-27 17:27:01',
            'trafficType': 1,
            'ex1': '',
            'o': '',
            'lastClick': -1,
            'page': page,
            'pagesize': '60',
            'classify_id': '',
            'searchkey': keyword,
            'onlyTcm': 0,
            'operationtype': 1,
            'qualifiedLoanee': 0,
            'drugId': -1,
            'tagId': '',
            'showRecentlyPurchasedFlag': True,
            'onlySimpleLoan': 0,
            'sn': '',
            'buttons': [],
            'buttonList': [],
            'synonymId': 0,
            'activityTypes': [],
            'provider_filter': '',
            'factoryNames': '',
            'tcmGradeNames': [],
            'tcmExeStandardIds': [],
            'specs': '',
            'deliverFloor': 0,
            'purchaseLimitFloor': 0,
            'nextRequestKey': '',
            'adConfigId': 0,
            'stateValue': '',
            'filterLeyoProvider': False,
            'firstSearch': False,
            'token': self.token,
        }
        return json_data

    def get_price(self, price_token):
        pattern = re.compile(r'(?<!\d)(\d+\.\d{2})(?!\d)')
        decoded = base64.b64decode(price_token)
        text_part = decoded.decode('utf-8', errors='ignore')
        numbers_from_text = pattern.findall(text_part)
        unique_prices = list(set(round(float(a), 2) for a in numbers_from_text))
        if not unique_prices:
            return ""
        last_prices = sorted(unique_prices, reverse=True)
        return last_prices[-1]

    def to_product(self, item, type_data):

        now = time.strftime("%Y-%m-%d %H:%M:%S")
        item_id = item.get("wholesaleid", "")
        provider_id = item.get("providerId", "")

        shop_url = f"https://dian.ysbang.cn/#/supplierstore?providerId={provider_id}&trafficType=4",

        city_str = item.get("warehouseCity", "")
        city_id = province_id = city = province = ""
        price = item.get("disPrice", "")

        if not price:
            price = item.get("minprice", "")
        if not price:
            price = item.get("price", "")

        if not price:
            price_token = item.get("priceToken", "")
            if price_token:
                price = self.get_price(price_token)
        if not price:
            city_str, price = self.parse_detail(item_id, type_data)

        if city_str:
            city_id, province_id, city, province = get_city(city_str)
        shop_name = item.get("provider_name", "")
        if not shop_name:
            shop_name = item.get("abbreviation", "")

        brand = item.get("brand","")

        # 销量: 优先 order_amount > orderAmountPlus > alreadysales
        sales = (
            item.get("order_amount")
            or item.get("orderAmountPlus")
            or item.get("alreadysales")
            or ""
        )

        product = {
            "platform": self.platform,
            "item_id": item_id,
            "enterprise_id": self.company_id,
            "product_name": item.get("drugname", ""),
            "spec": item.get("specification", ""),
            "one_price": '',
            "detail_url": f"https://dian.ysbang.cn/#/drugInfo?wholesaleid={item_id}&trafficType=1",
            "shop_name": shop_name,
            "anonymous_store_name": "",
            "shop_url": f"https://dian.ysbang.cn/#/supplierstore?providerId={provider_id}&trafficType=4",
            "city_name": city,
            "city_id": city_id,
            "province_name": province,
            "province_id": province_id,
            "area_info": "",
            "factory_name": item.get("manufacturer", ""),
            "scrape_date": time.strftime("%Y-%m-%d"),
            "price": price,
            "sales": sales,
            "stock_count": item.get("stockAvailable", ""),
            "snapshot_url": "",
            "approval_num": self.approval_num,
            "produced_time": item.get("prodDate", ""),
            "deadline": item.get("valid_date", ""),
            "update_time": now,
            "insert_time": now,
            "number": 1,
            "product_brand": brand,
            "collect_task_id": self.collect_task_id,
            "search_name": self.product,
            "company_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}),
            "account_id": self.account_id,
            "collect_region_id": self.collect_region_id,
            "collect_round": self.collect_round,
            "is_sold_out": 0
        }
        return product

    def parse_detail(self, product_id, type_data):

        date_str = time.strftime("%Y-%m-%d %H:%M:%S")

        json_data = {
            'platform': 'pc',
            'version': '6.0.0',
            'ua': 'Chrome146',
            'ex': '{} drugInfo {} {}'.format(self.start_date, date_str, date_str),
            'trafficType': 1,
            'ex1': 'qtrcqlxew',
            'wholesaleid': str(product_id),
            'showRecentlyPurchasedFlag': True,
            'isClinic': 0,
            'scene': [1, ],
            'adConfigId': 0,
            'token': self.token,
        }

        if type_data in [7]:
            json_data["wholesaleId"] = str(product_id)
            url = "https://dian.ysbang.cn/wholesale-drug/api/teambuy/getActivityDetail/v4260"
        else:
            json_data['wholesaleid'] = str(product_id)
            url = 'https://dian.ysbang.cn/wholesale-drug/sales/getPreferenceDetail/v5280'
        time.sleep(random.uniform(1, 3))

        response = requests.post(url, headers=self.headers, json=json_data)

        data_json = response.json()
        data = data_json.get("data", {})
        self.approval_num = data.get("druginfo", {}).get("approval", "")
        if not self.approval_num:
            team_info = (data.get("teamBuyDetailDrugInfo") or {}).get(
                "teamBuyDetailSingleSaleTypeDrugInfo"
            ) or {}
            self.approval_num = team_info.get("approval", "") or ""
        city_data = data.get("delivery_policy") or ""
        city_re = re.search(r"\[(\w+)\w+\]", str(city_data))
        price_dict = data.get("disPriceInfo", {})

        if not price_dict:
            price_dict = data.get("teamBuyDetailInfo", {})
        price = price_dict.get("disPrice", "")
        if not price:
            price = price_dict.get("minprice", "")
        if not price:
            price = data.get("price", "")
        if city_re:
            city_str = city_re.group(1)
        else:
            city_str = ""
        return city_str, price

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

        # 支持续传：从 task_dict 的 current_page（>0）或 start_page 开始
        page = int(self.task_dict.get("current_page", 0) or 0)
        if page <= 0:
            page = int(self.task_dict.get("start_page", 1) or 1)
        while page < 100:
            logger.info(f"药师帮爬取第{page}页")
            pair = self.gen_pair()
            payload = self.build_base_payload()
            payload["ex1"] = pair["ex1"]
            payload["o"] = pair["o"]
            payload["page"] = page

            response = None
            for attempt in range(3):
                try:
                    response = requests.post(
                        self.url, headers=self.headers, json=payload, timeout=30
                    )
                    if response.status_code == 200:
                        break
                except Exception as e:
                    logger.error("第%s页请求失败 (%s/3): %s", page, attempt + 1, e)
                    response = None
                    time.sleep(10)
            if not response or response.status_code != 200:
                self.is_success = False
                logger.error("第%s页请求失败，停止爬取", page)
                self._report_exception(5, f"第{page}页请求失败(status={response.status_code if response else 'None'})")
                return

            try:
                data_json = response.json()
            except json.JSONDecodeError:
                logger.exception("第%s页响应不是合法 JSON", page)
                self.is_success = False
                self._report_exception(5, f"第{page}页响应非合法JSON")
                return

            data_block = data_json.get("data") or {}
            if data_json.get("message", "") in ["该操作需要登录","需要前端行为验证！"] :
                logger.info("登录账号中。。。")
                YaoShiBangLogin().run()
                time.sleep(10)
                if not self.get_token():
                    logger.error("登录后仍未从库中读到有效 Token，停止重试")
                    self.is_success = False
                    self._report_exception(1, "账号登录过期，重登后仍无有效Token")
                    return
                logger.info("token 已刷新，重试第 %s 页", page)
                continue

            encrypted_o = data_block.get("o")
            if not encrypted_o:
                logger.warning("第%s页返回无加密 data.o: %s", page, data_json)
                break

            try:
                json_data = self.decrypt_payload(encrypted_o)
            except Exception as e:
                logger.exception("第%s页解密失败: %s", page, e)
                self.is_success = False
                self._report_exception(6, f"第{page}页数据解密失败")
                return
            wholesales = json_data.get("wholesales", [])
            if not wholesales:
                logger.info(f"第{page}页无数据，停止")
                break
            # 获取国药准字
            for item in wholesales[0:5]:
                product_id = item.get("wholesaleid", "")
                type_data = item.get("activitytype", 0)
                self.parse_detail(product_id, type_data)
                if self.approval_num:
                    break

            for item in wholesales:
                type_data = item.get("activitytype", 0)
                product = self.to_product(item, type_data)
                if not product.get("item_id"):
                    continue
                title = product.get("product_name","")
                if self.brand not in title:
                    self.is_product_count +=1
                if self.product not in title:
                    self.is_product_count +=1
                    continue
                if self.product in title and self.brand in title:
                    self.is_product_count = 0

                try:
                    self.pipeline.storge_data(product)
                    logger.info("%s", json.dumps(product, ensure_ascii=False))
                except Exception as e:
                    logger.exception("写入数据库失败: %s", e)
            if self.is_product_count >= 20:
                return

            page += 1
            self.page = page
            # 逐页回告进度
            if self._report_page_progress(page, self.pipeline.crawl_count) == True:
                break



    def run(self):
        if not self.get_token():
            logger.error("启动失败：无可用 token")
            self._report_exception(1, "无可用账号或Token")
            return 0, False
        try:
            self.search_data()
        except Exception as e:
            self.is_success = False
            logger.error(e)
            self._report_exception(5, f"未知异常: {e}")

        logger.info(f"爬取总数{self.pipeline.crawl_count}爬取页数{self.page}")


        return self.pipeline.crawl_count, self.is_success ,self.page