Lire le texte d'une page scannée

"""
Read the text layer the document may already carry, before reaching for OCR.

Rung N0. Standard library only: zlib to inflate the page streams, and a
reading of the text-showing operators inside them.

Most « scanned pages » were never scanned. 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. Asking the question first is one function call, and it
answers the whole need whenever the answer is yes.

The point is the question, not the extractor. When the answer is no, this
function says so — it does not hand back an empty string, which a caller would
read as « the page is blank ».
"""

from __future__ import annotations

import re
import zlib

# Under this many non-space 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

STREAM = re.compile(rb"stream\r?\n(.*?)[\r\n]*endstream", re.S)
PAGE = re.compile(rb"/Type\s*/Page[^s]")

# Inside a content stream: a string being shown, or an operator that moves the
# cursor to another line. Kerning numbers inside a TJ array are skipped on
# purpose — they space glyphs, they do not carry characters.
TOKEN = re.compile(rb"\((?:\\.|[^\\()])*\)|<[0-9A-Fa-f\s]*>|\bTd\b|\bTD\b|\bT\*|\bET\b", re.S)

ESCAPES = {b"n": b"\n", b"r": b"\r", b"t": b"\t", b"b": b"\b", b"f": b"\f"}
ESCAPE = re.compile(rb"\\(?:([0-7]{1,3})|(.))", re.S)


