Files
shopdb-flask/plugins/printers/plugin.py
cproudlock 0436e8b0af Asset-side printer save, status CRUD, real printer types, network fix
- Printers save through the asset blueprint (PUT /printers) instead of the
  legacy machines API; restrict supply-model picker to printer models.
- Asset statuses get full CRUD (PUT/DELETE with in-use guard); canonical set.
- Printer types reseeded to a real classification set + list filter.
- Equipment accepts gauge/maintenance references.
- Fix network list emitting network_device instead of networkdevice (View 404).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-26 08:35:32 -04:00

251 lines
8.7 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.extensions import db
from shopdb.core.models.machine import MachineType
from shopdb.core.models import AssetType
from .models import PrinterData, Printer, PrinterType, ModelSupply
from .api import printers_bp, 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 [
PrinterData, # Legacy Machine-based
Printer, # New Asset-based
PrinterType, # New printer type classification
ModelSupply, # model -> toner/drum/waste part numbers
]
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', '')
# Register legacy blueprint for backward compatibility
app.register_blueprint(printers_bp, url_prefix='/api/printers/legacy')
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._ensure_legacy_machine_types()
logger.info("Printers plugin installed")
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 _ensure_legacy_machine_types(self) -> None:
"""Ensure basic printer machine types exist (legacy architecture)."""
printertypes = [
('Laser Printer', 'Printer', 'Standard laser printer'),
('Inkjet Printer', 'Printer', 'Inkjet printer'),
('Label Printer', 'Printer', 'Label/barcode printer'),
('Multifunction Printer', 'Printer', 'MFP with scan/copy/fax'),
('Plotter', 'Printer', 'Large format plotter'),
]
for name, category, description in printertypes:
existing = MachineType.query.filter_by(machinetype=name).first()
if not existing:
mt = MachineType(
machinetype=name,
category=category,
description=description,
icon='printer'
)
db.session.add(mt)
logger.debug(f"Created machine type: {name}")
db.session.commit()
def on_uninstall(self, app: Flask) -> None:
"""Called when plugin is uninstalled."""
logger.info("Printers plugin uninstalled")
def get_cli_commands(self) -> List:
"""Return CLI commands for this plugin."""
@click.group('printers')
def printerscli():
"""Printers plugin commands."""
pass
@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,
},
]