feat: 특정 요소를 필터링하는 DOM 필터 도구 추가

CHANGELOG.md

@@ -12,6 +12,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - OpenAPI 제목/설명, README, `ArticleNode.headline` 문서를 일반 문서용 (section heading) 표현으로 재구성하여 특정 언어/신문 가정을 소비자에게 노출하지 않도록 함. 응답 스키마 필드는 하위 호환을 위해 변경하지 않음.
 
 ### Added
+- `tools/filter_dom.py`: 새로운 도구 추가. DOM JSON에서 ads, images, headers, footers 등 특정 요소를 필터링하는 기능 제공
 - [CLI] 단일 NewsDOM JSON 파일을 페이지 단위로 분리하는 `tools/split_dom.py` 도구를 추가했습니다.
 - [CLI] NewsDOM JSON 파일의 모든 텍스트 내용을 마스킹하여 익명화하는 `tools/anonymize_dom.py` 도구를 추가했습니다.
 - `/parse`에 대한 optional bearer 인증 게이트: 리프 서비스 자체 설정 `NEWSDOM_API_TOKEN`이 설정되면 `Authorization: Bearer <token>`을 요구(상수 시간 비교), 미설정 시 개방(개발용). `/health`는 항상 미인증 유지.

tests/test_tools_filter_dom.py

