Compare commits

..
5 Commits
Author SHA1 Message Date
hz4th_coder 32eebf3dd1 auto 任务运行中实时显示已爬/待爬缓存数; 人工停止后继续爬取验证
- engine: CrawlJob 增加 _auto_visited/_auto_pending 运行中实时状态 (含发现新链接入队后刷新)
- app.py 详情接口: 任务运行中优先读内存实时值, 不再显示上次运行结束的旧数字
- 端到端验证: 人工停止(已爬3,缓存18) → 继续爬取(跳过seed,爬完18页,缓存0) 
- 运行中采样: visited 1→2 实时增长, pending 15→14 同步递减 
2026-08-12 10:19:19 +08:00
hz4th_coder d9c9f0c633 chore: data/auto_state 为运行时爬取状态, 移出版本控制 (.gitignore) 2026-08-12 09:57:53 +08:00
hz4th_coder 909e8e01b5 新增继续爬取: auto 任务达 max_pages 上限停止后, 有待爬缓存链接时可在详情页一键继续
- POST /api/tasks/<tid>/continue: 校验 auto 模式 + pending 非空, skip_seed 启动
- engine skip_seed: 跳过起始网址直接从缓存队列消费 (不重爬 seed, 不重复入 visited)
- 关键修复: max_pages 原为累计上限(visited>=max_pages), 已达标任务无法继续; 继续模式改为按本次新增页数重新计算上限, 可反复继续直到缓存耗尽
- 详情弹窗 auto 任务且有缓存时显示 [ 继续爬取(缓存 N 条)] 按钮
- 端到端测试: max_pages=2 链路 执行(seed+p1,缓存4) → 继续(p2,p3,缓存2) → 继续(p4,p5,缓存0) → 无缓存报错 
2026-08-12 09:57:46 +08:00
hz4th_coder e749d70a43 chore: data/exports 为运行时打包产物, 移出版本控制 (.gitignore) 2026-08-12 09:43:32 +08:00
hz4th_coder 56f30f78c4 新增打包导出: 任务输出目录打包为 zip, 支持网页下载或发送到指定邮箱(默认 wlq@tphai.com)
- GET /api/export?task_id= 打包输出目录为 zip 并下载 (zip 内按任务名建顶层目录)
- POST /api/export/email 打包后通过 send_email.py 发送附件到邮箱, 邮箱可指定, 默认 wlq@tphai.com
- 详情弹窗新增 [📦 打包下载] [📧 发邮箱] 按钮, 打包中按钮禁用防重复点击
- notify.py 重构: send_attachment(subject, body, to, attach) 通用附件发送
- 打包文件存 data/exports/, 自动清理只保留最近 10 个
- 实测: cnblogs-auto 66MB/1606文件 -> 11.9MB zip 仅1.5s; 邮件发送成功 2.2s
2026-08-12 09:41:59 +08:00
6 changed files with 245 additions and 46901 deletions
+2
View File
@@ -4,3 +4,5 @@ data/*.json
data/cookies_*.json data/cookies_*.json
logs/ logs/
out/ out/
data/exports/
data/auto_state/
+133 -5
View File
@@ -4,14 +4,17 @@
启动: /home/hz1/miniconda3/envs/openclaw/bin/python app.py (默认端口 16062) 启动: /home/hz1/miniconda3/envs/openclaw/bin/python app.py (默认端口 16062)
""" """
import os import os
import re
import threading import threading
import time import time
import zipfile
from datetime import datetime from datetime import datetime
from flask import Flask, jsonify, request, send_file, send_from_directory from flask import Flask, jsonify, request, send_file, send_from_directory
import store import store
import db import db
import notify
from engine import CrawlJob, probe_links from engine import CrawlJob, probe_links
from scheduler import Scheduler, cron_next, interval_delta from scheduler import Scheduler, cron_next, interval_delta
@@ -101,13 +104,16 @@ def persist_cb(task_id, run):
print(f"[db] 同步失败: {e}", flush=True) print(f"[db] 同步失败: {e}", flush=True)
def start_run(task): def start_run(task, skip_seed=False):
"""为任务启动一次爬取, 返回 (run, error)""" """为任务启动一次爬取, 返回 (run, error)
skip_seed: auto 任务继续爬取模式, 跳过起始网址直接从缓存队列消费
"""
with JOBS_LOCK: with JOBS_LOCK:
job = JOBS.get(task["id"]) job = JOBS.get(task["id"])
if job and job.is_running(): if job and job.is_running():
return None, "该任务已有正在运行的爬取" return None, "该任务已有正在运行的爬取"
run = make_run(task) run = make_run(task)
run["skip_seed"] = bool(skip_seed)
store.add_run(task["id"], run) store.add_run(task["id"], run)
job = CrawlJob(task, run, persist_cb) job = CrawlJob(task, run, persist_cb)
JOBS[task["id"]] = job JOBS[task["id"]] = job
@@ -302,11 +308,19 @@ def api_task_detail(tid):
result["run_page"] = page result["run_page"] = page
result["run_pages"] = pages result["run_pages"] = pages
result["run_page_size"] = page_size result["run_page_size"] = page_size
# auto 状态数量 (独立文件) # auto 状态数量: 运行中优先读内存实时值, 否则读状态文件
if task.get("mode") == "auto": if task.get("mode") == "auto":
live_v = live_p = None
with JOBS_LOCK:
job = JOBS.get(tid)
if job:
live_v = getattr(job, "_auto_visited", None)
live_p = getattr(job, "_auto_pending", None)
st = store.load_auto_state(tid, task) st = store.load_auto_state(tid, task)
result["auto_pending_count"] = len(st.get("pending", [])) result["auto_pending_count"] = (
result["auto_visited_count"] = len(st.get("visited", [])) live_p if live_p is not None else len(st.get("pending", [])))
result["auto_visited_count"] = (
live_v if live_v is not None else len(st.get("visited", [])))
return jsonify(result) return jsonify(result)
@@ -421,6 +435,23 @@ def api_trash_clear():
# ---------------- API: 运行控制 ---------------- # ---------------- API: 运行控制 ----------------
@app.route("/api/tasks/<tid>/continue", methods=["POST"])
def api_continue(tid):
"""继续爬取: auto 任务从待爬缓存队列接着爬 (跳过起始网址, 保留已爬集合)"""
task = store.get_task(tid)
if not task:
return jsonify({"error": "任务不存在"}), 404
if task.get("mode") != "auto":
return jsonify({"error": "仅自动爬取任务支持继续爬取"}), 400
st = store.load_auto_state(tid, task)
if not st.get("pending"):
return jsonify({"error": "没有待爬缓存链接,无需继续"}), 400
run, err = start_run(task, skip_seed=True)
if err:
return jsonify({"error": err}), 409
return jsonify(run)
@app.route("/api/tasks/<tid>/start", methods=["POST"]) @app.route("/api/tasks/<tid>/start", methods=["POST"])
def api_start(tid): def api_start(tid):
task = store.get_task(tid) task = store.get_task(tid)
@@ -674,6 +705,103 @@ def api_file():
return send_file(full) return send_file(full)
# ---------------- API: 打包导出 ----------------
EXPORT_DIR = os.path.join(HERE, "data", "exports")
EXPORT_KEEP = 10 # 最多保留的打包文件数
def _safe_zip_name(task):
"""任务名安全化为文件名 (保留中文, 去非法字符)"""
name = re.sub(r'[\\/:*?"<>|\s]+', "_", task.get("name", "")).strip("_")
return (name or task["id"])[:60]
def _make_zip(task):
"""把任务输出目录打包为 zip, 返回 (zip_path, err) 或 (None, 错误信息)"""
out_dir = os.path.realpath(resolve_out_dir(task))
if not os.path.isdir(out_dir):
return None, "输出目录不存在"
os.makedirs(EXPORT_DIR, exist_ok=True)
zip_path = os.path.join(EXPORT_DIR, f"{task['id']}_{int(time.time())}.zip")
prefix = _safe_zip_name(task) + "/"
count = 0
try:
with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
for root, _dirs, files in os.walk(out_dir):
for f in files:
full = os.path.join(root, f)
rel = os.path.relpath(full, out_dir)
zf.write(full, prefix + rel)
count += 1
except Exception as e:
try:
os.remove(zip_path)
except OSError:
pass
return None, f"打包失败: {e}"
_cleanup_exports()
return zip_path, None
def _cleanup_exports():
"""清理旧的打包文件, 只保留最近 EXPORT_KEEP 个"""
try:
files = sorted(
(os.path.join(EXPORT_DIR, f) for f in os.listdir(EXPORT_DIR)
if f.endswith(".zip")),
key=os.path.getmtime, reverse=True,
)
for f in files[EXPORT_KEEP:]:
os.remove(f)
except Exception:
pass
@app.route("/api/export")
def api_export():
"""打包任务输出目录为 zip 并提供下载"""
tid = request.args.get("task_id", "")
task = store.get_task(tid)
if not task:
return jsonify({"error": "任务不存在"}), 404
zip_path, err = _make_zip(task)
if err:
return jsonify({"error": err}), 400
return send_file(
zip_path, as_attachment=True,
download_name=_safe_zip_name(task) + ".zip",
mimetype="application/zip",
)
@app.route("/api/export/email", methods=["POST"])
def api_export_email():
"""打包任务输出目录为 zip 并发送到指定邮箱 (默认 wlq@tphai.com)"""
body = request.get_json(force=True) or {}
tid = body.get("task_id", "")
email = (body.get("email") or "").strip() or "wlq@tphai.com"
task = store.get_task(tid)
if not task:
return jsonify({"error": "任务不存在"}), 404
if "@" not in email:
return jsonify({"error": "邮箱格式不正确"}), 400
zip_path, err = _make_zip(task)
if err:
return jsonify({"error": err}), 400
size_mb = round(os.path.getsize(zip_path) / 1048576, 1)
body_text = (f"项目: {task['name']}\n"
f"输出目录: {resolve_out_dir(task)}\n"
f"打包文件: {os.path.basename(zip_path)}\n"
f"压缩包大小: {size_mb} MB\n\n"
f"打包时间: {now_str()}")
ok, msg = notify.send_attachment(
f"[爬虫打包] {task['name']}", body_text, email, zip_path)
if not ok:
return jsonify({"error": f"邮件发送失败: {msg}"}), 500
return jsonify({"ok": True, "msg": f"已发送到 {email} (zip {size_mb} MB)"})
# ---------------- 启动 ---------------- # ---------------- 启动 ----------------
scheduler = Scheduler(start_run) scheduler = Scheduler(start_run)
File diff suppressed because it is too large Load Diff
+25 -5
View File
@@ -226,6 +226,9 @@ class CrawlJob:
self.persist = persist # callable(task_id, run) self.persist = persist # callable(task_id, run)
self._stop = threading.Event() self._stop = threading.Event()
self._pause = threading.Event() self._pause = threading.Event()
# 运行中的 auto 实时状态 (供详情接口读取; 任务结束时由状态文件兜底)
self._auto_visited = None
self._auto_pending = None
self._cfg_lock = threading.RLock() self._cfg_lock = threading.RLock()
self.thread = None self.thread = None
@@ -598,9 +601,10 @@ class CrawlJob:
def _crawl_auto(self): def _crawl_auto(self):
"""自动爬取: 持续递归 爬取->发现链接->爬取... 直到无新链接可爬或达到上限 """自动爬取: 持续递归 爬取->发现链接->爬取... 直到无新链接可爬或达到上限
- max_depth: 0=无限制, N=只爬 N 层 - max_depth: 0=无限制, N=只爬 N 层
- max_pages: 0=无限制, N=安全上限 - max_pages: 0=无限制, N=安全上限 (继续爬取模式按本次新增页数重新计算)
- 提取但未爬取的链接持久化到 auto.pending, 已爬集合持久化到 auto.visited - 提取但未爬取的链接持久化到 auto.pending, 已爬集合持久化到 auto.visited
- 停止后再次运行从缓存队列继续爬; 起始网址每次运行都重新爬(不去重) - 停止后再次运行从缓存队列继续爬; 起始网址每次运行都重新爬(不去重)
- skip_seed(继续爬取): 跳过起始网址直接消费缓存队列, 页数上限按本次新增重新计算
""" """
run = self.run run = self.run
auto = self.task.get("auto", {}) auto = self.task.get("auto", {})
@@ -615,8 +619,13 @@ class CrawlJob:
state = store.load_auto_state(self.task["id"], self.task) state = store.load_auto_state(self.task["id"], self.task)
visited = set(state.get("visited", []) or []) visited = set(state.get("visited", []) or [])
pending = state.get("pending", []) or [] pending = state.get("pending", []) or []
visited_base = len(visited)
# 起始网址每次运行都爬(不做去重), 缓存队列继续消费 # 起始网址每次运行都爬(不做去重), 缓存队列继续消费
queue = [(seed, 0, "")] # 继续爬取模式(skip_seed): 有缓存队列时跳过起始网址, 直接从待爬队列接着爬
skip_seed = bool(self.run.get("skip_seed"))
queue = []
if not (skip_seed and pending):
queue.append((seed, 0, ""))
if pending: if pending:
queue.extend((p["url"], p.get("depth", 0), p.get("source", "")) for p in pending) queue.extend((p["url"], p.get("depth", 0), p.get("source", "")) for p in pending)
queued = set(visited) queued = set(visited)
@@ -624,20 +633,30 @@ class CrawlJob:
queued.add(normalize_url(u)) queued.add(normalize_url(u))
run["progress"]["total"] = len(queue) run["progress"]["total"] = len(queue)
self._auto_visited = len(visited)
self._auto_pending = len(queue)
self._persist() self._persist()
p, browser, ctx, page, cookie_file = self._open_browser() p, browser, ctx, page, cookie_file = self._open_browser()
try: try:
while queue and not self._stop.is_set(): while queue and not self._stop.is_set():
self._wait_if_paused() self._wait_if_paused()
if max_pages > 0 and len(visited) >= max_pages: if max_pages > 0:
self._log("info", f"达到最大页数上限 {max_pages}, 停止") if skip_seed:
break # 继续爬取模式: 页数上限按本次新增页数重新计算 (累计 visited 不阻塞继续)
if len(visited) - visited_base >= max_pages:
self._log("info", f"本次继续爬取达到页数上限 {max_pages}, 停止")
break
elif len(visited) >= max_pages:
self._log("info", f"达到最大页数上限 {max_pages}, 停止")
break
url, depth, src = queue.pop(0) url, depth, src = queue.pop(0)
key = normalize_url(url) key = normalize_url(url)
if key in visited and url != seed: # 起始网址不去重, 其余已爬跳过 if key in visited and url != seed: # 起始网址不去重, 其余已爬跳过
continue continue
visited.add(key) visited.add(key)
self._auto_visited = len(visited)
self._auto_pending = len(queue)
idx = len(visited) idx = len(visited)
run["progress"]["current_url"] = url run["progress"]["current_url"] = url
run["progress"]["done"] = len(visited) run["progress"]["done"] = len(visited)
@@ -654,6 +673,7 @@ class CrawlJob:
if lk not in visited and lk not in queued: if lk not in visited and lk not in queued:
queued.add(lk) queued.add(lk)
queue.append((link, depth + 1, url)) queue.append((link, depth + 1, url))
self._auto_pending = len(queue) # 新链接入队后实时刷新
if entry["status"] == "OK": if entry["status"] == "OK":
self._delay() self._delay()
run["progress"]["total"] = len(visited) run["progress"]["total"] = len(visited)
+17 -4
View File
@@ -33,11 +33,24 @@ def notify_email(task, run):
if len(results) > 50: if len(results) > 50:
lines.append(f" ... 共 {len(results)}") lines.append(f" ... 共 {len(results)}")
body = "\n".join(lines) body = "\n".join(lines)
return send_attachment(f"[爬虫完成] {task['name']}", body, to)
def send_attachment(subject, body, to, attach_path=None):
"""发送带附件的邮件 (复用 send_email.py), 返回 (bool, msg)
attach_path: 附件文件路径, 可多个
"""
if not os.path.exists(SEND_EMAIL):
return False, "send_email.py 不存在"
cmd = [sys.executable, SEND_EMAIL, subject, body, "--to", to]
if attach_path:
if isinstance(attach_path, str):
attach_path = [attach_path]
for p in attach_path:
if os.path.isfile(p):
cmd += ["--attach", p]
try: try:
r = subprocess.run( r = subprocess.run(cmd, timeout=600, capture_output=True)
[sys.executable, SEND_EMAIL, f"[爬虫完成] {task['name']}", body, "--to", to],
timeout=60, capture_output=True,
)
return r.returncode == 0, (r.stderr or b"").decode(errors="ignore")[:200] return r.returncode == 0, (r.stderr or b"").decode(errors="ignore")[:200]
except Exception as e: except Exception as e:
return False, str(e) return False, str(e)
+68 -1
View File
@@ -602,10 +602,15 @@ function renderDetail() {
${t.running ? ` ${t.running ? `
<button class="btn" onclick="actTask('${t.id}','pause')">⏸ 暂停</button> <button class="btn" onclick="actTask('${t.id}','pause')">⏸ 暂停</button>
<button class="btn danger" onclick="actTask('${t.id}','stop')">⏹ 终止</button>` <button class="btn danger" onclick="actTask('${t.id}','stop')">⏹ 终止</button>`
: `<button class="btn primary" onclick="actTask('${t.id}','start')">▶ 立即执行</button>`} : `
<button class="btn primary" onclick="actTask('${t.id}','start')">▶ 立即执行</button>
${t.mode === "auto" && (t.auto_pending_count || 0) > 0 ? `
<button class="btn" onclick="contTask('${t.id}')">⏩ 继续爬取(缓存 ${t.auto_pending_count} 条)</button>` : ""}`}
${t.mode === "auto" ? `<button class="btn" onclick="probeTask('${t.id}')">🧪 试爬取</button> ${t.mode === "auto" ? `<button class="btn" onclick="probeTask('${t.id}')">🧪 试爬取</button>
<button class="btn" ${t.running ? "disabled" : ""} onclick="clearCache('${t.id}')">🧹 清空缓存</button>` : ""} <button class="btn" ${t.running ? "disabled" : ""} onclick="clearCache('${t.id}')">🧹 清空缓存</button>` : ""}
<button class="btn" onclick="openEdit('${t.id}')">✏️ 编辑配置</button> <button class="btn" onclick="openEdit('${t.id}')">✏️ 编辑配置</button>
<button class="btn" onclick="exportZip('${t.id}')">📦 打包下载</button>
<button class="btn" onclick="exportEmail('${t.id}')">📧 发邮箱</button>
</div>`; </div>`;
const chips = runs.map((r) => ` const chips = runs.map((r) => `
@@ -732,6 +737,68 @@ function stopLogPoll() {
if (state.logTimer) { clearInterval(state.logTimer); state.logTimer = null; } if (state.logTimer) { clearInterval(state.logTimer); state.logTimer = null; }
} }
async function contTask(tid) {
try {
const r = await api(`/api/tasks/${tid}/continue`, { method: "POST" });
toast("已开始继续爬取缓存队列");
loadTasks();
if (state.detail.task && state.detail.task.id === tid) openDetail(tid);
} catch (e) { toast(e.message, true); }
}
/* ---------------- 打包导出 ---------------- */
async function exportZip(tid) {
const btn = window.event && window.event.target;
if (btn) { btn.disabled = true; btn.textContent = "⏳ 打包中..."; }
try {
const res = await fetch(`/api/export?task_id=${tid}`);
if (!res.ok) {
let d = null;
try { d = await res.json(); } catch (e) { /* ignore */ }
throw new Error((d && d.error) || `HTTP ${res.status}`);
}
let fname = "export.zip";
const disp = res.headers.get("Content-Disposition") || "";
const m = disp.match(/filename\*?=(?:UTF-8'')?"?([^";]+)"?/i);
if (m) { try { fname = decodeURIComponent(m[1]); } catch (e) { fname = m[1]; } }
const blob = await res.blob();
const a = document.createElement("a");
a.href = URL.createObjectURL(blob);
a.download = fname;
document.body.appendChild(a);
a.click();
a.remove();
setTimeout(() => URL.revokeObjectURL(a.href), 5000);
toast(`打包下载完成 ${fname} (${(blob.size / 1048576).toFixed(1)} MB)`);
} catch (e) {
toast("打包失败: " + e.message, true);
} finally {
if (btn) { btn.disabled = false; btn.textContent = "📦 打包下载"; }
}
}
async function exportEmail(tid) {
const t = (state.detail.task && state.detail.task.id === tid)
? state.detail.task : state.tasks.find((x) => x.id === tid);
const def = (t && t.config && t.config.notify_email) || "wlq@tphai.com";
const email = prompt("发送到邮箱(留空默认 wlq@tphai.com:", def);
if (email === null) return; // 用户取消
const btn = window.event && window.event.target;
if (btn) { btn.disabled = true; btn.textContent = "⏳ 打包发送中..."; }
try {
const r = await api("/api/export/email", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ task_id: tid, email: email.trim() || "wlq@tphai.com" }),
});
toast(r.msg || "已发送");
} catch (e) {
toast("发送失败: " + e.message, true);
} finally {
if (btn) { btn.disabled = false; btn.textContent = "📧 发邮箱"; }
}
}
/* ---------------- 试爬取 ---------------- */ /* ---------------- 试爬取 ---------------- */
function collectProbeFromForm() { function collectProbeFromForm() {
const f = $("taskForm"); const f = $("taskForm");