"""文件夹批量处理引擎与 API 测试。

覆盖：视频扫描/同名文件夹推导、完成标记读写、批量任务创建校验、
引擎逐视频执行（建 run、复用现有调度器、复制最终产物、跳过已处理）、
暂停/继续断点续跑、失败视频不中断、孤儿清理不删批量运行、
批量 API 全端点（创建/列表/详情/暂停/继续/删除/下载）与 404/422 分支。
"""

import json
import time
from datetime import datetime, timezone
from pathlib import Path

import pytest
from fastapi.testclient import TestClient

from wov_app import batch as batch_engine
from wov_app.batch import (
    MARKER_NAME,
    PAUSE_FLAG,
    BatchWorker,
    create_job,
    load_marker,
    scan_videos,
    work_dir_for,
)
from wov_app.db import Database
from wov_app.main import app


def _now_iso() -> str:
    """返回当前 UTC 时间的 ISO 格式字符串。"""
    return datetime.now(timezone.utc).isoformat()


def _db(tmp_path) -> Database:
    """在临时目录创建独立数据库。"""
    return Database(tmp_path / "wov.db")


def _seed_echo_workflow(db: Database, workflow_id: str = "echo-app", published: bool = True) -> None:
    """创建引用 echo 节点的单节点工作流（真实数据流：输入文件复制为产物）。

    幂等：先删除同 ID 的旧工作流（含版本与任务），再重新创建。
    """
    definition = {
        "name": "echo-flow",
        "version": 1,
        "nodes": [
            {"id": "step", "node_type": "echo", "inputs": {"file_uri": "input.video_uri"}}
        ],
        "edges": [],
        "entry_inputs": {"video_uri": "file"},
        "final_outputs": {"result": "step.file_uri"},
    }
    # 批量测试会真实执行 echo 节点，注册表由 conftest 每测试隔离，需显式注册。
    _register_echo()
    if db.get_workflow(workflow_id) is not None:
        db.delete_workflow(workflow_id)
    db.upsert_workflow(
        {
            "id": workflow_id,
            "name": "Echo",
            "description": "",
            "published": 1 if published else 0,
            "latest_version": 1,
        }
    )
    db.create_workflow_version(workflow_id, 1, definition)


def _register_echo() -> None:
    """把内置 echo 节点注册到进程内注册表（conftest 每个测试隔离注册表）。"""
    from wov_app import registry
    from wov_sdk.models import NodeManifest

    root = Path(__file__).resolve().parent.parent
    from nodes.echo import invoke

    registry.register(NodeManifest.load(str(root / "manifests" / "echo.json")), invoke)

def _video_folder(tmp_path, names=("a.mp4", "b.mp4")) -> Path:
    """创建含视频文件的文件夹：内容为真实文本（echo 节点按文本读入）。"""
    folder = tmp_path / "videos"
    folder.mkdir()
    for index, name in enumerate(names, start=1):
        (folder / name).write_text(f"视频 {name} 的测试内容 {index}\n", encoding="utf-8")
    return folder


def _make_job(db: Database, folder: Path, workflow_id: str = "echo-app", recursive: bool = True) -> dict:
    """通过 create_job 创建批量任务并返回任务记录。"""
    job_id = create_job(db, str(folder), workflow_id, recursive)
    return db.get_batch_job(job_id)


# ---------------------------------------------------------------------------
# 扫描 / 目录推导 / 完成标记
# ---------------------------------------------------------------------------


def test_scan_videos_recursive_and_work_dir(tmp_path) -> None:
    """递归/非递归扫描只返回视频文件，同名文件夹去掉扩展名。"""
    folder = tmp_path / "media"
    (folder / "sub").mkdir(parents=True)
    (folder / "a.mp4").write_text("a", encoding="utf-8")
    (folder / "b.MKV").write_text("b", encoding="utf-8")
    (folder / "readme.txt").write_text("c", encoding="utf-8")
    (folder / "sub" / "c.avi").write_text("d", encoding="utf-8")
    (folder / "sub" / "notes.md").write_text("e", encoding="utf-8")

    recursive = scan_videos(folder, recursive=True)
    assert [p.name for p in recursive] == ["a.mp4", "b.MKV", "c.avi"]
    flat = scan_videos(folder, recursive=False)
    assert [p.name for p in flat] == ["a.mp4", "b.MKV"]
    # 同名文件夹：去掉扩展名，位于视频所在目录。
    assert work_dir_for(folder / "sub" / "c.avi") == folder / "sub" / "c"


def test_load_marker_variants(tmp_path) -> None:
    """完成标记缺失/损坏/非字典时返回 None。"""
    work = tmp_path / "movie"
    work.mkdir()
    assert load_marker(work) is None
    (work / MARKER_NAME).write_text("{broken json", encoding="utf-8")
    assert load_marker(work) is None
    (work / MARKER_NAME).write_text("[1,2]", encoding="utf-8")
    assert load_marker(work) is None
    (work / MARKER_NAME).write_text('{"workflow_id": "w", "finals": {"r": "m.txt"}}', encoding="utf-8")
    marker = load_marker(work)
    assert marker["workflow_id"] == "w"


# ---------------------------------------------------------------------------
# create_job 校验
# ---------------------------------------------------------------------------


