diff --git a/print/app.py b/print/app.py new file mode 100644 index 0000000..77b257e --- /dev/null +++ b/print/app.py @@ -0,0 +1,105 @@ +import os +import subprocess +from pathlib import Path + +import pypdf +from flask import Flask, flash, redirect, render_template, request, url_for +from werkzeug.utils import secure_filename + +UPLOAD_FOLDER = str(Path(__file__).parent / "uploads") +ALLOWED_EXTENSIONS = {"pdf"} +PRINTER_NAME = "HL-2270DW_series" +MAX_PAGES = 10 + +app = Flask(__name__) +app.config["UPLOAD_FOLDER"] = UPLOAD_FOLDER +app.config["SECRET_KEY"] = os.urandom(24) +app.config["MAX_CONTENT_LENGTH"] = 16 * 1024 * 1024 + + +def allowed_file(filename): + return "." in filename and filename.rsplit(".", 1)[1].lower() in ALLOWED_EXTENSIONS + + +def get_pdf_page_count(filepath): + try: + with open(filepath, "rb") as f: + reader = pypdf.PdfReader(f) + return len(reader.pages) + except Exception as e: + app.logger.error(f"Could not read PDF pages: {e}") + return None + + +@app.route("/", methods=["GET", "POST"]) +def upload_and_print(): + if request.method == "POST": + if "file" not in request.files: + flash("No file part in the request.", "error") + return redirect(request.url) + + file = request.files["file"] + + if file.filename == "": + flash("No file selected. Please choose a PDF to print.", "error") + return redirect(request.url) + + if file and allowed_file(file.filename): + filename = secure_filename(file.filename) + os.makedirs(app.config["UPLOAD_FOLDER"], exist_ok=True) + filepath = os.path.join(app.config["UPLOAD_FOLDER"], filename) + file.save(filepath) + + try: + page_count = get_pdf_page_count(filepath) + if page_count is None: + flash( + "Could not process the PDF file. It might be corrupt.", "error" + ) + return redirect(request.url) + + if page_count > MAX_PAGES: + flash( + f"Error: File has {page_count} pages. The maximum allowed is {MAX_PAGES}.", + "error", + ) + return redirect(request.url) + + app.logger.info( + f"Sending '{filename}' ({page_count} pages) to printer '{PRINTER_NAME}'" + ) + command = ["lp", "-d", PRINTER_NAME, filepath] + result = subprocess.run( + command, capture_output=True, text=True, check=True + ) + app.logger.info(f"lp command stdout: {result.stdout}") + flash(f"Success! '{filename}' has been sent to the printer.", "success") + + except subprocess.CalledProcessError as e: + app.logger.error(f"Error printing file: {e}") + app.logger.error(f"lp command stderr: {e.stderr}") + flash( + "Error sending file to printer. The printer may be offline or misconfigured.", + "error", + ) + + except Exception as e: + app.logger.error(f"An unexpected error occurred: {e}") + flash("An unexpected server error occurred.", "error") + + finally: + if os.path.exists(filepath): + os.remove(filepath) + app.logger.info(f"Cleaned up temporary file: {filepath}") + + return redirect(url_for("upload_and_print")) + + else: + flash("Invalid file type. Please upload a PDF file.", "error") + return redirect(request.url) + + return render_template("index.html", printer_name=PRINTER_NAME, max_pages=MAX_PAGES) + + +if __name__ == "__main__": + app.run(host="127.0.0.1", port=57928, debug=False) diff --git a/print/templates/index.html b/print/templates/index.html new file mode 100644 index 0000000..5564e1f --- /dev/null +++ b/print/templates/index.html @@ -0,0 +1,82 @@ + + + + + + Web Print Service + + + + + + + + +
+
+ +
+ +

PDF Web Print

+

Upload a PDF to print to {{ printer_name }}.

+

(Max {{ max_pages }} pages)

