japanese/tests/test_cleaner_workflow.py

"""
清洗工作流测试（三层架构版本）

测试策略：数据驱动 + 隔离临时任务，覆盖：
1. 底层单词分析（TangoAnalyser）：词条 → 状态分类
2. 中间层文件处理（TaskProcessor）：文件搬运、去重、格式校验
3. 工作流编排（CleanerWorkflow）：状态机推进、人工介入门禁

所有测试用隔离临时目录（tmp_path），不碰真实数据。
"""
from pathlib import Path
import pytest

from pl_japanese.cleaner import (
    TangoAnalyser, AnalysisStatus,
    TaskProcessor, TaskManager, CleanerWorkflow,
    ACTION_PROCESSED, ACTION_NEED_HUMAN, ACTION_COMPLETED, ACTION_EMPTY,
)
from pl_japanese.cleaner.task import (
    STATUS_CREATED, STATUS_PROCESSING, STATUS_REVIEWING, STATUS_READY, STATUS_MERGED,
)


# ==========================================================================
# 测试辅助
# ==========================================================================
def _make_task(tmp_path: Path, source_lines: list[str], task_id="test_task"):
    """创建隔离临时任务（数据源、任务目录、main/skipped 都在 tmp_path 下）
    
    返回: (tm, task, main_path, skipped_path)
    """
    source = tmp_path / "source.txt"
    source.write_text("\n".join(source_lines) + "\n", encoding="utf-8", newline="\n")
    
    tm = TaskManager(tasks_root=str(tmp_path / "tasks"))
    task = tm.create_task(
        task_id=task_id,
        source=str(source),
        name=f"test_{task_id}",
        start_line=1,
        count=len(source_lines),
    )
    main_path = str(tmp_path / "main.txt")
    skipped_path = str(tmp_path / "skipped.txt")
    return tm, task, main_path, skipped_path
    return tm, task


# ==========================================================================
# 1. 底层单词分析（TangoAnalyser）：词条 → 状态分类
# ==========================================================================
# (kanji, kana, 期望状态, 输出子串检查)
ANALYSE_CASES = [
    pytest.param("日本人", "にほんじん", AnalysisStatus.SUCCESS, "日|本|人:", id="success-basic"),
    pytest.param("中国人", "ちゅうごくじん", AnalysisStatus.SUCCESS, "中|国|人:", id="success-polyphone-resolved"),
    pytest.param("IT", "アイティー", AnalysisStatus.SKIP, "IT:", id="skip-no-kanji"),
    pytest.param("行列", "こうれつ", AnalysisStatus.SUCCESS, "行|列:こう|れつ:hang|lie", id="polyphone-resolved-by-dict"),
    pytest.param("女将", "おかみ", AnalysisStatus.SPLIT_FAILED, "女|将:おかみ:", id="split-failed"),
    pytest.param("食べる", "たべる", AnalysisStatus.SUCCESS, "食|べる:た|べる:", id="verb-auto"),
    pytest.param("Uターン", "ユーターン", AnalysisStatus.SKIP, "Uターン:", id="skip-no-kanji-with-latin"),
]


@pytest.mark.parametrize("kanji,kana,status,substring", ANALYSE_CASES)
def test_analyser_status_classification(kanji, kana, status, substring):
    """底层分析器：词条 → 状态分类正确，输出格式符合预期"""
    analyser = TangoAnalyser()
    result = analyser.analyze(kanji, kana)
    assert result.status == status
    assert substring in result.formatted_line


def test_analyser_needs_review_logic():
    """needs_review 属性：SUCCESS/SKIP 返回 False，其余返回 True"""
    analyser = TangoAnalyser()
    
    success = analyser.analyze("日本人", "にほんじん")
    assert success.is_success is True
    assert success.needs_review is False
    
    skip = analyser.analyze("IT", "アイティー")
    assert skip.status == AnalysisStatus.SKIP
    assert skip.needs_review is False
    
    # 用一个真正无法自动处理的词（分割失败）
    split_fail = analyser.analyze("女将", "おかみ")
    assert split_fail.needs_review is True