def test_create_job_validation_errors(tmp_path) -> None:
    """文件夹不存在、未发布工作流、无版本、无视频都拒绝创建。"""
    db = _db(tmp_path)
    folder = _video_folder(tmp_path)
    with pytest.raises(ValueError, match="folder not found"):
        create_job(db, str(tmp_path / "missing"), "echo-app")
    with pytest.raises(ValueError, match="published workflow not found"):
        create_job(db, str(folder), "ghost-flow")
    _seed_echo_workflow(db, published=False)
    with pytest.raises(ValueError, match="published workflow not found"):
        create_job(db, str(folder), "echo-app")
    # 有版本但文件夹里没有视频。
    _seed_echo_workflow(db, published=True)
    empty = tmp_path / "empty"
    empty.mkdir()
    with pytest.raises(ValueError, match="no videos found in folder"):
        create_job(db, str(empty), "echo-app")
    # 已发布但没有版本的工作流。
    db.delete_workflow("echo-app")
    db.upsert_workflow({"id": "echo-app", "name": "E", "description": "", "published": 1, "latest_version": 0})
    with pytest.raises(ValueError, match="workflow has no version"):
        create_job(db, str(folder), "echo-app")


def test_create_job_success_creates_rows(tmp_path) -> None:
    """创建成功：任务入队、视频明细齐全、递归标志与文件夹路径落库。"""
    db = _db(tmp_path)
    _seed_echo_workflow(db)
    (tmp_path / "videos" / "sub").mkdir(parents=True)
    (tmp_path / "videos" / "a.mp4").write_text("a", encoding="utf-8")
    (tmp_path / "videos" / "sub" / "b.mkv").write_text("b", encoding="utf-8")
    job = _make_job(db, tmp_path / "videos", recursive=True)
    assert job["status"] == "QUEUED"
    videos = db.list_batch_videos(job["id"])
    assert {v["video_path"] for v in videos} == {
        str(tmp_path / "videos" / "a.mp4"),
        str(tmp_path / "videos" / "sub" / "b.mkv"),
    }
    assert all(v["status"] == "PENDING" for v in videos)
    assert db.next_queued_batch_job()["id"] == job["id"]


# ---------------------------------------------------------------------------
# 引擎：完整处理 / 跳过 / 失败 / 暂停续跑
# ---------------------------------------------------------------------------


def test_batch_worker_processes_all_videos(tmp_path) -> None:
    """批量引擎逐个处理视频：建 source=batch 的 run、中间态进同名文件夹、
    最终产物复制到同名文件夹根目录并写完成标记，任务最终 COMPLETED。"""
    db = _db(tmp_path)
    _seed_echo_workflow(db)
    folder = _video_folder(tmp_path)
    job = _make_job(db, folder)
    worker = BatchWorker(db, interval_seconds=0.05)
    worker._process_job(job)

    job = db.get_batch_job(job["id"])
    assert job["status"] == "COMPLETED"
    assert job["done"] == 2 and job["failed"] == 0
    for name in ("a", "b"):
        work = folder / name
        marker = load_marker(work)
        assert marker is not None and marker["workflow_id"] == "echo-app"
        final_name = marker["finals"]["result"]
        final = work / final_name
        # 最终产物复制到同名文件夹根目录，内容与输入一致（真实数据流）。
        assert final.is_file()
        assert final.read_text(encoding="utf-8") == (folder / f"{name}.mp4").read_text(encoding="utf-8")
        # 中间态落在 <work>/runs/<run_id>/steps/ 下。
        # 中间态落在 <work>/runs/<run_id>/steps/ 下（收尾时产物已重命名）。
        assert list(work.glob("runs/*/steps/step/*.txt"))
    # run 记录为 batch 来源，且主调度器不会抢占。
    runs = [db.get_run(v["run_id"]) for v in db.list_batch_videos(job["id"])]
    assert all(run["source"] == "batch" and run["status"] == "COMPLETED" for run in runs)
    assert db.next_queued_run() is None


def test_batch_worker_skips_already_done_videos(tmp_path) -> None:
    """已有同工作流完成标记且产物齐全的视频直接跳过（不再处理）。"""
    db = _db(tmp_path)
    _seed_echo_workflow(db)
    folder = _video_folder(tmp_path, names=("a.mp4", "b.mp4"))
    work_a = folder / "a"
    work_a.mkdir()
    # 伪造 a.mp4 的完成标记与最终产物（模拟上次任务已处理完）。
    marker = {"workflow_id": "echo-app", "workflow_version": 1, "run_id": "run_old", "finals": {"result": "a.result.txt"}}
    (work_a / MARKER_NAME).write_text(json.dumps(marker), encoding="utf-8")
    (work_a / "a.result.txt").write_text("已处理", encoding="utf-8")

    job = _make_job(db, folder)
    BatchWorker(db, interval_seconds=0.05)._process_job(job)
    videos = {Path(v["video_path"]).name: v for v in db.list_batch_videos(job["id"])}
    assert videos["a.mp4"]["status"] == "SKIPPED"
    assert videos["a.mp4"]["run_id"] is None
    assert videos["b.mp4"]["status"] == "COMPLETED"
    assert db.get_batch_job(job["id"])["done"] == 2


def test_batch_worker_marker_workflow_mismatch_reprocesses(tmp_path) -> None:
    """不同工作流的完成标记不互相误判：换工作流后视频重新处理。"""
    db = _db(tmp_path)
    _seed_echo_workflow(db, workflow_id="echo-app")
    _seed_echo_workflow(db, workflow_id="echo-other")
    folder = _video_folder(tmp_path, names=("a.mp4",))
    work_a = folder / "a"
    work_a.mkdir()
    marker = {"workflow_id": "echo-other", "finals": {"result": "x.txt"}}
    (work_a / MARKER_NAME).write_text(json.dumps(marker), encoding="utf-8")
    (work_a / "x.txt").write_text("x", encoding="utf-8")

    job = _make_job(db, folder, workflow_id="echo-app")
    BatchWorker(db, interval_seconds=0.05)._process_job(job)
    video = db.list_batch_videos(job["id"])[0]
    # 标记工作流不匹配 → 重新处理并覆盖为新工作流的标记。
    assert video["status"] == "COMPLETED"
    new_marker = load_marker(work_a)
    assert new_marker["workflow_id"] == "echo-app"


