How to Build a Parsing Pipeline with Docling Parse for Layout-Aware Document Intelligence

import os, sys, subprocess, textwrap, json, time, shutil from pathlib import Path def run(cmd):    print(f"n$ {cmd}")    return subprocess.run(cmd, shell=True, text=True, capture_output=False) run(f'{sys.executable} -m pip install -q --no-cache-dir -U "pillow>=10.4.0,<12" reportlab pandas matplotlib docling-core docling-parse') try:    from PIL import Image, ImageDraw except ImportError:    print("nPillow import failed because Colab has a mixed PIL installation.")    print("Reinstalling Pillow and restarting runtime. After restart, run this same cell again.")    run(f'{sys.executable} -m pip uninstall -y pillow PIL')    run(f'{sys.executable} -m pip install -q --no-cache-dir --force-reinstall "pillow>=10.4.0,<12"')    os.kill(os.getpid(), 9) import pandas as pd import matplotlib.pyplot as plt from reportlab.lib.pagesizes import A4 from reportlab.lib import colors from reportlab.platypus import Table, TableStyle from reportlab.pdfgen import canvas from docling_core.types.doc.page import TextCellUnit from docling_parse.pdf_parser import DoclingPdfParser print("Environment ready.") print("Python:", sys.version.split()[0]) WORKDIR = Path("https://www.marktechpost.com/content/docling_parse_advanced_tutorial") WORKDIR.mkdir(parents=True, exist_ok=True) PDF_PATH = WORKDIR / "advanced_docling_parse_demo.pdf" OUT_DIR = WORKDIR / "outputs" OUT_DIR.mkdir(exist_ok=True) DEMO_IMAGE_PATH = WORKDIR / "demo_bitmap.png" 

def create_demo_image(path):    img = Image.new("RGB", (320, 180), "white")    draw = ImageDraw.Draw(img)    draw.rectangle([20, 20, 300, 160], outline="black", width=3)    draw.ellipse([55, 45, 145, 135], outline="black", width=4)    draw.line([180, 140, 285, 45], fill="black", width=4)    draw.text((45, 145), "Embedded bitmap image", fill="black")    img.save(path) create_demo_image(DEMO_IMAGE_PATH) def build_pdf(pdf_path):    c = canvas.Canvas(str(pdf_path), pagesize=A4)    width, height = A4    c.setFont("Helvetica-Bold", 20)    c.drawString(60, height - 70, "Docling Parse Advanced PDF Parsing Tutorial")    c.setFont("Helvetica", 11)    intro = (        "This generated document is designed for testing text extraction, coordinate parsing, "        "line grouping, vector path detection, bitmap resources, and layout-aware reconstruction."    )    text_obj = c.beginText(60, height - 105)    text_obj.setLeading(15)    for line in textwrap.wrap(intro, width=90):        text_obj.textLine(line)    c.drawText(text_obj)    c.setFont("Helvetica-Bold", 14)    c.drawString(60, height - 170, "1. Two-column text region")    left_para = (        "The left column contains compact explanatory text. A parser should expose words, "        "characters, and line-level cells along with coordinates. These coordinates allow us "        "to reconstruct reading order and inspect the spatial structure of a page."    )    right_para = (        "The right column contains a separate paragraph. In document AI pipelines, layout "        "features are useful for retrieval, table extraction, chunking, and downstream RAG "        "applications where page position can matter."    )    y_start = height - 200    left_text = c.beginText(60, y_start)    left_text.setFont("Helvetica", 10)    left_text.setLeading(13)    for line in textwrap.wrap(left_para, width=42):        left_text.textLine(line)    c.drawText(left_text)    right_text = c.beginText(325, y_start)    right_text.setFont("Helvetica", 10)    right_text.setLeading(13)    for line in textwrap.wrap(right_para, width=42):        right_text.textLine(line)    c.drawText(right_text)    c.setStrokeColor(colors.darkblue)    c.setLineWidth(2)    c.rect(55, height - 315, 225, 130, stroke=1, fill=0)    c.rect(320, height - 315, 225, 130, stroke=1, fill=0)    c.setStrokeColor(colors.darkgreen)    c.setLineWidth(3)    c.circle(140, height - 390, 40, stroke=1, fill=0)    c.line(220, height - 430, 310, height - 355)    c.setFont("Helvetica-Bold", 14)    c.setFillColor(colors.black)    c.drawString(60, height - 470, "2. Simple table-like structure")    data = [        ["Section", "Signal", "Expected parser behavior"],        ["Text", "Words and lines", "Return text cells with coordinates"],        ["Vector", "Boxes and lines", "Expose page path/vector resources"],        ["Bitmap", "Embedded image", "Expose or render image resources"],    ]    table = Table(data, colWidths=[100, 130, 260])    table.setStyle(TableStyle([        ("BACKGROUND", (0, 0), (-1, 0), colors.lightgrey),        ("GRID", (0, 0), (-1, -1), 0.7, colors.black),        ("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"),        ("FONTSIZE", (0, 0), (-1, -1), 9),        ("VALIGN", (0, 0), (-1, -1), "MIDDLE"),    ]))    table.wrapOn(c, width, height)    table.drawOn(c, 60, height - 590)    c.setFont("Helvetica", 9)    c.drawString(60, 55, "Page 1: generated programmatic PDF with text, table-like layout, and vector paths.")    c.showPage()    c.setFont("Helvetica-Bold", 18)    c.drawString(60, height - 70, "Page 2: Bitmap, Dense Text, and Reading Order")    c.setFont("Helvetica", 10)    dense = (        "This page includes an embedded bitmap image and several short blocks of text. "        "We use it to test whether rendering works, whether the parser preserves page-level "        "coordinates, and whether our own reconstruction logic can group words into lines."    )    y = height - 105    for para_idx in range(4):        tx = c.beginText(60, y)        tx.setFont("Helvetica", 10)        tx.setLeading(13)        for line in textwrap.wrap(f"Block {para_idx + 1}: {dense}", width=92):            tx.textLine(line)        c.drawText(tx)        y -= 70    c.drawImage(str(DEMO_IMAGE_PATH), 110, height - 510, width=320, height=180, preserveAspectRatio=True)    c.setStrokeColor(colors.red)    c.setLineWidth(2)    c.roundRect(95, height - 525, 350, 210, 10, stroke=1, fill=0)    c.setFillColor(colors.black)    c.setFont("Helvetica-Bold", 12)    c.drawString(60, height - 570, "Coordinate-aware extraction lets us keep page, text, and position together.")    c.setFont("Helvetica", 9)    c.drawString(60, 55, "Page 2: embedded bitmap image and multiple text blocks.")    c.save() build_pdf(PDF_PATH) print("Created PDF:", PDF_PATH) 

