"""药九九快照模式 — 一个浏览器从头到尾：检查登录 → 搜索 → 监听API → 匹配DOM → 截卡片""" import json import random import time import requests from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeoutError from commons.Logger import get_spider_logger from commons.conn_mysql import MySQLPoolOn2 from pipelines.drug_pipelines import DrugPipeline from oss_upload.oss_upload import AliyunOSSUploader from spiders.yaojiujiu.yyjzt_crawl import ( PLATFORM_ID, PAGE_SIZE, MAX_PAGES, SEARCH_API, PRODUCT_LINK_TPL, compute_one_price, generate_sign, ) logger = get_spider_logger("yaojiujiu_snapshot") import urllib.parse SEARCH_PAGE_URL = "https://www.yyjzt.com/search?keyword={keyword}" LOGIN_URL = "https://www.yyjzt.com/login" HOME_URL = "https://www.yyjzt.com" # 参考 main.py 的反爬/弹窗处理 USERNAME_SELECTOR = "input[name='username']" PASSWORD_SELECTOR = "input[name='password']" LOGIN_BTN_SELECTOR = "button:has-text('登录')" class YyjztSnapshotSpider: def __init__(self, task_dict=None): self.task_dict = task_dict or {} self.pipeline = DrugPipeline("yaojiujiu_snapshot") self.db = MySQLPoolOn2() self.is_success = True # 任务参数 self.task_id = "" self.company_id = "" self.product_name = "" self.product_specs = "" self.product_brand = "" self.keyword = "" self.collect_task_id = "" self.collect_round = "1" self.account_id = "" self.collect_region_id = "" self.sampling_cycle = "" self.sampling_start_time = "" self.sampling_end_time = "" # 账号 self._phone = "" self._password = "" if 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.collect_task_id = self.task_dict.get("collect_task_id", "") self.collect_round = self.task_dict.get("collect_round", "1") self.account_id = self.task_dict.get("collect_equipment_account_id", "") self.collect_region_id = self.task_dict.get("collect_region_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._phone = self.task_dict.get("account_phone", "") self._password = self.task_dict.get("account_password", "") # ============ 浏览器启动 + 登录 ============ def _get_account(self): sql = """SELECT `phone`, `password`, `cookie_str` FROM `accounts_platform` WHERE `platform`=%s AND `status`=1 ORDER BY `cookie_timestamp` ASC LIMIT 1""" rows = self.db.select_data(sql, (PLATFORM_ID,)) return rows[0] if rows else None def _is_login(self, page): """用 API 请求验证 token 是否真正有效""" try: page.goto(HOME_URL, timeout=15000) page.wait_for_load_state("networkidle") time.sleep(2) url = page.url.lower() logger.info("检测登录态: URL=%s", url) # 快速判断：URL 跳到登录页直接返回 False if "login" in url: logger.info("URL 含 login，未登录") return False # 提取 token cookies = page.context.cookies() token = "" for c in cookies: if c["name"] == "yjj-token": token = c["value"].replace("ssr_", "") break if not token: logger.info("cookie 中无 yjj-token，未登录") return False # 调搜索 API 验证 token logger.info("用 API 验证 token: %s...", token[:16]) body = {"pageIndex": 1, "pageSize": 1, "platformType": 1, "keyword": "999感冒灵", "searchNum": 0, "showSearchTips": 0, "isNeedAggregation": 0} ts = int(time.time() * 1000) headers = { "Content-Type": "application/json", "Origin": "https://www.yyjzt.com", "sy_plat": "B2B", "token_platform_client_type": "USER", "zhcaiToken": token, "deviceId": self._phone or "snapshot", "_s": generate_sign(token, body, ts), "_t": str(ts), } resp = requests.post(SEARCH_API, json=body, headers=headers, timeout=10) data = resp.json() code = data.get("code") msg = data.get("message") or data.get("msg") or "" logger.info("API 返回: code=%s msg=%s", code, msg) if code == 200: logger.info("API 验证通过，已登录") return True elif code == 401: logger.info("API 返回 401，未登录/token 过期") return False else: logger.warning("API 返回异常 code=%s，视为未登录", code) return False except Exception as e: logger.warning("检测登录态异常: %s", e) return False def _login_operation(self, page, phone, password, skip_goto=False): """参考 main.py login_operation()""" try: if not skip_goto: page.goto(LOGIN_URL, timeout=15000) page.wait_for_load_state("networkidle") time.sleep(2) # 切到密码登录 tab pwd_tab = page.locator("#tab-first") if pwd_tab.count() > 0: pwd_tab.click() page.wait_for_timeout(500) # 填手机号 page.fill(USERNAME_SELECTOR, "") for ch in str(phone): page.locator(USERNAME_SELECTOR).type(ch, delay=random.randint(60, 180)) logger.info("已输入账号") # 填密码 page.fill(PASSWORD_SELECTOR, "") for ch in str(password): page.locator(PASSWORD_SELECTOR).type(ch, delay=random.randint(60, 180)) logger.info("已输入密码") # 点登录 page.click(LOGIN_BTN_SELECTOR) logger.info("已点击登录") time.sleep(5) # 调 API 确认登录是否成功 cookies = page.context.cookies() token = "" for c in cookies: if c["name"] == "yjj-token": token = c["value"].replace("ssr_", "") break if not token: logger.error("登录后 cookie 中无 yjj-token") return False body = {"pageIndex": 1, "pageSize": 1, "platformType": 1, "keyword": "999感冒灵", "searchNum": 0, "showSearchTips": 0, "isNeedAggregation": 0} ts = int(time.time() * 1000) headers = { "Content-Type": "application/json", "Origin": "https://www.yyjzt.com", "sy_plat": "B2B", "token_platform_client_type": "USER", "zhcaiToken": token, "deviceId": str(phone), "_s": generate_sign(token, body, ts), "_t": str(ts), } resp = requests.post(SEARCH_API, json=body, headers=headers, timeout=10) data = resp.json() logger.info("登录验证 API: %s", json.dumps(data, ensure_ascii=False)) if data.get("success") and data.get("code") == 200: logger.info("登录成功") return True logger.error("登录失败，API 返回异常") return False except PlaywrightTimeoutError as e: logger.error("登录超时: %s", e) return False def _kill_popups(self, page): """参考 main.py: 先 JS 暴力清遮罩，再点按钮，三层防线不崩""" # 第一层：JS 先暴力移除遮罩 + 滑块验证 + 恢复滚动（main.py kill_masks） page.evaluate(""" () => { // 精准删已知遮罩 + 滑块验证 const selectors = [ '.v-modal', '.el-overlay', '.el-overlay-dialog', '.el-dialog__wrapper', '.el-message-box__wrapper', '.el-loading-mask', '.safety-slider-box', '[class*="safety"]', '[class*="slider"]', '[class*="captcha"]', '.el-popup-parent--hidden' ]; selectors.forEach(s => { document.querySelectorAll(s).forEach(e => { try { e.remove(); } catch(_) {} }); }); // 泛化兜底：近似全屏 + 高 z-index 的层强制隐藏 const all = Array.from(document.querySelectorAll('body *')); for (const el of all) { try { const s = getComputedStyle(el); const z = parseInt(s.zIndex || '0', 10); if ((s.position === 'fixed' || s.position === 'absolute') && z >= 1000 && s.pointerEvents !== 'none') { const r = el.getBoundingClientRect(); const nearFull = r.width >= innerWidth * 0.7 && r.height >= innerHeight * 0.7; if (nearFull) { el.style.pointerEvents = 'none'; el.style.display = 'none'; } } } catch(_) {} } // 恢复滚动 document.documentElement.style.overflow = 'auto'; document.body.style.overflow = 'auto'; document.body.style.position = 'static'; document.body.style.width = 'auto'; document.body.classList.remove('el-popup-parent--hidden'); } """) page.wait_for_timeout(300) # 第二层：连续点引导/关闭按钮（兼容多步引导），try 包住不崩 for _ in range(6): try: btn = page.locator( "xpath=//button[normalize-space()='下一步' or normalize-space()='完成' " "or normalize-space()='我知道了' or normalize-space()='关闭']" ).first if btn.count() > 0 and btn.is_visible(): btn.click(timeout=1500) page.wait_for_timeout(300) continue except Exception: pass try: close_icon = page.locator( ".el-dialog__headerbtn, .el-message-box__headerbtn, " ".close, .icon-close, .el-icon-close" ).first if close_icon.count() > 0 and close_icon.is_visible(): close_icon.click(timeout=1200, force=True) page.wait_for_timeout(300) continue except Exception: pass break # ============ API 搜索 ============ def _api_search_all(self): """API 搜全量，返回 [(item_dict, page), ...]""" all_items = [] for p in range(1, MAX_PAGES + 1): body = {"pageIndex": p, "pageSize": PAGE_SIZE, "platformType": 1, "keyword": self.keyword, "searchNum": 0, "showSearchTips": 0, "isNeedAggregation": 0} ts = int(time.time() * 1000) # 用 browser 当前 cookie 里的 token headers = { "Content-Type": "application/json", "Origin": "https://www.yyjzt.com", "sy_plat": "B2B", "token_platform_client_type": "USER", "zhcaiToken": self._token, "deviceId": self._phone, "_s": generate_sign(self._token, body, ts), "_t": str(ts), } try: resp = requests.post(SEARCH_API, json=body, headers=headers, timeout=30) data = resp.json() if data.get("success") and data.get("code") == 200: items = data["data"]["itemList"] for item in items: all_items.append((item, p)) total_count = int(data["data"].get("totalCount") or 0) if total_count > 0 and len(all_items) >= total_count: break if len(items) == 0: break else: logger.error("API page %s: %s", p, data.get("msg")) break except Exception as e: logger.error("API page %s: %s", p, e) break time.sleep(random.uniform(1, 2)) logger.info("API 共 %s 条", len(all_items)) return all_items # ============ DOM 匹配 ============ def _extract_dom_cards(self, page): """逐张卡片滚到可视区，触发懒加载后再取 innerText""" cards = [] card_els = page.locator(".goods_card") cnt = card_els.count() if cnt == 0: logger.warning("页面无 .goods_card 元素") return cards # 先滚回顶部 page.evaluate("window.scrollTo(0, 0)") page.wait_for_timeout(300) # 预估每张卡高度，用于判断是否需要继续滚 first_card_height = 0 try: box = card_els.nth(0).bounding_box() if box: first_card_height = box["height"] except Exception: pass # 每屏大约能显示多少张卡（假设卡高约 260px，视口 900px → ~3排 = 随实际布局调整） cards_per_screen = max(2, int(900 / max(first_card_height, 200))) if first_card_height else 3 for i in range(cnt): try: card = card_els.nth(i) # 滚到这张卡中间位置，确保它进入可视区 card.scroll_into_view_if_needed() page.wait_for_timeout(300) # 给懒加载时间 all_text = card.inner_text(timeout=2000).strip() if all_text: cards.append({"all_text": all_text, "locator": card, "index": i}) else: # 偶尔 inner_text 第一次拿不到（渲染慢），重试一次 page.wait_for_timeout(500) all_text = card.inner_text(timeout=2000).strip() if all_text: cards.append({"all_text": all_text, "locator": card, "index": i}) else: logger.warning("card[%s] 文本为空，可能的结构异常", i) except Exception as e: logger.warning("card[%s] 提取失败: %s", i, e) # 每屏结束时额外等一下，防止滚太快 if (i + 1) % cards_per_screen == 0: page.wait_for_timeout(400) logger.info("逐卡提取: %s/%s 个卡片有文本", len(cards), cnt) return cards @staticmethod def _price_variants(value): """同一个价格的不同字符串形态：15.0 → {"15.00","15.0","15"}""" vs = set() vs.add(f"{value:.2f}") vs.add(str(value)) vs.add(f"{value:.2f}".rstrip("0").rstrip(".")) vs.add(f"{value:.1f}".rstrip("0").rstrip(".")) return vs def _match(self, api_item, dom_cards, used): api_name = api_item.get("itemStoreName", "") api_store = api_item.get("storeName", "") api_specs = api_item.get("specs", "") api_price = compute_one_price(api_item) # sortPrice 是页面实际展示价（折后/拼团后），compute_one_price 是计算价， # DOM 可能显示其中任意一个，两者都纳入匹配池 api_sort_price = float(api_item.get("sortPrice") or 0) # 价格变体：compute_one_price + sortPrice 的所有字符串形态 price_pool = self._price_variants(api_price) if api_sort_price and abs(api_sort_price - api_price) > 0.001: price_pool |= self._price_variants(api_sort_price) fail_reasons = [] for c in dom_cards: if c["index"] in used: continue txt = c["all_text"] name_ok = bool(api_name) and (api_name in txt) store_ok = bool(api_store) and (api_store in txt) price_ok = any(p in txt for p in price_pool) # specs 辅助消歧：同名+同店+同价时，用规格区分 specs_ok = (not api_specs) or (api_specs in txt) if name_ok and store_ok and price_ok and specs_ok: return c, "" fail_reasons.append( f" card[{c['index']}]: name={name_ok} store={store_ok} price={price_ok} specs={specs_ok} " f"card_text={txt[:80]}" ) # 没匹配上，返回原因 detail = ( f"API: name={api_name!r} store={api_store!r} price={api_price}" f" sortPrice={api_sort_price} specs={api_specs!r} pool={price_pool}\n" + "\n".join(fail_reasons[:5]) ) return None, detail # ============ 翻页 ============ def _goto_next_page(self, page): """点击下一页按钮（参考 main.py goto_next_page），成功返回 True""" candidates = [ ".el-pagination button.btn-next:not(.is-disabled)", ".el-pagination__next:not(.is-disabled)", "button:has-text('下一页'):not([disabled])", "a:has-text('下一页')", ] for sel in candidates: btn = page.locator(sel).first if btn.count() > 0: try: page.evaluate("window.scrollTo(0, 0);") btn.click(timeout=5000) page.wait_for_load_state("networkidle") return True except Exception: continue return False # ============ 截图+入库 ============ def _process_page(self, page, items, page_no): self._kill_popups(page) if page_no == 1: # 第一页：URL 打开搜索 url = SEARCH_PAGE_URL.format(keyword=urllib.parse.quote(self.keyword)) page.goto(url, timeout=20000) else: # 后续页：点击"下一页"按钮（URL 翻页不生效，Vue 前端必须点按钮） if not self._goto_next_page(page): logger.warning("第 %s 页翻页失败，停止", page_no) return False page.wait_for_load_state("networkidle") time.sleep(random.uniform(1, 2)) self._kill_popups(page) # 页面加载完再清一次，防止延迟弹窗 # 截个整页图方便排查 page.screenshot(path=f"./snapshots/full_page_{page_no}.png", full_page=True) dom_cards = self._extract_dom_cards(page) logger.info("页%s: DOM=%s API=%s", page_no, len(dom_cards), len(items)) import os os.makedirs("./snapshots", exist_ok=True) os.makedirs("./snapshots/unmatched", exist_ok=True) os.makedirs("./snapshots/debug", exist_ok=True) # 先把所有 DOM 卡片截小图存 debug，方便排查 for c in dom_cards[:5]: # 只截前5个看结构 try: jpg = c["locator"].screenshot(type="jpeg", quality=60) fname = f"debug_card_{page_no}_{c['index']}.jpg" with open(os.path.join("./snapshots/debug", fname), "wb") as f: f.write(jpg) except Exception: pass used = set() matched = 0 unmatched_log = [] for item, _ in items: if self.product_brand and self.product_brand.strip() not in str(item.get("brandName") or ""): continue if self.product_name and self.product_name.strip() not in str(item.get("itemStoreName") or ""): continue if self.product_specs and self.product_specs.strip() not in str(item.get("specs") or ""): continue card, fail_detail = self._match(item, dom_cards, used) if not card: unmatched_log.append(fail_detail) continue used.add(card["index"]) try: card["locator"].scroll_into_view_if_needed() except Exception: pass time.sleep(0.3) # 截图 try: jpg = card["locator"].screenshot(type="jpeg", quality=80) except Exception as e: logger.warning("截图失败 item=%s: %s", item.get("itemStoreId"), e) continue # 上传 OSS fname = f"yjj_{item.get('itemStoreId','?')}_{page_no}_{card['index']}" try: snapshot_url = self.uploader.upload_from_bytes(jpg, fname) except Exception as e: logger.warning("OSS上传失败: %s, 存本地", e) local_path = os.path.join("./snapshots", fname + ".jpg") with open(local_path, "wb") as f: f.write(jpg) snapshot_url = local_path # 入库 product = self._to_product(item, snapshot_url) try: self.pipeline.storge_data(product) matched += 1 except Exception as e: logger.exception("入库失败: %s", e) # 同时写文件 self._all_results.append(product) # 未匹配的截全卡片图存 unmatched，同时写日志 skipped = len(items) - matched if unmatched_log: log_path = os.path.join("./snapshots", f"unmatched_page{page_no}.log") with open(log_path, "w", encoding="utf-8") as lf: lf.write(f"=== 页{page_no} 未匹配 {skipped} 条 ===\n\n") lf.write("\n---\n".join(unmatched_log)) logger.info("未匹配日志: %s", log_path) # 截几个未匹配的 DOM 卡片 for i, c in enumerate(dom_cards): if c["index"] not in used and i < 10: try: jpg = c["locator"].screenshot(type="jpeg", quality=70) fname = f"unmatched_{page_no}_{c['index']}.jpg" with open(os.path.join("./snapshots/unmatched", fname), "wb") as f: f.write(jpg) except Exception: pass if skipped: logger.info("页%s: 匹配入库 %s 条, 未匹配跳过 %s 条", page_no, matched, skipped) def _to_product(self, item, snapshot_url=""): now = time.strftime("%Y-%m-%d %H:%M:%S") today = time.strftime("%Y-%m-%d") sid = str(item.get("itemStoreId") or "") # min_price = gc-l2-price = 页面实际显示价 min_price = compute_one_price(item) return { "platform": PLATFORM_ID, "item_id": sid, "enterprise_id": self.company_id, "product_name": item.get("itemStoreName") or "", "spec": item.get("specs") or "", "one_price": 0, # 不存此字段，填0 "detail_url": PRODUCT_LINK_TPL.format(item_store_id=sid) if sid else "", "shop_name": item.get("storeName") or "", "anonymous_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": min_price, # → DB min_price = gc-l2-price "sales": item.get("totalCustCount") or "", "stock_count": item.get("itemStorage") or item.get("ownStorageNumber") or 0, "snapshot_url": snapshot_url, "approval_num": item.get("approvalNo") or "", "produced_time": "", "deadline": str(item.get("validateTimeEnd") or ""), "update_time": now, "insert_time": now, "number": 1, # 固定为1 "product_brand": item.get("brandName") or self.product_brand, "collect_task_id": self.collect_task_id, "search_name": self.product_name, "company_name": item.get("storeFullName") or "", "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 _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": 0, "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_snapshot(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 # ============ 主流程 ============ def run(self): self.uploader = AliyunOSSUploader() self._all_results = [] pages = {} # 提前声明，确保 finally 可见 # 1. 拿账号 acc = self._get_account() if not acc: logger.error("无可用账号") self.is_success = False self._report_exception(1, "无可用账号或Token") return 0, False phone = acc["phone"] password = acc["password"] # 2. 一个浏览器干到底（headless + Linux 兼容） with sync_playwright() as p: browser = p.chromium.launch( headless=True, channel="chrome", args=[ "--disable-blink-features=AutomationControlled", "--no-sandbox", "--disable-setuid-sandbox", "--disable-dev-shm-usage", "--disable-gpu", ], ) context = browser.new_context( locale="zh-CN", timezone_id="Asia/Shanghai", viewport={"width": 1600, "height": 900}, ) page = context.new_page() # 3. 加载 DB 缓存的 cookie（类似 main.py load_cookies） db_cookie_str = acc.get("cookie_str") or "" if db_cookie_str: try: db_cookies = json.loads(db_cookie_str) cookie_list = [ {"name": k, "value": str(v), "domain": ".yyjzt.com", "path": "/"} for k, v in db_cookies.items() if v and k not in ("Token",) ] if cookie_list: page.goto(HOME_URL, timeout=10000) # 先访问域名 context.add_cookies(cookie_list) logger.info("已加载 DB 缓存 %s 个 cookie", len(cookie_list)) except (json.JSONDecodeError, Exception) as e: logger.warning("加载缓存 cookie 失败: %s", e) # 4. 检查登录态 → 没登录当场登（同一个浏览器） if self._is_login(page): logger.info("Cookie 有效，已登录") else: logger.info("未登录，清 cookie 后重新登录...") context.clear_cookies() page.goto(LOGIN_URL, timeout=15000) page.wait_for_load_state("networkidle") time.sleep(2) if not self._login_operation(page, phone, password, skip_goto=True): logger.error("登录失败") self.is_success = False self._report_exception(4, "登录失败") browser.close() return 0, False logger.info("登录成功") # 登录后清弹窗 self._kill_popups(page) # 保存 cookie 到 DB cookies = context.cookies() cookie_map = {c["name"]: c["value"] for c in cookies} token = cookie_map.get("yjj-token", "").replace("ssr_", "") cookie_map["Token"] = token now_ts = int(time.time()) self.db.execute( "UPDATE accounts_platform SET cookie_str=%s, cookie_timestamp=%s WHERE phone=%s AND platform=%s", (json.dumps(cookie_map, ensure_ascii=False), now_ts, str(phone), PLATFORM_ID), ) logger.info("已保存 %s 个 cookie 到 DB", len(cookie_map)) # 4. 从 browser cookie 提取 token 给 API 用 cookies = context.cookies() for c in cookies: if c["name"] == "yjj-token": self._token = c["value"].replace("ssr_", "") break # 5. API 搜全量数据 all_items = self._api_search_all() if not all_items: logger.info("API 搜索无结果，关键词=%s，正常完成任务", self.keyword) # 正常回告：已跑完，数据为 0 self._report_page_progress_snapshot( page=1, crawled_count=0, is_finished=1, total_pages=0) browser.close() return 0, True # 6. 按页码分组，逐页截图 from collections import defaultdict pages = defaultdict(list) for item, page_no in all_items: pages[page_no].append((item, page_no)) for page_no in sorted(pages.keys()): if self._process_page(page, pages[page_no], page_no) is False: break time.sleep(random.uniform(2, 3)) # 正常完成回告 self._report_page_progress_snapshot( page=len(pages), crawled_count=self.pipeline.crawl_count, is_finished=1, total_pages=len(pages)) browser.close() logger.info("快照完成: %s 条, %s 页", self.pipeline.crawl_count, len(pages)) return self.pipeline.crawl_count, self.is_success, len(pages) if __name__ == "__main__": task = { "id": 999, "company_id": "1", "product_name": "蒲地蓝消炎片", "product_specs": "", "product_brand": "999", "product_keyword": "999 蒲地蓝消炎片", "current_page": 0, "start_page": 1, "end_page": 0, "scheduler": None, "_device_id": "test", "_platform": "7", "collect_task_id": "999", "snapshot_collect_status": 0, "collect_round": 1, "account_phone": "", "account_password": "", "account_name": "", "sampling_cycle": "", "sampling_start_time": "", "sampling_end_time": "", "collect_equipment_account_id": "", "collect_region_id": "", } s = YyjztSnapshotSpider(task) print(s.run())