Construire un analyseur PDF personnalisé avec PYPDF et Langchain

 Construire un analyseur PDF personnalisé avec PYPDF et Langchain



Image de l’auteur | Toile

Les fichiers PDF sont partout. Vous les avez probablement vus dans divers endroits, tels que les papiers universitaires, les factures d’électricité, les contrats de bureau, les manuels de produits, etc. Ils sont super communs, mais travailler avec eux n’est pas aussi facile qu’il n’y paraît. Disons que vous souhaitez extraire des informations utiles d’un PDF, comme lire le texte, les diviser en sections ou obtenir un résumé rapide. Cela peut sembler simple, mais vous verrez que ce n’est pas si lisse une fois que vous essayez.

Contrairement aux fichiers Word ou HTML, les PDF ne stockent pas de contenu d’une manière soignée et lisible. Au lieu de cela, ils sont conçus pour bien paraître, pour ne pas être lus par des programmes. Le texte peut être partout, divisé en blocs étranges, dispersés sur la page ou mélangés avec des tables et des images. Il est difficile de leur obtenir des données propres et structurées.

Dans cet article, nous allons construire quelque chose qui peut gérer ce gâchis. Nous créerons un analyseur PDF personnalisé qui peut:

  • Extraire et nettoyer le texte des PDF au niveau de la page, avec une conservation de mise en page facultative pour une meilleure mise en forme
  • Gérer l’extraction des métadonnées de l’image
  • Retirez les en-têtes et les pieds de page indésirables en détectant des lignes répétées à travers les pages pour réduire le bruit
  • Récupérer des métadonnées détaillées du document et du niveau de page, telles que l’auteur, le titre, la date de création, la rotation et la taille de la page
  • Chunk le contenu en pièces gérables pour plus de traitement NLP ou LLM

Commençons.

Structure de dossier

Avant de commencer, il est bon d’organiser vos fichiers de projet pour plus de clarté et d’évolutivité.

custom_pdf_parser/
│
├── parser.py           
├── langchain_loader.py  
├── pipeline.py          
├── example.py     
├── requirements.txt     # Dependencies list
└── __init__.py         # (Optional) to mark directory as Python package

Vous pouvez quitter le __init.py__ Fichier vide, car son objectif principal est simplement d’indiquer que ce répertoire doit être traité comme un package Python. J’expliquerai l’étape par étape de l’objectif de chacun des fichiers restants.

Outils requis (exigences.txt)

Les bibliothèques nécessaires sont:

  • Pypdf: Une bibliothèque Python pure à lire et à écrire des fichiers PDF. Il sera utilisé pour extraire le texte des fichiers PDF
  • Langchain: Un framework pour créer des applications contextuelles avec des modèles de langage (nous l’utiliserons pour traiter et chaîner des tâches de documents). Il sera utilisé pour traiter et organiser correctement le texte.

Installez-les avec:

pip install pypdf langchain

Si vous souhaitez gérer soigneusement les dépendances, créez un exigences.txt dossier avec:

Et courir:

pip install -r requirements.txt

Étape 1: Configurez le PDF Parser (Parser.py)

La classe principale CustomPdfParser Utilise le PYPDF pour extraire du texte et des métadonnées de chaque page PDF. Il comprend également des méthodes pour nettoyer le texte, extraire les informations d’image (facultative) et supprimer des en-têtes ou des pieds de page répétés qui apparaissent souvent sur chaque page.

  • Il prend en charge la préservation de la mise en forme de disposition
  • Il extrait les métadonnées comme le numéro de page, la rotation et les dimensions de la boîte de support
  • Il peut filtrer les pages avec trop peu de contenu
  • Le nettoyage de texte supprime les espaces excessifs tout en préservant les ruptures de paragraphes

La logique qui implémente tout cela est:

