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'(?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