Print Speed How to Fix Common Issues and Solutions

#!/usr/bin/env python3
"""
Print Queue Health Monitor
Scans local or remote print queues, identifies stalled jobs,
flags oversized spool files, and reports driver configuration issues.

Requirements: pip install pywin32 (Windows) or cups (Linux/macOS)
Usage: python print_queue_monitor.py [--host REMOTE_HOST] [--threshold-mb 100]
"""

import argparse
import json
import logging
import os
import platform
import subprocess
import sys
from datetime import datetime, timedelta
from dataclasses import dataclass, field, asdict
from typing import List, Optional

# Configure structured logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger("print_monitor")

# Threshold constants
DEFAULT_SIZE_THRESHOLD_MB = 100  # Flag jobs whose spool file exceeds this
DEFAULT_STALL_MINUTES = 30      # Flag jobs older than this threshold


@dataclass
class PrintJob:
    """Represents a single print job extracted from the spooler."""
    job_id: int
    printer_name: str
    document_name: str
    submitted_by: str
    pages: int
    size_bytes: int
    status: str
    submitted_at: Optional[datetime] = None
    is_stalled: bool = False
    is_oversized: bool = False
    notes: List[str] = field(default_factory=list)


def parse_size_to_bytes(size_str: str) -> int:
    """Convert human-readable size strings like '24.5 MB' to bytes."""
    units = {"B": 1, "KB": 1024, "MB": 1024**2, "GB": 1024**3}
    parts = size_str.strip().split()
    if len(parts) < 2:
        return 0
    try:
        value = float(parts[0])
        unit = parts[1].upper()
        return int(value * units.get(unit, 1))
    except (ValueError, IndexError):
        return 0


def get_jobs_windows(host: Optional[str] = None) -> List[PrintJob]:
    """
    Enumerate print jobs on Windows using win32print.
    Requires pywin32: pip install pywin32
    """
    jobs = []
    try:
        import win32print
        import win32con
        import win32api

        # Enumerate all printers visible to this machine
        if host:
            printers = win32print.EnumPrinters(
                win32print.PRINTER_ENUM_CONNECTIONS | win32print.PRINTER_ENUM_LOCAL,
                Name=host
            )
        else:
            printers = win32print.EnumPrinters(win32print.PRINTER_ENUM_LOCAL)

        for flags, description, name, comment in printers:
            try:
                # Open a handle to each printer to query its jobs
                hprinter = win32print.OpenPrinter(name)
                try:
                    # Level 2 gives us JOB_INFO_2 with full metadata
                    job_list = win32print.EnumJobs(hprinter, 0, -1, 2)
                    for j in job_list:
                        # Calculate job age
                        submitted = datetime.fromtimestamp(j['Submitted'])
                        age_minutes = (datetime.now() - submitted).total_seconds() / 60

                        job = PrintJob(
                            job_id=j['JobId'],
                            printer_name=name,
                            document_name=j['pDocument'],
                            submitted_by=j['pUserName'],
                            pages=j.get('TotalPages', 0),
                            size_bytes=j.get('Size', 0) * 1024,  # Size is in KB
                            status=j['Status'],
                            submitted_at=submitted
                        )

                        # Flag stalled jobs
                        if age_minutes > DEFAULT_STALL_MINUTES:
                            job.is_stalled = True
                            job.notes.append(
                                f"Job stalled for {age_minutes:.0f} minutes"
                            )

                        # Flag oversized jobs
                        if job.size_bytes > DEFAULT_SIZE_THRESHOLD_MB * 1024 * 1024:
                            job.is_oversized = True
                            job.notes.append(
                                f"Spool file is {job.size_bytes / 1024 / 1024:.1f} MB"
                            )

                        jobs.append(job)
                finally:
                    win32print.ClosePrinter(hprinter)
            except Exception as e:
                logger.warning(f"Could not query printer '{name}': {e}")
                continue
    except ImportError:
        logger.error("pywin32 not installed. Run: pip install pywin32")
    except Exception as e:
        logger.error(f"Windows print enumeration failed: {e}")

    return jobs


