from __future__ import annotations import json import re from typing import Any import bleach import markdown TABLE_RE = re.compile(r"", re.IGNORECASE) JSON_OBJECT_RE = re.compile( r'\{[^{}]*?"bbox"\s*:\s*\[[^\]]+\][^{}]*?\}', re.IGNORECASE, ) BBOX_TAG_RE = re.compile( r"""<(?:bbox|box)(?:\s+label=["']?([^"'>]+)["']?)?[^>]*> \s*\[?\s*([\d.]+)\s*,\s*([\d.]+)\s*,\s*([\d.]+)\s*,\s*([\d.]+)\s*\]? \s*""", re.IGNORECASE | re.VERBOSE, ) TABLE_TAGS = [ "table", "caption", "thead", "tbody", "tfoot", "tr", "th", "td", "br", "strong", "em", ] TABLE_ATTRIBUTES = { "th": ["colspan", "rowspan", "scope"], "td": ["colspan", "rowspan"], } TEXT_TAGS = [ "p", "h1", "h2", "h3", "h4", "h5", "ul", "ol", "li", "strong", "em", "code", "pre", "blockquote", "hr", "br", ] TABLE_TEXT_MARKER = "`...
`" def _normalize_box(candidate: dict[str, Any], index: int) -> dict | None: bbox = candidate.get("bbox") if not isinstance(bbox, list) or len(bbox) != 4: return None try: coords = [float(value) for value in bbox] except (TypeError, ValueError): return None maximum = max(coords) scale = 1 if maximum <= 1 else (1000 if maximum <= 1000 else maximum) x1, y1, x2, y2 = [min(1, max(0, value / scale)) for value in coords] if x2 <= x1 or y2 <= y1: return None label = str(candidate.get("label") or candidate.get("type") or "element") return { "id": f"{index}-{'-'.join(str(value) for value in coords)}", "label": label[:48], "bbox": [x1, y1, x2, y2], } def extract_bounding_boxes(raw_text: str) -> list[dict]: candidates: list[dict] = [] for match in JSON_OBJECT_RE.finditer(raw_text): try: candidate = json.loads(match.group(0)) except json.JSONDecodeError: continue if isinstance(candidate, dict): candidates.append(candidate) for match in BBOX_TAG_RE.finditer(raw_text): candidates.append( { "label": match.group(1) or "element", "bbox": [float(match.group(index)) for index in range(2, 6)], } ) boxes = [] for index, candidate in enumerate(candidates): normalized = _normalize_box(candidate, index) if normalized: boxes.append(normalized) return boxes[:250] def extract_tables(raw_text: str) -> str: tables = TABLE_RE.findall(raw_text) if not tables: return "" return bleach.clean( "\n".join(tables), tags=TABLE_TAGS, attributes=TABLE_ATTRIBUTES, protocols=[], strip=True, ) def render_text(raw_text: str) -> str: text_with_table_markers = TABLE_RE.sub( f"\n\n{TABLE_TEXT_MARKER}\n\n", raw_text, ) rendered = markdown.markdown( text_with_table_markers, extensions=["extra", "sane_lists"], output_format="html", ) return bleach.clean( rendered, tags=TEXT_TAGS, attributes={}, protocols=[], strip=True, ) def shape_parser_output(raw_text: str) -> dict: output = raw_text if isinstance(raw_text, str) else "" return { "raw_output": output, "output_chars": len(output), "boxes": extract_bounding_boxes(output), "tables_html": extract_tables(output), "text_html": render_text(output), } def _structured_box( payload: dict[str, Any], *, label: str, index: int, ) -> dict | None: bbox = payload.get("bounding_box_normalized") if not isinstance(bbox, dict): return None return _normalize_box( { "label": label, "bbox": [ bbox.get("top_left_x"), bbox.get("top_left_y"), bbox.get("bottom_right_x"), bbox.get("bottom_right_y"), ], }, index, ) def _shape_blocks_page(page: dict[str, Any]) -> tuple[str, list[dict]]: blocks = page.get("blocks") if not isinstance(blocks, list): raise ValueError("blocks page is missing blocks") content: list[str] = [] boxes: list[dict] = [] for block in blocks: if not isinstance(block, dict): raise ValueError("invalid parse block") block_type = block.get("type") if block_type == "text": text = block.get("text") if not isinstance(text, dict) or not isinstance(text.get("content"), str): raise ValueError("invalid text block") content.append(text["content"]) continue if block_type == "table": table = block.get("table") if not isinstance(table, dict) or not isinstance(table.get("html"), str): raise ValueError("invalid table block") content.append(table["html"]) box = _structured_box( table, label=str(table.get("title") or "table"), index=len(boxes), ) elif block_type == "image": image = block.get("image") if not isinstance(image, dict): raise ValueError("invalid image block") description = str(image.get("description") or "Image") content.append(f"*Image: {description}*") box = _structured_box( image, label=str(image.get("category") or image.get("id") or "image"), index=len(boxes), ) else: raise ValueError("unsupported parse block") if box: boxes.append(box) return "\n\n".join(content), boxes def _shape_markdown_page(page: dict[str, Any]) -> tuple[str, list[dict]]: markdown_page = page.get("markdown") if not isinstance(markdown_page, dict) or not isinstance( markdown_page.get("content"), str ): raise ValueError("invalid markdown page") boxes = [] for image in markdown_page.get("images") or []: if not isinstance(image, dict): raise ValueError("invalid markdown image") box = _structured_box( image, label=str(image.get("category") or image.get("id") or "image"), index=len(boxes), ) if box: boxes.append(box) return markdown_page["content"], boxes def shape_parse_response(body: dict[str, Any]) -> dict: if not isinstance(body, dict): raise TypeError("parse response must be an object") pages = body.get("pages") if not isinstance(pages, list) or not pages: raise ValueError("parse response is missing pages") page_content: list[str] = [] boxes: list[dict] = [] for page in pages: if not isinstance(page, dict): raise ValueError("invalid parse page") page_type = page.get("type") if page_type == "blocks": content, page_boxes = _shape_blocks_page(page) elif page_type == "markdown": content, page_boxes = _shape_markdown_page(page) elif page_type == "raw_generation" and isinstance( page.get("raw_generation"), str ): content, page_boxes = page["raw_generation"], [] else: raise ValueError("unsupported parse page") page_content.append(content) boxes.extend(page_boxes) text_output = "\n\n".join(page_content) shaped = shape_parser_output(text_output) return { **shaped, "raw_output": json.dumps(body, indent=2, ensure_ascii=False), "output_chars": len(text_output), "boxes": boxes[:250], "text_output": text_output, "response": body, }