"""药九九搜索爬虫 — 通过 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