def test_batch_worker_failed_video_continues(tmp_path) -> None:
    """缺失视频文件与失败 run 都记为 FAILED，任务继续处理后续视频。"""
    db = _db(tmp_path)
    _seed_echo_workflow(db)
    folder = _video_folder(tmp_path, names=("gone.mp4", "b.mp4"))
    job = _make_job(db, folder)
    # 任务创建后、处理前删除第一个视频（模拟外部移除），第二个正常处理。
    (folder / "gone.mp4").unlink()
    BatchWorker(db, interval_seconds=0.05)._process_job(job)

    videos = {Path(v["video_path"]).name: v for v in db.list_batch_videos(job["id"])}
    assert videos["gone.mp4"]["status"] == "FAILED"
    assert videos["gone.mp4"]["error"] == "video file not found"
    assert videos["b.mp4"]["status"] == "COMPLETED"
    job = db.get_batch_job(job["id"])
    assert job["status"] == "COMPLETED"
    assert job["done"] == 1 and job["failed"] == 1


class _PausingScheduler:
    """把 execute_run 模拟为"被暂停"的假调度器：置 run 为 PAUSED。"""

    def __init__(self, db: Database, work_dir: Path) -> None:
        self.db = db
        self.work_dir = work_dir

    def execute_run(self, run_id: str) -> None:
        self.db.update_run(run_id, status="PAUSED", updated_at=_now_iso())


def test_batch_worker_pause_then_resume_continues(tmp_path, monkeypatch) -> None:
    """暂停后重新开始：PAUSED 视频从断点续跑，未开始的视频接着处理。"""
    db = _db(tmp_path)
    _seed_echo_workflow(db)
    folder = _video_folder(tmp_path, names=("a.mp4", "b.mp4"))
    job = _make_job(db, folder)
    worker = BatchWorker(db, interval_seconds=0.05)

    # 第一次执行：第一个视频处理中被暂停（假调度器把 run 置为 PAUSED）。
    monkeypatch.setattr("wov_app.batch.WorkflowScheduler", _PausingScheduler)
    worker._process_job(job)
    videos = {Path(v["video_path"]).name: v for v in db.list_batch_videos(job["id"])}
    assert videos["a.mp4"]["status"] == "PAUSED"
    assert videos["b.mp4"]["status"] == "PENDING"
    assert db.get_batch_job(job["id"])["status"] == "PAUSED"

    # 恢复真实调度器并继续：a 从断点完成，b 接着处理，任务 COMPLETED。
    monkeypatch.undo()
    worker.resume_job(job["id"])
    worker._process_job(db.get_batch_job(job["id"]))
    videos = {Path(v["video_path"]).name: v for v in db.list_batch_videos(job["id"])}
    assert videos["a.mp4"]["status"] == "COMPLETED"
    assert videos["b.mp4"]["status"] == "COMPLETED"
    assert db.get_batch_job(job["id"])["status"] == "COMPLETED"

    # 再次处理（已完成视频在循环里直接 continue）：结果不变，幂等。
    worker._process_job(db.get_batch_job(job["id"]))
    videos = {Path(v["video_path"]).name: v for v in db.list_batch_videos(job["id"])}
    assert videos["a.mp4"]["status"] == "COMPLETED"
    assert videos["b.mp4"]["status"] == "COMPLETED"
    assert db.get_batch_job(job["id"])["done"] == 2


def test_batch_worker_paused_job_does_not_start_new_video(tmp_path, monkeypatch) -> None:
    """任务在视频之间被暂停：后续视频不开始，不创建 run。"""
    db = _db(tmp_path)
    _seed_echo_workflow(db)
    folder = _video_folder(tmp_path, names=("a.mp4", "b.mp4"))
    job = _make_job(db, folder)
    job_id = job["id"]
    from wov_app.scheduler import WorkflowScheduler

    class _PauseAfterFirstScheduler:
        """真实执行第一个视频后把批量任务置为 PAUSED（模拟用户处理中暂停）。"""

        def __init__(self, db: Database, work_dir: Path) -> None:
            self.db = db
            self.work_dir = work_dir

        def execute_run(self, run_id: str) -> None:
            WorkflowScheduler(self.db, self.work_dir).execute_run(run_id)
            self.db.update_batch_job(job_id, status="PAUSED", updated_at=_now_iso())

    monkeypatch.setattr("wov_app.batch.WorkflowScheduler", _PauseAfterFirstScheduler)
    BatchWorker(db, interval_seconds=0.05)._process_job(db.get_batch_job(job_id))
    videos = {Path(v["video_path"]).name: v for v in db.list_batch_videos(job_id)}
    # 第一个视频真实完成；第二个视频在开始前因任务已暂停而不处理。
    assert videos["a.mp4"]["status"] == "COMPLETED"
    assert videos["b.mp4"]["status"] == "PENDING"
    assert videos["b.mp4"]["run_id"] is None
    assert db.get_batch_job(job_id)["status"] == "PAUSED"

class _FailingScheduler:
    """把 execute_run 模拟为"节点失败"的假调度器。"""

    def __init__(self, db: Database, work_dir: Path) -> None:
        self.db = db

    def execute_run(self, run_id: str) -> None:
        self.db.update_run(run_id, status="FAILED", error="node boom", updated_at=_now_iso())


