155 lines
6.8 KiB
Python
155 lines
6.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Prepare ordered scanner pages and publish an optionally searchable PDF atomically."""
|
|
import argparse
|
|
import json
|
|
import math
|
|
from pathlib import Path
|
|
import re
|
|
import signal
|
|
import subprocess
|
|
import tempfile
|
|
|
|
from PIL import Image, PdfParser
|
|
|
|
|
|
def run_ocr_command(arguments):
|
|
process = subprocess.Popen(arguments, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
|
|
try:
|
|
stdout, stderr = process.communicate()
|
|
except BaseException:
|
|
process.kill()
|
|
process.wait()
|
|
raise
|
|
if process.returncode != 0:
|
|
raise RuntimeError("OCR fehlgeschlagen: " + stderr.strip())
|
|
return stdout
|
|
|
|
|
|
def check_ocr(language):
|
|
if not re.fullmatch(r"[a-zA-Z0-9_]+(?:\+[a-zA-Z0-9_]+)*", language):
|
|
raise ValueError("Ungültige OCR-Sprache")
|
|
try:
|
|
languages = run_ocr_command(["tesseract", "--list-langs"]).splitlines()
|
|
except FileNotFoundError as error:
|
|
raise RuntimeError("Tesseract fehlt. Bitte tesseract-ocr und die Sprachpakete tesseract-ocr-deu/tesseract-ocr-eng installieren.") from error
|
|
missing = set(language.split("+")) - {line.strip() for line in languages}
|
|
if missing:
|
|
raise RuntimeError("OCR-Sprachpakete fehlen: " + ", ".join(sorted(missing)))
|
|
|
|
|
|
def create_document(inputs, output, resolution, use_postprocess=False, profile="document", strict=False,
|
|
ocr=False, ocr_language="deu+eng", background_cleanup=False):
|
|
if not inputs:
|
|
raise ValueError("Keine Seiten vorhanden")
|
|
if not math.isfinite(resolution) or resolution <= 0:
|
|
raise ValueError("Ungültige Scan-Auflösung")
|
|
output = Path(output)
|
|
file_format = {".pdf": "PDF", ".png": "PNG", ".tiff": "TIFF"}.get(output.suffix.lower())
|
|
if not file_format:
|
|
raise ValueError("Unbekanntes Ausgabeformat")
|
|
if (len(inputs) != 1 or ocr) and file_format != "PDF":
|
|
raise ValueError("Mehrseitige Scans und OCR benötigen PDF als Ausgabeformat")
|
|
if ocr:
|
|
check_ocr(ocr_language)
|
|
|
|
warnings = []
|
|
processor = None
|
|
if use_postprocess:
|
|
try:
|
|
from opencv_postprocess import postprocess
|
|
processor = postprocess
|
|
except ImportError:
|
|
if strict:
|
|
raise
|
|
warnings.append("OpenCV fehlt; alle Seiten wurden ohne Bildkorrektur übernommen.")
|
|
if background_cleanup:
|
|
from scan_background import clean_scan_background
|
|
|
|
partial = output.with_name(output.name + ".partial")
|
|
try:
|
|
with tempfile.TemporaryDirectory(prefix="prepared-", dir=output.parent) as directory:
|
|
prepared = []
|
|
for index, input_path in enumerate(inputs):
|
|
image_path = Path(input_path)
|
|
corrected = Path(directory) / f"corrected-{index:06}.png"
|
|
if processor:
|
|
try:
|
|
processor(image_path, corrected, profile)
|
|
image_path = corrected
|
|
except Exception:
|
|
if strict:
|
|
raise
|
|
warnings.append(f"Seite {index + 1}: Bildkorrektur fehlgeschlagen; Rohscan übernommen.")
|
|
if background_cleanup:
|
|
cleaned = Path(directory) / f"cleaned-{index:06}.png"
|
|
clean_scan_background(image_path, cleaned)
|
|
image_path = cleaned
|
|
|
|
# Only one decoded page is kept in memory, even for duplex stacks.
|
|
with Image.open(image_path) as image:
|
|
with image.convert("RGB") as rgb:
|
|
if ocr:
|
|
page = Path(directory).resolve() / f"page-{index:06}.png"
|
|
rgb.save(page, "PNG", dpi=(resolution, resolution))
|
|
prepared.append(page)
|
|
elif file_format == "PDF":
|
|
rgb.save(partial, "PDF", resolution=resolution, quality=90, append=index > 0)
|
|
else:
|
|
rgb.save(partial, file_format, dpi=(resolution, resolution))
|
|
corrected.unlink(missing_ok=True)
|
|
if background_cleanup:
|
|
cleaned.unlink(missing_ok=True)
|
|
if ocr:
|
|
page_list = Path(directory) / "pages.txt"
|
|
if any("\n" in str(page) or "\r" in str(page) for page in prepared):
|
|
raise ValueError("Das Scan-Arbeitsverzeichnis darf keinen Zeilenumbruch enthalten")
|
|
page_list.write_text("\n".join(map(str, prepared)) + "\n", encoding="utf8")
|
|
prefix = Path(directory) / "searchable"
|
|
run_ocr_command(["tesseract", str(page_list), str(prefix), "-l", ocr_language,
|
|
"--dpi", str(round(resolution)), "pdf"])
|
|
searchable = prefix.with_suffix(".pdf")
|
|
with PdfParser.PdfParser(filename=str(searchable)) as pdf:
|
|
if len(pdf.pages) != len(inputs):
|
|
raise RuntimeError("OCR-PDF ist unvollständig; Rohscans bleiben erhalten")
|
|
searchable.replace(partial)
|
|
partial.replace(output)
|
|
finally:
|
|
partial.unlink(missing_ok=True)
|
|
return {"pageCount": len(inputs), "warnings": warnings}
|
|
|
|
|
|
def interrupted(*_):
|
|
raise SystemExit("Scan-Verarbeitung abgebrochen")
|
|
|
|
|
|
def main():
|
|
# Let finally blocks stop Tesseract and remove incomplete PDFs on agent timeout.
|
|
signal.signal(signal.SIGTERM, interrupted)
|
|
parser = argparse.ArgumentParser(description="Scan-Seiten zu einer Datei zusammenführen")
|
|
parser.add_argument("--inputs", nargs="+")
|
|
parser.add_argument("--output")
|
|
parser.add_argument("--resolution", type=float, default=300)
|
|
parser.add_argument("--profile", choices=["document", "receipt", "raw"], default="document")
|
|
parser.add_argument("--postprocess", action="store_true")
|
|
parser.add_argument("--strict", action="store_true")
|
|
parser.add_argument("--background-cleanup", action="store_true")
|
|
parser.add_argument("--ocr", action="store_true")
|
|
parser.add_argument("--ocr-language", default="deu+eng")
|
|
parser.add_argument("--check-runtime", action="store_true")
|
|
args = parser.parse_args()
|
|
if args.check_runtime:
|
|
if args.background_cleanup or (args.postprocess and args.strict):
|
|
import cv2, numpy
|
|
if args.ocr:
|
|
check_ocr(args.ocr_language)
|
|
print(json.dumps({"ready": True}))
|
|
return
|
|
if not args.inputs or not args.output:
|
|
parser.error("--inputs und --output sind erforderlich")
|
|
print(json.dumps(create_document(args.inputs, args.output, args.resolution, args.postprocess, args.profile,
|
|
args.strict, args.ocr, args.ocr_language, args.background_cleanup)))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|