import csv
import logging
import os
import pathlib
import sys
import tempfile
import time
import aspose.pdf as apdf
from dotenv import dotenv_values
MAX_PDF_BYTES = 1_000_000_000 # 1 GB; reject larger inputs before loading them.
REPORT_COLUMNS = ("input_file", "output_file", "status", "processing_seconds", "error")
class SkipPDF(Exception):
"""The input cannot be processed under the batch's safety policy."""
def optimize(input_file: str, output_file: str) -> None:
try:
document = apdf.Document(input_file)
except RuntimeError as exc:
# Aspose's Python bridge exposes .NET exceptions as RuntimeError.
if "InvalidPasswordException" in str(exc):
raise SkipPDF("Password-protected PDF; no password supplied") from exc
if "InvalidPdfFileFormatException" in str(exc):
raise ValueError(f"Invalid or corrupted PDF: {exc}") from exc
raise
try:
if document.is_encrypted:
raise SkipPDF("Encrypted PDF; password-protected files are skipped")
document.optimize()
document.save(output_file)
finally:
del document
def configure_logging(logs_folder: pathlib.Path) -> None:
logs_folder.mkdir(parents=True, exist_ok=True)
logging.basicConfig(
filename=str(logs_folder / "processing.log"),
encoding="utf-8",
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
)
def find_pdf_files(input_folder: pathlib.Path) -> list[pathlib.Path]:
return sorted(
path
for path in input_folder.iterdir()
if path.is_file() and path.suffix.lower() == ".pdf"
)
def process_file(input_file: pathlib.Path, output_file: pathlib.Path) -> tuple[str, float, str]:
"""Optimize one PDF, recording failures without interrupting the batch."""
started_at = time.perf_counter()
logging.info("Processing %s", input_file.name)
error = ""
try:
if os.path.lexists(output_file):
raise SkipPDF("Output already exists; left unchanged")
if input_file.stat().st_size > MAX_PDF_BYTES:
raise SkipPDF("Input exceeds the 1 GB size limit")
with input_file.open("rb") as source:
if b"%PDF-" not in source.read(1024):
raise ValueError("Invalid PDF: missing PDF header")
# Publish only complete PDFs. Hard-link creation cannot overwrite an
# output created by another process after the existence check.
with tempfile.TemporaryDirectory(prefix=".pdf-", dir=output_file.parent) as work:
temporary_output = pathlib.Path(work) / "optimized.pdf"
optimize(str(input_file), str(temporary_output))
os.link(temporary_output, output_file)
except (SkipPDF, FileExistsError) as exc:
status = "skipped"
error = str(exc)
logging.warning("Skipped %s: %s", input_file.name, error)
except PermissionError as exc:
status = "failed"
error = f"Permission denied reading input or writing output: {exc}"
logging.error("Failed to process %s: %s", input_file.name, error)
except Exception as exc:
status = "failed"
error = str(exc)
logging.exception("Failed to process %s", input_file.name)
else:
status = "success"
logging.info("Successfully processed %s", input_file.name)
return status, time.perf_counter() - started_at, error
def process_files(
input_files: list[pathlib.Path],
output_folder: pathlib.Path,
report_path: pathlib.Path,
) -> tuple[int, int, int]:
"""Write a report and return success, failure, and skipped counts."""
if input_files:
output_folder.mkdir(parents=True, exist_ok=True)
successfully_processed = 0
failed = 0
skipped = 0
logging.info("Files found: %d", len(input_files))
with report_path.open("w", newline="", encoding="utf-8") as report_file:
writer = csv.writer(report_file)
writer.writerow(REPORT_COLUMNS)
report_file.flush()
for pdf_in_file in input_files:
pdf_out_file = output_folder / pdf_in_file.name
status, processing_seconds, error = process_file(pdf_in_file, pdf_out_file)
if status == "success":
successfully_processed += 1
elif status == "skipped":
skipped += 1
else:
failed += 1
writer.writerow(
[
str(pdf_in_file),
str(pdf_out_file),
status,
f"{processing_seconds:.3f}",
error,
]
)
report_file.flush()
return successfully_processed, failed, skipped
def main(
base_folder: pathlib.Path | None = None,
license_path: str | None = None,
) -> int:
started_at = time.perf_counter()
config_path = pathlib.Path(__file__).resolve().parent / ".env"
with config_path.open(encoding="utf-8") as config_file:
config = dotenv_values(stream=config_file)
if base_folder is None:
configured_base_folder = config.get("BASE_FOLDER")
if not configured_base_folder or not configured_base_folder.strip():
raise ValueError("BASE_FOLDER must be set in .env")
base_folder = pathlib.Path(configured_base_folder).expanduser()
if not base_folder.is_absolute():
base_folder = config_path.parent / base_folder
if license_path is None:
configured_license_path = config.get("LICENSE_PATH")
if not configured_license_path or not configured_license_path.strip():
raise ValueError("LICENSE_PATH must be set in .env")
license_file = pathlib.Path(configured_license_path).expanduser()
if not license_file.is_absolute():
license_file = config_path.parent / license_file
license_path = str(license_file)
if not base_folder.is_dir():
raise NotADirectoryError(f"BASE_FOLDER is not a directory: {base_folder}")
if not pathlib.Path(license_path).is_file():
raise FileNotFoundError(f"LICENSE_PATH is not a file: {license_path}")
logs_folder = base_folder / "logs"
report_path = logs_folder / "processing-report.csv"
try:
configure_logging(logs_folder)
logging.info("Processing started")
pdf_in_files = find_pdf_files(base_folder / "input")
if pdf_in_files:
pdf_license = apdf.License()
pdf_license.set_license(license_path)
else:
logging.info("No PDF files found; nothing to process")
print("No PDF files found; nothing to process.")
successfully_processed, failed, skipped = process_files(
pdf_in_files, base_folder / "output", report_path
)
except (OSError, RuntimeError) as exc:
message = f"Batch could not complete: {exc}"
logging.error(message)
print(message, file=sys.stderr)
raise
total_seconds = time.perf_counter() - started_at
summary = (
f"Files found: {len(pdf_in_files)}\n"
f"Successfully processed: {successfully_processed}\n"
f"Failed: {failed}\n"
f"Skipped: {skipped}\n"
f"Total processing time: {total_seconds:.1f} seconds\n"
f"Report: {report_path.relative_to(base_folder).as_posix()}"
)
logging.info("Processing finished\n%s", summary)
print(summary)
return 1 if failed else 0
if __name__ == "__main__":
sys.exit(main())
Comments