import os
import logging
from pathlib import Path
from typing import List, Dict, Any
import pypdf
from pypdf import PdfReader
# Configure logging to show info and above messages
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class CustomPDFParser:
  def __init__(
      self,extract_images: bool = False,preserve_layout: bool = True,remove_headers_footers: bool = True,min_text_length: int = 10
  ):
      """
      Initialize the parser with options to extract images, preserve layout, remove repeated headers/footers, and minimum text length for pages.
      Args:
          extract_images: Whether to extract image info from pages
          preserve_layout: Whether to keep layout spacing in text extraction
          remove_headers_footers: Whether to detect and remove headers/footers
          min_text_length: Minimum length of text for a page to be considered valid
      """
      self.extract_images = extract_images
      self.preserve_layout = preserve_layout
      self.remove_headers_footers = remove_headers_footers
      self.min_text_length = min_text_length
  def extract_text_from_page(self, page: pypdf.PageObject, page_num: int) -> Dict(str, Any):
      """
      Extract text and metadata from a single PDF page.
      Args:
          page: PyPDF page object
          page_num: zero-based page number
      Returns:
          dict with keys:
              - 'text': extracted and cleaned text string,
              - 'metadata': page metadata dict,
              - 'word_count': number of words in extracted text
      """
      try:
 # Extract text, optionally preserving the layout for better formatting
          if self.preserve_layout:
              text = page.extract_text(extraction_mode="layout")
          else:
              text = page.extract_text()
        # Clean text: remove extra whitespace and normalize paragraphs
          text = self._clean_text(text)
        # Gather page metadata (page number, rotation angle, mediabox)
          metadata = {
              "page_number": page_num + 1,  # 1-based numbering
              "rotation": getattr(page, "rotation", 0),
              "mediabox": str(getattr(page, "mediabox", None)),
          }
          # Optionally, extract image info from page if requested
          if self.extract_images:
              metadata("images") = self._extract_image_info(page)
          # Return dictionary with text and metadata for this page
          return {
              "text": text,
              "metadata": metadata,
              "word_count": len(text.split()) if text else 0
          }
      except Exception as e:
          # Log error and return empty data for problematic pages
          logger.error(f"Error extracting page {page_num}: {e}")
          return {
              "text": "",
              "metadata": {"page_number": page_num + 1, "error": str(e)},
              "word_count": 0
          }
  def _clean_text(self, text: str) -> str:
      """
      Clean and normalize extracted text, preserving paragraph breaks.
      Args:
          text: raw text extracted from PDF page
      Returns:
          cleaned text string
      """
      if not text:
          return ""
      lines = text.split('n')
      cleaned_lines = ()
      for line in lines:
          line = line.strip()  # Remove leading/trailing whitespace
          if line:
              # Non-empty line; keep it
              cleaned_lines.append(line)
          elif cleaned_lines and cleaned_lines(-1):
              # Preserve paragraph break by keeping empty line only if previous line exists
              cleaned_lines.append("")
      cleaned_text="n".join(cleaned_lines)
