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
307 lines
11 KiB
Python
307 lines
11 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, Setting
|
|
|
|
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()
|
|
self._seed_settings()
|
|
logger.info("Printers plugin installed")
|
|
|
|
def on_enable(self, app: Flask) -> None:
|
|
# Idempotent re-seed so settings added in later versions reach sites
|
|
# that installed earlier (enable runs on every upgrade cycle).
|
|
with app.app_context():
|
|
self._seed_settings()
|
|
|
|
def _seed_settings(self) -> None:
|
|
"""Seed low-toner alert recipient settings (idempotent)."""
|
|
defaults = [
|
|
('printers_alert_email', '', 'string',
|
|
'Comma-separated low-toner alert recipients; empty uses the '
|
|
'site alert_recipients'),
|
|
('printers_alert_userids', '', 'string',
|
|
'Comma-separated shopdb user ids whose account emails receive '
|
|
'low-toner alerts'),
|
|
('printers_alert_roleids', '', 'string',
|
|
'Comma-separated role ids; every active member of these roles '
|
|
'receives low-toner alerts'),
|
|
('printers_alert_supportteamid', '', 'string',
|
|
'Support team whose webhook receives low-toner alerts; empty '
|
|
'uses the site alert_webhook_url'),
|
|
('printers_alert_warning_threshold', '5', 'integer',
|
|
'Toner percent remaining at or below which a warning email fires'),
|
|
('printers_alert_critical_threshold', '0', 'integer',
|
|
'Toner percent remaining at or below which a critical email fires'),
|
|
]
|
|
for key, value, valuetype, description in defaults:
|
|
if Setting.get(key) is None:
|
|
Setting.set(key, value, valuetype=valuetype,
|
|
category='printers', description=description)
|
|
db.session.commit()
|
|
|
|
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_dashboard_widgets(self) -> List[Dict]:
|
|
"""Return dashboard widget definitions."""
|
|
return [
|
|
{
|
|
'name': 'Printer Status',
|
|
'component': 'PrinterStatusWidget',
|
|
'endpoint': '/api/printers/dashboard/summary',
|
|
'size': 'medium',
|
|
'position': 10,
|
|
},
|
|
]
|
|
|
|
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'),
|
|
]
|