universal-crawler/app.py

# -*- coding: utf-8 -*-
"""
通用爬虫系统 - Web 管理后端
启动: /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, normalize_url, url_excluded
from scheduler import Scheduler, cron_next, interval_delta

HERE = os.path.dirname(os.path.abspath(__file__))
PORT = int(os.environ.get("CRAWLER_PORT", "16062"))

app = Flask(__name__, static_folder="static", static_url_path="")
JOBS = {}                       # task_id -> CrawlJob
JOBS_LOCK = threading.RLock()
STARTED = datetime.now()

DEFAULT_CONFIG = {
    "out_dir": "",              # 留空 -> out/<任务ID>
    "delay_min": 2,
    "delay_max": 5,
    "timeout": 60,
    "crawl_images": False,
    "retry_count": 2,
    "retry_interval": 3,
    "notify": False,
    "notify_email": "wlq@tphai.com",
}


def now_str():
    return store.now_str()


def _parse_first_run(s):
    """解析表单提交的首次执行时间 (datetime-local 格式), 非法返回 None"""
    if not s:
        return None
    try:
        return datetime.strptime(str(s), "%Y-%m-%dT%H:%M")
    except Exception:
        return None


def _schedule_next_run(sch):
    """根据调度配置 + 首次执行时间计算 next_run (str)"""
    first_run = _parse_first_run(sch.get("first_run"))
    base = first_run if (first_run and first_run > datetime.now()) else None
    if sch.get("type") == "cron":
        expr = sch.get("cron") or "0 * * * *"
        nn = cron_next(expr, base or datetime.now())
        if not nn:
            raise ValueError("cron 表达式在未来一年内无匹配时间")
        return nn.strftime("%Y-%m-%d %H:%M:%S")
    sch.setdefault("interval_unit", "hours")
    sch.setdefault("interval_value", 24)
    if base:
        return base.strftime("%Y-%m-%d %H:%M:%S")
    return (datetime.now() + interval_delta(sch)).strftime("%Y-%m-%d %H:%M:%S")


def resolve_out_dir(task):
    cfg = task.get("config", {}) or {}
    if cfg.get("out_dir", "").strip():
        return cfg["out_dir"].strip()
    return os.path.join(HERE, "out", task["id"])


def make_run(task):
    return {
        "id": store.new_id("r"),
        "task_id": task["id"],
        "mode": task.get("mode", "batch"),
        "status": "running",
        "progress": {"done": 0, "total": 0, "current_url": "", "percent": 0},
        "started_at": now_str(),
        "finished_at": "",
        "stats": {"ok": 0, "fail": 0, "images": 0},
        "results": [],
        "logs": [],
        "out_dir": resolve_out_dir(task),
    }


def persist_cb(task_id, run):
    total = run["progress"].get("total") or 0
    done = run["progress"].get("done") or 0
    run["progress"]["percent"] = round(done * 100 / total) if total else 0
    store.save_run(task_id, run)
    try:
        db.sync_run_async(run)      # 后台异步同步, 不再阻塞爬虫线程
    except Exception as e:
        print(f"[db] 同步失败: {e}", flush=True)


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
        job.start()
        db.upsert_task_async(task)        # 确保任务在库中 (异步)
        return run, None


def _stop_job(tid):
    with JOBS_LOCK:
        job = JOBS.get(tid)
        if job and job.is_running():
            job.stop()
            return True
    return False


# ---------------- 页面 ----------------

@app.route("/")
def index():
    return send_from_directory(app.static_folder, "index.html")


# ---------------- API: 状态 ----------------

@app.route("/api/status")
def api_status():
    with JOBS_LOCK:
        running = sum(1 for j in JOBS.values() if j.is_running())
    return jsonify({
        "app": "universal-crawler",
        "version": "1.0.0",
        "running": running,
        "uptime": str(datetime.now() - STARTED).split(".")[0],
    })


# ---------------- API: 任务 ----------------

def _strip_auto_state(task):
    """剥离任务对象中的待爬/已爬队列 (已独立存储, 避免大 JSON 拖慢接口)"""
    if task.get("mode") == "auto" and task.get("auto"):
        task["auto"] = {k: v for k, v in task["auto"].items()
                         if k not in ("pending", "visited")}
    return task


def _run_summary(run):
    """运行记录摘要 (不含 results/logs), 供列表/详情页使用"""
    return {
        "id": run["id"],
        "task_id": run.get("task_id", ""),
        "mode": run.get("mode", ""),
        "status": run.get("status", ""),
        "progress": run.get("progress", {}),
        "started_at": run.get("started_at", ""),
        "finished_at": run.get("finished_at", ""),
        "stats": run.get("stats", {}),
        "out_dir": run.get("out_dir", ""),
    }


@app.route("/api/tasks", methods=["GET"])
def api_tasks():
    tasks = [t for t in store.load_tasks() if not t.get("deleted_at")]
    runs_map = store.load_runs_map()      # 只全量读一次
    for t in tasks:
        _strip_auto_state(t)
        runs = runs_map.get(t["id"], [])
        t["latest_run"] = _run_summary(runs[-1]) if runs else None
        with JOBS_LOCK:
            job = JOBS.get(t["id"])
            t["running"] = bool(job and job.is_running())
    tasks.sort(key=lambda x: x.get("created_at", ""), reverse=True)
    return jsonify(tasks)


def _clean_urls(lines):
    """清洗网址行: 去空白, 去行内 # 注释 (URL 本身不含空格, 安全), 自动补 https 前缀"""
    out = []
    for u in lines or []:
        u = str(u).split(" #")[0].strip()
        if not u:
            continue
        if not u.startswith("http"):
            u = "https://" + u
        out.append(u)
    return out