#Reduce any instances of more than two consecutive blank lines to two
      while 'nnn' in cleaned_text:
          cleaned_text = cleaned_text.replace('nnn', 'nn')
      return cleaned_text.strip()
  def _extract_image_info(self, page: pypdf.PageObject) -> List(Dict(str, Any)):
      """
      Extract basic image metadata from page, if available.
      Args:
          page: PyPDF page object
      Returns:
          List of dictionaries with image info (index, name, width, height)
      """
      images = ()
      try:
          # PyPDF pages can have an 'images' attribute listing embedded images
          if hasattr(page, 'images'):
              for i, image in enumerate(page.images):
                  images.append({
                      "image_index": i,
                      "name": getattr(image, 'name', f"image_{i}"),
                      "width": getattr(image, 'width', None),
                      "height": getattr(image, 'height', None)
                  })
      except Exception as e:
          logger.warning(f"Image extraction failed: {e}")
      return images

  def _remove_headers_footers(self, pages_data: List(Dict(str, Any))) -> List(Dict(str, Any)):
      """
      Remove repeated headers and footers that appear on many pages.
      This is done by identifying lines appearing on over 50% of pages
      at the start or end of page text, then removing those lines.
      Args:
          pages_data: List of dictionaries representing each page's extracted data.
      Returns:
          Updated list of pages with headers/footers removed
      """
      # Only attempt removal if enough pages and option enabled
      if len(pages_data) < 3 or not self.remove_headers_footers:
          return pages_data
      # Collect first and last lines from each page's text for analysis
      first_lines = (page("text").split('n')(0) if page("text") else "" for page in pages_data)
      last_lines = (page("text").split('n')(-1) if page("text") else "" for page in pages_data)
      threshold = len(pages_data) * 0.5  # More than 50% pages
      # Identify candidate headers and footers appearing frequently
      potential_headers = (line for line in set(first_lines)
                          if first_lines.count(line) > threshold and line.strip())
      potential_footers = (line for line in set(last_lines)
                          if last_lines.count(line) > threshold and line.strip())
      # Remove identified headers and footers from each page's text
      for page_data in pages_data:
          lines = page_data("text").split('n')
          # Remove header if it matches a frequent header
          if lines and potential_headers:
              for header in potential_headers:
                  if lines(0).strip() == header.strip():
                      lines = lines(1:)
                      break
          # Remove footer if it matches a frequent footer
          if lines and potential_footers:
              for footer in potential_footers:
                  if lines(-1).strip() == footer.strip():
                      lines = lines(:-1)
                      break

          page_data("text") = 'n'.join(lines).strip()
      return pages_data
  def _extract_document_metadata(self, pdf_reader: PdfReader, pdf_path: str) -> Dict(str, Any):
      """
      Extract metadata from the PDF document itself.
      Args:
          pdf_reader: PyPDF PdfReader instance
          pdf_path: path to PDF file
      Returns:
          Dictionary of metadata including file info and PDF document metadata
      """
      metadata = {
          "file_path": pdf_path,
          "file_name": Path(pdf_path).name,
          "file_size": os.path.getsize(pdf_path) if os.path.exists(pdf_path) else None,
      }
      try:
          if pdf_reader.metadata:
              # Extract common PDF metadata keys if available
              metadata.update({
                  "title": pdf_reader.metadata.get('/Title', ''),
                  "author": pdf_reader.metadata.get('/Author', ''),
                  "subject": pdf_reader.metadata.get('/Subject', ''),
                  "creator": pdf_reader.metadata.get('/Creator', ''),
                  "producer": pdf_reader.metadata.get('/Producer', ''),
                  "creation_date": str(pdf_reader.metadata.get('/CreationDate', '')),
                  "modification_date": str(pdf_reader.metadata.get('/ModDate', '')),
              })
      except Exception as e:
          logger.warning(f"Metadata extraction failed: {e}")
      return metadata
  def parse_pdf(self, pdf_path: str) -> Dict(str, Any):
      """
      Parse the entire PDF file. Opens the file, extracts text and metadata page by page, removes headers/footers if configured, and aggregates results.
      Args:
          pdf_path: Path to the PDF file
      Returns:
          Dictionary with keys:
              - 'full_text': combined text from all pages,
              - 'pages': list of page-wise dicts with text and metadata,
              - 'document_metadata': file and PDF metadata,
              - 'total_pages': total pages in PDF,
              - 'processed_pages': number of pages kept after filtering,
              - 'total_words': total word count of parsed text
      """
      try:
          with open(pdf_path, 'rb') as file:
              pdf_reader = PdfReader(file)
              doc_metadata = self._extract_document_metadata(pdf_reader, pdf_path)
              pages_data = ()
              # Iterate over all pages and extract data
              for i, page in enumerate(pdf_reader.pages):
                  page_data = self.extract_text_from_page(page, i)
                  # Only keep pages with sufficient text length
                  if len(page_data("text")) >= self.min_text_length:
                      pages_data.append(page_data)
              # Remove repeated headers and footers
              pages_data = self._remove_headers_footers(pages_data)
           # Combine all page texts with a double newline as a separator
              full_text="nn".join(page("text") for page in pages_data if page("text"))
              # Return final structured data
              return {
                  "full_text": full_text,
                  "pages": pages_data,
                  "document_metadata": doc_metadata,
                  "total_pages": len(pdf_reader.pages),
                  "processed_pages": len(pages_data),
                  "total_words": sum(page("word_count") for page in pages_data)
              }
      except Exception as e:
          logger.error(f"Failed to parse PDF {pdf_path}: {e}")
          raise

Étape 2: Intégrez à Langchain (Langchain_loader.py)

Le Langchainpdfloader La classe enroule l’analyseur personnalisé et convertit les pages analysées en Langchain Document Objets, qui sont les éléments constitutifs des pipelines de Langchain.

  • Il permet de se lancer des documents en pièces plus petites en utilisant Langchain RecursiveCaracterTextSplitter
  • Vous pouvez personnaliser les tailles de morceaux et se chevaucher pour l’entrée LLM en aval
  • Ce chargeur prend en charge l’intégration propre entre le contenu PDF brut et l’abstraction du document de Langchain

La logique derrière cela est:

