update

spiders/yaojiujiu/yyjzt_crawl.py

@@ -387,6 +387,7 @@ class YyjztSpider:
 
         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)
@@ -400,9 +401,12 @@ class YyjztSpider:
                 return
 
             if not items:
-                logger.info("第 %s 页无数据，停止翻页", page)
+                logger.info("第 %s 页无数据，停止翻页")
                 break
 
+            pages.append(page)
+            self.page = page
+
             # 先补齐所有店铺信息
             for item in items:
                 store_id = item.get("storeId")
@@ -423,6 +427,8 @@ class YyjztSpider:
                                    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  # 匹配成功则重置
@@ -432,8 +438,6 @@ class YyjztSpider:
                 except Exception as e:
                     logger.exception("入库失败: %s", e)
 
-            self.page = page
-
             if self._report_page_progress(page, self.pipeline.crawl_count):
                 break
 
@@ -443,6 +447,12 @@ class YyjztSpider:
             # 翻页休息 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()

spiders/yaojiujiu/yyjzt_snapshot.py

@@ -86,25 +86,69 @@ class YyjztSnapshotSpider:
         return rows[0] if rows else None
 
     def _is_login(self, page):
-        """参考 main.py is_login()"""
+        """用 API 请求验证 token 是否真正有效"""
         try:
             page.goto(HOME_URL, timeout=15000)
             page.wait_for_load_state("networkidle")
             time.sleep(2)
-            if "login" in page.url.lower():
+            url = page.url.lower()
+            logger.info("检测登录态: URL=%s", url)
+
+            # 快速判断：URL 跳到登录页直接返回 False
+            if "login" in url:
+                logger.info("URL 含 login，未登录")
                 return False
-            if page.locator(":has-text('欢迎登录')").count() > 0:
+
+            # 提取 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
-            return True
-        except Exception:
+
+            # 调搜索 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):
+    def _login_operation(self, page, phone, password, skip_goto=False):
         """参考 main.py login_operation()"""
         try:
-            page.goto(LOGIN_URL, timeout=15000)
-            page.wait_for_load_state("networkidle")
-            time.sleep(2)
+            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")
@@ -127,20 +171,38 @@ class YyjztSnapshotSpider:
             # 点登录
             page.click(LOGIN_BTN_SELECTOR)
             logger.info("已点击登录")
-            time.sleep(3)
-
-            try:
-                page.wait_for_load_state("networkidle", timeout=15000)
-            except Exception:
-                pass
+            time.sleep(5)
 
-            # 检查是否登录成功
-            if "login" in page.url.lower():
-                err = page.locator(".el-message--error, [class*='error']").first
-                if err.count() > 0:
-                    logger.error("登录失败: %s", err.inner_text(timeout=3000))
+            # 调 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
-            return True
+
+            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
@@ -529,15 +591,54 @@ class YyjztSnapshotSpider:
             "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"]
@@ -580,9 +681,15 @@ class YyjztSnapshotSpider:
             if self._is_login(page):
                 logger.info("Cookie 有效，已登录")
             else:
-                logger.info("未登录，开始登录...")
-                if not self._login_operation(page, phone, password):
+                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("登录成功")
@@ -610,8 +717,12 @@ class YyjztSnapshotSpider:
             # 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, False
+                return 0, True
 
             # 6. 按页码分组，逐页截图
             from collections import defaultdict
@@ -624,15 +735,14 @@ class YyjztSnapshotSpider:
                     break
                 time.sleep(random.uniform(2, 3))
 
-            # 写结果文件
-            # out_path = f"./snapshots/result_{int(time.time())}.json"
-            # with open(out_path, "w", encoding="utf-8") as f:
-            #     json.dump(self._all_results, f, ensure_ascii=False, indent=2, default=str)
-            # logger.info("结果文件: %s (%s 条)", out_path, len(self._all_results))
+            # 正常完成回告
+            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 条", self.pipeline.crawl_count)
+        logger.info("快照完成: %s 条, %s 页", self.pipeline.crawl_count, len(pages))
         return self.pipeline.crawl_count, self.is_success, len(pages)