~kf5jwc/sms-printer

6a2e6e2b7591ba972574af4054db55dbf969bba0 — Kyle Jones 2 years ago 664f82c
Code linters are great.
M Pipfile => Pipfile +4 -8
@@ 1,26 1,22 @@
[[source]]

verify_ssl = true
url = "https://pypi.python.org/simple"
name = "pypi"


[packages]

gunicorn = "*"
flask = "*"
pyxdg = "*"
jsonschema = "*"


[dev-packages]

isort = "*"
yapf = "*"
black = "*"
pylint = "*"
mypy = "*"


[requires]

python_version = "3.6"

[pipenv]
allow_prereleases = true

M Pipfile.lock => Pipfile.lock +49 -14
@@ 1,7 1,7 @@
{
    "_meta": {
        "hash": {
            "sha256": "d3d9d2a887f020329d9691f227c06981a74b45b6a6b434b5c42368659ab574ae"
            "sha256": "4578baaf6d64898434be661bb76aace3ecb7d761e09333bfcb44cab47c1a9683"
        },
        "pipfile-spec": 6,
        "requires": {


@@ 28,6 28,7 @@
                "sha256:2271c0070dbcb5275fad4a82e29f23ab92682dc45f9dfbc22c02ba9b9322ce48",
                "sha256:a080b744b7e345ccfcbc77954861cb05b3c63786e93f2b3875e0913d44b43f05"
            ],
            "index": "pypi",
            "version": "==1.0.2"
        },
        "gunicorn": {


@@ 35,6 36,7 @@
                "sha256:7ef2b828b335ed58e3b64ffa84caceb0a7dd7c5ca12f217241350dec36a1d5dc",
                "sha256:bc59005979efb6d2dd7d5ba72d99f8a8422862ad17ff3a16e900684630dd2a10"
            ],
            "index": "pypi",
            "version": "==19.8.1"
        },
        "itsdangerous": {


@@ 55,6 57,7 @@
                "sha256:000e68abd33c972a5248544925a0cae7d1125f9bf6c58280d37546b946769a08",
                "sha256:6ff5f3180870836cae40f06fa10419f557208175f13ad7bc26caa77beb1f6e02"
            ],
            "index": "pypi",
            "version": "==2.6.0"
        },
        "markupsafe": {


@@ 68,6 71,7 @@
                "sha256:1948ff8e2db02156c0cccd2529b43c0cff56ebaa71f5f021bbd755bc1419190e",
                "sha256:fe2928d3f532ed32b39c32a482b54136fe766d19936afc96c8f00645f9da1a06"
            ],
            "index": "pypi",
            "version": "==0.26"
        },
        "werkzeug": {


@@ 79,12 83,41 @@
        }
    },
    "develop": {
        "appdirs": {
            "hashes": [
                "sha256:9e5896d1372858f8dd3344faf4e5014d21849c756c8d5701f78f8a103b372d92",
                "sha256:d8b24664561d0d34ddfaec54636d502d7cea6e29c3eaf68f3df6180863e2166e"
            ],
            "version": "==1.4.3"
        },
        "astroid": {
            "hashes": [
                "sha256:0ef2bf9f07c3150929b25e8e61b5198c27b0dca195e156f0e4d5bdd89185ca1a",
                "sha256:fc9b582dba0366e63540982c3944a9230cbc6f303641c51483fa547dcc22393a"
                "sha256:b7787562a4757d7101ca6270a06e5246c1dcd1a4851d4bb2421a6baae4b09c1b",
                "sha256:f14913e18b71e6a86e05aeec80b8bd09286dff8a6e34c3cc71d67fa3d726aec7"
            ],
            "version": "==2.0.0.dev1"
        },
        "attrs": {
            "hashes": [
                "sha256:4b90b09eeeb9b88c35bc642cbac057e45a5fd85367b985bd2809c62b7b939265",
                "sha256:e0d0eb91441a3b53dab4d9b743eafc1ac44476296a2053b6ca3af0b139faf87b"
            ],
            "version": "==18.1.0"
        },
        "black": {
            "hashes": [
                "sha256:3efe92eafbde15f8ac06478de11cfb84e47504896ccdde64507e751d2f91ec3a",
                "sha256:fc26c4ab28c541fb824f59fa83d5702f75829495d5a1dee603b29bc4fbe79095"
            ],
            "index": "pypi",
            "version": "==18.6b2"
        },
        "click": {
            "hashes": [
                "sha256:29f99fc6125fbc931b758dc053b3114e55c77a6e4c6c3a2674a2dc986016381d",
                "sha256:f15516df478d5a56180fbf80e68f206010e6d160fc39fa508b65e035fd75130b"
            ],
            "version": "==1.6.5"
            "version": "==6.7"
        },
        "isort": {
            "hashes": [


@@ 92,6 125,7 @@
                "sha256:b9c40e9750f3d77e6e4d441d8b0266cf555e7cdabdcff33c4fd06366ca761ef8",
                "sha256:ec9ef8f4a9bc6f71eec99e1806bfa2de401650d996c59330782b89a5555c1497"
            ],
            "index": "pypi",
            "version": "==4.3.4"
        },
        "lazy-object-proxy": {


@@ 140,14 174,16 @@
                "sha256:1b899802a89b67bb68f30d788bba49b61b1f28779436f06b75c03495f9d6ea5c",
                "sha256:f472645347430282d62d1f97d12ccb8741f19f1572b7cf30b58280e4e0818739"
            ],
            "index": "pypi",
            "version": "==0.610"
        },
        "pylint": {
            "hashes": [
                "sha256:a48070545c12430cfc4e865bf62f5ad367784765681b3db442d8230f0960aa3c",
                "sha256:fff220bcb996b4f7e2b0f6812fd81507b72ca4d8c4d05daf2655c333800cb9b3"
                "sha256:0990347c0f605927fadb2a9366a0b3d40bd19eb44e4312f0a1ef729a389b2f40",
                "sha256:19b902f93f2dc3fa45565e54b88702b28379be40107f509a8516dde152460d1f"
            ],
            "version": "==1.9.2"
            "index": "pypi",
            "version": "==2.0.0.dev1"
        },
        "six": {
            "hashes": [


@@ 156,6 192,12 @@
            ],
            "version": "==1.11.0"
        },
        "toml": {
            "hashes": [
                "sha256:8e86bd6ce8cc11b9620cb637466453d94f5d57ad86f17e98a98d1f73e3baab2d"
            ],
            "version": "==0.9.4"
        },
        "typed-ast": {
            "hashes": [
                "sha256:0948004fa228ae071054f5208840a1e88747a357ec1101c17217bfe99b299d58",


@@ 193,13 235,6 @@
                "sha256:d4d560d479f2c21e1b5443bbd15fe7ec4b37fe7e53d335d3b9b0a7b1226fe3c6"
            ],
            "version": "==1.10.11"
        },
        "yapf": {
            "hashes": [
                "sha256:6567745f0b6656f9c33a73c56a393071c699e6284a70d793798ab6e3769d25ec",
                "sha256:a98a6eacca64d2b920558f4a2f78150db9474de821227e60deaa29f186121c63"
            ],
            "version": "==0.22.0"
        }
    }
}