@@ -0,0 +1,199 @@
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+from tools.filter_dom import filter_dom, main
+
+
+def create_valid_parse_response_data() -> dict:
+    return {
+        "document_id": "test_doc",
+        "quality": {"status": "success", "parser": "test", "warnings": []},
+        "pages": [
+            {
+                "page_number": 1,
+                "ads": ["Ad 1"],
+                "headers": ["Header 1"],
+                "footers": ["Footer 1"],
+                "articles": [
+                    {
+                        "article_id": "art-1",
+                        "headline": "H1",
+                        "images": [{"path": "img1.png", "media_type": "image"}],
+                        "body_blocks": [],
+                    }
+                ],
+                "page_numbers": [],
+            }
+        ],
+    }
+
+
+def test_filter_dom_removes_elements(tmp_path: Path):
+    input_file = tmp_path / "input.json"
+    data = create_valid_parse_response_data()
+    input_file.write_text(json.dumps(data))
+
+    result = filter_dom(
+        input_file,
+        remove_ads=True,
+        remove_images=True,
+        remove_headers=True,
+        remove_footers=True,
+    )
+
+    page = result["pages"][0]
+    assert page["ads"] == []
+    assert page["headers"] == []
+    assert page["footers"] == []
+    assert page["articles"][0]["images"] == []
+
+
+def test_filter_dom_partial_removal(tmp_path: Path):
+    input_file = tmp_path / "input.json"
+    data = create_valid_parse_response_data()
+    input_file.write_text(json.dumps(data))
+
+    result = filter_dom(
+        input_file,
+        remove_ads=True,
+        remove_images=True,
+    )
+
+    page = result["pages"][0]
+    assert page["ads"] == []
+    assert page["headers"] == ["Header 1"]
+    assert page["footers"] == ["Footer 1"]
+    assert page["articles"][0]["images"] == []
+
+
+def test_filter_dom_no_removal(tmp_path: Path):
+    input_file = tmp_path / "input.json"
+    data = create_valid_parse_response_data()
+    input_file.write_text(json.dumps(data))
+
+    result = filter_dom(input_file)
+
+    page = result["pages"][0]
+    assert page["ads"] == ["Ad 1"]
+    assert page["headers"] == ["Header 1"]
+    assert page["footers"] == ["Footer 1"]
+    assert page["articles"][0]["images"] == [
+        {"path": "img1.png", "media_type": "image"}
+    ]
+
+
+def test_filter_dom_empty_pages(tmp_path: Path):
+    input_file = tmp_path / "input.json"
+    data = create_valid_parse_response_data()
+    data["pages"] = []
+    input_file.write_text(json.dumps(data))
+
+    result = filter_dom(input_file, remove_ads=True)
+    assert result["pages"] == []
+
+
+def test_filter_dom_article_without_images(tmp_path: Path):
+    input_file = tmp_path / "input.json"
+    data = create_valid_parse_response_data()
+    del data["pages"][0]["articles"][0]["images"]
+    input_file.write_text(json.dumps(data))
+
+    result = filter_dom(input_file, remove_images=True)
+    assert "images" not in result["pages"][0]["articles"][0]
+
+
+def test_filter_dom_not_a_file(tmp_path: Path):
+    non_existent = tmp_path / "non_existent.json"
+    with pytest.raises(FileNotFoundError):
+        filter_dom(non_existent)
+
+
+def test_filter_dom_wrong_extension(tmp_path: Path):
+    txt_file = tmp_path / "input.txt"
+    txt_file.write_text("{}")
+    with pytest.raises(ValueError, match="must be a .json file"):
+        filter_dom(txt_file)
+
+
+def test_filter_dom_invalid_json(tmp_path: Path):
+    input_file = tmp_path / "input.json"
+    input_file.write_text("{invalid")
+    with pytest.raises(ValueError, match="Invalid JSON file"):
+        filter_dom(input_file)
+
+
+def test_filter_dom_invalid_schema(tmp_path: Path):
+    input_file = tmp_path / "input.json"
+    input_file.write_text('{"pages": []}')
+    with pytest.raises(ValueError, match="does not match ParseResponse schema"):
+        filter_dom(input_file)
+
+
+def test_main_with_output(tmp_path: Path, capsys):
+    input_file = tmp_path / "input.json"
+    output_file = tmp_path / "output.json"
+
+    data = create_valid_parse_response_data()
+    input_file.write_text(json.dumps(data))
+
+    main([str(input_file), "-o", str(output_file), "--remove-ads"])
+
+    assert output_file.is_file()
+    result = json.loads(output_file.read_text())
+    assert result["pages"][0]["ads"] == []
+
+
+def test_main_without_output(tmp_path: Path, capsys):
+    input_file = tmp_path / "input.json"
+    data = create_valid_parse_response_data()
+    input_file.write_text(json.dumps(data))
+
+    main([str(input_file), "--remove-ads"])
+
+    captured = capsys.readouterr()
+    assert '"ads": []' in captured.out
+
+
+def test_main_exception_filenotfound(tmp_path: Path, capsys):
+    non_existent = tmp_path / "non_existent.json"
+    with pytest.raises(SystemExit) as exc:
+        main([str(non_existent)])
+
+    assert exc.value.code == 1
+    captured = capsys.readouterr()
+    assert "Error:" in captured.err
+
+
+def test_main_exception_valueerror(tmp_path: Path, capsys):
+    input_file = tmp_path / "input.txt"
+    input_file.write_text("")
+    with pytest.raises(SystemExit) as exc:
+        main([str(input_file)])
+
+    assert exc.value.code == 1
+    captured = capsys.readouterr()
+    assert "must be a .json file" in captured.err
+
+
+def test_sys_path_injection_local_scope(monkeypatch, tmp_path: Path):
+    """Test that sys.path injection is properly handled and cleaned up."""
+    input_file = tmp_path / "input.json"
+    data = create_valid_parse_response_data()
+    input_file.write_text(json.dumps(data))
+
+    _REPO_ROOT = Path(__file__).resolve().parents[1]
+    _SRC_ROOT = str(_REPO_ROOT / "src")
+
+    # Remove src from sys.path to force the function to inject it
+    monkeypatch.setattr(sys, "path", [p for p in sys.path if p != _SRC_ROOT])
+
+    assert _SRC_ROOT not in sys.path
+
+    # Run filter_dom, which should locally inject and then remove _SRC_ROOT
+    filter_dom(input_file)
+
+    # Verify cleanup occurred
+    assert _SRC_ROOT not in sys.path

