Files
shopdb-flask/plugins/printers/plugin.py
cproudlock 42c050a9f4
Some checks failed
CI / backend (push) Failing after 8s
CI / naming (push) Successful in 2s
CI / frontend (push) Successful in 8s
CI / migrations-mysql (push) Failing after 7s
dashboard: tighten the printer, warranty and notification cards
PRINTER CARD is now supplies at or below 5 percent, a new
printers_dashboardpercent setting. The report and the card want different
scopes: the report lists anything the thresholds call low, which is right for
planning an order, while the dashboard is asking what to walk out and change
today - and a cartridge at 18 percent is not that. Lowest first.

PRINTER LOCATION WAS ALWAYS EMPTY. The lookup went through db.session.get on
locationid and produced nothing even where a location is set; the printers list
has always read it through the asset relationship, so the card does too now.

WARRANTY ROWS are identified the way the floor identifies them: the PC's
hostname and the MACHINE it drives, reusing the same lookup behind the warranty
page's machine column so the board and the report cannot disagree about which
bay a PC belongs to. No dates - expired or expiring is the whole decision when
scanning a board, and the exact day belongs on the report you order from.

NOTIFICATIONS are stacked: the type in full on one line, the message beneath,
trimmed to 100 characters with the rest on hover. Inline, the type was
truncated to make room for prose that was then truncated anyway, and neither
read. A tooltip is omitted when the text was not trimmed, because one repeating
what is already on screen is noise.

Two general additions: layout: 'stacked' on a card, and map.detailtooltip.
2026-08-11 15:57:14 -04:00

365 lines
13 KiB
Python

