ske087
bff6d70b17
Aplicatie Flask + SQLite pentru productie alimentara: cantarirea componentelor unui produs finit si tiparirea etichetei prin QZ Tray. - produse finite (retete) formate din componente cantarite - designer vizual de etichete (mm/px, zona sigura, campuri variabile, cod QR cu continut din mai multe bucati) - legare produs finit -> sablon de eticheta - navigare pe doua taburi: Cantarire / Setari - administrare: componente, produse finite, sabloane, campuri, istoric - backup, resetare baza de date, checklist de pregatire, /healthz - CLI de administrare si serviciu systemd pentru productie - vendor/syncpulse-scale-bridge (GPL-3.0-or-later) pentru citirea cantarului
58 lines
1.9 KiB
Python
58 lines
1.9 KiB
Python
"""Logging: jurnal de acces lizibil, fara zgomotul polling-ului.
|
|
|
|
Flask/werkzeug scrie un rand pentru fiecare cerere. Pagina de cantarire
|
|
interogheaza greutatea de mai multe ori pe secunda, deci jurnalul devine
|
|
inutilizabil. Aici:
|
|
- cererile de polling (/api/weight) si fisierele statice nu sunt logate;
|
|
- erorile (4xx/5xx) sunt pastrate intotdeauna;
|
|
- jurnalul aplicatiei merge la stderr, cu timestamp.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import re
|
|
import sys
|
|
|
|
QUIET_PREFIXES = (
|
|
"GET /api/weight",
|
|
"GET /static/",
|
|
"GET /favicon.ico",
|
|
)
|
|
|
|
# Werkzeug access line: ... "GET /path HTTP/1.1" 200 -
|
|
_STATUS_RE = re.compile(r'"\s(\d{3})\s')
|
|
|
|
|
|
class _QuietPolling(logging.Filter):
|
|
"""Drop access-log records for polling and static assets, keep errors."""
|
|
|
|
def filter(self, record: logging.LogRecord) -> bool:
|
|
try:
|
|
message = record.getMessage()
|
|
except Exception: # noqa: BLE001 - never let logging break a request
|
|
return True
|
|
if not any(prefix in message for prefix in QUIET_PREFIXES):
|
|
return True
|
|
match = _STATUS_RE.search(message)
|
|
if match is None:
|
|
return True
|
|
# Keep only failures (4xx / 5xx); 2xx and 304 are noise.
|
|
return not match.group(1).startswith(("2", "3"))
|
|
|
|
|
|
def configure_logging(level: int = logging.INFO) -> None:
|
|
handler = logging.StreamHandler(sys.stderr)
|
|
handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)-7s %(name)s: %(message)s",
|
|
"%H:%M:%S"))
|
|
root = logging.getLogger()
|
|
if not root.handlers:
|
|
root.addHandler(handler)
|
|
root.setLevel(level)
|
|
|
|
werkzeug = logging.getLogger("werkzeug")
|
|
werkzeug.setLevel(logging.INFO)
|
|
if not any(isinstance(f, _QuietPolling) for f in werkzeug.filters):
|
|
werkzeug.addFilter(_QuietPolling())
|
|
|
|
logging.getLogger("scale_bridge").setLevel(logging.WARNING)
|