@app.route("/api/tasks", methods=["POST"])
def api_create_task():
    body = request.get_json(force=True) or {}
    name = (body.get("name") or "").strip()
    mode = body.get("mode", "batch")
    if not name:
        return jsonify({"error": "项目名不能为空"}), 400
    if mode not in ("batch", "auto", "scheduled"):
        return jsonify({"error": "无效的模式"}), 400

    task = {
        "id": store.new_id("t"),
        "name": name,
        "mode": mode,
        "created_at": now_str(),
        "updated_at": now_str(),
        "config": {**DEFAULT_CONFIG, **(body.get("config") or {})},
        "urls": _clean_urls(body.get("urls") or []),
    }

    if mode == "auto":
        auto = dict(body.get("auto") or {})
        seed = (auto.get("seed_url") or "").strip()
        if not seed:
            return jsonify({"error": "自动模式需要填写起始网址"}), 400
        if not seed.startswith("http"):
            seed = "https://" + seed
        auto["seed_url"] = seed
        task["auto"] = auto
    elif mode == "scheduled":
        sch = dict(body.get("schedule") or {})
        sch.setdefault("enabled", True)
        sch.setdefault("type", "interval")
        try:
            sch["next_run"] = _schedule_next_run(sch)
        except ValueError as e:
            return jsonify({"error": f"调度配置错误: {e}"}), 400
        sch.setdefault("last_run", "")
        sch.setdefault("runs_count", 0)
        task["schedule"] = sch
        if not task["urls"]:
            return jsonify({"error": "定时任务需要网址列表"}), 400
    else:
        if not task["urls"]:
            return jsonify({"error": "请至少填写一个网址"}), 400

    store.upsert_task(task)
    db.upsert_task_async(task)
    return jsonify(task), 201


@app.route("/api/tasks/<tid>", methods=["GET"])
def api_task_detail(tid):
    task = store.get_task(tid)
    if not task:
        return jsonify({"error": "任务不存在"}), 404
    # 分页参数: 指定 run + 页码 (默认最新 run 第 1 页)
    rid = request.args.get("run", "")
    try:
        page = max(int(request.args.get("page", 1)), 1)
    except ValueError:
        page = 1
    try:
        page_size = min(max(int(request.args.get("page_size", 100)), 10), 500)
    except ValueError:
        page_size = 100

    runs = store.get_runs(tid)
    if rid:
        cur = next((r for r in runs if r["id"] == rid), None)
        if cur is None:
            cur = runs[-1] if runs else None   # 指定的 run 不存在时回退最新
    else:
        cur = runs[-1] if runs else None
    with JOBS_LOCK:
        job = JOBS.get(tid)
        task["running"] = bool(job and job.is_running())

    result = _strip_auto_state(dict(task))
    result["runs"] = [_run_summary(r) for r in reversed(runs)]
    result["run"] = None
    result["run_id"] = ""
    result["run_total"] = 0
    result["run_page"] = 1
    result["run_pages"] = 1
    if cur:
        results = cur.get("results", [])
        total = len(results)
        pages = max((total + page_size - 1) // page_size, 1)
        page = min(page, pages)
        start = (page - 1) * page_size
        snap = {k: v for k, v in cur.items() if k != "logs"}   # logs 走独立轮询接口
        snap["results"] = results[start:start + page_size]
        snap["results_total"] = total
        snap["run_page"] = page
        snap["run_pages"] = pages
        snap["run_page_size"] = page_size
        result["run"] = snap
        result["run_id"] = cur["id"]
        result["run_total"] = total
        result["run_page"] = page
        result["run_pages"] = pages
        result["run_page_size"] = page_size
    # 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"] = (
            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)