"""Printers plugin main class."""
import json
import logging
from pathlib import Path
from typing import List, Dict, Optional, Type
from flask import Flask, Blueprint
import click
from shopdb.plugins.base import BasePlugin, PluginMeta
from shopdb.api import db, AssetType
from .models import (
Printer, PrinterType, ModelSupply, PrinterDriver, PrinterSupplyAlert
)
from .api import printers_asset_bp
from .services import ZabbixService
logger = logging.getLogger(__name__)
class PrintersPlugin(BasePlugin):
"""
Printers plugin - manages printer assets.
Supports both legacy Machine-based architecture and new Asset-based architecture:
- Legacy: PrinterData table linked to machines
- New: Printer table linked to assets
Features:
- PrinterType classification
- Windows/network naming
- Zabbix integration for real-time supply level lookups
"""
def __init__(self):
self._manifest = self._load_manifest()
self._zabbixservice = None
def _load_manifest(self) -> Dict:
"""Load plugin manifest from JSON file."""
manifestpath = Path(__file__).parent / 'manifest.json'
if manifestpath.exists():
with open(manifestpath, 'r') as f:
return json.load(f)
return {}
@property
def meta(self) -> PluginMeta:
"""Return plugin metadata."""
return PluginMeta(
name=self._manifest.get('name', 'printers'),
version=self._manifest.get('version', '2.0.0'),
description=self._manifest.get(
'description',
'Printer management with Zabbix integration'
),
author=self._manifest.get('author', 'ShopDB Team'),
dependencies=self._manifest.get('dependencies', []),
core_version=self._manifest.get('core_version', '>=1.0.0'),
api_prefix=self._manifest.get('api_prefix', '/api/printers'),
)
def get_blueprint(self) -> Optional[Blueprint]:
"""
Return Flask Blueprint with API routes.
Returns the new Asset-based blueprint.
Legacy Machine-based blueprint is registered separately in init_app.
"""
return printers_asset_bp
def get_models(self) -> List[Type]:
"""Return list of SQLAlchemy model classes."""
return [
Printer, # Asset-based
PrinterType, # printer type classification
PrinterDriver, # driver links (SMB/HTTP)
ModelSupply, # model -> toner/drum/waste part numbers
PrinterSupplyAlert, # per-printer toner alert crossing state
]
def get_services(self) -> Dict[str, Type]:
"""Return plugin services."""
return {
'zabbix': ZabbixService,
}
@property
def zabbixservice(self) -> ZabbixService:
"""Get Zabbix service instance."""
if self._zabbixservice is None:
self._zabbixservice = ZabbixService()
return self._zabbixservice
def init_app(self, app: Flask, db_instance) -> None:
"""Initialize plugin with Flask app."""
app.config.setdefault('ZABBIX_URL', '')
app.config.setdefault('ZABBIX_TOKEN', '')
logger.info(f"Printers plugin initialized (v{self.meta.version})")
def on_install(self, app: Flask) -> None:
"""Called when plugin is installed."""
with app.app_context():
self._ensure_asset_type()
self._ensure_printer_types()
logger.info("Printers plugin installed")
def get_settings_defaults(self) -> List[dict]:
"""Low-toner alert settings.
The framework seeds these at install, at enable, and on every
`flask plugin upgrade-all`, so a key added in a later version reaches a
site that installed an earlier one.
"""
return [
{
'key': 'printers_alert_email',
'value': '',
'valuetype': 'string',
'category': 'printers',
'description': 'Comma-separated low-toner alert recipients; '
'empty uses the site alert_recipients',
},
{
'key': 'printers_alert_userids',
'value': '',
'valuetype': 'string',
'category': 'printers',
'description': 'Comma-separated shopdb user ids whose account '
'emails receive low-toner alerts',
},
{
'key': 'printers_alert_roleids',
'value': '',
'valuetype': 'string',
'category': 'printers',
'description': 'Comma-separated role ids; every active member '
'of these roles receives low-toner alerts',
},
{
'key': 'printers_alert_supportteamid',
'value': '',
'valuetype': 'string',
'category': 'printers',
'description': 'Support team whose webhook receives low-toner '
'alerts; empty uses the site alert_webhook_url',
},
{
'key': 'printers_alert_warning_threshold',
'value': '5',
'valuetype': 'integer',
'category': 'printers',
'description': 'Toner percent remaining at or below which a '
'warning email fires',
},
{
'key': 'printers_alert_critical_threshold',
'value': '0',
'valuetype': 'integer',
'category': 'printers',
'description': 'Toner percent remaining at or below which a '
'critical email fires',
},
]
def _ensure_asset_type(self) -> None:
"""Ensure printer asset type exists."""
existing = AssetType.query.filter_by(assettype='printer').first()
if not existing:
at = AssetType(
assettype='printer',
pluginname='printers',
tablename='printers',
description='Printers (laser, inkjet, label, MFP, plotter)',
icon='printer'
)
db.session.add(at)
logger.debug("Created asset type: printer")
db.session.commit()
def _ensure_printer_types(self) -> None:
"""Ensure basic printer types exist (new architecture)."""
printer_types = [
('Laser', 'Standard laser printer', 'printer'),
('Inkjet', 'Inkjet printer', 'printer'),
('Label', 'Label/barcode printer', 'barcode'),
('Card', 'ID / card printer', 'id-card'),
('MFP', 'Multifunction printer with scan/copy/fax', 'printer'),
('Plotter', 'Large format plotter', 'drafting-compass'),
('Thermal', 'Thermal printer', 'temperature-high'),
('Dot Matrix', 'Dot matrix printer', 'th'),
('Other', 'Other printer type', 'printer'),
]
for name, description, icon in printer_types:
existing = PrinterType.query.filter_by(printertype=name).first()
if not existing:
pt = PrinterType(
printertype=name,
description=description,
icon=icon
)
db.session.add(pt)
logger.debug(f"Created printer type: {name}")
db.session.commit()
def on_uninstall(self, app: Flask) -> None:
"""Called when plugin is uninstalled."""
logger.info("Printers plugin uninstalled")
def get_settings_cards(self) -> List[dict]:
return [
{
'group': 'Printers',
'to': '/settings/printer-alerts',
'icon': 'bell',
'title': 'Low-Toner Alerts',
'description': 'Who gets warning (5%) and critical (0%) toner '
'emails, and which support team webhook',
'position': 48,
},
]
def get_cli_commands(self) -> List:
"""Return CLI commands for this plugin."""
@click.group('printers')
def printerscli():
"""Printers plugin commands."""
pass
@printerscli.command('check-toner-alerts')
def checktoneralerts():
"""Poll Zabbix for all printers and email/webhook low-toner crossings.
Run on a schedule (scheduled task / cron). Fires a warning at or
below 5 percent and a critical at 0 percent, once per crossing."""
from flask import current_app
from .services import check_supplies
with current_app.app_context():
summary = check_supplies()
click.echo(
f"Toner poll: {summary['polled']}/{summary['printers']} "
f"printers reachable, {summary['alerts']} alert(s) sent, "
f"{summary['rearmed']} re-armed.")
@printerscli.command('check-supplies')
@click.argument('ip')
def checksupplies(ip):
"""Check supply levels for a printer by IP (via Zabbix)."""
from flask import current_app
with current_app.app_context():
service = ZabbixService()
if not service.isconfigured:
click.echo('Error: Zabbix not configured. Set ZABBIX_URL and ZABBIX_TOKEN.')
return
supplies = service.getsuppliesbyip(ip)
if not supplies:
click.echo(f'No supply data found for {ip}')
return
click.echo(f'Supply levels for {ip}:')
for supply in supplies:
click.echo(f" {supply['name']}: {supply['level']}%")
@printerscli.command('seed-supplies')
def seedsuppliescommand():
"""Seed corrected model->toner part numbers into modelsupplies."""
from flask import current_app
from .services import seedsupplies
with current_app.app_context():
summary = seedsupplies()
click.echo(
f"Seeded supplies: {summary['suppliesadded']} added across "
f"{summary['modelstouched']} models."
)
return [printerscli]
def get_settings_defaults(self) -> List[Dict]:
"""Settings this plugin owns for the dashboard card."""
return [
{
'key': 'printers_dashboardpercent',
'value': '5',
'valuetype': 'integer',
'category': 'printers',
'description': 'Supply percentage at or below which a printer '
'appears on the dashboard. Tighter than the '
'low-supplies report, which is for planning an '
'order rather than walking out to change one.',
},
]
def get_dashboard_widgets(self) -> List[Dict]:
"""Dashboard card: printers needing a cartridge.
Replaces a declaration naming a component nobody wrote. Reuses the
low-supplies query and its cache - a Zabbix round-trip per printer on
every dashboard load would make this the slowest page in the app.
"""
return [
{
'id': 'printers-supplies',
'viewall': '/reports/toner',
'title': 'Printer supplies',
'endpoint': '/api/printers/dashboard/supplies',
'render': 'exceptions',
'severity': 'warning',
'permission': 'printers.view',
'empty': 'hide',
'position': 40,
'map': {
'title': 'printername',
# Location on hover rather than on the line: it is context
# for "where do I walk", not part of the finding, and it
# was the text pushing rows past the card edge.
'titletooltip': 'location',
'chips': 'supplies',
'link': '/printers/{printerid}',
},
},
]
def get_navigation_items(self) -> List[Dict]:
"""Return navigation menu items."""
return [
{
'name': 'Printers',
'icon': 'printer',
'route': '/printers',
'position': 20,
},
]
def get_reports(self) -> List[Dict]:
"""Return report card definitions for the Reports hub."""
return [
{
'id': 'toner',
'name': 'Toner Report',
'description': 'Printers with low or critical toner/supply levels',
'category': 'printers',
'route': '/reports/toner',
},
]
def get_permissions(self) -> List:
"""Return the RBAC permissions this plugin owns."""
return [
('printers.view', 'View printers', 'printers'),
('printers.create', 'Create printers', 'printers'),
('printers.edit', 'Edit printers', 'printers'),
('printers.delete', 'Delete printers', 'printers'),
]