# ==========================================================================
# 2. 中间层文件处理（TaskProcessor）：process_source 条数铁律
# ==========================================================================
# (源数据, 期望有效行数, 期望输出总数, 期望桶计数子集)
# (源行, 有效行数, 输出总数, {auto_pass期望, need_review期望})
PROCESS_CASES = [
    pytest.param(
        ["日本人:にほんじん:", "中国人:ちゅうごくじん:"],
        2, 2, {"auto_pass": 2, "need_review": 0},
        id="all-auto",
    ),
    pytest.param(
        ["中国人:ちゅうごくじん:", "", "学生:がくせい:"],
        2, 2, {"auto_pass": 2},
        id="blank-ignored",
    ),
    pytest.param(
        ["IT:アイティー:", "％:パーセント:"],
        2, 2, {"auto_pass": 0, "need_review": 2, "suggest_skip": 2},
        id="all-skip",
    ),
    pytest.param(
        ["日本人:にほんじん:", "IT:アイティー:"],
        2, 2, {"auto_pass": 1, "need_review": 1, "suggest_skip": 1},
        id="mixed",
    ),
    pytest.param(
        ["行列:こうれつ:", "Uターン:ユーターン:"],
        2, 2, {"auto_pass": 1, "need_review": 1},
        id="mixed-auto-skip",
    ),
]


@pytest.mark.parametrize("lines,valid,total,expect_subset", PROCESS_CASES)
def test_processor_count_law(tmp_path, lines, valid, total, expect_subset):
    """中间层文件处理器：条数铁律（有效输入 == 输出总数）+ 分区分布"""
    tm, task, main_path, skipped_path = _make_task(tmp_path, lines)
    proc = TaskProcessor(task, main=main_path, skipped=skipped_path)
    result = proc.process_source()
    
    assert result["valid_lines"] == valid
    assert result["output_total"] == total
    assert result["validation"] is True
    
    for key, cnt in expect_subset.items():
        assert result[key] == cnt


def test_processor_writes_single_review_file(tmp_path):
    """process_source 写单一待确认文件，条数=源行数"""
    tm, task, main_path, skipped_path = _make_task(tmp_path, ["日本人:にほんじん:", "IT:アイティー:"])
    proc = TaskProcessor(task, main=main_path, skipped=skipped_path)
    
    proc.process_source()
    # 待确认文件包含全部条目（自动通过段 + 待确认段）
    assert proc.total_entries() == 2
    # 待确认段只有 IT（建议skip）
    assert proc.review_count() == 1
    # auto_done/skip 此时还没写（等 apply_review）
    assert proc.snapshot_counts()["auto_done"] == 0


# ==========================================================================
# 3. 中间层：merge_final 去重 + 格式校验
# ==========================================================================
def _process_and_confirm(proc):
    """辅助：处理源文件后模拟人工确认（末尾加 done），落地 auto_done/skip。"""
    proc.process_source()
    review_file = proc.review
    content = review_file.read_text(encoding="utf-8")
    review_file.write_text(content + "done\n", encoding="utf-8", newline="\n")
    return proc.apply_review()


def test_merge_deduplication(tmp_path):
    """merge_final 去重：重复词条只保留一份"""
    tm, task, main_path, skipped_path = _make_task(tmp_path, ["日本人:にほんじん:", "日本人:にほんじん:"])
    proc = TaskProcessor(task, main=main_path, skipped=skipped_path)
    _process_and_confirm(proc)
    
    result = proc.merge_final(dry_run=False)
    assert result["merged_auto"] == 1  # 2 条输入去重后只合并 1 条
    assert result["duplicated_auto"] == 1
    assert result["main_total"] == 1