def get_jobs_linux(host: Optional[str] = None) -> List[PrintJob]:
    """
    Enumerate print jobs on Linux/macOS using lpstat.
    Falls back gracefully if CUPS is not available.
    """
    jobs = []
    try:
        cmd = ["lpstat", "-W", "completed", "-o"]
        if host:
            cmd.extend(["-h", host])

        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=15
        )

        if result.returncode != 0:
            # Try active jobs only
            result = subprocess.run(
                ["lpstat", "-o"] + (["-h", host] if host else []),
                capture_output=True,
                text=True,
                timeout=15
            )

        for line in result.stdout.strip().split("\n"):
            if not line.strip():
                continue
            # Parse lpstat output format: printer-name job-id user size date
            parts = line.split()
            if len(parts) >= 5:
                try:
                    jobs.append(PrintJob(
                        job_id=int(parts[2]),
                        printer_name=parts[0],
                        document_name=parts[1] if len(parts) > 1 else "unknown",
                        submitted_by=parts[3] if len(parts) > 3 else "unknown",
                        pages=0,
                        size_bytes=parse_size_to_bytes(parts[4]) if len(parts) > 4 else 0,
                        status="active",
                        submitted_at=datetime.now()
                    ))
                except (ValueError, IndexError):
                    continue
    except FileNotFoundError:
        logger.warning("lpstat not found — this script requires CUPS on Linux.")
    except subprocess.TimeoutExpired:
        logger.error("lpstat timed out after 15 seconds")
    except Exception as e:
        logger.error(f"Linux print enumeration failed: {e}")

    return jobs


def generate_health_report(jobs: List[PrintJob]) -> dict:
    """Generate a summary report from the collected job data."""
    total_jobs = len(jobs)
    stalled = [j for j in jobs if j.is_stalled]
    oversized = [j for j in jobs if j.is_oversized]
    total_bytes = sum(j.size_bytes for j in jobs)

    # Group by printer to find hot spots
    printer_stats = {}
    for j in jobs:
        if j.printer_name not in printer_stats:
            printer_stats[j.printer_name] = {
                "job_count": 0,
                "total_bytes": 0,
                "stalled_count": 0
            }
        printer_stats[j.printer_name]["job_count"] += 1
        printer_stats[j.printer_name]["total_bytes"] += j.size_bytes
        if j.is_stalled:
            printer_stats[j.printer_name]["stalled_count"] += 1

    report = {
        "timestamp": datetime.now().isoformat(),
        "summary": {
            "total_jobs": total_jobs,
            "stalled_jobs": len(stalled),
            "oversized_jobs": len(oversized),
            "total_spool_bytes": total_bytes,
            "total_spool_mb": round(total_bytes / 1024 / 1024, 2)
        },
        "printer_breakdown": printer_stats,
        "flagged_jobs": [
            {
                "job_id": j.job_id,
                "printer": j.printer_name,
                "document": j.document_name,
                "flags": j.notes
            }
            for j in jobs if j.is_stalled or j.is_oversized
        ]
    }
    return report


def main():
    parser = argparse.ArgumentParser(
        description="Print Queue Health Monitor — diagnose spooler bottlenecks"
    )
    parser.add_argument(
        "--host", "-H",
        help="Remote host to query (omit for local)",
        default=None
    )
    parser.add_argument(
        "--threshold-mb",
        type=int,
        default=DEFAULT_SIZE_THRESHOLD_MB,
        help="Flag jobs larger than this threshold in MB"
    )
    parser.add_argument(
        "--output", "-o",
        help="Write JSON report to file",
        default=None
    )
    args = parser.parse_args()

    global DEFAULT_SIZE_THRESHOLD_MB
    DEFAULT_SIZE_THRESHOLD_MB = args.threshold_mb

    logger.info(f"Scanning print queues (threshold: {args.threshold_mb} MB)...")

    # Dispatch based on OS
    system = platform.system()
    if system == "Windows":
        jobs = get_jobs_windows(host=args.host)
    elif system in ("Linux", "Darwin"):
        jobs = get_jobs_linux(host=args.host)
    else:
        logger.error(f"Unsupported OS: {system}")
        sys.exit(1)

    if not jobs:
        logger.info("No active print jobs found.")
        sys.exit(0)

    report = generate_health_report(jobs)

    # Output results
    output = json.dumps(report, indent=2, default=str)
    if args.output:
        with open(args.output, "w") as f:
            f.write(output)
        logger.info(f"Report written to {args.output}")
    else:
        print(output)

    # Exit with error code if problems found
    if report["summary"]["stalled_jobs"] > 0:
        sys.exit(2)  # Nagios/Icinga compatible


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Network Print Bandwidth Benchmark
Measures effective throughput to IPP printers or CUPS servers.
Generates synthetic print data to avoid wasting real paper/toner.