def test_batch_worker_run_failed_marks_video_failed(tmp_path, monkeypatch) -> None:
    """节点执行失败：run FAILED → 视频 FAILED 带错误信息，任务继续。"""
    db = _db(tmp_path)
    _seed_echo_workflow(db)
    folder = _video_folder(tmp_path, names=("a.mp4", "b.mp4"))
    job = _make_job(db, folder)
    monkeypatch.setattr("wov_app.batch.WorkflowScheduler", _FailingScheduler)
    BatchWorker(db, interval_seconds=0.05)._process_job(job)
    videos = {Path(v["video_path"]).name: v for v in db.list_batch_videos(job["id"])}
    assert videos["a.mp4"]["status"] == "FAILED"
    assert videos["a.mp4"]["error"] == "node boom"
    assert videos["b.mp4"]["status"] == "FAILED"
    assert db.get_batch_job(job["id"])["status"] == "COMPLETED"
    assert db.get_batch_job(job["id"])["failed"] == 2


def test_batch_worker_resumes_failed_and_running_runs(tmp_path) -> None:
    """已失败的 run 重跑、上次进程残留的 RUNNING run 恢复后继续。"""
    db = _db(tmp_path)
    _seed_echo_workflow(db)
    folder = _video_folder(tmp_path, names=("a.mp4", "b.mp4"))
    job = _make_job(db, folder)
    now = _now_iso()
    # 手动构造 run：a 失败（重跑）、b 残留 RUNNING（进程被杀后恢复）。
    for name, status in (("a", "FAILED"), ("b", "RUNNING")):
        run_id = f"run_{name}"
        db.create_run(
            {
                "id": run_id,
                "workflow_id": "echo-app",
                "workflow_version": 1,
                "status": status,
                "progress": 0,
                "input_uri": str(folder / f"{name}.mp4"),
                "param_overrides": None,
                "source": "batch",
                "created_at": now,
                "updated_at": now,
            }
        )
        video = next(v for v in db.list_batch_videos(job["id"]) if v["video_path"].endswith(f"{name}.mp4"))
        db.update_batch_video(video["id"], run_id=run_id, updated_at=now)

    BatchWorker(db, interval_seconds=0.05)._process_job(db.get_batch_job(job["id"]))
    videos = {Path(v["video_path"]).name: v for v in db.list_batch_videos(job["id"])}
    assert videos["a.mp4"]["status"] == "COMPLETED"
    assert videos["b.mp4"]["status"] == "COMPLETED"


def test_batch_worker_retry_failed_keeps_completed_node_artifacts(tmp_path) -> None:
    """批量重试 FAILED 视频**保留已完成节点产物**：只重跑失败节点。

    回归：此前 FAILED 走 reset_run 清空全部产物记录，重跑时 extract/ocr 等
    长耗时节点从头重做（run_e2b74e89e232 的 22222 帧 OCR 被白白丢弃）。
    现改为保留产物恢复 QUEUED，execute_run 从产物表跳过已完成节点。
    """
    db = _db(tmp_path)
    _register_echo()
    # 双节点串联工作流：step1 成功、step2 失败的场景下验证只重跑 step2。
    definition = {
        "name": "two-step",
        "version": 1,
        "nodes": [
            {"id": "s1", "node_type": "echo", "inputs": {"file_uri": "input.video_uri"}},
            {"id": "s2", "node_type": "echo", "inputs": {"file_uri": "s1.file_uri"}},
        ],
        "edges": [{"from": "s1", "to": "s2"}],
        "entry_inputs": {"video_uri": "file"},
        "final_outputs": {"result": "s2.file_uri"},
    }
    db.upsert_workflow({"id": "echo-app", "name": "Echo", "description": "", "published": 1, "latest_version": 1})
    db.create_workflow_version("echo-app", 1, definition)
    folder = _video_folder(tmp_path, names=("a.mp4",))
    job = _make_job(db, folder)
    now = _now_iso()
    run_id = "run_retry"
    db.create_run(
        {
            "id": run_id,
            "workflow_id": "echo-app",
            "workflow_version": 1,
            "status": "FAILED",
            "progress": 0.5,
            "error": "节点2炸了",
            "input_uri": str(folder / "a.mp4"),
            "param_overrides": None,
            "source": "batch",
            "created_at": now,
            "updated_at": now,
        }
    )
    # 模拟 step1 已完成并登记产物（step2 失败时 step1 的成果）。
    s1_out = folder / "a" / "runs" / run_id / "steps" / "s1" / "echo.txt"
    s1_out.parent.mkdir(parents=True)
    s1_out.write_text("step1 产物", encoding="utf-8")
    db.create_artifact(
        {
            "run_id": run_id,
            "node_id": "s1",
            "name": "s1.file_uri",
            "uri": str(s1_out),
            "mime_type": "text/plain",
            "size": 10,
        }
    )
    video = db.list_batch_videos(job["id"])[0]
    db.update_batch_video(video["id"], run_id=run_id, updated_at=now)

    BatchWorker(db, interval_seconds=0.05)._process_job(db.get_batch_job(job["id"]))
    video = db.list_batch_videos(job["id"])[0]
    assert video["status"] == "COMPLETED"
    run = db.get_run(run_id)
    assert run["status"] == "COMPLETED" and run["error"] is None
    # step1 产物记录保留且未被重写（节点没有重新执行）。
    artifacts = {a["name"]: a for a in db.list_artifacts(run_id)}
    assert artifacts["s1.file_uri"]["uri"] == str(s1_out)
    # step2 本次补做完成；收尾时最终产物被重命名为 <片名>.result.<时间戳>.txt。
    assert "s2.file_uri" in artifacts
    assert list((folder / "a" / "runs" / run_id / "steps" / "s2").glob("*"))
    finals = {a["name"]: a for a in db.list_artifacts(run_id)}
    assert "result" in finals
    assert Path(finals["result"]["uri"]).is_file()

