"""Printers plugin main class.""" import json import logging from pathlib import Path import re from typing import List, Dict, Optional, Type from flask import Flask, Blueprint, g import click from shopdb.plugins.base import BasePlugin, PluginMeta from shopdb.api import db, AssetType from .models import ( Printer, PrinterType, ModelSupply, PrinterDriver, PrinterSupplyAlert, PrinterObservedQueue ) from .api import printers_asset_bp from .services import ZabbixService logger = logging.getLogger(__name__) # Widths of the text columns on printerobservedqueues. A Windows queue name # tops out well below this, but a report is unattended machine input: one # oversized string must not turn into a 500 the bay retries every cycle. OBSERVEDTEXTLIMIT = 255 def _observed_text(value, fieldname, warnings): """Trim one reported string to what the column holds, or None if blank.""" if value is None: return None text = str(value).strip() if not text: return None if len(text) > OBSERVEDTEXTLIMIT: warnings.append('truncated {} longer than {} characters'.format( fieldname, OBSERVEDTEXTLIMIT)) text = text[:OBSERVEDTEXTLIMIT] return text def _observed_bool(value): """Coerce a reported flag to bool. PowerShell's ConvertTo-Json emits real booleans, but hand-built payloads and older clients send 'True'/'true'/1, and a bare truthiness test would read the string 'False' as a default printer. """ if isinstance(value, bool): return value if isinstance(value, (int, float)): return bool(value) if isinstance(value, str): return value.strip().lower() in ('true', '1', 'yes') return False 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 PrinterObservedQueue, # what a bay reports it ACTUALLY has ] 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 plus the dashboard threshold. 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. ONE definition only. There were two, and the later one shadowed this list outright: every alert setting below was silently never declared, so the alerts settings page wrote keys the plugin did not own and a new site seeded none of them. Add keys here; do not add a second method. """ 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', }, { '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 _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: level = supply.get('level') shown = 'unknown' if level is None else f'{level}%' click.echo(f" {supply['name']}: {shown}") @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]: """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', 'shortname': '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', # Hovering the name shows the floor-plan preview, the # same component the printer's own page uses: a # location name tells you the room, the map tells you # where to walk. 'maphover': {'x': 'mapx', 'y': 'mapy', 'label': 'printername'}, 'chips': 'supplies', 'link': '/printers/{printerid}', }, }, ] def get_navigation_items(self) -> List[Dict]: """Return navigation menu items.""" return [ { 'name': 'Printers', 'icon': 'printer', 'route': '/printers', 'permission': 'printers.view', 'position': 20, 'section': 'assets', }, ] 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', }, { # Separate from the toner report on purpose: that one is an # exceptions list a tech acts on today, this is an ordering # view read monthly, and it runs a heavier history query. 'id': 'tonerforecast', 'name': 'Toner Forecast', 'description': 'Estimated days until each printer runs out, ' 'and how many cartridges it has been through', 'category': 'printers', 'route': '/reports/toner-forecast', }, ] 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'), ] # ---- ADR-006 collector contract ------------------------------------- def get_collector_schema(self) -> Optional[dict]: """What a bay reports it ACTUALLY has (POST /api/collector/printers). The observed half of the printer story. ShopDB already knows what a host SHOULD have (/api/printers/for-host); this is what enumerating the host found, kept apart from the assignment so drift stays visible. Declaring this schema is what registers the endpoint - the dispatcher in shopdb/core/api/collector.py discovers it, and brings the collector key / managed-token auth and the audit row with it. """ return { 'identityfield': 'hostname', 'acceptsagentcredential': True, 'fields': { 'type': 'object', 'required': ['hostname', 'queues'], 'properties': { 'hostname': { 'type': 'string', 'description': 'Reporting PC (COMPUTERNAME or its ' 'FQDN). The identity of the report: ' 'the PC asset is resolved from it, but ' 'the rows are keyed by the name, so an ' 'unenrolled bay still reports.', }, 'queues': { 'type': 'array', 'description': "Every real print queue on the host. " "This REPLACES the host's previous set, " "so an empty array is a valid report " "that clears it. A client whose " "enumeration FAILED must send nothing " "at all - never an empty array.", 'items': { 'type': 'object', 'required': ['queuename'], 'properties': { 'queuename': { 'type': 'string', 'description': 'Windows printer name.', }, 'drivername': { 'type': 'string', 'description': 'Driver name verbatim, as ' 'the INF spells it.', }, 'portname': { 'type': 'string', 'description': 'Windows port name.', }, 'portaddress': { 'type': 'string', 'description': 'PrinterHostAddress of a ' 'TCP/IP port - an IP or ' 'FQDN. The primary key for ' 'matching this queue to a ' 'printer asset; omit it for ' 'a non-TCP port.', }, 'isdefault': { 'type': 'boolean', 'description': 'True on the one queue that ' 'is the default printer.', }, 'isshared': { 'type': 'boolean', 'description': 'True when the queue is ' 'shared off this PC.', }, }, }, }, 'observedat': { 'type': 'string', 'format': 'date-time', 'description': 'Accepted and ignored. The server stamps ' 'observedat at ingest, so a bay with a ' 'wrong clock cannot report itself fresh ' 'or stale.', }, }, }, } def apply_collector_payload(self, payload: dict) -> dict: """Replace one host's observed queue set (ADR-006). THIS NEVER WRITES AN ASSIGNMENT. Observed and assigned are separate tables on purpose: the moment a drifted bay's report is allowed to become what that bay is told to install, enforcement stops meaning anything. Seeding an assignment from observed state is a human action through PUT /api/printers/assignments/for-asset/. Replace, not append: this is current state, so the latest report is the whole truth for that host. Nothing is matched to a printer asset here either - resolution happens at read time, so a printer added to ShopDB tomorrow matches yesterday's report without the bay reporting again. """ from datetime import datetime, timezone warnings = [] hostname = (payload.get('hostname') or '').strip() if not hostname: raise ValueError('hostname is required') queues = payload.get('queues') if queues is None: # Absent and empty are NOT the same thing. [] is a host that # genuinely has no queues and clears its rows; a missing key is a # malformed report, and treating it as a wipe would let one client # bug erase the observed state of the fleet host by host. raise ValueError('queues is required; send an empty array for a ' 'host with no queues') if not isinstance(queues, list): raise ValueError('queues must be an array') # One stamp for the whole report, so "when did this bay last report" # reads off any row of it rather than a MAX over the set. observedat = datetime.now(timezone.utc).replace(tzinfo=None) # Resolved BEFORE the rows are written: assetid is a convenience for # the read paths, and hostname stays the identity that the replace # keys on, so an unresolved host still records everything it reported. assetid = self._observed_assetid(hostname, warnings) # A per-PC credential reports for its own PC only. if (getattr(g, 'agentauth', None) == 'credential' and assetid != getattr(g, 'agentassetid', None)): raise PermissionError( 'The payload names a different endpoint than the credential') # Case-folded on both sides: one script sends COMPUTERNAME uppercase # and another the lowercase FQDN, and MySQL forgives that while SQLite # does not. Uncompared, the replace would leave the other spelling's # rows in place and the host would appear to have every queue twice. # # A bulk delete so the DELETE reaches the database BEFORE the inserts # below: a re-report repeats the same queue names, and the # (hostname, queuename) unique index rejects the new rows if the old # ones are still there. synchronize_session='fetch' costs one select # and keeps the session's identity map honest, so a caller that read # these rows earlier in the same request does not keep deleted ones. # Every spelling of this host, not just the one it sent. The READ path # treats a short name and its FQDN as the same machine, so a delete that # matched only the exact string would leave the other spelling's rows # behind and the host would appear to have every queue twice - the bug # this replace exists to prevent. A PC that enrolls short and later # reports fully qualified is normal, not exotic. shortname = hostname.lower().split('.')[0] predicate = db.or_( db.func.lower(PrinterObservedQueue.hostname) == hostname.lower(), db.func.lower(PrinterObservedQueue.hostname) == shortname) if re.match(r'^[a-z0-9-]+$', shortname): # Prefix match only for a plain name, as the read path does: a # wildcard built from arbitrary input would delete another PC's rows. predicate = db.or_( predicate, db.func.lower(PrinterObservedQueue.hostname).like(shortname + '.%')) db.session.query(PrinterObservedQueue).filter(predicate).delete( synchronize_session='fetch') seennames = set() defaultqueue = None stored = 0 for entry in queues: if not isinstance(entry, dict): warnings.append('ignored a queue entry that was not an object') continue queuename = _observed_text(entry.get('queuename'), 'queuename', warnings) if not queuename: warnings.append('ignored a queue with no queuename') continue if queuename.lower() in seennames: # Windows cannot hold two queues of one name on a host, so this # is a doubled line in the report. Dropping it keeps the # (hostname, queuename) unique index from failing the whole # report over one bad row. warnings.append( 'ignored duplicate queue {!r}'.format(queuename)) continue seennames.add(queuename.lower()) isdefault = _observed_bool(entry.get('isdefault')) if isdefault and defaultqueue is not None: # A host has exactly one default printer. Two means the client # misread it, and keeping both would leave the seed candidate # picking one at random. warnings.append( 'more than one queue reported as default; kept {!r}'.format( defaultqueue)) isdefault = False if isdefault: defaultqueue = queuename db.session.add(PrinterObservedQueue( hostname=hostname, queuename=queuename, drivername=_observed_text(entry.get('drivername'), 'drivername', warnings), portname=_observed_text(entry.get('portname'), 'portname', warnings), portaddress=_observed_text(entry.get('portaddress'), 'portaddress', warnings), isdefault=isdefault, isshared=_observed_bool(entry.get('isshared')), assetid=assetid, observedat=observedat, )) stored += 1 # flush, not commit: the collector dispatcher owns the transaction and # commits after writing its AuditLog row. Committing here would leave # an unaudited report behind if that write then failed. db.session.flush() # Always 'updated'. This endpoint replaces observed rows and creates no # asset, so 'created' never applies, and calling an identical re-report # 'noop' would hide that the bay is still checking in. return { 'action': 'updated', 'assetid': assetid, 'warnings': warnings, 'extra': {'queuecount': stored}, } def _observed_assetid(self, hostname, warnings): """Computer asset this hostname belongs to, or None with a warning. Reuses the resolver behind /api/printers/for-host rather than repeating it: if the two ever disagreed, a bay would be compared against the assignment of a different PC than the one it was told to install from. An unknown hostname is a WARNING, not an error. A bay reporting before its PC record exists is normal on a new build, the rows are keyed by hostname and resolve the moment that record appears, and a 500 here would make the client retry and log a failure on every cycle forever. """ try: from .api.asset_routes import _computer_by_hostname row = _computer_by_hostname(hostname) except ImportError: # A lean site can run without the computers plugin (ADR-013). The # observed rows are still worth keeping - they just stay unresolved. warnings.append('computers plugin not installed; observed queues ' 'stored against the hostname only') return None if row is None: warnings.append( 'hostname {!r} does not match a known PC; observed queues ' 'stored unresolved'.format(hostname)) return None # _computer_by_hostname returns the (Computer, Asset) pair. return row[1].assetid def get_asset_type_descriptor(self): """Return descriptor for printer types management in Asset Type Hub.""" return { 'assettype': 'printer', 'subtype_endpoint': '/printers/types', 'subtype_id_field': 'printertypeid', 'subtype_label_field': 'printertype', 'subtype_description_field': 'description', 'subtype_color_field': 'color', 'optional_tabs': [], 'edit_permission': 'printers.manage', }