"""OTSL -> HTML table conversion. Ported from the official MiniCPM-OCR SDK (OpenBMB/MiniCPM-OCR, sdk/minicpmvocr/postprocess/otsl/), Apache License 2.0. MiniCPM-OCR 1.0 emits tables in OTSL-v1.0 token format; the SDK converts them to HTML after recognition, so we do the same here. """ from __future__ import annotations import html import itertools import re 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, ) _ALL_TAGS = {OTSL_NL, OTSL_FCEL, OTSL_ECEL, OTSL_LCEL, OTSL_UCEL, OTSL_XCEL} def _otsl_pad_to_sqr_v2(otsl_str: str) -> str: """Pad an OTSL string so every row contains the same number of cells.""" 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 = [] 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) max_total_len = max(row["total_len"] for row in row_data) 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): cost = sum(abs(row["total_len"] - width) for row in row_data) if cost < min_total_cost: min_total_cost = cost optimal_width = width repaired_lines = [] for row in row_data: cells = row["raw_cells"] current_len = len(cells) if current_len > optimal_width: new_cells = cells[:optimal_width] else: new_cells = cells + [OTSL_ECEL] * (optimal_width - current_len) repaired_lines.append("".join(new_cells)) return OTSL_NL.join(repaired_lines) + OTSL_NL def _otsl_extract_tokens_and_text(s: str) -> tuple[list[str], list[str]]: pattern = "(" + "|".join([OTSL_NL, OTSL_FCEL, OTSL_ECEL, OTSL_LCEL, OTSL_UCEL, OTSL_XCEL]) + ")" tokens = re.findall(pattern, s) text_parts = re.split(pattern, s) text_parts = [t for t in text_parts if t.strip()] return tokens, text_parts def _otsl_parse_texts(texts: list[str], tokens: list[str]): split_row_tokens = [ list(y) for x, y in itertools.groupby(tokens, lambda z: z == OTSL_NL) if not x ] 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: list[str] = [] 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 _ALL_TAGS: 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_2d, c, r, which): span = 0 ci = c while ci < len(tokens_2d[r]) and tokens_2d[r][ci] in which: ci += 1 span += 1 return span def _count_down(tokens_2d, c, r, which): span = 0 ri = r while ri < len(tokens_2d) and c < len(tokens_2d[ri]) and tokens_2d[ri][c] in which: ri += 1 span += 1 return span table_cells: list[dict] = [] r_idx = 0 c_idx = 0 for i, text in enumerate(texts): if text in [OTSL_FCEL, OTSL_ECEL]: row_span = 1 col_span = 1 right_offset = 1 cell_text = "" if text != OTSL_ECEL: cell_text = texts[i + 1] if i + 1 < len(texts) else "" right_offset = 2 next_right = texts[i + right_offset] if i + right_offset < len(texts) else "" next_bottom = "" if r_idx + 1 < len(split_row_tokens) and c_idx < len(split_row_tokens[r_idx + 1]): next_bottom = split_row_tokens[r_idx + 1][c_idx] if next_right in [OTSL_LCEL, OTSL_XCEL]: col_span += _count_right(split_row_tokens, c_idx + 1, r_idx, [OTSL_LCEL, OTSL_XCEL]) if next_bottom in [OTSL_UCEL, OTSL_XCEL]: row_span += _count_down(split_row_tokens, c_idx, r_idx + 1, [OTSL_UCEL, OTSL_XCEL]) table_cells.append( { "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 convert_otsl_to_html(otsl_content: str) -> str: """Convert an OTSL-v1.0 string to an HTML ````.""" otsl_content = _otsl_pad_to_sqr_v2(html.unescape(otsl_content)) tokens, mixed_texts = _otsl_extract_tokens_and_text(otsl_content) table_cells, split_row_tokens = _otsl_parse_texts(mixed_texts, tokens) if not table_cells: return "" num_rows = len(split_row_tokens) num_cols = max(len(row) for row in split_row_tokens) if split_row_tokens else 0 grid = [ [ { "text": "", "start_row_offset_idx": i, "start_col_offset_idx": j, "row_span": 1, "col_span": 1, } for j in range(num_cols) ] for i in range(num_rows) ] for cell in table_cells: for i in range(min(cell["start_row_offset_idx"], num_rows), min(cell["end_row_offset_idx"], num_rows)): for j in range(min(cell["start_col_offset_idx"], num_cols), min(cell["end_col_offset_idx"], num_cols)): grid[i][j] = cell body = "" for i in range(num_rows): body += "" for j in range(num_cols): cell = grid[i][j] if cell["start_row_offset_idx"] != i or cell["start_col_offset_idx"] != j: continue content = html.escape(cell["text"].strip()) opening = "td" if cell["row_span"] > 1: opening += f' rowspan="{cell["row_span"]}"' if cell["col_span"] > 1: opening += f' colspan="{cell["col_span"]}"' body += f"<{opening}>{content}" body += "" return f"
{body}
"