def test_batch_worker_already_completed_run_copies_finals(tmp_path) -> None:
    """run 已完成但视频未标记（收尾前中断）：直接复制产物并标记完成。

    覆盖 _copy_finals 的三个分支：产物齐全（复制）、产物记录存在但文件丢失
    （跳过）、无产物记录（跳过）——完成后均写出完成标记。
    """
    db = _db(tmp_path)
    _seed_echo_workflow(db)
    folder = _video_folder(tmp_path, names=("a.mp4", "b.mp4", "c.mp4"))
    job = _make_job(db, folder)
    now = _now_iso()
    videos = {Path(v["video_path"]).stem: v for v in db.list_batch_videos(job["id"])}

    def complete_run(run_id: str, name: str, with_file: bool, with_artifact: bool) -> None:
        """创建 COMPLETED run；可选产物文件与产物记录。"""
        db.create_run(
            {
                "id": run_id,
                "workflow_id": "echo-app",
                "workflow_version": 1,
                "status": "COMPLETED",
                "progress": 1,
                "input_uri": str(folder / f"{name}.mp4"),
                "param_overrides": None,
                "source": "batch",
                "created_at": now,
                "updated_at": now,
            }
        )
        if with_artifact:
            db.create_artifact(
                {
                    "run_id": run_id,
                    "node_id": "step",
                    "name": "result",
                    "uri": str(folder / name / "runs" / run_id / "steps" / "step" / f"{name}.result.txt"),
                    "mime_type": "text/plain",
                    "size": 6,
                }
            )
        if with_file:
            path = folder / name / "runs" / run_id / "steps" / "step" / f"{name}.result.txt"
            path.parent.mkdir(parents=True, exist_ok=True)
            path.write_text("产物", encoding="utf-8")
        db.update_batch_video(videos[name]["id"], run_id=run_id, updated_at=now)

    # a：产物齐全；b：产物记录存在但文件丢失；c：没有任何产物记录。
    complete_run("run_done_a", "a", with_file=True, with_artifact=True)
    complete_run("run_done_b", "b", with_file=False, with_artifact=True)
    complete_run("run_done_c", "c", with_file=False, with_artifact=False)

    BatchWorker(db, interval_seconds=0.05)._process_job(db.get_batch_job(job["id"]))
    videos = {Path(v["video_path"]).stem: v for v in db.list_batch_videos(job["id"])}
    assert all(videos[name]["status"] == "COMPLETED" for name in ("a", "b", "c"))
    # a 的产物被复制到同名文件夹根目录；b/c 无产物可复制，标记为空。
    marker_a = load_marker(folder / "a")
    assert marker_a["finals"]["result"] == "a.result.txt"
    assert (folder / "a" / "a.result.txt").is_file()
    assert load_marker(folder / "b")["finals"] == {}
    assert load_marker(folder / "c")["finals"] == {}


# ---------------------------------------------------------------------------
# 引擎：任务级异常与校验失败
# ---------------------------------------------------------------------------


def test_batch_worker_job_validation_failures(tmp_path) -> None:
    """文件夹缺失/工作流未发布/无版本时任务置为 FAILED 并记录错误。"""
    db = _db(tmp_path)
    now = _now_iso()

    def add_job(job_id, folder, workflow_id="echo-app"):
        db.create_batch_job(
            {
                "id": job_id, "folder_path": folder, "workflow_id": workflow_id,
                "recursive": 1, "status": "QUEUED", "progress": 0, "total": 0,
                "done": 0, "failed": 0, "error": None,
                "created_at": now, "updated_at": now,
            }
        )

    worker = BatchWorker(db, interval_seconds=0.05)
    add_job("job_nofolder", str(tmp_path / "missing"))
    worker._process_job(db.get_batch_job("job_nofolder"))
    assert db.get_batch_job("job_nofolder")["status"] == "FAILED"
    assert "folder not found" in db.get_batch_job("job_nofolder")["error"]

    folder = _video_folder(tmp_path)
    _seed_echo_workflow(db, published=False)
    add_job("job_unpublished", str(folder))
    worker._process_job(db.get_batch_job("job_unpublished"))
    assert db.get_batch_job("job_unpublished")["status"] == "FAILED"
    assert "not found or unpublished" in db.get_batch_job("job_unpublished")["error"]

    # 已发布但没有任何版本。
    _seed_echo_workflow(db, published=True)
    db.delete_workflow("echo-app")
    db.upsert_workflow({"id": "echo-app", "name": "E", "description": "", "published": 1, "latest_version": 0})
    add_job("job_noversion", str(folder))
    worker._process_job(db.get_batch_job("job_noversion"))
    assert db.get_batch_job("job_noversion")["status"] == "FAILED"
    assert "has no version" in db.get_batch_job("job_noversion")["error"]


def test_batch_worker_catches_unexpected_job_error(tmp_path) -> None:
    """任务级兜底：DAG 解析失败（缺 name）时任务 FAILED 而不是卡死在 RUNNING。"""
    db = _db(tmp_path)
    _seed_echo_workflow(db)
    folder = _video_folder(tmp_path, names=("a.mp4",))
    # 覆盖为缺失 name 的非法定义：from_dict 抛 KeyError。
    bad_definition = {"version": 1, "nodes": [], "edges": []}
    db.create_workflow_version("echo-app", 2, bad_definition)
    db.upsert_workflow({"id": "echo-app", "name": "E", "description": "", "published": 1, "latest_version": 2})
    job = _make_job(db, folder)
    BatchWorker(db, interval_seconds=0.05)._process_job(job)
    assert db.get_batch_job(job["id"])["status"] == "FAILED"
    assert "name" in db.get_batch_job(job["id"])["error"]