def read_text_layer(pdf_bytes: bytes, *, min_characters: int = MIN_CHARACTERS) -> 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.
    """
    chunks = [t for t in (_read_stream(s) for s in STREAM.findall(pdf_bytes)) if t]
    text = "\n".join(chunks)
    characters = sum(1 for c in text if not c.isspace())
    has_text_layer = characters >= min_characters
    return {
        "has_text_layer": has_text_layer,
        "text": text,
        "characters": characters,
        "pages": len(PAGE.findall(pdf_bytes)),
        "reason": None if has_text_layer else "no text layer: this page is an image, and needs OCR",
    }


def _read_stream(raw: bytes) -> str:
    """Inflate the stream if it is compressed, then read what it shows."""
    try:
        # decompressobj, not decompress: a stream may carry padding after the
        # deflated data, and a raw decompress would refuse it.
        data = zlib.decompressobj().decompress(raw)
    except zlib.error:
        data = raw  # an uncompressed content stream is perfectly legal

    lines: list[str] = []
    current: list[str] = []
    for token in TOKEN.findall(data):
        if token.startswith(b"("):
            current.append(_literal(token[1:-1]))
        elif token.startswith(b"<"):
            current.append(_hex(token[1:-1]))
        elif current:
            lines.append("".join(current))
            current = []
    if current:
        lines.append("".join(current))
    return "\n".join(lines)


def _literal(body: bytes) -> str:
    """A literal string: backslash escapes, and octal for everything else."""

    def replace(match: re.Match) -> bytes:
        octal, char = match.groups()
        if octal:
            return bytes([int(octal, 8) & 0xFF])
        return ESCAPES.get(char, char)

    # Latin-1, because a simple font encodes one byte per character. A font
    # with its own encoding table needs that table, which is another job.
    return ESCAPE.sub(replace, body).decode("latin-1")


def _hex(body: bytes) -> str:
    """A hex string, as PDF writers emit for anything beyond ASCII."""
    digits = re.sub(rb"\s", b"", body)
    if len(digits) % 2:
        digits += b"0"  # the spec pads a lone last digit with zero
    raw = bytes.fromhex(digits.decode("ascii"))
    if raw[:2] == b"\xfe\xff":
        return raw[2:].decode("utf-16-be", errors="replace")
    return raw.decode("latin-1")

/**
 * Read the text layer the document may already carry, before reaching for OCR.
 *
 * Rung N0. Standard library only: node:zlib to inflate the page streams, and
 * a reading of the text-showing operators inside them.
 *
 * Most « scanned pages » were never scanned. 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. Asking the question first is one function
 * call, and it answers the whole need whenever the answer is yes.
 *
 * The point is the question, not the extractor. When the answer is no, this
 * function says so — it does not hand back an empty string, which a caller
 * would read as « the page is blank ».
 */
import zlib from 'node:zlib';

// Under this many non-space 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;

const STREAM = /stream\r?\n([\s\S]*?)[\r\n]*endstream/g;
const PAGE = /\/Type\s*\/Page[^s]/g;

// Inside a content stream: a string being shown, or an operator that moves the
// cursor to another line. Kerning numbers inside a TJ array are skipped on
// purpose — they space glyphs, they do not carry characters.
const TOKEN = /\((?:\\[\s\S]|[^\\()])*\)|<[0-9A-Fa-f\s]*>|\bTd\b|\bTD\b|\bT\*|\bET\b/g;

const ESCAPES = { n: '\n', r: '\r', t: '\t', b: '\b', f: '\f' };
const ESCAPE = /\\(?:([0-7]{1,3})|([\s\S]))/g;

/**
 * Say whether the document carries a text layer, and return it if it does.
 *
 * The answer is a report, not a string: `hasTextLayer` is the decision the
 * caller acts on, and `reason` is what to tell them when it is false.
 *
 * @param {Buffer|Uint8Array} pdfBytes
 * @param {{minCharacters?: number}} [options]
 */
export function readTextLayer(pdfBytes, { minCharacters = MIN_CHARACTERS } = {}) {
  // Latin-1 maps each byte to one character, so a regular expression can walk
  // the file without ever corrupting the compressed parts it steps over.
  const document = Buffer.from(pdfBytes).toString('latin1');

  const chunks = [];
  for (const [, stream] of document.matchAll(STREAM)) {
    const text = readStream(Buffer.from(stream, 'latin1'));
    if (text) chunks.push(text);
  }

  const text = chunks.join('\n');
  const characters = [...text].filter((c) => !/\s/.test(c)).length;
  const hasTextLayer = characters >= minCharacters;
  return {
    hasTextLayer,
    text,
    characters,
    pages: (document.match(PAGE) ?? []).length,
    reason: hasTextLayer ? null : 'no text layer: this page is an image, and needs OCR',
  };
}

/** Inflate the stream if it is compressed, then read what it shows. */
function readStream(raw) {
  let data;
  try {
    // Z_SYNC_FLUSH, because a stream may carry padding after the deflated
    // data, and a strict inflate would refuse it.
    data = zlib.inflateSync(raw, { finishFlush: zlib.constants.Z_SYNC_FLUSH }).toString('latin1');
  } catch {
    data = raw.toString('latin1'); // an uncompressed content stream is legal
  }

  const lines = [];
  let current = [];
  for (const [token] of data.matchAll(TOKEN)) {
    if (token.startsWith('(')) current.push(literal(token.slice(1, -1)));
    else if (token.startsWith('<')) current.push(hexString(token.slice(1, -1)));
    else if (current.length) {
      lines.push(current.join(''));
      current = [];
    }
  }
  if (current.length) lines.push(current.join(''));
  return lines.join('\n');
}

/**
 * A literal string: backslash escapes, and octal for everything else.
 *
 * The result stays in Latin-1, because a simple font encodes one byte per
 * character. A font with its own encoding table needs that table, which is
 * another job.
 */
function literal(body) {
  return body.replace(ESCAPE, (_, octal, char) =>
    octal ? String.fromCharCode(parseInt(octal, 8) & 0xff) : (ESCAPES[char] ?? char),
  );
}

/** A hex string, as PDF writers emit for anything beyond ASCII. */
function hexString(body) {
  let digits = body.replace(/\s/g, '');
  if (digits.length % 2) digits += '0'; // the spec pads a lone last digit with zero
  const raw = Buffer.from(digits, 'hex');
  if (raw[0] === 0xfe && raw[1] === 0xff) {
    return Buffer.from(raw.subarray(2)).swap16().toString('utf16le');
  }
  return raw.toString('latin1');
}

"""
Read a scanned page with a self-hosted optical recognition engine.

Rung N2. When N0 answered « no text layer », something has to look at the
pixels. An engine like Tesseract does exactly that, on your machine: the page
never leaves it, there is no key and no quota, and the same image gives the
same reading every time.