from typing import List, Optional, Dict, Any
from langchain.schema import Document
from langchain.document_loaders.base import BaseLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from parser import CustomPDFParser  # import the parser defined above
class LangChainPDFLoader(BaseLoader):
   def __init__(
       self,file_path: str,parser_config: Optional(Dict(str, Any)) = None,chunk_size: int = 500, chunk_overlap: int = 50
   ):
       """
       Initialize the loader with the PDF file path, parser configuration, and chunking parameters.
       Args:
           file_path: path to PDF file
           parser_config: dictionary of parser options
           chunk_size: chunk size for splitting long texts
           chunk_overlap: chunk overlap for splitting
       """
       self.file_path = file_path
       self.parser_config = parser_config or {}
       self.chunk_size = chunk_size
       self.chunk_overlap = chunk_overlap
       self.parser = CustomPDFParser(**self.parser_config)
   def load(self) -> List(Document):
       """
       Load PDF, parse pages, and convert each page to a LangChain Document.
       Returns:
           List of Document objects with page text and combined metadata.
       """
       parsed_data = self.parser.parse_pdf(self.file_path)
       documents = ()
       # Convert each page dict to a LangChain Document
       for page_data in parsed_data("pages"):
           if page_data("text"):
               # Merge document-level and page-level metadata
               metadata = {**parsed_data("document_metadata"), **page_data("metadata")}
               doc = Document(page_content=page_data("text"), metadata=metadata)
               documents.append(doc)
       return documents
   def load_and_split(self) -> List(Document):
       """
       Load the PDF and split large documents into smaller chunks.
       Returns:
           List of Document objects after splitting large texts.
       """
       documents = self.load()
       # Initialize a text splitter with the desired chunk size and overlap
       text_splitter = RecursiveCharacterTextSplitter(
           chunk_size=self.chunk_size,
           chunk_overlap=self.chunk_overlap,
           separators=("nn", "n", " ", "")  # hierarchical splitting
       )
       # Split documents into smaller chunks
       split_docs = text_splitter.split_documents(documents)
       return split_docs

Étape 3: Construire un pipeline de traitement (pipeline.py)

Le PDFProcessingPipeline La classe fournit une interface de niveau supérieur pour:

  • Traitement un seul PDF
  • Sélection du format de sortie (dict brut, documents Langchain ou texte brut)
  • Activer ou désactiver la section avec des tailles de morceaux configurables
  • Gestion des erreurs et journalisation

Cette abstraction permet une intégration facile dans des applications ou des flux de travail plus importants. La logique derrière cela est:

from typing import List, Optional, Dict, Any
from langchain.schema import Document
from parser import CustomPDFParser
from langchain_loader import LangChainPDFLoader
import logging
logger = logging.getLogger(__name__)
class PDFProcessingPipeline:
   def __init__(self, parser_config: Optional(Dict(str, Any)) = None):
       """
       Args:
          parser_config: dictionary of options passed to CustomPDFParser
       """
       self.parser_config = parser_config or {}
   def process_single_pdf(
       self,pdf_path: str,output_format: str = "langchain",chunk_documents: bool = True,chunk_size: int = 500,chunk_overlap: int = 50
   ) -> Any:
       """
       Args:
           pdf_path: path to PDF file
           output_format: "raw" (dict), "langchain" (Documents), or "text" (string)
           chunk_documents: whether to split LangChain documents into chunks
           chunk_size: chunk size for splitting
           chunk_overlap: chunk overlap for splitting
       Returns:
           Parsed content in the requested format
       """
       if output_format == "raw":
           # Use raw CustomPDFParser output
           parser = CustomPDFParser(**self.parser_config)
           return parser.parse_pdf(pdf_path)
       elif output_format == "langchain":
           # Use LangChain loader, optionally chunked
           loader = LangChainPDFLoader(pdf_path, self.parser_config, chunk_size, chunk_overlap)
           if chunk_documents:
               return loader.load_and_split()
           else:
               return loader.load()
       elif output_format == "text":
           # Return combined plain text only
           parser = CustomPDFParser(**self.parser_config)
           parsed_data = parser.parse_pdf(pdf_path)
           return parsed_data.get("full_text", "")
       else:
           raise ValueError(f"Unknown output_format: {output_format}")

Étape 4: Testez l’analyseur (exemple.py)

Testons l’analyseur comme suit:

