feat(export): emit schema-validated JSONL atomically

.github/workflows/tests.yml

@@ -35,4 +35,4 @@ jobs:
         env:
           PYTHONWARNINGS: error
           PYTHONPATH: src
-        run: uv run pytest --cov=src/newsdom_api --cov-branch --cov-report=term-missing --cov-fail-under=100
+        run: uv run pytest --cov --cov-branch --cov-report=term-missing --cov-fail-under=100

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- NewsDOM JSON을 schema validation 후 article/body-block 단위 JSONL로 변환하는 `tools/export_jsonl.py` 도구를 추가했습니다. 출력은 같은 디렉터리의 임시 파일에 완전히 기록된 뒤 교체되므로 encoding/write 실패 시 기존 published JSONL을 보존하고 partial 결과를 완료 산출물로 노출하지 않습니다.
+
+
 > **Planned 0.3.0 deployment migration:** parser authentication changes from
 > **default-open** to **default-required**. Production must configure
 > `NEWSDOM_AUTH_MODE=required`, `NEWSDOM_RUNTIME_PROFILE=production`, and

tests/test_ci_coverage_contract.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+import re
+import shlex
+from pathlib import Path
+
+import yaml
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def _coverage_source_paths() -> set[str]:
+    pyproject = (ROOT / "pyproject.toml").read_text(encoding="utf-8")
+    match = re.search(
+        r"(?ms)^\[tool\.coverage\.run\]\s*$\n(?P<section>.*?)(?=^\[|\Z)",
+        pyproject,
+    )
+    assert match is not None, "pyproject.toml must declare [tool.coverage.run]"
+
+    source_match = re.search(r"(?m)^source\s*=\s*\[(?P<paths>[^]]+)\]", match["section"])
+    assert source_match is not None, "coverage run config must declare source paths"
+    return set(re.findall(r'"([^"]+)"', source_match["paths"]))
+
+
+def test_ci_coverage_uses_project_sources_without_narrowing_them() -> None:
+    """CI must measure every production source declared by coverage configuration."""
+    declared_sources = _coverage_source_paths()
+    assert {"src/newsdom_api", "tools"}.issubset(declared_sources)
+
+    workflow = yaml.safe_load(
+        (ROOT / ".github/workflows/tests.yml").read_text(encoding="utf-8")
+    )
+    pytest_job = workflow["jobs"]["pytest"]
+    coverage_step = next(
+        step for step in pytest_job["steps"] if step.get("name") == "Run tests with coverage"
+    )
+    command = coverage_step["run"]
+    tokens = shlex.split(command)
+
+    assert "--cov" in tokens, (
+        "pytest-cov must use bare --cov so [tool.coverage.run].source remains authoritative; "
+        "--cov=<value> overrides the configured source set"
+    )
+    assert not any(token.startswith("--cov=") for token in tokens)

tests/test_tools_export_jsonl.py