Requirements: pip install pyipp requests
Usage: python print_benchmark.py --printer-uri http://printer:631/ipp/print
"""

import argparse
import hashlib
import logging
import statistics
import sys
import time
import uuid
from dataclasses import dataclass, asdict
from typing import List, Optional

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s"
)
logger = logging.getLogger("print_benchmark")

# Test payload sizes (in bytes) — simulate real-world job variety
PAYLOAD_SIZES = [
    64 * 1024,        # 64 KB  — small text document
    256 * 1024,       # 256 KB — medium memo with images
    1 * 1024 * 1024,  # 1 MB   — standard report
    5 * 1024 * 1024,  # 5 MB   — presentation deck
    20 * 1024 * 1024, # 20 MB  — high-res photo book
]

# Number of iterations per payload size for statistical significance
ITERATIONS = 5


@dataclass
class BenchmarkResult:
    """Stores the result of a single benchmark iteration."""
    payload_size_bytes: int
    duration_seconds: float
    throughput_mbps: float
    success: bool
    error_message: Optional[str] = None


@dataclass
class BenchmarkSummary:
    """Aggregated statistics for one payload size."""
    payload_size_bytes: int
    iterations: int
    mean_throughput_mbps: float
    median_throughput_mbps: float
    p95_throughput_mbps: float
    min_throughput_mbps: float
    max_throughput_mbps: float
    success_rate: float
    mean_latency_ms: float


def generate_test_payload(size_bytes: int) -> bytes:
    """
    Generate deterministic test data using repeated SHA-256 hashes.
    This ensures consistent payloads across runs for reproducible benchmarks.
    """
    payload = bytearray()
    seed = b"print-benchmark-seed-v1"
    while len(payload) < size_bytes:
        seed = hashlib.sha256(seed).digest()
        payload.extend(seed)
    return bytes(payload[:size_bytes])


def create_session_with_retries(max_retries=3, backoff=0.5):
    """Create a requests session with automatic retry on transient failures."""
    session = requests.Session()
    retry_strategy = Retry(
        total=max_retries,
        backoff_factor=backoff,
        status_forcelist=[500, 502, 503, 504],
        allowed_methods=["POST", "GET", "PUT"],
        raise_on_status=False
    )
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    return session


def send_ipp_test(
    session: requests.Session,
    printer_uri: str,
    payload: bytes,
    job_name: str
) -> BenchmarkResult:
    """
    Send a single test job via IPP Create-Job + Send-Document.
    Returns timing and throughput data.
    """
    start_time = time.monotonic()

    try:
        # Step 1: Create a print job
        create_url = f"{printer_uri.rstrip('/')}/jobs"
        headers = {
            "Content-Type": "application/ipp",
            "Accept": "application/ipp"
        }

        # Minimal IPP Create-Job request body
        # This is a simplified IPP/1.1 Create-Job with attributes
        ipp_create_body = _build_ipp_create_job(job_name, len(payload))

        response = session.post(
            create_url,
            data=ipp_create_body,
            headers=headers,
            timeout=60
        )

        if response.status_code >= 400:
            # Fallback: try direct POST of document data
            response = session.post(
                printer_uri,
                data=payload,
                headers={"Content-Type": "application/octet-stream"},
                timeout=120
            )

        end_time = time.monotonic()
        duration = end_time - start_time
        throughput_mbps = (len(payload) / duration) / (1024 * 1024) if duration > 0 else 0

        return BenchmarkResult(
            payload_size_bytes=len(payload),
            duration_seconds=round(duration, 4),
            throughput_mbps=round(throughput_mbps, 2),
            success=response.status_code < 400,
            error_message=None if response.status_code < 400 else f"HTTP {response.status_code}"
        )

    except requests.exceptions.Timeout:
        return BenchmarkResult(
            payload_size_bytes=len(payload),
            duration_seconds=60.0,
            throughput_mbps=0.0,
            success=False,
            error_message="Request timed out after 60s"
        )
    except requests.exceptions.ConnectionError as e:
        return BenchmarkResult(
            payload_size_bytes=len(payload),
            duration_seconds=0.0,
            throughput_mbps=0.0,
            success=False,
            error_message=f"Connection failed: {e}"
        )
    except Exception as e:
        return BenchmarkResult(
            payload_size_bytes=len(payload),
            duration_seconds=0.0,
            throughput_mbps=0.0,
            success=False,
            error_message=str(e)
        )


def _build_ipp_create_job(job_name: str, document_size: int) -> bytes:
    """
    Build a minimal IPP/1.1 Create-Job request.
    IPP uses binary encoding with tag-delimited attributes.
    """
    # IPP operation ID for Create-Job is 0x0002
    operation_id = 0x0002
    request_id = 1

    # Build attribute groups
    # Operation attributes tag (0x01)
    attrs = bytearray()
    attrs.extend(b'\x01')  # operation-attributes-tag

    # attributes-charset (type 0x47)
    attrs.extend(b'\x47\x00\x17attributes-charset')
    charset = b'us-ascii'
    attrs.extend((len(charset) + 1).to_bytes(2, 'big'))
    attrs.extend(charset + b'\x00')

    # attributes-natural-language (type 0x48)
    attrs.extend(b'\x48\x00\x1battributes-natural-language')
    lang = b'en-us'
    attrs.extend((len(lang) + 1).to_bytes(2, 'big'))
    attrs.extend(lang + b'\x00')

    # job-name (type 0x49)
    name_bytes = job_name.encode('utf-8')
    attrs.extend(b'\x49\x00\x08job-name')
    attrs.extend((len(name_bytes) + 1).to_bytes(2, 'big'))
    attrs.extend(name_bytes + b'\x00')

    # assembling the full IPP message
    version = b'\x02\x00'  # IPP/1.1
    op_bytes = operation_id.to_bytes(2, 'big')
    req_id_bytes = request_id.to_bytes(4, 'big')

    return version + op_bytes + req_id_bytes + bytes(attrs) + b'\x03'


def run_benchmark(printer_uri: str) -> List[BenchmarkResult]:
    """Run the full benchmark suite against the target printer."""
    session = create_session_with_retries()
    results = []

    for size in PAYLOAD_SIZES:
        logger.info(f"Testing payload size: {size / 1024:.0f} KB ({ITERATIONS} iterations)...")
        payload = generate_test_payload(size)

        for i in range(ITERATIONS):
            job_name = f"benchmark-{uuid.uuid4().hex[:8]}-{i}"
            result = send_ipp_test(session, printer_uri, payload, job_name)
            results.append(result)
            status = "OK" if result.success else f"FAIL: {result.error_message}"
            logger.info(
                f"  Run {i+1}: {result.duration_seconds:.2f}s, "
                f"{result.throughput_mbps:.2f} MB/s — {status}"
            )

    return results


def summarize_results(results: List[BenchmarkResult]) -> List[BenchmarkSummary]:
    """Compute aggregated statistics per payload size."""
    from collections import defaultdict
    grouped = defaultdict(list)

    for r in results:
        grouped[r.payload_size_bytes].append(r)

    summaries = []
    for size in sorted(grouped.keys()):
        group = grouped[size]
        successful = [r for r in group if r.success]
        throughputs = [r.throughput_mbps for r in successful]
        latencies = [r.duration_seconds * 1000 for r in successful]

        if not throughputs:
            summaries.append(BenchmarkSummary(
                payload_size_bytes=size,
                iterations=len(group),
                mean_throughput_mbps=0.0,
                median_throughput_mbps=0.0,
                p95_throughput_mbps=0.0,
                min_throughput_mbps=0.0,
                max_throughput_mbps=0.0,
                success_rate=0.0,
                mean_latency_ms=0.0
            ))
            continue

        sorted_tp = sorted(throughputs)
        p95_idx = int(len(sorted_tp) * 0.95)

        summaries.append(BenchmarkSummary(
            payload_size_bytes=size,
            iterations=len(group),
            mean_throughput_mbps=round(statistics.mean(throughputs), 2),
            median_throughput_mbps=round(statistics.median(throughputs), 2),
            p95_throughput_mbps=round(sorted_tp[min(p95_idx, len(sorted_tp) - 1)], 2),
            min_throughput_mbps=round(min(throughputs), 2),
            max_throughput_mbps=round(max(throughputs), 2),
            success_rate=round(len(successful) / len(group) * 100, 1),
            mean_latency_ms=round(statistics.mean(latencies), 1)
        ))

    return summaries


def main():
    parser = argparse.ArgumentParser(
        description="Benchmark network print throughput to IPP/CUPS printers"
    )
    parser.add_argument(
        "--printer-uri", "-u",
        required=True,
        help="IPP printer URI, e.g., http://printer:631/ipp/print"
    )
    parser.add_argument(
        "--iterations", "-n",
        type=int,
        default=ITERATIONS,
        help=f"Iterations per payload size (default: {ITERATIONS})"
    )
    parser.add_argument(
        "--json-output", "-j",
        help="Write raw results to JSON file"
    )
    args = parser.parse_args()

    global ITERATIONS
    ITERATIONS = args.iterations

    logger.info(f"Starting print benchmark against {args.printer_uri}")
    results = run_benchmark(args.printer_uri)

    summaries = summarize_results(results)

    print("\n" + "=" * 72)
    print("NETWORK PRINT BANDWIDTH BENCHMARK RESULTS")
    print("=" * 72)
    print(f"{'Payload Size':<15} {'Mean (MB/s)':<14} {'Median':<10} {'P95':<10} {'Success %':<10} {'Avg Latency'}")
    print("-" * 72)
    for s in summaries:
        size_label = f"{s.payload_size_bytes / 1024:.0f}KB" if s.payload_size_bytes < 1024 * 1024 else f"{s.payload_size_bytes / 1024 / 1024:.0f}MB"
        print(
            f"{size_label:<15} {s.mean_throughput_mbps:<14.2f} "
            f"{s.median_throughput_mbps:<10.2f} {s.p95_throughput_mbps:<10.2f} "
            f"{s.success_rate:<10.1f} {s.mean_latency_ms:.0f}ms"
        )

    if args.json_output:
        import json
        output = {
            "printer_uri": args.printer_uri,
            "timestamp": time.strftime("%Y-%m-%dT%H:%M:%S"),
            "raw_results": [asdict(r) for r in results],
            "summaries": [asdict(s) for s in summaries]
        }
        with open(args.json_output, "w") as f:
            json.dump(output, f, indent=2)
        logger.info(f"Detailed results written to {args.json_output}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Print Job Batch Optimizer
Combines multiple print jobs into optimized batches to minimize
per-job overhead. Supports PDF merging, page reordering, and
automatic duplex pairing.

Requirements: pip install pypdf2
Usage:
    python batch_optimizer.py --input-dir ./jobs/ --output merged_batch.pdf
"""