def test_batch_worker_cycle_fails_video_not_job(tmp_path) -> None:
    """DAG 环在执行期抛错：单个视频 FAILED 记录错误，批量任务继续并完成。"""
    db = _db(tmp_path)
    _seed_echo_workflow(db)
    folder = _video_folder(tmp_path, names=("a.mp4",))
    # 覆盖为带环的定义：validate 通过、调度器拓扑排序时抛"contains a cycle"。
    cycle_definition = {
        "name": "bad",
        "version": 2,
        "nodes": [
            {"id": "x", "node_type": "echo", "inputs": {"file_uri": "y.file_uri"}},
            {"id": "y", "node_type": "echo", "inputs": {"file_uri": "x.file_uri"}},
        ],
        "edges": [{"from": "x", "to": "y"}, {"from": "y", "to": "x"}],
        "entry_inputs": {"video_uri": "file"},
        "final_outputs": {"result": "x.file_uri"},
    }
    db.create_workflow_version("echo-app", 2, cycle_definition)
    db.upsert_workflow({"id": "echo-app", "name": "E", "description": "", "published": 1, "latest_version": 2})
    job = _make_job(db, folder)
    BatchWorker(db, interval_seconds=0.05)._process_job(job)
    video = db.list_batch_videos(job["id"])[0]
    assert video["status"] == "FAILED"
    assert "cycle" in video["error"]
    assert db.get_batch_job(job["id"])["status"] == "COMPLETED"
    assert db.get_batch_job(job["id"])["failed"] == 1


def test_batch_worker_loop_processes_and_survives_exceptions(tmp_path, monkeypatch) -> None:
    """轮询线程：处理排队任务；轮询异常不杀死线程；重复启动/幽灵任务无害。"""
    db = _db(tmp_path)
    _seed_echo_workflow(db)
    folder = _video_folder(tmp_path, names=("a.mp4",))
    job = _make_job(db, folder)
    # 不存在的任务 ID：_run_job 直接返回，不报错。
    BatchWorker(db, interval_seconds=0.05)._process_job({"id": "ghost_job"})
    # 第一次轮询抛异常（模拟数据库抖动），后续正常。
    # 第一次轮询抛异常（模拟数据库抖动），后续正常。
    calls = {"n": 0}
    real_next = db.next_queued_batch_job

    def flaky_next():
        calls["n"] += 1
        if calls["n"] == 1:
            raise RuntimeError("transient error")
        return real_next()

    monkeypatch.setattr(db, "next_queued_batch_job", flaky_next)
    worker = BatchWorker(db, interval_seconds=0.05)
    worker.start()
    # 重复启动无副作用：线程已存在时直接返回。
    worker.start()
    try:
        deadline = time.monotonic() + 10
        while time.monotonic() < deadline:
            if db.get_batch_job(job["id"])["status"] in {"COMPLETED", "FAILED"}:
                break
            time.sleep(0.1)
    finally:
        worker.stop()
    assert db.get_batch_job(job["id"])["status"] == "COMPLETED"


def test_batch_worker_pause_job_writes_flag_and_pauses_run(tmp_path) -> None:
    """pause_job：任务置 PAUSED、排队/运行中的 run 暂停并写 paused.flag。"""
    db = _db(tmp_path)
    _seed_echo_workflow(db)
    folder = _video_folder(tmp_path, names=("a.mp4", "b.mp4", "c.mp4"))
    job = _make_job(db, folder)
    now = _now_iso()
    videos = {Path(v["video_path"]).stem: v for v in db.list_batch_videos(job["id"])}

    # a：QUEUED 运行（应被暂停并写 flag）；b：run 记录不存在；c：已完成的 run。
    db.create_run(
        {
            "id": "run_pause_me",
            "workflow_id": "echo-app",
            "workflow_version": 1,
            "status": "QUEUED",
            "progress": 0,
            "input_uri": str(folder / "a.mp4"),
            "param_overrides": None,
            "source": "batch",
            "created_at": now,
            "updated_at": now,
        }
    )
    db.create_run(
        {
            "id": "run_done_c",
            "workflow_id": "echo-app",
            "workflow_version": 1,
            "status": "COMPLETED",
            "progress": 1,
            "input_uri": str(folder / "c.mp4"),
            "param_overrides": None,
            "source": "batch",
            "created_at": now,
            "updated_at": now,
        }
    )
    db.update_batch_video(videos["a"]["id"], run_id="run_pause_me", updated_at=now)
    db.update_batch_video(videos["b"]["id"], run_id="run_ghost", updated_at=now)
    db.update_batch_video(videos["c"]["id"], run_id="run_done_c", updated_at=now)

    worker = BatchWorker(db, interval_seconds=0.05)
    worker.pause_job(job["id"])
    assert db.get_batch_job(job["id"])["status"] == "PAUSED"
    assert db.get_run("run_pause_me")["status"] == "PAUSED"
    assert (folder / "a" / "runs" / "run_pause_me" / PAUSE_FLAG).is_file()
    # b 的 run 不存在、c 的 run 已完成：都被跳过，不写 flag。
    assert not (folder / "b" / "runs" / "run_ghost" / PAUSE_FLAG).exists()
    assert not (folder / "c" / "runs" / "run_done_c" / PAUSE_FLAG).exists()


# ---------------------------------------------------------------------------
# 批量 API
# ---------------------------------------------------------------------------


