How to Extract Pages from PDF

"""Extract selected physical pages into a PDF and an audit manifest.

Install: pip install aspose-pdf
Run: python "src/06-pdf-extract pages/pdf-page-extractor.py"

Edit configuration below. "2,5,9-11" produces five pages; "5,2" preserves
that order. Repeated pages are rejected. Numbers are one-based physical page
positions, not printed labels. Set ASPOSE_PDF_LICENSE for the full example:
evaluation mode adds watermarks and limits processing to the first four pages.

Each successful run creates a new directory containing evidence-packet.pdf
and manifest.json. Files are staged beside that directory and published by
directory rename after verification. No existing destination is overwritten;
this is not a guarantee of atomic publication on every filesystem.

Verification compares page counts and extracted text in requested order.
For a synthetic 20-page report, put a unique "Source page N" marker on each
page and inspect the five resulting markers: 2, 5, 9, 10, 11. Scanned pages
need visual review; matching empty text does not prove correct page content.
Forms, annotations, bookmarks, links to omitted pages, and signatures need
separate review. This script does not repair those document features.

API references:
https://docs.aspose.com/pdf/python-net/extract-pages/
https://reference.aspose.com/pdf/python-net/aspose.pdf.text/textabsorber/
https://docs.aspose.com/pdf/python-net/licensing/
https://docs.aspose.com/pdf/python-net/system-requirements/
"""

from datetime import datetime, timezone
import hashlib
from importlib.metadata import version
import json
import os
from pathlib import Path
import re
import sys
from tempfile import TemporaryDirectory
from uuid import uuid4

import aspose.pdf as apdf


# Configuration ---------------------------------------------------------------
BASE_DIR = Path(__file__).resolve().parent
INPUT_FILE = BASE_DIR / "data" / "input" / "report.pdf"
OUTPUT_ROOT = BASE_DIR / "data" / "output"
PAGE_SELECTION = "2,5,9-11"
OUTPUT_FILENAME = "evidence-packet.pdf"
PASSWORD = os.environ.get("PDF_PASSWORD")
LICENSE_FILE = os.environ.get("ASPOSE_PDF_LICENSE")


def parse_selection(selection: str, page_count: int) -> list[int]:
    """Expand inclusive ranges, retaining order and rejecting duplicates.

    Invalid examples: "", "2,,5", "two", "5-2", "0", "2,2".
    Validate range bounds before expansion to avoid allocating huge ranges.
    """
    if page_count < 1:
        raise ValueError("Source PDF has no pages")
    pages = []
    seen = set()
    for token in selection.split(","):
        match = re.fullmatch(r"\s*([0-9]+)\s*(?:-\s*([0-9]+)\s*)?", token)
        if not match:
            raise ValueError(f"Invalid page selection item: {token!r}")
        start = int(match[1])
        end = int(match[2]) if match[2] is not None else start
        if not 1 <= start <= end <= page_count:
            raise ValueError(
                f"Invalid range {start}-{end}; source has {page_count} pages"
            )
        for page in range(start, end + 1):
            if page in seen:
                raise ValueError(f"Repeated source page: {page}")
            seen.add(page)
            pages.append(page)
    return pages


def source_digest(path: Path) -> str:
    """Identify source bytes without loading the entire PDF into memory."""
    with path.open("rb") as stream:
        return hashlib.file_digest(stream, "sha256").hexdigest()


def page_text(document: apdf.Document, page_number: int) -> str:
    absorber = apdf.text.TextAbsorber()
    absorber.visit(document.pages[page_number])
    return absorber.text


def copy_pages(source: apdf.Document, pages: list[int], output_file: Path) -> None:
    """Copy requested pages using Aspose's one-based page collection."""
    with apdf.Document() as output:
        output.ignore_corrupted_objects = False
        for position, page_number in enumerate(pages, start=1):
            try:
                output.pages.add(source.pages[page_number])
            except Exception as exc:
                raise RuntimeError(
                    f"Copying source page {page_number} to output page "
                    f"{position} failed: {exc}"
                ) from exc
        try:
            output.save(str(output_file))
        except Exception as exc:
            raise RuntimeError(f"Saving {output_file.name} failed: {exc}") from exc