def test_merge_idempotent(tmp_path):
    """幂等性：同一批数据再次合并，主库不增长"""
    tm, task, main_path, skipped_path = _make_task(tmp_path, ["日本人:にほんじん:"])
    proc = TaskProcessor(task, main=main_path, skipped=skipped_path)
    
    _process_and_confirm(proc)
    r1 = proc.merge_final(dry_run=False)
    assert r1["merged_auto"] == 1
    
    proc.clear_single_batch()
    _process_and_confirm(proc)  # 再次处理同样数据
    r2 = proc.merge_final(dry_run=False)
    assert r2["merged_auto"] == 0  # 无新增
    assert r2["duplicated_auto"] == 1  # 全部重复
    assert r2["main_total"] == 1  # 总数不变


def test_merge_validation_rejects_bad_format(tmp_path):
    """merge_final 格式校验：非法行被 reject，不进主库"""
    tm, task, main_path, skipped_path = _make_task(tmp_path, ["日本人:にほんじん:"])
    proc = TaskProcessor(task, main=main_path, skipped=skipped_path)
    # 先不 process，直接手工写 auto_done 来测试校验
    
    # 手工塞一条合法行（段数匹配） + 一条格式非法行（只有 1 个冒号）
    proc.auto_done.write_text(
        "日|本|人:に|ほん|じん:ri|ben|ren\n非法行\n",
        encoding="utf-8", newline="\n",
    )
    
    result = proc.merge_final(dry_run=False)
    assert result["rejected"] == 1
    assert result["merged_auto"] == 1  # 合法的那条成功
    assert any("非法行" in r[0] for r in result["rejects"])


# ==========================================================================
# 4. 中间层：apply_review 应用人工裁决
# ==========================================================================
def test_apply_review_requires_done_signal(tmp_path):
    """apply_review：没有 done 信号时拒绝应用"""
    tm, task, main_path, skipped_path = _make_task(tmp_path, ["中国人:ちゅうごくじん:"])
    proc = TaskProcessor(task, main=main_path, skipped=skipped_path)
    proc.process_source()
    
    # 没加 done → 拒绝
    result = proc.apply_review()
    assert result["success"] is False


def test_apply_review_accepts_and_skips(tmp_path):
    """apply_review：待确认段无skip前缀进auto，有skip前缀进skip"""
    tm, task, main_path, skipped_path = _make_task(tmp_path, ["女将:おかみ:", "IT:アイティー:"])
    proc = TaskProcessor(task, main=main_path, skipped=skipped_path)
    proc.process_source()
    
    # 女将分割失败进待确认（格式：女|将:おかみ:nv|jiang），IT进建议skip
    # 人工：女将手动分割假名（改成：女|将:おん|な:nv|jiang），IT保持skip，加done
    content = proc.review.read_text(encoding="utf-8")
    # 将"女|将:おかみ:"修正为"女|将:おん|な:"（简化测试，不追求正确性）
    content = content.replace("女|将:おかみ:", "女|将:おん|な:")
    proc.review.write_text(content + "done\n", encoding="utf-8", newline="\n")
    
    result = proc.apply_review()
    # 验证 IT 进 skip，女将进auto_done
    skip_lines = proc._read_clean_lines(proc.skip)
    assert any("IT" in s for s in skip_lines)


def test_apply_review_add_marker(tmp_path):
    """apply_review：add 前缀的人工新增行也入主库（如一词多音的第二个读音）"""
    tm, task, main_path, skipped_path = _make_task(tmp_path, ["農作物:のうさくぶつ:"])
    proc = TaskProcessor(task, main=main_path, skipped=skipped_path)
    proc.process_source()

    content = proc.review.read_text(encoding="utf-8")
    # 主行手动拆好假名，并用 add 追加第二个读音条目
    content = content.replace(
        "農|作|物:のうさくぶつ:nong|zuo|wu",
        "農|作|物:のう|さく|ぶつ:nong|zuo|wu",
    )
    content += "add 農|作|物:のう|さく|もつ:nong|zuo|wu\n"
    proc.review.write_text(content + "done\n", encoding="utf-8", newline="\n")

    result = proc.apply_review()
    assert result["success"] is True
    # 两条读音都应进 auto_done
    accepted = proc._read_clean_lines(proc.auto_done)
    assert any("ぶつ" in s for s in accepted)
    assert any("もつ" in s for s in accepted)