def safe_to_dict(obj, max_depth=2):    if obj is None:        return None    if isinstance(obj, (str, int, float, bool)):        return obj    if isinstance(obj, (list, tuple)):        return [safe_to_dict(x, max_depth=max_depth - 1) for x in obj[:50]]    if isinstance(obj, dict):        return {            str(k): safe_to_dict(v, max_depth=max_depth - 1)            for k, v in list(obj.items())[:50]        }    if hasattr(obj, "model_dump"):        try:            return obj.model_dump()        except Exception:            pass    if hasattr(obj, "__dict__") and max_depth > 0:        try:            return {                k: safe_to_dict(v, max_depth=max_depth - 1)                for k, v in obj.__dict__.items()                if not k.startswith("_")            }        except Exception:            pass    return str(obj) def rect_to_dict(rect):    d = safe_to_dict(rect)    if isinstance(d, dict):        return d    attrs = {}    for name in [        "l", "t", "r", "b",        "left", "top", "right", "bottom",        "x0", "y0", "x1", "y1",        "width", "height"    ]:        if hasattr(rect, name):            try:                attrs[name] = getattr(rect, name)            except Exception:                pass    return attrs if attrs else {"raw": str(rect)} def get_text_cell_records(page_no, pred_page, unit_type):    records = []    try:        cells = list(pred_page.iterate_cells(unit_type=unit_type))    except Exception as e:        print(f"Could not iterate {unit_type} cells on page {page_no}: {e}")        return records    for idx, cell in enumerate(cells):        text = getattr(cell, "text", "")        rect = getattr(cell, "rect", None)        records.append({            "page": page_no,            "unit": str(unit_type).split(".")[-1],            "index": idx,            "text": text,            "rect": rect_to_dict(rect),            "raw_cell": safe_to_dict(cell, max_depth=1),        })    return records def count_possible_resources(pred_page):    resource_summary = {}    names = dir(pred_page)    keywords = ["path", "bitmap", "image", "resource", "line", "rect"]    for name in names:        lname = name.lower()        if any(k in lname for k in keywords) and not name.startswith("_"):            try:                value = getattr(pred_page, name)                if callable(value):                    continue                try:                    resource_summary[name] = len(value)                except Exception:                    resource_summary[name] = type(value).__name__            except Exception:                pass    return resource_summary parser = DoclingPdfParser() start = time.perf_counter() pdf_doc = parser.load(path_or_stream=str(PDF_PATH)) load_time = time.perf_counter() - start print(f"nLoaded PDF in {load_time:.3f} seconds.") all_records = [] page_summaries = [] rendered_paths = [] parse_start = time.perf_counter() for page_no, pred_page in pdf_doc.iterate_pages():    print(f"n--- Page {page_no} ---")    word_records = get_text_cell_records(page_no, pred_page, TextCellUnit.WORD)    char_records = get_text_cell_records(page_no, pred_page, TextCellUnit.CHAR)    line_records = get_text_cell_records(page_no, pred_page, TextCellUnit.LINE)    all_records.extend(word_records)    all_records.extend(char_records)    all_records.extend(line_records)    resource_summary = count_possible_resources(pred_page)    page_summaries.append({        "page": page_no,        "words": len(word_records),        "chars": len(char_records),        "lines": len(line_records),        "possible_resource_attributes": resource_summary,    })    print("Words:", len(word_records))    print("Characters:", len(char_records))    print("Lines:", len(line_records))    print("Possible resource attributes:", resource_summary)    print("nFirst 20 extracted words:")    print(" ".join([r["text"] for r in word_records[:20]]))    for unit_name, unit_type in [        ("word", TextCellUnit.WORD),        ("char", TextCellUnit.CHAR),        ("line", TextCellUnit.LINE),    ]:        try:            img = pred_page.render_as_image(cell_unit=unit_type)            out_img = OUT_DIR / f"page_{page_no}_{unit_name}_overlay.png"            img.save(out_img)            rendered_paths.append(out_img)            print("Saved rendered overlay:", out_img)        except Exception as e:            print(f"Could not render {unit_name} overlay for page {page_no}: {e}") parse_time = time.perf_counter() - parse_start 

records_path = OUT_DIR / "docling_parse_cells.json" with open(records_path, "w", encoding="utf-8") as f:    json.dump(all_records, f, indent=2, ensure_ascii=False) summary_path = OUT_DIR / "page_summaries.json" with open(summary_path, "w", encoding="utf-8") as f:    json.dump(page_summaries, f, indent=2, ensure_ascii=False) flat_rows = [] for r in all_records:    rect = r.get("rect", {})    row = {        "page": r["page"],        "unit": r["unit"],        "index": r["index"],        "text": r["text"],    }    if isinstance(rect, dict):        for k, v in rect.items():            if isinstance(v, (str, int, float, bool)) or v is None:                row[f"rect_{k}"] = v            else:                row[f"rect_{k}"] = str(v)    flat_rows.append(row) df = pd.DataFrame(flat_rows) csv_path = OUT_DIR / "docling_parse_cells.csv" df.to_csv(csv_path, index=False) summary_df = pd.DataFrame(page_summaries) summary_csv_path = OUT_DIR / "page_summaries.csv" summary_df.to_csv(summary_csv_path, index=False) print("nSaved structured outputs:") print(records_path) print(csv_path) print(summary_path) print(summary_csv_path) print("nPage summary:") display(summary_df) print("nCell dataframe sample:") display(df.head(20)) def extract_rect_numbers(rect):    if not isinstance(rect, dict):        return None    possible_sets = [        ("l", "t", "r", "b"),        ("left", "top", "right", "bottom"),        ("x0", "y0", "x1", "y1"),    ]    for keys in possible_sets:        if all(k in rect for k in keys):            try:                vals = [float(rect[k]) for k in keys]                return vals            except Exception:                pass    numeric = []    for v in rect.values():        try:            numeric.append(float(v))        except Exception:            pass    if len(numeric) >= 4:        return numeric[:4]    return None word_df = df[df["unit"].str.contains("WORD", case=False, na=False)].copy() if len(word_df) == 0:    word_df = df[df["unit"].str.contains("word", case=False, na=False)].copy() coords = [] for _, row in word_df.iterrows():    rect_data = {}    for col in word_df.columns:        if col.startswith("rect_"):            rect_data[col.replace("rect_", "")] = row[col]    nums = extract_rect_numbers(rect_data)    coords.append(nums) word_df["coord_numbers"] = coords word_df = word_df[word_df["coord_numbers"].notna()].copy() if len(word_df) > 0:    word_df["x0"] = word_df["coord_numbers"].apply(lambda x: min(x[0], x[2]))    word_df["x1"] = word_df["coord_numbers"].apply(lambda x: max(x[0], x[2]))    word_df["y0"] = word_df["coord_numbers"].apply(lambda x: min(x[1], x[3]))    word_df["y1"] = word_df["coord_numbers"].apply(lambda x: max(x[1], x[3]))    word_df["y_mid"] = (word_df["y0"] + word_df["y1"]) / 2    reconstructed_pages = {}    for page, g in word_df.groupby("page"):        g = g.sort_values(["y_mid", "x0"]).copy()        y_values = sorted(g["y_mid"].tolist())        line_bins = []        threshold = 8.0        for y in y_values:            placed = False            for line in line_bins:                if abs(line["center"] - y) <= threshold:                    line["values"].append(y)                    line["center"] = sum(line["values"]) / len(line["values"])                    placed = True                    break            if not placed:                line_bins.append({"center": y, "values": [y]})        def assign_line(y):            return min(range(len(line_bins)), key=lambda i: abs(line_bins[i]["center"] - y))        g["line_id"] = g["y_mid"].apply(assign_line)        lines = []        for line_id, lg in g.groupby("line_id"):            lg = lg.sort_values("x0")            line_text = " ".join(lg["text"].astype(str).tolist())            lines.append((lg["y_mid"].mean(), line_text))        lines = sorted(lines, key=lambda x: x[0])        reconstructed_text = "n".join([line for _, line in lines])        reconstructed_pages[int(page)] = reconstructed_text    recon_path = OUT_DIR / "layout_aware_reconstructed_text.json"    with open(recon_path, "w", encoding="utf-8") as f:        json.dump(reconstructed_pages, f, indent=2, ensure_ascii=False)    print("nLayout-aware reconstructed text:")    for page, text in reconstructed_pages.items():        print(f"n===== PAGE {page} =====")        print(text[:2500])    print("nSaved reconstruction:", recon_path) else:    print("nCould not build coordinate-based reconstruction because rectangle coordinates were not exposed in a numeric form.") 

print("nAttempting threaded parsing benchmark...") threaded_results = [] threaded_available = True try:    from docling_parse.pdf_parser import DoclingThreadedPdfParser, ThreadedPdfParserConfig    from docling_parse.pdf_parsers import DecodePageConfig    parser_config = ThreadedPdfParserConfig(        loglevel="fatal",        threads=4,        max_concurrent_results=32,    )    decode_config = DecodePageConfig()    threaded_parser = DoclingThreadedPdfParser(        parser_config=parser_config,        decode_config=decode_config,    )    t0 = time.perf_counter()    doc_key = threaded_parser.load(str(PDF_PATH))    page_count = threaded_parser.page_count(doc_key)    print("Threaded doc key:", doc_key)    print("Threaded page count:", page_count)    for result in threaded_parser.iterate_results():        item = {            "doc_key": str(getattr(result, "doc_key", "")),            "page_number": getattr(result, "page_number", None),            "success": getattr(result, "success", None),            "error_message": getattr(result, "error_message", None),        }        if getattr(result, "success", False):            seg_page = result.get_page()            timings = result.get_timings()            item["word_count"] = len(getattr(seg_page, "word_cells", []))            try:                item["total_time"] = timings.total()            except Exception:                item["total_time"] = str(timings)        threaded_results.append(item)    threaded_time = time.perf_counter() - t0 except Exception as e:    threaded_available = False    threaded_time = None    print("Threaded parser is not available or failed in this environment.")    print("Error:", repr(e)) if threaded_available:    threaded_path = OUT_DIR / "threaded_parse_results.json"    with open(threaded_path, "w", encoding="utf-8") as f:        json.dump(threaded_results, f, indent=2, ensure_ascii=False)    threaded_df = pd.DataFrame(threaded_results)    print("nThreaded parsing results:")    display(threaded_df)    print("Saved threaded results:", threaded_path) benchmark = {    "standard_load_time_seconds": load_time,    "standard_iterate_parse_time_seconds": parse_time,    "threaded_total_time_seconds": threaded_time,    "total_cells_extracted": len(all_records),    "output_dir": str(OUT_DIR), } benchmark_path = OUT_DIR / "benchmark.json" with open(benchmark_path, "w", encoding="utf-8") as f:    json.dump(benchmark, f, indent=2) print("nBenchmark:") print(json.dumps(benchmark, indent=2)) print("nChecking CLI availability...") cli = shutil.which("docling-parse") print("docling-parse CLI:", cli) if cli:    cli_result = subprocess.run(        "docling-parse -h",        shell=True,        text=True,        capture_output=True,    )    print(cli_result.stdout[:1000]) else:    print("CLI was not found on PATH, but the Python API worked.") print("nGenerated files:") for p in sorted(OUT_DIR.glob("*")):    print(p) print("nRendering saved overlay images in notebook...") for img_path in rendered_paths[:6]:    try:        img = Image.open(img_path)        plt.figure(figsize=(9, 12))        plt.imshow(img)        plt.axis("off")        plt.title(img_path.name)        plt.show()    except Exception as e:        print("Could not display:", img_path, e) print("nTutorial complete.") print("Main outputs are stored in:", OUT_DIR)