import argparse
import hashlib
import json
import logging
import os
import sys
import time
from dataclasses import dataclass, field, asdict
from pathlib import Path
from typing import Dict, List, Optional, Tuple

# pypdf for PDF manipulation
from pypdf import PdfReader, PdfWriter, Transformation

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s"
)
logger = logging.getLogger("batch_optimizer")

# Configuration constants
DEFAULT_BATCH_SIZE = 50           # Max pages per merged batch
DEFAULT_DUPLEX_PAIR = True       # Pair odd/even for duplex printing
COMPRESSION_LEVEL = 6            # zlib compression level for output
MEMORY_LIMIT_MB = 512            # Soft memory limit for processing


@dataclass
class PrintJobMetadata:
    """Metadata for an individual input print job."""
    filepath: str
    page_count: int
    file_size_bytes: int
    sha256: str
    created_at: str
    tags: List[str] = field(default_factory=list)


@dataclass
class BatchPlan:
    """A planned batch of jobs to merge together."""
    batch_id: int
    jobs: List[PrintJobMetadata] = field(default_factory=list)
    total_pages: int = 0
    total_input_bytes: int = 0
    estimated_output_bytes: int = 0
    duplex_pairs: int = 0
    blank_inserts: int = 0

    @property
    def page_count(self) -> int:
        return self.total_pages