@app.route("/api/tasks/<tid>", methods=["PUT"])
def api_update_task(tid):
    task = store.get_task(tid)
    if not task:
        return jsonify({"error": "任务不存在"}), 404
    if task.get("deleted_at"):
        return jsonify({"error": "任务在回收站中，请先恢复"}), 400
    body = request.get_json(force=True) or {}
    with JOBS_LOCK:
        job = JOBS.get(tid)
        running = bool(job and job.is_running())
        if "name" in body and str(body["name"]).strip():
            task["name"] = str(body["name"]).strip()
        if "urls" in body:
            task["urls"] = _clean_urls(body["urls"])
        if "config" in body:
            merged = {**task.get("config", {}), **body["config"]}
            task["config"] = merged
            if running:
                job.update_config(body["config"])       # 运行中热更新
        if "auto" in body and task.get("mode") == "auto":
            old_exclude = task.get("auto", {}).get("exclude") or []
            task["auto"] = {**task.get("auto", {}), **body["auto"]}
            new_exclude = task["auto"].get("exclude") or []
            # 排除规则发生变化时, 同步清理待爬队列中已命中的 URL (已爬 visited 保留)
            if new_exclude and new_exclude != old_exclude and not running:
                st = store.load_auto_state(tid, task)
                kept = [p for p in st.get("pending", [])
                        if not url_excluded(p.get("url", ""), task["auto"])]
                removed = len(st.get("pending", [])) - len(kept)
                if removed:
                    store.save_auto_state(tid, {
                        "pending": kept,
                        "visited": st.get("visited", []),
                    })
                    if "exclude" in body.get("auto", {}):
                        task["auto"]["_cleaned"] = removed
        if "schedule" in body and task.get("mode") == "scheduled":
            sch = {**task.get("schedule", {}), **body["schedule"]}
            try:
                sch["next_run"] = _schedule_next_run(sch)
            except ValueError as e:
                return jsonify({"error": f"调度配置错误: {e}"}), 400
            task["schedule"] = sch
        task["updated_at"] = now_str()
        store.upsert_task(task)
        db.upsert_task_async(task)
    return jsonify(task)


@app.route("/api/tasks/<tid>", methods=["DELETE"])
def api_delete_task(tid):
    """删除任务 -> 移入回收站 (可恢复)"""
    task = store.get_task(tid)
    if not task:
        return jsonify({"error": "任务不存在"}), 404
    if task.get("deleted_at"):
        return jsonify({"error": "任务已在回收站中"}), 400
    _stop_job(tid)
    with JOBS_LOCK:
        JOBS.pop(tid, None)
    store.soft_delete_task(tid, now_str())
    db.upsert_task_async(store.get_task(tid))
    return jsonify({"ok": True, "msg": "已移入回收站"})


# ---------------- API: 回收站 ----------------

def _purge_out_dir(out_dir):
    """删除任务输出目录; 仅当目录位于项目 out/ 下才删 (自定义目录保留), 返回是否删除"""
    try:
        d = os.path.realpath(out_dir)
        base = os.path.realpath(os.path.join(HERE, "out"))
        if d.startswith(base + os.sep) and os.path.isdir(d):
            import shutil
            shutil.rmtree(d, ignore_errors=True)
            return True
    except Exception:
        pass
    return False


@app.route("/api/trash", methods=["GET"])
def api_trash_list():
    items = store.list_trash()
    runs_map = store.load_runs_map()
    for t in items:
        t["runs_count"] = len(runs_map.get(t["id"], []))
        t["out_dir"] = resolve_out_dir(t)
    return jsonify(items)


@app.route("/api/trash/<tid>/restore", methods=["POST"])
def api_trash_restore(tid):
    task = store.get_task(tid)
    if not task or not task.get("deleted_at"):
        return jsonify({"error": "任务不在回收站中"}), 404
    store.restore_task(tid)
    db.upsert_task_async(store.get_task(tid))
    return jsonify({"ok": True, "msg": "已恢复"})


