Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
32eebf3dd1 | ||
|
|
d9c9f0c633 | ||
|
|
909e8e01b5 | ||
|
|
e749d70a43 | ||
|
|
56f30f78c4 |
@@ -4,3 +4,5 @@ data/*.json
|
||||
data/cookies_*.json
|
||||
logs/
|
||||
out/
|
||||
data/exports/
|
||||
data/auto_state/
|
||||
@@ -4,14 +4,17 @@
|
||||
启动: /home/hz1/miniconda3/envs/openclaw/bin/python app.py (默认端口 16062)
|
||||
"""
|
||||
import os
|
||||
import re
|
||||
import threading
|
||||
import time
|
||||
import zipfile
|
||||
from datetime import datetime
|
||||
|
||||
from flask import Flask, jsonify, request, send_file, send_from_directory
|
||||
|
||||
import store
|
||||
import db
|
||||
import notify
|
||||
from engine import CrawlJob, probe_links
|
||||
from scheduler import Scheduler, cron_next, interval_delta
|
||||
|
||||
@@ -101,13 +104,16 @@ def persist_cb(task_id, run):
|
||||
print(f"[db] 同步失败: {e}", flush=True)
|
||||
|
||||
|
||||
def start_run(task):
|
||||
"""为任务启动一次爬取, 返回 (run, error)"""
|
||||
def start_run(task, skip_seed=False):
|
||||
"""为任务启动一次爬取, 返回 (run, error)
|
||||
skip_seed: auto 任务继续爬取模式, 跳过起始网址直接从缓存队列消费
|
||||
"""
|
||||
with JOBS_LOCK:
|
||||
job = JOBS.get(task["id"])
|
||||
if job and job.is_running():
|
||||
return None, "该任务已有正在运行的爬取"
|
||||
run = make_run(task)
|
||||
run["skip_seed"] = bool(skip_seed)
|
||||
store.add_run(task["id"], run)
|
||||
job = CrawlJob(task, run, persist_cb)
|
||||
JOBS[task["id"]] = job
|
||||
@@ -302,11 +308,19 @@ def api_task_detail(tid):
|
||||
result["run_page"] = page
|
||||
result["run_pages"] = pages
|
||||
result["run_page_size"] = page_size
|
||||
# auto 状态数量 (独立文件)
|
||||
# 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)
|
||||
result["auto_pending_count"] = len(st.get("pending", []))
|
||||
result["auto_visited_count"] = len(st.get("visited", []))
|
||||
result["auto_pending_count"] = (
|
||||
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)
|
||||
|
||||
|
||||
@@ -421,6 +435,23 @@ def api_trash_clear():
|
||||
|
||||
# ---------------- 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"])
|
||||
def api_start(tid):
|
||||
task = store.get_task(tid)
|
||||
@@ -674,6 +705,103 @@ def api_file():
|
||||
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)
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -226,6 +226,9 @@ class CrawlJob:
|
||||
self.persist = persist # callable(task_id, run)
|
||||
self._stop = threading.Event()
|
||||
self._pause = threading.Event()
|
||||
# 运行中的 auto 实时状态 (供详情接口读取; 任务结束时由状态文件兜底)
|
||||
self._auto_visited = None
|
||||
self._auto_pending = None
|
||||
self._cfg_lock = threading.RLock()
|
||||
self.thread = None
|
||||
|
||||
@@ -598,9 +601,10 @@ class CrawlJob:
|
||||
def _crawl_auto(self):
|
||||
"""自动爬取: 持续递归 爬取->发现链接->爬取... 直到无新链接可爬或达到上限
|
||||
- max_depth: 0=无限制, N=只爬 N 层
|
||||
- max_pages: 0=无限制, N=安全上限
|
||||
- max_pages: 0=无限制, N=安全上限 (继续爬取模式按本次新增页数重新计算)
|
||||
- 提取但未爬取的链接持久化到 auto.pending, 已爬集合持久化到 auto.visited
|
||||
- 停止后再次运行从缓存队列继续爬; 起始网址每次运行都重新爬(不去重)
|
||||
- skip_seed(继续爬取): 跳过起始网址直接消费缓存队列, 页数上限按本次新增重新计算
|
||||
"""
|
||||
run = self.run
|
||||
auto = self.task.get("auto", {})
|
||||
@@ -615,8 +619,13 @@ class CrawlJob:
|
||||
state = store.load_auto_state(self.task["id"], self.task)
|
||||
visited = set(state.get("visited", []) 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:
|
||||
queue.extend((p["url"], p.get("depth", 0), p.get("source", "")) for p in pending)
|
||||
queued = set(visited)
|
||||
@@ -624,20 +633,30 @@ class CrawlJob:
|
||||
queued.add(normalize_url(u))
|
||||
|
||||
run["progress"]["total"] = len(queue)
|
||||
self._auto_visited = len(visited)
|
||||
self._auto_pending = len(queue)
|
||||
self._persist()
|
||||
|
||||
p, browser, ctx, page, cookie_file = self._open_browser()
|
||||
try:
|
||||
while queue and not self._stop.is_set():
|
||||
self._wait_if_paused()
|
||||
if max_pages > 0 and len(visited) >= max_pages:
|
||||
self._log("info", f"达到最大页数上限 {max_pages}, 停止")
|
||||
break
|
||||
if max_pages > 0:
|
||||
if skip_seed:
|
||||
# 继续爬取模式: 页数上限按本次新增页数重新计算 (累计 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)
|
||||
key = normalize_url(url)
|
||||
if key in visited and url != seed: # 起始网址不去重, 其余已爬跳过
|
||||
continue
|
||||
visited.add(key)
|
||||
self._auto_visited = len(visited)
|
||||
self._auto_pending = len(queue)
|
||||
idx = len(visited)
|
||||
run["progress"]["current_url"] = url
|
||||
run["progress"]["done"] = len(visited)
|
||||
@@ -654,6 +673,7 @@ class CrawlJob:
|
||||
if lk not in visited and lk not in queued:
|
||||
queued.add(lk)
|
||||
queue.append((link, depth + 1, url))
|
||||
self._auto_pending = len(queue) # 新链接入队后实时刷新
|
||||
if entry["status"] == "OK":
|
||||
self._delay()
|
||||
run["progress"]["total"] = len(visited)
|
||||
|
||||
@@ -33,11 +33,24 @@ def notify_email(task, run):
|
||||
if len(results) > 50:
|
||||
lines.append(f" ... 共 {len(results)} 条")
|
||||
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:
|
||||
r = subprocess.run(
|
||||
[sys.executable, SEND_EMAIL, f"[爬虫完成] {task['name']}", body, "--to", to],
|
||||
timeout=60, capture_output=True,
|
||||
)
|
||||
r = subprocess.run(cmd, timeout=600, capture_output=True)
|
||||
return r.returncode == 0, (r.stderr or b"").decode(errors="ignore")[:200]
|
||||
except Exception as e:
|
||||
return False, str(e)
|
||||
+68
-1
@@ -602,10 +602,15 @@ function renderDetail() {
|
||||
${t.running ? `
|
||||
<button class="btn" onclick="actTask('${t.id}','pause')">⏸ 暂停</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>
|
||||
<button class="btn" ${t.running ? "disabled" : ""} onclick="clearCache('${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>`;
|
||||
|
||||
const chips = runs.map((r) => `
|
||||
@@ -732,6 +737,68 @@ function stopLogPoll() {
|
||||
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() {
|
||||
const f = $("taskForm");
|
||||
|
||||
Reference in New Issue
Block a user