LPR client written in Python

pylpr.py

"""\
License: MIT

See also:
  - RFC 1179 - Line Printer Daemon Protocol
    http://www.ietf.org/rfc/rfc1179.txt

  - David Boddie's lpr.py (Thanks!)
    http://www.boddie.org.uk/david/Projects/Python/lptools/lpr.py
"""

from argparse import ArgumentParser
from contextlib import closing
from getpass import getuser
from logging import getLogger
from random import randint
from socket import create_connection
from socket import gethostname
import sys
from os.path import basename


class LinePrinterRemote(object):
    BUFFSIZE=1024

    def __init__(self, host, port=515, queue="lp", encoding="ascii"):
        self.host = host
        self.port = port
        self.queue = queue
        self.encoding = encoding

    @staticmethod
    def assertNL(res):
        assert res == b"\x00", "Unexpected response, %r" % res

    def pack(self, *chunks):
        return b"".join([
            chunk if isinstance(chunk, (bytes, bytearray)) else
                str(chunk).encode(encoding=self.encoding, errors="replace")
            for chunk in chunks
        ])

    def send(self, data, filename, user, host):
        logger = getLogger(__name__)
        SP, LF, NL= b"\x20", b"\x0a", b"\x00"
        dfA_NNN_host = "dfA%03d%s" % (randint(1, 999), host)

        # Windows emulation
        control = self.pack(
            b"H", host, LF,
            b"P", user, LF,
            b"l", dfA_NNN_host, LF,
            b"U", dfA_NNN_host, LF,
            b"N", filename, LF,
        )

        logger.debug(control)
        logger.debug("Connecting to %s:%d", self.host, self.port)
        with closing(create_connection((self.host, self.port))) as sock:
            logger.debug("Sending queue info, %s", self.queue)
            sock.send(self.pack(b"\x02", self.queue, LF))
            self.assertNL(sock.recv(self.BUFFSIZE))

            logger.debug("Sending control info")
            sock.send(self.pack(b"\x02", len(control), SP, dfA_NNN_host, LF))
            self.assertNL(sock.recv(self.BUFFSIZE))

            logger.debug("Sending control, %d bytes", len(control))
            sock.send(self.pack(control, NL))
            self.assertNL(sock.recv(self.BUFFSIZE))

            logger.debug("Sending data info")
            sock.send(self.pack(b"\x03", len(data), SP, dfA_NNN_host, LF))
            self.assertNL(sock.recv(self.BUFFSIZE))

            logger.debug("Sending data, %d bytes", len(data))
            sock.send(self.pack(data, NL))
            self.assertNL(sock.recv(self.BUFFSIZE))
        logger.debug("Completed")


def getargs(args):
    default_encoding = sys.getdefaultencoding()
    default_username = getuser()
    default_hostname = gethostname()
    parser = ArgumentParser(description="LPR Client",)
    parser.add_argument("-v", "--verbose", action="store_true",
                        help="enable verbose logging")
    parser.add_argument("-E", "--encoding", type=str, default=default_encoding,
                        help="text encoding")
    parser.add_argument("-Q", "--queue", type=str, default="lp",
                        help="destination queue")
    parser.add_argument("-P", "--port", type=int, default=515,
                        help="destination port")
    parser.add_argument("-U", "--username", type=str, default=default_username,
                        help="user name producing the job")
    parser.add_argument("-H", "--hostname", type=str, default=default_hostname,
                        help="host name producing the job")
    parser.add_argument("host", metavar="HOST", type=str,
                        help="destination host")
    parser.add_argument("files", metavar="FILE", type=str, nargs="+",
                        help="file recording print job data")
    return parser.parse_args(args)


def main():
    from logging import basicConfig
    from logging import DEBUG
    from logging import INFO

    args = getargs(sys.argv[1:])
    basicConfig(level=args.verbose and DEBUG or INFO)
    logger = getLogger(__name__)

    lpr = LinePrinterRemote(args.host, args.port, args.queue, args.encoding)
    for f in args.files:
        logger.info("Printing %s", f)
        with open(f, "rb") as fp:
            try:
                lpr.send(fp.read(), basename(f), args.username, args.hostname)
            except AssertionError as e:
                logger.exception(e)


if __name__ == "__main__":
    main()