import base64 import hashlib import json import random import signal import socket import sys import time from pathlib import Path from urllib.parse import quote import requests from Crypto.Cipher import AES from DrissionPage import ChromiumPage, ChromiumOptions from commons.Logger import get_spider_logger from commons.scheduler import CrawlerScheduler from commons.config import YYC_ACCOUNT, YYC_DEVICE_ID from oss_upload.oss_upload import AliyunOSSUploader from pipelines.drug_pipelines import DrugPipeline from area_info.city_name_to_id import get_city from Crypto.Util.Padding import unpad logger = get_spider_logger("yaoex_snapshot") # 平台常量（API 调度用） PLATFORM_ID = 6 HEARTBEAT_INTERVAL_SECONDS = 60 TASK_API_BASE = "http://192.168.2.246:8080/api/collect_task" TASK_API_TOKEN = "zhijiayun_crawler_2026" TASK_REPORT_URL = TASK_API_BASE + "/report" CAPTCHA_TOKEN = "zPzmt1mG1ouCU6GTzsZN2Lmm8pdZypapPcLJTBRETco" CAPTCHA_API_URL = "http://api.jfbym.com/api/YmServer/customApi" chrome_path = r"C:\Program Files\Google\Chrome\Application\chrome.exe" # 项目根目录 → spiders/yaoex（与从哪执行脚本无关） PROJECT_ROOT = Path(__file__).resolve().parents[2] YAOEX_SPIDER_DIR = PROJECT_ROOT / "spiders" / "yaoex" SLIDER_OFFSET_FIX = 10 DETAIL_GET_TIMEOUT = 15 DETAIL_URL_WAIT = 10 DETAIL_DOM_WAIT = 8 DETAIL_NAV_RETRIES = 3 DETAIL_CONTENT_XPATH = "xpath://div[contains(@class,'yaoex-product-detail__content')]" REQUEST_RETRY_COUNT = 3 REQUEST_TIMEOUT_SEC = 20 NOT_PRODUCT_BREAK = 15 headers = { "Accept": "application/json, text/plain, */*", "Accept-Language": "zh-CN,zh;q=0.9", "Connection": "keep-alive", "Content-Type": "application/x-www-form-urlencoded", "Origin": "https://mall.yaoex.com", "Referer": "https://mall.yaoex.com/", "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "cross-site", "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" ), "X-Request-Agent": "Axios", "X-Requested-With": "XMLHttpRequest", "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"', } class YaoexSnapshotCrawl: def __init__(self, drug_dict=None): self.driver = None self.platform = 6 self.pipeline = DrugPipeline("yaoex") self.task_dict = drug_dict or {} self.ossuploader = AliyunOSSUploader() self.start_page = 1 self.end_page = 1 self.account_name = YYC_ACCOUNT.get("username", "yyc_default") self._shop_cache = {} self._register_signal_handler() if self.task_dict: self.get_product_data() self.success = True self.is_not_product = 0 self.user_id = YYC_ACCOUNT["user_id"] self.token = YYC_ACCOUNT["token"] self.scheduler = CrawlerScheduler(YYC_DEVICE_ID, str(PLATFORM_ID), heartbeat_interval=HEARTBEAT_INTERVAL_SECONDS) 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) self.start_page = self._parse_page(self.task_dict.get("start_page"), 1) self.end_page = self._parse_page(self.task_dict.get("end_page"), 100) # ---------- API report helpers ---------- def _send_report(self, params: dict, log_msg: str): try: resp = requests.post( TASK_REPORT_URL, json=params, timeout=10, headers={'X-Crawler-Token': TASK_API_TOKEN}, ) if resp.status_code == 200: try: data = resp.json() if data.get("code") == "success": logger.info("回告成功：%s", log_msg) return "success" else: msg = data.get("msg", "") if "已达平台限额" in msg or "限额" in msg: logger.warning("回告接口返回限额错误：code=%s, msg=%s", data.get('code'), msg) return "limit_reached" else: logger.warning("回告接口返回错误：code=%s, msg=%s", data.get('code'), msg) return "error" except ValueError: logger.warning("回告响应非 JSON：%s", resp.text[:200]) return "error" else: logger.warning("回告 HTTP %s", resp.status_code) return "error" except Exception as e: logger.error("回告失败：%s，错误：%s", log_msg, str(e)) return "error" def _report_start(self): payload = { "task_id": self.task_id, "platform": str(self.platform), "username": YYC_DEVICE_ID, "current_page": 0, "total_pages": 0, "is_finished": 0, } logger.info("上报开始 task_id=%s", self.task_id) return self._send_report(payload, "开始上报（is_finished=0）") def _report_progress(self, current_page: int, total_pages: int = 0): payload = { "task_id": self.task_id, "platform": str(self.platform), "username": YYC_DEVICE_ID, "current_page": current_page, "total_pages": total_pages, "is_finished": 0, } return self._send_report(payload, f"进度上报 page={current_page}") def _report_end(self, success: bool, real_count: int, page_no: int = 0, total_pages: int = 0): payload = { "task_id": self.task_id, "platform": str(self.platform), "username": YYC_DEVICE_ID, "current_page": page_no, "total_pages": total_pages if total_pages > 0 else page_no, "crawled_count": real_count, "is_finished": 1 if success else 0, } logger.info("上报结束 task_id=%s payload=%s", self.task_id, payload) self._send_report(payload, "结束上报（is_finished=1）") @staticmethod def _parse_page(value, default=1): try: page = int(value) return page if page >= 1 else default except (TypeError, ValueError): return default def _register_signal_handler(self): def handler(signum, frame): logger.info("收到退出信号，正在关闭浏览器...") self._quit_browser() sys.exit(0) signal.signal(signal.SIGINT, handler) if hasattr(signal, "SIGTERM"): signal.signal(signal.SIGTERM, handler) @staticmethod def _timestamp_ms() -> str: return str(int(time.time() * 1000)) def _quit_browser(self): if self.driver: try: self.driver.quit() except Exception: pass self.driver = None @staticmethod def _get_free_port(): with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.bind(("127.0.0.1", 0)) return s.getsockname()[1] def _resolve_browser_profile_dir(self): """浏览器数据目录: <项目根>/spiders/yaoex/<账号>""" profile_dir = YAOEX_SPIDER_DIR / self.account_name # 仅兼容历史误路径，新建不再使用 chrome_profile legacy_nested = YAOEX_SPIDER_DIR / "spiders" / "yaoex" / self.account_name legacy_chrome_profile = YAOEX_SPIDER_DIR / "chrome_profile" / self.account_name for candidate in (profile_dir, legacy_nested, legacy_chrome_profile): if (candidate / "Default").is_dir() or (candidate / "Local State").is_file(): logger.info("使用已有浏览器配置目录: %s", candidate) return candidate profile_dir.mkdir(parents=True, exist_ok=True) logger.info("新建浏览器配置目录: %s", profile_dir) return profile_dir def init_browser(self): co = ChromiumOptions().set_browser_path(chrome_path) debug_port = self._get_free_port() profile_dir = self._resolve_browser_profile_dir() profile_dir.mkdir(parents=True, exist_ok=True) co.set_user_data_path(str(profile_dir)) logger.info("浏览器用户目录(绝对路径): %s", profile_dir.resolve()) co.set_local_port(debug_port) co.set_argument(f"--remote-debugging-port={debug_port}") co.set_argument("--remote-debugging-address=127.0.0.1") co.set_argument("--disable-dev-shm-usage") co.set_argument("--start-maximized") co.set_argument("--no-first-run") co.set_argument("--no-default-browser-check") self.driver = ChromiumPage(co) def _is_logged_in(self): return bool(self.driver.ele("xpath=//a[@id='logout']", timeout=5)) def _call_captcha_api(self, image_bytes): try: b64 = base64.b64encode(image_bytes).decode() resp = requests.post( CAPTCHA_API_URL, json={"token": CAPTCHA_TOKEN, "type": "22222", "image": b64}, headers={"Content-Type": "application/json"}, timeout=15, ).json() logger.info("验证码 API 返回: %s", resp) return resp["data"]["data"] except Exception as e: logger.exception("验证码识别失败: %s", e) return None @staticmethod def _generate_human_track(distance): tracks = [] current = 0 mid = distance * 0.7 t = 0.2 v = 0 move_points = [] while current < mid: a = random.uniform(2, 4) v0 = v v = v0 + a * t move = v0 * t + 0.5 * a * t * t current += move move_points.append(move) while current < distance: a = -random.uniform(0.5, 1.5) v0 = v v = v0 + a * t if v < 0.5: v = 0.5 move = v0 * t + 0.5 * a * t * t current += move move_points.append(move) total_points = len(move_points) for i, move in enumerate(move_points): y_offset = random.randint(-2, 2) if i % random.randint(2, 4) == 0 else 0 if i < total_points * 0.3: duration = random.uniform(0.01, 0.03) elif i > total_points * 0.7: duration = random.uniform(0.03, 0.08) else: duration = random.uniform(0.02, 0.05) if random.random() < 0.05: duration += random.uniform(0.05, 0.1) tracks.append((move, y_offset, duration)) if random.random() < 0.7: tracks.append((-random.randint(1, 3), 0, 0.05)) return tracks def _simulate_slider_drag(self, slider_element, target_distance): self.driver.actions.move_to(slider_element).hold() for offset_x, offset_y, duration in self._generate_human_track(target_distance): self.driver.actions.move(offset_x, offset_y, duration=duration / 1000) self.driver.actions.release() def _solve_slider_if_present(self): modal = self.driver.ele("xpath://div[@class='yidun_modal']", timeout=3) if not modal: return True logger.info("检测到滑块验证码，开始处理") jpg_bytes = modal.get_screenshot(as_bytes="jpg") distance = self._call_captcha_api(jpg_bytes) if distance is None: return False slider = self.driver.ele( "xpath://div[contains(@class,'yidun_slider--hover')]", timeout=5 ) if not slider: logger.error("未找到滑块元素") return False self._simulate_slider_drag(slider, float(distance) + SLIDER_OFFSET_FIX) time.sleep(3) return True def login(self): self.driver.get("https://mall.yaoex.com/login", timeout=15) self.driver.wait.doc_loaded(timeout=10) input_name = self.driver.ele("xpath://input[@name='username']", timeout=5) if not input_name: logger.error("未找到用户名输入框") return False input_name.input(YYC_ACCOUNT["username"]) time.sleep(random.uniform(1.2, 2.0)) input_pass = self.driver.ele("xpath://input[@name='password']", timeout=5) if not input_pass: logger.error("未找到密码输入框") return False input_pass.input(YYC_ACCOUNT["password"]) time.sleep(random.uniform(1.2, 2.0)) geetest_click = self.driver.ele( "xpath=//div[contains(@class,'geetest_btn_click')]", timeout=3 ) if geetest_click: geetest_click.click() time.sleep(1.5) login_button = self.driver.ele("xpath://input[@id='login-btn']", timeout=5) if not login_button: logger.error("未找到登录按钮") return False login_button.click() self.driver.wait.doc_loaded(timeout=10) time.sleep(2) if not self._solve_slider_if_present(): return False return self._is_logged_in() def decrypt_price(self, ciphertext_b64): if not ciphertext_b64 or not str(ciphertext_b64).strip(): return "" _KEY_FIXED = "GDLSAUO1KUMIIBCE" if not self.user_id: key = _KEY_FIXED.encode("utf-8") else: uid = str(self.user_id)[:6].rjust(6, "0") key = (_KEY_FIXED[:10] + uid).encode("utf-8") raw = base64.b64decode(ciphertext_b64.strip()) cipher = AES.new(key, AES.MODE_ECB) plain = unpad(cipher.decrypt(raw), AES.block_size) return plain.decode("utf-8") def _post_with_retry(self, url, payload, retries=REQUEST_RETRY_COUNT, timeout=REQUEST_TIMEOUT_SEC): last_err = None for attempt in range(1, retries + 1): try: resp = requests.post( url, headers=headers, data=payload, timeout=timeout, ) resp.raise_for_status() return resp except Exception as e: last_err = e if attempt < retries: logger.warning("请求失败，第%s/%s次重试: %s", attempt, retries, e) time.sleep(min(2 * attempt, 5)) else: logger.error("请求失败，已达最大重试次数(%s): %s", retries, e) raise last_err def _shop_payload(self, enterprise_id): return { "traderName": "yaoex_pc", "trader": "pc", "closesignature": "yes", "signature_method": "md5", "signature": "****", "timestamp": self._timestamp_ms(), "token": self.token, "userToken": self.token, "enterpriseId": enterprise_id, } def _list_payload(self, keyword, page): return { "traderName": "yaoex_pc", "trader": "pc", "closesignature": "yes", "signature_method": "md5", "signature": "****", "timestamp": self._timestamp_ms(), "token": self.token, "userToken": self.token, "userId": self.user_id, "roleId": "101", "userType": "下游客户", "buyerCode": self.user_id, "nowPage": str(page), "per": "20", "keyword": keyword, "catSearchId": "", "specs": "", "factoryIds": "", "sellerCodes": "", "sellerFilterMode": "0", "sortColumn": "default", "sortMode": "default", "ver": "1", "stock_mode": "1", "showExtendCard": "true", "needDinnerPrice": "true", "limitStart": "", "limitEnd": "", "deadLineStart": "", "deadLineEnd": "", "filterDtos": "", "showWholePurchase": "true", } def fetch_list_page(self, keyword, page): list_url = "https://gateway-b2b.fangkuaiyi.com/home/search/homeSearchList" resp = self._post_with_retry(list_url, self._list_payload(keyword, page)) data = resp.json() recall_status = data.get("data", {}).get("recallStatus", 0) data_block = data.get("data", {}) or {} # 尝试提取总页数（常见字段名依次尝试） total_pages = 0 for key in ("totalPage", "totalPages", "pageCount", "total", "totalCount", "totalNum", "totalSize"): val = data_block.get(key) if val is not None and int(val) > 0: total_pages = int(val) break if total_pages: logger.info("📊 列表 API 检测到总页数：%s (page=%s)", total_pages, page) else: logger.debug("列表 API 未返回总页数字段，data keys: %s", list(data_block.keys())[:20]) if int(recall_status) == 1: return data_block.get("shopProducts", []) or [], total_pages else: return [], 0 def fetch_shop(self, seller_code): detail_url = "https://gateway-b2b.fangkuaiyi.com/ycapp/shop/enterpriseQualification" resp = self._post_with_retry(detail_url, self._shop_payload(seller_code)) shop_res = resp.json().get("data", {}) base_info = shop_res.get("baseInfo", {}) return base_info.get("address", ""), base_info.get("enterpriseName", "") def _get_shop_info(self, seller_code): if seller_code in self._shop_cache: return self._shop_cache[seller_code] try: shop_info = self.fetch_shop(seller_code) except Exception as e: logger.warning("fetch_shop 失败 seller_code=%s: %s", seller_code, e) shop_info = ("", "") self._shop_cache[seller_code] = shop_info return shop_info def _current_url(self): try: return self.driver.url or "" except Exception: return "" def _url_has_product(self, spu_code, seller_code): url = self._current_url() spu_code = str(spu_code or "") seller_code = str(seller_code or "") if spu_code and seller_code: return spu_code in url and seller_code in url return bool(spu_code and spu_code in url) def _wait_detail_ready(self, spu_code, seller_code, timeout=DETAIL_URL_WAIT): deadline = time.time() + timeout while time.time() < deadline: if self._url_has_product(spu_code, seller_code): if self.driver.ele(DETAIL_CONTENT_XPATH, timeout=1): time.sleep(0.3) return True time.sleep(0.4) return False def _build_detail_url(self, item): if not item.get("productId") and item.get("groupBuyProductDto"): item = item.get("groupBuyProductDto") or {} spu_code = item.get("spuCode", "") seller_code = item.get("sellerCode", "") group_buying_id = item.get("groupBuyingId", "") p_json = json.dumps( {"id": group_buying_id, "s": seller_code, "sp": spu_code}, separators=(",", ":"), ) detail_url = ( f"https://mall.yaoex.com/groupBuying/#/productDetail?p={quote(p_json)}" ) else: seller_code = item.get("sellerCode") spu_code = item.get("spuCode") detail_url = ( f"https://mall.yaoex.com/v2/product/#/spuCode/{spu_code}/sellerCode/{seller_code}" ) return item, detail_url, spu_code, seller_code def _goto_detail_page(self, detail_url, spu_code, seller_code): """get 后 refresh 一次，让 SPA 按当前 URL 重新渲染详情。""" for attempt in range(1, DETAIL_NAV_RETRIES + 1): try: self.driver.get(detail_url, timeout=DETAIL_GET_TIMEOUT) time.sleep(0.5) self.driver.refresh() time.sleep(2) ele = self.driver.ele("xpath=//div[@class='yaoex-product-detail__product-detail']") if ele: return True else: continue except Exception as e: logger.warning( "跳转详情异常 spu=%s seller=%s attempt=%s: %s", spu_code, seller_code, attempt, e, ) time.sleep(random.uniform(0.8, 1.5)) return False def _take_snapshot(self, upload_key): time.sleep(1) try: detail_ele = self.driver.ele(DETAIL_CONTENT_XPATH, timeout=2) if detail_ele: jpg_bytes = detail_ele.get_screenshot(as_bytes="jpg") else: jpg_bytes = self.driver.get_screenshot(as_bytes="jpg") if not jpg_bytes: logger.warning("截图为空 upload_key=%s", upload_key) return "" img_url = self.ossuploader.upload_from_bytes(jpg_bytes, str(upload_key)) except Exception: logger.exception("截图或 OSS 上传失败 upload_key=%s", upload_key) return "" if not img_url: logger.warning("OSS 未返回有效地址 upload_key=%s", upload_key) return "" logger.info("截图上传完成 upload_key=%s url=%s", upload_key, img_url) time.sleep(random.uniform(1, 2)) return img_url def parse_product(self, item, detail_url, snap_url): seller_code = item.get("sellerCode") spu_code = item.get("spuCode") name_part = (item.get("productName") or "").strip() short_part = (item.get("shortName") or "").strip() product_name = f"{name_part} {short_part}".strip() shop_url = f"https://mall.yaoex.com/v2/store/#/detail/{seller_code}/home" company_address, company_name = self._get_shop_info(seller_code) address = item.get("cityName", "") city_id = province_id = city = province = "" if address: city_id, province_id, city, province = get_city(address.split("市")[0]) price = self.decrypt_price(item.get("price")) hash_text = f"{seller_code}_{spu_code}_{price}" item_id = hashlib.md5(hash_text.encode("utf-8")).hexdigest() is_sold_out = 1 if "商品已售罄" in (item.get("statusDescription") or "") else 0 shop_name = item.get("storeName") or item.get("shopName") anonymous_store_name = "" if shop_name == "预约配送中心": anonymous_store_name = item.get("supplyName", "") inventory = item.get("currentInventory") or item.get("stockCount") now = time.strftime("%Y-%m-%d %H:%M:%S") return { "platform": self.platform, "item_id": item_id, "enterprise_id": self.company_id, "product_name": product_name, "spec": item.get("spec"), "one_price": "", "detail_url": detail_url, "shop_name": shop_name, "anonymous_store_name": anonymous_store_name, "shop_url": shop_url, "city_name": city, "city_id": city_id, "province_name": province, "province_id": province_id, "shipment_city_name": "", "shipment_city_id": "", "shipment_province_name": "", "shipment_province_id": "", "area_info": company_address or "", "factory_name": item.get("factoryName"), "scrape_date": time.strftime("%Y-%m-%d"), "price": price, "sales": "", "stock_count": inventory, "snapshot_url": snap_url, "approval_num": item.get("approvalNum"), "produced_time": item.get("productionTime"), "deadline": item.get("deadLine"), "update_time": now, "insert_time": now, "number": 1, "product_brand": self.brand or "", "collect_task_id": self.collect_task_id, "search_name": self.product, "company_name": 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": is_sold_out, } def search(self): self.driver.get("https://mall.yaoex.com/", timeout=15) self.driver.wait.doc_loaded(timeout=10) if not self._is_logged_in(): if not self.login(): logger.error("登录失败") return False keyword = self.product if self.brand: keyword = (self.brand + " " + self.product).strip() if self.product_desc: keyword = (keyword + " " + self.product_desc).strip() for page in range(self.start_page, self.end_page + 1): if self.scheduler.end: logger.warning("收到停止信号，终止采集") break logger.info("正在爬取 %s %s，第%s页", self.brand, self.product, page) page_items, api_total_pages = self.fetch_list_page(keyword=keyword, page=page) if not page_items: logger.info("第%s页无数据，停止", page) break for item in page_items: if self.scheduler.end: logger.warning("收到停止信号，终止当前页面采集") break item, detail_url, spu_code, seller_code = self._build_detail_url(item) name_part = (item.get("productName") or "").strip() short_part = (item.get("shortName") or "").strip() product_name = f"{name_part} {short_part}".strip() if self.product not in product_name: self.is_not_product += 1 continue if self.brand not in product_name: self.is_not_product += 1 continue self.is_not_product = 0 if not self._goto_detail_page(detail_url, spu_code, seller_code): logger.warning( "详情页跳转失败，跳过 spu=%s seller=%s url=%s", spu_code, seller_code, detail_url, ) continue upload_key = hashlib.md5(detail_url.encode("utf-8")).hexdigest() snap_url = self._take_snapshot(upload_key) product = self.parse_product(item, detail_url, snap_url) if not product.get("item_id"): continue try: self.pipeline.storge_data(product) logger.info("%s", json.dumps(product, ensure_ascii=False, default=str)) except Exception as e: logger.exception("写入数据库失败: %s", e) time.sleep(random.uniform(1, 2)) if self.is_not_product > NOT_PRODUCT_BREAK: logger.info("连续不匹配商品过多，停止搜索") break # 每页结束上报进度，检查限额 self._page_no = page prog_status = self._report_progress(page, total_pages=api_total_pages) if prog_status == "limit_reached": logger.warning("⛔ 进度上报检测到限额，通知停止") self.scheduler.set_flag(True) time.sleep(random.uniform(1, 3)) return True def run(self): self._page_no = 0 try: self.scheduler.start() self._report_start() self.init_browser() self.search() except Exception as e: logger.exception("运行异常: %s", e) self.success = False finally: self.scheduler.stop() try: self._report_end( success=self.success, real_count=self.pipeline.crawl_count, page_no=self._page_no, ) except Exception: pass self._quit_browser() logger.info("爬取总数%s", self.pipeline.crawl_count) return self.pipeline.crawl_count, self.success