class BatchOptimizer:
    """
    Core optimizer that analyzes a set of print jobs and produces
    an optimal batching strategy.
    """

    def __init__(
        self,
        batch_size: int = DEFAULT_BATCH_SIZE,
        duplex: bool = DEFAULT_DUPLEX_PAIR,
        memory_limit_mb: int = MEMORY_LIMIT_MB
    ):
        self.batch_size = batch_size
        self.duplex = duplex
        self.memory_limit_bytes = memory_limit_mb * 1024 * 1024
        self._batches: List[BatchPlan] = []

    def analyze_jobs(self, filepaths: List[str]) -> List[PrintJobMetadata]:
        """
        Scan input files and extract metadata for each job.
        Validates that files are valid PDFs before adding to queue.
        """
        jobs = []
        for fp in filepaths:
            path = Path(fp)
            if not path.exists():
                logger.warning(f"File not found, skipping: {fp}")
                continue

            if path.suffix.lower() not in (".pdf", ".PDF"):
                logger.warning(f"Non-PDF file skipped (support PDF only): {fp}")
                continue

            try:
                file_size = path.stat().st_size
                # Compute hash for deduplication
                file_hash = self._compute_hash(path)
                # Count pages using pypdf (streaming, low memory)
                page_count = self._count_pages(path)
                created = time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime(path.stat().st_mtime))

                metadata = PrintJobMetadata(
                    filepath=str(path),
                    page_count=page_count,
                    file_size_bytes=file_size,
                    sha256=file_hash,
                    created_at=created
                )
                jobs.append(metadata)
                logger.info(f"Analyzed: {path.name} — {page_count} pages, {file_size / 1024:.1f} KB")

            except Exception as e:
                logger.error(f"Failed to analyze {fp}: {e}")
                continue

        # Deduplicate identical files
        return self._deduplicate(jobs)

    def _compute_hash(self, filepath: Path) -> str:
        """Compute SHA-256 hash of a file for deduplication."""
        h = hashlib.sha256()
        with open(filepath, "rb") as f:
            while chunk := f.read(8192):
                h.update(chunk)
        return h.hexdigest()

    def _count_pages(self, filepath: Path) -> int:
        """Count pages in a PDF using streaming read for memory efficiency."""
        try:
            reader = PdfReader(str(filepath))
            return len(reader.pages)
        except Exception as e:
            logger.error(f"Cannot read PDF {filepath}: {e}")
            return 0

    def _deduplicate(self, jobs: List[PrintJobMetadata]) -> List[PrintJobMetadata]:
        """Remove duplicate files based on SHA-256 hash."""
        seen = set()
        unique = []
        removed = 0
        for job in jobs:
            if job.sha256 in seen:
                removed += 1
                logger.info(f"Dedup removed: {job.filepath}")
                continue
            seen.add(job.sha256)
            unique.append(job)
        if removed:
            logger.info(f"Removed {removed} duplicate job(s)")
        return unique

    def plan_batches(self, jobs: List[PrintJobMetadata]) -> List[BatchPlan]:
        """
        Create optimal batch plans using a first-fit decreasing algorithm.
        Sorts jobs by page count (largest first) and packs them into bins
        of size self.batch_size.
        """
        # Sort by page count descending for better packing
        sorted_jobs = sorted(jobs, key=lambda j: j.page_count, reverse=True)
        bins: List[BatchPlan] = []

        for job in sorted_jobs:
            placed = False
            # Try to fit in existing batch
            for batch in bins:
                if batch.total_pages + job.page_count <= self.batch_size:
                    batch.jobs.append(job)
                    batch.total_pages += job.page_count
                    batch.total_input_bytes += job.file_size_bytes
                    placed = True
                    break

            # Create new batch if it doesn't fit anywhere
            if not placed:
                new_batch = BatchPlan(batch_id=len(bins) + 1)
                new_batch.jobs.append(job)
                new_batch.total_pages = job.page_count
                new_batch.total_input_bytes = job.file_size_bytes
                bins.append(new_batch)

        # Calculate estimates for each batch
        for i, batch in enumerate(bins):
            batch.batch_id = i + 1
            batch.duplex_pairs = batch.total_pages // 2 if self.duplex else 0
            batch.blank_inserts = 1 if (self.duplex and batch.total_pages % 2 != 0) else 0
            # Estimate ~60% compression ratio for typical office documents
            batch.estimated_output_bytes = int(batch.total_input_bytes * 0.6)

        self._batches = bins
        return bins

    def generate_report(self) -> dict:
        """Generate a summary report of the batch plan."""
        if not self._batches:
            return {"error": "No batches planned. Run plan_batches() first."}

        total_input = sum(b.total_input_bytes for b in self._batches)
        total_output = sum(b.estimated_output_bytes for b in self._batches)
        total_pages = sum(b.total_pages for b in self._batches)
        total_jobs = sum(len(b.jobs) for b in self._batches)

        # Calculate overhead savings
        # Without batching: each input file is a separate print job
        # With batching: only len(self._batches) jobs
        jobs_without_batching = total_jobs
        jobs_with_batching = len(self._batches)
        per_job_overhead_seconds = 3.2  # Measured empirically
        time_without = jobs_without_batching * per_job_overhead_seconds
        time_with = jobs_with_batching * per_job_overhead_seconds

        return {
            "total_input_files": total_jobs,
            "total_pages": total_pages,
            "total_input_bytes": total_input,
            "estimated_output_bytes": total_output,
            "compression_ratio": round(total_output / total_input, 2) if total_input else 0,
            "batches_created": len(self._batches),
            "jobs_without_batching": jobs_without_batching,
            "jobs_with_batching": jobs_with_batching,
            "overhead_time_without_batching_seconds": round(time_without, 1),
            "overhead_time_with_batching_seconds": round(time_with, 1),
            "time_saved_seconds": round(time_without - time_with, 1),
            "batches": [
                {
                    "batch_id": b.batch_id,
                    "pages": b.total_pages,
                    "jobs": len(b.jobs),
                    "duplex_pairs": b.duplex_pairs,
                    "blank_inserts": b.blank_inserts
                }
                for b in self._batches
            ]
        }