M sms_printer/application.py => sms_printer/application.py +3 -3
@@ 9,12 9,12 @@ APP = Flask(__name__)
logging.basicConfig(level=logging.DEBUG)


@APP.route("/", methods=['GET'])
@APP.route("/", methods=["GET"])
def main():
    return Response(status=204)


@APP.route("/sms-printer", methods=['POST'])
@APP.route("/sms-printer", methods=["POST"])
def sms_printer() -> Response:
    # If this throws, flask takes care of it
    sms_input = request.get_json(force=True)


@@ 32,7 32,7 @@ def sms_printer() -> Response:

        logging.info("Recieved sms from %s", sms.sender)
        if not allowed_sender(sms.sender):
            logging.warn("Unauthorized sender %s", sms.sender)
            logging.warning("Unauthorized sender %s", sms.sender)
            return Response(status=401)

        logging.info("Printing sms from %s", sms.sender)

M sms_printer/config.py => sms_printer/config.py +29 -34
@@ 4,10 4,14 @@ from typing import List

from xdg import BaseDirectory


def first_defined_argument(*args):
    return next((item for item in args if item is not None), None)


first = first_defined_argument


class Environment(object):
    """
    This is where all of the configuration options (should) live.


@@ 16,63 20,54 @@ class Environment(object):
    options through their respective environment variables.
    """

    ARCHIVE = getenv('ARCHIVE_DIR', BaseDirectory.save_data_path(__package__))
    ARCHIVE_SUFFIX_LENGTH = int(getenv('SUFFIX_LENGTH', 6))
    ALLOWED_SENDERS = [
        num for num in getenv('ALLOWED_SENDERS', '').split(',') if num
    ]
    ARCHIVE = getenv("ARCHIVE_DIR", BaseDirectory.save_data_path(__package__))
    ARCHIVE_SUFFIX_LENGTH = int(getenv("SUFFIX_LENGTH", "6"))
    ALLOWED_SENDERS = [num for num in getenv("ALLOWED_SENDERS", "").split(",") if num]

    MESSAGE_FORMAT = getenv('MESSAGE_FORMAT', '{message}')
    MESSAGE_FORMAT = getenv("MESSAGE_FORMAT", "{message}")

    PRINT_COMMAND = getenv('PRINT_COMMAND', 'lp')
    PRINTER_NAME = getenv('PRINTER_NAME', None)
    PRINT_JOB_TIMEOUT = int(getenv('PRINT_JOB_TIMEOUT', 5)) # seconds
    PRINT_COMMAND = getenv("PRINT_COMMAND", "lp")
    PRINTER_NAME = getenv("PRINTER_NAME", None)
    PRINT_JOB_TIMEOUT = int(getenv("PRINT_JOB_TIMEOUT", "5"))  # seconds

    CPI = int(getenv('CPI', 10))
    LPI = int(getenv('LPI', 6))
    TEXT_SCALE = float(getenv('TEXT_SCALE', 1.0))
    CPI = int(getenv("CPI", "10"))
    LPI = int(getenv("LPI", "6"))
    TEXT_SCALE = float(getenv("TEXT_SCALE", "1"))

    # Set in points; each point is 1/72 inch or 0.35mm.
    MARGINS = getenv('MARGINS', None)
    MARGIN_SIDES = first(getenv('MARGIN_SIDES', None), MARGINS)
    MARGIN_TOP = first(getenv('MARGIN_TOP', None), MARGINS)
    MARGIN_BOTTOM = first(getenv('MARGIN_BOTTOM', None), MARGINS)
    MARGIN_LEFT = first(getenv('MARGIN_LEFT', None), MARGIN_SIDES, MARGINS)
    MARGIN_RIGHT = first(getenv('MARGIN_RIGHT', None), MARGIN_SIDES, MARGINS)
    MARGINS = getenv("MARGINS", None)
    MARGIN_SIDES = first(getenv("MARGIN_SIDES", None), MARGINS)
    MARGIN_TOP = first(getenv("MARGIN_TOP", None), MARGINS)
    MARGIN_BOTTOM = first(getenv("MARGIN_BOTTOM", None), MARGINS)
    MARGIN_LEFT = first(getenv("MARGIN_LEFT", None), MARGIN_SIDES, MARGINS)
    MARGIN_RIGHT = first(getenv("MARGIN_RIGHT", None), MARGIN_SIDES, MARGINS)

    def print_command(self, file_to_print=None) -> List[str]:
        command = [self.PRINT_COMMAND]

        if self.PRINTER_NAME is not None:
            command.append(
                "-d {}".format(self.PRINTER_NAME))
            command.append("-d {}".format(self.PRINTER_NAME))

        if self.TEXT_SCALE is not 1.0:
            # Scale horizontally
            cpi = floor(self.CPI*(1/self.TEXT_SCALE))
            command.append(
                "-o cpi={}".format(cpi))
            cpi = floor(self.CPI * (1 / self.TEXT_SCALE))
            command.append("-o cpi={}".format(cpi))

            # Scale vertically
            lpi = floor(self.LPI*(1/self.TEXT_SCALE))
            command.append(
                "-o lpi={}".format(lpi))
            lpi = floor(self.LPI * (1 / self.TEXT_SCALE))
            command.append("-o lpi={}".format(lpi))

        if self.MARGIN_TOP is not None:
            command.append(
                "-o page-top={}".format(self.MARGIN_TOP))
            command.append("-o page-top={}".format(self.MARGIN_TOP))

        if self.MARGIN_BOTTOM is not None:
            command.append(
                "-o page-bottom={}".format(self.MARGIN_BOTTOM))
            command.append("-o page-bottom={}".format(self.MARGIN_BOTTOM))

        if self.MARGIN_LEFT is not None:
            command.append(
                "-o page-left={}".format(self.MARGIN_LEFT))
            command.append("-o page-left={}".format(self.MARGIN_LEFT))

        if self.MARGIN_RIGHT is not None:
            command.append(
                "-o page-right={}".format(self.MARGIN_RIGHT))
            command.append("-o page-right={}".format(self.MARGIN_RIGHT))

        if file_to_print is not None:
            command.append(file_to_print)