@app.route("/api/trash/<tid>", methods=["DELETE"])
def api_trash_purge(tid):
    task = store.get_task(tid)
    if not task or not task.get("deleted_at"):
        return jsonify({"error": "任务不在回收站中"}), 404
    out_dir = resolve_out_dir(task)
    removed = _purge_out_dir(out_dir)
    store.purge_task(tid)
    db.purge_task_db(tid)
    return jsonify({"ok": True, "purged": True, "files_removed": removed, "out_dir": out_dir})


@app.route("/api/trash", methods=["DELETE"])
def api_trash_clear():
    items = store.list_trash()
    dirs = [resolve_out_dir(t) for t in items]
    store.purge_trash()
    db.purge_trash_db()
    removed = sum(1 for d in dirs if _purge_out_dir(d))
    return jsonify({"ok": True, "purged": len(items), "dirs_removed": removed})


# ---------------- API: 运行控制 ----------------

@app.route("/api/tasks/<tid>/retry-failed", methods=["POST"])
def api_retry_failed(tid):
    """重爬失败页: 提取指定 run (默认最新) 中失败的 URL, 从已爬集合解除标记并注入
    待爬队列头部; 返回注入数量, 调用方可随后点「继续爬取」重爬这些页面"""
    task = store.get_task(tid)
    if not task:
        return jsonify({"error": "任务不存在"}), 404
    if task.get("mode") != "auto":
        return jsonify({"error": "仅自动模式任务支持重爬失败页"}), 400
    try:
        body = request.get_json(force=True) or {}
    except Exception:
        body = {}
    rid = request.args.get("run", "") or body.get("run", "")
    runs = store.get_runs(tid)
    cur = None
    if rid:
        cur = next((r for r in runs if r["id"] == rid), None)
    else:
        cur = runs[-1] if runs else None
    if not cur:
        return jsonify({"error": "运行记录不存在"}), 404
    failed = [res.get("url") for res in cur.get("results", [])
              if res.get("status") == "FAIL" and res.get("url")]
    if not failed:
        return jsonify({"error": "该运行记录没有失败页面", "injected": 0})
    st = store.load_auto_state(tid, task)
    visited = set(st.get("visited", []))
    pending = st.get("pending", [])
    pending_urls = {normalize_url(p.get("url", "")) for p in pending}
    # 待重爬: 已爬过且不在待爬队列中的失败 URL (去重)
    to_inject, seen = [], set()
    for u in failed:
        key = normalize_url(u)
        if key in visited and key not in pending_urls and key not in seen:
            seen.add(key)
            to_inject.append({"url": u, "depth": 0, "source": "retry-failed"})
    if not to_inject:
        return jsonify({"error": "失败页面均已爬或已在待爬队列中", "injected": 0})
    # 解除已爬标记
    remove_keys = {normalize_url(u) for u in failed}
    visited = {v for v in visited if normalize_url(v) not in remove_keys}
    # 注入队列头部, 优先重爬
    pending = to_inject + pending
    store.save_auto_state(tid, {"pending": pending, "visited": sorted(visited)})
    return jsonify({"injected": len(to_inject), "pending_total": len(pending)})


@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)
    if not task:
        return jsonify({"error": "任务不存在"}), 404
    run, err = start_run(task)
    if err:
        return jsonify({"error": err}), 409
    return jsonify(run)


@app.route("/api/tasks/<tid>/stop", methods=["POST"])
def api_stop(tid):
    if _stop_job(tid):
        return jsonify({"ok": True, "msg": "已发送终止信号"})
    return jsonify({"ok": True, "msg": "任务未在运行"})


@app.route("/api/tasks/<tid>/pause", methods=["POST"])
def api_pause(tid):
    with JOBS_LOCK:
        job = JOBS.get(tid)
        if job and job.is_running():
            job.pause()
            return jsonify({"ok": True})
    return jsonify({"error": "任务未在运行"}), 409


@app.route("/api/tasks/<tid>/resume", methods=["POST"])
def api_resume(tid):
    with JOBS_LOCK:
        job = JOBS.get(tid)
        if job and job.is_running():
            job.resume()
            return jsonify({"ok": True})
    return jsonify({"error": "任务未在运行"}), 409


# ---------------- API: 统计 ----------------