def _client_with_echo_workflow(tmp_path):
    """返回 TestClient 与包含真实视频文件的文件夹（echo 工作流已发布）。"""
    folder = _video_folder(tmp_path, names=("a.mp4", "b.mp4"))
    client = TestClient(app)
    client.__enter__()
    client.post(
        "/api/admin/workflows",
        json={
            "id": "echo-app",
            "name": "Echo App",
            "description": "batch test",
            "definition": {
                "name": "echo-flow",
                "version": 1,
                "nodes": [
                    {"id": "step", "node_type": "echo", "inputs": {"file_uri": "input.video_uri"}}
                ],
                "edges": [],
                "entry_inputs": {"video_uri": "file"},
                "final_outputs": {"result": "step.file_uri"},
            },
        },
    )
    client.post("/api/admin/workflows/echo-app/publish")
    return client, folder


def test_batch_api_create_list_detail(tmp_path) -> None:
    """批量 API：创建任务、列表、详情（含视频明细与完成标记产物）。"""
    client, folder = _client_with_echo_workflow(tmp_path)
    try:
        response = client.post(
            "/api/batch/jobs",
            json={"folder": str(folder), "workflow_id": "echo-app", "recursive": True},
        )
        assert response.status_code == 200
        job = response.json()
        assert job["status"] == "QUEUED"
        assert len(job["videos"]) == 2
        assert job["videos"][0]["finals"] == {}

        listed = client.get("/api/batch/jobs").json()
        assert any(item["id"] == job["id"] for item in listed)

        detail = client.get(f"/api/batch/jobs/{job['id']}").json()
        assert detail["workflow_id"] == "echo-app"
        assert len(detail["videos"]) == 2

        missing = client.get("/api/batch/jobs/ghost")
        assert missing.status_code == 404
    finally:
        client.__exit__(None, None, None)


def test_batch_api_creation_errors(tmp_path) -> None:
    """批量 API 拒绝：文件夹不存在、无视频、未发布工作流。"""
    client, folder = _client_with_echo_workflow(tmp_path)
    try:
        bad_folder = client.post(
            "/api/batch/jobs",
            json={"folder": str(tmp_path / "missing"), "workflow_id": "echo-app"},
        )
        assert bad_folder.status_code == 422
        empty = tmp_path / "empty"
        empty.mkdir()
        no_videos = client.post(
            "/api/batch/jobs",
            json={"folder": str(empty), "workflow_id": "echo-app"},
        )
        assert no_videos.status_code == 422
        not_published = client.post(
            "/api/batch/jobs",
            json={"folder": str(folder), "workflow_id": "ghost"},
        )
        assert not_published.status_code == 422
    finally:
        client.__exit__(None, None, None)


def test_batch_api_pause_resume_delete(tmp_path) -> None:
    """批量 API：暂停/继续切换任务状态，删除清理数据库记录（含 run）。"""
    client, folder = _client_with_echo_workflow(tmp_path)
    try:
        job = client.post(
            "/api/batch/jobs",
            json={"folder": str(folder), "workflow_id": "echo-app"},
        ).json()

        paused = client.post(f"/api/batch/jobs/{job['id']}/pause")
        assert paused.status_code == 200
        assert client.get(f"/api/batch/jobs/{job['id']}").json()["status"] == "PAUSED"

        resumed = client.post(f"/api/batch/jobs/{job['id']}/resume")
        assert resumed.status_code == 200
        assert client.get(f"/api/batch/jobs/{job['id']}").json()["status"] == "QUEUED"

        # 给第一个视频挂一个 run，验证删除任务时级联删除 run 记录。
        db = app.state.db
        db.create_run(
            {
                "id": "run_del",
                "workflow_id": "echo-app",
                "workflow_version": 1,
                "status": "QUEUED",
                "progress": 0,
                "input_uri": str(folder / "a.mp4"),
                "param_overrides": None,
                "source": "batch",
                "created_at": _now_iso(),
                "updated_at": _now_iso(),
            }
        )
        video = job["videos"][0]
        db.update_batch_video(video["id"], run_id="run_del", updated_at=_now_iso())
        assert db.get_run("run_del") is not None

        deleted = client.delete(f"/api/batch/jobs/{job['id']}")
        assert deleted.status_code == 200
        assert client.get(f"/api/batch/jobs/{job['id']}").status_code == 404
        assert db.get_run("run_del") is None

        assert client.post("/api/batch/jobs/ghost/pause").status_code == 404
        assert client.post("/api/batch/jobs/ghost/resume").status_code == 404
        assert client.delete("/api/batch/jobs/ghost").status_code == 404
    finally:
        client.__exit__(None, None, None)


def test_batch_api_worker_unavailable(tmp_path, monkeypatch) -> None:
    """批量引擎不可用时，暂停/继续接口返回 503。"""
    client, folder = _client_with_echo_workflow(tmp_path)
    try:
        job = client.post(
            "/api/batch/jobs",
            json={"folder": str(folder), "workflow_id": "echo-app"},
        ).json()
        monkeypatch.setattr("wov_app.routers.batch._get_worker", lambda: None)
        assert client.post(f"/api/batch/jobs/{job['id']}/pause").status_code == 503
        assert client.post(f"/api/batch/jobs/{job['id']}/resume").status_code == 503
    finally:
        client.__exit__(None, None, None)