def merge_batch(batch: BatchPlan, output_path: str) -> bool:
    """
    Merge all PDFs in a batch into a single output file.
    Applies duplex blank-page insertion if configured.
    """
    writer = PdfWriter()

    try:
        for job in batch.jobs:
            reader = PdfReader(job.filepath)
            for page in reader.pages:
                writer.add_page(page)

        # Insert blank page for duplex alignment if needed
        if batch.blank_inserts > 0:
            from pypdf import PageObject
            blank = PageObject.create_blank_page(width=612, height=792)
            writer.add_page(blank)
            logger.info(f"Inserted {batch.blank_inserts} blank page(s) for duplex alignment")

        with open(output_path, "wb") as f:
            writer.write(f)

        logger.info(f"Wrote batch {batch.batch_id} to {output_path} "
                     f"({batch.total_pages} pages)")
        return True

    except Exception as e:
        logger.error(f"Failed to merge batch {batch.batch_id}: {e}")
        return False


def main():
    parser = argparse.ArgumentParser(
        description="Batch optimizer — merge and optimize print jobs"
    )
    parser.add_argument(
        "--input-dir", "-i",
        required=True,
        help="Directory containing PDF files to batch"
    )
    parser.add_argument(
        "--output", "-o",
        default="./merged_batch",
        help="Output directory for merged batch files"
    )
    parser.add_argument(
        "--batch-size", "-b",
        type=int,
        default=DEFAULT_BATCH_SIZE,
        help=f"Max pages per batch (default: {DEFAULT_BATCH_SIZE})"
    )
    parser.add_argument(
        "--no-duplex",
        action="store_true",
        help="Disable duplex page pairing"
    )
    parser.add_argument(
        "--report-only",
        action="store_true",
        help="Print batch plan without merging"
    )
    args = parser.parse_args()

    # Discover all PDF files in input directory
    input_dir = Path(args.input_dir)
    if not input_dir.is_dir():
        logger.error(f"Not a directory: {args.input_dir}")
        sys.exit(1)

    pdf_files = sorted([
        str(f) for f in input_dir.glob("*.pdf")
    ])

    if not pdf_files:
        logger.error(f"No PDF files found in {args.input_dir}")
        sys.exit(1)

    logger.info(f"Found {len(pdf_files)} PDF files in {args.input_dir}")

    # Initialize optimizer and run analysis
    optimizer = BatchOptimizer(
        batch_size=args.batch_size,
        duplex=not args.no_duplex
    )

    jobs = optimizer.analyze_jobs(pdf_files)
    if not jobs:
        logger.error("No valid jobs to process")
        sys.exit(1)

    batches = optimizer.plan_batches(jobs)
    report = optimizer.generate_report()

    # Print report
    print("\n" + "=" * 60)
    print("BATCH OPTIMIZATION REPORT")
    print("=" * 60)
    print(f"Input files:              {report['total_input_files']}")
    print(f"Total pages:              {report['total_pages']}")
    print(f"Batches created:          {report['batches_created']}")
    print(f"Jobs without batching:    {report['jobs_without_batching']}")
    print(f"Jobs with batching:       {report['jobs_with_batching']}")
    print(f"Overhead time (no batch): {report['overhead_time_without_batching_seconds']}s")
    print(f"Overhead time (batched):  {report['overhead_time_with_batching_seconds']}s")
    print(f"Time saved:               {report['time_saved_seconds']}s")
    print()
    for b in report["batches"]:
        print(f"  Batch {b['batch_id']}: {b['pages']} pages, "
              f"{b['jobs']} files, {b['duplex_pairs']} duplex pairs")

    if args.report_only:
        sys.exit(0)

    # Merge batches
    output_dir = Path(args.output)
    output_dir.mkdir(parents=True, exist_ok=True)

    success_count = 0
    for batch in batches:
        out_file = output_dir / f"batch_{batch.batch_id:04d}.pdf"
        if merge_batch(batch, str(out_file)):
            success_count += 1

    print(f"\nMerged {success_count}/{len(batches)} batches to {output_dir}")