What you own on this rung is not the engine, it is everything around it: the
language you tell it to expect, the confidence under which a page goes to a
human, the cleaning of a text that comes back broken across lines, and the
answer to « what does the code do when the engine says nothing usable ».
"""

from __future__ import annotations

import re

# The language matters more than anything else you can tune here: an engine
# reading French with an English model invents accents it has never seen.
LANGUAGE = "fra"

# The engine reports how sure it is of each word. Below this, the page is
# still returned, but flagged: the text is a draft, not a reading.
DEFAULT_MIN_CONFIDENCE = 0.70


class OCRUnavailable(Exception):
    """The engine failed, or answered something no caller can act on."""


class TesseractOCR:
    """The real engine, started once and kept for the process."""

    def __init__(self, language: str = LANGUAGE) -> None:
        import pytesseract  # wraps the tesseract binary installed on the host

        self._pytesseract = pytesseract
        self.language = language

    def read(self, image_path: str) -> dict:
        """The text of one image, and how sure the engine is of it."""
        from PIL import Image

        data = self._pytesseract.image_to_data(
            Image.open(image_path),
            lang=self.language,
            output_type=self._pytesseract.Output.DICT,
        )
        # Word-level scores, on a hundred-point scale, and a score of -1 for
        # the layout boxes that carry no word at all.
        scores = [float(c) for w, c in zip(data["text"], data["conf"]) if w.strip() and float(c) >= 0]
        return {
            "text": "\n".join(data["text"]),
            "confidence": sum(scores) / (100 * len(scores)) if scores else 0.0,
        }


def read_page(image_path, engine=None, *, min_confidence: float = DEFAULT_MIN_CONFIDENCE, attempts: int = 2) -> dict:
    """
    Read one page image, and say whether a human should check the result.

    `engine` is injected so this can be tested without installing the binary
    or the language data. In production it defaults to the real engine above.
    """
    engine = engine or TesseractOCR()
    reading = _read(engine, image_path, attempts)
    if not isinstance(reading, dict) or not isinstance(reading.get("text"), str):
        raise OCRUnavailable("the engine owed a reading, and did not give one")

    text = clean(reading["text"])
    confidence = _confidence(reading.get("confidence"))
    # A doubtful page is not thrown away: it goes to a human with the text and
    # the score that earned the doubt. Throwing it away would cost the reading
    # that was, most of the time, almost right.
    return {"text": text, "confidence": confidence, "review": not text or confidence < min_confidence}


def _read(engine, image_path, attempts: int) -> object:
    """One page per call, and a failed call is retried, not swallowed."""
    last_error: Exception | None = None
    for _ in range(attempts):
        try:
            return engine.read(image_path)
        except Exception as error:  # noqa: BLE001 - any engine failure is retried
            last_error = error
    raise OCRUnavailable(str(last_error))


def clean(text: str) -> str:
    """Whitespace, and the hyphen a page break leaves inside a word."""
    lines = [re.sub(r"[ \t\xa0]+", " ", line).strip() for line in text.splitlines()]
    joined = "\n".join(line for line in lines if line)
    # « exemp-\nlaire » is one word the scanner cut in two, not two words.
    return re.sub(r"(\w)-\n(\w)", r"\1\2", joined)


def _confidence(value) -> float:
    """A number the caller can act on, rather than whatever came back."""
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        return 0.0
    return float(value) if 0.0 <= value <= 1.0 else 0.0

/**
 * Read a scanned page with a self-hosted optical recognition engine.
 *
 * Rung N2. When N0 answered « no text layer », something has to look at the
 * pixels. An engine like Tesseract does exactly that, on your machine: the
 * page never leaves it, there is no key and no quota, and the same image gives
 * the same reading every time.
 *
 * What you own on this rung is not the engine, it is everything around it: the
 * language you tell it to expect, the confidence under which a page goes to a
 * human, the cleaning of a text that comes back broken across lines, and the
 * answer to « what does the code do when the engine says nothing usable ».
 */

// The language matters more than anything else you can tune here: an engine
// reading French with an English model invents accents it has never seen.
export const LANGUAGE = 'fra';

// The engine reports how sure it is of each word. Below this, the page is
// still returned, but flagged: the text is a draft, not a reading.
export const DEFAULT_MIN_CONFIDENCE = 0.7;

export class OCRUnavailable extends Error {}

/** The real engine, started once and kept for the process. */
export class TesseractOCR {
  static async load(language = LANGUAGE) {
    const { createWorker } = await import('tesseract.js'); // pulls the language data once
    return new TesseractOCR(await createWorker(language));
  }

  constructor(worker) {
    this.worker = worker;
  }

  /** The text of one image, and how sure the engine is of it. */
  async read(imagePath) {
    const { data } = await this.worker.recognize(imagePath);
    // Word-level scores, on a hundred-point scale.
    const scores = data.words.filter((w) => w.text.trim()).map((w) => w.confidence);
    const total = scores.reduce((sum, score) => sum + score, 0);
    return { text: data.text, confidence: scores.length ? total / (100 * scores.length) : 0 };
  }
}

/**
 * Read one page image, and say whether a human should check the result.
 *
 * `engine` is injected so this can be tested without installing the binary or
 * the language data. In production it defaults to the real engine above.
 *
 * @param {string} imagePath
 * @param {{read: Function}} [engine]
 * @param {{minConfidence?: number, attempts?: number}} [options]
 */
export async function readPage(imagePath, engine, { minConfidence = DEFAULT_MIN_CONFIDENCE, attempts = 2 } = {}) {
  const reader = engine ?? (await TesseractOCR.load());
  const reading = await read(reader, imagePath, attempts);
  if (typeof reading?.text !== 'string') {
    throw new OCRUnavailable('the engine owed a reading, and did not give one');
  }

  const text = clean(reading.text);
  const confidence = confidenceOf(reading.confidence);
  // A doubtful page is not thrown away: it goes to a human with the text and
  // the score that earned the doubt. Throwing it away would cost the reading
  // that was, most of the time, almost right.
  return { text, confidence, review: text === '' || confidence < minConfidence };
}

/** One page per call, and a failed call is retried, not swallowed. */
async function read(engine, imagePath, attempts) {
  let lastError;
  for (let i = 0; i < attempts; i += 1) {
    try {
      return await engine.read(imagePath);
    } catch (error) {
      lastError = error;
    }
  }
  throw new OCRUnavailable(String(lastError));
}

/** Whitespace, and the hyphen a page break leaves inside a word. */
export function clean(text) {
  const lines = text
    .split('\n')
    .map((line) => line.replace(/[ \t\xa0]+/g, ' ').trim())
    .filter((line) => line !== '');
  // « exemp-\nlaire » is one word the scanner cut in two, not two words.
  return lines.join('\n').replace(/(\w)-\n(\w)/g, (_, before, after) => before + after);
}

/** A number the caller can act on, rather than whatever came back. */
function confidenceOf(value) {
  return typeof value === 'number' && value >= 0 && value <= 1 ? value : 0;
}

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

Rung N3. This is the option people reach for first, and on this entry it does
buy something real: a model that sees the page reads a handwritten annotation
in the margin, a stamp across a table, a column layout an OCR engine flattens.

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

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 signatures a scanner produces. Anything else is refused rather than sent
# and charged for, because a provider will reject it too.
SIGNATURES = ((b"\x89PNG\r\n\x1a\n", "image/png"), (b"\xff\xd8\xff", "image/jpeg"),
              (b"II*\x00", "image/tiff"), (b"MM\x00*", "image/tiff"))

# A page scan larger than this is a photograph of a desk, not a page. A model
# charges by what it is given, so refusing it is a cost control, not an
# optimisation.
MAX_IMAGE_BYTES = 8 * 1024 * 1024


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.
    """
    if client is None:  # pragma: no cover - needs a key and a network
        from openai import OpenAI

        client = OpenAI()

    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")

    # What the model admits it could not read is the only doubt it reports.
    # It is worth having, and it is not a confidence: see the test file.
    return {"text": text, "unreadable": [str(u) for u in unreadable], "review": bool(unreadable)}


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 image_bytes.startswith(signature):
            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")},
                # Temperature zero: a transcription that changes between two
                # identical calls cannot be checked by anyone.
                temperature=0,
            )
            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. This is the option people reach for first, and on this entry it
 * does buy something real: a model that sees the page reads a handwritten
 * annotation in the margin, a stamp across a table, a column layout an OCR
 * engine flattens.
 *
 * 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 signatures a scanner produces. Anything else is refused rather than sent
