b211e817d5
Poll Zabbix for toner levels on a schedule and email/webhook on a downward crossing. Warning fires at or below the warning threshold (default 5%), critical at the critical threshold (default 0%); both thresholds are settings. State lives in printersupplyalerts so an alert fires once per crossing and re-arms after a refill. Recipients mirror the printedparts pattern: plugin-scoped shopdb users + roles + free-text emails (falling back to the site alert_recipients), and a chosen support team's webhook (falling back to the site alert_webhook_url). - PrinterSupplyAlert model + migration printers0002supplyalerts - alerttier(remaining, warning, critical) + check_supplies poller - flask printers check-toner-alerts CLI (run via scheduled task/cron) - printers alert settings + Low-Toner Alerts settings page - 7 tests: tier boundaries, once-per-crossing + re-arm, toner-only scope, custom thresholds, support-team webhook routing
132 lines
4.4 KiB
Python
132 lines
4.4 KiB
Python
"""Printer supply classification and part-number lookup.
|
|
|
|
Part numbers now live in the modelsupplies table (see seed_supplies.py),
|
|
managed through the API/UI. This module keeps the runtime logic that is not
|
|
per-model data: classifying a reported level into ok/low/critical (waste
|
|
cartridges invert), deriving supply type and color from a Zabbix item name,
|
|
and reading the matching part numbers out of the database.
|
|
"""
|
|
|
|
from typing import Dict, List, Optional
|
|
|
|
|
|
# display thresholds (percent remaining) for the ok/low/critical report badge
|
|
CRITICAL_THRESHOLD = 5
|
|
LOW_THRESHOLD = 10
|
|
|
|
# toner email-alert tiers (percent remaining). Distinct from the display
|
|
# badge above: a warning email fires at or below TONER_WARNING_THRESHOLD, a
|
|
# critical email at or below TONER_CRITICAL_THRESHOLD (empty).
|
|
TONER_WARNING_THRESHOLD = 5
|
|
TONER_CRITICAL_THRESHOLD = 0
|
|
|
|
|
|
def alerttier(remaining: float, warning: float = TONER_WARNING_THRESHOLD,
|
|
critical: float = TONER_CRITICAL_THRESHOLD) -> str:
|
|
"""Map a toner percent-remaining to an email-alert tier.
|
|
|
|
critical at or below the critical threshold (default 0 = empty), warning at
|
|
or below the warning threshold (default 5), else ok. Both are configurable
|
|
(printers_alert_critical_threshold / printers_alert_warning_threshold)."""
|
|
if remaining <= critical:
|
|
return 'critical'
|
|
if remaining <= warning:
|
|
return 'warning'
|
|
return 'ok'
|
|
|
|
|
|
# tier severity rank; an alert fires only when the rank increases (worsens)
|
|
TIER_RANK = {'ok': 0, 'warning': 1, 'critical': 2}
|
|
|
|
|
|
def derivesupplytype(name: str) -> str:
|
|
"""Map a Zabbix item name to a supply type."""
|
|
lowername = (name or "").lower()
|
|
if "waste" in lowername:
|
|
return "waste"
|
|
if "drum" in lowername or "imaging" in lowername:
|
|
return "drum"
|
|
if "maintenance" in lowername or "fuser" in lowername:
|
|
return "maintenance"
|
|
return "toner"
|
|
|
|
|
|
def derivecolor(name: str, tagcolor: Optional[str] = None) -> str:
|
|
"""Best-effort supply color from a Zabbix color tag, then the item name."""
|
|
color = (tagcolor or "").lower()
|
|
if "black" in color:
|
|
return "black"
|
|
if color in ("cyan", "magenta", "yellow"):
|
|
return color
|
|
if color in ("grey", "gray"):
|
|
return "gray"
|
|
|
|
lowername = (name or "").lower()
|
|
for candidate in ("cyan", "magenta", "yellow", "black"):
|
|
if candidate in lowername:
|
|
return candidate
|
|
return "none"
|
|
|
|
|
|
def classifysupply(level: float, name: str, vendor: Optional[str]) -> Dict:
|
|
"""Classify one supply item into ok/low/critical.
|
|
|
|
Waste cartridge fill is inverted vs a toner level: a full waste cartridge
|
|
is bad. Standard vendors report waste as percent FULL (high = bad). Xerox
|
|
EC/AltaLink series report waste as percent capacity REMAINING (low = bad),
|
|
same direction as toner. Normalise everything to percent remaining first.
|
|
"""
|
|
lowername = (name or "").lower()
|
|
iswaste = "waste" in lowername
|
|
isdrum = "drum" in lowername or "imaging" in lowername
|
|
isxerox = bool(vendor) and "xerox" in vendor.lower()
|
|
|
|
if iswaste and not isxerox:
|
|
remaining = 100 - level
|
|
else:
|
|
remaining = level
|
|
|
|
if remaining <= CRITICAL_THRESHOLD:
|
|
status = "critical"
|
|
elif remaining <= LOW_THRESHOLD:
|
|
status = "low"
|
|
else:
|
|
status = "ok"
|
|
|
|
return {
|
|
"status": status,
|
|
"remaining": round(remaining, 1),
|
|
"iswaste": iswaste,
|
|
"isdrum": isdrum,
|
|
}
|
|
|
|
|
|
def lookupsupplies(modelnumberid: Optional[int], color: str,
|
|
supplytype: str) -> List[Dict]:
|
|
"""Part-number options for a model + color + supply type, from the DB.
|
|
|
|
Returns every matching capacity tier (standard / high / metered / ...) so
|
|
the report can show all reorder options, like the classic report did.
|
|
"""
|
|
if not modelnumberid:
|
|
return []
|
|
|
|
from ..models import ModelSupply
|
|
|
|
query = ModelSupply.query.filter_by(
|
|
modelnumberid=modelnumberid,
|
|
supplytype=supplytype,
|
|
isactive=True,
|
|
)
|
|
# toners are color-specific; drum/waste/maintenance are not
|
|
if supplytype == 'toner' and color and color != 'none':
|
|
query = query.filter_by(color=color)
|
|
|
|
rows = query.order_by(ModelSupply.capacitytier).all()
|
|
return [{
|
|
'partnumber': row.partnumber,
|
|
'marketingname': row.marketingname,
|
|
'capacitytier': row.capacitytier,
|
|
'pageyield': row.pageyield,
|
|
} for row in rows]
|