Read the text of a scanned page

"""
Ask the document whether it already carries its text, before reaching for OCR.

Rung N0. One question, put to a PDF library, and an answer the caller can act
on.

A PDF produced by an accounting tool, a word processor or a print-to-PDF driver
carries its text next to its drawing instructions, already correct, with no
recognition step and therefore nothing to get wrong. A scan of the same
document carries pixels. The two look alike in a mailbox, and nothing
downstream tells them apart unless somebody asks.

The question is the point of this rung; the extractor is not. Writing the
extractor is the trap: inflating page streams and decoding the text-showing
operators is two hundred lines that still lose to the first font the file
renumbers for itself, which is what a word processor does on every export.
That work is done and shipped — `pypdf` here, `pdf.js` in JavaScript,
`pdftotext` at a shell prompt — and what is left to write is the decision.

Which is this: a page that carries no readable text is not an empty page, and
the function says so instead of handing back an empty string a caller would
read as « the page is blank ». A no comes with its next step.
"""

from __future__ import annotations

import io

# Under this many readable characters, what was found is a stamp, a page number
# or a stray label, not a text layer. Raise it for dense documents.
MIN_CHARACTERS = 24


def read_text_layer(pdf_bytes: bytes, *, min_characters: int = MIN_CHARACTERS, reader=None) -> dict:
    """
    Say whether the document carries a text layer, and return it if it does.

    The answer is a report, not a string: `has_text_layer` is the decision the
    caller acts on, and `reason` is what to tell them when it is false.

    `reader` is injected by the tests; in production it is `pypdf.PdfReader`.
    """
    if reader is None:  # imported here so the rest of this file needs nothing
        from pypdf import PdfReader as reader

    try:
        document = reader(io.BytesIO(pdf_bytes))
        pages = [page.extract_text() or "" for page in document.pages]
    except Exception as error:  # noqa: BLE001 - a broken file is an answer too
        return _report("", 0, 0, f"this file could not be opened as a PDF: {error}")

    text = "\n".join(page for page in pages if page.strip())
    characters = sum(1 for character in text if character.isprintable() and not character.isspace())
    if characters >= min_characters:
        return _report(text, characters, len(pages), None)
    return _report(text, characters, len(pages),
                   "no text layer: this page is an image, and needs OCR")


def _report(text: str, characters: int, pages: int, reason: str | None) -> dict:
    return {
        "has_text_layer": reason is None,
        "text": text,
        "characters": characters,
        "pages": pages,
        "reason": reason,
    }

/**
 * Ask the document whether it already carries its text, before reaching for OCR.
 *
 * Rung N0. One question, put to a PDF library, and an answer the caller can
 * act on.
 *
 * A PDF produced by an accounting tool, a word processor or a print-to-PDF
 * driver carries its text next to its drawing instructions, already correct,
 * with no recognition step and therefore nothing to get wrong. A scan of the
 * same document carries pixels. The two look alike in a mailbox, and nothing
 * downstream tells them apart unless somebody asks.
 *
 * The question is the point of this rung; the extractor is not. Writing the
 * extractor is the trap: inflating page streams and decoding the text-showing
 * operators is two hundred lines that still lose to the first font the file
 * renumbers for itself, which is what a word processor does on every export.
 * That work is done and shipped — `pdf.js` here, `pypdf` in Python,
 * `pdftotext` at a shell prompt — and what is left to write is the decision.
 *
 * Which is this: a page that carries no readable text is not an empty page,
 * and the function says so instead of handing back an empty string a caller
 * would read as « the page is blank ». A no comes with its next step.
 */

// Under this many readable characters, what was found is a stamp, a page
// number or a stray label, not a text layer. Raise it for dense documents.
export const MIN_CHARACTERS = 24;

/**
 * Say whether the document carries a text layer, and return it if it does.
 *
 * The answer is a report, not a string: `has_text_layer` is the decision the
 * caller acts on, and `reason` is what to tell them when it is false.
 *
 * `load` is injected by the tests; in production it is pdf.js.
 */
export async function readTextLayer(pdfBytes, { minCharacters = MIN_CHARACTERS, load } = {}) {
  // Imported here, and only here: nothing else in this file needs it.
  const getDocument = load ?? (await import('pdfjs-dist/legacy/build/pdf.mjs')).getDocument;
  let text;
  let pages = 0;
  try {
    const document = await getDocument({ data: new Uint8Array(pdfBytes), isEvalSupported: false }).promise;
    pages = document.numPages;
    const read = [];
    for (let number = 1; number <= pages; number += 1) {
      // eslint-disable-next-line no-await-in-loop -- pdf.js hands pages out one at a time
      const content = await (await document.getPage(number)).getTextContent();
      read.push(content.items.map((item) => item.str + (item.hasEOL ? '\n' : '')).join(''));
    }
    text = read.filter((page) => page.trim()).join('\n');
  } catch (error) {
    return report('', 0, 0, `this file could not be opened as a PDF: ${error.message}`);
  }

  const characters = [...text].filter((character) => !/\s/u.test(character)).length;
  if (characters >= minCharacters) return report(text, characters, pages, null);
  return report(text, characters, pages, 'no text layer: this page is an image, and needs OCR');
}