@@ -0,0 +1,214 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+from pydantic import ValidationError
+
+from tools.export_jsonl import export_jsonl, main
+
+VALID_JSON_DATA = {
+    "document_id": "test_doc",
+    "pages": [
+        {
+            "page_number": 1,
+            "articles": [
+                {
+                    "article_id": "art_1",
+                    "headline": "Test Headline 1",
+                    "body_blocks": ["Block 1", "Block 2"],
+                },
+                {
+                    "article_id": "art_2",
+                    "headline": "Test Headline 2",
+                    "body_blocks": [],
+                },
+            ],
+        },
+        {
+            "page_number": 2,
+            "articles": [
+                {
+                    "article_id": "art_3",
+                    "headline": "Test Headline 3",
+                    "body_blocks": ["Block 3"],
+                },
+            ],
+        },
+    ],
+}
+
+
+def test_export_jsonl_success(tmp_path: Path) -> None:
+    input_file = tmp_path / "input.json"
+    input_file.write_text(json.dumps(VALID_JSON_DATA), encoding="utf-8")
+    output_file = tmp_path / "output.jsonl"
+
+    export_jsonl(input_file, output_file)
+
+    assert output_file.exists()
+
+    with output_file.open("r", encoding="utf-8") as f:
+        lines = f.readlines()
+
+        assert len(lines) == 4
+        row0 = json.loads(lines[0])
+        assert row0["document_id"] == "test_doc"
+        assert row0["page_number"] == 1
+        assert row0["article_id"] == "art_1"
+        assert row0["headline"] == "Test Headline 1"
+        assert row0["body_block_index"] == 0
+        assert row0["body_block_text"] == "Block 1"
+
+        row1 = json.loads(lines[1])
+        assert row1["body_block_index"] == 1
+        assert row1["body_block_text"] == "Block 2"
+
+        row2 = json.loads(lines[2])
+        assert row2["headline"] == "Test Headline 2"
+        assert row2["body_block_index"] is None
+        assert row2["body_block_text"] == ""
+
+        row3 = json.loads(lines[3])
+        assert row3["page_number"] == 2
+        assert row3["article_id"] == "art_3"
+        assert row3["body_block_index"] == 0
+        assert row3["body_block_text"] == "Block 3"
+
+
+@pytest.mark.parametrize(
+    "invalid_pages",
+    [
+        ["not_a_dict_page"],
+        [{"page_number": 1, "articles": ["not_a_dict_article"]}],
+        [
+            {
+                "page_number": 1,
+                "articles": [
+                    {
+                        "article_id": "art_1",
+                        "headline": "Headline",
+                        "body_blocks": "not-a-list",
+                    }
+                ],
+            }
+        ],
+    ],
+)
+def test_export_jsonl_rejects_schema_invalid_dom_before_output(
+    tmp_path: Path, invalid_pages: list[object]
+) -> None:
+    input_file = tmp_path / "input.json"
+    input_file.write_text(
+        json.dumps({"document_id": "test_doc", "pages": invalid_pages}),
+        encoding="utf-8",
+    )
+    output_file = tmp_path / "output.jsonl"
+
+    with pytest.raises(ValidationError):
+        export_jsonl(input_file, output_file)
+
+    assert not output_file.exists()
+
+
+def test_export_jsonl_preserves_published_output_when_encoding_fails(
+    tmp_path: Path,
+) -> None:
+    input_file = tmp_path / "input.json"
+    input_file.write_text(
+        json.dumps(
+            {
+                "document_id": "test_doc",
+                "pages": [
+                    {
+                        "page_number": 1,
+                        "articles": [
+                            {
+                                "article_id": "good",
+                                "headline": "Good",
+                                "body_blocks": ["first row"],
+                            },
+                            {
+                                "article_id": "bad",
+                                "headline": "Bad",
+                                "body_blocks": ["\ud800"],
+                            },
+                        ],
+                    }
+                ],
+            }
+        ),
+        encoding="utf-8",
+    )
+    output_file = tmp_path / "output.jsonl"
+    published = '{"existing": true}\n'
+    output_file.write_text(published, encoding="utf-8")
+
+    with pytest.raises(UnicodeEncodeError):
+        export_jsonl(input_file, output_file)
+
+    assert output_file.read_text(encoding="utf-8") == published
+    assert {path.name for path in tmp_path.iterdir()} == {"input.json", "output.jsonl"}
+
+
+def test_export_jsonl_missing_output_parent_leaves_no_temporary_artifact(
+    tmp_path: Path,
+) -> None:
+    input_file = tmp_path / "input.json"
+    input_file.write_text(json.dumps(VALID_JSON_DATA), encoding="utf-8")
+    missing_parent = tmp_path / "missing"
+    output_file = missing_parent / "output.jsonl"
+
+    with pytest.raises(FileNotFoundError):
+        export_jsonl(input_file, output_file)
+
+    assert not missing_parent.exists()
+    assert {path.name for path in tmp_path.iterdir()} == {"input.json"}
+
+
+def test_export_jsonl_invalid_file(tmp_path: Path) -> None:
+    output_file = tmp_path / "output.jsonl"
+
+    non_existent = tmp_path / "not_exist.json"
+    with pytest.raises(FileNotFoundError, match="File not found"):
+        export_jsonl(non_existent, output_file)
+
+    not_json = tmp_path / "input.txt"
+    not_json.write_text("plain text", encoding="utf-8")
+    with pytest.raises(ValueError, match="must be a .json file"):
+        export_jsonl(not_json, output_file)
+
+    invalid_json = tmp_path / "invalid.json"
+    invalid_json.write_text("{invalid_json:", encoding="utf-8")
+    with pytest.raises(ValueError, match="Invalid JSON file"):
+        export_jsonl(invalid_json, output_file)
+
+
+def test_export_jsonl_cli_success(
+    tmp_path: Path, capsys: pytest.CaptureFixture
+) -> None:
+    input_file = tmp_path / "input.json"
+    input_file.write_text(json.dumps(VALID_JSON_DATA), encoding="utf-8")
+    output_file = tmp_path / "output.jsonl"
+
+    main([str(input_file), str(output_file)])
+
+    assert output_file.exists()
+    captured = capsys.readouterr()
+    assert "JSONL successfully written" in captured.out
+
+
+def test_export_jsonl_cli_invalid_file(
+    tmp_path: Path, capsys: pytest.CaptureFixture
+) -> None:
+    not_json = tmp_path / "input.txt"
+    not_json.write_text("plain text", encoding="utf-8")
+    output_file = tmp_path / "output.jsonl"
+
+    with pytest.raises(SystemExit) as exc_info:
+        main([str(not_json), str(output_file)])
+
+    assert exc_info.value.code == 1
+    captured = capsys.readouterr()
+    assert "Error exporting JSONL:" in captured.err

tests/test_tools_export_jsonl_interrupt_cleanup.py

@@ -0,0 +1,52 @@
+"""Lifecycle cleanup contracts for the NewsDOM JSONL exporter."""
+
+from __future__ import annotations
+
+import importlib
+import json
+from pathlib import Path
+
+import pytest
+
+export_module = importlib.import_module("tools.export_jsonl")
+
+
+def test_export_jsonl_cleans_temporary_artifact_on_keyboard_interrupt(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A user cancellation must not leak a temporary publication artifact."""
+    input_file = tmp_path / "input.json"
+    input_file.write_text(
+        json.dumps(
+            {
+                "document_id": "test_doc",
+                "pages": [
+                    {
+                        "page_number": 1,
+                        "articles": [
+                            {
+                                "article_id": "art_1",
+                                "headline": "Headline",
+                                "body_blocks": ["Block"],
+                            }
+                        ],
+                    }
+                ],
+            }
+        ),
+        encoding="utf-8",
+    )
+    output_file = tmp_path / "output.jsonl"
+    output_file.write_text("previous publication\n", encoding="utf-8")
+
+    def _interrupt(*_args: object, **_kwargs: object) -> str:
+        raise KeyboardInterrupt
+
+    monkeypatch.setattr(export_module.json, "dumps", _interrupt)
+
+    with pytest.raises(KeyboardInterrupt):
+        export_module.export_jsonl(input_file, output_file)
+
+    assert output_file.read_text(encoding="utf-8") == "previous publication\n"
+    assert list(tmp_path.glob(".output.jsonl.*.tmp")) == []