# app.py from flask import Flask, request, render_template, redirect, url_for, send_from_directory from print_utils import generate_qr_image, generate_text_image from printer_queue import print_job_queue, start_printer_worker from flask_httpauth import HTTPBasicAuth from werkzeug.security import generate_password_hash, check_password_hash from print_utils import save_history from print_utils import save_history, save_history_multi import os import json from datetime import datetime from picamera2 import Picamera2 from io import BytesIO from time import sleep import cv2 import numpy as np from flask import Response picam2 = Picamera2() picam2.configure(picam2.create_video_configuration(main={"size": (640, 480)})) picam2.start() os.makedirs("static", exist_ok=True) os.makedirs("history", exist_ok=True) HISTORY_DIR = "history" app = Flask(__name__) UPLOAD_FOLDER = 'static' app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER auth = HTTPBasicAuth() users = { "admin": generate_password_hash("123456") # 可自定义用户名密码 } from threading import Thread, Lock frame_lock = Lock() latest_frame = None def capture_frames(): global latest_frame while True: frame_rgb = picam2.capture_array() frame_bgr = cv2.cvtColor(frame_rgb, cv2.COLOR_RGB2BGR) ret, jpeg = cv2.imencode('.jpg', frame_bgr, [int(cv2.IMWRITE_JPEG_QUALITY), 90]) if ret: with frame_lock: latest_frame = jpeg.tobytes() sleep(0.01) # 启动摄像头捕获线程 capture_thread = Thread(target=capture_frames, daemon=True) capture_thread.start() def generate_video_stream(): global latest_frame while True: with frame_lock: frame = latest_frame if frame is not None: yield (b'--frame\r\n' b'Content-Type: image/jpeg\r\n\r\n' + frame + b'\r\n') else: # 如果还没准备好帧，等待 sleep(0.05) @auth.verify_password def verify_password(username, password): if username in users and check_password_hash(users.get(username), password): return username @app.route("/", methods=["GET", "POST"]) @auth.login_required def index(): if request.method == "POST": file = request.files.get("image") if file and file.filename != '': # ✅ 生成唯一文件名（如 upload_20250602_183045.png） timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"upload_{timestamp}.png" save_path = os.path.join(app.config['UPLOAD_FOLDER'], filename) file.save(save_path) size = request.form.get("size", "30x20").split("x") speed = int(request.form.get("speed", 4)) density = int(request.form.get("density", 8)) size_mm = (int(size[0]), int(size[1])) copies = int(request.form.get("copies", 1)) # ✅ 这里也要用空格缩进 for _ in range(copies): print_job_queue.put({ "image_path": save_path, "scale": 1.0, "size_mm": size_mm, "speed": speed, "density": density }) from print_utils import save_history_multi save_history_multi([save_path], { "size": f"{size_mm[0]}x{size_mm[1]}", "speed": speed, "density": density, "mode": "image" }) # ✅ 跳转时保留参数 return redirect(url_for("index", preview=filename, size=f"{size_mm[0]}x{size_mm[1]}", speed=speed, density=density)) # 页面初始加载时的默认参数 form_data = { "size": request.args.get("size", "40x30"), "speed": request.args.get("speed", 4), "density": request.args.get("density", 8), "font_size": request.args.get("font_size", 32), "text": request.args.get("text", ""), "mode": request.args.get("mode", "text"), "preview": request.args.get("preview", "test.png") } return render_template("index.html", **form_data) from html2image import Html2Image import bleach from datetime import datetime @app.route("/generate", methods=["POST"]) @auth.login_required def generate(): raw_text = request.form.get("text", "") mode = request.form.get("mode", "text") size = request.form.get("size", "30x20").split("x") speed = int(request.form.get("speed", 4)) density = int(request.form.get("density", 8)) font_size = int(request.form.get("font_size", 32)) copies = int(request.form.get("copies", 1)) size_mm = (int(size[0]), int(size[1])) if mode == "qr": from print_utils import generate_qr_image path = generate_qr_image(raw_text) image_paths = [path] else: # ✅ 清洗 HTML，防止 XSS 注入 allowed_tags = bleach.sanitizer.ALLOWED_TAGS + ["table", "thead", "tbody", "tr", "td", "th", "style", "span"] allowed_attrs = { "*": ["style", "class"], "td": ["colspan", "rowspan"], "th": ["colspan", "rowspan"] } safe_html = bleach.clean(raw_text, tags=allowed_tags, attributes=allowed_attrs) # ✅ 用 html2image 生成图片 hti = Html2Image(browser_executable='/usr/bin/chromium-browser') # 指定浏览器路径 timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"generated_{timestamp}.png" output_path = os.path.join("static", filename) hti.screenshot( html=f"{safe_html}", save_as=filename, size=(800, 600) # 可根据打印实际需求调整 ) image_paths = [output_path] for path in image_paths: for _ in range(copies): print_job_queue.put({ "image_path": path, "scale": 1.0, "size_mm": size_mm, "speed": speed, "density": density }) from print_utils import save_history_multi save_history_multi(image_paths, { "size": f"{size_mm[0]}x{size_mm[1]}", "speed": speed, "density": density, "font_size": font_size, "raw_text": raw_text, "mode": mode }) return redirect(url_for("index", size=f"{size_mm[0]}x{size_mm[1]}", speed=speed, density=density, font_size=font_size, text=raw_text, mode=mode)) @app.route("/history") @auth.login_required def history(): entries = [] for task_dir in sorted(os.listdir(HISTORY_DIR), reverse=True): task_path = os.path.join(HISTORY_DIR, task_dir) if os.path.isdir(task_path): meta_path = os.path.join(task_path, "metadata.json") if os.path.exists(meta_path): with open(meta_path, encoding="utf-8") as f: meta = json.load(f) # 用第一页图作为缩略图 first_page = os.path.join(task_dir, "page_1.png") entries.append({ "image": "/history/" + first_page.replace("\\", "/"), "meta": meta, "base": task_dir }) return render_template("history.html", entries=entries) @app.route("/history/") @auth.login_required def serve_history_image(filename): return send_from_directory(HISTORY_DIR, filename) @app.route("/reprint/") @auth.login_required def reprint(base): task_path = os.path.join(HISTORY_DIR, base) meta_path = os.path.join(task_path, "metadata.json") if not os.path.exists(meta_path): return "历史记录不存在", 404 with open(meta_path, encoding="utf-8") as f: meta = json.load(f) pages = meta.get("pages", []) for rel_path in pages: img_path = os.path.join(HISTORY_DIR, rel_path) print_job_queue.put({ "image_path": img_path, "scale": 1.0, "size_mm": tuple(map(int, meta.get("size", "30x20").split("x"))), "speed": meta.get("speed", 4), "density": meta.get("density", 8) }) return redirect("/history") @app.route("/reprint/latest") @auth.login_required def reprint_latest(): latest_time = None latest_dir = None for d in os.listdir(HISTORY_DIR): task_path = os.path.join(HISTORY_DIR, d) meta_path = os.path.join(task_path, "metadata.json") if os.path.isdir(task_path) and os.path.exists(meta_path): try: t = datetime.strptime(d, "%Y%m%d_%H%M%S") if latest_time is None or t > latest_time: latest_time = t latest_dir = d except ValueError: continue if latest_dir: return redirect(f"/reprint/{latest_dir}") else: return "没有可重打印的历史记录", 404 @app.route("/video_feed") @auth.login_required def video_feed(): return Response(generate_video_stream(), mimetype="multipart/x-mixed-replace; boundary=frame") if __name__ == "__main__": start_printer_worker() # 启动打印线程 app.run(host="0.0.0.0", port=5000, debug=False)