"""Mini-HTTP-Server für Handy-basiertes Backcover-Upload.
Startet einen lokalen HTTP-Server (kein HTTPS nötig, da nur file-input,
kein getUserMedia), der eine mobile Upload-Seite ausliefert und
hochgeladene Fotos in einer Queue bereitstellt.
"""
from __future__ import annotations
import base64
import json
import logging
import queue
import socket
import tempfile
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path
logger = logging.getLogger(__name__)
_UPLOAD_HTML = """\
CD-Backcover
CD-Backcover hochladen
Fotografiere die Rückseite der CD-Hülle und lade das Bild hoch.
"""
def _get_local_ip() -> str:
"""Ermittelt die lokale LAN-IP-Adresse."""
try:
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
s.connect(("8.8.8.8", 80))
return s.getsockname()[0]
except OSError:
return "127.0.0.1"
def print_qr(url: str) -> None:
"""Gibt einen QR-Code für die URL im Terminal aus.
Zeigt alternativ nur die URL, wenn qrcode nicht installiert ist.
"""
try:
import qrcode # type: ignore[import-untyped]
qr = qrcode.QRCode(border=1)
qr.add_data(url)
qr.make(fit=True)
qr.print_ascii(invert=True)
except ImportError:
pass # qrcode optional — URL wird sowieso separat ausgegeben
class ScannerServer:
"""Lokaler HTTP-Server für Backcover-Foto-Upload vom Handy.
Beispiel:
server = ScannerServer(port=8765)
server.start()
print(server.url())
photo_path = server.get_photo(timeout=300)
server.stop()
"""
def __init__(self, port: int = 8765, upload_dir: Path | None = None) -> None:
self._port = port
self._queue: queue.Queue[Path] = queue.Queue()
self._upload_dir = upload_dir or Path(tempfile.mkdtemp(prefix="ms_scan_"))
self._server: HTTPServer | None = None
self._thread: threading.Thread | None = None
# ------------------------------------------------------------------
# Public API
# ------------------------------------------------------------------
def start(self) -> None:
"""Startet den HTTP-Server im Hintergrund-Thread."""
server_instance = self
class _Handler(BaseHTTPRequestHandler):
def log_message(self, fmt: str, *args: object) -> None: # suppress output
logger.debug("Scanner HTTP: " + fmt, *args)
def do_GET(self) -> None: # noqa: N802
if self.path in ("/", "/index.html"):
body = _UPLOAD_HTML.encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
else:
self.send_response(404)
self.end_headers()
def do_POST(self) -> None: # noqa: N802
if self.path != "/upload":
self.send_response(404)
self.end_headers()
return
length = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(length)
try:
data = json.loads(body)
img_data: str = data["image"] # "data:;base64,"
# Mime-Typ und Erweiterung bestimmen
mime = "image/jpeg"
if img_data.startswith("data:"):
mime = img_data.split(";")[0][5:]
ext = {
"image/jpeg": ".jpg",
"image/png": ".png",
"image/webp": ".webp",
}.get(mime, ".jpg")
_, b64 = img_data.split(",", 1)
img_bytes = base64.b64decode(b64)
path = server_instance._upload_dir / f"back{ext}"
path.write_bytes(img_bytes)
logger.info("Backcover hochgeladen: %s (%d bytes)", path, len(img_bytes))
server_instance._queue.put(path)
resp = json.dumps({"status": "ok"}).encode()
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(resp)))
self.end_headers()
self.wfile.write(resp)
except Exception as exc:
logger.warning("Upload-Fehler: %s", exc)
resp = json.dumps({"status": "error", "message": str(exc)}).encode()
self.send_response(400)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(resp)))
self.end_headers()
self.wfile.write(resp)
self._server = HTTPServer(("0.0.0.0", self._port), _Handler)
self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)
self._thread.start()
logger.info("Scanner-Server gestartet: %s", self.url())
def stop(self) -> None:
"""Beendet den HTTP-Server."""
if self._server:
self._server.shutdown()
self._server = None
logger.info("Scanner-Server gestoppt.")
def get_photo(self, timeout: float | None = None) -> Path | None:
"""Gibt den Pfad zum hochgeladenen Foto zurück (blockierend mit Timeout).
Args:
timeout: Wartezeit in Sekunden. None = unbegrenzt. 0 = sofort.
Returns:
Pfad zur Bilddatei oder None wenn Timeout abgelaufen.
"""
try:
if timeout == 0:
return self._queue.get_nowait()
return self._queue.get(timeout=timeout)
except queue.Empty:
return None
def url(self) -> str:
"""Gibt die URL des Servers im LAN zurück."""
return f"http://{_get_local_ip()}:{self._port}"