d109314123
The toner report says what is empty now. It could not say what to order, and nothing recorded how fast anything drains - every level read was cached for five minutes and then discarded. Zabbix has been keeping the history all along; we simply never asked. One history.get gives both answers, because a cartridge only goes DOWN while it is in use: a rise is a replacement. Count the rises and you have how many cartridges a printer has been through; fit a slope to the readings SINCE the last rise and you have days-to-empty. Fitting across a replacement averages a spent cartridge with a fresh one and describes neither. Sorted by days left, which is the point. A cartridge at 60% dropping 5% a day needs ordering before one sitting at 8% that has not moved in months, and a level-sorted list ranks those backwards. It refuses to guess. Too few readings, a level that has not moved enough - many printers report in 10% steps and sit on a plateau for a fortnight - or a recent replacement each produce no estimate and say which. Those printers are listed separately rather than sorted in as 0 or as 999, since a printer without an estimate is neither urgent nor safe. Estimates show what they rest on, because "9 days from 21 days of readings" and "9 days from 2 readings" are not the same claim. A separate report card, not an extension of the toner report: that one is an exceptions list a tech acts on today, this is an ordering view read monthly, and the history query is heavier than the live read it would have slowed down. The analysis is pure arithmetic over a list of readings, so the 14 tests cover the noise wobble, the plateau, the swap, junk rows and division by zero without needing Zabbix. Zabbix being unreachable is reported as such rather than rendering an empty table that reads as "nothing is due".
1584 lines
57 KiB
Python
1584 lines
57 KiB
Python
"""Printers API routes - new Asset-based architecture."""
|
|
|
|
import logging
|
|
import re
|
|
|
|
from flask import Blueprint, request, Response
|
|
from flask_jwt_extended import jwt_required
|
|
|
|
from shopdb.api import db, cache, Setting, Asset, AssetType, Vendor, Model, Communication, CommunicationType, AssetRelationship, RelationshipType, success_response, error_response, paginated_response, ErrorCodes, get_pagination_params, paginate_query
|
|
|
|
from ..models import Printer, PrinterType, ModelSupply, PrinterDriver
|
|
from ..models.model_supply import SUPPLY_TYPES, SUPPLY_COLORS, CAPACITY_TIERS
|
|
from ..services import (
|
|
ZabbixService,
|
|
classifysupply,
|
|
derivesupplytype,
|
|
derivecolor,
|
|
lookupsupplies,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
from shopdb.api import require_permission, apply_import_timestamps
|
|
|
|
printers_asset_bp = Blueprint('printers_asset', __name__)
|
|
|
|
|
|
# =============================================================================
|
|
# Printer Types
|
|
# =============================================================================
|
|
|
|
@printers_asset_bp.route('/types', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def list_printer_types():
|
|
"""List all printer types."""
|
|
page, per_page = get_pagination_params(request)
|
|
|
|
query = PrinterType.query
|
|
|
|
if request.args.get('active', 'true').lower() != 'false':
|
|
query = query.filter(PrinterType.isactive == True)
|
|
|
|
if search := request.args.get('search'):
|
|
query = query.filter(PrinterType.printertype.ilike(f'%{search}%'))
|
|
|
|
query = query.order_by(PrinterType.printertype)
|
|
|
|
items, total = paginate_query(query, page, per_page)
|
|
data = [t.to_dict() for t in items]
|
|
|
|
return paginated_response(data, page, per_page, total)
|
|
|
|
|
|
@printers_asset_bp.route('/types/<int:type_id>', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def get_printer_type(type_id: int):
|
|
"""Get a single printer type."""
|
|
t = db.session.get(PrinterType, type_id)
|
|
|
|
if not t:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Printer type with ID {type_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
return success_response(t.to_dict())
|
|
|
|
|
|
@printers_asset_bp.route('/types', methods=['POST'])
|
|
@jwt_required()
|
|
@require_permission('printers.create')
|
|
def create_printer_type():
|
|
"""Create a new printer type."""
|
|
data = request.get_json()
|
|
|
|
if not data or not data.get('printertype'):
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'printertype is required')
|
|
|
|
existing = PrinterType.query.filter_by(printertype=data['printertype']).first()
|
|
if existing:
|
|
if not existing.isactive:
|
|
existing.isactive = True
|
|
for key in ('description', 'icon', 'color'):
|
|
if data.get(key) is not None:
|
|
setattr(existing, key, data[key])
|
|
db.session.commit()
|
|
return success_response(existing.to_dict(), message='Reactivated existing type')
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"Printer type '{data['printertype']}' already exists",
|
|
http_code=409
|
|
)
|
|
|
|
t = PrinterType(
|
|
printertype=data['printertype'],
|
|
description=data.get('description'),
|
|
icon=data.get('icon'), color=data.get('color')
|
|
)
|
|
|
|
db.session.add(t)
|
|
db.session.commit()
|
|
|
|
return success_response(t.to_dict(), message='Printer type created', http_code=201)
|
|
|
|
|
|
@printers_asset_bp.route('/types/<int:type_id>', methods=['PUT'])
|
|
@jwt_required()
|
|
@require_permission('printers.edit')
|
|
def update_printer_type(type_id: int):
|
|
"""Update a printer type."""
|
|
t = db.session.get(PrinterType, type_id)
|
|
if not t:
|
|
return error_response(ErrorCodes.NOT_FOUND,
|
|
f'Printer type with ID {type_id} not found', http_code=404)
|
|
|
|
data = request.get_json() or {}
|
|
if 'printertype' in data and data['printertype'] != t.printertype:
|
|
if PrinterType.query.filter_by(printertype=data['printertype']).first():
|
|
return error_response(ErrorCodes.CONFLICT,
|
|
f"Printer type '{data['printertype']}' already exists", http_code=409)
|
|
|
|
for key in ['printertype', 'description', 'icon', 'color', 'isactive']:
|
|
if key in data:
|
|
setattr(t, key, data[key])
|
|
|
|
db.session.commit()
|
|
return success_response(t.to_dict(), message='Printer type updated')
|
|
|
|
|
|
@printers_asset_bp.route('/types/<int:type_id>', methods=['DELETE'])
|
|
@jwt_required()
|
|
@require_permission('printers.delete')
|
|
def delete_printer_type(type_id: int):
|
|
"""Delete a printer type. Refused if any printer still uses it."""
|
|
t = db.session.get(PrinterType, type_id)
|
|
if not t:
|
|
return error_response(ErrorCodes.NOT_FOUND, 'Printer type not found', http_code=404)
|
|
inuse = Printer.query.filter_by(printertypeid=type_id).count()
|
|
if inuse:
|
|
return error_response(ErrorCodes.CONFLICT,
|
|
f"Cannot delete: {inuse} printer(s) still use this type", http_code=409)
|
|
db.session.delete(t)
|
|
db.session.commit()
|
|
return success_response(message='Printer type deleted')
|
|
|
|
|
|
# =============================================================================
|
|
# Printer Drivers (named SMB / HTTP links to driver packages)
|
|
# =============================================================================
|
|
|
|
@printers_asset_bp.route('/drivers', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def list_drivers():
|
|
"""List printer drivers. ?active=false includes inactive ones."""
|
|
query = PrinterDriver.query
|
|
if request.args.get('active', 'true').lower() != 'false':
|
|
query = query.filter_by(isactive=True)
|
|
drivers = query.order_by(PrinterDriver.name).all()
|
|
return success_response([d.to_dict() for d in drivers])
|
|
|
|
|
|
@printers_asset_bp.route('/drivers', methods=['POST'])
|
|
@jwt_required()
|
|
@require_permission('printers.create')
|
|
def create_driver():
|
|
data = request.get_json() or {}
|
|
if not (data.get('name') and data.get('location')):
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'name and location are required')
|
|
d = PrinterDriver(
|
|
name=data['name'],
|
|
location=data['location'],
|
|
description=data.get('description'),
|
|
modelnumberid=data.get('modelnumberid') or None,
|
|
isactive=data.get('isactive', True),
|
|
)
|
|
db.session.add(d)
|
|
db.session.commit()
|
|
return success_response(d.to_dict(), message='Driver created', http_code=201)
|
|
|
|
|
|
@printers_asset_bp.route('/drivers/<int:driver_id>', methods=['PUT'])
|
|
@jwt_required()
|
|
@require_permission('printers.edit')
|
|
def update_driver(driver_id):
|
|
d = db.session.get(PrinterDriver, driver_id)
|
|
if not d:
|
|
return error_response(ErrorCodes.NOT_FOUND, 'Driver not found', http_code=404)
|
|
data = request.get_json() or {}
|
|
for key in ('name', 'location', 'description', 'isactive'):
|
|
if key in data:
|
|
setattr(d, key, data[key])
|
|
if 'modelnumberid' in data:
|
|
d.modelnumberid = data['modelnumberid'] or None
|
|
db.session.commit()
|
|
return success_response(d.to_dict(), message='Driver updated')
|
|
|
|
|
|
@printers_asset_bp.route('/drivers/<int:driver_id>', methods=['DELETE'])
|
|
@jwt_required()
|
|
@require_permission('printers.delete')
|
|
def delete_driver(driver_id):
|
|
d = db.session.get(PrinterDriver, driver_id)
|
|
if not d:
|
|
return error_response(ErrorCodes.NOT_FOUND, 'Driver not found', http_code=404)
|
|
db.session.delete(d)
|
|
db.session.commit()
|
|
return success_response(message='Driver deleted')
|
|
|
|
|
|
# =============================================================================
|
|
# Printers CRUD
|
|
# =============================================================================
|
|
|
|
@printers_asset_bp.route('', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def list_printers():
|
|
"""
|
|
List all printers with filtering and pagination.
|
|
|
|
Query parameters:
|
|
- page, per_page: Pagination
|
|
- active: Filter by active status
|
|
- search: Search by asset number, name, or hostname
|
|
- type_id: Filter by printer type ID
|
|
- vendor_id: Filter by vendor ID
|
|
- location_id: Filter by location ID
|
|
- businessunit_id: Filter by business unit ID
|
|
"""
|
|
page, per_page = get_pagination_params(request)
|
|
|
|
# Join Printer with Asset
|
|
query = db.session.query(Printer).join(Asset)
|
|
|
|
# Active filter
|
|
if request.args.get('active', 'true').lower() != 'false':
|
|
query = query.filter(Asset.isactive == True)
|
|
|
|
# Exact-match natural-key lookup for idempotent import (asset number).
|
|
if exactassetnumber := request.args.get('assetnumber'):
|
|
query = query.filter(Asset.assetnumber == exactassetnumber)
|
|
|
|
# Search filter. Type and model are columns in the list, so searching
|
|
# 'Thermal' must find the thermal printers. Outer joins so a printer missing
|
|
# either still matches on its own fields.
|
|
if search := request.args.get('search'):
|
|
pattern = f'%{search}%'
|
|
query = query.outerjoin(
|
|
PrinterType, Printer.printertypeid == PrinterType.printertypeid
|
|
).outerjoin(
|
|
Model, Printer.modelnumberid == Model.modelnumberid
|
|
).filter(
|
|
db.or_(
|
|
Asset.assetnumber.ilike(pattern),
|
|
Asset.name.ilike(pattern),
|
|
Asset.serialnumber.ilike(pattern),
|
|
Printer.hostname.ilike(pattern),
|
|
Printer.windowsname.ilike(pattern),
|
|
PrinterType.printertype.ilike(pattern),
|
|
Model.modelnumber.ilike(pattern)
|
|
)
|
|
)
|
|
|
|
# Type filter
|
|
if typeid := request.args.get('typeid', request.args.get('type_id')):
|
|
query = query.filter(Printer.printertypeid == int(typeid))
|
|
|
|
# Vendor filter
|
|
if vendor_id := request.args.get('vendorid', request.args.get('vendor_id')):
|
|
query = query.filter(Printer.vendorid == int(vendor_id))
|
|
|
|
# Location filter
|
|
if location_id := request.args.get('locationid', request.args.get('location_id')):
|
|
query = query.filter(Asset.locationid == int(location_id))
|
|
|
|
# Business unit filter
|
|
if bu_id := request.args.get('businessunitid', request.args.get('businessunit_id')):
|
|
query = query.filter(Asset.businessunitid == int(bu_id))
|
|
|
|
# Sorting
|
|
sort_by = request.args.get('sort', 'hostname')
|
|
sort_dir = request.args.get('dir', 'asc')
|
|
|
|
if sort_by == 'hostname':
|
|
col = Printer.hostname
|
|
elif sort_by == 'assetnumber':
|
|
col = Asset.assetnumber
|
|
elif sort_by == 'name':
|
|
col = Asset.name
|
|
else:
|
|
col = Printer.hostname
|
|
|
|
query = query.order_by(col.desc() if sort_dir == 'desc' else col)
|
|
|
|
items, total = paginate_query(query, page, per_page)
|
|
|
|
# Build response with both asset and printer data
|
|
data = []
|
|
for printer in items:
|
|
item = printer.asset.to_dict() if printer.asset else {}
|
|
item['printer'] = printer.to_dict()
|
|
|
|
# Add primary IP address
|
|
if printer.asset:
|
|
primary_comm = Communication.query.filter_by(
|
|
assetid=printer.asset.assetid,
|
|
isprimary=True
|
|
).first()
|
|
if not primary_comm:
|
|
primary_comm = Communication.query.filter_by(
|
|
assetid=printer.asset.assetid
|
|
).first()
|
|
item['ipaddress'] = primary_comm.ipaddress if primary_comm else None
|
|
|
|
data.append(item)
|
|
|
|
return paginated_response(data, page, per_page, total)
|
|
|
|
|
|
@printers_asset_bp.route('/install-list', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def printer_install_list():
|
|
"""Flat, unpaginated list of network printers for the printer installer.
|
|
|
|
Shopfloor 2.0 PCs cannot run unsigned .bat maps, so the signed installer EXE
|
|
pulls printer data + floor-map positions from here and renders the picker.
|
|
Replaces the classic apiprinters.asp contract, adding mapx/mapy. Network
|
|
printers only (USB-only printers are excluded).
|
|
"""
|
|
rows = []
|
|
query = db.session.query(Printer).join(Asset).filter(Asset.isactive == True)
|
|
for printer in query.all():
|
|
asset = printer.asset
|
|
if not asset:
|
|
continue
|
|
primary = Communication.query.filter_by(
|
|
assetid=asset.assetid, isprimary=True).first() \
|
|
or Communication.query.filter_by(assetid=asset.assetid).first()
|
|
ipaddress = primary.ipaddress if primary else None
|
|
|
|
# Network printers only: must have a hostname or a non-USB IP.
|
|
is_network = bool(printer.hostname) or (ipaddress and ipaddress != 'USB')
|
|
if not is_network:
|
|
continue
|
|
|
|
data = printer.to_dict()
|
|
rows.append({
|
|
'printerid': printer.printerid,
|
|
'name': asset.name or asset.assetnumber,
|
|
'machinenumber': asset.assetnumber,
|
|
'windowsname': printer.windowsname,
|
|
'sharename': printer.sharename,
|
|
'hostname': printer.hostname,
|
|
'ipaddress': ipaddress,
|
|
'vendorname': _printer_vendor(printer),
|
|
'modelnumber': data.get('modelname'),
|
|
'installpath': printer.installpath,
|
|
'iscsf': printer.iscsf,
|
|
'locationname': asset.location.locationname if asset.location else None,
|
|
'mapx': asset.mapx,
|
|
'mapy': asset.mapy,
|
|
})
|
|
|
|
# A pipe-delimited text variant for the Inno installers: one printer per
|
|
# line, fixed field order, so the Pascal side is a split() instead of a
|
|
# hand-rolled JSON parser. The web map uses the default JSON.
|
|
if request.args.get('format') == 'text':
|
|
fields = ('printerid', 'windowsname', 'vendorname', 'modelnumber',
|
|
'hostname', 'ipaddress', 'mapx', 'mapy')
|
|
lines = [_text_line(row, fields) for row in rows]
|
|
return Response('\n'.join(lines), mimetype='text/plain')
|
|
|
|
return success_response(rows)
|
|
|
|
|
|
# Vendors whose printers install through the universal PrinterInstaller.exe
|
|
# (single call with a comma-separated /PRINTER list). Everything else installs
|
|
# from its own installpath .exe, or is flagged for manual install. Mirrors the
|
|
# classic installprinter.asp grouping rule.
|
|
UNIVERSAL_INSTALL_VENDORS = frozenset({'HP', 'Xerox'})
|
|
|
|
|
|
def _batch_base_url():
|
|
"""Base URL the generated .bat downloads installers from. Prefer the
|
|
configured site_base_url (it already includes scheme + the /shopdb mount).
|
|
|
|
Fallback matters: behind IIS the app sees http on a loopback port and its
|
|
url_root drops the mount, so a naive request.url_root yields a broken
|
|
http://127.0.0.1/installers/... URL. Rebuild from the forwarded Host + the
|
|
mount (script_root) and force https instead."""
|
|
base = (Setting.get('site_base_url') or '').strip().rstrip('/')
|
|
if base:
|
|
return base
|
|
host = request.headers.get('X-Forwarded-Host') or request.host
|
|
root = (request.script_root or '').rstrip('/')
|
|
return 'https://%s%s' % (host, root)
|
|
|
|
|
|
def _text_line(row, fields):
|
|
"""One pipe-delimited line for the installer text format. None -> empty;
|
|
any pipe/newline in a value is neutralized so the field count stays fixed."""
|
|
parts = []
|
|
for key in fields:
|
|
value = row.get(key)
|
|
text = '' if value is None else str(value)
|
|
parts.append(text.replace('|', ' ').replace('\r', ' ').replace('\n', ' '))
|
|
return '|'.join(parts)
|
|
|
|
|
|
def _printer_vendor(printer):
|
|
"""Vendor name for install grouping. The legacy import sets the printer's
|
|
model but not its direct vendorid, so resolve via the model's vendor (as the
|
|
classic installprinter.asp did) when the printer has no direct vendor."""
|
|
if printer.vendor:
|
|
return (printer.vendor.vendor or '').strip()
|
|
if printer.model and printer.model.vendor:
|
|
return (printer.model.vendor.vendor or '').strip()
|
|
return ''
|
|
|
|
|
|
def _installer_url(installpath, base):
|
|
"""Absolute URL for a specific installer. Full URLs and UNC paths pass
|
|
through; a stored relative path ('./installers/printers/X.exe') mounts under
|
|
the site base -> base + '/installers/printers/X.exe'."""
|
|
path = (installpath or '').strip()
|
|
if not path:
|
|
return None
|
|
if re.match(r'^[a-z][a-z0-9+.-]*:', path, re.I) or path.startswith('\\\\'):
|
|
return path
|
|
return base + '/' + re.sub(r'^(\.?/)+', '', path)
|
|
|
|
|
|
def _install_name(printer, asset):
|
|
"""Name to install the printer as: the standardized Windows Name, else the
|
|
share/CSF name, else the asset name/number."""
|
|
return ((printer.windowsname or '').strip()
|
|
or (printer.sharename or '').strip()
|
|
or (asset.name or asset.assetnumber or '').strip())
|
|
|
|
|
|
@printers_asset_bp.route('/install-batch', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def printer_install_batch():
|
|
"""Generate a Windows .bat that installs the selected printers.
|
|
|
|
?printerids=1,2,3 (the printers the user clicked on the installer map). The
|
|
batch groups them the same way the classic installprinter.asp did:
|
|
- HP / Xerox -> one universal PrinterInstaller.exe /PRINTER="a,b,c" call
|
|
- has .exe installpath -> download + run that installer /SILENT
|
|
- anything else (no installpath, or a .zip) -> listed as manual install
|
|
Downloads use PowerShell Invoke-WebRequest with the caller's Windows creds,
|
|
against the site base URL + the IIS-served /installers folder.
|
|
"""
|
|
raw = (request.args.get('printerids') or '').strip()
|
|
ids = []
|
|
for token in raw.split(','):
|
|
token = token.strip()
|
|
if token.isdigit():
|
|
ids.append(int(token))
|
|
if not ids:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR,
|
|
'printerids is required (comma-separated)')
|
|
|
|
base = _batch_base_url()
|
|
universal = [] # (name,) install via PrinterInstaller.exe
|
|
specific = [] # (name, url) install via own .exe /SILENT
|
|
manual = [] # (name, vendor) cannot auto-install
|
|
|
|
for printer in (db.session.query(Printer).join(Asset)
|
|
.filter(Printer.printerid.in_(ids))
|
|
.filter(Asset.isactive == True).all()):
|
|
asset = printer.asset
|
|
if not asset:
|
|
continue
|
|
name = _install_name(printer, asset)
|
|
if not name:
|
|
continue
|
|
vendor = _printer_vendor(printer)
|
|
installpath = (printer.installpath or '').strip()
|
|
if vendor in UNIVERSAL_INSTALL_VENDORS:
|
|
universal.append(name)
|
|
elif installpath.lower().endswith('.exe'):
|
|
specific.append((name, _installer_url(installpath, base)))
|
|
else:
|
|
# No installer, or a non-.exe payload (e.g. .zip) we will not run
|
|
# blindly with /SILENT - surface it for a human instead.
|
|
manual.append((name, vendor or 'unknown'))
|
|
|
|
facility = (Setting.get('facility_name') or 'GE Aerospace').strip() or 'GE Aerospace'
|
|
total = len(universal) + len(specific) + len(manual)
|
|
bat = _render_install_bat(facility, base, universal, specific, manual, total)
|
|
|
|
count = total if total else 0
|
|
filename = ('Install_%d_Printers.bat' % count) if count != 1 else 'Install_Printer.bat'
|
|
return Response(bat, mimetype='application/octet-stream',
|
|
headers={'Content-Disposition': 'attachment; filename=%s' % filename})
|
|
|
|
|
|
# PowerShell one-liner that downloads a URL to a temp file using the caller's
|
|
# Windows credentials (the installers share/site is integrated-auth on the LAN).
|
|
_PS_DOWNLOAD = ("powershell -NoProfile -Command \""
|
|
"$ProgressPreference='SilentlyContinue'; "
|
|
"[Net.ServicePointManager]::SecurityProtocol=[Net.SecurityProtocolType]::Tls12; "
|
|
"Invoke-WebRequest -Uri '%s' -OutFile '%s' "
|
|
"-UseBasicParsing -UseDefaultCredentials\"")
|
|
|
|
|
|
def _render_install_bat(facility, base, universal, specific, manual, total):
|
|
"""Build the .bat text (CRLF line endings for cmd.exe)."""
|
|
out = []
|
|
add = out.append
|
|
add('@echo off')
|
|
add('setlocal enabledelayedexpansion')
|
|
add('')
|
|
add('echo ========================================')
|
|
add('echo %s Printer Installer' % facility)
|
|
add('echo ========================================')
|
|
add('echo.')
|
|
|
|
if total == 0:
|
|
add('echo No installable printers were selected.')
|
|
add('pause')
|
|
add('exit /b 1')
|
|
return '\r\n'.join(out) + '\r\n'
|
|
|
|
add('echo Installing %d printer(s)...' % total)
|
|
add('echo.')
|
|
|
|
if manual:
|
|
add('echo *** The following require MANUAL installation (no silent installer): ***')
|
|
for name, vendor in manual:
|
|
add('echo - %s (%s)' % (name, vendor))
|
|
add('echo.')
|
|
|
|
for name, url in specific:
|
|
add('echo ----------------------------------------')
|
|
add('echo Installing: %s' % name)
|
|
add('echo Downloading installer...')
|
|
add(_PS_DOWNLOAD % (url, '%TEMP%\\printer_installer.exe'))
|
|
add('if exist "%TEMP%\\printer_installer.exe" (')
|
|
add(' echo Running installer...')
|
|
add(' "%TEMP%\\printer_installer.exe" /SILENT')
|
|
add(' del "%TEMP%\\printer_installer.exe" 2>nul')
|
|
add(') else (')
|
|
add(' echo ERROR: Could not download installer for %s' % name)
|
|
add(')')
|
|
add('echo.')
|
|
|
|
if universal:
|
|
add('echo ----------------------------------------')
|
|
add('echo Installing %d printer(s) via the universal installer:' % len(universal))
|
|
for name in universal:
|
|
add('echo - %s' % name)
|
|
add('echo ----------------------------------------')
|
|
add('echo Downloading PrinterInstaller.exe...')
|
|
add(_PS_DOWNLOAD % (base + '/installers/PrinterInstaller.exe',
|
|
'%TEMP%\\PrinterInstaller.exe'))
|
|
add('if exist "%TEMP%\\PrinterInstaller.exe" (')
|
|
add(' echo Running installer...')
|
|
add(' "%TEMP%\\PrinterInstaller.exe" /PRINTER="' + ','.join(universal) + '"')
|
|
add(' del "%TEMP%\\PrinterInstaller.exe" 2>nul')
|
|
add(') else (')
|
|
add(' echo ERROR: Could not download PrinterInstaller.exe')
|
|
add(')')
|
|
add('echo.')
|
|
|
|
add('echo ========================================')
|
|
add('echo Installation Complete!')
|
|
add('echo ========================================')
|
|
add('echo.')
|
|
add('')
|
|
add(':: Self-delete this batch file')
|
|
add('(goto) 2>nul & del "%~f0"')
|
|
return '\r\n'.join(out) + '\r\n'
|
|
|
|
|
|
@printers_asset_bp.route('/pc-default', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def pc_default_printer():
|
|
"""Default printer for a PC, by machine (asset) number.
|
|
|
|
Parity with classic apipcdefaultprinter.asp: the signed installer EXE
|
|
preselects a PC's default-printer hotspot on the site-map wizard using the
|
|
machine number persisted at PXE enrollment. The link is a `defaultprinter`
|
|
asset relationship (PC asset -> printer asset), so this stays inside the
|
|
contract surface (no cross-plugin model import).
|
|
|
|
Returns {printerid, windowsname}, or {} when the machine is unknown or has
|
|
no active default printer set.
|
|
"""
|
|
as_text = request.args.get('format') == 'text'
|
|
|
|
def _empty():
|
|
# Text variant returns an empty body (no default) so the installer's
|
|
# split yields nothing; JSON keeps the {} contract.
|
|
return Response('', mimetype='text/plain') if as_text else success_response({})
|
|
|
|
machine = (request.args.get('machine') or '').strip()
|
|
if not machine:
|
|
return _empty()
|
|
|
|
pc = Asset.query.filter_by(assetnumber=machine, isactive=True).first()
|
|
dp_type = RelationshipType.query.filter_by(relationshiptype='defaultprinter').first()
|
|
if not pc or not dp_type:
|
|
return _empty()
|
|
|
|
rel = AssetRelationship.query.filter_by(
|
|
sourceassetid=pc.assetid,
|
|
relationshiptypeid=dp_type.relationshiptypeid,
|
|
isactive=True,
|
|
).first()
|
|
if not rel:
|
|
return _empty()
|
|
|
|
printer = db.session.query(Printer).join(Asset).filter(
|
|
Printer.assetid == rel.targetassetid,
|
|
Asset.isactive == True,
|
|
).first()
|
|
if not printer:
|
|
return _empty()
|
|
|
|
if as_text:
|
|
return Response(
|
|
_text_line({'printerid': printer.printerid,
|
|
'windowsname': printer.windowsname},
|
|
('printerid', 'windowsname')),
|
|
mimetype='text/plain')
|
|
|
|
return success_response({
|
|
'printerid': printer.printerid,
|
|
'windowsname': printer.windowsname,
|
|
})
|
|
|
|
|
|
@printers_asset_bp.route('/<int:printer_id>', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def get_printer(printer_id: int):
|
|
"""Get a single printer with full details."""
|
|
printer = db.session.get(Printer, printer_id)
|
|
|
|
if not printer:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Printer with ID {printer_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
result = printer.asset.to_dict() if printer.asset else {}
|
|
result['printer'] = printer.to_dict()
|
|
|
|
# Add communications
|
|
if printer.asset:
|
|
comms = Communication.query.filter_by(assetid=printer.asset.assetid).all()
|
|
result['communications'] = [c.to_dict() for c in comms]
|
|
|
|
# Attach active drivers that match this printer's model
|
|
if printer.modelnumberid:
|
|
drivers = PrinterDriver.query.filter_by(
|
|
modelnumberid=printer.modelnumberid, isactive=True
|
|
).order_by(PrinterDriver.name).all()
|
|
result['drivers'] = [d.to_dict() for d in drivers]
|
|
else:
|
|
result['drivers'] = []
|
|
|
|
return success_response(result)
|
|
|
|
|
|
@printers_asset_bp.route('/by-asset/<int:asset_id>', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def get_printer_by_asset(asset_id: int):
|
|
"""Get printer data by asset ID."""
|
|
printer = Printer.query.filter_by(assetid=asset_id).first()
|
|
|
|
if not printer:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Printer for asset {asset_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
result = printer.asset.to_dict() if printer.asset else {}
|
|
result['printer'] = printer.to_dict()
|
|
|
|
return success_response(result)
|
|
|
|
|
|
@printers_asset_bp.route('', methods=['POST'])
|
|
@jwt_required()
|
|
@require_permission('printers.create')
|
|
def create_printer():
|
|
"""
|
|
Create new printer (creates both Asset and Printer records).
|
|
|
|
Required fields:
|
|
- assetnumber: Business identifier
|
|
|
|
Optional fields:
|
|
- name, serialnumber, statusid, locationid, businessunitid
|
|
- printertypeid, vendorid, modelnumberid, hostname
|
|
- windowsname, sharename, iscsf, installpath, pin
|
|
- iscolor, isduplex, isnetwork
|
|
- mapx, mapy, notes
|
|
"""
|
|
data = request.get_json()
|
|
|
|
if not data:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
|
|
|
|
if not data.get('assetnumber'):
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'assetnumber is required')
|
|
|
|
# Check for duplicate assetnumber
|
|
if Asset.query.filter_by(assetnumber=data['assetnumber']).first():
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"Asset with number '{data['assetnumber']}' already exists",
|
|
http_code=409
|
|
)
|
|
|
|
# Get printer asset type
|
|
printer_type = AssetType.query.filter_by(assettype='printer').first()
|
|
if not printer_type:
|
|
return error_response(
|
|
ErrorCodes.INTERNAL_ERROR,
|
|
'Printer asset type not found. Plugin may not be properly installed.',
|
|
http_code=500
|
|
)
|
|
|
|
# Create the core asset
|
|
asset = Asset(
|
|
assetnumber=data['assetnumber'],
|
|
name=data.get('name'),
|
|
serialnumber=data.get('serialnumber'),
|
|
gaugelabreference=data.get('gaugelabreference'),
|
|
maintenancereference=data.get('maintenancereference'),
|
|
assettypeid=printer_type.assettypeid,
|
|
statusid=data.get('statusid', 1),
|
|
locationid=data.get('locationid'),
|
|
businessunitid=data.get('businessunitid'),
|
|
mapx=data.get('mapx'),
|
|
mapy=data.get('mapy'),
|
|
notes=data.get('notes')
|
|
)
|
|
|
|
db.session.add(asset)
|
|
db.session.flush() # Get the assetid
|
|
|
|
# Create the printer extension
|
|
printer = Printer(
|
|
assetid=asset.assetid,
|
|
printertypeid=data.get('printertypeid'),
|
|
vendorid=data.get('vendorid'),
|
|
modelnumberid=data.get('modelnumberid'),
|
|
hostname=data.get('hostname'),
|
|
windowsname=data.get('windowsname'),
|
|
sharename=data.get('sharename'),
|
|
iscsf=data.get('iscsf', False),
|
|
installpath=data.get('installpath'),
|
|
pin=data.get('pin'),
|
|
iscolor=data.get('iscolor', False),
|
|
isduplex=data.get('isduplex', False),
|
|
isnetwork=data.get('isnetwork', True)
|
|
)
|
|
|
|
db.session.add(printer)
|
|
|
|
# Create communication record if IP provided
|
|
if data.get('ipaddress'):
|
|
ip_comtype = CommunicationType.query.filter_by(comtype='IP').first()
|
|
if ip_comtype:
|
|
comm = Communication(
|
|
assetid=asset.assetid,
|
|
comtypeid=ip_comtype.comtypeid,
|
|
ipaddress=data['ipaddress'],
|
|
isprimary=True
|
|
)
|
|
db.session.add(comm)
|
|
|
|
# Preserve legacy timestamps in import mode (no-op otherwise)
|
|
apply_import_timestamps(asset, data)
|
|
|
|
db.session.commit()
|
|
|
|
result = asset.to_dict()
|
|
result['printer'] = printer.to_dict()
|
|
|
|
return success_response(result, message='Printer created', http_code=201)
|
|
|
|
|
|
@printers_asset_bp.route('/<int:printer_id>', methods=['PUT'])
|
|
@jwt_required()
|
|
@require_permission('printers.edit')
|
|
def update_printer(printer_id: int):
|
|
"""Update printer (both Asset and Printer records)."""
|
|
printer = db.session.get(Printer, printer_id)
|
|
|
|
if not printer:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Printer with ID {printer_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
data = request.get_json()
|
|
if not data:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
|
|
|
|
asset = printer.asset
|
|
|
|
# Check for conflicting assetnumber
|
|
if 'assetnumber' in data and data['assetnumber'] != asset.assetnumber:
|
|
if Asset.query.filter_by(assetnumber=data['assetnumber']).first():
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"Asset with number '{data['assetnumber']}' already exists",
|
|
http_code=409
|
|
)
|
|
|
|
# Update asset fields (optional identifiers gated per-type in Settings)
|
|
asset_fields = ['assetnumber', 'name', 'serialnumber', 'gaugelabreference',
|
|
'maintenancereference', 'statusid',
|
|
'locationid', 'businessunitid', 'mapx', 'mapy',
|
|
'notes', 'isactive']
|
|
for key in asset_fields:
|
|
if key in data:
|
|
setattr(asset, key, data[key])
|
|
|
|
# Update printer fields
|
|
printer_fields = ['printertypeid', 'vendorid', 'modelnumberid', 'hostname',
|
|
'windowsname', 'sharename', 'iscsf', 'installpath', 'pin',
|
|
'iscolor', 'isduplex', 'isnetwork']
|
|
for key in printer_fields:
|
|
if key in data:
|
|
setattr(printer, key, data[key])
|
|
|
|
# Upsert the primary IP communication when an ipaddress is supplied, so a
|
|
# single PUT updates core, extension, and network in one call.
|
|
if 'ipaddress' in data:
|
|
ip = (data.get('ipaddress') or '').strip()
|
|
comm = Communication.query.filter_by(
|
|
assetid=asset.assetid, isprimary=True).first()
|
|
if ip:
|
|
if comm:
|
|
comm.ipaddress = ip
|
|
else:
|
|
ip_comtype = CommunicationType.query.filter_by(comtype='IP').first()
|
|
if ip_comtype:
|
|
db.session.add(Communication(
|
|
assetid=asset.assetid,
|
|
comtypeid=ip_comtype.comtypeid,
|
|
ipaddress=ip,
|
|
isprimary=True,
|
|
))
|
|
elif comm:
|
|
comm.ipaddress = None
|
|
|
|
apply_import_timestamps(asset, data)
|
|
db.session.commit()
|
|
|
|
result = asset.to_dict()
|
|
result['printer'] = printer.to_dict()
|
|
|
|
return success_response(result, message='Printer updated')
|
|
|
|
|
|
@printers_asset_bp.route('/<int:printer_id>', methods=['DELETE'])
|
|
@jwt_required()
|
|
@require_permission('printers.delete')
|
|
def delete_printer(printer_id: int):
|
|
"""Delete (soft delete) printer."""
|
|
printer = db.session.get(Printer, printer_id)
|
|
|
|
if not printer:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Printer with ID {printer_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
# Soft delete the asset
|
|
printer.asset.isactive = False
|
|
db.session.commit()
|
|
|
|
return success_response(message='Printer deleted')
|
|
|
|
|
|
# =============================================================================
|
|
# Supply Levels (Zabbix Integration)
|
|
# =============================================================================
|
|
|
|
@printers_asset_bp.route('/<int:printer_id>/supplies', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def get_printer_supplies(printer_id: int):
|
|
"""Get supply levels from Zabbix (real-time lookup)."""
|
|
printer = db.session.get(Printer, printer_id)
|
|
|
|
if not printer:
|
|
return error_response(ErrorCodes.NOT_FOUND, 'Printer not found', http_code=404)
|
|
|
|
# Get IP address from communications
|
|
comm = Communication.query.filter_by(
|
|
assetid=printer.assetid,
|
|
isprimary=True
|
|
).first()
|
|
if not comm:
|
|
comm = Communication.query.filter_by(assetid=printer.assetid).first()
|
|
|
|
if not comm or not comm.ipaddress:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'Printer has no IP address')
|
|
|
|
service = ZabbixService()
|
|
if not service.isconfigured or not service.isreachable:
|
|
# fail soft when zabbix off or down
|
|
return success_response({
|
|
'ipaddress': comm.ipaddress,
|
|
'pingstatus': '-1',
|
|
'supplies': []
|
|
})
|
|
|
|
# vendor drives waste-cartridge rules; modelnumberid drives part lookup
|
|
vendor_name = printer.vendor.vendor if getattr(printer, 'vendor', None) else None
|
|
|
|
raw_supplies = service.getsuppliesbyip(comm.ipaddress) or []
|
|
supplies = [
|
|
_annotate_supply(s, vendor_name, printer.modelnumberid) for s in raw_supplies
|
|
]
|
|
|
|
return success_response({
|
|
'ipaddress': comm.ipaddress,
|
|
'pingstatus': service.getpingstatus(comm.ipaddress),
|
|
'supplies': supplies
|
|
})
|
|
|
|
|
|
# =============================================================================
|
|
# Low Supplies
|
|
# =============================================================================
|
|
|
|
def _annotate_supply(supply, vendor_name, modelnumberid):
|
|
"""Add status, remaining percent, and part numbers to a raw supply dict.
|
|
|
|
Waste cartridge direction depends on vendor, so classification lives in
|
|
the supply_parts helper. Part numbers come from the modelsupplies table.
|
|
"""
|
|
level = supply.get('level', 0)
|
|
name = supply.get('name', 'Unknown')
|
|
supplytype = derivesupplytype(name)
|
|
color = derivecolor(name, supply.get('color'))
|
|
cls = classifysupply(level, name, vendor_name)
|
|
return {
|
|
'name': name,
|
|
'level': level,
|
|
'color': color,
|
|
'supplytype': supplytype,
|
|
'status': cls['status'],
|
|
'remaining': cls['remaining'],
|
|
'iswaste': cls['iswaste'],
|
|
'isdrum': cls['isdrum'],
|
|
'partnumbers': lookupsupplies(modelnumberid, color, supplytype),
|
|
}
|
|
|
|
|
|
def _get_low_supplies_data():
|
|
"""Build low supplies data (cached for 5 minutes)."""
|
|
cached = cache.get('printers_low_supplies')
|
|
if cached is not None:
|
|
return cached
|
|
|
|
service = ZabbixService()
|
|
if not service.isconfigured or not service.isreachable:
|
|
return {'printers': [], 'summary': {'total_checked': 0, 'low': 0, 'critical': 0}}
|
|
|
|
# active printers with an IP, with vendor and model for waste/part rules
|
|
rows = (
|
|
db.session.query(Printer, Asset, Communication, Vendor, Model)
|
|
.join(Asset, Asset.assetid == Printer.assetid)
|
|
.join(Communication, Communication.assetid == Asset.assetid)
|
|
.outerjoin(Vendor, Vendor.vendorid == Printer.vendorid)
|
|
.outerjoin(Model, Model.modelnumberid == Printer.modelnumberid)
|
|
.filter(Asset.isactive == True)
|
|
.filter(Communication.ipaddress.isnot(None))
|
|
.filter(Communication.ipaddress != '')
|
|
.all()
|
|
)
|
|
|
|
# dedupe by printer id (a printer may have several comms)
|
|
seen = set()
|
|
unique_printers = []
|
|
for printer, asset, comm, vendor, model in rows:
|
|
if printer.printerid not in seen:
|
|
seen.add(printer.printerid)
|
|
unique_printers.append((printer, asset, comm, vendor, model))
|
|
|
|
results = []
|
|
total_checked = 0
|
|
|
|
for printer, asset, comm, vendor, model in unique_printers:
|
|
supplies = service.getsuppliesbyip_cached(comm.ipaddress)
|
|
if supplies is None:
|
|
continue
|
|
|
|
total_checked += 1
|
|
|
|
vendor_name = vendor.vendor if vendor else None
|
|
model_number = model.modelnumber if model else None
|
|
modelnumberid = model.modelnumberid if model else None
|
|
|
|
# ONLY the supplies that need attention. A printer reporting one empty
|
|
# black cartridge alongside three full colour ones was listing all four,
|
|
# so the reader had to find the problem inside the row rather than being
|
|
# shown it. The whole report exists to answer "what needs replacing".
|
|
annotated = []
|
|
has_low = False
|
|
for s in supplies:
|
|
item = _annotate_supply(s, vendor_name, modelnumberid)
|
|
if item['status'] == 'ok':
|
|
continue
|
|
has_low = True
|
|
annotated.append(item)
|
|
|
|
if has_low:
|
|
# location name for the report row
|
|
# Via the relationship, the way the printers list does it. The
|
|
# previous lookup went through db.session.get on locationid, and
|
|
# every row came back with no location even where one is set.
|
|
location_name = (asset.location.locationname
|
|
if asset.location else None)
|
|
|
|
results.append({
|
|
'printerid': printer.printerid,
|
|
'printername': asset.name or printer.hostname or '',
|
|
'assetnumber': asset.assetnumber or '',
|
|
'ipaddress': comm.ipaddress,
|
|
'vendor': vendor_name,
|
|
'model': model_number,
|
|
'location': location_name,
|
|
'mapx': asset.mapx,
|
|
'mapy': asset.mapy,
|
|
'supplies': annotated
|
|
})
|
|
|
|
low_count = 0
|
|
critical_count = 0
|
|
for p in results:
|
|
has_critical = any(s['status'] == 'critical' for s in p['supplies'])
|
|
has_low = any(s['status'] == 'low' for s in p['supplies'])
|
|
if has_critical:
|
|
critical_count += 1
|
|
elif has_low:
|
|
low_count += 1
|
|
|
|
data = {
|
|
'printers': results,
|
|
'summary': {
|
|
'total_checked': total_checked,
|
|
'low': low_count,
|
|
'critical': critical_count
|
|
}
|
|
}
|
|
|
|
cache.set('printers_low_supplies', data, timeout=300)
|
|
return data
|
|
|
|
|
|
@printers_asset_bp.route('/lowsupplies', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def low_supplies():
|
|
"""Get printers with low or critical supply levels."""
|
|
data = _get_low_supplies_data()
|
|
return success_response(data)
|
|
|
|
|
|
@printers_asset_bp.route('/lookup', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def printer_lookup():
|
|
"""Find a printer by IP or FQDN. Parity with the classic printerlookup.asp.
|
|
|
|
Zabbix uses this to jump straight to a printer record. Query with
|
|
?ip=x.x.x.x or ?fqdn=hostname; returns the matching printer id.
|
|
"""
|
|
ip = (request.args.get('ip') or '').strip()
|
|
fqdn = (request.args.get('fqdn') or '').strip()
|
|
lookup_value = ip or fqdn
|
|
|
|
if not lookup_value:
|
|
return error_response(
|
|
ErrorCodes.VALIDATION_ERROR,
|
|
'Provide ip or fqdn'
|
|
)
|
|
|
|
# match the IP against any active printer communication
|
|
row = (
|
|
db.session.query(Printer, Asset)
|
|
.join(Asset, Asset.assetid == Printer.assetid)
|
|
.join(Communication, Communication.assetid == Asset.assetid)
|
|
.filter(Asset.isactive == True)
|
|
.filter(Communication.ipaddress == lookup_value)
|
|
.first()
|
|
)
|
|
|
|
if not row:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Printer not found: {lookup_value}',
|
|
http_code=404
|
|
)
|
|
|
|
printer, asset = row
|
|
return success_response({
|
|
'printerid': printer.printerid,
|
|
'assetid': asset.assetid,
|
|
'assetnumber': asset.assetnumber,
|
|
'name': asset.name or printer.hostname,
|
|
})
|
|
|
|
|
|
@printers_asset_bp.route('/supplies/refresh', methods=['POST'])
|
|
@jwt_required()
|
|
@require_permission('printers.create')
|
|
def refresh_supplies_cache():
|
|
"""Clear cached Zabbix supply data so the next read pulls fresh values.
|
|
|
|
Backs the toner report Refresh button (parity with adminclearcache.asp
|
|
type=zabbix).
|
|
"""
|
|
ZabbixService().clearcache()
|
|
return success_response(message='Supply cache cleared')
|
|
|
|
|
|
# =============================================================================
|
|
# Dashboard
|
|
# =============================================================================
|
|
|
|
@printers_asset_bp.route('/dashboard/summary', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def dashboard_summary():
|
|
"""Get printer dashboard summary data."""
|
|
# Total active printers
|
|
total = db.session.query(Printer).join(Asset).filter(
|
|
Asset.isactive == True
|
|
).count()
|
|
|
|
# Count by printer type
|
|
by_type = db.session.query(
|
|
PrinterType.printertype,
|
|
db.func.count(Printer.printerid)
|
|
).join(Printer, Printer.printertypeid == PrinterType.printertypeid
|
|
).join(Asset, Asset.assetid == Printer.assetid
|
|
).filter(Asset.isactive == True
|
|
).group_by(PrinterType.printertype
|
|
).all()
|
|
|
|
# Count by vendor
|
|
by_vendor = db.session.query(
|
|
Vendor.vendor,
|
|
db.func.count(Printer.printerid)
|
|
).join(Printer, Printer.vendorid == Vendor.vendorid
|
|
).join(Asset, Asset.assetid == Printer.assetid
|
|
).filter(Asset.isactive == True
|
|
).group_by(Vendor.vendor
|
|
).all()
|
|
|
|
# Get real low/critical supply counts (skip if Zabbix not reachable)
|
|
low_count = 0
|
|
critical_count = 0
|
|
service = ZabbixService()
|
|
if service.isconfigured and service.isreachable:
|
|
try:
|
|
supply_data = _get_low_supplies_data()
|
|
low_count = supply_data['summary']['low']
|
|
critical_count = supply_data['summary']['critical']
|
|
except Exception as e:
|
|
logger.warning(f"Could not fetch supply data for dashboard: {e}")
|
|
|
|
return success_response({
|
|
'total': total,
|
|
'totalprinters': total,
|
|
'online': total,
|
|
'lowsupplies': low_count,
|
|
'criticalsupplies': critical_count,
|
|
'bytype': [{'type': t, 'count': c} for t, c in by_type],
|
|
'byvendor': [{'vendor': v, 'count': c} for v, c in by_vendor],
|
|
})
|
|
|
|
|
|
# =============================================================================
|
|
# Model Supplies (data-driven toner/drum/waste part numbers)
|
|
# =============================================================================
|
|
|
|
def _validate_supply_payload(data):
|
|
"""Return an error message if the supply payload is invalid, else None."""
|
|
if not data:
|
|
return 'No data provided'
|
|
if not data.get('partnumber'):
|
|
return 'partnumber is required'
|
|
supplytype = data.get('supplytype', 'toner')
|
|
if supplytype not in SUPPLY_TYPES:
|
|
return f"supplytype must be one of {', '.join(SUPPLY_TYPES)}"
|
|
color = data.get('color', 'none')
|
|
if color not in SUPPLY_COLORS:
|
|
return f"color must be one of {', '.join(SUPPLY_COLORS)}"
|
|
capacitytier = data.get('capacitytier', 'standard')
|
|
if capacitytier not in CAPACITY_TIERS:
|
|
return f"capacitytier must be one of {', '.join(CAPACITY_TIERS)}"
|
|
return None
|
|
|
|
|
|
@printers_asset_bp.route('/supplies/meta', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def supplies_meta():
|
|
"""Allowed values for supply type, color, and capacity tier (for the UI)."""
|
|
return success_response({
|
|
'supplytypes': list(SUPPLY_TYPES),
|
|
'colors': list(SUPPLY_COLORS),
|
|
'capacitytiers': list(CAPACITY_TIERS),
|
|
})
|
|
|
|
|
|
@printers_asset_bp.route('/models', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def list_supply_models():
|
|
"""List models with a supply count, for the supply-management picker.
|
|
|
|
Query parameters:
|
|
- search: filter by model number
|
|
- vendor_id: filter by vendor
|
|
- withsupplies: 'true' to only return models that already have supplies
|
|
"""
|
|
page, per_page = get_pagination_params(request)
|
|
|
|
supplycount = db.func.count(ModelSupply.modelsupplyid).label('supplycount')
|
|
query = (
|
|
db.session.query(Model, Vendor.vendor, supplycount)
|
|
.outerjoin(Vendor, Vendor.vendorid == Model.vendorid)
|
|
.outerjoin(ModelSupply, db.and_(
|
|
ModelSupply.modelnumberid == Model.modelnumberid,
|
|
ModelSupply.isactive == True,
|
|
))
|
|
.group_by(Model.modelnumberid, Vendor.vendor)
|
|
)
|
|
|
|
# Toner/drum/waste only apply to printers, so restrict the picker to
|
|
# printer models: those attached to a printer asset, or those that already
|
|
# carry supply mappings. Keeps machine/controller models out of the list.
|
|
printer_model_ids = (
|
|
db.session.query(Printer.modelnumberid)
|
|
.filter(Printer.modelnumberid.isnot(None))
|
|
)
|
|
supply_model_ids = db.session.query(ModelSupply.modelnumberid)
|
|
query = query.filter(db.or_(
|
|
Model.modelnumberid.in_(printer_model_ids),
|
|
Model.modelnumberid.in_(supply_model_ids),
|
|
))
|
|
|
|
if search := request.args.get('search'):
|
|
query = query.filter(Model.modelnumber.ilike(f'%{search}%'))
|
|
if vendor_id := request.args.get('vendorid', request.args.get('vendor_id')):
|
|
query = query.filter(Model.vendorid == int(vendor_id))
|
|
if request.args.get('withsupplies', '').lower() == 'true':
|
|
query = query.having(supplycount > 0)
|
|
|
|
query = query.order_by(Model.modelnumber)
|
|
|
|
total = query.count()
|
|
rows = query.limit(per_page).offset((page - 1) * per_page).all()
|
|
|
|
data = [{
|
|
'modelnumberid': model.modelnumberid,
|
|
'modelnumber': model.modelnumber,
|
|
'vendor': vendor,
|
|
'vendorid': model.vendorid,
|
|
'supplycount': count,
|
|
} for model, vendor, count in rows]
|
|
|
|
return paginated_response(data, page, per_page, total)
|
|
|
|
|
|
@printers_asset_bp.route('/models/<int:modelnumberid>/supplies', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def list_model_supplies(modelnumberid: int):
|
|
"""List all supplies mapped to a model."""
|
|
model = db.session.get(Model, modelnumberid)
|
|
if not model:
|
|
return error_response(ErrorCodes.NOT_FOUND, 'Model not found', http_code=404)
|
|
|
|
supplies = (
|
|
ModelSupply.query
|
|
.filter_by(modelnumberid=modelnumberid, isactive=True)
|
|
.order_by(ModelSupply.supplytype, ModelSupply.color, ModelSupply.capacitytier)
|
|
.all()
|
|
)
|
|
return success_response({
|
|
'modelnumberid': modelnumberid,
|
|
'modelnumber': model.modelnumber,
|
|
'supplies': [s.to_dict() for s in supplies],
|
|
})
|
|
|
|
|
|
@printers_asset_bp.route('/models/<int:modelnumberid>/supplies', methods=['POST'])
|
|
@jwt_required()
|
|
@require_permission('printers.create')
|
|
def create_model_supply(modelnumberid: int):
|
|
"""Add a supply to a model."""
|
|
model = db.session.get(Model, modelnumberid)
|
|
if not model:
|
|
return error_response(ErrorCodes.NOT_FOUND, 'Model not found', http_code=404)
|
|
|
|
data = request.get_json()
|
|
message = _validate_supply_payload(data)
|
|
if message:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, message)
|
|
|
|
existing = ModelSupply.query.filter_by(
|
|
modelnumberid=modelnumberid,
|
|
partnumber=data['partnumber'],
|
|
).first()
|
|
if existing:
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"Part number '{data['partnumber']}' already mapped to this model",
|
|
http_code=409,
|
|
)
|
|
|
|
supply = ModelSupply(
|
|
modelnumberid=modelnumberid,
|
|
supplytype=data.get('supplytype', 'toner'),
|
|
color=data.get('color', 'none'),
|
|
capacitytier=data.get('capacitytier', 'standard'),
|
|
partnumber=data['partnumber'],
|
|
marketingname=data.get('marketingname'),
|
|
pageyield=data.get('pageyield'),
|
|
notes=data.get('notes'),
|
|
)
|
|
db.session.add(supply)
|
|
db.session.commit()
|
|
|
|
return success_response(supply.to_dict(), message='Supply added', http_code=201)
|
|
|
|
|
|
@printers_asset_bp.route('/supplies/<int:modelsupplyid>', methods=['PUT'])
|
|
@jwt_required()
|
|
@require_permission('printers.edit')
|
|
def update_model_supply(modelsupplyid: int):
|
|
"""Update a model supply."""
|
|
supply = db.session.get(ModelSupply, modelsupplyid)
|
|
if not supply:
|
|
return error_response(ErrorCodes.NOT_FOUND, 'Supply not found', http_code=404)
|
|
|
|
data = request.get_json()
|
|
if not data:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
|
|
|
|
# validate only the fields present
|
|
merged = {
|
|
'partnumber': data.get('partnumber', supply.partnumber),
|
|
'supplytype': data.get('supplytype', supply.supplytype),
|
|
'color': data.get('color', supply.color),
|
|
'capacitytier': data.get('capacitytier', supply.capacitytier),
|
|
}
|
|
message = _validate_supply_payload(merged)
|
|
if message:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, message)
|
|
|
|
if 'partnumber' in data and data['partnumber'] != supply.partnumber:
|
|
clash = ModelSupply.query.filter_by(
|
|
modelnumberid=supply.modelnumberid,
|
|
partnumber=data['partnumber'],
|
|
).first()
|
|
if clash:
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"Part number '{data['partnumber']}' already mapped to this model",
|
|
http_code=409,
|
|
)
|
|
|
|
for field in ('supplytype', 'color', 'capacitytier', 'partnumber',
|
|
'marketingname', 'pageyield', 'notes'):
|
|
if field in data:
|
|
setattr(supply, field, data[field])
|
|
|
|
db.session.commit()
|
|
return success_response(supply.to_dict(), message='Supply updated')
|
|
|
|
|
|
@printers_asset_bp.route('/supplies/<int:modelsupplyid>', methods=['DELETE'])
|
|
@jwt_required()
|
|
@require_permission('printers.delete')
|
|
def delete_model_supply(modelsupplyid: int):
|
|
"""Delete a model supply."""
|
|
supply = db.session.get(ModelSupply, modelsupplyid)
|
|
if not supply:
|
|
return error_response(ErrorCodes.NOT_FOUND, 'Supply not found', http_code=404)
|
|
|
|
db.session.delete(supply)
|
|
db.session.commit()
|
|
return success_response(message='Supply deleted')
|
|
|
|
|
|
def _shortsupplyname(name):
|
|
"""'Black Toner Level' -> 'Black'. The card has one line per printer, and
|
|
the words Toner and Level carry no information when every row is a toner
|
|
level."""
|
|
text = (name or 'supply').strip()
|
|
for noise in (' Cartridge Level', ' Toner Level', ' Level', ' Cartridge'):
|
|
if text.endswith(noise):
|
|
text = text[:-len(noise)]
|
|
break
|
|
return text or 'supply'
|
|
|
|
|
|
def _reordertip(supply):
|
|
"""What to order, for the chip's tooltip.
|
|
|
|
A percentage says a cartridge is nearly out; the part number says what to
|
|
buy, which is the next thing someone needs and today means opening the
|
|
printer's page to find it. Every capacity tier is listed, because the
|
|
report has always shown all reorder options.
|
|
"""
|
|
parts = supply.get('partnumbers') or []
|
|
if not parts:
|
|
return 'No part number on file for this model'
|
|
return ', '.join(
|
|
'{}{}'.format(part['partnumber'],
|
|
' ({})'.format(part['capacitytier'])
|
|
if part.get('capacitytier') else '')
|
|
for part in parts)
|
|
|
|
|
|
@printers_asset_bp.route('/dashboard/supplies', methods=['GET'])
|
|
@jwt_required()
|
|
@require_permission('printers.view')
|
|
def dashboard_supplies():
|
|
"""Printers needing a cartridge, flattened to one row per printer.
|
|
|
|
Reuses the existing low-supplies query and its five-minute cache, so the
|
|
card costs nothing extra: a Zabbix round-trip per printer on every dashboard
|
|
load would make this the slowest page in the app.
|
|
|
|
Critical first, then low. A printer with several depleted cartridges appears
|
|
once, listing them - a row per cartridge would report one printer three
|
|
times and read as three problems.
|
|
"""
|
|
data = _get_low_supplies_data()
|
|
|
|
threshold = 5
|
|
setting = Setting.query.filter_by(key='printers_dashboardpercent').first()
|
|
if setting and (setting.value or '').strip():
|
|
try:
|
|
threshold = int(setting.value)
|
|
except (TypeError, ValueError):
|
|
pass
|
|
|
|
rows = []
|
|
for printer in data.get('printers', []):
|
|
# The CARD is tighter than the report. The report lists anything the
|
|
# thresholds call low, which is the right scope for planning an order;
|
|
# the dashboard is asking what to walk out and change today, and a
|
|
# cartridge at 18% is not that. Anything at or below the threshold.
|
|
depleted = [supply for supply in printer['supplies']
|
|
if isinstance(supply.get('remaining'), (int, float))
|
|
and supply['remaining'] <= threshold]
|
|
if not depleted:
|
|
continue
|
|
depleted.sort(key=lambda supply: supply['remaining'])
|
|
rows.append({
|
|
'printerid': printer['printerid'],
|
|
'printername': printer['printername'] or printer['assetnumber'],
|
|
'location': printer['location'] or 'No location set',
|
|
# Coordinates for the hover preview. Either may be None - a
|
|
# printer never placed on the floor plan still belongs on the
|
|
# card, it just has nothing to preview.
|
|
'mapx': printer.get('mapx'),
|
|
'mapy': printer.get('mapy'),
|
|
'iscritical': any(s['status'] == 'critical' for s in depleted),
|
|
'supplies': [{
|
|
'text': '{} {}%'.format(_shortsupplyname(supply.get('name')),
|
|
supply.get('remaining')),
|
|
'title': _reordertip(supply),
|
|
'level': supply.get('status'),
|
|
} for supply in depleted],
|
|
})
|
|
|
|
rows.sort(key=lambda r: (not r['iscritical'], r['printername']))
|
|
return success_response(rows)
|
|
|
|
|
|
# =============================================================================
|
|
# Supply forecast
|
|
#
|
|
# The toner report answers "what is empty now". This answers "what will be, and
|
|
# what have we been getting through" - a purchasing question, on a different
|
|
# cadence, off data Zabbix has been keeping all along.
|
|
# =============================================================================
|
|
|
|
@printers_asset_bp.route('/supplies/forecast', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def supplies_forecast():
|
|
"""Days-to-empty and replacement counts per printer.
|
|
|
|
?days=90 how far back to read (Zabbix retention is the real ceiling)
|
|
|
|
Printers sort by their soonest supply. Anything without an honest estimate
|
|
is returned separately with the reason, rather than sorted as though it
|
|
were fine or dropped as though it did not exist.
|
|
"""
|
|
from ..services.supply_history import analyse, soonest
|
|
|
|
try:
|
|
days = max(1, min(365, int(request.args.get('days', 90))))
|
|
except (TypeError, ValueError):
|
|
days = 90
|
|
|
|
service = ZabbixService()
|
|
if not service.isconfigured or not service.isreachable:
|
|
return success_response({
|
|
'printers': [], 'unestimated': [], 'days': days,
|
|
'available': False,
|
|
'reason': 'Zabbix is not configured or not reachable',
|
|
})
|
|
|
|
rows = (
|
|
db.session.query(Printer, Asset, Communication, Vendor)
|
|
.join(Asset, Asset.assetid == Printer.assetid)
|
|
.join(Communication, Communication.assetid == Asset.assetid)
|
|
.outerjoin(Vendor, Vendor.vendorid == Printer.vendorid)
|
|
.filter(Asset.isactive == True,
|
|
Communication.ipaddress.isnot(None),
|
|
Communication.ipaddress != '')
|
|
.all()
|
|
)
|
|
|
|
seen = set()
|
|
estimated, unestimated = [], []
|
|
|
|
for printer, asset, comm, vendor in rows:
|
|
if printer.printerid in seen:
|
|
continue
|
|
seen.add(printer.printerid)
|
|
|
|
supplies = service.getsuppliesbyip_cached(comm.ipaddress)
|
|
if not supplies:
|
|
continue
|
|
itemids = [s['itemid'] for s in supplies if s.get('itemid')]
|
|
history = service.gethistory(itemids, days=days)
|
|
|
|
analysed = []
|
|
for supply in supplies:
|
|
points = history.get(str(supply.get('itemid')), [])
|
|
detail = analyse(points)
|
|
detail['name'] = supply.get('name')
|
|
detail['color'] = supply.get('color')
|
|
# Trust the live read for the level; history can lag a poll behind.
|
|
detail['currentlevel'] = supply.get('level', detail['currentlevel'])
|
|
analysed.append(detail)
|
|
|
|
entry = {
|
|
'printerid': printer.printerid,
|
|
'printername': asset.name or printer.hostname or '',
|
|
'assetnumber': asset.assetnumber or '',
|
|
'ipaddress': comm.ipaddress,
|
|
'vendor': vendor.vendor if vendor else None,
|
|
'supplies': analysed,
|
|
'daysleft': soonest(analysed),
|
|
'replacements': sum(s['replacements'] for s in analysed),
|
|
}
|
|
(estimated if entry['daysleft'] is not None else unestimated).append(entry)
|
|
|
|
# Soonest first: the point of the report is what to order next.
|
|
estimated.sort(key=lambda p: p['daysleft'])
|
|
unestimated.sort(key=lambda p: p['printername'])
|
|
|
|
return success_response({
|
|
'printers': estimated,
|
|
'unestimated': unestimated,
|
|
'days': days,
|
|
'available': True,
|
|
'summary': {
|
|
'estimated': len(estimated),
|
|
'unestimated': len(unestimated),
|
|
'replacements': sum(p['replacements'] for p in estimated + unestimated),
|
|
'duewithin30': sum(1 for p in estimated if p['daysleft'] <= 30),
|
|
},
|
|
})
|