M sms_printer/sms/__init__.py => sms_printer/sms/__init__.py +4 -1
@@ 5,9 5,11 @@ from flask import json
from .schemas import SCHEMAS
from .schemas.types import Parser_Base


class NoMatchingSchema(Exception):
    pass


def parse_sms(sms_input: str) -> Parser_Base:
    for schema, sms_parser in SCHEMAS.items():
        try:


@@ 18,4 20,5 @@ def parse_sms(sms_input: str) -> Parser_Base:

    raise NoMatchingSchema

__all__ = ['parse_sms', 'NoMatchingSchema']

__all__ = ["parse_sms", "NoMatchingSchema"]

M sms_printer/sms/schemas/__init__.py => sms_printer/sms/schemas/__init__.py +2 -2
@@ 16,7 16,7 @@ I use `genson` to generate schemas based on an example message.

from typing import Dict, Type

import types
from . import types

from . import bandwidth_v1
from . import bandwidth_v2


@@ 28,4 28,4 @@ SCHEMAS[bandwidth_v1.SCHEMA] = bandwidth_v1.Parser
SCHEMAS[bandwidth_v2.SCHEMA] = bandwidth_v2.Parser


__all__ = ['SCHEMAS', 'types']
__all__ = ["SCHEMAS", "types"]

