craw_spider/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 MySQLPoolOnline
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.approval_num = ""
        self.task_dict = drug_dict or {}
        self.collect_task_id = None
        self.token = None
        self.account_name = "17097980383"
        self.pipeline = DrugPipeline("yaoshibang")
        if self.task_dict:
            self.get_product_data()
        self.is_success = True
        self.db_online = MySQLPoolOnline()
        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 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","")
        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": "",
            "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

        page = 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)
                return

            try:
                data_json = response.json()
            except json.JSONDecodeError:
                logger.exception("第%s页响应不是合法 JSON", page)
                self.is_success = False
                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
                    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
                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

    def run(self):
        if not self.get_token():
            logger.error("启动失败：无可用 token")
            return 0, False
        try:
            self.search_data()
        except Exception as e:

            self.is_success = False
            logger.error(e)

        logger.info(f"爬取总数{self.pipeline.crawl_count}")
        return self.pipeline.crawl_count, self.is_success