# 磁盘占用统计缓存 (避免每次轮询都逐个 stat 文件)
_DISK_CACHE = {"ts": 0.0, "mb": 0.0}
DISK_CACHE_TTL = 30   # 秒


@app.route("/api/stats")
def api_stats():
    tasks = store.load_tasks()
    active = [t for t in tasks if not t.get("deleted_at")]
    trash_count = len(tasks) - len(active)
    runs_map = store.load_runs_map()      # 只全量读一次
    total_runs = ok = fail = imgs = 0
    for t in active:
        for r in runs_map.get(t["id"], []):
            total_runs += 1
            st = r.get("stats") or {}
            ok += st.get("ok", 0)
            fail += st.get("fail", 0)
            imgs += st.get("images", 0)
    with JOBS_LOCK:
        running = sum(1 for j in JOBS.values() if j.is_running())
    now = time.time()
    if now - _DISK_CACHE["ts"] > DISK_CACHE_TTL:
        size = 0
        seen = set()
        for t in active:
            d = os.path.realpath(resolve_out_dir(t))
            if d in seen or not os.path.isdir(d):
                continue
            seen.add(d)
            for root, _dirs, files in os.walk(d):
                for f in files:
                    try:
                        size += os.path.getsize(os.path.join(root, f))
                    except OSError:
                        pass
        _DISK_CACHE.update(ts=now, mb=round(size / 1048576, 1))
    return jsonify({
        "tasks": len(active),
        "running": running,
        "runs": total_runs,
        "ok": ok,
        "fail": fail,
        "images": imgs,
        "disk_mb": _DISK_CACHE["mb"],
        "trash": trash_count,
    })


# ---------------- API: 试爬取 ----------------

@app.route("/api/probe", methods=["POST"])
def api_probe():
    """试爬取: 按表单给出的规则探测起始页, 返回将爬取的链接清单"""
    body = request.get_json(force=True) or {}
    seed = (body.get("seed_url") or "").strip()
    if not seed:
        return jsonify({"error": "请填写起始网址"}), 400
    if not seed.startswith("http"):
        seed = "https://" + seed
    result = probe_links(
        seed,
        include=[x.strip() for x in (body.get("include") or []) if x.strip()],
        exclude=[x.strip() for x in (body.get("exclude") or []) if x.strip()],
        same_domain=body.get("same_domain", True),
        use_regex=bool(body.get("use_regex", False)),
        timeout=int(body.get("timeout") or 45),
    )
    return jsonify(result)


@app.route("/api/tasks/<tid>/clear_cache", methods=["POST"])
def api_clear_cache(tid):
    """清空自动任务的待爬缓存队列与已爬集合 (规则变更后重新开始用)"""
    task = store.get_task(tid)
    if not task:
        return jsonify({"error": "任务不存在"}), 404
    if task.get("mode") != "auto":
        return jsonify({"error": "仅自动爬取任务支持清空缓存"}), 400
    with JOBS_LOCK:
        job = JOBS.get(tid)
        if job and job.is_running():
            return jsonify({"error": "任务正在运行，无法清空缓存"}), 409
    auto = task.setdefault("auto", {})
    auto["pending"] = []
    auto["visited"] = []
    store.clear_auto_state(tid)          # 同时清空独立状态文件
    task["updated_at"] = now_str()
    store.upsert_task(task)
    db.upsert_task_async(task)
    return jsonify({"ok": True, "msg": "缓存队列已清空"})


@app.route("/api/tasks/<tid>/probe", methods=["POST"])
def api_task_probe(tid):
    """对已保存的自动任务执行试爬取 (使用保存的规则)"""
    task = store.get_task(tid)
    if not task:
        return jsonify({"error": "任务不存在"}), 404
    if task.get("mode") != "auto":
        return jsonify({"error": "仅自动爬取任务支持试爬取"}), 400
    auto = task.get("auto", {})
    result = probe_links(
        auto.get("seed_url", ""),
        include=auto.get("include", []),
        exclude=auto.get("exclude", []),
        same_domain=auto.get("same_domain", True),
        use_regex=bool(auto.get("use_regex", False)),
        timeout=int((task.get("config") or {}).get("timeout", 60)),
    )
    return jsonify(result)


# ---------------- API: 搜索 ----------------

