From 1ca5a15f226f45a0e9bd20b07b136db6fc253f0e Mon Sep 17 00:00:00 2001 From: chottokun <29515187+chottokun@users.noreply.github.com> Date: Mon, 12 Oct 2026 01:08:37 +0000 Subject: [PATCH 1/2] refactor: improve code quality and separation of concerns across modules Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com> --- src/docling_lib/config.py | 1 + src/docling_lib/converter.py | 132 ++++++++------ src/docling_lib/utils.py | 225 +++++++++++++++--------- tests/test_refactor_characterization.py | 67 +++++++ 4 files changed, 283 insertions(+), 142 deletions(-) create mode 100644 tests/test_refactor_characterization.py diff --git a/src/docling_lib/config.py b/src/docling_lib/config.py index 7aa1d4c..c9b218a 100644 --- a/src/docling_lib/config.py +++ b/src/docling_lib/config.py @@ -35,6 +35,7 @@ # Model cache & artifacts configurations _default_cache_models_dir = Path(".cache_models").resolve() _env_artifacts_path = os.getenv("DOCLING_ARTIFACTS_PATH") +DOCLING_ARTIFACTS_PATH: Path | None if _env_artifacts_path: DOCLING_ARTIFACTS_PATH = Path(_env_artifacts_path) elif _default_cache_models_dir.exists() and _default_cache_models_dir.is_dir(): diff --git a/src/docling_lib/converter.py b/src/docling_lib/converter.py index e732d5c..8f631f4 100644 --- a/src/docling_lib/converter.py +++ b/src/docling_lib/converter.py @@ -195,39 +195,39 @@ def __init__( options: DocumentConversionOptions | None = None, ): self.options = options or DocumentConversionOptions() + pipeline_options = self._configure_pipeline_options(self.options) + format_options = self._get_format_options(pipeline_options) + self.doc_converter = DocumentConverter(format_options=format_options) - # Configure pipeline options + def _configure_pipeline_options( + self, options: DocumentConversionOptions + ) -> PdfPipelineOptions: + """Constructs and configures PdfPipelineOptions from DocumentConversionOptions.""" pipeline_options = PdfPipelineOptions() - if self.options.artifacts_path: - pipeline_options.artifacts_path = Path(self.options.artifacts_path) + if options.artifacts_path: + pipeline_options.artifacts_path = Path(options.artifacts_path) pipeline_options.generate_picture_images = True - pipeline_options.images_scale = self.options.image_scale - pipeline_options.do_formula_enrichment = self.options.do_formula - pipeline_options.do_ocr = self.options.do_ocr - pipeline_options.do_chart_extraction = self.options.do_chart - pipeline_options.do_code_enrichment = self.options.do_code + pipeline_options.images_scale = options.image_scale + pipeline_options.do_formula_enrichment = options.do_formula + pipeline_options.do_ocr = options.do_ocr + pipeline_options.do_chart_extraction = options.do_chart + pipeline_options.do_code_enrichment = options.do_code - if self.options.include_page_breaks: + if options.include_page_breaks: pipeline_options.generate_page_images = True pipeline_options.generate_parsed_pages = True - # Configure accelerator options (GPU fallback to CPU) - if is_cuda_compatible(): - acc_device = AcceleratorDevice.AUTO - else: - acc_device = AcceleratorDevice.CPU - + acc_device = ( + AcceleratorDevice.AUTO if is_cuda_compatible() else AcceleratorDevice.CPU + ) pipeline_options.accelerator_options = AcceleratorOptions( device=acc_device, - num_threads=self.options.num_threads, - cuda_use_flash_attention2=self.options.cuda_use_flash_attention, + num_threads=options.num_threads, + cuda_use_flash_attention2=options.cuda_use_flash_attention, ) - # Configure DocumentConverter with multi-format support - format_options = self._get_format_options(pipeline_options) - - self.doc_converter = DocumentConverter(format_options=format_options) + return pipeline_options def _get_format_options(self, pipeline_options: PdfPipelineOptions) -> dict: """ @@ -248,6 +248,26 @@ def _get_format_options(self, pipeline_options: PdfPipelineOptions) -> dict: InputFormat.VTT: HTMLFormatOption(pipeline_options=pipeline_options), } + def _hash_existing_pictures(self, doc: DoclingDocument) -> set[str]: + """Computes SHA-256 hashes for existing picture items in doc.pictures.""" + import hashlib + import io + + existing_hashes: set[str] = set() + for existing_pic in doc.pictures: + if ( + existing_pic.image + and hasattr(existing_pic.image, "pil_image") + and existing_pic.image.pil_image + ): + try: + buf = io.BytesIO() + existing_pic.image.pil_image.save(buf, format="PNG") + existing_hashes.add(hashlib.sha256(buf.getvalue()).hexdigest()) + except Exception: + pass + return existing_hashes + def _enrich_excel_pictures(self, doc: DoclingDocument, input_path: Path) -> None: """ Inspects Excel (.xlsx) input files for embedded or pasted images in the ZIP archive @@ -262,8 +282,6 @@ def _enrich_excel_pictures(self, doc: DoclingDocument, input_path: Path) -> None if not extracted_media: return - import io - from docling_core.types.doc import ( BoundingBox, CoordOrigin, @@ -271,22 +289,7 @@ def _enrich_excel_pictures(self, doc: DoclingDocument, input_path: Path) -> None ProvenanceItem, ) - # Compute sha256 of existing images in doc.pictures to prevent duplication securely - existing_hashes: set[str] = set() - for existing_pic in doc.pictures: - if ( - existing_pic.image - and hasattr(existing_pic.image, "pil_image") - and existing_pic.image.pil_image - ): - try: - buf = io.BytesIO() - existing_pic.image.pil_image.save(buf, format="PNG") - import hashlib - - existing_hashes.add(hashlib.sha256(buf.getvalue()).hexdigest()) - except Exception: - pass + existing_hashes = self._hash_existing_pictures(doc) for item in extracted_media: pil_img = item.get("pil_image") @@ -534,7 +537,7 @@ def _apply_metadata_frontmatter( ) if metadata: - import yaml + import yaml # type: ignore[import-untyped] # default_flow_style=False ensures it doesn't use {key: value} block style # which might have caused some assertion issues with quoting @@ -633,6 +636,29 @@ def _save_markdown( return output_dir / md_output_name + def _process_and_save_single_image( + self, + i: int, + element: Any, + images_dir: Path, + doc_slug: str, + ) -> None: + """Saves a single extracted picture to disk using the standard naming scheme.""" + page_no = get_picture_page_no(element) + image_filename = generate_image_filename(doc_slug, page_no, i + 1) + image_path = images_dir / image_filename + try: + element.image.pil_image.save(image_path) + try: + image_path.chmod(0o666) + except Exception: + pass + logger.debug(f"Saved image: {image_path}") + except Exception as e: + logger.warning( + f"Failed to save image {image_path}: {sanitize_log_message(e)}" + ) + def _save_images( self, doc: DoclingDocument, images_dir: Path, slug: str | None = None ) -> None: @@ -651,32 +677,22 @@ def _save_images( doc_slug = slug or (generate_doc_slug(doc.name) if getattr(doc, "name", None) else "document") - def save_image(i, element): - page_no = get_picture_page_no(element) - image_filename = generate_image_filename(doc_slug, page_no, i + 1) - image_path = images_dir / image_filename - try: - element.image.pil_image.save(image_path) - try: - image_path.chmod(0o666) - except Exception: - pass - logger.debug(f"Saved image: {image_path}") - except Exception as e: - logger.warning( - f"Failed to save image {image_path}: {sanitize_log_message(e)}" - ) - if len(valid_pictures) == 1: # Sequential execution for single-image documents to avoid threadpool overhead i, element = valid_pictures[0] - save_image(i, element) + self._process_and_save_single_image(i, element, images_dir, doc_slug) else: # Parallel execution for multi-image files, dynamically bounding workers max_workers = min(len(valid_pictures), 32) with ThreadPoolExecutor(max_workers=max_workers) as executor: for i, element in valid_pictures: - executor.submit(save_image, i, element) + executor.submit( + self._process_and_save_single_image, + i, + element, + images_dir, + doc_slug, + ) class EnhancedDoclingConverter: diff --git a/src/docling_lib/utils.py b/src/docling_lib/utils.py index dfe8f5e..bde392a 100644 --- a/src/docling_lib/utils.py +++ b/src/docling_lib/utils.py @@ -49,6 +49,129 @@ def sanitize_log_message(message: Any) -> str: return sanitized +def _parse_workbook_relationships(wb_rels_tree: Any, ns_rel: str) -> dict[str, str]: + """Extracts target paths from workbook relationships XML.""" + wb_rels: dict[str, str] = {} + for rel in wb_rels_tree.findall(f"{{{ns_rel}}}Relationship"): + r_id = rel.attrib.get("Id") + target = rel.attrib.get("Target", "") + if r_id and target: + target = target.lstrip("/") + if not target.startswith("xl/"): + target = "xl/" + target + wb_rels[r_id] = target + return wb_rels + + +def _parse_sheet_relationships( + sheet_rels_tree: Any, sheet_dir: str, ns_rel: str +) -> dict[str, str]: + """Extracts drawing target paths from sheet relationships XML.""" + sheet_rels: dict[str, str] = {} + for rel in sheet_rels_tree.findall(f"{{{ns_rel}}}Relationship"): + r_id_s = rel.attrib.get("Id") + target_s = rel.attrib.get("Target", "") + if r_id_s and target_s: + target_s = target_s.lstrip("/") + if not target_s.startswith("xl/") and not target_s.startswith("../"): + target_s = f"{sheet_dir}/{target_s}" + elif target_s.startswith("../"): + target_s = "xl/" + target_s.split("../", 1)[1] + sheet_rels[r_id_s] = target_s + return sheet_rels + + +def _parse_drawing_relationships( + drawing_rels_tree: Any, drawing_dir: str, ns_rel: str +) -> dict[str, str]: + """Extracts media target paths from drawing relationships XML.""" + drawing_rels: dict[str, str] = {} + for rel in drawing_rels_tree.findall(f"{{{ns_rel}}}Relationship"): + r_id_d = rel.attrib.get("Id") + target_d = rel.attrib.get("Target", "") + if r_id_d and target_d: + target_d = target_d.lstrip("/") + if "media/" in target_d: + target_d = "xl/media/" + target_d.rsplit("media/", 1)[1] + elif not target_d.startswith("xl/"): + target_d = f"{drawing_dir}/{target_d}" + drawing_rels[r_id_d] = target_d + return drawing_rels + + +def _extract_anchors_from_drawing( + zf: Any, + drawing_tree: Any, + drawing_rels: dict[str, str], + sheet_name: str, + sheet_idx: int, + namelist: list[str], + ns_xdr: str, + ns_a: str, + rel_embed_attr: str, +) -> list[dict[str, Any]]: + """Parses image anchors from a drawing XML tree and returns extracted media items.""" + import hashlib + import io + + from PIL import Image as PILImage + + extracted: list[dict[str, Any]] = [] + anchors = drawing_tree.findall( + f"{{{ns_xdr}}}twoCellAnchor" + ) + drawing_tree.findall(f"{{{ns_xdr}}}oneCellAnchor") + + for anchor in anchors: + from_node = anchor.find(f"{{{ns_xdr}}}from") + row_idx = 0 + col_idx = 0 + if from_node is not None: + col_elem = from_node.find(f"{{{ns_xdr}}}col") + row_elem = from_node.find(f"{{{ns_xdr}}}row") + if col_elem is not None and col_elem.text: + col_idx = int(col_elem.text) + if row_elem is not None and row_elem.text: + row_idx = int(row_elem.text) + + blip = anchor.find(f".//{{{ns_a}}}blip") + if blip is None: + continue + + embed_id = blip.attrib.get(rel_embed_attr) + if not embed_id: + continue + + media_path = drawing_rels.get(embed_id) + if not media_path or media_path not in namelist: + continue + + img_bytes = zf.read(media_path) + filename = media_path.rsplit("/", 1)[-1] + pil_img = None + try: + pil_img = PILImage.open(io.BytesIO(img_bytes)) + pil_img.load() + except Exception: + pass + + sha256_hash = hashlib.sha256(img_bytes).hexdigest() + extracted.append( + { + "sheet_name": sheet_name, + "sheet_index": sheet_idx, + "row": row_idx, + "col": col_idx, + "filename": filename, + "image_bytes": img_bytes, + "pil_image": pil_img, + "media_path": media_path, + "sha256": sha256_hash, + } + ) + + return extracted + + def extract_excel_images(file_path_or_bytes: Any) -> list[dict[str, Any]]: """ Extracts embedded and pasted images from an Excel (.xlsx) file by traversing @@ -57,12 +180,10 @@ def extract_excel_images(file_path_or_bytes: Any) -> list[dict[str, Any]]: Returns a list of dicts containing sheet_name, sheet_index (1-based), row, col, filename, image_bytes, pil_image, media_path, and sha256 hash. """ - import hashlib import io import zipfile import defusedxml.ElementTree as ET - from PIL import Image as PILImage ns_main = "http://schemas.openxmlformats.org/spreadsheetml/2006/main" ns_rel = "http://schemas.openxmlformats.org/package/2006/relationships" @@ -100,15 +221,7 @@ def extract_excel_images(file_path_or_bytes: Any) -> list[dict[str, Any]]: wb_tree = ET.fromstring(zf.read("xl/workbook.xml")) wb_rels_tree = ET.fromstring(zf.read("xl/_rels/workbook.xml.rels")) - wb_rels: dict[str, str] = {} - for rel in wb_rels_tree.findall(f"{{{ns_rel}}}Relationship"): - r_id = rel.attrib.get("Id") - target = rel.attrib.get("Target", "") - if r_id and target: - target = target.lstrip("/") - if not target.startswith("xl/"): - target = "xl/" + target - wb_rels[r_id] = target + wb_rels = _parse_workbook_relationships(wb_rels_tree, ns_rel) sheets = wb_tree.find(f"{{{ns_main}}}sheets") if sheets is None: @@ -129,17 +242,7 @@ def extract_excel_images(file_path_or_bytes: Any) -> list[dict[str, Any]]: continue sheet_rels_tree = ET.fromstring(zf.read(sheet_rels_path)) - sheet_rels: dict[str, str] = {} - for rel in sheet_rels_tree.findall(f"{{{ns_rel}}}Relationship"): - r_id_s = rel.attrib.get("Id") - target_s = rel.attrib.get("Target", "") - if r_id_s and target_s: - target_s = target_s.lstrip("/") - if not target_s.startswith("xl/") and not target_s.startswith("../"): - target_s = f"{sheet_dir}/{target_s}" - elif target_s.startswith("../"): - target_s = "xl/" + target_s.split("../", 1)[1] - sheet_rels[r_id_s] = target_s + sheet_rels = _parse_sheet_relationships(sheet_rels_tree, sheet_dir, ns_rel) sheet_tree = ET.fromstring(zf.read(sheet_path)) drawing_node = sheet_tree.find(f"{{{ns_main}}}drawing") @@ -157,70 +260,24 @@ def extract_excel_images(file_path_or_bytes: Any) -> list[dict[str, Any]]: continue drawing_rels_tree = ET.fromstring(zf.read(drawing_rels_path)) - drawing_rels: dict[str, str] = {} - for rel in drawing_rels_tree.findall(f"{{{ns_rel}}}Relationship"): - r_id_d = rel.attrib.get("Id") - target_d = rel.attrib.get("Target", "") - if r_id_d and target_d: - target_d = target_d.lstrip("/") - if "media/" in target_d: - target_d = "xl/media/" + target_d.rsplit("media/", 1)[1] - elif not target_d.startswith("xl/"): - target_d = f"{drawing_dir}/{target_d}" - drawing_rels[r_id_d] = target_d + drawing_rels = _parse_drawing_relationships( + drawing_rels_tree, drawing_dir, ns_rel + ) drawing_tree = ET.fromstring(zf.read(drawing_path)) - anchors = drawing_tree.findall( - f"{{{ns_xdr}}}twoCellAnchor" - ) + drawing_tree.findall(f"{{{ns_xdr}}}oneCellAnchor") - - for anchor in anchors: - from_node = anchor.find(f"{{{ns_xdr}}}from") - row_idx = 0 - col_idx = 0 - if from_node is not None: - col_elem = from_node.find(f"{{{ns_xdr}}}col") - row_elem = from_node.find(f"{{{ns_xdr}}}row") - if col_elem is not None and col_elem.text: - col_idx = int(col_elem.text) - if row_elem is not None and row_elem.text: - row_idx = int(row_elem.text) - - blip = anchor.find(f".//{{{ns_a}}}blip") - if blip is None: - continue - - embed_id = blip.attrib.get(rel_embed_attr) - if not embed_id: - continue - - media_path = drawing_rels.get(embed_id) - if not media_path or media_path not in namelist: - continue - - img_bytes = zf.read(media_path) - filename = media_path.rsplit("/", 1)[-1] - pil_img = None - try: - pil_img = PILImage.open(io.BytesIO(img_bytes)) - pil_img.load() - except Exception: - pass - - sha256_hash = hashlib.sha256(img_bytes).hexdigest() - extracted.append( - { - "sheet_name": sheet_name, - "sheet_index": sheet_idx, - "row": row_idx, - "col": col_idx, - "filename": filename, - "image_bytes": img_bytes, - "pil_image": pil_img, - "media_path": media_path, - "sha256": sha256_hash, - } - ) + drawing_anchors = _extract_anchors_from_drawing( + zf=zf, + drawing_tree=drawing_tree, + drawing_rels=drawing_rels, + sheet_name=sheet_name, + sheet_idx=sheet_idx, + namelist=namelist, + ns_xdr=ns_xdr, + ns_a=ns_a, + rel_embed_attr=rel_embed_attr, + ) + extracted.extend(drawing_anchors) + except Exception as exc: logger.warning( f"Error parsing OpenXML drawing relationships: {sanitize_log_message(exc)}" diff --git a/tests/test_refactor_characterization.py b/tests/test_refactor_characterization.py new file mode 100644 index 0000000..9fd5629 --- /dev/null +++ b/tests/test_refactor_characterization.py @@ -0,0 +1,67 @@ +import io +import zipfile +from pathlib import Path +from unittest.mock import MagicMock + +import pytest + +from src.docling_lib.config import DOCLING_ARTIFACTS_PATH +from src.docling_lib.converter import DocumentConversionOptions, PDFConverter +from src.docling_lib.utils import ( + extract_excel_images, + generate_doc_slug, + generate_image_filename, + get_picture_page_no, +) + + +def test_characterization_extract_excel_images_invalid_inputs(): + """Verify extract_excel_images handling of invalid files or non-zip bytes.""" + assert extract_excel_images(b"invalid_bytes_content") == [] + assert extract_excel_images(Path("non_existent_file_path.xlsx")) == [] + + +def test_characterization_extract_excel_images_empty_zip(): + """Verify extract_excel_images handling of zip archives lacking OpenXML structure.""" + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w") as zf: + zf.writestr("dummy.txt", "content") + assert extract_excel_images(buf.getvalue()) == [] + + +def test_characterization_slug_and_filename_generation(): + """Verify boundary cases for slug and image filename generation.""" + assert generate_doc_slug(None) == "document" + assert generate_doc_slug("") == "document" + assert generate_doc_slug(" ") == "document" + assert generate_doc_slug("My/Doc:Name*.pdf") == "Doc-Name" + assert generate_doc_slug("Doc Name.pdf") == "Doc-Name" + + assert generate_image_filename("slug", 1, 1) == "slug_p1_1.png" + # Negative/zero values bounded to 1 + assert generate_image_filename("slug", -5, 0) == "slug_p1_1.png" + + +def test_characterization_get_picture_page_no(): + """Verify page number extraction from picture item provenance.""" + pic = MagicMock() + pic.prov = [] + assert get_picture_page_no(pic) == 1 + + prov_item = MagicMock() + prov_item.page_no = 3 + pic.prov = [prov_item] + assert get_picture_page_no(pic) == 3 + + +def test_characterization_pdf_converter_format_options(): + """Verify PDFConverter initializes format options properly.""" + options = DocumentConversionOptions() + converter = PDFConverter(options) + assert converter.options.image_dir_name == "images" + assert converter.doc_converter is not None + + +def test_characterization_config_artifacts_path(): + """Verify DOCLING_ARTIFACTS_PATH configuration exists and is properly typed.""" + assert DOCLING_ARTIFACTS_PATH is None or isinstance(DOCLING_ARTIFACTS_PATH, Path) From 651a8856aa6d233292b042d5e9f37e80edffa38b Mon Sep 17 00:00:00 2001 From: chottokun Date: Thu, 15 Oct 2026 22:50:52 +0900 Subject: [PATCH 2/2] test: remove unused imports, add adversarial tests, and harden anchor coordinate parsing --- src/docling_lib/utils.py | 10 +- tests/test_refactor_characterization.py | 181 +++++++++++++++++++++++- 2 files changed, 188 insertions(+), 3 deletions(-) diff --git a/src/docling_lib/utils.py b/src/docling_lib/utils.py index bde392a..a2adb13 100644 --- a/src/docling_lib/utils.py +++ b/src/docling_lib/utils.py @@ -129,9 +129,15 @@ def _extract_anchors_from_drawing( col_elem = from_node.find(f"{{{ns_xdr}}}col") row_elem = from_node.find(f"{{{ns_xdr}}}row") if col_elem is not None and col_elem.text: - col_idx = int(col_elem.text) + try: + col_idx = int(col_elem.text) + except (ValueError, TypeError): + col_idx = 0 if row_elem is not None and row_elem.text: - row_idx = int(row_elem.text) + try: + row_idx = int(row_elem.text) + except (ValueError, TypeError): + row_idx = 0 blip = anchor.find(f".//{{{ns_a}}}blip") if blip is None: diff --git a/tests/test_refactor_characterization.py b/tests/test_refactor_characterization.py index 9fd5629..90093f3 100644 --- a/tests/test_refactor_characterization.py +++ b/tests/test_refactor_characterization.py @@ -3,11 +3,15 @@ from pathlib import Path from unittest.mock import MagicMock -import pytest +import defusedxml.ElementTree as ET from src.docling_lib.config import DOCLING_ARTIFACTS_PATH from src.docling_lib.converter import DocumentConversionOptions, PDFConverter from src.docling_lib.utils import ( + _extract_anchors_from_drawing, + _parse_drawing_relationships, + _parse_sheet_relationships, + _parse_workbook_relationships, extract_excel_images, generate_doc_slug, generate_image_filename, @@ -65,3 +69,178 @@ def test_characterization_pdf_converter_format_options(): def test_characterization_config_artifacts_path(): """Verify DOCLING_ARTIFACTS_PATH configuration exists and is properly typed.""" assert DOCLING_ARTIFACTS_PATH is None or isinstance(DOCLING_ARTIFACTS_PATH, Path) + + +# --- 批判的テスト（Adversarial & Boundary Tests） --- + + +def test_adversarial_parse_workbook_relationships(): + """XMLツリーの欠落・異常値・パス解決の耐性を検証。""" + ns_rel = "http://schemas.openxmlformats.org/package/2006/relationships" + + # 空のツリー + empty_tree = ET.fromstring("") + assert _parse_workbook_relationships(empty_tree, ns_rel) == {} + + # Id欠落、Target欠落、xl/プレフィックス有無、先頭スラッシュ + xml = f""" + + + + + + + + """ + tree = ET.fromstring(xml) + parsed = _parse_workbook_relationships(tree, ns_rel) + assert parsed["rId1"] == "xl/worksheets/sheet1.xml" + assert parsed["rId2"] == "xl/worksheets/sheet2.xml" + assert parsed["rId3"] == "xl/worksheets/sheet3.xml" + assert "rId4" not in parsed + assert len(parsed) == 3 + + +def test_adversarial_parse_sheet_relationships(): + """シートリレーションシップの相対パス解決と異常値の耐性を検証。""" + ns_rel = "http://schemas.openxmlformats.org/package/2006/relationships" + + xml = f""" + + + + + + + """ + tree = ET.fromstring(xml) + parsed = _parse_sheet_relationships(tree, sheet_dir="xl/worksheets", ns_rel=ns_rel) + assert parsed["rId1"] == "xl/drawings/drawing1.xml" + assert parsed["rId2"] == "xl/worksheets/drawings/drawing2.xml" + assert parsed["rId3"] == "xl/drawings/drawing3.xml" + assert "rId4" not in parsed + + +def test_adversarial_parse_drawing_relationships(): + """ドローイングリレーションシップのメディアパス正規化を検証。""" + ns_rel = "http://schemas.openxmlformats.org/package/2006/relationships" + + xml = f""" + + + + + + + """ + tree = ET.fromstring(xml) + parsed = _parse_drawing_relationships(tree, drawing_dir="xl/drawings", ns_rel=ns_rel) + assert parsed["rId1"] == "xl/media/image1.png" + assert parsed["rId2"] == "xl/media/image2.jpeg" + assert parsed["rId3"] == "xl/media/image3.png" + assert parsed["rId4"] == "xl/drawings/custom/dir/other.png" + + +def test_adversarial_extract_anchors_corrupt_data(): + """壊れたXML要素、不正な座標、壊れた画像バイナリへの耐性を検証。""" + ns_xdr = "http://schemas.openxmlformats.org/drawingml/2006/spreadsheetDrawing" + ns_a = "http://schemas.openxmlformats.org/drawingml/2006/main" + rel_embed_attr = "{http://schemas.openxmlformats.org/officeDocument/2006/relationships}embed" + + drawing_xml = f""" + + + + + invalid_col + + + + + + + + + + + + 1 + 2 + + + + + + 3 + 4 + + + + + + + + + """ + drawing_tree = ET.fromstring(drawing_xml) + + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w") as zf: + # 壊れた画像（無効なPNGバイナリ） + zf.writestr("xl/media/corrupt.png", b"not_a_valid_image_binary") + + with zipfile.ZipFile(io.BytesIO(buf.getvalue()), "r") as zf: + drawing_rels = { + "rId1": "xl/media/corrupt.png", + "rIdMissing": "xl/media/not_exist.png", + } + anchors = _extract_anchors_from_drawing( + zf=zf, + drawing_tree=drawing_tree, + drawing_rels=drawing_rels, + sheet_name="TestSheet", + sheet_idx=1, + namelist=zf.namelist(), + ns_xdr=ns_xdr, + ns_a=ns_a, + rel_embed_attr=rel_embed_attr, + ) + + # 破損画像でも例外を吐かずに抽出され、pil_imageはNoneとしてハンドリングされること + assert len(anchors) == 1 + assert anchors[0]["sheet_name"] == "TestSheet" + assert anchors[0]["row"] == 0 # 空文字はデフォルト0 + assert anchors[0]["col"] == 0 # int変換失敗時はデフォルト0 + assert anchors[0]["pil_image"] is None + assert anchors[0]["image_bytes"] == b"not_a_valid_image_binary" + assert len(anchors[0]["sha256"]) == 64 + + +def test_adversarial_pdf_converter_helpers(tmp_path: Path): + """PDFConverter の分離メソッド（_hash_existing_pictures, _process_and_save_single_image）の耐性を検証。""" + converter = PDFConverter(DocumentConversionOptions()) + + # 1. _hash_existing_pictures の異常系耐性 + mock_doc = MagicMock() + mock_pic_broken = MagicMock() + mock_pic_broken.image.pil_image.save.side_effect = RuntimeError("Save failure") + mock_doc.pictures = [mock_pic_broken] + + # 例外がスルーされクラッシュしないこと + hashes = converter._hash_existing_pictures(mock_doc) + assert isinstance(hashes, set) + assert len(hashes) == 0 + + # 2. _process_and_save_single_image の異常系耐性 + mock_element = MagicMock() + mock_element.prov = [] + mock_element.image.pil_image.save.side_effect = PermissionError("Cannot write") + + # 例外が内部でハンドリングされ、呼び出し元に波及しないこと + converter._process_and_save_single_image( + i=0, + element=mock_element, + images_dir=tmp_path, + doc_slug="test_doc", + ) +