import os
from pathlib import Path
def main():
   print("👋 Welcome to the Custom PDF Parser!")
   print("What would you like to do?")
   print("1. View full parsed raw data")
   print("2. Extract full plain text")
   print("3. Get LangChain documents (no chunking)")
   print("4. Get LangChain documents (with chunking)")
   print("5. Show document metadata")
   print("6. Show per-page metadata")
   print("7. Show cleaned page text (header/footer removed)")
   print("8. Show extracted image metadata")
   choice = input("Enter the number of your choice: ").strip()
   if choice not in {'1', '2', '3', '4', '5', '6', '7', '8'}:
       print("❌ Invalid option.")
       return
   file_path = input("Enter the path to your PDF file: ").strip()
   if not Path(file_path).exists():
       print("❌ File not found.")
       return
   # Initialize pipeline
   pipeline = PDFProcessingPipeline({
       "preserve_layout": False,
       "remove_headers_footers": True,
       "extract_images": True,
       "min_text_length": 20
   })
   # Raw data is needed for most options
   parsed = pipeline.process_single_pdf(file_path, output_format="raw")
   if choice == '1':
       print("nFull Raw Parsed Output:")
       for k, v in parsed.items():
           print(f"{k}: {str(v)(:300)}...")
   elif choice == '2':
       print("nFull Cleaned Text (truncated preview):")
       print("Previewing the first 1000 characters:n"+parsed("full_text")(:1000), "...")
   elif choice == '3':
       docs = pipeline.process_single_pdf(file_path, output_format="langchain", chunk_documents=False)
       print(f"nLangChain Documents: {len(docs)}")
       print("Previewing the first 500 characters:n", docs(0).page_content(:500), "...")
   elif choice == '4':
       docs = pipeline.process_single_pdf(file_path, output_format="langchain", chunk_documents=True)
       print(f"nLangChain Chunks: {len(docs)}")
       print("Sample chunk content (first 500 chars):")
       print(docs(0).page_content(:500), "...")
   elif choice == '5':
       print("nDocument Metadata:")
       for key, value in parsed("document_metadata").items():
           print(f"{key}: {value}")
   elif choice == '6':
       print("nPer-page Metadata:")
       for i, page in enumerate(parsed("pages")):
           print(f"Page {i+1}: {page('metadata')}")
   elif choice == '7':
       print("nCleaned Text After Header/Footer Removal.")
       print("Showing the first 3 pages and first 500 characters of the text from each page.")
       for i, page in enumerate(parsed("pages")(:3)):  # First 3 pages
           print(f"n--- Page {i+1} ---")
           print(page("text")(:500), "...")
   elif choice == '8':
       print("nExtracted Image Metadata (if available):")
       found = False
       for i, page in enumerate(parsed("pages")):
           images = page("metadata").get("images", ())
           if images:
               found = True
               print(f"n--- Page {i+1} ---")
               for img in images:
                   print(img)
       if not found:
           print("No image metadata found.")
if __name__ == "__main__":
   main()

Exécutez ceci et vous serez invité à saisir le choix du choix et le chemin du PDF. Entrez cela. Le PDF que j’utilise est accessible au public, et vous pouvez le télécharger en utilisant le lien.

👋 Welcome to the Custom PDF Parser!
What would you like to do?
1. View full parsed raw data
2. Extract full plain text
3. Get LangChain documents (no chunking)
4. Get LangChain documents (with chunking)
5. Show document metadata
6. Show per-page metadata
7. Show cleaned page text (header/footer removed)
8. Show extracted image metadata.
Enter the number of your choice: 5
Enter the path to your PDF file: /content/articles.pdf

Output:
LangChain Chunks: 16
First chunk preview:
San José State University Writing Center
www.sjsu.edu/writingcenter
Written by Ben Aldridge

Articles (a/an/the), Spring 2014.                                                                                   1 of 4
Articles (a/an/the)

There are three articles in the English language: a, an, and the. They are placed before nouns
and show whether a given noun is general or specific.

Examples of Articles

Conclusion

Dans ce guide, vous avez appris à construire un pipeline de traitement PDF flexible et puissant à l’aide d’outils open-source uniquement. Parce qu’il est modulaire, vous pouvez facilement l’étendre, peut-être ajouter une barre de recherche à l’aide de rationalisation, stocker des morceaux dans une base de données vectorielle comme Faish pour des recherches plus intelligentes, ou même le brancher sur un chatbot. Vous n’avez rien à reconstruire, vous connectez simplement la prochaine pièce. Avec cette approche, vous pouvez transformer n’importe quel document en quelque chose que vous pouvez lire, rechercher et comprendre vos termes.

Kanwal Mehreen Kanwal est ingénieur d’apprentissage automatique et écrivain technique avec une profonde passion pour la science des données et l’intersection de l’IA avec la médecine. Elle a co-écrit l’ebook « Maximiser la productivité avec Chatgpt ». En tant que Google Generation Scholar 2022 pour APAC, elle défend la diversité et l’excellence académique. Elle est également reconnue comme une diversité de Teradata dans Tech Scholar, le boursier de recherche Mitacs Globalink et le savant de Harvard WECODE. Kanwal est un ardent défenseur du changement, après avoir fondé des femmes pour autonomiser les femmes dans les champs STEM.



Source link

Related post