@app.route("/api/search")
def api_search():
    """基本搜索: 对任务名/网址/标题做子串匹配 (仅遍历本地内存数据, 轻量)"""
    q = (request.args.get("q") or "").strip()
    if not q:
        return jsonify({"q": q, "count": 0, "results": []})
    ql = q.lower()
    results = []
    seen = set()
    runs_map = store.load_runs_map()      # 只全量读一次
    for task in store.load_tasks():
        if task.get("deleted_at"):
            continue                       # 回收站任务不参与搜索
        auto = task.get("auto") or {}
        task_hit = (
            ql in (task.get("name") or "").lower()
            or any(ql in u.lower() for u in task.get("urls", []))
            or ql in ((auto.get("seed_url") or "").lower())
        )
        for run in runs_map.get(task["id"], []):
            for r in run.get("results", []):
                url = r.get("url") or ""
                title = r.get("title") or ""
                if ql not in url.lower() and ql not in title.lower():
                    continue
                key = (task["id"], run.get("id"), url)
                if key in seen:
                    continue
                seen.add(key)
                results.append({
                    "type": "page",
                    "task_id": task["id"],
                    "task_name": task.get("name", ""),
                    "mode": task.get("mode", ""),
                    "run_id": run.get("id", ""),
                    "run_status": run.get("status", ""),
                    "url": url,
                    "title": title,
                    "status": r.get("status", ""),
                    "crawl_time": r.get("crawl_time", ""),
                    "html_file": r.get("html_file", ""),
                    "txt_file": r.get("txt_file", ""),
                    "meta_file": r.get("meta_file", ""),
                    "images": len(r.get("images", []) or []),
                })
                if len(results) >= 100:
                    break
            if len(results) >= 100:
                break
        if task_hit and len(results) < 100:
            results.append({
                "type": "task",
                "task_id": task["id"],
                "task_name": task.get("name", ""),
                "mode": task.get("mode", ""),
                "urls_count": len(task.get("urls", [])),
                "created_at": task.get("created_at", ""),
                "seed_url": auto.get("seed_url", ""),
            })
    return jsonify({"q": q, "count": len(results), "results": results})


# ---------------- API: 运行记录与文件 ----------------

@app.route("/api/runs/<rid>")
def api_run_detail(rid):
    run = store.get_run(rid)
    if not run:
        return jsonify({"error": "运行记录不存在"}), 404
    return jsonify(run)


@app.route("/api/runs/<rid>/logs")
def api_run_logs(rid):
    run = store.get_run(rid)
    if not run:
        return jsonify({"error": "运行记录不存在"}), 404
    offset = int(request.args.get("offset", 0))
    logs = run.get("logs", [])
    return jsonify({"logs": logs[offset:], "count": len(logs)})


@app.route("/api/file")
def api_file():
    tid = request.args.get("task_id", "")
    path = request.args.get("path", "")
    task = store.get_task(tid)
    if not task or not path:
        return jsonify({"error": "参数错误"}), 400
    out_dir = os.path.realpath(resolve_out_dir(task))
    full = os.path.realpath(os.path.join(out_dir, path))
    if not full.startswith(out_dir + os.sep) and full != out_dir:
        return jsonify({"error": "路径越界"}), 403
    if not os.path.isfile(full):
        return jsonify({"error": "文件不存在"}), 404
    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)


def migrate_auto_state():
    """启动时把旧任务对象里的 auto pending/visited 迁移到独立文件, 给 tasks.json 瘦身"""
    migrated = 0
    for t in store.load_tasks():
        if t.get("mode") != "auto":
            continue
        auto = t.get("auto") or {}
        if auto.get("pending") or auto.get("visited"):
            store.save_auto_state(t["id"], {
                "pending": auto.get("pending", []),
                "visited": auto.get("visited", []),
            })
            auto.pop("pending", None)
            auto.pop("visited", None)
            t["updated_at"] = now_str()
            store.upsert_task(t)
            migrated += 1
    if migrated:
        print(f"[universal-crawler] 已迁移 {migrated} 个 auto 任务的队列状态到独立文件", flush=True)


def main():
    os.makedirs(os.path.join(HERE, "data"), exist_ok=True)
    os.makedirs(os.path.join(HERE, "out"), exist_ok=True)
    db.init_db()
    migrate_auto_state()
    # 全量同步存量任务到数据库 (异步, 不阻塞启动)
    for t in store.load_tasks():
        db.upsert_task_async(t)
    scheduler.start()
    print(f"[universal-crawler] 启动完成, 管理界面: http://0.0.0.0:{PORT}/")
    app.run(host="0.0.0.0", port=PORT, threaded=True, debug=False)


if __name__ == "__main__":
    main()