~kf5jwc/sms-printer

ref: bc020437189df722dfb68d41c6c409bda4add52a sms-printer/sms_printer/application.py -rw-r--r-- 2.5 KiB View raw
bc020437 — Kyle Jones Add linters and type checking 2 years ago
                                                                                
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
import os
from math import floor
from subprocess import CalledProcessError, run
from tempfile import NamedTemporaryFile

from flask import Flask, Response, json, request
from xdg import BaseDirectory

APP = Flask(__name__)
ARCHIVE = os.getenv('ARCHIVE_DIR', BaseDirectory.save_data_path(__name__))
ALLOWED_SENDERS = [
    num for num in os.getenv('ALLOWED_SENDERS', '').split(',') if num
]
PRINTER_FORMAT = "\n\n" + "      {message}"*2
TEXT_SCALE = 0.6
CPI_DEFAULT = 10
LPI_DEFAULT = 6


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


@APP.route("/sms-printer", methods=['POST'])
def sms_printer() -> Response:
    def sms_main() -> Response:
        user_agent = (request.user_agent.platform, request.user_agent.version)
        if user_agent is ("BandwidthAPI", "v2"):
            return bandwidth_v2()

        # Default, since they report no UA
        return bandwidth_v1()

    def allowed_sender(sender: str) -> bool:
        return (bool(ALLOWED_SENDERS) and sender not in ALLOWED_SENDERS)

    def archive_request(sms_id: str, data: json):
        with open(os.path.join(ARCHIVE, sms_id), 'w+') as archive_file:
            json.dump(data, archive_file)

    def print_message(message: str) -> bool:
        with NamedTemporaryFile(mode='w+') as tmp_file:
            print(PRINTER_FORMAT.format(message=message), file=tmp_file)
            tmp_file.flush()

            command = [
                "lp",
                "-o cpi={}".format(floor(CPI_DEFAULT*(1/TEXT_SCALE))),
                "-o lpi={}".format(floor(LPI_DEFAULT*(1/TEXT_SCALE))),
                tmp_file.name]
            try:
                run(command, check=True, timeout=10)

            except CalledProcessError:
                return False

        return True

    def bandwidth_v1() -> Response:
        sms = request.get_json(force=True)
        archive_request(sms['messageId'], sms)

        if not allowed_sender(sms['from']):
            return Response(status=401)

        if not print_message(sms['text']):
            return Response(status=500)

        return Response(status=200)

    def bandwidth_v2() -> Response:
        for sms in request.get_json(force=True):
            archive_request(sms['message']['id'], sms)

            if not allowed_sender(sms['message']['from']):
                return Response(status=401)

            if not print_message(sms['message']['text']):
                return Response(status=500)

        return Response(status=200)

    sms_main()