function report(text, characters, pages, reason) {
  return { has_text_layer: reason === null, text, characters, pages, reason };
}

"""
Read a scanned page by handing the image to a general-purpose multimodal model.

Rung N3. The page goes out whole, to a model that returns a transcription.

Note what the code has to do that N0 did not: recognise the image format, cap
what it sends, encode the bytes, retry on failure, parse an answer that is only
probably valid JSON, and refuse an answer it cannot use. That plumbing is the
real cost of this rung, and it is the part the tests have to cover, because the
model itself is not testable.

And note what none of that plumbing can do: tell whether the transcription is
what the page says. A model that reads is also a model that writes.
"""

from __future__ import annotations

import base64
import json
import re

PROMPT = (
    "Transcribe the page in the image, exactly as it is printed, keeping the\n"
    "line breaks. Answer with JSON only: an object with keys `text` and\n"
    "`unreadable`, where `text` is the transcription and `unreadable` is the\n"
    "list of fragments you could not read. Never guess at a fragment you\n"
    "cannot read: leave « ... » in the text and name it in `unreadable`."
)

# The formats the provider's vision input lists: PNG, JPEG, WEBP and GIF (not
# animated, which this check does not see). Anything else, a TIFF included, is
# refused before it is sent: convert it first.
SIGNATURES = ((rb"\x89PNG\r\n\x1a\n", "image/png"), (rb"\xff\xd8\xff", "image/jpeg"),
              (rb"RIFF.{4}WEBP", "image/webp"), (rb"GIF8[79]a", "image/gif"))

# A cap on what is encoded and uploaded; base64 makes the request a third larger
# than the file. It is not a cost control: the provider bills an image by its
# dimensions, not by its bytes.
MAX_IMAGE_BYTES = 8 * 1024 * 1024

# The provider named here is an example, not a recommendation: the reasoning
# holds for any general-purpose model API, and the client is swappable. Pass
# any object with a `complete(prompt=..., image=..., temperature=...)` method.
MODEL = "gpt-4.1-mini"  # an example id: check the parameters your model accepts


class ProviderClient:
    """The one call this snippet makes, on top of the provider's SDK."""

    def __init__(self, sdk=None, model: str = MODEL):
        if sdk is None:  # pragma: no cover - needs a key and a network
            from openai import OpenAI

            sdk = OpenAI()
        self.sdk, self.model = sdk, model

    def complete(self, *, prompt: str, image: dict, temperature: float) -> str:
        # The image travels inside the message, as a data URL.
        url = f"data:{image['media_type']};base64,{image['data']}"
        response = self.sdk.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": [
                {"type": "text", "text": prompt},
                {"type": "image_url", "image_url": {"url": url}},
            ]}],
            temperature=temperature,
        )
        return response.choices[0].message.content


class ReadingUnavailable(Exception):
    """The provider could not be reached, or answered something unusable."""


def read_page(image_bytes: bytes, client=None, *, attempts: int = 3, max_bytes: int = MAX_IMAGE_BYTES) -> dict:
    """
    Transcribe one page image, and say what the model admits it could not read.

    `client` is injected so this function can be tested without a network call.
    In production it defaults to a real provider client.
    """
    client = client or ProviderClient()
    media_type = _media_type(image_bytes)
    if len(image_bytes) > max_bytes:
        raise ValueError(f"image larger than {max_bytes} bytes")

    answer = _ask(client, image_bytes, media_type, attempts)
    text, unreadable = answer.get("text"), answer.get("unreadable", [])
    if not isinstance(text, str) or not isinstance(unreadable, list):
        raise ReadingUnavailable("the model answered JSON that is not a transcription")

    # Two doubts, and neither is a confidence: what the model admits it could
    # not read, and a page that came back empty. See the test file.
    review = bool(unreadable) or not text.strip()
    return {"text": text, "unreadable": [str(u) for u in unreadable], "review": review}


def _media_type(image_bytes: bytes) -> str:
    """Read the format from the bytes, rather than trusting a file extension."""
    for signature, media_type in SIGNATURES:
        if re.match(signature, image_bytes, re.S):
            return media_type
    raise ValueError("unrecognised image format")


def _ask(client, image_bytes: bytes, media_type: str, attempts: int) -> dict:
    last_error: Exception | None = None
    for _ in range(attempts):
        try:
            answer = client.complete(
                prompt=PROMPT,
                # Base64 is how an image travels in a JSON request body.
                image={"media_type": media_type, "data": base64.b64encode(image_bytes).decode("ascii")},
                # The lowest temperature: the SDK documents lower values as more
                # focused and deterministic.
                temperature=0,
            )
            if not isinstance(answer, str):  # a refusal comes back as no content
                last_error = ValueError("the model answered no text")
                continue
            parsed = json.loads(answer)
            if isinstance(parsed, dict):
                return parsed
            last_error = ValueError("the model answered something that is not an object")
        except Exception as error:  # noqa: BLE001 - any provider failure is retried
            last_error = error
    raise ReadingUnavailable(str(last_error))