def test_batch_api_download_artifact(tmp_path) -> None:
    """批量 API：从完成标记下载最终产物，缺失别名/文件返回 404。"""
    client, folder = _client_with_echo_workflow(tmp_path)
    try:
        job = client.post(
            "/api/batch/jobs",
            json={"folder": str(folder), "workflow_id": "echo-app"},
        ).json()
        video = job["videos"][0]
        db = app.state.db
        # 直接构造完成状态：run + 产物 + 同名文件夹完成标记。
        # 产物文件名与下载端点约定一致：完成标记里的文件名位于同名文件夹根。
        run_id = "run_dl"
        work = folder / "a"
        steps = work / "runs" / run_id / "steps" / "step"
        steps.mkdir(parents=True)
        (steps / "a.result.20260819000000.txt").write_text("下载内容", encoding="utf-8")
        (work / "a.result.20260819000000.txt").write_text("下载内容", encoding="utf-8")
        db.create_run(
            {
                "id": run_id,
                "workflow_id": "echo-app",
                "workflow_version": 1,
                "status": "COMPLETED",
                "progress": 1,
                "input_uri": str(folder / "a.mp4"),
                "param_overrides": None,
                "source": "batch",
                "created_at": _now_iso(),
                "updated_at": _now_iso(),
            }
        )
        db.create_artifact(
            {
                "run_id": run_id,
                "node_id": "step",
                "name": "result",
                "uri": str(steps / "a.result.20260819000000.txt"),
                "mime_type": "text/plain",
                "size": 8,
            }
        )
        marker = {"workflow_id": "echo-app", "workflow_version": 1, "run_id": run_id, "finals": {"result": "a.result.20260819000000.txt"}}
        (work / MARKER_NAME).write_text(json.dumps(marker), encoding="utf-8")

        ok = client.get(f"/api/batch/jobs/{job['id']}/videos/{video['id']}/download?alias=result")
        assert ok.status_code == 200
        assert ok.content == "下载内容".encode("utf-8")

        bad_alias = client.get(f"/api/batch/jobs/{job['id']}/videos/{video['id']}/download?alias=nope")
        assert bad_alias.status_code == 404
        bad_video = client.get(f"/api/batch/jobs/{job['id']}/videos/bv_ghost/download?alias=result")
        assert bad_video.status_code == 404
        # 产物文件缺失 → 404。
        (work / "a.result.20260819000000.txt").unlink()
        gone = client.get(f"/api/batch/jobs/{job['id']}/videos/{video['id']}/download?alias=result")
        assert gone.status_code == 404
    finally:
        client.__exit__(None, None, None)


def test_batch_api_roots(tmp_path, monkeypatch) -> None:
    """目录树选择器：返回可浏览根目录（含根/家目录与 Windows 盘符分支）。"""
    client, _folder = _client_with_echo_workflow(tmp_path)
    try:
        roots = client.get("/api/batch/roots").json()
        assert isinstance(roots, list) and len(roots) >= 1
        # POSIX 必有 /；Windows 必有盘符；两者都有家目录。
        assert any(item["path"] in ("/", str(Path.home())) for item in roots)
        assert all(item["name"] for item in roots)

        # Windows 分支：模拟 os.name=nt，存在盘符时返回该驱动器。
        import os

        real_exists = Path.exists

        def fake_exists(path):
            # 盘符形式（如 C:\）视为存在，其余走真实判断。
            return str(path).endswith(":\\") or real_exists(path)

        monkeypatch.setattr(os, "name", "nt")
        monkeypatch.setattr(Path, "exists", fake_exists)
        nt_roots = client.get("/api/batch/roots").json()
        assert any(str(item["path"]).endswith(":\\") for item in nt_roots)
    finally:
        client.__exit__(None, None, None)


def test_batch_api_dirs(tmp_path, monkeypatch) -> None:
    """目录树选择器：列出子目录、隐藏目录过滤、不存在/不可读返回空。"""
    client, _folder = _client_with_echo_workflow(tmp_path)
    try:
        # 真实目录结构：普通子目录、隐藏目录、文件。
        target = tmp_path / "media"
        (target / "movies").mkdir(parents=True)
        (target / "series").mkdir(parents=True)
        (target / ".hidden").mkdir(parents=True)
        (target / "note.txt").write_text("x", encoding="utf-8")

        data = client.get(f"/api/batch/dirs?path={target}").json()
        assert [item["name"] for item in data["dirs"]] == ["movies", "series"]
        assert data["path"] == str(target)

        # 路径指向文件 → 空列表。
        file_data = client.get(f"/api/batch/dirs?path={target / 'note.txt'}").json()
        assert file_data["dirs"] == []
        # 目录不存在 → 空列表。
        missing = client.get(f"/api/batch/dirs?path={tmp_path / 'ghost'}").json()
        assert missing["dirs"] == []

        from pathlib import Path as RealPath

        # 单个子项不可读（is_dir 抛 OSError）→ 跳过该项，其余目录正常返回。
        class _PoisonPath(RealPath):
            """is_dir 恒抛权限错误的子类，模拟不可读的子目录。"""

            def is_dir(self):
                raise OSError("denied")

        real_iterdir = RealPath.iterdir

        def mixed_iterdir(path):
            return list(real_iterdir(path)) + [_PoisonPath(str(tmp_path / "poison"))]

        monkeypatch.setattr(RealPath, "iterdir", mixed_iterdir)
        mixed = client.get(f"/api/batch/dirs?path={target}").json()
        assert [item["name"] for item in mixed["dirs"]] == ["movies", "series"]

        # 整个目录不可读（iterdir 抛 OSError）→ 空列表而不是 500。
        def deny(path):
            raise OSError("denied")

        monkeypatch.setattr(RealPath, "iterdir", deny)
        denied = client.get(f"/api/batch/dirs?path={target}").json()
        assert denied["dirs"] == []
    finally:
        client.__exit__(None, None, None)