"""Vendored deterministic pipeline for KDL-Frontier-Parser-nano / rakedoc-nano. Source: run-llama/ParseBench, src/parse_bench/inference/providers/parse/ kdl_frontier_nano.py (Apache-2.0). Reproduced here verbatim except that the ParseBench framework imports, the httpx transport helper and the `Provider` subclass have been removed -- this Space drives the exact same stage prompts, image preprocessing, OTSL conversion and markdown assembly against a local `transformers` model instead of a vLLM HTTP endpoint. """ import asyncio import base64 import enum import html import io import itertools import json import logging import math import os import re import unicodedata from collections import Counter from datetime import datetime from decimal import Decimal from pathlib import Path from typing import Any, Dict, List, Literal, Match, Tuple, Union, cast import markdown as _markdown_lib from PIL import Image from pydantic import BaseModel, Field, computed_field, model_validator logger = logging.getLogger("kdl_frontier_nano") # vendored modules below were written against `from enum import Enum`-style # imports; provide the same names without the original package layout. Enum = enum.Enum # from deepparser_v2/config/model_flow.py (verbatim values) from typing import Literal as _Literal # explicit: alias below needs Literal at module init RecognitionStage = _Literal["text", "table", "picture", "formula"] RECOGNITION_STAGES: tuple = ("text", "table", "picture", "formula") # ========================================================================== # [vendored] element_schema(ElementCategory) # ========================================================================== class ElementCategory(str, Enum): """문서 요소 카테고리""" # 텍스트 요소 TITLE = "Title" SECTION_HEADER = "Section-header" TEXT = "Text" PAGE_HEADER = "Page-header" PAGE_FOOTER = "Page-footer" LIST_ITEM = "List-item" CAPTION = "Caption" FOOTNOTE = "Footnote" # 시각적 요소 TABLE = "Table" PICTURE = "Picture" DATA_CHART = "Chart" FLOW_DIAGRAM = "Flowchart" # 수학 요소 FORMULA = "Formula" # ========================================================================== # [vendored] recognition_contract # ========================================================================== """Recognition bucket contract shared by planning and layout grouping.""" # [vendor-strip] from __future__ import annotations from typing import Sequence # [vendor-strip] from ..schemas.element_schema import ElementCategory # [vendor-strip] from .model_flow import RECOGNITION_STAGES, RecognitionStage RecognitionBucket = RecognitionStage RECOGNITION_BUCKETS: tuple[RecognitionBucket, ...] = RECOGNITION_STAGES CATEGORY_TO_RECOGNITION_BUCKET: dict[str, RecognitionBucket] = { ElementCategory.TITLE.value: "text", ElementCategory.SECTION_HEADER.value: "text", ElementCategory.TEXT.value: "text", ElementCategory.PAGE_HEADER.value: "text", ElementCategory.PAGE_FOOTER.value: "text", ElementCategory.LIST_ITEM.value: "text", ElementCategory.CAPTION.value: "text", ElementCategory.FOOTNOTE.value: "text", ElementCategory.TABLE.value: "table", ElementCategory.PICTURE.value: "picture", ElementCategory.DATA_CHART.value: "picture", ElementCategory.FLOW_DIAGRAM.value: "picture", ElementCategory.FORMULA.value: "formula", } def category_to_recognition_bucket( category: str | ElementCategory, ) -> RecognitionBucket: """Return the recognition bucket for a public element category.""" return CATEGORY_TO_RECOGNITION_BUCKET.get(_category_value(category), "text") def target_categories_to_recognition_buckets( target_categories: Sequence[str] | None, ) -> frozenset[RecognitionBucket] | None: """Map an optional public category filter to executable recognition buckets.""" if not target_categories: return None return frozenset( CATEGORY_TO_RECOGNITION_BUCKET[category] for category in (_category_value(category) for category in target_categories) if category in CATEGORY_TO_RECOGNITION_BUCKET ) def _category_value(category: str | ElementCategory) -> str: if isinstance(category, ElementCategory): return category.value return str(category).strip() # ========================================================================== # [vendored] layout_contract # ========================================================================== """Layout category contract shared by layout provider adapters.""" from collections.abc import Mapping from typing import Any # [vendor-strip] from ..config.recognition_contract import category_to_recognition_bucket # [vendor-strip] from ..schemas.element_schema import ElementCategory CANONICAL_LAYOUT_CATEGORIES = ( ElementCategory.TITLE.value, ElementCategory.SECTION_HEADER.value, ElementCategory.TEXT.value, ElementCategory.LIST_ITEM.value, ElementCategory.TABLE.value, ElementCategory.PICTURE.value, ElementCategory.DATA_CHART.value, ElementCategory.FLOW_DIAGRAM.value, ElementCategory.FORMULA.value, ElementCategory.CAPTION.value, ElementCategory.FOOTNOTE.value, ElementCategory.PAGE_HEADER.value, ElementCategory.PAGE_FOOTER.value, ) DEFAULT_LAYOUT_CATEGORY = ElementCategory.TEXT.value _CANONICAL_BY_CASEFOLD = { category.casefold(): category for category in CANONICAL_LAYOUT_CATEGORIES } def normalize_layout_category( raw_category: Any, provider_category_map: Mapping[str, str] | None = None, ) -> str: """Normalize a provider label into the layout-stage category contract.""" mapped_category = _map_provider_category(raw_category, provider_category_map) return _canonicalize_category(mapped_category) def layout_recognition_bucket(category: Any) -> str: """Return the recognition bucket used immediately after layout detection.""" normalized_category = normalize_layout_category(category) return category_to_recognition_bucket(normalized_category) def _map_provider_category( raw_category: Any, provider_category_map: Mapping[str, str] | None, ) -> str: category = _stringify_category(raw_category) if not provider_category_map: return category return ( provider_category_map.get(category) or provider_category_map.get(category.casefold()) or category ) def _canonicalize_category(category: Any) -> str: canonical = _CANONICAL_BY_CASEFOLD.get(_stringify_category(category).casefold()) if canonical is None: return DEFAULT_LAYOUT_CATEGORY return canonical def _stringify_category(category: Any) -> str: if category is None: return "" return str(category).strip() # ========================================================================== # [vendored] image_preprocessing # ========================================================================== # [vendor-strip] from __future__ import annotations import math from typing import Tuple # [vendor-strip] from loguru import logger from PIL import Image from pydantic import BaseModel, Field # 이미지 처리 설정값 class ImageConfig: """이미지 처리 관련 설정값""" # DotsOCR 설정 FACTOR = 28 MIN_PIXELS = 147384 * 2 MAX_PIXELS = 2822400 # MAX_PIXELS = 1638400 MAX_ASPECT_RATIO = 200 # 공통 설정 DEFAULT_BACKGROUND_COLOR = (255, 255, 255) # 흰색 배경 class PageContentMetrics(BaseModel): """Lightweight page-content signals used to separate blank pages from misses.""" foreground_ratio: float = Field(..., ge=0.0, le=1.0) edge_ratio: float = Field(..., ge=0.0, le=1.0) intensity_variance: float = Field(..., ge=0.0) is_blank: bool CONTENT_ANALYSIS_MAX_DIMENSION = 192 FOREGROUND_DELTA_THRESHOLD = 12 FOREGROUND_DARK_THRESHOLD = 250 EDGE_DELTA_THRESHOLD = 18 BLANK_FOREGROUND_RATIO_THRESHOLD = 0.002 BLANK_EDGE_RATIO_THRESHOLD = 0.001 BLANK_INTENSITY_VARIANCE_THRESHOLD = 4.0 def _round_by_factor(number: int, factor: int) -> int: """주어진 factor의 배수로 반올림""" return round(number / factor) * factor def _ceil_by_factor(number: int, factor: int) -> int: """주어진 factor의 배수로 올림""" return math.ceil(number / factor) * factor def _floor_by_factor(number: int, factor: int) -> int: """주어진 factor의 배수로 내림""" return math.floor(number / factor) * factor def normalize_image_mode(image: Image.Image, target_mode: str = "RGB") -> Image.Image: """이미지를 지정된 모드로 변환 (투명 배경 처리 포함)""" if image.mode == target_mode: return image if image.mode == "RGBA" and target_mode == "RGB": # RGBA를 RGB로 변환 시 흰색 배경 추가 background = Image.new("RGB", image.size, ImageConfig.DEFAULT_BACKGROUND_COLOR) background.paste(image, mask=image.split()[3]) # 알파 채널을 마스크로 사용 return background return image.convert(target_mode) def calculate_dimensions( width: int, height: int, factor: int = ImageConfig.FACTOR, min_pixels: int = ImageConfig.MIN_PIXELS, max_pixels: int = ImageConfig.MAX_PIXELS, ) -> Tuple[int, int]: """이미지 크기 계산""" # 종횡비 검증 aspect_ratio = max(height, width) / min(height, width) if aspect_ratio > ImageConfig.MAX_ASPECT_RATIO: raise ValueError( f"종횡비가 너무 큽니다. 최대 {ImageConfig.MAX_ASPECT_RATIO}:1, " f"현재 {aspect_ratio:.1f}:1" ) # factor의 배수로 조정 target_height = max(factor, _round_by_factor(height, factor)) target_width = max(factor, _round_by_factor(width, factor)) # 최대 픽셀 수 초과 시 축소 if target_height * target_width > max_pixels: scale = math.sqrt((height * width) / max_pixels) target_height = max(factor, _floor_by_factor(int(height / scale), factor)) target_width = max(factor, _floor_by_factor(int(width / scale), factor)) # 최소 픽셀 수 미만 시 확대 elif target_height * target_width < min_pixels: scale = math.sqrt(min_pixels / (height * width)) target_height = _ceil_by_factor(int(height * scale), factor) target_width = _ceil_by_factor(int(width * scale), factor) # 확대 후 최대 픽셀 수 재검사 if target_height * target_width > max_pixels: scale = math.sqrt((target_height * target_width) / max_pixels) target_height = max( factor, _floor_by_factor(int(target_height / scale), factor) ) target_width = max( factor, _floor_by_factor(int(target_width / scale), factor) ) return target_width, target_height def smart_resize(image: Image.Image) -> Image.Image: """단일 PIL.Image 객체를 받아 리사이즈 후 반환""" w, h = image.size try: target_width, target_height = calculate_dimensions(w, h) if (target_width, target_height) != (w, h): resized_img = image.resize( (target_width, target_height), resample=Image.Resampling.BICUBIC, ) return resized_img else: return image except Exception: raise ValueError(f"Failed to resize image: {image}") def is_monochromatic(image: Image.Image) -> bool: """ 이미지가 완전 단색인지 확인합니다. 모든 픽셀이 동일한 색상인 경우 True를 반환합니다. Args: image: PIL 이미지 Returns: bool: 단색이면 True, 아니면 False """ try: # getcolors()는 [(픽셀수, (R,G,B)), ...] 형태로 반환 # 색상이 1개만 있으면 단색 colors = image.getcolors() return colors is not None and len(colors) == 1 except Exception as e: logger.warning(f"Failed to check monochromatic: {e}") return False def analyze_page_content(image: Image.Image) -> PageContentMetrics: """Estimate whether a page has visible content without calling a detector.""" try: normalized = normalize_image_mode(image, "RGB") width, height = normalized.size scale = min(1.0, CONTENT_ANALYSIS_MAX_DIMENSION / max(width, height)) if scale < 1.0: sample_size = (max(1, int(width * scale)), max(1, int(height * scale))) normalized = normalized.resize( sample_size, resample=Image.Resampling.BILINEAR, ) grayscale = normalized.convert("L") pixels = list(grayscale.getdata()) total_pixels = len(pixels) if total_pixels == 0: return PageContentMetrics( foreground_ratio=0.0, edge_ratio=0.0, intensity_variance=0.0, is_blank=True, ) sorted_pixels = sorted(pixels) background_index = min( total_pixels - 1, int(total_pixels * 0.95), ) background_level = sorted_pixels[background_index] foreground_pixels = sum( 1 for value in pixels if ( background_level - value >= FOREGROUND_DELTA_THRESHOLD and value < FOREGROUND_DARK_THRESHOLD ) ) foreground_ratio = foreground_pixels / total_pixels mean = sum(pixels) / total_pixels intensity_variance = sum((value - mean) ** 2 for value in pixels) / total_pixels sample_width, sample_height = grayscale.size edge_pairs = 0 edge_hits = 0 for y in range(sample_height): row_offset = y * sample_width for x in range(sample_width): value = pixels[row_offset + x] if x + 1 < sample_width: edge_pairs += 1 if abs(value - pixels[row_offset + x + 1]) >= EDGE_DELTA_THRESHOLD: edge_hits += 1 if y + 1 < sample_height: edge_pairs += 1 if ( abs(value - pixels[row_offset + sample_width + x]) >= EDGE_DELTA_THRESHOLD ): edge_hits += 1 edge_ratio = edge_hits / edge_pairs if edge_pairs else 0.0 is_blank = ( foreground_ratio < BLANK_FOREGROUND_RATIO_THRESHOLD and edge_ratio < BLANK_EDGE_RATIO_THRESHOLD and intensity_variance < BLANK_INTENSITY_VARIANCE_THRESHOLD ) return PageContentMetrics( foreground_ratio=round(foreground_ratio, 6), edge_ratio=round(edge_ratio, 6), intensity_variance=round(float(intensity_variance), 6), is_blank=is_blank, ) except Exception as e: logger.warning(f"Failed to analyze page content: {e}") return PageContentMetrics( foreground_ratio=0.0, edge_ratio=0.0, intensity_variance=0.0, is_blank=False, ) def preprocess_for_vlm(image: Image.Image) -> Image.Image: """ VLM 입력을 위한 이미지 전처리 Args: image: 원본 PIL 이미지 Returns: 전처리된 PIL 이미지 """ # 1. 이미지 모드 정규화 image = normalize_image_mode(image, "RGB") # 2. 스마트 리사이즈 image = smart_resize(image) return image # ========================================================================== # [vendored] native_layout # ========================================================================== """the model layout token parsing utilities.""" import re from typing import Any from PIL import Image from pydantic import BaseModel, Field # [vendor-strip] from .layout_contract import normalize_layout_category NATIVE_LAYOUT_IMAGE_SIZE = (1036, 1036) _NATIVE_LAYOUT_RE = re.compile( r"<\|box_start\|>(\d+)\s+(\d+)\s+(\d+)\s+(\d+)" r"<\|box_end\|><\|ref_start\|>([^<]+?)<\|ref_end\|>" r"(?:(<\|rotate_(?:up|right|down|left)\|>))?", re.DOTALL, ) _ROTATION_ANGLES = { "<|rotate_up|>": 0, "<|rotate_right|>": 90, "<|rotate_down|>": 180, "<|rotate_left|>": 270, } NATIVE_LAYOUT_CATEGORY_MAP = { "algorithm": "Text", "aside_text": "Text", "chart": "Chart", "code": "Text", "code_caption": "Caption", "equation": "Formula", "equation_block": "Formula", "footer": "Page-footer", "header": "Page-header", "image": "Picture", "image_block": "Picture", "image_caption": "Caption", "image_footnote": "Footnote", "inline_formula": "Formula", "list": "List-item", "list_item": "List-item", "page_footnote": "Footnote", "page_number": "Page-footer", "phonetic": "Text", "ref_text": "List-item", "table": "Table", "table_caption": "Caption", "table_footnote": "Footnote", "text": "Text", "title": "Title", "unknown": "Text", } _LIST_CHILD_CATEGORIES = {"text", "list_item", "ref_text"} _IMAGE_BLOCK_CHILD_CATEGORIES = { "chart", "image", "image_caption", "image_footnote", } _EQUATION_CHILD_CATEGORIES = {"equation", "inline_formula"} _ATTACHMENT_PARENT_RAW_CATEGORIES = { "table_caption": {"table"}, "table_footnote": {"table"}, "image_caption": {"image", "chart"}, "image_footnote": {"image", "chart"}, "code_caption": {"code", "algorithm"}, } class NativeLayoutItem(BaseModel): """Raw the model layout item parsed from special tokens.""" bbox: list[float] raw_bbox: list[int] raw_category: str layout_order: int angle: int = 0 class NormalizedNativeLayoutItem(BaseModel): """the model layout item normalized to DeepParser categories.""" bbox: list[float] category: str layout_order: int raw_category: str angle: int = 0 metadata: dict[str, Any] = Field(default_factory=dict) def to_output_dict(self) -> dict[str, Any]: data = self.model_dump() if not data["metadata"]: data.pop("metadata") return data def is_native_layout_response(content: Any) -> bool: return isinstance(content, str) and "<|box_start|>" in content def prepare_native_layout_image(image: Image.Image) -> Image.Image: return image.convert("RGB").resize( NATIVE_LAYOUT_IMAGE_SIZE, Image.Resampling.BICUBIC, ) def parse_native_raw_layout_tokens(content: str) -> list[NativeLayoutItem]: items: list[NativeLayoutItem] = [] for match in _NATIVE_LAYOUT_RE.finditer(content): converted_bbox = _convert_native_layout_bbox( (match.group(1), match.group(2), match.group(3), match.group(4)) ) if converted_bbox is None: continue raw_bbox, bbox = converted_bbox raw_category = match.group(5).strip().lower() angle = _ROTATION_ANGLES.get(match.group(6) or "", 0) items.append( NativeLayoutItem( bbox=bbox, raw_bbox=raw_bbox, raw_category=raw_category, layout_order=len(items), angle=angle, ) ) return items def parse_native_layout_tokens(content: str) -> list[dict[str, Any]]: raw_items = parse_native_raw_layout_tokens(content) normalized_items = normalize_native_layout_items(raw_items) return [item.to_output_dict() for item in normalized_items] def normalize_native_layout_items( raw_items: list[NativeLayoutItem], ) -> list[NormalizedNativeLayoutItem]: list_containers = [ item for item in raw_items if item.raw_category == "list" and _has_contained_child(item, raw_items, _LIST_CHILD_CATEGORIES) ] image_blocks = [ item for item in raw_items if item.raw_category == "image_block" and _has_contained_child(item, raw_items, _IMAGE_BLOCK_CHILD_CATEGORIES) ] equation_blocks = [ item for item in raw_items if item.raw_category == "equation_block" ] equation_child_orders = { child.layout_order for block in equation_blocks for child in raw_items if child.raw_category in _EQUATION_CHILD_CATEGORIES and _bbox_contains(block.bbox, child.bbox) } normalized_items: list[NormalizedNativeLayoutItem] = [] for item in raw_items: metadata: dict[str, Any] = {} if item.raw_category == "list" and item in list_containers: continue if item.raw_category == "image_block" and item in image_blocks: continue if item.layout_order in equation_child_orders: continue if item.raw_category == "equation_block": children = [ _attachment_from_raw_item(child) for child in raw_items if child.raw_category in _EQUATION_CHILD_CATEGORIES and _bbox_contains(item.bbox, child.bbox) ] if children: metadata["children"] = children else: list_parent = _find_containing_parent(item, list_containers) if list_parent is not None and item.raw_category in _LIST_CHILD_CATEGORIES: metadata.update(_parent_metadata(list_parent)) image_parent = _find_containing_parent(item, image_blocks) if image_parent is not None: metadata.update(_parent_metadata(image_parent)) normalized_items.append( NormalizedNativeLayoutItem( bbox=item.bbox, category=_category_for_item(item, metadata), layout_order=item.layout_order, raw_category=item.raw_category, angle=item.angle, metadata=metadata, ) ) _attach_caption_and_footnote_refs(normalized_items) return normalized_items def _convert_native_layout_bbox( raw_bbox: tuple[str, str, str, str], ) -> tuple[list[int], list[float]] | None: try: coords = [int(value) for value in raw_bbox] except ValueError: return None if any(coord < 0 or coord > 1000 for coord in coords): return None x1, y1, x2, y2 = coords if x2 < x1: x1, x2 = x2, x1 if y2 < y1: y1, y2 = y2, y1 if x1 == x2 or y1 == y2: return None raw_coords = [x1, y1, x2, y2] normalized_coords = [round(value / 1000.0, 4) for value in raw_coords] return raw_coords, normalized_coords def _category_for_item(item: NativeLayoutItem, metadata: dict[str, Any]) -> str: if item.raw_category in _LIST_CHILD_CATEGORIES and ( item.raw_category in {"list_item", "ref_text"} or metadata.get("parent_raw_category") == "list" ): return "List-item" return normalize_layout_category(item.raw_category, NATIVE_LAYOUT_CATEGORY_MAP) def _bbox_area(bbox: list[float]) -> float: return max(0.0, bbox[2] - bbox[0]) * max(0.0, bbox[3] - bbox[1]) def _bbox_contains( parent_bbox: list[float], child_bbox: list[float], *, min_child_overlap: float = 0.85, ) -> bool: child_area = _bbox_area(child_bbox) if child_area <= 0: return False inter_x1 = max(parent_bbox[0], child_bbox[0]) inter_y1 = max(parent_bbox[1], child_bbox[1]) inter_x2 = min(parent_bbox[2], child_bbox[2]) inter_y2 = min(parent_bbox[3], child_bbox[3]) intersection = _bbox_area([inter_x1, inter_y1, inter_x2, inter_y2]) return intersection / child_area >= min_child_overlap def _has_contained_child( parent: NativeLayoutItem, items: list[NativeLayoutItem], child_categories: set[str], ) -> bool: return any( child.layout_order != parent.layout_order and child.raw_category in child_categories and _bbox_contains(parent.bbox, child.bbox) for child in items ) def _find_containing_parent( item: NativeLayoutItem, parents: list[NativeLayoutItem], ) -> NativeLayoutItem | None: candidates = [ parent for parent in parents if parent.layout_order != item.layout_order and _bbox_contains(parent.bbox, item.bbox) ] return min(candidates, key=lambda parent: _bbox_area(parent.bbox), default=None) def _parent_metadata(parent: NativeLayoutItem) -> dict[str, Any]: return { "parent_raw_category": parent.raw_category, "parent_bbox": parent.bbox, "parent_layout_order": parent.layout_order, } def _attachment_from_raw_item(item: NativeLayoutItem) -> dict[str, Any]: return { "bbox": item.bbox, "raw_category": item.raw_category, "layout_order": item.layout_order, "angle": item.angle, } def _attachment_from_normalized_item( item: NormalizedNativeLayoutItem, ) -> dict[str, Any]: return { "bbox": item.bbox, "raw_category": item.raw_category, "layout_order": item.layout_order, "angle": item.angle, } def _attach_caption_and_footnote_refs( items: list[NormalizedNativeLayoutItem], ) -> None: for item in items: parent_raw_categories = _ATTACHMENT_PARENT_RAW_CATEGORIES.get(item.raw_category) if not parent_raw_categories: continue parent = _find_nearest_attachment_parent(item, items, parent_raw_categories) if parent is None: continue key = ( "attached_footnotes" if item.raw_category.endswith("_footnote") else "attached_captions" ) parent.metadata.setdefault(key, []).append( _attachment_from_normalized_item(item) ) item.metadata.update( { "attached_to_raw_category": parent.raw_category, "attached_to_category": parent.category, "attached_to_bbox": parent.bbox, "attached_to_layout_order": parent.layout_order, } ) def _find_nearest_attachment_parent( item: NormalizedNativeLayoutItem, items: list[NormalizedNativeLayoutItem], parent_raw_categories: set[str], ) -> NormalizedNativeLayoutItem | None: candidates = [ candidate for candidate in items if candidate.layout_order != item.layout_order and candidate.raw_category in parent_raw_categories ] return min( candidates, key=lambda candidate: _attachment_distance(candidate, item), default=None, ) def _attachment_distance( parent: NormalizedNativeLayoutItem, child: NormalizedNativeLayoutItem, ) -> float: parent_center_x = (parent.bbox[0] + parent.bbox[2]) / 2 child_center_x = (child.bbox[0] + child.bbox[2]) / 2 horizontal_gap = abs(parent_center_x - child_center_x) if child.bbox[1] >= parent.bbox[3]: vertical_gap = child.bbox[1] - parent.bbox[3] elif parent.bbox[1] >= child.bbox[3]: vertical_gap = parent.bbox[1] - child.bbox[3] else: vertical_gap = 0.0 return vertical_gap + horizontal_gap # ========================================================================== # [vendored] otsl_converter # ========================================================================== # Copyright (c) 2024 PaddlePaddle Authors. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import html import itertools import re from typing import Any, Dict, List, Tuple, cast # [vendor-strip] from loguru import logger from pydantic import BaseModel, computed_field, model_validator class TableCell(BaseModel): """ TableCell represents a single cell in a table. Attributes: row_span (int): Number of rows spanned. col_span (int): Number of columns spanned. start_row_offset_idx (int): Start row index. end_row_offset_idx (int): End row index (exclusive). start_col_offset_idx (int): Start column index. end_col_offset_idx (int): End column index (exclusive). text (str): Cell text content. column_header (bool): Whether this cell is a column header. row_header (bool): Whether this cell is a row header. row_section (bool): Whether this cell is a row section. """ row_span: int = 1 col_span: int = 1 start_row_offset_idx: int end_row_offset_idx: int start_col_offset_idx: int end_col_offset_idx: int text: str column_header: bool = False row_header: bool = False row_section: bool = False @model_validator(mode="before") @classmethod def from_dict_format(cls, data: Any) -> Any: """ Create TableCell from dict, extracting 'text' property correctly. Args: data (Any): Input data. Returns: Any: TableCell-compatible dict. """ if isinstance(data, Dict): if "text" in data: return data text = data["bbox"].get("token", "") if not len(text): text_cells = data.pop("text_cell_bboxes", None) if text_cells: for el in text_cells: text += el["token"] + " " text = text.strip() data["text"] = text return data class TableData(BaseModel): """ TableData holds a table's cells, row and column counts, and provides a grid property. Attributes: table_cells (List[TableCell]): List of table cells. num_rows (int): Number of rows. num_cols (int): Number of columns. """ table_cells: List[TableCell] = [] num_rows: int = 0 num_cols: int = 0 @computed_field def grid(self) -> List[List[TableCell]]: """ Returns a 2D grid of TableCell objects for the table. Returns: List[List[TableCell]]: Table as 2D grid. """ table_data = [ [ TableCell( text="", start_row_offset_idx=i, end_row_offset_idx=i + 1, start_col_offset_idx=j, end_col_offset_idx=j + 1, ) for j in range(self.num_cols) ] for i in range(self.num_rows) ] for cell in self.table_cells: for i in range( min(cell.start_row_offset_idx, self.num_rows), min(cell.end_row_offset_idx, self.num_rows), ): for j in range( min(cell.start_col_offset_idx, self.num_cols), min(cell.end_col_offset_idx, self.num_cols), ): table_data[i][j] = cell return table_data # OTSL tag constants OTSL_NL = "" OTSL_FCEL = "" OTSL_ECEL = "" OTSL_LCEL = "" OTSL_UCEL = "" OTSL_XCEL = "" NON_CAPTURING_TAG_GROUP = "(?:|||||)" OTSL_FIND_PATTERN = re.compile( f"{NON_CAPTURING_TAG_GROUP}.*?(?={NON_CAPTURING_TAG_GROUP}|$)", flags=re.DOTALL ) IGNORABLE_OTSL_TAIL_PATTERN = re.compile( r"(?:\s|||)+", flags=re.IGNORECASE, ) def otsl_extract_tokens_and_text(s: str) -> Tuple[List[str], List[str]]: """ Extract OTSL tags and text parts from the input string. Args: s (str): OTSL string. Returns: Tuple[List[str], List[str]]: (tokens, text_parts) """ pattern = ( r"(" + r"|".join([OTSL_NL, OTSL_FCEL, OTSL_ECEL, OTSL_LCEL, OTSL_UCEL, OTSL_XCEL]) + r")" ) tokens = re.findall(pattern, s) text_parts = re.split(pattern, s) text_parts = [token for token in text_parts if token.strip()] return tokens, text_parts def otsl_parse_texts( texts: List[str], tokens: List[str] ) -> Tuple[List[TableCell], List[List[str]]]: """ Parse OTSL text and tags into TableCell objects and tag structure. Args: texts (List[str]): List of tokens and text. tokens (List[str]): List of OTSL tags. Returns: Tuple[List[TableCell], List[List[str]]]: (table_cells, split_row_tokens) """ split_word = OTSL_NL split_row_tokens = [ list(y) for x, y in itertools.groupby(tokens, lambda z: z == split_word) if not x ] table_cells: List[TableCell] = [] r_idx = 0 c_idx = 0 # Ensure matrix completeness if split_row_tokens: max_cols = max(len(row) for row in split_row_tokens) for row in split_row_tokens: while len(row) < max_cols: row.append(OTSL_ECEL) new_texts = [] text_idx = 0 for row in split_row_tokens: for token in row: new_texts.append(token) if text_idx < len(texts) and texts[text_idx] == token: text_idx += 1 if text_idx < len(texts) and texts[text_idx] not in [ OTSL_NL, OTSL_FCEL, OTSL_ECEL, OTSL_LCEL, OTSL_UCEL, OTSL_XCEL, ]: new_texts.append(texts[text_idx]) text_idx += 1 new_texts.append(OTSL_NL) if text_idx < len(texts) and texts[text_idx] == OTSL_NL: text_idx += 1 texts = new_texts def count_right( tokens: List[List[str]], c_idx: int, r_idx: int, which_tokens: List[str] ) -> int: span = 0 c_idx_iter = c_idx while tokens[r_idx][c_idx_iter] in which_tokens: c_idx_iter += 1 span += 1 if c_idx_iter >= len(tokens[r_idx]): return span return span def count_down( tokens: List[List[str]], c_idx: int, r_idx: int, which_tokens: List[str] ) -> int: span = 0 r_idx_iter = r_idx while tokens[r_idx_iter][c_idx] in which_tokens: r_idx_iter += 1 span += 1 if r_idx_iter >= len(tokens): return span return span for i, text in enumerate(texts): cell_text = "" if text in [OTSL_FCEL, OTSL_ECEL]: row_span = 1 col_span = 1 right_offset = 1 if text != OTSL_ECEL: cell_text = texts[i + 1] right_offset = 2 next_right_cell = ( texts[i + right_offset] if i + right_offset < len(texts) else "" ) next_bottom_cell = "" if r_idx + 1 < len(split_row_tokens): if c_idx < len(split_row_tokens[r_idx + 1]): next_bottom_cell = split_row_tokens[r_idx + 1][c_idx] if next_right_cell in [OTSL_LCEL, OTSL_XCEL]: col_span += count_right( split_row_tokens, c_idx + 1, r_idx, [OTSL_LCEL, OTSL_XCEL] ) if next_bottom_cell in [OTSL_UCEL, OTSL_XCEL]: row_span += count_down( split_row_tokens, c_idx, r_idx + 1, [OTSL_UCEL, OTSL_XCEL] ) table_cells.append( TableCell( text=cell_text.strip(), row_span=row_span, col_span=col_span, start_row_offset_idx=r_idx, end_row_offset_idx=r_idx + row_span, start_col_offset_idx=c_idx, end_col_offset_idx=c_idx + col_span, ) ) if text in [OTSL_FCEL, OTSL_ECEL, OTSL_LCEL, OTSL_UCEL, OTSL_XCEL]: c_idx += 1 if text == OTSL_NL: r_idx += 1 c_idx = 0 return table_cells, split_row_tokens def export_to_html(table_data: TableData) -> str: """ Export TableData to HTML table. Args: table_data (TableData): TableData object. Returns: str: HTML string. """ nrows = table_data.num_rows ncols = table_data.num_cols if len(table_data.table_cells) == 0: return "" body = "" grid = table_data.grid() if callable(table_data.grid) else table_data.grid for i in range(nrows): body += "" for j in range(ncols): cell: TableCell = grid[i][j] rowspan, rowstart = (cell.row_span, cell.start_row_offset_idx) colspan, colstart = (cell.col_span, cell.start_col_offset_idx) if rowstart != i or colstart != j: continue content = html.escape(cell.text.strip()).replace("\r\n", "\n") content = content.replace("\r", "\n").replace("\n", "
") celltag = "th" if cell.column_header else "td" opening_tag = f"{celltag}" if rowspan > 1: opening_tag += f' rowspan="{rowspan}"' if colspan > 1: opening_tag += f' colspan="{colspan}"' body += f"<{opening_tag}>{content}" body += "" body = f"{body}
" return body def otsl_pad_to_sqr_v2(otsl_str: str) -> str: """ Pad OTSL string to a square (rectangular) format, ensuring each row has equal number of cells. Args: otsl_str (str): OTSL string. Returns: str: Padded OTSL string. """ assert isinstance(otsl_str, str) otsl_str = otsl_str.strip() if OTSL_NL not in otsl_str: return otsl_str + OTSL_NL lines = otsl_str.split(OTSL_NL) row_data: List[Dict[str, Any]] = [] for line in lines: if not line: continue raw_cells = OTSL_FIND_PATTERN.findall(line) if not raw_cells: continue total_len = len(raw_cells) min_len = 0 for i, cell_str in enumerate(raw_cells): if cell_str.startswith(OTSL_FCEL): min_len = i + 1 row_data.append( {"raw_cells": raw_cells, "total_len": total_len, "min_len": min_len} ) if not row_data: return OTSL_NL global_min_width = max(row["min_len"] for row in row_data) if row_data else 0 max_total_len = max(row["total_len"] for row in row_data) if row_data else 0 search_start = global_min_width search_end = max(global_min_width, max_total_len) min_total_cost = float("inf") optimal_width = search_end for width in range(search_start, search_end + 1): current_total_cost = sum(abs(row["total_len"] - width) for row in row_data) if current_total_cost < min_total_cost: min_total_cost = current_total_cost optimal_width = width repaired_lines = [] for row in row_data: cells = cast(List[str], row["raw_cells"]) current_len = len(cells) # Never truncate rows here. Merge markers such as // # carry structural information, so cutting longer rows can drop content. if current_len > optimal_width: new_cells = cells else: padding = [OTSL_ECEL] * (optimal_width - current_len) new_cells = cells + padding repaired_lines.append("".join(new_cells)) return OTSL_NL.join(repaired_lines) + OTSL_NL def convert_otsl_to_html(otsl_content: str) -> str: """ Convert OTSL-v1.0 string to HTML. Only 6 tags allowed: , , , , , . Args: otsl_content (str): OTSL string. Returns: str: HTML table. """ otsl_content = otsl_pad_to_sqr_v2(otsl_content) tokens, mixed_texts = otsl_extract_tokens_and_text(otsl_content) table_cells, split_row_tokens = otsl_parse_texts(mixed_texts, tokens) table_data = TableData( num_rows=len(split_row_tokens), num_cols=(max(len(row) for row in split_row_tokens) if split_row_tokens else 0), table_cells=table_cells, ) return export_to_html(table_data) def _collapse_placeholder_with_ignorable_tail( otsl_content: str, placeholders: List[str] ) -> str: stripped = otsl_content.strip() for placeholder in placeholders: if stripped == placeholder: return placeholder if stripped.startswith(placeholder): tail = stripped[len(placeholder) :].strip() if tail and IGNORABLE_OTSL_TAIL_PATTERN.fullmatch(tail): return placeholder return otsl_content _OTSL_STRUCTURAL_TAG_RE = re.compile(r"<(?:otsl|fcel|ecel|lcel|ucel|xcel)>") def _flatten_top_level_table_placeholders( otsl_content: str, placeholder_to_html: Dict[str, str] ) -> str | None: """If the top level is one or more table placeholders plus only non-OTSL text (sibling tables and/or a the model-appended caption/note after ````), return the concatenated HTML of those tables in order of appearance; otherwise ``None``. Without this, a placeholder-only top-level string has no OTSL cell tags, so ``convert_otsl_to_html`` yields an empty table and every table — sibling tables and all cell line breaks included — is silently dropped (issue #176). A genuinely nested table keeps structural OTSL tags at the top level, so the remainder check returns ``None`` and it falls through to the normal render path unchanged. """ present = [p for p in placeholder_to_html if p in otsl_content] if not present: return None remainder = otsl_content for placeholder in present: remainder = remainder.replace(placeholder, "") if _OTSL_STRUCTURAL_TAG_RE.search(remainder): return None ordered = sorted(present, key=otsl_content.index) return "".join(placeholder_to_html[p] for p in ordered) def convert_otsl_to_html_v2(otsl_content: str, debug: bool = False) -> str: """ Convert OTSL-v1.0 string to HTML with support for nested tables. Recursively processes nested ... blocks from innermost to outermost. Args: otsl_content (str): OTSL string, possibly with nested blocks. debug (bool): Enable debug output. Returns: str: HTML table with nested tables properly rendered. """ import uuid if debug: logger.debug("OTSL conversion started", input_preview=otsl_content[:200]) # Bug G guard: a genuinely nested table needs >= 2 opens. When # there is at most one open but more closes than opens, the # input is a single (malformed) table with a stray/premature # (the model sometimes emits this). Running the non-greedy nested regex # on it splits the table at the first and silently drops the # rest. Strip every otsl tag and convert it flat instead. Well-formed # single (1 open / 1 close) and genuine nested (>= 2 opens) are NOT # affected — they fall through to the unchanged path below. n_open = otsl_content.count("") if n_open <= 1 and otsl_content.count("") > n_open: flat = otsl_content.replace("", "").replace("", "") try: return convert_otsl_to_html(flat) except Exception as e: # noqa: BLE001 if debug: logger.error("Bug-G flat conversion failed", error=str(e)) return "
" # 1단계: 모든 중첩 테이블을 placeholder로 치환하고 OTSL 저장 # placeholder -> OTSL 매핑 (아직 HTML 아님!) placeholder_to_otsl: Dict[str, str] = {} # 가장 안쪽 ... 블록을 찾는 정규식 nested_pattern = r"((?:(?!).)*?)" iteration = 0 max_iterations = 100 while iteration < max_iterations: match = re.search(nested_pattern, otsl_content, re.DOTALL) if not match: break inner_otsl = match.group(1) # 태그 제외한 OTSL 내용 if debug: logger.debug( "Found nested OTSL", iteration=iteration, nested_preview=inner_otsl[:100], ) # placeholder 생성하고 OTSL 저장 (HTML 변환은 나중에!) placeholder = f"__NESTED_TABLE_{uuid.uuid4().hex}__" placeholder_to_otsl[placeholder] = inner_otsl # 원본에서 해당 블록을 placeholder로 치환 otsl_content = ( otsl_content[: match.start()] + placeholder + otsl_content[match.end() :] ) iteration += 1 if debug: logger.debug("After all nested replacement", content_preview=otsl_content[:200]) logger.debug("Total nested levels found", total_levels=len(placeholder_to_otsl)) # 최상위 태그 제거 otsl_content = re.sub(r"^|$", "", otsl_content.strip()) otsl_content = _collapse_placeholder_with_ignorable_tail( otsl_content, list(placeholder_to_otsl) ) # 2단계: 모든 레벨의 OTSL을 HTML로 변환 (placeholder 포함된 채로) placeholder_to_html: Dict[str, str] = {} for placeholder, otsl in placeholder_to_otsl.items(): try: inner_html = convert_otsl_to_html(otsl) placeholder_to_html[placeholder] = inner_html if debug: logger.debug( f"Converted {placeholder[:30]}... to HTML: {inner_html[:80]}..." ) except Exception as e: placeholder_to_html[placeholder] = "
" if debug: logger.error( "OTSL conversion failed", placeholder=placeholder, error=str(e) ) # 최상위 레벨도 HTML로 변환 top_stripped = otsl_content.strip() flat_html = _flatten_top_level_table_placeholders(top_stripped, placeholder_to_html) if top_stripped in placeholder_to_otsl: # 전체가 하나의 중첩 테이블인 경우 html_result = placeholder_to_html.get(top_stripped, "
") elif flat_html is not None: # placeholder(들) + (표 아닌) 잡텍스트: 형제 표를 통째로 버리지 말고 # 등장 순서대로 이어붙인다. the model full-page 출력은 표를 여러 개 쌓거나 # 표 뒤에 비고/캡션을 덧붙인다 (이슈 #176). html_result = flat_html else: try: html_result = convert_otsl_to_html(otsl_content) if debug: logger.debug("Top-level HTML generated", html_preview=html_result[:150]) except Exception as e: if debug: logger.error("Top-level conversion failed", error=str(e)) return "
" # 3단계: 모든 placeholder를 실제 HTML로 치환 (가장 안쪽부터) # 여러 번 반복하여 모든 레벨의 placeholder를 치환 max_replace_iterations = 10 for _ in range(max_replace_iterations): replaced = False for placeholder, inner_html in placeholder_to_html.items(): escaped_placeholder = html.escape(placeholder) if placeholder in html_result or escaped_placeholder in html_result: html_result = html_result.replace(placeholder, inner_html) html_result = html_result.replace(escaped_placeholder, inner_html) replaced = True if not replaced: break if debug: logger.debug("OTSL conversion completed", result_preview=html_result[:200]) return str(html_result) if __name__ == "__main__": # 기존 함수 테스트 (중첩 테이블 없는 경우) simple_otsl = "ABCD" print("=== Simple table (v1) ===") print(convert_otsl_to_html(simple_otsl)) print() # v2 함수 테스트 (중첩 테이블 있는 경우) nested_otsl = "연 번품명규격신고인증정국(인)선물
가액사진품목1(작성
예시)󰄫다기24cm×26cm❙김동연
(도지사)중국
(천리)추정가
(약 15만원
또는
가액 추정 불가장관항 목200세외수입210경상적 세외수입220임시적 세외수입221재산매각수입222자치단체간 부담금223보조금 반환수입224기타수입225지난년도 수입230지방행정제재부과금등240지난년도 수입241지난년도 수입241-01지난년도 수입" print("=== Nested table (v2) ===") print(convert_otsl_to_html_v2(nested_otsl)) print() # 간단한 중첩 테이블 테스트 simple_nested = "Outer AInner 1Inner 2Outer BOuter C" print("=== Simple nested (v2) ===") print(convert_otsl_to_html_v2(simple_nested, debug=True)) # ========================================================================== # [vendored] truncate_repeat # ========================================================================== import re import unicodedata from collections import Counter from typing import Tuple, Union def find_shortest_repeating_substring(s: str) -> Union[str, None]: """ Find the shortest substring that repeats to form the entire string. Args: s (str): Input string. Returns: str or None: Shortest repeating substring, or None if not found. """ n = len(s) for i in range(1, n // 2 + 1): if n % i == 0: substring = s[:i] if substring * (n // i) == s: return substring return None def find_repeating_suffix( s: str, min_len: int = 8, min_repeats: int = 5 ) -> Union[Tuple[str, str, int], None]: """ Detect if string ends with a repeating phrase. Args: s (str): Input string. min_len (int): Minimum length of unit. min_repeats (int): Minimum repeat count. Returns: Tuple[str, str, int] or None: (prefix, unit, count) if found, else None. """ for i in range(len(s) // (min_repeats), min_len - 1, -1): unit = s[-i:] if s.endswith(unit * min_repeats): count = 0 temp_s = s while temp_s.endswith(unit): temp_s = temp_s[:-i] count += 1 start_index = len(s) - (count * i) return s[:start_index], unit, count return None def remove_ngram_repeats(text: str, ngram_size: int = 2, min_repeats: int = 3) -> str: """ Remove intermediate n-gram repetitions in text. Args: text: Input text. ngram_size: Size of n-gram to detect (default: 2 for bigram). min_repeats: Minimum number of repeats to trigger removal. Returns: Text with intermediate n-gram repetitions removed. Example: >>> remove_ngram_repeats("재직한 기간에 재직한 기간에 재직한 기간에") "재직한 기간에" """ if not text: return text words = text.split() if len(words) < ngram_size * min_repeats: return text # n-gram 패턴 추출 및 반복 탐지 i = 0 result_words = [] while i < len(words): # 남은 단어가 ngram_size보다 적으면 그대로 추가 if i + ngram_size > len(words): result_words.extend(words[i:]) break # 현재 n-gram current_ngram = tuple(words[i : i + ngram_size]) # 앞으로 이 n-gram이 얼마나 반복되는지 확인 repeat_count = 1 j = i + ngram_size while j + ngram_size <= len(words): next_ngram = tuple(words[j : j + ngram_size]) if next_ngram == current_ngram: repeat_count += 1 j += ngram_size else: break # 반복이 최소 횟수 이상이면 첫 번째만 남기고 건너뜀 if repeat_count >= min_repeats: result_words.extend(words[i : i + ngram_size]) # 첫 번째만 추가 i = j # 반복된 부분 모두 건너뜀 else: result_words.append(words[i]) # 한 단어씩 추가 i += 1 return " ".join(result_words) def truncate_repetitive_content( content: str, line_threshold: int = 10, char_threshold: int = 10, min_len: int = 10, preserve_line_breaks: bool = False, ) -> str: """ Detect and truncate character-level, phrase-level, or line-level repetition in content. Args: content (str): Input text. line_threshold (int): Min lines for line-level truncation. char_threshold (int): Min repeats for char-level truncation. min_len (int): Min length for char-level check. preserve_line_breaks (bool): If True, preserve line breaks (for markdown tables). Returns: Union[str, str]: (truncated_content, info_string) """ # 줄바꿈 유지 모드: 라인 단위로 처리 if preserve_line_breaks: lines = content.split("\n") processed_lines = [] for line in lines: # 빈 줄은 그대로 유지 if not line.strip(): processed_lines.append(line) else: # 각 라인을 개별적으로 처리 (줄바꿈 유지 안 함) processed = truncate_repetitive_content( line, line_threshold, char_threshold, min_len, preserve_line_breaks=False, ) processed_lines.append(processed) return "\n".join(processed_lines) stripped_content = content.strip() # Priority 0: 1-gram repetition (single word repeats like "안녕 안녕 안녕") if " " in stripped_content and len(stripped_content.split()) >= 4: # 같은 단어가 4번 이상 반복되면 첫 번째만 남김 words = stripped_content.split() if len(set(words)) == 1: # 모든 단어가 같음 content = words[0] stripped_content = content.strip() # Priority 0.5: N-gram repetition (intermediate phrase repeats like "재직한 기간에 재직한 기간에") # 여러 n-gram 크기를 시도 (가장 긴 패턴부터) if " " in stripped_content and len(stripped_content.split()) >= 6: def _dedup_ngrams(line: str) -> str: # 3-gram 우선, 그 다음 2-gram을 시도해 더 많이 제거된(더 짧은) 결과 선택 line_3gram = remove_ngram_repeats(line, ngram_size=3, min_repeats=3) line_2gram = remove_ngram_repeats(line, ngram_size=2, min_repeats=3) candidate = line_3gram if len(line_3gram) < len(line_2gram) else line_2gram # remove_ngram_repeats는 반복이 없어도 항상 공백을 평탄화하므로, # 반복이 실제로 제거된 경우(평탄화된 원본보다 짧을 때)에만 교체한다. # 그렇지 않으면 원본 줄을 그대로 유지해 줄바꿈/공백 손실을 막는다 (이슈 #171). normalized = " ".join(line.split()) return candidate if len(candidate) < len(normalized) else line # 줄 단위로 처리해 줄바꿈 구조를 보존한다 (n-gram 반복은 줄 내부 구문 루프가 대상). content = "\n".join( _dedup_ngrams(line) if line.strip() else line for line in content.split("\n") ) stripped_content = content.strip() # Priority 0: Remove long sequences of separator/symbol characters # Uses Unicode category to detect punctuation (P), space (Zs), and symbol (S) characters # This handles dots, underscores, dashes, box drawings, waves, etc. elegantly def is_separator_symbol(char: str) -> bool: """Check if character is a separator/symbol that should be removed when repeated.""" cat = unicodedata.category(char) # P*: Punctuation, Zs: Space separator, S*: All symbols (Sm, Sc, Sk, So) return cat.startswith("P") or cat == "Zs" or cat.startswith("S") # Find and remove repeated separator/symbol characters (5+ repeats) match = re.search(r"(.)\1{4,}", stripped_content) if match and is_separator_symbol(match.group(1)): repeated_char = match.group(1) content = re.sub(re.escape(repeated_char) + r"{5,}", "", content) stripped_content = content.strip() if not stripped_content: return content # Priority 1: Phrase-level suffix repetition in long single lines. if "\n" not in stripped_content and len(stripped_content) > 100: suffix_match = find_repeating_suffix(stripped_content, min_len=8, min_repeats=5) if suffix_match: prefix, repeating_unit, count = suffix_match if len(repeating_unit) * count > len(stripped_content) * 0.5: return prefix # Priority 2: Full-string character-level repetition (e.g., 'ababab') if "\n" not in stripped_content and len(stripped_content) > min_len: repeating_substring = find_shortest_repeating_substring(stripped_content) if repeating_substring: count = len(stripped_content) // len(repeating_substring) if count >= char_threshold: return repeating_substring # Priority 3: Line-level repetition (e.g., same line repeated many times) lines = [line.strip() for line in content.split("\n") if line.strip()] if not lines: return content total_lines = len(lines) if total_lines < line_threshold: return content line_counts = Counter(lines) most_common_line, count = line_counts.most_common(1)[0] if count >= line_threshold and (count / total_lines) >= 0.8: return most_common_line return content # ========================================================================== # [vendored] picture_response_validator # ========================================================================== """ Picture Response Validator VLM에서 반환된 picture/chart/flowchart 응답의 유효성을 검증합니다. """ import re from decimal import Decimal # [vendor-strip] from loguru import logger _SEPARATOR_PATTERN = re.compile(r"^\s*\|(?:\s*:?-+:?\s*\|)+\s*$") _KRW_EXPR_PATTERN = re.compile( r"(?P(?:\d[\d,]*\s*(?:조원|억원|만원|조|억|만|원)\s*)+)" ) _KRW_SEGMENT_PATTERN = re.compile(r"(\d[\d,]*)\s*(조원|억원|만원|조|억|만|원)") _KRW_UNIT_MULTIPLIERS = { "조원": 10**12, "억원": 10**8, "만원": 10**4, "조": 10**12, "억": 10**8, "만": 10**4, "원": 1, } _YEAR_HEADER_PATTERN = re.compile(r"^(year|년도|연도|年)$", re.IGNORECASE) _AMOUNT_HEADER_PATTERN = re.compile( r"(amount|value|금액|예산|budget|won|krw|억원|억\s*원|億ウォン|兆ウォン|trillion|billion|million)", re.IGNORECASE, ) _PERCENT_PATTERN = re.compile(r"^\d[\d,]*(?:\.\d+)?%$") _PLAIN_NUMBER_PATTERN = re.compile(r"^\d[\d,]*(?:\.\d+)?$") _KRW_VALUE_PATTERN = re.compile(r"^KRW [\d,]+$") def split_markdown_table_row(line: str) -> list[str]: """Split a Markdown table row whether the trailing pipe is present or not.""" row = line.strip() if row.startswith("|"): row = row[1:] if row.endswith("|"): row = row[:-1] return [cell.strip() for cell in row.split("|")] def normalize_markdown_table_content(content: str) -> str: """ 차트/표 응답에서 흔히 발생하는 Markdown table 포맷 문제를 정규화합니다. 정규화 항목: 1. 전각 파이프(｜) -> ASCII 파이프(|) 2. 헤더 구분선이 없는 경우 자동 삽입 """ if not content or not isinstance(content, str): return content normalized = content.replace("｜", "|") lines = normalized.splitlines() table_line_indices = [ idx for idx, line in enumerate(lines) if line.strip().startswith("|") ] if len(table_line_indices) < 2: return normalized first_idx = table_line_indices[0] header_line = lines[first_idx].strip() if first_idx + 1 < len(lines) and _SEPARATOR_PATTERN.match( lines[first_idx + 1].strip() ): return normalized header_cells = split_markdown_table_row(header_line) if not header_cells: return normalized separator_line = "|" + "|".join([" --- "] * len(header_cells)) + "|" lines.insert(first_idx + 1, separator_line) return "\n".join(lines) def normalize_krw_expressions(content: str) -> str: """ 한국식 금액 표기(조/억/만/원)를 full numeric KRW 형식으로 정규화합니다. 예: - 34조 7,398억원 -> KRW 34,739,800,000,000 - 26,656억원 -> KRW 2,665,600,000,000 """ if not content or not isinstance(content, str): return content def _replace(match: re.Match[str]) -> str: expr = match.group("expr") total = 0 found = False for raw_number, unit in _KRW_SEGMENT_PATTERN.findall(expr): found = True total += int(raw_number.replace(",", "")) * _KRW_UNIT_MULTIPLIERS[unit] if not found: return expr return f"KRW {total:,}" return _KRW_EXPR_PATTERN.sub(_replace, content) def normalize_chart_table_currency_values(content: str) -> str: """ 차트 마크다운 테이블에서 금액 컬럼을 full numeric KRW 형식으로 정규화합니다. 전제: - 비한국어 chart 출력용 후처리 - 헤더에 금액 단위 힌트가 있거나, 첫 컬럼이 연도이고 나머지 컬럼이 숫자 중심인 경우 적용 """ if not content or not isinstance(content, str): return content normalized = normalize_markdown_table_content(normalize_krw_expressions(content)) lines = normalized.splitlines() table_lines = [line for line in lines if line.strip().startswith("|")] if len(table_lines) < 3: return normalized rows = [] for line in table_lines: cells = split_markdown_table_row(line) rows.append(cells) header = rows[0] body_rows = rows[2:] if not header or not body_rows: return normalized default_multiplier = 10**8 column_multipliers: dict[int, int] = {} convert_columns: set[int] = set() for idx, header_cell in enumerate(header): if _AMOUNT_HEADER_PATTERN.search(header_cell): convert_columns.add(idx) # 배수를 컬럼별로 그 컬럼 헤더 단위에서 도출한다. 단일 변수를 덮어쓰면 # 마지막 헤더 단위가 모든 컬럼을 지배해 혼합 단위 표가 오염된다 (이슈 #175). if re.search(r"(trillion|조|兆)", header_cell, re.IGNORECASE): column_multipliers[idx] = 10**12 elif re.search(r"(billion|억|億)", header_cell, re.IGNORECASE): column_multipliers[idx] = 10**8 elif re.search(r"(million|백만)", header_cell, re.IGNORECASE): column_multipliers[idx] = 10**6 else: column_multipliers[idx] = default_multiplier if not convert_columns and header and _YEAR_HEADER_PATTERN.search(header[0]): numeric_columns = set(range(1, len(header))) for row in body_rows: for idx in list(numeric_columns): if idx >= len(row): numeric_columns.discard(idx) continue cell = row[idx] if cell in ("", "-"): continue if cell.startswith("KRW "): continue if not _PLAIN_NUMBER_PATTERN.fullmatch(cell): numeric_columns.discard(idx) convert_columns = numeric_columns if not convert_columns: return normalized for row in body_rows: for idx in convert_columns: if idx >= len(row): continue cell = normalize_krw_expressions(row[idx]).strip() if not cell or cell == "-": row[idx] = cell continue if cell.startswith("KRW ") or _PERCENT_PATTERN.fullmatch(cell): row[idx] = cell continue if _PLAIN_NUMBER_PATTERN.fullmatch(cell): multiplier = column_multipliers.get(idx, default_multiplier) amount = int(Decimal(cell.replace(",", "")) * multiplier) row[idx] = f"KRW {amount:,}" else: row[idx] = cell rebuilt_lines = [] for line_idx, cells in enumerate(rows): if line_idx == 1: rebuilt_lines.append("| " + " | ".join(["---"] * len(header)) + " |") else: rebuilt_lines.append("| " + " | ".join(cells) + " |") rebuilt_iter = iter(rebuilt_lines) output_lines = [] for line in lines: if line.strip().startswith("|"): output_lines.append(next(rebuilt_iter)) else: output_lines.append(line) return "\n".join(output_lines) def merge_translated_table_with_source_values( translated_content: str, source_content: str, ) -> str: """ 번역된 chart table의 라벨은 유지하고, 값 셀은 원본 정규화 값으로 덮어씁니다. 목적: - 비한국어 번역 시에도 KRW full numeric / 퍼센트 / 숫자 정밀도를 유지 """ translated = normalize_markdown_table_content(translated_content) source = normalize_chart_table_currency_values(source_content) translated_lines = [ line for line in translated.splitlines() if line.strip().startswith("|") ] source_lines = [ line for line in source.splitlines() if line.strip().startswith("|") ] if len(translated_lines) < 3 or len(translated_lines) != len(source_lines): return translated def _split(line: str) -> list[str]: return split_markdown_table_row(line) translated_rows = [_split(line) for line in translated_lines] source_rows = [_split(line) for line in source_lines] if not translated_rows or len(translated_rows[0]) != len(source_rows[0]): return translated merged_rows: list[list[str]] = [] for row_idx, (translated_row, source_row) in enumerate( zip(translated_rows, source_rows) ): if row_idx == 1: merged_rows.append(["---"] * len(translated_rows[0])) continue merged_row = [] for translated_cell, source_cell in zip(translated_row, source_row): if ( _KRW_VALUE_PATTERN.fullmatch(source_cell) or _PERCENT_PATTERN.fullmatch(source_cell) or _PLAIN_NUMBER_PATTERN.fullmatch(source_cell) ): merged_row.append(source_cell) else: merged_row.append(translated_cell) merged_rows.append(merged_row) merged_lines = ["| " + " | ".join(row) + " |" for row in merged_rows] merged_iter = iter(merged_lines) output_lines = [] for line in translated.splitlines(): if line.strip().startswith("|"): output_lines.append(next(merged_iter)) else: output_lines.append(line) return "\n".join(output_lines) def is_valid_markdown_table(content: str) -> bool: """ 마크다운 테이블이 유효한지 검증합니다. 검증 항목: 1. 너무 많은 연속된 빈 셀 (| | | | 패턴) 2. 테이블 구조의 무결성 Args: content: VLM에서 반환된 마크다운 테이블 문자열 Returns: bool: 유효하면 True, 깨진 테이블이면 False """ # 빈 내용이나 None은 통과 (chart/flow가 아닌 경우) if not content or not isinstance(content, str): return True content = normalize_markdown_table_content(content) # 빈 내용은 통과 if not content.strip(): return True # 마크다운 테이블이 아닌 경우 통과 (일반 텍스트 등) if not content.strip().startswith("|"): return True lines = content.strip().split("\n") table_lines = [line for line in lines if line.strip().startswith("|")] if not table_lines: return True # 1. 연속된 빈 셀 패턴 검사 for line in table_lines: # 구분자 라인은 건너뜀 (|---|---|) if _SEPARATOR_PATTERN.search(line): continue stripped_line = line.strip() # 연속된 빈 셀 패턴 (| | | |)을 직접 검사 # 빈 셀이 5개 이상 연속되면 깨진 테이블로 간주 # 정상 테이블은 "| A | B |" 형태라 빈 셀이 2개 연속은 있을 수 있음 if re.search(r"(\|\s+){5,}", stripped_line): # 실제로 빈 셀인지 확인 (공백만 있는지) consecutive_empty = re.findall(r"\|\s+(?=\|)", stripped_line) if len(consecutive_empty) >= 5: logger.warning( f"Detected corrupted table with {len(consecutive_empty)} consecutive empty cells: {line[:100]}..." ) return False return True # --------------------------------------------------------------------------- # Raw the model chart-token parser (chart pipeline fix) # # The the model base model, when prompted with "\nImage Analysis:" for a chart # region, emits its native classified format WITHOUT a JSON envelope, e.g.: # # <|class_start|>chart<|class_end|> # <|sub_class_start|>Bar Chart<|sub_class_end|> # <|caption_start|> ... <|caption_end|> # <|content_start|>| Country | Value |\n|---|---|\n| Netherlands | 215 |<|content_end|> # # These responses are neither valid JSON nor start with "|", so # parse_classified_picture_output() returns None and the plain-"|" fallback in # apply_picture_result() also misses them, leaving the raw token-wrapped block # to leak into the assembled markdown (sometimes HTML-escaped inside

..

, # or jammed into an image alt-text "![...]") where ChartDataPointRule's pandas # table parser cannot read it. This helper extracts (image_type, table) so the # chart table reaches the markdown as a clean, parseable table. # --------------------------------------------------------------------------- import html as _html _NATIVE_CLASS_PATTERN = re.compile( r"<\|class_start\|>\s*(?P[^<|]*?)\s*<\|class_end\|>", re.S ) _NATIVE_CONTENT_PATTERN = re.compile( r"<\|content_start\|>(?P.*?)<\|content_end\|>", re.S ) _NATIVE_TOKEN_PATTERN = re.compile(r"<\|[^|>]*\|>") def parse_native_chart_tokens(content): """Parse a raw the model classified picture/chart response. Returns (image_type, table_text) when the response uses the the model <|...|> token format, else None (so callers fall through to their existing JSON / plain-markdown handling). table_text is the markdown table extracted from the <|content_start|>..<|content_end|> span with all special tokens stripped; image_type is the <|class_start|>..<|class_end|> class ("chart"/"flow"/"picture"), defaulting to "picture". """ if not content or not isinstance(content, str): return None # Unescape HTML entities and drop wrapping image/paragraph syntax so the # token markers are visible (e.g. "![<|class_start|>..." or "

<|.."). text = _html.unescape(content) if "<|content_start|>" not in text and "<|class_start|>" not in text: return None cls_match = _NATIVE_CLASS_PATTERN.search(text) raw_cls = cls_match.group("cls").strip().lower() if cls_match else "" if raw_cls.startswith("chart"): image_type = "chart" elif raw_cls.startswith("flow"): image_type = "flow" else: image_type = "picture" body_match = _NATIVE_CONTENT_PATTERN.search(text) if body_match: table = body_match.group("body") else: # No explicit content span: strip tokens and keep whatever remains. table = text table = _NATIVE_TOKEN_PATTERN.sub("", table) # Drop residual image/paragraph wrapper artifacts. table = table.replace("

", "").replace("

", "") table = re.sub(r"^!\[", "", table.strip()).rstrip("]").strip() return image_type, table # ========================================================================== # [vendored] table_processor(functions) # ========================================================================== """ TableElementPostProcessor 테이블 요소의 OTSL 태그를 HTML로 변환합니다. """ import html import json import re from typing import Match # [vendor-strip] from loguru import logger # [vendor-strip] from ..schemas.element_schema import DocumentElement, ElementCategory # [vendor-strip] from ..utils.otsl_converter import convert_otsl_to_html_v2 # [vendor-strip] from ..utils.truncate_repeat import truncate_repetitive_content # [vendor-strip] from .base import ElementPostProcessor def looks_like_otsl(content: str) -> bool: """Return True when table content starts with wrapped or bare OTSL tokens.""" return bool( content.lstrip().startswith( ("", "", "", "", "", "", "") ) ) def looks_like_html_table(content: str) -> bool: """Return True when content appears to contain an HTML table.""" stripped = content.strip() if stripped.startswith("\"" or "". lowered = stripped.lower() return " str: """rowspan/colspan 속성값의 따옴표를 제거해 출력 형식을 통일한다.""" return re.sub( r'\b(rowspan|colspan)\s*=\s*(?:"|\')?(\d+)(?:"|\')?', r"\1=\2", content, flags=re.IGNORECASE, ) def normalize_html_table_content(content: str) -> str: """Normalize escaped quotes and wrappers in HTML table content. NOTE: provider가 직접 반환한 HTML(escaped quote, JSON 문자열 래핑, code fence 등) 정규화용이다. ``html.unescape``를 포함하므로, 이미 ``html.escape`` 된 출력 (예: ``convert_otsl_to_html_v2`` 결과)에는 적용하면 안 된다. 그 경로에서는 ``normalize_span_attributes``만 사용한다 (이슈 #172). """ normalized = content.strip() # Some providers return HTML table content as a JSON string literal. # Unwrap it repeatedly while it still looks like a serialized string. for _ in range(3): if len(normalized) < 2 or normalized[0] != '"' or normalized[-1] != '"': break try: parsed = json.loads(normalized) except json.JSONDecodeError: break if not isinstance(parsed, str): break normalized = parsed.strip() normalized = html.unescape(normalized) normalized = normalized.replace("\\u0022", '"').replace("\\n", "") normalized = re.sub(r'\\+(")', '"', normalized) normalized = re.sub(r"\\+(/)", r"\1", normalized) normalized = re.sub( r"^\s*```(?:html)?\s*(.*?)\s*```\s*$", r"\1", normalized, flags=re.IGNORECASE | re.DOTALL, ) stripped = normalized.strip() if len(stripped) >= 2 and stripped[0] == stripped[-1] == '"': inner = stripped[1:-1].strip() if inner.startswith(" str: """ HTML 테이블의 " # 패턴을 찾아서 처리 (태그 속성도 고려) result = re.sub( r"]*)>(.*?)", clean_dots_in_td, html_content, flags=re.DOTALL ) return result # ========================================================================== # [vendored] chart_processor(functions) # ========================================================================== """ DataChartElementPostProcessor 차트 요소를 후처리합니다 (현재는 bypass). """ import html import re import markdown # [vendor-strip] from loguru import logger # [vendor-strip] from ..schemas.element_schema import DocumentElement, ElementCategory # [vendor-strip] from ..utils.picture_response_validator import normalize_markdown_table_content # [vendor-strip] from .base import ElementPostProcessor _SEPARATOR_ROW_PATTERN = re.compile(r"\|\s*:?-+:?\s*(?:\|\s*:?-+:?\s*)+\|") def _extract_pipe_row(segment: str, expected_pipe_count: int) -> tuple[str, str]: """Extract a single markdown row from a flattened pipe-table segment.""" stripped = segment.lstrip() if not stripped.startswith("|"): return "", segment pipe_count = 0 for index, char in enumerate(stripped): if char == "|": pipe_count += 1 if pipe_count == expected_pipe_count: row = stripped[: index + 1].strip() remaining = stripped[index + 1 :].lstrip() return row, remaining return stripped.strip(), "" def normalize_inline_markdown_table(content: str) -> str: """Split a single-line pipe table into markdown rows before HTML conversion.""" normalized = normalize_markdown_table_content(content) if "\n" in normalized or not normalized.strip().startswith("|"): return normalized separator_match = _SEPARATOR_ROW_PATTERN.search(normalized) if not separator_match: return normalized separator_row = separator_match.group(0).strip() expected_pipe_count = separator_row.count("|") if expected_pipe_count < 2: return normalized header_segment = normalized[: separator_match.start()].strip() data_segment = normalized[separator_match.end() :].strip() header_row, leftover_header = _extract_pipe_row(header_segment, expected_pipe_count) if not header_row or leftover_header.strip(): return normalized rows = [header_row, separator_row] remaining = data_segment while remaining: row, remaining = _extract_pipe_row(remaining, expected_pipe_count) if not row: break rows.append(row) return "\n".join(rows) if len(rows) >= 2 else normalized # ========================================================================== # [vendored] kdl_postprocess # ========================================================================== """Pure-rule markdown postprocessing for the ``kdl_frontier`` mode. These are GT-agnostic, structure-preserving text-surface rewrites applied to the fully-assembled document markdown right before the parse response is returned. They are the production port of the kdl-frontier-parser prototype's ``_apply_stack`` rules (validated on ParseBench: header-marking +0.0156 5D, title+quote +0.0132 5D). Three rules, applied in this exact order (matching the prototype): 1. header_mark - mark a table's auto-detected multi-level header region as ``
셀에서 과도한 점 패턴을 제거합니다. truncate_repeat.py의 로직을 참고하여 점 5개 이상 연속된 패턴을 제거합니다. ASCII dot(.) 또는 Unicode ellipsis(…, U+2026)를 처리합니다. Args: html_content: HTML 테이블 문자열 Returns: str: 점 패턴이 제거된 HTML 테이블 """ if not html_content or not isinstance(html_content, str): return html_content def clean_dots_in_td(match: Match[str]) -> str: attrs = match.group(1) # 태그 속성 td_content = match.group(2) # 셀 내용 original = td_content # 태그 내부의 텍스트에서 점 패턴 제거 # ASCII dot(.) 또는 Unicode ellipsis(…, U+2026)가 5개 이상 연속되면 제거 td_content = re.sub(r"(\.|\u2026|\u00B7){5,}", "", td_content) # 중간생략문자(···) 같은 패턴도 제거 (U+00B7은 middle dot) td_content = re.sub(r"(\u00B7){5,}", "", td_content) if td_content != original: logger.debug("Removed dot patterns from TD cell") return f"{td_content}...``. 2. quote_fold - fold curly quotes/apostrophes to ASCII (SemFormat literal-match fix). 3. title_promote- de-blockquote + promote standalone title-ish lines to ``# ``. All three are idempotent and never reorder / merge / split elements, so document STRUCTURE (and TEDS-Structure) is unchanged. Stdlib-only (no model, no I/O). """ # [vendor-strip] from __future__ import annotations import re from collections import Counter # -------------------------------------------------------------------------------------- # Rule 1: multi-level table-header marking (port of header_transform.transform_markdown) # -------------------------------------------------------------------------------------- _TABLE_RE = re.compile(r"]*>.*?
", re.I | re.S) _TR_RE = re.compile(r"(]*>.*?)", re.I | re.S) _COLSPAN_RE = re.compile(r"colspan\s*=\s*[\"']?([2-9]|\d\d)", re.I) def _row_has_colspan(tr: str) -> bool: return bool(_COLSPAN_RE.search(tr)) def _auto_header_n(table_html: str) -> int: """Number of leading rows that form the header: contiguous colspan>1 rows + 1 leaf row. Single-level tables (no colspan>1 in row 0) -> 0 (first-row marking was net-negative). """ trs = _TR_RE.findall(table_html) if not trs or not _row_has_colspan(trs[0]): return 0 n = 0 for tr in trs: if _row_has_colspan(tr): n += 1 else: break return min(n + 1, len(trs)) def _mark_n(table_html: str, n: int) -> str: out = table_html for tr in _TR_RE.findall(table_html)[:n]: m = re.sub(r"]*)?>", r"", tr, flags=re.I) m = re.sub(r"", "", m, flags=re.I) out = out.replace(tr, m, 1) return out def header_mark(md: str) -> str: """Mark auto-detected multi-level header rows of every block as
.""" if not md: return md def _repl(match: "re.Match[str]") -> str: table_html = match.group(0) n = _auto_header_n(table_html) if n >= 1: marked = _mark_n(table_html, n) if marked != table_html: return marked return table_html return _TABLE_RE.sub(_repl, md) # -------------------------------------------------------------------------------------- # Rule 2: curly-quote fold (port of wf_rule_4_transform.fold_md) # -------------------------------------------------------------------------------------- _QUOTE_FOLD = { "‘": "'", "’": "'", "‚": "'", "‛": "'", # single quotes "“": '"', "”": '"', "„": '"', "‟": '"', # double quotes } _QUOTE_TABLE = str.maketrans(_QUOTE_FOLD) def quote_fold(md: str) -> str: """Fold curly quotes/apostrophes to ASCII (regression-free; dash/space fold excluded).""" if not md: return md return md.translate(_QUOTE_TABLE) # -------------------------------------------------------------------------------------- # Rule 3: title promotion + de-blockquote (port of wf_rule_0_transform.transform_markdown) # -------------------------------------------------------------------------------------- _HEADING_RE = re.compile(r"^\s*#{1,6}\s+") _LIST_RE = re.compile(r"^\s*[-*+]\s+") _NUMLIST_RE = re.compile(r"^\s*\d+\.\s+") _TABLEROW_RE = re.compile(r"^\s*\|") _LABELVALUE_RE = re.compile(r"^.{1,40}:\s") _BLOCKQUOTE_RE = re.compile(r"^\s*>\s+(.+)$") _FENCE_RE = re.compile(r"^\s*(```|~~~)") _TERMINAL_PUNCT = tuple(".!?:;,") _LETTER_RE = re.compile(r"[A-Za-z]") _UPPER_RE = re.compile(r"[A-Z]") # Multi-page docs inject "---\n\n**Page N**" separators (response_builder._build_full_markdown); # never promote those to headings. _PAGE_MARKER_RE = re.compile(r"^\*\*Page\s+\d+\*\*$") # Shipped variant params (verbatim from the prototype). "aggressive" is the chosen one # (conservative + defensible vs ultra2; 5D delta vs ultra2 is ~ -0.0008). _TITLE_VARIANTS = { "aggressive": (12, 0.60, False), "strict_caps_bq": (8, 0.85, True), "deblockquote_only": (12, 0.60, False), "ultra": (22, 0.25, False), "ultra2": (30, 0.0, False), } def _is_titleish(text: str, max_words: int, caps_ratio: float, require_all_caps: bool) -> bool: s = text.strip() if not s: return False if _HEADING_RE.match(s) or _LIST_RE.match(s) or _NUMLIST_RE.match(s) or _TABLEROW_RE.match(s): return False if len(s.split()) > max_words: return False if s.endswith(_TERMINAL_PUNCT): return False if _LABELVALUE_RE.match(s): return False letters = _LETTER_RE.findall(s) if not letters: return False caps_frac = len(_UPPER_RE.findall(s)) / len(letters) if require_all_caps: return caps_frac >= caps_ratio first_alpha = next((c for c in s if c.isalpha()), "") return first_alpha.isupper() or caps_frac > caps_ratio def title_promote(md: str, variant: str = "aggressive") -> str: """De-blockquote and promote standalone title-ish lines to ``# `` headings.""" if not md: return md max_words, caps_ratio, require_all_caps = _TITLE_VARIANTS[variant] do_promote = variant != "deblockquote_only" lines = md.split("\n") n = len(lines) out = list(lines) in_fence = False for i, raw in enumerate(lines): if _FENCE_RE.match(raw): in_fence = not in_fence continue if in_fence: continue above = lines[i - 1] if i > 0 else "" below = lines[i + 1] if i + 1 < n else "" standalone = (i == 0 or above.strip() == "") and (i + 1 >= n or below.strip() == "") if not standalone: continue if _PAGE_MARKER_RE.match(raw.strip()): continue bq = _BLOCKQUOTE_RE.match(raw) if bq: inner = bq.group(1) if _is_titleish(inner, max_words, caps_ratio, require_all_caps): out[i] = ("# " + inner.strip()) if do_promote else inner.strip() continue if do_promote and _is_titleish(raw, max_words, caps_ratio, require_all_caps): out[i] = "# " + raw.strip() return "\n".join(out) # -------------------------------------------------------------------------------------- # Public entry point # -------------------------------------------------------------------------------------- # Guard: skip rewriting pathologically large/repetitive markdown (avoids catastrophic # regex cost on runaway model output). Generation already uses no_repeat_ngram_size, so # this should rarely trigger; when it does we return the markdown UNCHANGED (never drop # content -> structure stays intact). _MAX_MD_LEN = 2_000_000 def _looks_runaway(md: str) -> bool: if len(md) > _MAX_MD_LEN: return True lines = [ln.strip() for ln in md.splitlines() if ln.strip()] if len(lines) > 200: top = Counter(lines).most_common(1) if top and top[0][1] > 1000: return True return False def postprocess_markdown(md: str, *, title_variant: str = "aggressive") -> str: """Apply the kdl-frontier pure-rule stack to assembled document markdown. Order: header_mark -> quote_fold -> title_promote(variant). Each rule is wrapped so a failure in one never discards the document; on a runaway document all rules are skipped. Returns the rewritten markdown (or the input unchanged on guard/failure). """ if not md or _looks_runaway(md): return md try: md = header_mark(md) except Exception: # noqa: BLE001 - never fail the response on a postprocess rule pass try: md = quote_fold(md) md = title_promote(md, variant=title_variant) except Exception: # noqa: BLE001 pass return md # ========================================================================== # [core] KDL-Frontier-Parser-nano orchestration # ========================================================================== # Faithful port of the closed-source orchestrator's inference flow for the # submitted run. Stage prompts/params and every deterministic transform above # are vendored verbatim; this section wires them together: # render -> layout (1036x1036) -> crop/bucket -> per-category recognition # -> element post-processing -> markdown assembly -> rule post-processing. _NANO_PROMPTS = { # byte-exact stage prompts (leading newline included; formula has no # trailing newline) — these are the templates the run was measured with. "layout": "\nLayout Detection:\n", "text": "\nText Recognition:\n", "table": "\nTable Recognition:\n", "picture": "\nImage Analysis:\n", "formula": "\nFormula Recognition:", } _NANO_EXTRA_PAYLOAD = { # merged TOP-LEVEL into the chat/completions body (vLLM accepts these). "layout": { "skip_special_tokens": False, "top_p": 0.01, "top_k": 1, "presence_penalty": 0.0, "frequency_penalty": 0.0, "repetition_penalty": 1.0, "extra_args": {"no_repeat_ngram_size": 100}, }, "text": { "top_p": 0.01, "top_k": 1, "repetition_penalty": 1.0, "extra_args": {"no_repeat_ngram_size": 100}, }, "table": { "skip_special_tokens": False, "top_p": 0.01, "top_k": 1, "presence_penalty": 1.0, "frequency_penalty": 0.005, "repetition_penalty": 1.0, "extra_args": {"no_repeat_ngram_size": 100}, }, "picture": { "top_p": 0.01, "top_k": 1, "repetition_penalty": 1.0, "extra_args": {"no_repeat_ngram_size": 100}, }, "formula": None, # formula stage sends temperature+max_tokens only } _NANO_MAX_TOKENS_ENV = { # submitted-run budgets (layout/table must fit --max-model-len 8192, # otherwise long pages return HTTP 400 -> empty layout) "layout": ("KDL_NANO_LAYOUT_MAX_TOKENS", 6000), "text": ("KDL_NANO_TEXT_MAX_TOKENS", 2048), "table": ("KDL_NANO_TABLE_MAX_TOKENS", 5500), "picture": ("KDL_NANO_PICTURE_MAX_TOKENS", 4096), "formula": ("KDL_NANO_FORMULA_MAX_TOKENS", 128), } # table crops are encoded lossless PNG (JPEG q95 collapses multi-line cells); # every other stage uses JPEG quality=95. _NANO_LOSSLESS_STAGES = {"table"} _TEXT_BUCKET_CATEGORIES = { # TextElementPostProcessor.TEXT_CATEGORIES (n-gram repetition cleanup) "Title", "Section-header", "Text", "Page-header", "Page-footer", "List-item", "Caption", "Footnote", } _BLOCKQUOTE_CATEGORIES = {"Caption", "Footnote", "Page-header", "Page-footer"} def _nano_image_to_data_uri(image: Image.Image, *, lossless: bool = False) -> str: """Port of message_builder.image_to_bytes + base64 data-URI encoding.""" buffered = io.BytesIO() if ( lossless or image.mode in ("RGBA", "LA") or (image.mode == "P" and "transparency" in image.info) ): mime_type = "image/png" image.save(buffered, format="PNG") else: if image.mode not in ("L", "RGB", "CMYK"): image = image.convert("RGB") mime_type = "image/jpeg" image.save(buffered, format="JPEG", quality=95) b64 = base64.b64encode(buffered.getvalue()).decode("ascii") return f"data:{mime_type};base64,{b64}" def _nano_payload(stage: str, model: str, image: Image.Image) -> dict: payload = { "model": model, "messages": [ { "role": "user", "content": [ { "type": "image_url", "image_url": { "url": _nano_image_to_data_uri( image, lossless=stage in _NANO_LOSSLESS_STAGES ) }, }, {"type": "text", "text": _NANO_PROMPTS[stage]}, ], } ], "temperature": 0.0, "max_tokens": int( os.getenv(_NANO_MAX_TOKENS_ENV[stage][0], str(_NANO_MAX_TOKENS_ENV[stage][1])) ), } extra = _NANO_EXTRA_PAYLOAD[stage] if extra is not None: payload.update(extra) return payload def _nano_group_by_bucket( content: List[Dict[str, Any]], original_image: Image.Image ) -> Dict[str, List[Dict[str, Any]]]: """Verbatim port of result_parser.group_content_by_category.""" result: Dict[str, List[Dict[str, Any]]] = { "text": [], "table": [], "picture": [], "formula": [], } im_w, im_h = original_image.size for item in content: cat = item.get("category", "Text") mapped_cat = layout_recognition_bucket(cat) bbox = item.get("bbox", []) if not bbox or len(bbox) != 4: continue try: x1, y1, x2, y2 = bbox left = int(round(x1 * im_w)) upper = int(round(y1 * im_h)) right = int(round(x2 * im_w)) lower = int(round(y2 * im_h)) left = max(0, min(left, im_w)) upper = max(0, min(upper, im_h)) right = max(left + 1, min(right, im_w)) lower = max(upper + 1, min(lower, im_h)) pixel_width = right - left pixel_height = lower - upper if pixel_width < 5 or pixel_height < 5: continue cropped_img = original_image.crop((left, upper, right, lower)) preprocessed_img = preprocess_for_vlm(cropped_img) if is_monochromatic(preprocessed_img): continue except Exception: continue element_info = { "bbox": bbox, "category": cat, "layout_order": item.get("layout_order", 0), "page_number": item.get("page_number", 1), "preprocessed_image": preprocessed_img, } if "angle" in item: element_info["angle"] = item.get("angle") if mapped_cat == "picture": element_info["cropped_image"] = cropped_img result[mapped_cat].append(element_info) return result def _nano_is_single_clean_otsl(content: Any) -> bool: return ( isinstance(content, str) and content.count("") == 1 and ("" in content or "" in content) ) def _nano_apply_picture_result(el: Dict[str, Any], content: str | None) -> None: """Port of picture_recognition.apply_picture_result for the single-model run (enable_chart=True, caption_language='en'). The classified-JSON branch of the original is unreachable here: this preset has no response_format, so the model answers '\\nImage Analysis:' either with native chart tokens or plain text — never schema JSON. Hangul caption translation (a recognition-time convenience for Korean documents) is likewise inert on this benchmark and not ported. """ if content is None: el["content"] = "" return content_str = content.strip() native_parsed = parse_native_chart_tokens(content_str) if native_parsed is not None: image_type, table = native_parsed if table: # language_code 'en' != 'ko' table = normalize_krw_expressions(table) if image_type == "chart": el["category"] = "Chart" elif image_type == "flow": el["category"] = "Flowchart" normalized = normalize_markdown_table_content(table) if is_valid_markdown_table(normalized) and normalized.strip().startswith("|"): el["content"] = normalized else: el["content"] = table return if content_str.startswith("|") and is_valid_markdown_table(content_str): el["content"] = normalize_markdown_table_content(content_str) el["category"] = "Chart" else: el["content"] = content_str def _nano_postprocess_element(el: Dict[str, Any]) -> None: """Port of postprocessing.PostProcessor (first matching processor wins; failures keep the original content).""" content = el.get("content") or "" category = el.get("category", "Text") if not content.strip(): return try: if category == "Table": if looks_like_html_table(content): el["content"] = normalize_html_table_content(content) elif looks_like_otsl(content): cleaned = truncate_repetitive_content(content, preserve_line_breaks=True) html_content = convert_otsl_to_html_v2(cleaned) html_content = normalize_span_attributes(html_content) el["content"] = remove_dots_from_html_cells(html_content) elif category in _TEXT_BUCKET_CATEGORIES: el["content"] = truncate_repetitive_content(content) elif category in ("Picture", "Flowchart"): el["content"] = f"![{truncate_repetitive_content(content)}]" elif category == "Chart": cleaned = normalize_inline_markdown_table(content) el["content"] = _markdown_lib.markdown( html.escape(cleaned), extensions=["tables"] ) elif category == "Formula": cleaned = truncate_repetitive_content(content) if ("\\(" in cleaned and "\\)" in cleaned) or ( "\\[" in cleaned and "\\]" in cleaned ): cleaned = cleaned.replace("$", "") cleaned = ( cleaned.replace("\\(", " $ ") .replace("\\)", " $ ") .replace("\\[", " $$ ") .replace("\\]", " $$ ") ) el["content"] = cleaned except Exception as e: # keep original content on post-processing failure logger.warning("element postprocess failed (%s): %s", category, e) # --- markdown formatting (port of formatters.markdown_formatter over dicts) --- # verbatim from deepparser_v2/formatters/markdown_formatter.py _HTML_STRIKE_OPEN_RE = re.compile(r"<\s*(?:s|strike|del)(?:\s[^>]*)?>", re.IGNORECASE) _HTML_STRIKE_CLOSE_RE = re.compile(r"<\s*/\s*(?:s|strike|del)\s*>", re.IGNORECASE) _LEADING_MD_HEADING_RE = re.compile(r"^[ \t]*#{1,6}[ \t]+") def _strip_leading_heading_marker(content: str) -> str: if not content: return content return _LEADING_MD_HEADING_RE.sub("", content, count=1) def _preserve_inline_markup(content: str) -> str: if not content: return content content = _HTML_STRIKE_OPEN_RE.sub("~~", content) content = _HTML_STRIKE_CLOSE_RE.sub("~~", content) return content def _nano_format_blockquote(content: str) -> str: stripped = content.strip() if not stripped: return "" lines = stripped.splitlines() quoted = [f"> {lines[0]}"] quoted.extend(f"> {line}" if line else ">" for line in lines[1:]) return "\n".join(quoted) def _nano_format_formula(content: str) -> str: stripped = content.strip() if not stripped: return "" if stripped.startswith("```"): return stripped return f"```latex\n{stripped}\n```" def _nano_image_markdown(el: Dict[str, Any]) -> str: image_src = el.get("picture_path") or "" if not image_src: return "" alt = {"Chart": "chart", "Flowchart": "flowchart"}.get(el["category"], "picture") return f"![{alt}]({image_src})" def _nano_format_element(el: Dict[str, Any]) -> str: category = el.get("category", "Text") content = el.get("content") or "" if category in ("Title", "Section-header"): prefix = "#" if category == "Title" else "##" return f"{prefix} {_preserve_inline_markup(_strip_leading_heading_marker(content))}" if category == "Table": return content if category == "Chart": image_markdown = _nano_image_markdown(el) blocks = [b for b in (image_markdown, content) if b] return "\n\n".join(blocks) if category in ("Picture", "Flowchart"): image_markdown = _nano_image_markdown(el) if not image_markdown: return content stripped = content.strip() if stripped.startswith("|") or stripped.startswith(" Tuple[str, List[Dict[str, Any]]]: """Port of response_builder._build_markdown_context/_build_full_markdown: sort by (page, layout_order), group contiguous same-page List-items into one block, drop empty blocks, page separators '---\\n\\n**Page N**' in the full document string only.""" valid = [ e for e in sorted(elements, key=lambda e: (e.get("page_number", 1), e.get("layout_order", 0))) if isinstance(e.get("page_number"), int) and not isinstance(e.get("page_number"), bool) and e.get("page_number", 0) >= 1 ] blocks: List[Tuple[int, str]] = [] # (page_number, content) index = 0 while index < len(valid): el = valid[index] page = el["page_number"] if el.get("category") == "List-item": items = [] while ( index < len(valid) and valid[index].get("category") == "List-item" and valid[index]["page_number"] == page ): formatted = _nano_format_element(valid[index]) if formatted: items.append(formatted) index += 1 content = "\n".join(items).strip() else: content = _nano_format_element(el).strip() index += 1 if content: blocks.append((page, content)) md_parts: List[str] = [] current_page: int | None = None for page, content in blocks: if current_page is not None and page != current_page: md_parts.append(f"---\n\n**Page {page}**") md_parts.append(content) current_page = page pages_md: Dict[int, List[str]] = {} for page, content in blocks: pages_md.setdefault(page, []).append(content) markdown_pages = [ {"page_number": page, "content": "\n\n".join(parts)} for page, parts in sorted(pages_md.items()) ] return "\n\n".join(md_parts), markdown_pages