+
+ + {% with messages = get_flashed_messages(with_categories=true) %} + {% if messages %} +
+ {% for category, message in messages %} + + {% endfor %} +
+ {% endif %} + {% endwith %} + +
+
+ + +
+ +
+ +
+
+
+ + +
+ + + + + diff --git a/print/tests/__init__.py b/print/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/print/tests/conftest.py b/print/tests/conftest.py new file mode 100644 index 0000000..441d010 --- /dev/null +++ b/print/tests/conftest.py @@ -0,0 +1,59 @@ +import io +import tempfile + +import pypdf +import pytest + +from app import app as flask_app, get_pdf_page_count + + +@pytest.fixture +def app(): + flask_app.config["TESTING"] = True + flask_app.config["UPLOAD_FOLDER"] = tempfile.mkdtemp() + flask_app.config["SECRET_KEY"] = "test-secret" + yield flask_app + + +@pytest.fixture +def client(app): + return app.test_client() + + +def _make_pdf(num_pages=3): + writer = pypdf.PdfWriter() + for _ in range(num_pages): + writer.add_blank_page(width=612, height=792) + buf = io.BytesIO() + writer.write(buf) + buf.seek(0) + return buf + + +@pytest.fixture +def sample_pdf(): + return _make_pdf(3).read() + + +@pytest.fixture +def large_pdf(): + return _make_pdf(15).read() + + +@pytest.fixture +def corrupt_file(): + return b"This is not a PDF at all" + + +@pytest.fixture +def sample_pdf_path(tmp_path): + path = tmp_path / "test.pdf" + path.write_bytes(_make_pdf(3).read()) + return str(path) + + +@pytest.fixture +def corrupt_file_path(tmp_path): + path = tmp_path / "corrupt.pdf" + path.write_bytes(b"not a pdf") + return str(path) diff --git a/print/tests/test_pdf_utils.py b/print/tests/test_pdf_utils.py new file mode 100644 index 0000000..7054f51 --- /dev/null +++ b/print/tests/test_pdf_utils.py @@ -0,0 +1,13 @@ +from app import get_pdf_page_count + + +def test_page_count_valid_pdf(sample_pdf_path): + assert get_pdf_page_count(sample_pdf_path) == 3 + + +def test_page_count_corrupt_file(corrupt_file_path): + assert get_pdf_page_count(corrupt_file_path) is None + + +def test_page_count_nonexistent_file(): + assert get_pdf_page_count("/nonexistent/file.pdf") is None diff --git a/print/tests/test_routes.py b/print/tests/test_routes.py new file mode 100644 index 0000000..4414a63 --- /dev/null +++ b/print/tests/test_routes.py @@ -0,0 +1,69 @@ +import io +import os +from unittest.mock import patch + + +def test_get_index(client): + resp = client.get("/") + assert resp.status_code == 200 + assert b"printer" in resp.data.lower() or b"upload" in resp.data.lower() + + +def test_post_no_file(client): + resp = client.post("/", data={}, follow_redirects=True) + assert resp.status_code == 200 + assert b"No file part" in resp.data + + +def test_post_empty_filename(client): + resp = client.post("/", data={"file": (io.BytesIO(b""), "")}, follow_redirects=True) + assert resp.status_code == 200 + assert b"No file selected" in resp.data + + +def test_post_invalid_extension(client): + data = {"file": (io.BytesIO(b"hello"), "document.txt")} + resp = client.post("/", data=data, follow_redirects=True) + assert resp.status_code == 200 + assert b"Invalid file type" in resp.data + + +@patch("app.subprocess.run") +def test_post_valid_pdf(mock_run, client, app, sample_pdf): + mock_run.return_value.stdout = "request-id-123" + mock_run.return_value.stderr = "" + mock_run.return_value.returncode = 0 + + data = {"file": (io.BytesIO(sample_pdf), "test.pdf")} + resp = client.post("/", data=data, follow_redirects=True) + assert resp.status_code == 200 + assert b"Success" in resp.data + + upload_dir = app.config["UPLOAD_FOLDER"] + remaining = os.listdir(upload_dir) + assert len(remaining) == 0 + + mock_run.assert_called_once() + cmd = mock_run.call_args[0][0] + assert cmd[0] == "lp" + + +@patch("app.subprocess.run") +def test_post_pdf_exceeds_max_pages(mock_run, client, large_pdf): + data = {"file": (io.BytesIO(large_pdf), "big.pdf")} + resp = client.post("/", data=data, follow_redirects=True) + assert resp.status_code == 200 + assert b"maximum allowed" in resp.data.lower() + + mock_run.assert_not_called() + + +def test_post_corrupt_pdf(client, app, corrupt_file): + data = {"file": (io.BytesIO(corrupt_file), "bad.pdf")} + resp = client.post("/", data=data, follow_redirects=True) + assert resp.status_code == 200 + assert b"corrupt" in resp.data.lower() + + upload_dir = app.config["UPLOAD_FOLDER"] + remaining = os.listdir(upload_dir) + assert len(remaining) == 0 diff --git a/print/uploads/.gitkeep b/print/uploads/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/pyproject.toml b/pyproject.toml index 670a846..d4abdb1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -6,4 +6,5 @@ dependencies = [ "flask>=3.0", "gunicorn>=23.0", "pymongo>=4.0", + "pypdf>=4.0", ] diff --git a/uv.lock b/uv.lock index 9e21f02..9e73450 100644 --- a/uv.lock +++ b/uv.lock @@ -173,6 +173,7 @@ dependencies = [ { name = "flask" }, { name = "gunicorn" }, { name = "pymongo" }, + { name = "pypdf" }, ] [package.metadata] @@ -180,6 +181,7 @@ requires-dist = [ { name = "flask", specifier = ">=3.0" }, { name = "gunicorn", specifier = ">=23.0" }, { name = "pymongo", specifier = ">=4.0" }, + { name = "pypdf", specifier = ">=4.0" }, ] [[package]] @@ -252,6 +254,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/dc/cb/d9780b66939c4fc1f024bcc7be23a2abcfe06a9745ca8fa76dc73395482e/pymongo-4.17.0-cp314-cp314t-win_arm64.whl", hash = "sha256:9543d8f84c2e5608565c08ac679774811e6730770d8a645439b073422a4276fb", size = 1058526, upload-time = "2026-04-20T16:39:27.924Z" }, ] +[[package]] +name = "pypdf" +version = "6.10.2" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/7b/3f/9f2167401c2e94833ca3b69535bad89e533b5de75fefe4197a2c224baec2/pypdf-6.10.2.tar.gz", hash = "sha256:7d09ce108eff6bf67465d461b6ef352dcb8d84f7a91befc02f904455c6eea11d", size = 5315679, upload-time = "2026-04-15T16:37:36.978Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/0c/d6/1d5c60cc17bbdf37c1552d9c03862fc6d32c5836732a0415b2d637edc2d0/pypdf-6.10.2-py3-none-any.whl", hash = "sha256:aa53be9826655b51c96741e5d7983ca224d898ac0a77896e64636810517624aa", size = 336308, upload-time = "2026-04-15T16:37:34.851Z" }, +] + [[package]] name = "werkzeug" version = "3.1.8"