ragflow/deepdoc/parser/mistral_parser.py

#
#  Copyright 2025 The InfiniFlow 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.
#
"""Mistral OCR PDF parser.

Calls Mistral's ``POST /v1/ocr`` endpoint (a dedicated document-OCR API, not
chat-completions), then normalizes the per-block response into the same shape
SoMark produces so the proven section-building contract is reused verbatim.
"""

import base64
import logging
import re
from io import BytesIO
from os import PathLike
from pathlib import Path
from typing import Optional

import numpy as np
import pdfplumber
import requests
from PIL import Image

from deepdoc.parser.pdf_parser import MAXIMUM_PAGE_NUMBER, RAGFlowPdfParser
from deepdoc.parser.utils import extract_pdf_outlines

# Mistral block "type" -> RAGFlow internal layout type.
# header/footer are resolved at runtime via keep_header_footer.
MISTRAL_TYPE_TO_RAGFLOW = {
    "text": "text",
    "title": "text",
    "list": "text",
    "table": "table",
    "image": "image",
    "equation": "equation",
    "code": "code",
}


class MistralParser(RAGFlowPdfParser):
    """Parse a PDF through Mistral OCR into (sections, tables)."""

    def __init__(
        self,
        base_url: str,
        api_key: str = "",
        *,
        model: str = "mistral-ocr-latest",
        table_format: str = "html",
        keep_header_footer: bool = False,
        timeout: int = 600,
        inline_max_bytes: int = 20 * 1024 * 1024,
    ):
        self.base_url = (base_url or "https://api.mistral.ai/v1").strip().rstrip("/")
        self.api_key = api_key
        self.model = model
        self.table_format = table_format.strip().lower() if table_format else "html"
        self.keep_header_footer = bool(keep_header_footer)
        self.timeout = int(timeout)
        self.inline_max_bytes = int(inline_max_bytes)
        self.outlines: list = []
        self.logger = logging.getLogger(self.__class__.__name__)
        # RAGFlowPdfParser.__init__ sets this default; MistralParser skips
        # that heavy __init__, so crop() needs it set before __images__ runs.
        self.page_from = 0
        # Optional tenant vision model (LLMBundle) for figure description;
        # set from parse_pdf's vision_model kwarg. None -> no enrichment.
        self.vision_model = None

    # ------------------------------------------------------------------
    # Page image rendering
    # ------------------------------------------------------------------
    def __images__(self, fnm, zoomin: int = 1, page_from: int = 0, page_to: int = MAXIMUM_PAGE_NUMBER, callback=None):
        self.page_from = page_from
        self.page_to = page_to
        try:
            ctx = pdfplumber.open(fnm) if isinstance(fnm, (str, PathLike)) else pdfplumber.open(BytesIO(fnm))
            with ctx as pdf:
                self.pdf = pdf
                self.page_images = [p.to_image(resolution=72 * zoomin, antialias=True).original for _, p in enumerate(self.pdf.pages[page_from:page_to])]
        except Exception as exc:
            self.page_images = None
            self.total_page = 0
            self.logger.exception(exc)

    # ------------------------------------------------------------------
    # Block classification
    # ------------------------------------------------------------------
    def _resolve_internal_type(self, block_type: Optional[str]) -> Optional[str]:
        """Return the RAGFlow internal layout type, or None to discard.

        header/footer obey keep_header_footer; unknown types fall back to text
        to avoid silent loss.
        """
        if block_type in ("header", "footer"):
            return "text" if self.keep_header_footer else None
        return MISTRAL_TYPE_TO_RAGFLOW.get(block_type, "text")

    @staticmethod
    def _block_text(block: dict, internal_type: str) -> str:
        """Textual payload for a block. image blocks contribute no text."""
        if internal_type == "image":
            return ""
        return (block.get("content") or "").strip()

    # ------------------------------------------------------------------
    # Response -> normalized pages (SoMark-compatible block shape)
    # ------------------------------------------------------------------
    def _normalize_pages(self, response: dict) -> list[dict]:
        """Map a /v1/ocr response into per-page dicts whose blocks carry a
        [x0, top, x1, bott] bbox and a {w,h} page_size, so downstream section
        building is identical to SoMark's."""
        out: list[dict] = []
        for page in (response or {}).get("pages") or []:
            dims = page.get("dimensions") or {}
            page_size = {"w": dims.get("width") or 1, "h": dims.get("height") or 1}
            blocks = []
            for b in page.get("blocks") or []:
                coord_keys = ("top_left_x", "top_left_y", "bottom_right_x", "bottom_right_y")
                if any(k in b for k in coord_keys):
                    bbox = [
                        b.get("top_left_x", 0),
                        b.get("top_left_y", 0),
                        b.get("bottom_right_x", 0),
                        b.get("bottom_right_y", 0),
                    ]
                else:
                    bbox = None
                blocks.append(
                    {
                        "type": b.get("type"),
                        "content": b.get("content"),
                        "bbox": bbox,
                    }
                )
            out.append(
                {
                    "page_num": page.get("index") or 0,
                    "page_size": page_size,
                    "blocks": blocks,
                }
            )
        return out

    # ------------------------------------------------------------------
    # Position tagging (compatible with RAGFlow extract_positions/crop)
    # ------------------------------------------------------------------
    def _line_tag(self, bx: dict) -> str:
        """Build a ``@@page\tx0\tx1\ttop\tbott##`` tag (page 1-based, order
        x0,x1,top,bott), rescaling bbox from Mistral's dimensions space to the
        locally rendered page pixels."""
        page_idx = bx.get("page_idx", 0)
        pn = [page_idx + 1]
        bbox = bx.get("bbox") or [0, 0, 0, 0]
        if len(bbox) != 4:
            bbox = [0, 0, 0, 0]
        x0, top, x1, bott = bbox
        page_size = bx.get("page_size") or {}
        src_w = page_size.get("w") or 1
        src_h = page_size.get("h") or 1

        if x0 > x1:
            x0, x1 = x1, x0
        if top > bott:
            top, bott = bott, top

        if getattr(self, "page_images", None) and len(self.page_images) > page_idx:
            page_width, page_height = self.page_images[page_idx].size
            x0 = max(0.0, (x0 / src_w) * page_width)
            x1 = max(0.0, (x1 / src_w) * page_width)
            top = max(0.0, (top / src_h) * page_height)
            bott = max(0.0, (bott / src_h) * page_height)

        return "@@{}\t{:.1f}\t{:.1f}\t{:.1f}\t{:.1f}##".format("-".join(str(x) for x in pn), x0, x1, top, bott)

    @staticmethod
    def extract_positions(txt: str):
        poss = []
        for tag in re.findall(r"@@[0-9-]+\t[0-9.\t]+##", txt):
            pn, left, right, top, bottom = tag.strip("#").strip("@").split("\t")
            left, right, top, bottom = float(left), float(right), float(top), float(bottom)
            poss.append(([int(p) - 1 for p in pn.split("-")], left, right, top, bottom))
        return poss

    def crop(self, text, ZM=1, need_position=False):
        """Image crop based on tags."""
        imgs = []
        poss = self.extract_positions(text)
        if not poss:
            return (None, None) if need_position else None
        if not getattr(self, "page_images", None):
            self.logger.warning("[Mistral OCR] crop called without page images; skip.")
            return (None, None) if need_position else None

        page_count = len(self.page_images)
        filtered = []
        for pns, left, right, top, bottom in poss:
            valid_pns = [p for p in pns if 0 <= p < page_count]
            if valid_pns:
                filtered.append((valid_pns, left, right, top, bottom))
        if not filtered:
            return (None, None) if need_position else None
        poss = filtered

        max_width = max(np.max([right - left for (_, left, right, _, _) in poss]), 6)
        GAP = 6
        first = poss[0]
        poss.insert(
            0,
            (
                [first[0][0]],
                first[1],
                first[2],
                max(0, first[3] - 120),
                max(first[3] - GAP, 0),
            ),
        )
        last = poss[-1]
        last_pn = last[0][-1]
        if not (0 <= last_pn < page_count):
            return (None, None) if need_position else None
        last_h = self.page_images[last_pn].size[1]
        poss.append(
            (
                [last_pn],
                last[1],
                last[2],
                min(last_h, last[4] + GAP),
                min(last_h, last[4] + 120),
            )
        )

        positions = []
        for ii, (pns, left, right, top, bottom) in enumerate(poss):
            right = left + max_width
            if bottom <= top:
                bottom = top + 2
            for pn in pns[1:]:
                if 0 <= pn - 1 < page_count:
                    bottom += self.page_images[pn - 1].size[1]
            if not (0 <= pns[0] < page_count):
                continue
            base_img = self.page_images[pns[0]]
            x0, y0, x1, y1 = (
                int(left),
                int(top),
                int(right),
                int(min(bottom, base_img.size[1])),
            )
            if x0 > x1:
                x0, x1 = x1, x0
            if y0 > y1:
                y0, y1 = y1, y0
            if x1 <= x0 or y1 <= y0:
                continue
            imgs.append(base_img.crop((x0, y0, x1, y1)))
            if 0 < ii < len(poss) - 1:
                positions.append((pns[0] + self.page_from, x0, x1, y0, y1))
            bottom -= base_img.size[1]
            for pn in pns[1:]:
                if not (0 <= pn < page_count):
                    continue
                page = self.page_images[pn]
                x0, y0, x1, y1 = (
                    int(left),
                    0,
                    int(right),
                    int(min(bottom, page.size[1])),
                )
                if x0 > x1:
                    x0, x1 = x1, x0
                if y0 > y1:
                    y0, y1 = y1, y0
                if x1 <= x0 or y1 <= y0:
                    bottom -= page.size[1]
                    continue
                imgs.append(page.crop((x0, y0, x1, y1)))
                if 0 < ii < len(poss) - 1:
                    positions.append((pn + self.page_from, x0, x1, y0, y1))
                bottom -= page.size[1]

        if not imgs:
            return (None, None) if need_position else None

        height = sum(img.size[1] + GAP for img in imgs)
        width = int(np.max([i.size[0] for i in imgs]))
        pic = Image.new("RGB", (width, int(height)), (245, 245, 245))
        offset = 0
        for ii, img in enumerate(imgs):
            if ii == 0 or ii + 1 == len(imgs):
                img = img.convert("RGBA")
                overlay = Image.new("RGBA", img.size, (0, 0, 0, 0))
                overlay.putalpha(128)
                img = Image.alpha_composite(img, overlay).convert("RGB")
            pic.paste(img, (0, int(offset)))
            offset += img.size[1] + GAP

        return (pic, positions) if need_position else pic

    # ------------------------------------------------------------------
    # Sections / tables
    # ------------------------------------------------------------------
    def _describe_image(self, line_tag: str) -> str:
        """Best-effort VLM caption of the figure at ``line_tag`` using the tenant
        vision model (``self.vision_model``). Crops the figure from the locally
        rendered page and describes it. Returns "" on any failure so figure
        enrichment never breaks parsing (mirrors MinerU's image enhancement)."""
        try:
            img = self.crop("figure" + line_tag, ZM=1)
            if img is None:
                return ""
            from rag.app.picture import vision_llm_chunk
            from rag.prompts.generator import vision_llm_figure_describe_prompt

            # vision_llm_chunk expects a PIL Image (it calls img.size / img.save),
            # not raw bytes — pass the crop directly.
            desc = vision_llm_chunk(binary=img, vision_model=self.vision_model, prompt=vision_llm_figure_describe_prompt())
            return (desc or "").strip()
        except Exception:
            self.logger.info("[Mistral OCR] figure description skipped", exc_info=True)
            return ""

    def _transfer_to_sections(self, pages: list[dict], parse_method: Optional[str] = None) -> list[tuple]:
        """manual/pipeline (rag/flow DAG) want typed 3-tuples
        (text, layout_type, line_tag); every other caller (naive.py) wants
        2-tuples (text, line_tag) that naive_merge consumes directly."""
        typed = parse_method in {"manual", "pipeline"}
        sections: list[tuple] = []
        image_seq = 0
        for page in pages or []:
            page_idx = page.get("page_num", 0)
            page_size = page.get("page_size") or {}
            for block in page.get("blocks") or []:
                internal = self._resolve_internal_type(block.get("type"))
                if internal is None:
                    continue
                bbox = block.get("bbox")
                tag_input = {"page_idx": page_idx, "bbox": bbox, "page_size": page_size}
                if internal == "image":
                    if not bbox or len(bbox) != 4:
                        continue  # no geometry -> nothing to crop
                    line_tag = self._line_tag(tag_input)
                    image_seq += 1
                    caption = (block.get("content") or "").strip()
                    description = self._describe_image(line_tag) if self.vision_model is not None else ""
                    label = description or caption or f"image {image_seq}"
                    if typed:
                        sections.append((label, internal, line_tag))
                    else:
                        # chunk id is hash(content + doc_id): empty image text would
                        # collide across figures. Keep a unique caption; append the
                        # tag so crop() recovers the figure, remove_tag() strips it.
                        sections.append((f"{label}{line_tag}", ""))
                    continue
                text = self._block_text(block, internal)
                if not text:
                    continue
                line_tag = self._line_tag(tag_input)
                if typed:
                    sections.append((text, internal, line_tag))
                else:
                    sections.append((text, line_tag))
        return sections

    def _transfer_to_tables(self, pages: list[dict]) -> list:
        # Tables are inlined as HTML in section text; no separate extraction.
        return []

    # ------------------------------------------------------------------
    # HTTP client
    # ------------------------------------------------------------------
    def _headers(self) -> dict:
        return {"Authorization": f"Bearer {self.api_key}"}

    def check_installation(self) -> tuple[bool, str]:
        if not self.api_key:
            return False, "Mistral API key is not configured."
        return True, ""

    def _raise_for_status(self, resp, action: str) -> None:
        if resp.status_code != 200:
            raise RuntimeError(f"Mistral {action} failed: {resp.status_code} {resp.text[:300]}")

    def _ocr_payload(self, document: dict, pages: Optional[list[int]]) -> dict:
        payload = {
            "model": self.model,
            "document": document,
            "include_blocks": True,
            "table_format": self.table_format,
            "include_image_base64": False,
        }
        if pages:
            payload["pages"] = pages
        return payload

    def _call_ocr(self, pdf_bytes: bytes, filename: str, pages: Optional[list[int]], callback=None) -> dict:
        ok, reason = self.check_installation()
        if not ok:
            raise RuntimeError(reason)

        if len(pdf_bytes) <= self.inline_max_bytes:
            b64 = base64.b64encode(pdf_bytes).decode()
            document = {"type": "document_url", "document_url": f"data:application/pdf;base64,{b64}"}
            return self._post_ocr(self._ocr_payload(document, pages))

        # Large file: upload -> signed url -> ocr -> delete.
        file_id = None
        try:
            r = requests.post(f"{self.base_url}/files", headers=self._headers(), files={"file": (filename, pdf_bytes, "application/pdf")}, data={"purpose": "ocr"}, timeout=self.timeout)
            self._raise_for_status(r, "/files upload")
            file_id = r.json().get("id")
            r = requests.get(f"{self.base_url}/files/{file_id}/url", headers=self._headers(), params={"expiry": 24}, timeout=self.timeout)
            self._raise_for_status(r, "signed-url fetch")
            signed = r.json().get("url")
            document = {"type": "document_url", "document_url": signed}
            return self._post_ocr(self._ocr_payload(document, pages))
        finally:
            if file_id:
                try:
                    resp = requests.delete(f"{self.base_url}/files/{file_id}", headers=self._headers(), timeout=self.timeout)
                    if not 200 <= resp.status_code < 300:
                        self.logger.warning("failed to delete uploaded file %s: %s %s", file_id, resp.status_code, resp.text[:200])
                except Exception:
                    self.logger.warning("failed to delete uploaded file %s", file_id)

    def _post_ocr(self, payload: dict) -> dict:
        r = requests.post(f"{self.base_url}/ocr", headers=self._headers(), json=payload, timeout=self.timeout)
        self._raise_for_status(r, "OCR")
        return r.json()

    # ------------------------------------------------------------------
    # Public entry point
    # ------------------------------------------------------------------
    def parse_pdf(self, filepath: str | PathLike[str], binary=None, callback=None, parse_method: str = "raw", from_page: int = 0, to_page: int = MAXIMUM_PAGE_NUMBER, **kwargs) -> tuple[list, list]:
        # Optional tenant vision model for figure description (best-effort).
        self.vision_model = kwargs.pop("vision_model", None)

        # Load bytes.
        if binary is not None:
            pdf_bytes = binary.getvalue() if hasattr(binary, "getvalue") else bytes(binary)
        else:
            pdf_bytes = Path(filepath).read_bytes()

        self.outlines = extract_pdf_outlines(pdf_bytes)

        # Render the WHOLE document locally: _line_tag/crop index page_images in
        # absolute page order, so a sliced render would crop the wrong page.
        self.__images__(pdf_bytes, zoomin=1)

        # pages is a selector: include only when the caller restricted the range.
        pages: Optional[list[int]] = None
        if from_page > 0 or to_page < MAXIMUM_PAGE_NUMBER:
            # Bound the selector by the real page count, never by the sentinel
            # to_page (MAXIMUM_PAGE_NUMBER) — otherwise a failed render would
            # build a ~100k-element list. Fall back to the PDF page count.
            rendered = len(self.page_images) if getattr(self, "page_images", None) else 0
            total = rendered or self.total_page_number(Path(filepath).name, pdf_bytes)
            end = min(to_page, total) if total else from_page
            pages = list(range(from_page, end))
            if not pages:
                return [], []

        if callback:
            callback(0.15, "[Mistral OCR] submitting document")

        response = self._call_ocr(pdf_bytes, Path(filepath).name, pages, callback=callback)
        norm = self._normalize_pages(response)

        if callback:
            n_blocks = sum(len(p.get("blocks") or []) for p in norm)
            callback(0.75, f"[Mistral OCR] parsed {n_blocks} blocks across {len(norm)} pages")

        sections = self._transfer_to_sections(norm, parse_method)
        tables = self._transfer_to_tables(norm)
        return sections, tables