tools/filter_dom.py

@@ -0,0 +1,112 @@
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+
+def filter_dom(
+    json_path: Path,
+    remove_ads: bool = False,
+    remove_images: bool = False,
+    remove_headers: bool = False,
+    remove_footers: bool = False,
+) -> dict:
+    """Filter specific elements from NewsDOM JSON."""
+    if not json_path.is_file():
+        raise FileNotFoundError(f"File not found or is not a file: {json_path}")
+    if json_path.suffix.lower() != ".json":
+        raise ValueError("File must be a .json file.")
+
+    try:
+        data = json.loads(json_path.read_text(encoding="utf-8"))
+    except json.JSONDecodeError as exc:
+        raise ValueError(f"Invalid JSON file ({json_path}): {exc}") from exc
+
+    # Validate using the canonical schema without modifying global sys.path.
+    _validate_schema_locally(data, json_path)
+
+    for page in data.get("pages", []):
+        if remove_ads and "ads" in page:
+            page["ads"] = []
+        if remove_headers and "headers" in page:
+            page["headers"] = []
+        if remove_footers and "footers" in page:
+            page["footers"] = []
+
+        if remove_images:
+            for article in page.get("articles", []):
+                if "images" in article:
+                    article["images"] = []
+
+    return data
+
+
+def _validate_schema_locally(data: dict, json_path: Path) -> None:
+    """Validate data against ParseResponse schema locally."""
+    # Ensure sys.path modification is scoped and does not leak globally.
+    _REPO_ROOT = Path(__file__).resolve().parents[1]
+    _SRC_ROOT = str(_REPO_ROOT / "src")
+
+    path_injected = False
+    if _SRC_ROOT not in sys.path:
+        sys.path.insert(0, _SRC_ROOT)
+        path_injected = True
+
+    try:
+        from pydantic import ValidationError
+        from newsdom_api.schemas import ParseResponse
+
+        ParseResponse.model_validate(data)
+    except ValidationError as exc:
+        raise ValueError(
+            f"File {json_path} does not match ParseResponse schema: {exc}"
+        ) from exc
+    finally:
+        if path_injected:
+            sys.path.remove(_SRC_ROOT)
+
+
+def main(argv: list[str] | None = None) -> None:
+    """Run filter_dom main entry point."""
+    parser = argparse.ArgumentParser(description="Filter elements from NewsDOM JSON.")
+    parser.add_argument("input", type=Path, help="Path to the JSON DOM file.")
+    parser.add_argument(
+        "-o",
+        "--output",
+        type=Path,
+        help="Path to save the filtered JSON.",
+        default=None,
+    )
+    parser.add_argument("--remove-ads", action="store_true", help="Remove ads.")
+    parser.add_argument("--remove-images", action="store_true", help="Remove images.")
+    parser.add_argument("--remove-headers", action="store_true", help="Remove headers.")
+    parser.add_argument("--remove-footers", action="store_true", help="Remove footers.")
+
+    args = parser.parse_args(argv)
+
+    try:
+        filtered_data = filter_dom(
+            args.input,
+            remove_ads=args.remove_ads,
+            remove_images=args.remove_images,
+            remove_headers=args.remove_headers,
+            remove_footers=args.remove_footers,
+        )
+        output_json = json.dumps(filtered_data, ensure_ascii=False, indent=2)
+
+        if args.output:
+            args.output.write_text(output_json, encoding="utf-8")
+        else:
+            print(output_json)
+    except FileNotFoundError as e:
+        print(f"Error: {e}", file=sys.stderr)
+        sys.exit(1)
+    except ValueError as e:
+        print(f"Error: {e}", file=sys.stderr)
+        sys.exit(1)
+
+
+if __name__ == "__main__":  # pragma: no cover
+    main()