# ==========================================================================
# 5. 工作流编排（CleanerWorkflow）：状态机推进
# ==========================================================================
def test_workflow_run_auto_complete(tmp_path):
    """全自动流程：created → ready（全部自动通过）→ merged → completed"""
    tm, task, main_path, skipped_path = _make_task(tmp_path, ["日本人:にほんじん:", "中国人:ちゅうごくじん:"])
    wf = CleanerWorkflow(task, main=main_path, skipped=skipped_path)
    
    # 第一步：created → process → ready（全部 auto_pass，待确认段为0）
    r1 = wf.run()
    assert r1["action"] == ACTION_PROCESSED
    assert r1["status"] == STATUS_READY
    
    # 第二步：ready → merge → merged
    r2 = wf.run()
    assert r2["action"] == ACTION_PROCESSED
    assert r2["status"] == STATUS_MERGED
    assert r2["merge"]["merged_auto"] == 2
    
    # 第三步：merged → completed（终态）
    r3 = wf.run()
    assert r3["action"] == ACTION_COMPLETED
    assert r3["status"] == STATUS_MERGED


def test_workflow_run_with_review_gate(tmp_path):
    """带人工介入流程：created → reviewing → 无done时返回待人工（不推进）"""
    # 使用真正会进待确认的词：女将（分割失败）
    tm, task, main_path, skipped_path = _make_task(tmp_path, ["女将:おかみ:", "日本人:にほんじん:"])
    wf = CleanerWorkflow(task, main=main_path, skipped=skipped_path)
    
    # 第一步：created → process → reviewing（有待确认条目）
    r1 = wf.run()
    assert r1["action"] == ACTION_PROCESSED
    assert r1["status"] == STATUS_REVIEWING
    assert r1["review_count"] > 0
    
    # 第二步：reviewing 且无 done → 返回待人工，不推进
    r2 = wf.run()
    assert r2["action"] == ACTION_NEED_HUMAN
    assert r2["status"] == STATUS_REVIEWING
    assert "done" in r2["message"]


def test_workflow_review_done_advances_to_ready(tmp_path):
    """人工加 done 后，run() 应用修改并推进到 ready"""
    tm, task, main_path, skipped_path = _make_task(tmp_path, ["女将:おかみ:"])
    wf = CleanerWorkflow(task, main=main_path, skipped=skipped_path)
    proc = wf.processor
    
    # 处理 → reviewing
    r1 = wf.run()
    assert r1["status"] == STATUS_REVIEWING
    
    # 人工：把女将改成合格三段式（模拟人工分割修正），加 done
    review_content = (
        "# ===== 自动通过（无需确认）=====\n\n"
        "# ===== 待确认（请检查，改拼音或加/删 skip）=====\n"
        "女|将:おか|み:nv|jiang    # 无法拆分, L1\n"
        "done\n"
    )
    proc.review.write_text(review_content, encoding="utf-8", newline="\n")
    
    # 再次 run()：应用 → 转 ready
    r2 = wf.run()
    assert r2["action"] == ACTION_PROCESSED
    assert r2["status"] == STATUS_READY