/**
 * Read a scanned page by handing the image to a general-purpose multimodal
 * model.
 *
 * Rung N3. The page goes out whole, to a model that returns a transcription.
 *
 * Note what the code has to do that N0 did not: recognise the image format,
 * cap what it sends, encode the bytes, retry on failure, parse an answer that
 * is only probably valid JSON, and refuse an answer it cannot use. That
 * plumbing is the real cost of this rung, and it is the part the tests have to
 * cover, because the model itself is not testable.
 *
 * And note what none of that plumbing can do: tell whether the transcription
 * is what the page says. A model that reads is also a model that writes.
 */

export const PROMPT = [
  'Transcribe the page in the image, exactly as it is printed, keeping the',
  'line breaks. Answer with JSON only: an object with keys `text` and',
  '`unreadable`, where `text` is the transcription and `unreadable` is the',
  'list of fragments you could not read. Never guess at a fragment you',
  'cannot read: leave « ... » in the text and name it in `unreadable`.',
].join('\n');

// The formats the provider's vision input lists: PNG, JPEG, WEBP and GIF (not
// animated, which this check does not see). Anything else, a TIFF included, is
// refused before it is sent: convert it first.
const SIGNATURES = [
  [/^\x89PNG\r\n\x1a\n/, 'image/png'],
  [/^\xff\xd8\xff/, 'image/jpeg'],
  [/^RIFF[\s\S]{4}WEBP/, 'image/webp'],
  [/^GIF8[79]a/, 'image/gif'],
];

// A cap on what is encoded and uploaded; base64 makes the request a third
// larger than the file. It is not a cost control: the provider bills an image
// by its dimensions, not by its bytes.
export const MAX_IMAGE_BYTES = 8 * 1024 * 1024;

// The provider named here is an example, not a recommendation: the reasoning
// holds for any general-purpose model API, and the client is swappable. Pass
// any object with a `complete({ prompt, image, temperature })` method.
export const MODEL = 'gpt-4.1-mini'; // an example id: check the parameters your model accepts

export async function providerClient(sdk, model = MODEL) {
  if (!sdk) {
    const { OpenAI } = await import('openai');
    sdk = new OpenAI();
  }
  return {
    async complete({ prompt, image, temperature }) {
      // The image travels inside the message, as a data URL.
      const url = `data:${image.mediaType};base64,${image.data}`;
      const response = await sdk.chat.completions.create({
        model,
        messages: [{
          role: 'user',
          content: [{ type: 'text', text: prompt }, { type: 'image_url', image_url: { url } }],
        }],
        temperature,
      });
      return response.choices[0].message.content;
    },
  };
}

export class ReadingUnavailable extends Error {}

/**
 * Transcribe one page image, and say what the model admits it could not read.
 *
 * @param {Buffer|Uint8Array} imageBytes
 * @param {object} options
 * @param {{complete: Function}} [options.client] injected so this can be
 *   tested without a network call; defaults to a real provider client
 * @param {number} [options.attempts]
 * @param {number} [options.maxBytes]
 */
export async function readPage(imageBytes, { client, attempts = 3, maxBytes = MAX_IMAGE_BYTES } = {}) {
  client ??= await providerClient();
  const bytes = Buffer.from(imageBytes);
  const mediaType = mediaTypeOf(bytes);
  if (bytes.length > maxBytes) throw new RangeError(`image larger than ${maxBytes} bytes`);

  const answer = await ask(client, bytes, mediaType, attempts);
  const { text, unreadable = [] } = answer;
  if (typeof text !== 'string' || !Array.isArray(unreadable)) {
    throw new ReadingUnavailable('the model answered JSON that is not a transcription');
  }

  // Two doubts, and neither is a confidence: what the model admits it could
  // not read, and a page that came back empty. See the test file.
  const review = unreadable.length > 0 || text.trim() === '';
  return { text, unreadable: unreadable.map(String), review };
}

/** Read the format from the bytes, rather than trusting a file extension. */
function mediaTypeOf(bytes) {
  const head = bytes.subarray(0, 12).toString('latin1');
  for (const [signature, mediaType] of SIGNATURES) {
    if (signature.test(head)) return mediaType;
  }
  throw new TypeError('unrecognised image format');
}

async function ask(client, bytes, mediaType, attempts) {
  let lastError;
  for (let i = 0; i < attempts; i += 1) {
    try {
      const answer = await client.complete({
        prompt: PROMPT,
        // Base64 is how an image travels in a JSON request body.
        image: { mediaType, data: bytes.toString('base64') },
        // The lowest temperature: the SDK documents lower values as more
        // focused and deterministic.
        temperature: 0,
      });
      if (typeof answer !== 'string') {
        lastError = new Error('the model answered no text'); // a refusal comes back as no content
        continue;
      }
      const parsed = JSON.parse(answer);
      if (parsed && typeof parsed === 'object' && !Array.isArray(parsed)) return parsed;
      lastError = new Error('the model answered something that is not an object');
    } catch (error) {
      lastError = error;
    }
  }
  throw new ReadingUnavailable(String(lastError));
}