v1.0.2: 自动爬取试爬取功能 + 每个页面/图片生成操作信息元数据(模式/时间/网址/来源链接/深度等)

engine.py

@@ -3,6 +3,8 @@
 通用爬虫引擎 (Playwright + stealth + 系统 Chrome)
 - 批量模式: 逐条爬取网址列表
 - 自动模式: 从起始网址按匹配规则自动发现链接并 BFS 爬取
+- 试爬取: 仅抓取起始页, 列出按规则将爬取的链接(不保存文件)
+- 每个页面/图片生成 .meta.json 操作信息(模式/时间/网址/来源链接/深度等)
 - 支持: 随机延迟 / 重试 / 图片下载 / 暂停恢复 / 终止 / 配置热更新 / cookie 复用
 """
 import json
@@ -40,7 +42,6 @@ def is_challenge_page(title, html):
     for mark in _CHALLENGE_MARKS:
         if mark in t or mark in low:
             return True
-    # 极小页面 + 无正文结构 -> 疑似验证壳
     if len(html) < 5000 and ("<article" not in low and "<main" not in low):
         return True
     return False
@@ -52,6 +53,122 @@ def safe_name(url, idx):
     return f"{idx:04d}_{host}_{ts}"
 
 
+def _settle_wait(page, timeout_s):
+    """等待页面稳定(无 stop/pause 检查, 供试爬取等独立流程使用)"""
+    last_title, stable = "", 0
+    start = time.time()
+    while time.time() - start < timeout_s:
+        time.sleep(1)
+        try:
+            title = page.title()
+            html = page.content()
+        except Exception:
+            continue
+        if is_challenge_page(title, html):
+            stable = 0
+            continue
+        if title == last_title:
+            stable += 1
+            if stable >= 2 and len(html) > 1000:
+                return True, title, html
+        else:
+            stable = 0
+        last_title = title
+    return True, page.title(), page.content()
+
+
+def filter_links(hrefs, seed_url, include=None, exclude=None,
+                 same_domain=True, use_regex=False):
+    """按规则过滤链接, 返回 (included, excluded); excluded 含排除原因"""
+    include = include or []
+    exclude = exclude or []
+    included, excluded = [], []
+    seed_host = urllib.parse.urlparse(seed_url).hostname or ""
+    for h in hrefs:
+        if not str(h).startswith("http"):
+            continue
+        if same_domain and seed_host:
+            host = urllib.parse.urlparse(h).hostname or ""
+            if host != seed_host and not host.endswith("." + seed_host):
+                excluded.append({"url": h, "reason": "不在同域名内"})
+                continue
+        if use_regex:
+            if include and not any(re.search(p, h) for p in include):
+                excluded.append({"url": h, "reason": "未匹配包含规则"})
+                continue
+            hit = next((p for p in exclude if re.search(p, h)), None)
+            if hit:
+                excluded.append({"url": h, "reason": f"命中排除规则: {hit}"})
+                continue
+        else:
+            if include and not any(p.lower() in h.lower() for p in include):
+                excluded.append({"url": h, "reason": "未匹配包含规则"})
+                continue
+            hit = next((p for p in exclude if p.lower() in h.lower()), None)
+            if hit:
+                excluded.append({"url": h, "reason": f"命中排除规则: {hit}"})
+                continue
+        included.append(h)
+    return included, excluded
+
+
+def probe_links(seed_url, include=None, exclude=None,
+                same_domain=True, use_regex=False, timeout=45):
+    """试爬取: 抓取起始页并列出按规则将爬取的链接 (不保存任何文件)"""
+    result = {"ok": False, "error": "", "seed_url": seed_url, "title": "",
+              "crawled_at": "", "total_links": 0,
+              "total_included": 0, "total_excluded": 0,
+              "included": [], "excluded": []}
+    try:
+        p = sync_playwright().start()
+        browser = p.chromium.launch(
+            headless=True, executable_path=CHROME,
+            args=["--disable-blink-features=AutomationControlled",
+                  "--no-sandbox", "--disable-gpu", "--disable-dev-shm-usage"],
+        )
+        ctx = browser.new_context(
+            user_agent=DEFAULT_UA,
+            viewport={"width": 1920, "height": 1080},
+            locale="en-US",
+        )
+        Stealth().apply_stealth_sync(ctx)
+        page = ctx.new_page()
+        try:
+            page.goto(seed_url, wait_until="domcontentloaded", timeout=timeout * 1000)
+            _ok, title, html = _settle_wait(page, timeout)
+            if is_challenge_page(title, html):
+                result["error"] = f"起始页被反爬拦截: title={title!r}"
+            else:
+                try:
+                    hrefs = page.evaluate(
+                        "() => Array.from(document.querySelectorAll('a[href]')).map(a => a.href)"
+                    )
+                except Exception:
+                    hrefs = []
+                included, excluded = filter_links(
+                    hrefs, seed_url, include, exclude, same_domain, use_regex)
+                result.update(
+                    ok=True, title=title, crawled_at=store.now_str(),
+                    total_links=len(hrefs),
+                    total_included=len(included), total_excluded=len(excluded),
+                    included=included[:200], excluded=excluded[:200],
+                )
+        except Exception as e:
+            result["error"] = f"试爬取失败: {e}"
+        finally:
+            try:
+                browser.close()
+            except Exception:
+                pass
+            try:
+                p.stop()
+            except Exception:
+                pass
+    except Exception as e:
+        result["error"] = f"浏览器启动失败: {e}"
+    return result
+
+
 class CrawlJob:
     """一次爬取执行 (独立线程运行)"""
 
@@ -195,8 +312,8 @@ class CrawlJob:
             text = ""
         return title, html, text
 
-    def _crawl_images(self, ctx, page, out_dir, base):
-        """下载页面图片, 返回 [{file, url, size}]"""
+    def _crawl_images(self, ctx, page, out_dir, base, page_url, source_url):
+        """下载页面图片并生成图片集 meta.json, 返回 [{file,url,size,download_time}]"""
         try:
             urls = page.evaluate(
                 "() => Array.from(document.querySelectorAll('img'))"
@@ -222,26 +339,78 @@ class CrawlJob:
                     os.makedirs(img_dir, exist_ok=True)
                     with open(os.path.join(img_dir, fname), "wb") as f:
                         f.write(resp.body())
-                    saved.append({"file": f"{base}_img/{fname}", "url": u, "size": len(resp.body())})
+                    saved.append({
+                        "file": f"{base}_img/{fname}", "url": u,
+                        "size": len(resp.body()), "download_time": store.now_str(),
+                    })
             except Exception:
                 continue
+        if saved:
+            try:
+                meta = {
+                    "type": "images",
+                    "mode": self.run.get("mode"),
+                    "task_id": self.task["id"],
+                    "task_name": self.task.get("name", ""),
+                    "run_id": self.run.get("id"),
+                    "crawl_time": store.now_str(),
+                    "page_url": page_url,
+                    "source_url": source_url,
+                    "images": saved,
+                }
+                with open(os.path.join(img_dir, "meta.json"), "w", encoding="utf-8") as f:
+                    json.dump(meta, f, ensure_ascii=False, indent=2)
+            except Exception:
+                pass
         return saved
 
-    def _retry_crawl(self, page, ctx, url, idx, out_dir):
-        """带重试的单页爬取, 返回结果 entry"""
+    def _write_page_meta(self, out_dir, base, entry):
+        """为每个爬取页面生成操作信息 meta.json"""
+        meta = {
+            "type": "page",
+            "mode": self.run.get("mode"),
+            "task_id": self.task["id"],
+            "task_name": self.task.get("name", ""),
+            "run_id": self.run.get("id"),
+            "crawl_time": entry.get("crawl_time", ""),
+            "url": entry.get("url", ""),
+            "source_url": entry.get("source_url", ""),
+            "depth": entry.get("depth"),
+            "title": entry.get("title", ""),
+            "status": entry.get("status", ""),
+            "error": entry.get("error", ""),
+            "attempts": entry.get("attempts", 1),
+            "html_file": entry.get("html_file", ""),
+            "txt_file": entry.get("txt_file", ""),
+            "images": entry.get("images", []),
+        }
+        try:
+            with open(os.path.join(out_dir, base + ".meta.json"), "w", encoding="utf-8") as f:
+                json.dump(meta, f, ensure_ascii=False, indent=2)
+        except Exception:
+            pass
+
+    def _retry_crawl(self, page, ctx, url, idx, out_dir, source_url="", depth=None):
+        """带重试的单页爬取, 返回结果 entry (含 meta 信息)"""
         timeout = int(self._cfg("timeout", 60))
         retries = int(self._cfg("retry_count", 2))
         retry_wait = float(self._cfg("retry_interval", 3))
         crawl_images = bool(self._cfg("crawl_images", False))
 
-        entry = {"url": url, "title": "", "status": "FAIL", "error": "",
-                 "html_file": "", "txt_file": "", "images": []}
         base = safe_name(url, idx)
+        entry = {
+            "url": url, "title": "", "status": "FAIL", "error": "",
+            "html_file": "", "txt_file": "", "meta_file": base + ".meta.json",
+            "crawl_time": store.now_str(),
+            "source_url": source_url, "depth": depth,
+            "images": [], "attempts": 0,
+        }
         for attempt in range(retries + 1):
             if self._stop.is_set():
                 entry["error"] = "任务已终止"
                 break
             self._wait_if_paused()
+            entry["attempts"] += 1
             try:
                 title, html, text = self._crawl_one(page, url, timeout)
                 html_path = os.path.join(out_dir, base + ".html")
@@ -251,13 +420,15 @@ class CrawlJob:
                 with open(txt_path, "w", encoding="utf-8") as f:
                     f.write(text)
                 entry.update(title=title, status="OK",
-                             html_file=base + ".html", txt_file=base + ".txt")
+                             html_file=base + ".html", txt_file=base + ".txt",
+                             error="", crawl_time=store.now_str())
                 if crawl_images:
-                    entry["images"] = self._crawl_images(ctx, page, out_dir, base)
+                    entry["images"] = self._crawl_images(ctx, page, out_dir, base, url, source_url)
                 self._log("info", f"OK {title[:50]!r} html={len(html)//1024}KB 图片={len(entry['images'])}")
                 break
             except Exception as e:
                 entry["error"] = str(e)
+                entry["crawl_time"] = store.now_str()
                 self._log("warn", f"第{attempt + 1}次失败 {url}: {e}")
                 if attempt < retries:
                     self._wait_if_paused()
@@ -267,6 +438,7 @@ class CrawlJob:
                             break
                         self._wait_if_paused()
                         time.sleep(0.3)
+        self._write_page_meta(out_dir, base, entry)
         return entry
 
     def _delay(self):
@@ -317,7 +489,6 @@ class CrawlJob:
                     self._log("error", f"邮件通知失败: {msg}")
             except Exception as e:
                 self._log("error", f"邮件通知异常: {e}")
-                self._persist()
         self._log("info", f"任务结束: {run['status']} 成功{run['stats'].get('ok', 0)} 失败{run['stats'].get('fail', 0)}")
         self._persist()
 
@@ -353,37 +524,18 @@ class CrawlJob:
     def _discover_links(self, page):
         """从当前页面提取符合规则的链接"""
         auto = self.task.get("auto", {})
-        include = [x.strip() for x in (auto.get("include") or []) if x.strip()]
-        exclude = [x.strip() for x in (auto.get("exclude") or []) if x.strip()]
-        use_regex = bool(auto.get("use_regex", False))
-        same_domain = auto.get("same_domain", True)
-        seed_host = urllib.parse.urlparse(auto.get("seed_url", "")).hostname or ""
         try:
             hrefs = page.evaluate(
                 "() => Array.from(document.querySelectorAll('a[href]')).map(a => a.href)"
             )
         except Exception:
             return []
-        out = []
-        for h in hrefs:
-            if not str(h).startswith("http"):
-                continue
-            if same_domain and seed_host:
-                host = urllib.parse.urlparse(h).hostname or ""
-                if host != seed_host and not host.endswith("." + seed_host):
-                    continue
-            if use_regex:
-                if include and not any(re.search(p, h) for p in include):
-                    continue
-                if any(re.search(p, h) for p in exclude):
-                    continue
-            else:
-                if include and not any(p.lower() in h.lower() for p in include):
-                    continue
-                if any(p.lower() in h.lower() for p in exclude):
-                    continue
-            out.append(h)
-        return out
+        included, _excluded = filter_links(
+            hrefs, auto.get("seed_url", ""),
+            auto.get("include", []), auto.get("exclude", []),
+            auto.get("same_domain", True), bool(auto.get("use_regex", False)),
+        )
+        return included
 
     def _crawl_auto(self):
         run = self.run
@@ -398,14 +550,14 @@ class CrawlJob:
         self._persist()
 
         p, browser, ctx, page, cookie_file = self._open_browser()
-        queue = [(seed, 0)]
+        queue = [(seed, 0, "")]               # (url, depth, 来源链接)
         visited = set()
         queued = set([seed])
         idx = 0
         try:
             while queue and not self._stop.is_set():
                 self._wait_if_paused()
-                url, depth = queue.pop(0)
+                url, depth, src = queue.pop(0)
                 if url in visited:
                     continue
                 if len(visited) >= max_pages:
@@ -415,7 +567,8 @@ class CrawlJob:
                 run["progress"]["current_url"] = url
                 run["progress"]["done"] = len(visited)
                 self._persist()
-                entry = self._retry_crawl(page, ctx, url, idx, out_dir)
+                entry = self._retry_crawl(page, ctx, url, idx, out_dir,
+                                          source_url=src, depth=depth)
                 run["results"].append(entry)
                 self._bump_stats(entry)
                 self._persist()
@@ -423,7 +576,7 @@ class CrawlJob:
                     for link in self._discover_links(page):
                         if link not in visited and link not in queued:
                             queued.add(link)
-                            queue.append((link, depth + 1))
+                            queue.append((link, depth + 1, url))
                 if entry["status"] == "OK":
                     self._delay()
         finally: