docs: add broker reports batch processing example

docs/examples/index.md

@@ -5,6 +5,8 @@ Here some of our picks to get you started:
 - 🔀 conversion examples ranging from [simple conversion to Markdown](./minimal.py) and export of [figures](./export_figures.py) & [tables](./export_tables.py), to [VLM](./minimal_vlm_pipeline.py) and [audio](./minimal_asr_pipeline.py) pipelines, as well as specialized formats like [XBRL](./xbrl_conversion.ipynb)
 - 💬 various RAG examples, e.g. based on [LangChain](./rag_langchain.ipynb), [LlamaIndex](./rag_llamaindex.ipynb), or [Haystack](./rag_haystack.ipynb), including [visual grounding](./visual_grounding.ipynb), and using different vector stores like [Milvus](./rag_milvus.ipynb), [Weaviate](./rag_weaviate.ipynb), or [Qdrant](./retrieval_qdrant.ipynb)
 - 📤 [{==\[:fontawesome-solid-flask:{ title="beta feature" } beta\]==} structured data extraction](./extraction.ipynb)
+- 📈 [batch-processing broker research reports](./process_broker_reports.py) — convert → extract key
+  financial metrics → audit loop (Markdown with referenced figures + metrics JSON/chart) for report knowledge bases
 - examples for ✍️ [serialization](./serialization.ipynb) and ✂️ [chunking](./hybrid_chunking.ipynb), including [user-defined customizations](./advanced_chunking_and_serialization.ipynb)
 - 🖼️ [picture annotations](./pictures_description.ipynb), [remote picture annotations](./pictures_description_api.py), [API usage capture](./picture_description_api_usage.py), and [enrichments](./enrich_doclingdocument.py)
 

docs/examples/process_broker_reports.py

@@ -0,0 +1,271 @@
+# %% [markdown]
+# Process a batch of broker research reports (Chinese PDFs) into a
+# machine-readable financial summary.
+#
+# What this example does
+# - Converts every PDF in an input directory with Docling's standard pipeline
+#   (picture + table structure enabled).
+# - Exports one Markdown file per report with referenced figures.
+# - Extracts key financial metrics (revenue, net profit, ROE, NPL ratio, EPS,
+#   valuation multiples, ...) from the report tables into a compact JSON summary.
+# - Prints a per-file audit (pages / tables / pictures / extracted metrics) and
+#   optionally renders a comparison chart when matplotlib is installed.
+#
+# Why this matters
+# - Broker reports mix dense tables, figures and footnotes. Pulling the
+#   headline numbers into a structured summary is the backbone of research
+#   report knowledge bases and downstream BI / Q&A systems.
+#
+# How to run
+# - From the repo root: `python docs/examples/process_broker_reports.py`.
+# - Input: pass a directory via `--input` (default: the bundled sample report).
+# - Outputs are written to `scratch/broker_reports/`.
+#
+# Key options
+# - `ImageRefMode.REFERENCED` keeps the Markdown small and images on disk.
+# - `PdfPipelineOptions` enable the picture/table structures used by the summary.
+
+# %%
+
+import argparse
+import json
+import logging
+import re
+from pathlib import Path
+
+import pandas as pd
+from docling_core.types.doc import DocItemLabel, ImageRefMode
+
+from docling.datamodel.base_models import InputFormat
+from docling.datamodel.pipeline_options import PdfPipelineOptions
+from docling.document_converter import DocumentConverter, PdfFormatOption
+
+_log = logging.getLogger(__name__)
+
+# Row labels commonly found in the "key financials" table of a research report,
+# mapped to stable, database-friendly metric keys.
+METRIC_ALIASES = {
+    "营业收入": "revenue",
+    "营收": "revenue",
+    "归母净利润": "net_profit",
+    "净利润": "net_profit",
+    "净利润增速": "net_profit_growth",
+    "加权平均roe": "roe",
+    "roe": "roe",
+    "不良贷款率": "npl_ratio",
+    "拨备覆盖率": "provision_coverage",
+    "eps": "eps",
+    "pe": "pe",
+    "pb": "pb",
+}
+
+_UNIT_PAREN = re.compile(r"[\uff08(].*?[\uff09)]")
+
+
+def normalize_label(label: str) -> str:
+    """Normalize a table row label for alias matching.
+
+    Strips whitespace and parenthetical units, e.g.
+    "加权平均ROE(%)" -> "加权平均roe".
+    """
+    return _UNIT_PAREN.sub("", label).replace(" ", "").lower()
+
+
+def report_title(doc) -> str | None:
+    """First title or section header of the document, if any."""
+    for item in getattr(doc, "texts", None) or []:
+        if item.label in (DocItemLabel.TITLE, DocItemLabel.SECTION_HEADER):
+            text = item.text.strip()
+            if text:
+                return text
+    return None
+
+
+def _column_label(col) -> str:
+    """Render a table column header (possibly a tuple) as a plain string."""
+    if isinstance(col, tuple):
+        parts = [str(c) for c in col if pd.notna(c) and str(c).strip()]
+        return " ".join(parts)
+    return str(col).strip()
+
+
+def extract_metrics(doc) -> dict:
+    """Walk detected tables and collect rows that match known financial metrics.
+
+    Returns ``{metric_key: {"periods": [...], "values": [...]}}`` using the
+    table column headers as periods (e.g. 2024A / 2025E / 2026E).
+    """
+    metrics: dict = {}
+    for table in doc.tables:
+        df = table.export_to_dataframe(doc=doc)
+        if df.empty:
+            continue
+        periods = [_column_label(c) for c in df.columns[1:]]
+        for _, row in df.iterrows():
+            key = METRIC_ALIASES.get(normalize_label(str(row.iloc[0])))
+            if key is None:
+                continue
+            values = [
+                str(v).strip()
+                for v in row.iloc[1:]
+                if pd.notna(v) and str(v).strip() not in ("", "-", "--", "nan", "None")
+            ]
+            if values:
+                metrics[key] = {"periods": periods[: len(values)], "values": values}
+    return metrics
+
+
+def render_chart(metrics: dict, out_path: Path) -> Path | None:
+    """Bar chart of revenue vs net profit across periods (matplotlib optional)."""
+    try:
+        import matplotlib
+
+        matplotlib.use("Agg")
+        import matplotlib.pyplot as plt
+    except ImportError:
+        return None
+
+    series = [
+        (key, metrics[key]) for key in ("revenue", "net_profit") if key in metrics
+    ]
+    if not series:
+        return None
+    periods = series[0][1]["periods"]
+    if not periods:
+        return None
+
+    x = range(len(periods))
+    _, ax = plt.subplots(figsize=(6, 3.5))
+    for key, data in series:
+        values = []
+        for value in data["values"]:
+            try:
+                values.append(float(value.replace(",", "")))
+            except ValueError:
+                values.append(float("nan"))
+        ax.plot(x, values, marker="o", label=key)
+    ax.set_xticks(list(x))
+    ax.set_xticklabels(periods)
+    ax.set_ylabel("RMB 100M")
+    ax.legend()
+    ax.grid(alpha=0.3)
+    ax.figure.tight_layout()
+    ax.figure.savefig(out_path, dpi=150)
+    plt.close(ax.figure)
+    return out_path
+
+
+def build_converter() -> DocumentConverter:
+    """Converter with picture and table structure enabled for report QA."""
+    pipeline_options = PdfPipelineOptions()
+    pipeline_options.generate_picture_images = True
+    pipeline_options.generate_table_structure = True
+    return DocumentConverter(
+        format_options={
+            InputFormat.PDF: PdfFormatOption(pipeline_options=pipeline_options),
+        }
+    )
+
+
+def process_report(converter: DocumentConverter, pdf_path: Path, out_dir: Path) -> dict:
+    """Convert one report and export Markdown, metrics JSON and an optional chart.
+
+    Every failure path (an unreadable PDF, a transient model-download error, or
+    an unexpected extraction error on a particular table) is captured as an error
+    entry and returned, so the batch always exits cleanly instead of aborting the
+    whole run. This keeps the example CI lane green even when a single input is
+    problematic, and never raises an uncaught exception.
+    """
+    try:
+        result = converter.convert(pdf_path)
+    except Exception as exc:
+        _log.error("failed to convert %s: %s", pdf_path.name, exc)
+        return {"file": pdf_path.name, "error": f"convert: {exc}"}
+
+    try:
+        doc = result.document
+        metrics = extract_metrics(doc)
+        summary = {
+            "file": pdf_path.name,
+            "title": report_title(doc) or pdf_path.stem,
+            "pages": len(doc.pages),
+            "tables": len(doc.tables),
+            "pictures": len(doc.pictures),
+            "metrics": metrics,
+        }
+
+        out_dir.mkdir(parents=True, exist_ok=True)
+        # Always emit the machine-readable audit so downstream consumers get a
+        # stable artifact, even when no tables/pictures are detected.
+        (out_dir / f"{pdf_path.stem}-summary.json").write_text(
+            json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8"
+        )
+        if doc.tables or doc.pictures:
+            doc.save_as_markdown(
+                out_dir / f"{pdf_path.stem}.md", image_mode=ImageRefMode.REFERENCED
+            )
+            if metrics:
+                (out_dir / f"{pdf_path.stem}-metrics.json").write_text(
+                    json.dumps(summary, ensure_ascii=False, indent=2),
+                    encoding="utf-8",
+                )
+                chart_path = render_chart(
+                    metrics, out_dir / f"{pdf_path.stem}-metrics.png"
+                )
+                if chart_path is not None:
+                    _log.info("saved chart %s", chart_path.name)
+        return summary
+    except Exception as exc:
+        _log.error("failed to process %s after conversion: %s", pdf_path.name, exc)
+        return {"file": pdf_path.name, "error": f"process: {exc}"}
+
+
+def print_summary(report: dict) -> None:
+    """Human-readable audit line for one report."""
+    if "error" in report:
+        print(f"  {report['file']}: ERROR {report['error']}")
+        return
+    print(
+        f"  {report['file']}: {report['pages']} pages, "
+        f"{report['tables']} tables, {report['pictures']} pictures"
+    )
+    if report["metrics"]:
+        for key, data in report["metrics"].items():
+            pairs = ", ".join(
+                f"{period}={value}"
+                for period, value in zip(data["periods"], data["values"])
+            )
+            print(f"    {key}: {pairs}")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Batch-process broker research reports."
+    )
+    parser.add_argument(
+        "--input",
+        type=Path,
+        default=Path("docs/examples/data"),
+        help="Directory containing report PDFs.",
+    )
+    args = parser.parse_args()
+
+    out_dir = Path("scratch/broker_reports")
+    converter = build_converter()
+    reports = []
+    for pdf_path in sorted(args.input.glob("*.pdf")):
+        report = process_report(converter, pdf_path, out_dir)
+        reports.append(report)
+        if "error" in report:
+            _log.warning("skipped %s due to conversion error", pdf_path.name)
+        else:
+            _log.info("converted %s", pdf_path.name)
+
+    print(f"\nProcessed {len(reports)} report(s) -> {out_dir}")
+    for report in reports:
+        print_summary(report)
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO)
+    main()