M sms_printer/sms/schemas/bandwidth_v1.py => sms_printer/sms/schemas/bandwidth_v1.py +1 -3
@@ 6,9 6,7 @@ from .types import Parser_Base, sms
class Parser(Parser_Base):
    def __init__(self, sms_input: Dict) -> None:
        super().__init__()
        msg = sms(
            sms_input['from'],
            sms_input['text'])
        msg = sms(sms_input["from"], sms_input["text"])
        self.messages.append(msg)



M sms_printer/sms/schemas/bandwidth_v2.py => sms_printer/sms/schemas/bandwidth_v2.py +1 -3
@@ 7,9 7,7 @@ class Parser(Parser_Base):
    def __init__(self, sms_json: List) -> None:
        super().__init__()
        for msg_input in sms_json:
            msg = sms(
                msg_input['message']['from'],
                msg_input['message']['text'])
            msg = sms(msg_input["message"]["from"], msg_input["message"]["text"])
            self.messages.append(msg)



M sms_printer/utils.py => sms_printer/utils.py +11 -7
@@ 11,9 11,11 @@ from .config import Environment

ENV = Environment()

def suffix(length=ENV.ARCHIVE_SUFFIX_LENGTH, chars=ascii_uppercase+digits):

def suffix(length=ENV.ARCHIVE_SUFFIX_LENGTH, chars=ascii_uppercase + digits):
    "Random suffix string"
    return ''.join(rand_choice(chars) for x in range(length))
    return "".join(rand_choice(chars) for x in range(length))


def allowed_sender(sender: str) -> bool:
    """


@@ 22,8 24,8 @@ def allowed_sender(sender: str) -> bool:
    ALLOWED_SENDERS is a comma-delimited list, with numbers strictly
    in the format copied from your provider. +12223334444 (string)
    """
    return (not bool(ENV.ALLOWED_SENDERS)
            or sender in ENV.ALLOWED_SENDERS)
    return not bool(ENV.ALLOWED_SENDERS) or sender in ENV.ALLOWED_SENDERS


def log_request(data: str):
    """


@@ 33,9 35,10 @@ def log_request(data: str):
    with a random suffix.
    """
    msg_id = "{}-{}".format(strftime("%Y%m%d-%H%M%S"), suffix())
    with open(path.join(ENV.ARCHIVE, msg_id), 'w+') as archive_file:
    with open(path.join(ENV.ARCHIVE, msg_id), "w+") as archive_file:
        json.dump(data, archive_file)


def print_message(message: str) -> bool:
    """
    Sends a plaintext message to CUPS for printing.


@@ 44,7 47,7 @@ def print_message(message: str) -> bool:
    MESSAGE_FORMAT to a string, where `{message}` is replaced with
    the message text.
    """
    with NamedTemporaryFile(mode='w+') as tmp_file:
    with NamedTemporaryFile(mode="w+") as tmp_file:
        tmp_file.write(ENV.MESSAGE_FORMAT.format(message=message))
        tmp_file.flush()



@@ 52,7 55,8 @@ def print_message(message: str) -> bool:
            run(
                ENV.print_command(tmp_file.name),
                check=True,
                timeout=ENV.PRINT_JOB_TIMEOUT)
                timeout=ENV.PRINT_JOB_TIMEOUT,
            )

        except CalledProcessError:
            return False