def verify_packet(
    source: apdf.Document, pages: list[int], output_file: Path
) -> None:
    """Reopen the staged PDF and compare page count and text per position."""
    with apdf.Document(str(output_file)) as output:
        output.ignore_corrupted_objects = False
        if len(output.pages) != len(pages):
            raise ValueError(
                f"{output_file.name}: expected {len(pages)} pages, "
                f"found {len(output.pages)}"
            )
        for position, source_page in enumerate(pages, start=1):
            if page_text(source, source_page) != page_text(output, position):
                raise ValueError(
                    f"Text verification failed for output page {position} "
                    f"(source page {source_page})"
                )


def extract_packet(
    input_file: Path,
    output_root: Path,
    selection: str,
    password: str | None = None,
) -> Path:
    """Validate, stage, verify, and publish a PDF and its manifest together."""
    input_file = input_file.resolve(strict=True)
    if not input_file.is_file():
        raise ValueError(f"Input is not a file: {input_file}")
    digest = source_digest(input_file)
    try:
        source = (
            apdf.Document(str(input_file))
            if password is None
            else apdf.Document(str(input_file), password)
        )
    except Exception as exc:
        raise RuntimeError(
            f"Cannot open {input_file}; check PDF contents, permissions, "
            f"and PDF_PASSWORD. Library error: {exc}"
        ) from exc

    with source:
        source.ignore_corrupted_objects = False
        page_count = len(source.pages)
        pages = parse_selection(selection, page_count)
        output_root.mkdir(parents=True, exist_ok=True)
        destination = output_root / f"packet-{uuid4().hex}"
        if os.path.lexists(destination):
            raise FileExistsError(f"Refusing to overwrite {destination}")
        with TemporaryDirectory(prefix=".pdf-extract-", dir=output_root) as temp:
            staging_dir = Path(temp) / "results"
            staging_dir.mkdir()
            output_file = staging_dir / OUTPUT_FILENAME
            copy_pages(source, pages, output_file)
            verify_packet(source, pages, output_file)
            if source_digest(input_file) != digest:
                raise RuntimeError(f"Source changed during extraction: {input_file}")
            manifest = {
                "created_utc": datetime.now(timezone.utc).isoformat(),
                "aspose_pdf_version": version("aspose-pdf"),
                "source": {
                    "path": str(input_file),
                    "sha256": digest,
                    "page_count": page_count,
                },
                "selection": selection,
                "requested_pages": pages,
                "output_filename": OUTPUT_FILENAME,
                "output_sha256": source_digest(output_file),
                "page_mapping": [
                    {"output_page": position, "source_page": page}
                    for position, page in enumerate(pages, start=1)
                ],
                "verification": "page count and per-page extracted text matched",
            }
            manifest_file = staging_dir / "manifest.json"
            manifest_file.write_text(
                json.dumps(manifest, indent=2) + "\n", encoding="utf-8"
            )
            if json.loads(manifest_file.read_text(encoding="utf-8")) != manifest:
                raise ValueError("Staged manifest failed read-back verification")
            if os.path.lexists(destination):
                raise FileExistsError(f"Refusing to overwrite {destination}")
            staging_dir.rename(destination)
    return destination


def main() -> int:
    try:
        if LICENSE_FILE:
            pdf_license = apdf.License()
            pdf_license.set_license(LICENSE_FILE)

        destination = extract_packet(INPUT_FILE, OUTPUT_ROOT, PAGE_SELECTION, PASSWORD)
        print(f"Extraction completed: {destination / OUTPUT_FILENAME}")
        print(f"Manifest: {destination / 'manifest.json'}")
        print("Verified page count and text; review layout and document features.")
        return 0
    except Exception as exc:
        print(f"Extraction failed: {exc}", file=sys.stderr)
        return 1

if __name__ == "__main__":
    sys.exit(main())