// and charged for, because a provider will reject it too.
const SIGNATURES = [
  [[0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a], 'image/png'],
  [[0xff, 0xd8, 0xff], 'image/jpeg'],
  [[0x49, 0x49, 0x2a, 0x00], 'image/tiff'],
  [[0x4d, 0x4d, 0x00, 0x2a], 'image/tiff'],
];

// A page scan larger than this is a photograph of a desk, not a page. A model
// charges by what it is given, so refusing it is a cost control, not an
// optimisation.
export const MAX_IMAGE_BYTES = 8 * 1024 * 1024;

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 } = {}) {
  if (!client) {
    // Needs a key and a network, so it is never reached in the tests.
    const { OpenAI } = await import('openai');
    client = new OpenAI();
  }

  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');
  }

  // What the model admits it could not read is the only doubt it reports. It
  // is worth having, and it is not a confidence: see the test file.
  return { text, unreadable: unreadable.map(String), review: unreadable.length > 0 };
}

/** Read the format from the bytes, rather than trusting a file extension. */
function mediaTypeOf(bytes) {
  for (const [signature, mediaType] of SIGNATURES) {
    if (signature.every((byte, i) => bytes[i] === byte)) 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') },
        // Temperature zero: a transcription that changes between two identical
        // calls cannot be checked by anyone.
        temperature: 0,
      });
      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));
}