def test_workflow_merge_rejects_bad_format(tmp_path):
    """合并时格式校验失败：不推进状态，保持 ready，提示人工修正"""
    tm, task, main_path, skipped_path = _make_task(tmp_path, ["日本人:にほんじん:"])
    wf = CleanerWorkflow(task, main=main_path, skipped=skipped_path)
    proc = wf.processor
    
    r1 = wf.run()
    assert r1["status"] == STATUS_READY
    
    # 手工塞一条非法格式
    proc.auto_done.write_text(
        "日|本|人:にほん|じん:ri|ben|ren\n非法行",
        encoding="utf-8", newline="\n",
    )
    
    r2 = wf.run()
    assert r2["action"] == ACTION_NEED_HUMAN
    assert r2["status"] == STATUS_READY
    assert "格式非法" in r2["message"] or "校验" in r2["message"]


def test_workflow_empty_source(tmp_path):
    """空源文件：返回 empty，不推进"""
    tm, task, main_path, skipped_path = _make_task(tmp_path, ["", "   "])
    wf = CleanerWorkflow(task, main=main_path, skipped=skipped_path)
    
    result = wf.run()
    assert result["action"] == ACTION_EMPTY
    assert result["process"]["valid_lines"] == 0


# ==========================================================================
# 6. 任务管理（TaskManager）：CRUD + 多任务隔离
# ==========================================================================
def test_task_manager_create_and_load(tmp_path):
    """任务创建和加载：配置持久化正确"""
    tm, task, main_path, skipped_path = _make_task(tmp_path, ["日本人:にほんじん:"], task_id="task1")
    
    loaded = tm.load_task("task1")
    assert loaded.task_id == "task1"
    assert loaded.config.source == task.config.source
    assert loaded.state.status == STATUS_CREATED


def test_task_manager_list_summaries(tmp_path):
    """list_task_summaries：返回所有任务摘要"""
    source1 = tmp_path / "s1.txt"
    source1.write_text("日本人:にほんじん:\n", encoding="utf-8")
    source2 = tmp_path / "s2.txt"
    source2.write_text("中国人:ちゅうごくじん:\n", encoding="utf-8")
    
    tm = TaskManager(tasks_root=str(tmp_path / "tasks"))
    tm.create_task("t1", source=str(source1), name="任务1")
    tm.create_task("t2", source=str(source2), name="任务2")
    
    summaries = tm.list_task_summaries()
    assert len(summaries) == 2
    ids = {s["task_id"] for s in summaries}
    assert ids == {"t1", "t2"}


def test_task_manager_duplicate_rejected(tmp_path):
    """重复 task_id 创建被拒绝"""
    tm, task, main_path, skipped_path = _make_task(tmp_path, ["日本人:にほんじん:"])
    with pytest.raises(FileExistsError):
        tm.create_task(task.task_id, source=task.config.source)


def test_multi_task_isolation(tmp_path):
    """多任务隔离：各任务的单批文件独立，互不干扰"""
    source = tmp_path / "source.txt"
    source.write_text("日本人:にほんじん:\n中国人:ちゅうごくじん:\n", encoding="utf-8")
    
    tm = TaskManager(tasks_root=str(tmp_path / "tasks"))
    t1 = tm.create_task("t1", source=str(source), count=1)
    t2 = tm.create_task("t2", source=str(source), start_line=2, count=1)
    
    main_path = str(tmp_path / "main.txt")
    skipped_path = str(tmp_path / "skip.txt")
    p1 = TaskProcessor(t1, main=main_path, skipped=skipped_path)
    p2 = TaskProcessor(t2, main=main_path, skipped=skipped_path)
    
    # 处理 + 应用（加done）
    _process_and_confirm(p1)
    _process_and_confirm(p2)
    
    # 各自单批文件独立
    assert p1.snapshot_counts()["auto_done"] == 1
    assert p2.snapshot_counts()["auto_done"] == 1
    
    # 但 main 库是共享的（都指向同一文件）
    p1.merge_final(dry_run=False)
    p2.merge_final(dry_run=False)
    assert p1.final_counts()["main"] == 2
    assert p2.final_counts()["main"] == 2