pdf2txt_parser_converter / parser_sevenof9_v1_52_en.py
kalle07's picture
with GUI - only windows, import libs if necessary
da51eb3 verified
import os # OS module for interacting with the operating system (file management, etc.)
import re
import sys # Provides access to system-specific parameters and functions
import tkinter as tk # GUI module for creating desktop applications
from tkinter import filedialog, messagebox # Additional tkinter components for file dialogs and message boxes
import subprocess # Module to run system commands
import threading # Threading module to run tasks concurrently
import json # JSON module for working with JSON data
import logging # Logging module for tracking events and errors
import pdfplumber # Library for extracting text and tables from PDFs
#from pdfplumber.utils import get_bbox_overlap, obj_to_bbox # Helper functions from pdfplumber for working with bounding boxes
from pdfplumber.utils.exceptions import PdfminerException # Exception related to PDF processing
from joblib import delayed, cpu_count, parallel_backend, Parallel # Joblib for parallel processing and optimization
import customtkinter as ctk
import tkinter.font as tkfont
# ========================
# Parser Configuration
# ========================
# Function to suppress PDFMiner logging, reducing verbosity
def suppress_pdfminer_logging():
for logger_name in [
#"pdfminer", # Various pdfminer modules to suppress logging from
#"pdfminer.pdfparser",
#"pdfminer.pdfdocument",
"pdfminer.pdfpage",
#"pdfminer.converter",
#"pdfminer.layout",
#"pdfminer.cmapdb",
#"pdfminer.utils"
]:
logging.getLogger(logger_name).setLevel(logging.ERROR) # Set logging level to ERROR to suppress lower levels
PARALLEL_THRESHOLD = 16 # Number of pages to use for deciding between serial or parallel processing
TEXT_EXTRACTION_SETTINGS = {
"x_tolerance": 1.5, # Horizontal tolerance for text extraction
"y_tolerance": 2.5, # Vertical tolerance for text extraction
"char_dir": "ltr",
"keep_blank_chars": False, # Option to retain blank characters in the extracted text
"use_text_flow": True, # Option to use text flow for better structure
}
# Regex zur Entfernung unerwünschter Zeichen:
EUROPEAN_PRINTABLES_PATTERN = re.compile(r"[^\n\r\t \w\u0000-\uFFFF]")
# EUROPEAN_PRINTABLES_PATTERN = re.compile(r"[^ \t\n\r0-9a-zA-ZäöüÄÖÜßÀ-ÿ.,;:!?(){}\[\]„“‚‘\"'´`\-_/+*=%^~|&§#@€$£<>\\]")
CID_PATTERN = re.compile(r"\(cid:\d+\)")
# Function to clean up text by removing unwanted hyphenations and newlines
def clean_cell_text(text):
if not isinstance(text, str):
return ""
# Entferne Silbentrennzeichen am Zeilenende
text = text.replace("-\n", "")
text = text.replace("\n", " ")
text = CID_PATTERN.sub("", text)
# Entferne alle Zeichen, die nicht zu den definierten druckbaren Zeichen gehören
cleaned_text = EUROPEAN_PRINTABLES_PATTERN.sub("", text)
return cleaned_text
# Function to safely clean and join row cell data
def safe_join(row):
return [clean_cell_text(str(cell)) if cell is not None else "" for cell in row] # Clean each cell in the row, or return empty if None
# Function to clamp bounding box coordinates within page boundaries
def clamp_bbox(bbox, page_width, page_height, precision=3):
x0, top, x1, bottom = bbox
x0 = max(0, min(x0, page_width))
x1 = max(0, min(x1, page_width))
top = max(0, min(top, page_height))
bottom = max(0, min(bottom, page_height))
# Hier runden
return (
round(x0, precision),
round(top, precision),
round(x1, precision),
round(bottom, precision)
)
# Hauptfunktion zur Verarbeitung einer einzelnen PDF-Seite
def process_page(args):
# suppress_pdfminer_logging()
try:
page_number, pdf_path, TEXT_EXTRACTION_SETTINGS = args
with pdfplumber.open(pdf_path) as pdf:
page = pdf.pages[page_number]
output = f"\n\nPage {page_number + 1}\n"
width, height = page.width, page.height
# Seitenränder (5 %) abschneiden
margin_x = width * 0.05
margin_y = height * 0.05
content_bbox = (margin_x, margin_y, width - margin_x, height - margin_y)
cropped_page = page.crop(content_bbox)
# 1. Bounding Boxes aus find_tables(für späteres ausschließen der fließtexterkennung)
table_bboxes = []
for table in cropped_page.find_tables():
bbox = clamp_bbox(table.bbox, width, height)
# if cropped_page.crop(bbox).chars:
# table_bboxes.append(bbox)
cropped_chars = cropped_page.crop(bbox).chars
valid_chars = [
c for c in cropped_chars
if not EUROPEAN_PRINTABLES_PATTERN.search(c["text"])
]
if valid_chars:
table_bboxes.append(bbox)
# 2. Tabelleninhalt unabhängig extrahieren
table_json_outputs = []
for table_data_raw in cropped_page.extract_tables({"text_x_tolerance": 1.5}):
if table_data_raw and len(table_data_raw) >= 1:
# Bereinige bereits hier alle Zellen der rohen Tabelle
table_data = [[clean_cell_text(cell) for cell in row] for row in table_data_raw]
headers = table_data[0]
rows = table_data[1:]
json_table = [dict(zip(headers, row)) for row in rows]
table_json_outputs.append(json.dumps(json_table, indent=1, ensure_ascii=False))
# fließtexterkennunf ohne tabellenbereiche
words_outside_tables = [
word for word in cropped_page.extract_words(**TEXT_EXTRACTION_SETTINGS)
if not any(
bbox[0] <= float(word['x0']) <= bbox[2] and
bbox[1] <= float(word['top']) <= bbox[3]
for bbox in table_bboxes
) and not EUROPEAN_PRINTABLES_PATTERN.search(word['text']) # <== Filter auf gültige Zeichen
]
# alle buchstaben je seite
characters = [
c for c in cropped_page.chars
if not any(
bbox[0] <= float(c['x0']) <= bbox[2] and
bbox[1] <= float(c['top']) <= bbox[3]
for bbox in table_bboxes
) and not EUROPEAN_PRINTABLES_PATTERN.search(c['text']) # <== Filter auf gültige Zeichen
]
# durchschnittliche schriftgröße je seite
letter_chars = [c for c in characters if c.get('text', '').isalpha()]
average_font_size = (
sum(float(c.get('size', 0)) for c in letter_chars) / len(letter_chars)
if letter_chars else 0
)
# wann ist ein wort bold oder groß-geschrieben, um das label wichtig oder kapitel zu erhalten
def classify_word(word, is_first_word_in_line):
"""Klassifiziere ein Wort individuell mit Stil, nur beim ersten Wort der Zeile."""
word_top = float(word['top'])
word_mid = (float(word['x0']) + float(word['x1'])) / 2
# Alle Zeichen in Zeilenhöhe
line_chars = sorted([
c for c in characters
if abs(c['top'] - word_top) < 2
], key=lambda c: c['x0'])
# Funktion: Finde mind. 3 aufeinanderfolgende fettgedruckte Zeichen
def has_consecutive_bold(chars):
count = 0
for c in chars:
if "bold" in c.get("fontname", "").lower() and float(c.get("size", 0)) >= average_font_size:
count += 1
if count >= 3:
return True
else:
count = 0
return False
# Funktion: Finde mind. 3 aufeinanderfolgende große Buchstaben
def has_consecutive_large_alpha(chars):
count = 0
for c in chars:
if c.get('text', '').isalpha() and float(c.get("size", 0)) >= average_font_size * 1.16:
count += 1
if count >= 3:
return True
else:
count = 0
return False
prefix = ""
if is_first_word_in_line:
if has_consecutive_bold(line_chars):
prefix += "important: "
if has_consecutive_large_alpha(line_chars):
prefix += "chapter: "
return prefix + word['text']
# Gruppierung in Zeilen mit Wortbasierter Analyse
current_y = None
line = []
text_content = ""
for word in words_outside_tables:
word_y = float(word['top'])
if current_y is None or abs(word_y - current_y) > 10:
if line:
text_content += " ".join(line).strip() + "\n"
current_y = word_y
line = [classify_word(word, is_first_word_in_line=True)]
else:
line.append(classify_word(word, is_first_word_in_line=False))
if line:
text_content += " ".join(line).strip() + "\n"
output += text_content.strip() + "\n"
# Add table JSON outputs to the page output
for idx, table in enumerate(table_json_outputs, start=1):
output += f'"table {idx}":\n{table}\n'
return page_number, output # Return the processed page number and output content
# except Exception as e:
# return args[0], f"[ERROR] Page {args[0]+1} ({args[1]}): {str(e)}" # Return an error message if an exception occurs
except Exception as e:
error_msg = str(e)
if "Cannot set gray non-stroke color because" in error_msg and "invalid float value" in error_msg:
friendly_msg = f"[ERROR] Seite {args[0]+1} ({args[1]}): Ungültiger Farbwert in PDF-Inhalt erkannt (möglicherweise beschädigte Farbdefinition). Verarbeitung nicht möglich."
return args[0], friendly_msg
else:
return args[0], f"[ERROR] Seite {args[0]+1} ({args[1]}): {error_msg}"
# Function to process the entire PDF document
def process_pdf(pdf_path):
suppress_pdfminer_logging() # Suppress unnecessary logging
try:
if not os.path.exists(pdf_path): # Check if the file exists
return f"[ERROR] File not found: {pdf_path}" # Return error message if file does not exist
print(f"[INFO] Starting processing: {pdf_path}") # Log the start of processing
try:
with pdfplumber.open(pdf_path) as pdf: # Open the PDF using pdfplumber
num_pages = len(pdf.pages) # Get the number of pages in the PDF
except PdfminerException as e:
return f"[ERROR] Cannot open PDF: {pdf_path}{str(e)}" # Return error if the PDF cannot be opened
except Exception as e:
return f"[ERROR] General error opening PDF: {pdf_path}{str(e)}" # Return general error if any exception occurs
pages = [(i, pdf_path, TEXT_EXTRACTION_SETTINGS) for i in range(num_pages)] # Prepare the pages for processing
try:
results = run_serial(pages) if num_pages <= PARALLEL_THRESHOLD else run_parallel(pages) # Run serial or parallel processing
except (EOFError, BrokenPipeError, KeyboardInterrupt):
return "[INFO] Processing was interrupted." # Handle interruptions during processing
sorted_results = sorted(results, key=lambda x: x[0]) # Sort results by page number
final_output = "\n".join(text for _, text in sorted_results) # Combine all page results into a single string
base_name = os.path.splitext(os.path.basename(pdf_path))[0] # Get the base name of the PDF file
output_dir = os.path.dirname(pdf_path) # Get the directory of the PDF file
output_path = os.path.join(output_dir, f"{base_name}.txt") # Generate the output file path
with open(output_path, "w", encoding="utf-8", errors="ignore") as f: # Open the output file for writing
f.write(final_output) # Write the final output to the file
print(f"[INFO] Processing complete: {output_path}") # Log the successful processing completion
return "complete"
except (EOFError, BrokenPipeError, KeyboardInterrupt):
return "[INFO] Processing interrupted by user." # Handle user interruptions
except Exception as e:
return f"[ERROR] Unexpected error with '{pdf_path}': {str(e)}" # Handle unexpected errors during processing
# Function to run the PDF processing serially (one page at a time)
def run_serial(pages):
return [process_page(args) for args in pages] # Process each page in sequence
# Function to run the PDF processing in parallel (across multiple cores)
def run_parallel(pages):
available_cores = max(1, cpu_count() - 2)
num_cores = min(available_cores, len(pages))
print(f"Starting parallel processing with {num_cores} cores...")
with parallel_backend('loky'): # 'loky' ist der Standard-Backend bei joblib
return Parallel(n_jobs=num_cores)(
delayed(process_page)(args) for args in pages
)
# Main function to process a list of PDFs
def process_pdfs_main():
suppress_pdfminer_logging() # Suppress unnecessary logging
pdf_files = sys.argv[1:] # Get PDF file paths from command-line arguments
if not pdf_files: # Check if any PDFs are provided
print("No PDF files provided.") # Log message if no PDFs are provided
return
small_pdfs = [] # List to store small PDFs (less than the parallel threshold)
large_pdfs = [] # List to store large PDFs (greater than the parallel threshold)
# Categorize PDFs into small and large based on the number of pages
for path in pdf_files:
if not os.path.exists(path): # Check if the file exists
print(f"File not found: {path}") # Log error if file does not exist
continue
try:
with pdfplumber.open(path) as pdf: # Open the PDF
if len(pdf.pages) <= PARALLEL_THRESHOLD: # If the PDF has fewer pages than the threshold
small_pdfs.append(path) # Add to small PDFs list
else:
large_pdfs.append(path) # Add to large PDFs list
except PdfminerException:
print(f"[ERROR] Password-protected PDF skipped: {path}") # Log if the PDF is password-protected
except Exception as e:
print(f"[ERROR] Error opening {path}: {str(e)}") # Log any other errors when opening the PDF
# Process small PDFs in parallel (each on one core)
if small_pdfs:
available_cores = max(1, cpu_count() - 2) # Determine the number of available cores
num_cores = min(available_cores, len(small_pdfs)) # Use the lesser of available cores or small PDFs count
print(f"\n[Phase 1] Starting parallel processing of small PDFs with {num_cores} cores, 2 leaving for system processes...") # Log processing start
results = Parallel(n_jobs=num_cores)( # Run parallel processing for small PDFs
delayed(process_pdf)(path) for path in small_pdfs
)
for r in results:
print(r) # Print the results for each small PDF
# Process large PDFs one by one (in serial on all cores)
for path in large_pdfs:
print(f"\n[Phase 2] Processing large PDF: {os.path.basename(path)}") # Log processing of large PDF
print(process_pdf(path)) # Process the large PDF
# GUI
ctk.set_appearance_mode("System")
ctk.set_default_color_theme("dark-blue")
class FileManager:
def __init__(self, master):
self.master = master
self.master.protocol("WM_DELETE_WINDOW", self.on_close)
self.master.title("Parser-Sevenof9")
self.master.geometry("1000x800+200+100")
self.master.minsize(1000, 800)
custom_font = tkfont.Font(family="Courier New", size=14)
self.files = []
self.last_selected_index = None
self.parser_process = None
self.master.grid_rowconfigure(1, weight=0)
self.master.grid_columnconfigure(0, weight=1)
# Label: "Selected PDF files"
self.label = ctk.CTkLabel(master, text="Selected PDF files: (right mouse, you can copy path or open PDF)", height=30)
self.label.grid(row=0, column=0, sticky="nw", padx=10, pady=(10, 0))
# Listbox Frame
listbox_frame = ctk.CTkFrame(master, height=200)
listbox_frame.grid(row=1, column=0, sticky="nsew", padx=10)
listbox_frame.grid_propagate(False)
listbox_frame.grid_rowconfigure(0, weight=1)
listbox_frame.grid_columnconfigure(0, weight=1)
self.listbox = tk.Listbox(listbox_frame, selectmode=tk.MULTIPLE, font=custom_font,)
scrollbar_listbox = tk.Scrollbar(listbox_frame, command=self.listbox.yview)
self.listbox.config(yscrollcommand=scrollbar_listbox.set)
self.listbox.grid(row=0, column=0, sticky="nsew")
scrollbar_listbox.grid(row=0, column=1, sticky="ns")
# Context Menu
self.context_menu = tk.Menu(master, tearoff=0)
self.context_menu.add_command(label="Remove selected", command=self.remove_file)
self.context_menu.add_separator()
self.context_menu.add_command(label="Copy file location", command=self.copy_file_location)
self.context_menu.add_command(label="Open in default PDF app", command=self.open_file_in_default_app)
self.listbox.bind("<Button-3>", self.show_context_menu)
self.listbox.bind("<<ListboxSelect>>", self.show_text_file)
self.listbox.bind("<Button-1>", self.on_listbox_click)
self.listbox.bind("<Shift-Button-1>", self.on_listbox_shift_click)
# Button Frame
button_frame = ctk.CTkFrame(master, height=40)
button_frame.grid(row=2, column=0, sticky="nsew", padx=10, pady=5)
button_frame.grid_propagate(False)
button_frame.grid_columnconfigure((0, 1, 2, 3, 4, 5), weight=1)
ctk.CTkButton(button_frame, text="Add Folder", command=self.add_folder).grid(row=0, column=0, padx=5, pady=5)
ctk.CTkButton(button_frame, text="Select Files", command=self.add_file).grid(row=0, column=1, padx=5, pady=5)
ctk.CTkButton(button_frame, text="Remove Selected", command=self.remove_file).grid(row=0, column=2, padx=5, pady=5)
ctk.CTkButton(button_frame, text="Remove All", command=self.remove_all).grid(row=0, column=3, padx=5, pady=5)
ctk.CTkButton(button_frame, text="Stop", command=self.stop_parser, fg_color="darkred", hover_color="red").grid(row=0, column=4, padx=5, pady=5)
ctk.CTkButton(button_frame, text="Start Parser", command=self.start_parser, fg_color="darkgreen", hover_color="green").grid(row=0, column=5, padx=5, pady=5)
# Label: "Text Frame"
self.progress_label = ctk.CTkLabel(master, text="Text Frame: (select a PDF, you can copy text parts)", height=30)
self.progress_label.grid(row=3, column=0, sticky="nw", padx=10)
# Text Frame
text_frame = ctk.CTkFrame(master, height=250)
text_frame.grid(row=4, column=0, sticky="nsew", padx=10, pady=5)
text_frame.grid_propagate(False)
text_frame.grid_rowconfigure(0, weight=1)
text_frame.grid_columnconfigure(0, weight=1)
self.text_widget = tk.Text(text_frame, wrap=tk.WORD, font=custom_font,)
scrollbar_text = tk.Scrollbar(text_frame, command=self.text_widget.yview)
self.text_widget.config(yscrollcommand=scrollbar_text.set)
self.text_widget.grid(row=0, column=0, sticky="nsew")
scrollbar_text.grid(row=0, column=1, sticky="ns")
# Label: "Progress"
self.progress_label = ctk.CTkLabel(master, text="Progress: (Error and success messages are not always correct)", height=30)
self.progress_label.grid(row=5, column=0, sticky="nw", padx=10)
# Progress Frame
progress_frame = ctk.CTkFrame(master, height=160)
progress_frame.grid(row=6, column=0, sticky="nsew", padx=10, pady=(0, 10))
progress_frame.grid_propagate(False)
progress_frame.grid_rowconfigure(0, weight=1)
progress_frame.grid_columnconfigure(0, weight=1)
self.progress_text = tk.Text(progress_frame, state=tk.DISABLED)
scrollbar_progress = tk.Scrollbar(progress_frame, command=self.progress_text.yview)
self.progress_text.config(yscrollcommand=scrollbar_progress.set)
self.progress_text.grid(row=0, column=0, sticky="nsew")
scrollbar_progress.grid(row=0, column=1, sticky="ns")
def on_close(self):
if self.parser_process:
self.stop_parser()
# ggf. geplante after-Aufrufe canceln
if hasattr(self, 'after_id'):
self.master.after_cancel(self.after_id)
self.master.destroy()
def on_listbox_click(self, event):
# Handle single left-click selection; clear previous selection
index = self.listbox.nearest(event.y)
self.listbox.selection_clear(0, tk.END)
self.listbox.selection_set(index)
self.last_selected_index = index
self.show_text_file(None)
return "break" # Prevent default event propagation
def on_listbox_shift_click(self, event):
# Handle shift-click for range selection
index = self.listbox.nearest(event.y)
if self.last_selected_index is None:
self.last_selected_index = index
start, end = sorted((self.last_selected_index, index))
self.listbox.selection_clear(0, tk.END)
for i in range(start, end + 1):
self.listbox.selection_set(i)
return "break"
def show_context_menu(self, event):
# Show right-click context menu if any item is selected
if self.listbox.curselection():
self.context_menu.tk_popup(event.x_root, event.y_root)
def add_folder(self):
# Add all PDFs from a selected folder
folder = filedialog.askdirectory(title="Select Folder")
if not folder:
return
for root, _, files in os.walk(folder):
for file in files:
if file.lower().endswith(".pdf"):
path = os.path.normpath(os.path.join(root, file))
if path not in self.files:
self.files.append(path)
self.listbox.insert(tk.END, path)
def add_file(self):
# Add selected individual PDF files
paths = filedialog.askopenfilenames(title="Select PDF Files", filetypes=[("PDF Files", "*.pdf")])
for path in paths:
path = os.path.normpath(path)
if path not in self.files:
self.files.append(path)
self.listbox.insert(tk.END, path)
def remove_file(self):
# Remove selected files from list and internal storage
selection = self.listbox.curselection()
if not selection:
messagebox.showwarning("Notice", "Please select an entry to remove.")
return
for index in reversed(selection): # Reverse to avoid index shifting
self.listbox.delete(index)
del self.files[index]
self.text_widget.delete(1.0, tk.END)
def copy_file_location(self):
selection = self.listbox.curselection()
if not selection:
return
index = selection[0]
path = self.files[index]
self.master.clipboard_clear()
self.master.clipboard_append(path)
self.master.update() # erforderlich, um Inhalt im Clipboard zu halten
def open_file_in_default_app(self):
selection = self.listbox.curselection()
if not selection:
return
index = selection[0]
path = self.files[index]
if os.path.exists(path):
try:
os.startfile(path)
except Exception as e:
messagebox.showerror("Error", f"Cannot open file:\n{e}")
else:
messagebox.showwarning("File not found", "The selected file could not be found.")
def remove_all(self):
# Remove all files from the list
self.listbox.delete(0, tk.END)
self.files.clear()
self.text_widget.delete(1.0, tk.END)
def start_parser(self):
# Validate input and launch parser in separate thread
if not self.files:
messagebox.showinfo("No Files", "Please select at least one file.")
return
self.progress_text.config(state=tk.NORMAL)
self.progress_text.delete(1.0, tk.END)
self.progress_text.insert(tk.END, "Starting parser...\n")
self.progress_text.config(state=tk.DISABLED)
# Launch parsing in background to avoid UI freeze
thread = threading.Thread(target=self.run_parser)
thread.start()
def stop_parser(self):
# Terminate running parser process if active
if self.parser_process and self.parser_process.poll() is None:
self.parser_process.terminate()
self.append_progress_text("Parser process was stopped.\n")
else:
self.append_progress_text("No active parser process to stop.\n")
def run_parser(self):
# Internal method to run the external parser script
try:
self.parser_process = subprocess.Popen(
[sys.executable, __file__] + self.files,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
encoding='utf-8',
errors='ignore',
bufsize=4096
)
for line in self.parser_process.stdout:
self.append_progress_text(line)
self.parser_process.stdout.close()
self.parser_process.wait()
if self.parser_process.returncode == 0:
self.append_progress_text("\nParser finished successfully.\n")
self.show_messagebox_threadsafe("Parser Done", "The parser was executed successfully.")
else:
self.append_progress_text("\nError while running the parser.\n")
self.show_messagebox_threadsafe("Error", "Error while running the parser.")
except Exception as e:
self.append_progress_text(f"Error: {e}\n")
self.show_messagebox_threadsafe("Error", f"Error during execution:\n{e}")
finally:
self.parser_process = None
def append_progress_text(self, text):
# Thread-safe method to append text to the progress view
self.progress_text.after(0, lambda: self._insert_text(text))
def _insert_text(self, text):
# Append text and scroll to bottom
self.progress_text.config(state=tk.NORMAL)
self.progress_text.insert(tk.END, text)
self.progress_text.see(tk.END)
self.progress_text.config(state=tk.DISABLED)
def show_messagebox_threadsafe(self, title, message):
# Display a messagebox from a background thread
self.master.after(0, lambda: messagebox.showinfo(title, message))
def show_text_file(self, event):
# Load and show the content of the corresponding .txt file (if available)
selection = self.listbox.curselection()
if not selection:
return
index = selection[0]
path = self.files[index]
txt_path = os.path.splitext(path)[0] + ".txt"
self.text_widget.delete(1.0, tk.END)
if os.path.exists(txt_path):
try:
with open(txt_path, "r", encoding="utf-8", errors="ignore") as f:
self.text_widget.insert(tk.END, f.read())
except Exception as e:
self.text_widget.insert(tk.END, f"Error loading text file:\n{e}")
else:
self.text_widget.insert(tk.END, "[No corresponding .txt file found]")
def main():
if len(sys.argv) > 1:
process_pdfs_main()
else:
launch_gui()
def launch_gui():
root = ctk.CTk()
app = FileManager(root)
root.mainloop()
# MAIN
if __name__ == "__main__":
main()