diff --git a/CHANGELOG.md b/CHANGELOG.md index 7c3337c..a7a953d 100644 --- a/CHANGELOG.md +++ b/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 `을 요구(상수 시간 비교), 미설정 시 개방(개발용). `/health`는 항상 미인증 유지. diff --git a/tests/test_tools_filter_dom.py b/tests/test_tools_filter_dom.py new file mode 100644 index 0000000..bcc6f37 --- /dev/null +++ b/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 diff --git a/tools/filter_dom.py b/tools/filter_dom.py new file mode 100644 index 0000000..39738cd --- /dev/null +++ b/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()