if __name__ == "__main__":
    main()

# Mount a tmpfs for CUPS spool (adjust size for your workload)
mount -t tmpfs -o size=2G tmpfs /var/spool/cups

# Make persistent via /etc/fstab
tmpfs /var/spool/cups tmpfs defaults,size=2G 0 0

# /etc/cups/cupsd.conf — Enable and secure IPP
Listen 0.0.0.0:631

<Location />
  Order allow,deny
  Allow 192.168.1.0/24
  # Require encryption for external networks
</Location>

<Policy default>
  EncryptionRequired Yes
</Policy>

#!/usr/bin/env python3
"""
Pre-rasterize and cache template pages.
Generate once, reference many times in print jobs.
"""

import hashlib
import os
import subprocess
from pathlib import Path
from typing import Dict

CACHE_DIR = Path("/var/cache/print_raster_cache")
CACHE_DIR.mkdir(parents=True, exist_ok=True)


def rasterize_page(pdf_path: str, page_num: int, dpi: int = 600) -> Path:
    """
    Rasterize a single PDF page to a compressed TIFF.
    Uses Ghostscript under the hood for broad format support.
    """
    # Create a content hash for cache key
    with open(pdf_path, "rb") as f:
        content_hash = hashlib.sha256(f.read()).hexdigest()[:16]

    cache_key = f"{content_hash}_p{page_num:04d}_{dpi}dpi.tiff"
    cache_path = CACHE_DIR / cache_key

    # Return cached version if it exists
    if cache_path.exists():
        return cache_path

    # Otherwise, rasterize with Ghostscript
    try:
        subprocess.run(
            [
                "gs", "-dQUIET", "-dNOPAUSE", "-dBATCH",
                f"-r{dpi}",
                "-sDEVICE=tiff24nc",
                "-dFirstPage=" + str(page_num),
                "-dLastPage=" + str(page_num),
                f"-sOutputFile={cache_path}",
                pdf_path
            ],
            check=True,
            capture_output=True,
            timeout=120
        )
        logger.info(f"Rasterized {pdf_path} page {page_num} to {cache_path}")
        return cache_path
    except subprocess.CalledProcessError as e:
        logger.error(f"Ghostscript failed: {e.stderr.decode()}")
        raise
    except subprocess.TimeoutExpired:
        logger.error(f"Rasterization timed out for {pdf_path}")
        raise


def print_cached_page(cache_path: Path, printer_name: str) -> bool:
    """
    Send a pre-rasterized page directly to the printer,
    bypassing the driver rasterization pipeline.
    """
    try:
        result = subprocess.run(
            ["lp", "-d", printer_name, str(cache_path)],
            capture_output=True,
            text=True,
            timeout=60
        )
        if result.returncode != 0:
            logger.error(f"Print failed: {result.stderr}")
            return False
        return True
    except Exception as e:
        logger.error(f"Failed to send cached page: {e}")
        return False


# Example: pre-rasterize a 50-page invoice template
if __name__ == "__main__":
    import logging
    logging.basicConfig(level=logging.INFO)
    logger = logging.getLogger("raster_cache")

    template = "/opt/templates/monthly_invoice.pdf"
    for page in range(1, 51):
        try:
            rasterize_page(template, page, dpi=300)
        except Exception as e:
            logger.warning(f"Page {page} failed: {e}")

    print(f"Cache populated. Files in {CACHE_DIR}: {len(list(CACHE_DIR.iterdir()))}")