ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add custom fields + warranty plugin, rework settings into two-pane shell

Browse Source 
Feature work from the 2026-07 session:

Settings IA
- Replace the flat 27-card settings hub with a persistent two-pane shell
  (SettingsLayout.vue): grouped, searchable left rail + content pane.
- Nest all settings/* routes under the shell via router post-processing;
  shared nav catalog in settingsNav.js. Group by asset class (PCs, Printers,
  Equipment, Network) so per-type settings stop scattering.

Custom fields (core)
- customfields + customfieldvalues tables (migration 7d14), CRUD API at
  /api/customfields, per-asset value get/save.
- Settings management page + reusable CustomFieldsSection (detail) and
  CustomFieldsInputs (form) wired into all four asset types.

Warranty (new plugin)
- plugins/warranty: warranties + warrantyassets (migration 7d15), derived
  coverage status, provider abstraction (manual now; Dell/Lenovo/HP stubs).
- API CRUD + per-asset panel + report buckets; WarrantyPanel on all four
  detail pages; Warranties management page; Warranty report + Reports card.
- Seed warranty.* permissions.

Printer drivers
- printerdrivers table (migration 7d13) linked to printer models; drivers now
  surface on the matching printer's detail page.

Other
- PCDetail rebalanced (Network + Status + Warranty + custom fields on the right).
- Rename PCs list "Features" column to "Remote Access"; fix badge hover underline.
- Drop equipment islocationonly field.
- Centralize asset-type label/route maps into utils/assetTypes.js.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
cproudlock
2026-07-09 15:37:21 -04:00
parent 419f26107d
commit 78a0ee8d83
154 changed files with 9479 additions and 1098 deletions
Download Patch File Download Diff File Expand all files Collapse all files

220
plugins/computers/api/routes.py
View File

@@ -5,7 +5,9 @@ from flask_jwt_extended import jwt_required
from shopdb.api import db, Asset, AssetType, OperatingSystem, Application, AppVersion, AuditLog, Communication, CommunicationType, success_response, error_response, paginated_response, ErrorCodes, get_pagination_params, paginate_query
from ..models import Computer, ComputerType, ComputerInstalledApp
from ..models import Computer, ComputerType, ComputerInstalledApp, AccessProtocol, ComputerAccess
from shopdb.api import require_permission, require_role
computers_bp = Blueprint('computers', __name__)
@@ -54,6 +56,7 @@ def get_computer_type(type_id: int):
@computers_bp.route('/types', methods=['POST'])
@jwt_required()
@require_permission('computers.create')
def create_computer_type():
"""Create a new computer type."""
data = request.get_json()
@@ -61,7 +64,16 @@ def create_computer_type():
if not data or not data.get('computertype'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'computertype is required')
if ComputerType.query.filter_by(computertype=data['computertype']).first():
existing = ComputerType.query.filter_by(computertype=data['computertype']).first()
if existing:
if not existing.isactive:
# Adding a name that matches a deactivated type revives it.
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 computer type')
return error_response(
ErrorCodes.CONFLICT,
f"Computer type '{data['computertype']}' already exists",
@@ -71,7 +83,7 @@ def create_computer_type():
t = ComputerType(
computertype=data['computertype'],
description=data.get('description'),
icon=data.get('icon')
icon=data.get('icon'), color=data.get('color')
)
db.session.add(t)
@@ -82,6 +94,7 @@ def create_computer_type():
@computers_bp.route('/types/<int:type_id>', methods=['PUT'])
@jwt_required()
@require_permission('computers.edit')
def update_computer_type(type_id: int):
"""Update a computer type."""
t = ComputerType.query.get(type_id)
@@ -105,7 +118,7 @@ def update_computer_type(type_id: int):
http_code=409
)
for key in ['computertype', 'description', 'icon', 'isactive']:
for key in ['computertype', 'description', 'icon', 'color', 'isactive']:
if key in data:
setattr(t, key, data[key])
@@ -113,6 +126,168 @@ def update_computer_type(type_id: int):
return success_response(t.to_dict(), message='Computer type updated')
@computers_bp.route('/types/<int:type_id>', methods=['DELETE'])
@jwt_required()
@require_permission('computers.delete')
def delete_computer_type(type_id: int):
"""Delete a computer type. Refused if any PC still uses it."""
t = ComputerType.query.get(type_id)
if not t:
return error_response(ErrorCodes.NOT_FOUND, 'Computer type not found', http_code=404)
inuse = Computer.query.filter_by(computertypeid=type_id).count()
if inuse:
return error_response(ErrorCodes.CONFLICT,
f"Cannot delete: {inuse} PC(s) still use this type", http_code=409)
db.session.delete(t)
db.session.commit()
return success_response(message='Computer type deleted')
# =============================================================================
# Access protocol catalog (VNC / WinRM / RDP / ...) - admin-managed
# =============================================================================
@computers_bp.route('/protocols', methods=['GET'])
@jwt_required(optional=True)
def list_protocols():
"""List access protocols. ?active=false includes disabled ones."""
query = AccessProtocol.query
if request.args.get('active', 'true').lower() != 'false':
query = query.filter(AccessProtocol.isactive == True)
protocols = query.order_by(AccessProtocol.name).all()
return success_response([p.to_dict() for p in protocols])
@computers_bp.route('/protocols', methods=['POST'])
@jwt_required()
@require_permission('computers.edit')
def create_protocol():
data = request.get_json() or {}
if not (data.get('name') and data.get('scheme') and data.get('linktemplate')):
return error_response(ErrorCodes.VALIDATION_ERROR, 'name, scheme and linktemplate are required')
if AccessProtocol.query.filter_by(name=data['name']).first():
return error_response(ErrorCodes.CONFLICT, f"Protocol '{data['name']}' already exists", http_code=409)
p = AccessProtocol(
name=data['name'],
scheme=data['scheme'],
defaultport=data.get('defaultport') or None,
linktemplate=data['linktemplate'],
isactive=data.get('isactive', True),
)
db.session.add(p)
db.session.commit()
return success_response(p.to_dict(), message='Protocol created', http_code=201)
@computers_bp.route('/protocols/<int:protocol_id>', methods=['PUT', 'PATCH'])
@jwt_required()
@require_permission('computers.edit')
def update_protocol(protocol_id):
p = AccessProtocol.query.get(protocol_id)
if not p:
return error_response(ErrorCodes.NOT_FOUND, 'Protocol not found', http_code=404)
data = request.get_json() or {}
for field in ('name', 'scheme', 'linktemplate'):
if data.get(field):
setattr(p, field, data[field])
if 'defaultport' in data:
p.defaultport = data['defaultport'] or None
if 'isactive' in data:
p.isactive = bool(data['isactive'])
db.session.commit()
return success_response(p.to_dict(), message='Protocol updated')
@computers_bp.route('/protocols/<int:protocol_id>', methods=['DELETE'])
@jwt_required()
@require_permission('computers.edit')
def delete_protocol(protocol_id):
p = AccessProtocol.query.get(protocol_id)
if not p:
return error_response(ErrorCodes.NOT_FOUND, 'Protocol not found', http_code=404)
# If any PC still references it, deactivate rather than hard-delete.
if ComputerAccess.query.filter_by(protocolid=protocol_id).first():
p.isactive = False
db.session.commit()
return success_response(message='Protocol is in use; deactivated instead of deleted')
db.session.delete(p)
db.session.commit()
return success_response(message='Protocol deleted')
def _computer_access_links(comp):
"""Resolved remote-access links for a computer: each enabled protocol's
template filled with the PC hostname joined to the pc_access_domain setting.
A hostname that is already an FQDN (has a dot) is used as-is."""
from shopdb.core.api.settings import get_cached_settings
settings = get_cached_settings()
domain = (settings.get('pc_access_domain') or '').strip()
hostname = (comp.hostname or '').strip()
if not hostname:
host = ''
elif '.' in hostname or not domain:
host = hostname
else:
host = f"{hostname}.{domain}"
links = []
for am in comp.accessmethods:
protocol = am.protocol
if not (am.isactive and protocol and protocol.isactive):
continue
port = am.portoverride or protocol.defaultport
link = None
if host:
try:
link = protocol.linktemplate.format(
host=host,
port=(port if port is not None else ''),
scheme=protocol.scheme,
)
except (KeyError, IndexError, ValueError):
link = None
links.append({
'id': am.id,
'protocolid': protocol.protocolid,
'name': protocol.name,
'scheme': protocol.scheme,
'port': port,
'portoverride': am.portoverride,
'link': link,
})
return links
def _sync_access_methods(comp, data):
"""Replace a computer's enabled protocols from data['accessmethods'] (a list
of {protocolid, portoverride?}). No-op if the key is absent, so callers that
don't touch access aren't affected."""
if 'accessmethods' not in data:
return
desired = data.get('accessmethods') or []
ComputerAccess.query.filter_by(computerid=comp.computerid).delete()
seen = set()
for m in desired:
try:
pid = int(m.get('protocolid'))
except (TypeError, ValueError):
continue
if pid in seen:
continue
seen.add(pid)
port = m.get('portoverride')
try:
port = int(port) if port not in (None, '') else None
except (TypeError, ValueError):
port = None
db.session.add(ComputerAccess(
computerid=comp.computerid,
protocolid=pid,
portoverride=port,
isactive=True,
))
# =============================================================================
# Computers CRUD
# =============================================================================
@@ -169,9 +344,15 @@ def list_computers():
if bu_id := request.args.get('businessunitid', request.args.get('businessunit_id')):
query = query.filter(Asset.businessunitid == int(bu_id))
# Shopfloor filter
# Shopfloor filter (by the Shopfloor computer type)
if shopfloor := request.args.get('shopfloor'):
query = query.filter(Computer.isshopfloor == (shopfloor.lower() == 'true'))
sf = ComputerType.query.filter_by(computertype='Shopfloor').first()
sf_id = sf.computertypeid if sf else -1
if shopfloor.lower() == 'true':
query = query.filter(Computer.computertypeid == sf_id)
else:
query = query.filter(db.or_(Computer.computertypeid != sf_id,
Computer.computertypeid.is_(None)))
# Sorting
sort_by = request.args.get('sort', 'hostname')
@@ -197,6 +378,7 @@ def list_computers():
for comp in items:
item = comp.asset.to_dict() if comp.asset else {}
item['computer'] = comp.to_dict()
item['accessmethods'] = _computer_access_links(comp)
data.append(item)
return paginated_response(data, page, per_page, total)
@@ -221,6 +403,7 @@ def get_computer(computer_id: int):
c.to_dict() for c in
Communication.query.filter_by(assetid=comp.assetid).all()
]
result['accessmethods'] = _computer_access_links(comp)
return success_response(result)
@@ -265,6 +448,7 @@ def get_computer_by_hostname(hostname: str):
@computers_bp.route('', methods=['POST'])
@jwt_required()
@require_permission('computers.create')
def create_computer():
"""
Create new computer (creates both Asset and Computer records).
@@ -275,7 +459,6 @@ def create_computer():
Optional fields:
- name, serialnumber, statusid, locationid, businessunitid
- computertypeid, hostname, osid
- isvnc, iswinrm, isshopfloor
- mapx, mapy, notes
"""
data = request.get_json()
@@ -341,10 +524,7 @@ def create_computer():
modelnumberid=data.get('modelnumberid'),
loggedinuser=data.get('loggedinuser'),
lastreporteddate=data.get('lastreporteddate'),
lastboottime=data.get('lastboottime'),
isvnc=data.get('isvnc', False),
iswinrm=data.get('iswinrm', False),
isshopfloor=data.get('isshopfloor', False)
lastboottime=data.get('lastboottime')
)
db.session.add(comp)
@@ -361,6 +541,9 @@ def create_computer():
isprimary=True,
))
# Remote-access protocols
_sync_access_methods(comp, data)
# Audit log
AuditLog.log('created', 'Computer', entityid=comp.computerid,
entityname=data.get('hostname') or data['assetnumber'])
@@ -369,12 +552,14 @@ def create_computer():
result = asset.to_dict()
result['computer'] = comp.to_dict()
result['accessmethods'] = _computer_access_links(comp)
return success_response(result, message='Computer created', http_code=201)
@computers_bp.route('/<int:computer_id>', methods=['PUT'])
@jwt_required()
@require_permission('computers.edit')
def update_computer(computer_id: int):
"""Update computer (both Asset and Computer records)."""
comp = Computer.query.get(computer_id)
@@ -429,7 +614,7 @@ def update_computer(computer_id: int):
# Update computer fields
computer_fields = ['computertypeid', 'hostname', 'osid', 'vendorid',
'modelnumberid', 'loggedinuser', 'lastreporteddate',
'lastboottime', 'isvnc', 'iswinrm', 'isshopfloor']
'lastboottime']
for key in computer_fields:
if key in data:
old_val = getattr(comp, key)
@@ -455,6 +640,9 @@ def update_computer(computer_id: int):
elif primary:
primary.ipaddress = None
# Remote-access protocols
_sync_access_methods(comp, data)
# Audit log if there were changes
if changes:
AuditLog.log('updated', 'Computer', entityid=comp.computerid,
@@ -464,12 +652,14 @@ def update_computer(computer_id: int):
result = asset.to_dict()
result['computer'] = comp.to_dict()
result['accessmethods'] = _computer_access_links(comp)
return success_response(result, message='Computer updated')
@computers_bp.route('/<int:computer_id>', methods=['DELETE'])
@jwt_required()
@require_permission('computers.delete')
def delete_computer(computer_id: int):
"""Delete (soft delete) computer."""
comp = Computer.query.get(computer_id)
@@ -522,6 +712,7 @@ def get_installed_apps(computer_id: int):
@computers_bp.route('/<int:computer_id>/apps', methods=['POST'])
@jwt_required()
@require_permission('computers.create')
def add_installed_app(computer_id: int):
"""Add an installed application to a computer."""
comp = Computer.query.get(computer_id)
@@ -578,6 +769,7 @@ def add_installed_app(computer_id: int):
@computers_bp.route('/<int:computer_id>/apps/<int:app_id>', methods=['DELETE'])
@jwt_required()
@require_permission('computers.delete')
def remove_installed_app(computer_id: int, app_id: int):
"""Remove an installed application from a computer."""
installed = ComputerInstalledApp.query.filter_by(
@@ -605,6 +797,7 @@ def remove_installed_app(computer_id: int, app_id: int):
@computers_bp.route('/<int:computer_id>/report', methods=['POST'])
@jwt_required()
@require_permission('computers.create')
def report_status(computer_id: int):
"""
Report computer status (for agent-based reporting).
@@ -671,9 +864,10 @@ def dashboard_summary():
).all()
# Count shopfloor vs non-shopfloor
sf = ComputerType.query.filter_by(computertype='Shopfloor').first()
shopfloor_count = db.session.query(Computer).join(Asset).filter(
Asset.isactive == True,
Computer.isshopfloor == True
Computer.computertypeid == (sf.computertypeid if sf else -1)
).count()
return success_response({

10
plugins/computers/models/__init__.py
View File

@@ -1,9 +1,17 @@
"""Computers plugin models."""
from .computer import Computer, ComputerType, ComputerInstalledApp
from .computer import (
Computer,
ComputerType,
ComputerInstalledApp,
AccessProtocol,
ComputerAccess,
)
__all__ = [
'Computer',
'ComputerType',
'ComputerInstalledApp',
'AccessProtocol',
'ComputerAccess',
]

90
plugins/computers/models/computer.py
View File

@@ -15,6 +15,7 @@ class ComputerType(BaseModel):
computertype = db.Column(db.String(100), unique=True, nullable=False)
description = db.Column(db.Text)
icon = db.Column(db.String(50), comment='Icon name for UI')
color = db.Column(db.String(20), comment='CSS color for UI/map markers')
def __repr__(self):
return f"<ComputerType {self.computertype}>"
@@ -78,24 +79,8 @@ class Computer(BaseModel):
lastreporteddate = db.Column(db.DateTime, nullable=True)
lastboottime = db.Column(db.DateTime, nullable=True)
# Remote access features
isvnc = db.Column(
db.Boolean,
default=False,
comment='VNC remote access enabled'
)
iswinrm = db.Column(
db.Boolean,
default=False,
comment='WinRM enabled'
)
# Classification flags
isshopfloor = db.Column(
db.Boolean,
default=False,
comment='Shopfloor PC (vs office PC)'
)
# Remote access is now modeled per-protocol via the accessmethods
# relationship (AccessProtocol / ComputerAccess), replacing isvnc/iswinrm.
# Relationships
asset = db.relationship(
@@ -115,6 +100,14 @@ class Computer(BaseModel):
lazy='dynamic'
)
# Remote-access protocols enabled on this PC (replaces isvnc/iswinrm)
accessmethods = db.relationship(
'ComputerAccess',
back_populates='computer',
cascade='all, delete-orphan',
lazy='selectin'
)
__table_args__ = (
db.Index('idx_computer_type', 'computertypeid'),
db.Index('idx_computer_hostname', 'hostname'),
@@ -138,6 +131,12 @@ class Computer(BaseModel):
if self.model:
result['modelname'] = self.model.modelnumber
# Names of enabled remote-access protocols (for list badges)
result['accessprotocolnames'] = [
am.protocol.name for am in self.accessmethods
if am.isactive and am.protocol and am.protocol.isactive
]
return result
@@ -182,6 +181,61 @@ class ComputerInstalledApp(db.Model):
db.Index('idx_compapp_app', 'appid'),
)
class AccessProtocol(db.Model):
"""
Catalog of remote-access protocols a PC can expose (VNC, WinRM, RDP, SSH...).
linktemplate builds a connection URL from placeholders {host}, {port},
{scheme}. {host} is the PC hostname joined to the pc_access_domain setting.
Admin-managed; replaces the old fixed isvnc/iswinrm booleans.
"""
__tablename__ = 'accessprotocols'
protocolid = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(50), unique=True, nullable=False)
scheme = db.Column(db.String(20), nullable=False)
defaultport = db.Column(db.Integer, nullable=True)
linktemplate = db.Column(db.String(255), nullable=False)
isactive = db.Column(db.Boolean, default=True, nullable=False)
def to_dict(self):
return {
'protocolid': self.protocolid,
'name': self.name,
'scheme': self.scheme,
'defaultport': self.defaultport,
'linktemplate': self.linktemplate,
'isactive': bool(self.isactive),
}
class ComputerAccess(db.Model):
"""A protocol enabled on a specific PC, with an optional port override."""
__tablename__ = 'computeraccess'
id = db.Column(db.Integer, primary_key=True)
computerid = db.Column(
db.Integer,
db.ForeignKey('computers.computerid', ondelete='CASCADE'),
nullable=False
)
protocolid = db.Column(
db.Integer,
db.ForeignKey('accessprotocols.protocolid'),
nullable=False
)
portoverride = db.Column(db.Integer, nullable=True)
isactive = db.Column(db.Boolean, default=True, nullable=False)
protocol = db.relationship('AccessProtocol')
computer = db.relationship('Computer', back_populates='accessmethods')
__table_args__ = (
db.UniqueConstraint('computerid', 'protocolid', name='uq_computer_protocol'),
db.Index('idx_compaccess_computer', 'computerid'),
)
def to_dict(self):
"""Convert to dictionary."""
return {

9
plugins/computers/plugin.py
View File

@@ -11,7 +11,7 @@ import click
from shopdb.plugins.base import BasePlugin, PluginMeta
from shopdb.api import db, AssetType
from .models import Computer, ComputerType, ComputerInstalledApp
from .models import Computer, ComputerType, ComputerInstalledApp, AccessProtocol, ComputerAccess
from .api import computers_bp
logger = logging.getLogger(__name__)
@@ -58,7 +58,7 @@ class ComputersPlugin(BasePlugin):
def get_models(self) -> List[Type]:
"""Return list of SQLAlchemy model classes."""
return [Computer, ComputerType, ComputerInstalledApp]
return [Computer, ComputerType, ComputerInstalledApp, AccessProtocol, ComputerAccess]
def init_app(self, app: Flask, db_instance) -> None:
"""Initialize plugin with Flask app."""
@@ -329,10 +329,11 @@ class ComputersPlugin(BasePlugin):
click.echo(f"Total active computers: {total}")
# Shopfloor count
# Shopfloor count (by the Shopfloor computer type)
sf = ComputerType.query.filter_by(computertype='Shopfloor').first()
shopfloor = db.session.query(Computer).join(Asset).filter(
Asset.isactive == True,
Computer.isshopfloor == True
Computer.computertypeid == (sf.computertypeid if sf else -1)
).count()
click.echo(f" Shopfloor PCs: {shopfloor}")

38
plugins/equipment/api/routes.py
View File

@@ -7,6 +7,8 @@ from shopdb.api import db, Asset, AssetType, Vendor, Model, AuditLog, success_re
from ..models import Equipment, EquipmentType
from shopdb.api import require_permission, require_role
equipment_bp = Blueprint('equipment', __name__)
@@ -54,6 +56,7 @@ def get_equipment_type(type_id: int):
@equipment_bp.route('/types', methods=['POST'])
@jwt_required()
@require_permission('equipment.create')
def create_equipment_type():
"""Create a new equipment type."""
data = request.get_json()
@@ -61,7 +64,15 @@ def create_equipment_type():
if not data or not data.get('equipmenttype'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'equipmenttype is required')
if EquipmentType.query.filter_by(equipmenttype=data['equipmenttype']).first():
existing = EquipmentType.query.filter_by(equipmenttype=data['equipmenttype']).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"Equipment type '{data['equipmenttype']}' already exists",
@@ -71,7 +82,7 @@ def create_equipment_type():
t = EquipmentType(
equipmenttype=data['equipmenttype'],
description=data.get('description'),
icon=data.get('icon')
icon=data.get('icon'), color=data.get('color')
)
db.session.add(t)
@@ -82,6 +93,7 @@ def create_equipment_type():
@equipment_bp.route('/types/<int:type_id>', methods=['PUT'])
@jwt_required()
@require_permission('equipment.edit')
def update_equipment_type(type_id: int):
"""Update an equipment type."""
t = EquipmentType.query.get(type_id)
@@ -105,7 +117,7 @@ def update_equipment_type(type_id: int):
http_code=409
)
for key in ['equipmenttype', 'description', 'icon', 'isactive']:
for key in ['equipmenttype', 'description', 'icon', 'color', 'isactive']:
if key in data:
setattr(t, key, data[key])
@@ -113,6 +125,23 @@ def update_equipment_type(type_id: int):
return success_response(t.to_dict(), message='Equipment type updated')
@equipment_bp.route('/types/<int:type_id>', methods=['DELETE'])
@jwt_required()
@require_permission('equipment.delete')
def delete_equipment_type(type_id: int):
"""Delete an equipment type. Refused if any asset still uses it."""
t = EquipmentType.query.get(type_id)
if not t:
return error_response(ErrorCodes.NOT_FOUND, 'Equipment type not found', http_code=404)
inuse = Equipment.query.filter_by(equipmenttypeid=type_id).count()
if inuse:
return error_response(ErrorCodes.CONFLICT,
f"Cannot delete: {inuse} asset(s) still use this type", http_code=409)
db.session.delete(t)
db.session.commit()
return success_response(message='Equipment type deleted')
# =============================================================================
# Equipment CRUD
# =============================================================================
@@ -232,6 +261,7 @@ def get_equipment_by_asset(asset_id: int):
@equipment_bp.route('', methods=['POST'])
@jwt_required()
@require_permission('equipment.create')
def create_equipment():
"""
Create new equipment (creates both Asset and Equipment records).
@@ -321,6 +351,7 @@ def create_equipment():
@equipment_bp.route('/<int:equipment_id>', methods=['PUT'])
@jwt_required()
@require_permission('equipment.edit')
def update_equipment(equipment_id: int):
"""Update equipment (both Asset and Equipment records)."""
equip = Equipment.query.get(equipment_id)
@@ -391,6 +422,7 @@ def update_equipment(equipment_id: int):
@equipment_bp.route('/<int:equipment_id>', methods=['DELETE'])
@jwt_required()
@require_permission('equipment.delete')
def delete_equipment(equipment_id: int):
"""Delete (soft delete) equipment."""
equip = Equipment.query.get(equipment_id)

1
plugins/equipment/models/equipment.py
View File

@@ -15,6 +15,7 @@ class EquipmentType(BaseModel):
equipmenttype = db.Column(db.String(100), unique=True, nullable=False)
description = db.Column(db.Text)
icon = db.Column(db.String(50), comment='Icon name for UI')
color = db.Column(db.String(20), comment='CSS color for UI/map markers')
def __repr__(self):
return f"<EquipmentType {self.equipmenttype}>"

5
plugins/knowledgebase/api/routes.py
View File

@@ -16,6 +16,8 @@ from shopdb.api import (
from ..models import KnowledgeBase
from shopdb.api import require_permission, require_role
knowledgebase_bp = Blueprint('knowledgebase', __name__)
@@ -137,6 +139,7 @@ def track_click(link_id: int):
@knowledgebase_bp.route('', methods=['POST'])
@jwt_required()
@require_permission('kb.create')
def create_article():
"""Create a new knowledge base article."""
data = request.get_json()
@@ -169,6 +172,7 @@ def create_article():
@knowledgebase_bp.route('/<int:link_id>', methods=['PUT'])
@jwt_required()
@require_permission('kb.edit')
def update_article(link_id: int):
"""Update a knowledge base article."""
article = KnowledgeBase.query.get(link_id)
@@ -197,6 +201,7 @@ def update_article(link_id: int):
@knowledgebase_bp.route('/<int:link_id>', methods=['DELETE'])
@jwt_required()
@require_permission('kb.delete')
def delete_article(link_id: int):
"""Delete (deactivate) a knowledge base article."""
article = KnowledgeBase.query.get(link_id)

44
plugins/network/api/routes.py
View File

@@ -7,6 +7,8 @@ from shopdb.api import db, Asset, AssetType, Vendor, AuditLog, success_response,
from ..models import NetworkDevice, NetworkDeviceType, Subnet, VLAN
from shopdb.api import require_permission, require_role
network_bp = Blueprint('network', __name__)
@@ -54,6 +56,7 @@ def get_network_device_type(type_id: int):
@network_bp.route('/types', methods=['POST'])
@jwt_required()
@require_permission('network.create')
def create_network_device_type():
"""Create a new network device type."""
data = request.get_json()
@@ -61,7 +64,15 @@ def create_network_device_type():
if not data or not data.get('networkdevicetype'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'networkdevicetype is required')
if NetworkDeviceType.query.filter_by(networkdevicetype=data['networkdevicetype']).first():
existing = NetworkDeviceType.query.filter_by(networkdevicetype=data['networkdevicetype']).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"Network device type '{data['networkdevicetype']}' already exists",
@@ -71,7 +82,7 @@ def create_network_device_type():
t = NetworkDeviceType(
networkdevicetype=data['networkdevicetype'],
description=data.get('description'),
icon=data.get('icon')
icon=data.get('icon'), color=data.get('color')
)
db.session.add(t)
@@ -82,6 +93,7 @@ def create_network_device_type():
@network_bp.route('/types/<int:type_id>', methods=['PUT'])
@jwt_required()
@require_permission('network.edit')
def update_network_device_type(type_id: int):
"""Update a network device type."""
t = NetworkDeviceType.query.get(type_id)
@@ -105,7 +117,7 @@ def update_network_device_type(type_id: int):
http_code=409
)
for key in ['networkdevicetype', 'description', 'icon', 'isactive']:
for key in ['networkdevicetype', 'description', 'icon', 'color', 'isactive']:
if key in data:
setattr(t, key, data[key])
@@ -113,6 +125,23 @@ def update_network_device_type(type_id: int):
return success_response(t.to_dict(), message='Network device type updated')
@network_bp.route('/types/<int:type_id>', methods=['DELETE'])
@jwt_required()
@require_permission('network.delete')
def delete_network_device_type(type_id: int):
"""Delete a network device type. Refused if any device still uses it."""
t = NetworkDeviceType.query.get(type_id)
if not t:
return error_response(ErrorCodes.NOT_FOUND, 'Network device type not found', http_code=404)
inuse = NetworkDevice.query.filter_by(networkdevicetypeid=type_id).count()
if inuse:
return error_response(ErrorCodes.CONFLICT,
f"Cannot delete: {inuse} device(s) still use this type", http_code=409)
db.session.delete(t)
db.session.commit()
return success_response(message='Network device type deleted')
# =============================================================================
# Network Devices CRUD
# =============================================================================
@@ -264,6 +293,7 @@ def get_network_device_by_hostname(hostname: str):
@network_bp.route('', methods=['POST'])
@jwt_required()
@require_permission('network.create')
def create_network_device():
"""
Create new network device (creates both Asset and NetworkDevice records).
@@ -360,6 +390,7 @@ def create_network_device():
@network_bp.route('/<int:device_id>', methods=['PUT'])
@jwt_required()
@require_permission('network.edit')
def update_network_device(device_id: int):
"""Update network device (both Asset and NetworkDevice records)."""
netdev = NetworkDevice.query.get(device_id)
@@ -437,6 +468,7 @@ def update_network_device(device_id: int):
@network_bp.route('/<int:device_id>', methods=['DELETE'])
@jwt_required()
@require_permission('network.delete')
def delete_network_device(device_id: int):
"""Delete (soft delete) network device."""
netdev = NetworkDevice.query.get(device_id)
@@ -567,6 +599,7 @@ def get_vlan(vlan_id: int):
@network_bp.route('/vlans', methods=['POST'])
@jwt_required()
@require_permission('network.create')
def create_vlan():
"""Create a new VLAN."""
data = request.get_json()
@@ -608,6 +641,7 @@ def create_vlan():
@network_bp.route('/vlans/<int:vlan_id>', methods=['PUT'])
@jwt_required()
@require_permission('network.edit')
def update_vlan(vlan_id: int):
"""Update a VLAN."""
vlan = VLAN.query.get(vlan_id)
@@ -653,6 +687,7 @@ def update_vlan(vlan_id: int):
@network_bp.route('/vlans/<int:vlan_id>', methods=['DELETE'])
@jwt_required()
@require_permission('network.delete')
def delete_vlan(vlan_id: int):
"""Delete (soft delete) a VLAN."""
vlan = VLAN.query.get(vlan_id)
@@ -746,6 +781,7 @@ def get_subnet(subnet_id: int):
@network_bp.route('/subnets', methods=['POST'])
@jwt_required()
@require_permission('network.create')
def create_subnet():
"""Create a new subnet."""
data = request.get_json()
@@ -811,6 +847,7 @@ def create_subnet():
@network_bp.route('/subnets/<int:subnet_id>', methods=['PUT'])
@jwt_required()
@require_permission('network.edit')
def update_subnet(subnet_id: int):
"""Update a subnet."""
subnet = Subnet.query.get(subnet_id)
@@ -861,6 +898,7 @@ def update_subnet(subnet_id: int):
@network_bp.route('/subnets/<int:subnet_id>', methods=['DELETE'])
@jwt_required()
@require_permission('network.delete')
def delete_subnet(subnet_id: int):
"""Delete (soft delete) a subnet."""
subnet = Subnet.query.get(subnet_id)

1
plugins/network/models/network_device.py
View File

@@ -15,6 +15,7 @@ class NetworkDeviceType(BaseModel):
networkdevicetype = db.Column(db.String(100), unique=True, nullable=False)
description = db.Column(db.Text)
icon = db.Column(db.String(50), comment='Icon name for UI')
color = db.Column(db.String(20), comment='CSS color for UI/map markers')
def __repr__(self):
return f"<NetworkDeviceType {self.networkdevicetype}>"

325
plugins/notifications/api/routes.py
View File

@@ -1,6 +1,10 @@
"""Notifications plugin API endpoints - adapted to existing schema."""
from datetime import datetime
import hashlib
import os
from datetime import datetime, timedelta, timezone
from zoneinfo import ZoneInfo
from flask import Blueprint, request
from flask_jwt_extended import jwt_required
@@ -8,8 +12,157 @@ from shopdb.api import db, success_response, error_response, paginated_response,
from ..models import Notification, NotificationType
from shopdb.api import require_permission, require_role
notifications_bp = Blueprint('notifications', __name__)
# Notification types whose multi-employee cards get split into one card per
# employee on the shopfloor dashboard. typecolor drives this (not typename), so
# recognition, training and recertification all fan out; every other type
# stays one card.
SPLIT_TYPECOLORS = frozenset({'recognition', 'training', 'recertification'})
# How long a card stays up on the shopfloor board when the creator does not set
# an explicit end time, by notification typecolor.
# recognition - clears at the next 8:00 AM Eastern (daily reset)
# recertification - stays up two weeks (employees have time to book the course)
EASTERN = ZoneInfo('America/New_York')
RECERTIFICATION_DAYS = 14
def _next_eastern_time(after, hour, minute=0):
"""Next hour:minute America/New_York strictly after `after` (naive UTC),
returned as naive UTC. Uses the tz database so it is correct across EST/EDT."""
after_east = after.replace(tzinfo=timezone.utc).astimezone(EASTERN)
target = after_east.replace(hour=int(hour), minute=int(minute), second=0, microsecond=0)
if target <= after_east:
target += timedelta(days=1)
return target.astimezone(timezone.utc).replace(tzinfo=None)
def _auto_endtime(ntype, starttime):
"""Default display window for a notification with no explicit end time, from
the notification type's configured expiry rule:
'dailytime' -> next expiryhour:expiryminute Eastern (daily reset)
'duration' -> starttime + expirydays days
'none' -> None (show indefinitely)
Falls back to the legacy typecolor rules when the expiry columns are unset,
so it is safe before/after the 7d03 migration."""
mode = getattr(ntype, 'expirymode', None) or 'none'
if mode == 'dailytime':
hour = ntype.expiryhour if ntype.expiryhour is not None else 8
return _next_eastern_time(starttime, hour, ntype.expiryminute or 0)
if mode == 'duration' and ntype.expirydays:
return starttime + timedelta(days=int(ntype.expirydays))
if mode == 'none':
# legacy fallback for rule-bearing types created before the expiry columns
if getattr(ntype, 'typecolor', None) == 'recognition':
return _next_eastern_time(starttime, 8, 0)
if getattr(ntype, 'typecolor', None) == 'recertification':
return starttime + timedelta(days=RECERTIFICATION_DAYS)
return None
_EXPIRY_MODES = ('none', 'duration', 'dailytime')
def _apply_expiry_fields(t, data):
"""Set expiry-rule columns on a NotificationType from request data. Only
touches fields that are present. Returns an error string, or None on ok."""
if 'expirymode' in data:
mode = data.get('expirymode') or 'none'
if mode not in _EXPIRY_MODES:
return "expirymode must be one of: %s" % ", ".join(_EXPIRY_MODES)
t.expirymode = mode
if 'expirydays' in data:
v = data.get('expirydays')
if v in (None, ''):
t.expirydays = None
else:
try:
t.expirydays = int(v)
except (TypeError, ValueError):
return "expirydays must be an integer"
if t.expirydays < 1:
return "expirydays must be >= 1"
if 'expiryhour' in data:
v = data.get('expiryhour')
if v in (None, ''):
t.expiryhour = None
else:
try:
t.expiryhour = int(v)
except (TypeError, ValueError):
return "expiryhour must be an integer"
if not (0 <= t.expiryhour <= 23):
return "expiryhour must be 0-23"
if 'expiryminute' in data:
v = data.get('expiryminute')
try:
t.expiryminute = int(v) if v not in (None, '') else 0
except (TypeError, ValueError):
return "expiryminute must be an integer"
if not (0 <= t.expiryminute <= 59):
return "expiryminute must be 0-59"
# cross-field consistency
mode = t.expirymode or 'none'
if mode == 'dailytime' and t.expiryhour is None:
t.expiryhour = 8
if mode == 'duration' and not t.expirydays:
return "duration expiry requires expirydays >= 1"
return None
_DISPLAY_STYLES = ('standard', 'carousel', 'grid', 'banner')
def _apply_display_fields(t, data):
"""Set shopfloor display-behavior columns on a NotificationType from request
data. Only touches fields that are present. Returns an error string, or None."""
if 'splitperemployee' in data:
t.splitperemployee = bool(data.get('splitperemployee'))
if 'showemployeephoto' in data:
t.showemployeephoto = bool(data.get('showemployeephoto'))
if 'displaystyle' in data:
ds = data.get('displaystyle') or 'standard'
if ds not in _DISPLAY_STYLES:
return "displaystyle must be one of: %s" % ", ".join(_DISPLAY_STYLES)
t.displaystyle = ds
return None
def _config_version():
"""Short hash of everything that changes the board's LAYOUT: per-type display
config plus an optional deploy stamp (SHOPFLOOR_BUILD env). The shopfloor
kiosks reload when this changes, so type/layout edits and frontend deploys
reach pages that are already open."""
types = NotificationType.query.order_by(NotificationType.notificationtypeid).all()
parts = [
"%s|%s|%s|%d|%d|%s|%s|%s|%d" % (
t.notificationtypeid, t.typecolor, t.displaystyle,
int(bool(t.splitperemployee)), int(bool(t.showemployeephoto)),
t.expirymode, t.expirydays, t.expiryhour, int(bool(t.isactive)),
)
for t in types
]
parts.append(os.environ.get('SHOPFLOOR_BUILD', ''))
return hashlib.md5('||'.join(parts).encode()).hexdigest()[:12]
def _employee_picture(sso):
"""Best-effort Picture blob for an SSO from the HR directory. None on any miss."""
if not (sso and str(sso).isdigit()):
return None
try:
conn = employee_connection()
with conn.cursor() as cur:
cur.execute('SELECT Picture FROM employees WHERE SSO = %s', (int(sso),))
emp = cur.fetchone()
conn.close()
return emp.get('Picture') if emp else None
except Exception:
return None
# =============================================================================
# Notification Types
@@ -35,6 +188,7 @@ def list_notification_types():
@notifications_bp.route('/types', methods=['POST'])
@jwt_required()
@require_permission('notifications.create')
def create_notification_type():
"""Create a new notification type."""
data = request.get_json()
@@ -55,12 +209,57 @@ def create_notification_type():
typecolor=data.get('typecolor') or data.get('color', '#17a2b8')
)
err = _apply_expiry_fields(t, data)
if err:
return error_response(ErrorCodes.VALIDATION_ERROR, err)
err = _apply_display_fields(t, data)
if err:
return error_response(ErrorCodes.VALIDATION_ERROR, err)
db.session.add(t)
db.session.commit()
return success_response(t.to_dict(), message='Notification type created', http_code=201)
@notifications_bp.route('/types/<int:type_id>', methods=['PUT', 'PATCH'])
@jwt_required()
@require_permission('notifications.create')
def update_notification_type(type_id: int):
"""Update a notification type, including its auto-expiry rule."""
t = NotificationType.query.get(type_id)
if not t:
return error_response(ErrorCodes.NOT_FOUND, f'Notification type {type_id} not found', http_code=404)
data = request.get_json() or {}
if data.get('typename'):
dup = NotificationType.query.filter(
NotificationType.typename == data['typename'],
NotificationType.notificationtypeid != type_id
).first()
if dup:
return error_response(ErrorCodes.CONFLICT,
f"Notification type '{data['typename']}' already exists", http_code=409)
t.typename = data['typename']
if 'typedescription' in data or 'description' in data:
t.typedescription = data.get('typedescription') or data.get('description')
if 'typecolor' in data or 'color' in data:
t.typecolor = data.get('typecolor') or data.get('color')
if 'isactive' in data:
t.isactive = bool(data['isactive'])
err = _apply_expiry_fields(t, data)
if err:
return error_response(ErrorCodes.VALIDATION_ERROR, err)
err = _apply_display_fields(t, data)
if err:
return error_response(ErrorCodes.VALIDATION_ERROR, err)
db.session.commit()
return success_response(t.to_dict(), message='Notification type updated')
# =============================================================================
# Notifications CRUD
# =============================================================================
@@ -132,6 +331,7 @@ def get_notification(notification_id: int):
@notifications_bp.route('', methods=['POST'])
@jwt_required()
@require_permission('notifications.create')
def create_notification():
"""Create a new notification."""
data = request.get_json()
@@ -161,6 +361,13 @@ def create_notification():
except ValueError:
return error_response(ErrorCodes.VALIDATION_ERROR, 'Invalid endtime format')
# No explicit end time: apply the per-type display window (recognition
# clears at the next 8 AM Eastern, recertification runs two weeks).
if endtime is None and data.get('notificationtypeid'):
ntype = NotificationType.query.get(data['notificationtypeid'])
if ntype:
endtime = _auto_endtime(ntype, starttime)
n = Notification(
notification=notification_text,
notificationtypeid=data.get('notificationtypeid'),
@@ -184,6 +391,7 @@ def create_notification():
@notifications_bp.route('/<int:notification_id>', methods=['PUT'])
@jwt_required()
@require_permission('notifications.edit')
def update_notification(notification_id: int):
"""Update a notification."""
n = Notification.query.get(notification_id)
@@ -250,6 +458,7 @@ def update_notification(notification_id: int):
@notifications_bp.route('/<int:notification_id>', methods=['DELETE'])
@jwt_required()
@require_permission('notifications.delete')
def delete_notification(notification_id: int):
"""Delete (soft delete) a notification."""
n = Notification.query.get(notification_id)
@@ -427,7 +636,7 @@ def get_shopfloor_notifications():
Get notifications for shopfloor TV dashboard.
Returns current and upcoming notifications with isshopfloor=1.
Splits multi-employee recognition into separate entries.
Splits multi-employee recognition and training into separate entries.
Query parameters:
- businessunit: Filter by business unit ID (null = all units)
@@ -487,6 +696,8 @@ def get_shopfloor_notifications():
def notification_to_shopfloor(n, employee_override=None):
"""Convert notification to shopfloor format."""
is_resolved = n.endtime and n.endtime < now
ntype = n.notificationtype
show_photo = bool(ntype and ntype.showemployeephoto)
result = {
'notificationid': n.notificationid,
@@ -498,11 +709,13 @@ def get_shopfloor_notifications():
'isactive': n.isactive,
'isshopfloor': True,
'resolved': is_resolved,
'typename': n.notificationtype.typename if n.notificationtype else None,
'typecolor': n.notificationtype.typecolor if n.notificationtype else None,
'typename': ntype.typename if ntype else None,
'typecolor': ntype.typecolor if ntype else None,
# Per-type display behavior the dashboard groups/renders by.
'displaystyle': (ntype.displaystyle or 'standard') if ntype else 'standard',
}
# Employee info
# Employee info (photo only when the type wants it)
if employee_override:
result['employeesso'] = employee_override.get('sso')
result['employeename'] = employee_override.get('name')
@@ -510,92 +723,36 @@ def get_shopfloor_notifications():
else:
result['employeesso'] = n.employeesso
result['employeename'] = n.employeename
result['employeepicture'] = None
# Try to get picture from wjf_employees
if n.employeesso and n.employeesso.isdigit():
try:
conn = employee_connection()
with conn.cursor() as cur:
cur.execute('SELECT Picture FROM employees WHERE SSO = %s', (int(n.employeesso),))
emp = cur.fetchone()
if emp and emp.get('Picture'):
result['employeepicture'] = emp['Picture']
conn.close()
except Exception:
pass
result['employeepicture'] = _employee_picture(n.employeesso) if show_photo else None
return result
# Process current notifications (split multi-employee recognition)
current_data = []
for n in current_notifications:
is_recognition = n.notificationtype and n.notificationtype.typecolor == 'recognition'
def expand(n):
"""One shopfloor card per notification, or one per employee when the
type is configured to split multi-employee lists."""
ntype = n.notificationtype
is_split = bool(ntype and ntype.splitperemployee)
if not (is_split and n.employeesso and ',' in n.employeesso):
return [notification_to_shopfloor(n)]
if is_recognition and n.employeesso and ',' in n.employeesso:
# Split into individual cards for each employee
ssos = [s.strip() for s in n.employeesso.split(',')]
names = n.employeename.split(', ') if n.employeename else []
show_photo = bool(ntype and ntype.showemployeephoto)
ssos = [s.strip() for s in n.employeesso.split(',')]
names = n.employeename.split(', ') if n.employeename else []
return [
notification_to_shopfloor(n, {
'sso': sso,
'name': names[i] if i < len(names) else sso,
'picture': _employee_picture(sso) if show_photo else None,
})
for i, sso in enumerate(ssos)
]
for i, sso in enumerate(ssos):
name = names[i] if i < len(names) else sso
# Look up picture
picture = None
if sso.isdigit():
try:
conn = employee_connection()
with conn.cursor() as cur:
cur.execute('SELECT Picture FROM employees WHERE SSO = %s', (int(sso),))
emp = cur.fetchone()
if emp:
picture = emp.get('Picture')
conn.close()
except Exception:
pass
current_data.append(notification_to_shopfloor(n, {
'sso': sso,
'name': name,
'picture': picture
}))
else:
current_data.append(notification_to_shopfloor(n))
# Process upcoming notifications
upcoming_data = []
for n in upcoming_notifications:
is_recognition = n.notificationtype and n.notificationtype.typecolor == 'recognition'
if is_recognition and n.employeesso and ',' in n.employeesso:
ssos = [s.strip() for s in n.employeesso.split(',')]
names = n.employeename.split(', ') if n.employeename else []
for i, sso in enumerate(ssos):
name = names[i] if i < len(names) else sso
picture = None
if sso.isdigit():
try:
conn = employee_connection()
with conn.cursor() as cur:
cur.execute('SELECT Picture FROM employees WHERE SSO = %s', (int(sso),))
emp = cur.fetchone()
if emp:
picture = emp.get('Picture')
conn.close()
except Exception:
pass
upcoming_data.append(notification_to_shopfloor(n, {
'sso': sso,
'name': name,
'picture': picture
}))
else:
upcoming_data.append(notification_to_shopfloor(n))
current_data = [card for n in current_notifications for card in expand(n)]
upcoming_data = [card for n in upcoming_notifications for card in expand(n)]
return success_response({
'timestamp': now.isoformat(),
'current': current_data,
'upcoming': upcoming_data
'upcoming': upcoming_data,
'configversion': _config_version(),
})

56
plugins/notifications/models/notification.py
View File

@@ -17,6 +17,25 @@ class NotificationType(db.Model):
typecolor = db.Column(db.String(20), default='#17a2b8')
isactive = db.Column(db.Boolean, default=True)
# Auto-expiry rule: when a notification of this type has no explicit end time,
# how long it stays up on the shopfloor board.
# 'none' -> indefinite (never auto-expires)
# 'duration' -> starttime + expirydays days
# 'dailytime' -> next expiryhour:expiryminute Eastern (daily reset)
expirymode = db.Column(db.String(20), default='none')
expirydays = db.Column(db.Integer, nullable=True)
expiryhour = db.Column(db.SmallInteger, nullable=True)
expiryminute = db.Column(db.SmallInteger, nullable=True, default=0)
# Shopfloor display behavior (data-driven; replaces hardcoded per-type logic).
# splitperemployee -> one card per listed employee SSO
# showemployeephoto -> resolve + show each employee's photo + name
# displaystyle -> 'standard' (rows) | 'carousel' (rotating photo card)
# | 'grid' (cycling row of tiles) | 'banner'
splitperemployee = db.Column(db.Boolean, default=False)
showemployeephoto = db.Column(db.Boolean, default=False)
displaystyle = db.Column(db.String(20), default='standard')
def __repr__(self):
return f"<NotificationType {self.typename}>"
@@ -26,7 +45,14 @@ class NotificationType(db.Model):
'typename': self.typename,
'typedescription': self.typedescription,
'typecolor': self.typecolor,
'isactive': self.isactive
'isactive': self.isactive,
'expirymode': self.expirymode or 'none',
'expirydays': self.expirydays,
'expiryhour': self.expiryhour,
'expiryminute': self.expiryminute if self.expiryminute is not None else 0,
'splitperemployee': bool(self.splitperemployee),
'showemployeephoto': bool(self.showemployeephoto),
'displaystyle': self.displaystyle or 'standard'
}
@@ -52,8 +78,11 @@ class Notification(db.Model):
link = db.Column(db.String(500), nullable=True)
isactive = db.Column(db.Boolean, default=True)
isshopfloor = db.Column(db.Boolean, default=False)
employeesso = db.Column(db.String(100), nullable=True)
employeename = db.Column(db.String(100), nullable=True)
# TEXT (not VARCHAR): recognition/recertification notifications comma-join
# every employee's SSO/name into one field, which overflows 100 chars once
# ~11 people are listed.
employeesso = db.Column(db.Text, nullable=True)
employeename = db.Column(db.Text, nullable=True)
# Relationships
notificationtype = db.relationship('NotificationType', backref='notifications')
@@ -114,24 +143,25 @@ class Notification(db.Model):
def to_calendar_event(self):
"""Convert to FullCalendar event format."""
# Map Bootstrap color names to hex colors
color_map = {
# Color is data-driven: types store a hex typecolor. Only the legacy
# Bootstrap color-name aliases still need translating; hex passes through.
color_aliases = {
'success': '#04b962',
'warning': '#ff8800',
'danger': '#f5365c',
'info': '#14abef',
'primary': '#7934f3',
'secondary': '#94614f',
'recognition': '#14abef', # Blue for recognition
}
raw_color = self.notificationtype.typecolor if self.notificationtype else 'info'
# Use mapped color if it's a Bootstrap name, otherwise use as-is (hex)
color = color_map.get(raw_color, raw_color if raw_color.startswith('#') else '#14abef')
ntype = self.notificationtype
raw_color = ntype.typecolor if ntype else '#14abef'
color = color_aliases.get(raw_color, raw_color if raw_color.startswith('#') else '#14abef')
show_photo = bool(ntype and getattr(ntype, 'showemployeephoto', False))
# For recognition notifications, include employee name (or SSO as fallback) in title
# Employee-photo types prefix the card with the person's name/SSO.
title = self.title
if raw_color == 'recognition':
if show_photo:
employee_display = self.employeename or self.employeesso
if employee_display:
title = f"{employee_display}: {title}"
@@ -147,8 +177,10 @@ class Notification(db.Model):
'extendedProps': {
'notificationid': self.notificationid,
'message': self.notification,
'typename': self.notificationtype.typename if self.notificationtype else None,
'typename': ntype.typename if ntype else None,
'typecolor': raw_color,
'showemployeephoto': show_photo,
'displaystyle': (ntype.displaystyle or 'standard') if ntype else 'standard',
'linkurl': self.link,
'ticketnumber': self.ticketnumber,
'employeename': self.employeename,

38
plugins/notifications/plugin.py
View File

@@ -73,23 +73,41 @@ class NotificationsPlugin(BasePlugin):
logger.info("Notifications plugin installed")
def _ensure_notification_types(self) -> None:
"""Ensure default notification types exist."""
"""Ensure default notification types exist.
For the special shopfloor types (recognition, training,
recertification) the typecolor is a keyword the dashboard maps to a
style and the feed uses to split multi-employee cards per employee;
generic types use a hex color.
"""
# (typename, typedescription, typecolor, expirymode, expirydays,
# expiryhour, splitperemployee, showemployeephoto, displaystyle)
default_types = [
('Awareness', 'General awareness notification', '#17a2b8', 'info-circle'),
('Change', 'Planned change notification', '#ffc107', 'exchange-alt'),
('Incident', 'Incident or outage notification', '#dc3545', 'exclamation-triangle'),
('Maintenance', 'Scheduled maintenance notification', '#6c757d', 'wrench'),
('General', 'General announcement', '#28a745', 'bullhorn'),
('Awareness', 'General awareness notification', '#17a2b8', 'none', None, None, False, False, 'standard'),
('Change', 'Planned change notification', '#ffc107', 'none', None, None, False, False, 'standard'),
('Incident', 'Incident or outage notification', '#dc3545', 'none', None, None, False, False, 'standard'),
('Maintenance', 'Scheduled maintenance notification', '#6c757d', 'none', None, None, False, False, 'standard'),
('General', 'General announcement', '#28a745', 'none', None, None, False, False, 'standard'),
('Recognition', 'Employee recognition (clears at 8 AM Eastern)', '#ffc107', 'dailytime', None, 8, True, True, 'carousel'),
('Training', 'Training notice (one card per employee)', '#17a2b8', 'none', None, None, True, True, 'carousel'),
('Recertification', 'Employees due to retake a training course (shows two weeks)', '#0d6efd', 'duration', 14, None, True, True, 'grid'),
]
for typename, description, color, icon in default_types:
for (typename, typedescription, typecolor, expirymode, expirydays,
expiryhour, splitperemployee, showemployeephoto, displaystyle) in default_types:
existing = NotificationType.query.filter_by(typename=typename).first()
if not existing:
t = NotificationType(
typename=typename,
description=description,
color=color,
icon=icon
typedescription=typedescription,
typecolor=typecolor,
expirymode=expirymode,
expirydays=expirydays,
expiryhour=expiryhour,
expiryminute=0,
splitperemployee=splitperemployee,
showemployeephoto=showemployeephoto,
displaystyle=displaystyle
)
db.session.add(t)
logger.debug(f"Created notification type: {typename}")

183
plugins/printers/api/asset_routes.py
View File

@@ -5,9 +5,9 @@ import logging
from flask import Blueprint, request
from flask_jwt_extended import jwt_required
from shopdb.api import db, cache, Asset, AssetType, Vendor, Model, Communication, CommunicationType, success_response, error_response, paginated_response, ErrorCodes, get_pagination_params, paginate_query
from shopdb.api import db, cache, 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
from ..models import Printer, PrinterType, ModelSupply, PrinterDriver
from ..models.model_supply import SUPPLY_TYPES, SUPPLY_COLORS, CAPACITY_TIERS
from ..services import (
ZabbixService,
@@ -19,6 +19,8 @@ from ..services import (
logger = logging.getLogger(__name__)
from shopdb.api import require_permission, require_role
printers_asset_bp = Blueprint('printers_asset', __name__)
@@ -66,6 +68,7 @@ def get_printer_type(type_id: int):
@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()
@@ -73,7 +76,15 @@ def create_printer_type():
if not data or not data.get('printertype'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'printertype is required')
if PrinterType.query.filter_by(printertype=data['printertype']).first():
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",
@@ -83,7 +94,7 @@ def create_printer_type():
t = PrinterType(
printertype=data['printertype'],
description=data.get('description'),
icon=data.get('icon')
icon=data.get('icon'), color=data.get('color')
)
db.session.add(t)
@@ -92,6 +103,110 @@ def create_printer_type():
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 = PrinterType.query.get(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 = PrinterType.query.get(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 = PrinterDriver.query.get(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 = PrinterDriver.query.get(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
# =============================================================================
@@ -235,6 +350,50 @@ def printer_install_list():
return success_response(rows)
@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.
"""
machine = (request.args.get('machine') or '').strip()
if not machine:
return success_response({})
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 success_response({})
rel = AssetRelationship.query.filter_by(
sourceassetid=pc.assetid,
relationshiptypeid=dp_type.relationshiptypeid,
isactive=True,
).first()
if not rel:
return success_response({})
printer = db.session.query(Printer).join(Asset).filter(
Printer.assetid == rel.targetassetid,
Asset.isactive == True,
).first()
if not printer:
return success_response({})
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):
@@ -256,6 +415,15 @@ def get_printer(printer_id: int):
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)
@@ -280,6 +448,7 @@ def get_printer_by_asset(asset_id: int):
@printers_asset_bp.route('', methods=['POST'])
@jwt_required()
@require_permission('printers.create')
def create_printer():
"""
Create new printer (creates both Asset and Printer records).
@@ -379,6 +548,7 @@ def create_printer():
@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 = Printer.query.get(printer_id)
@@ -453,6 +623,7 @@ def update_printer(printer_id: int):
@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 = Printer.query.get(printer_id)
@@ -697,6 +868,7 @@ def printer_lookup():
@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.
@@ -878,6 +1050,7 @@ def list_model_supplies(modelnumberid: int):
@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 = Model.query.get(modelnumberid)
@@ -918,6 +1091,7 @@ def create_model_supply(modelnumberid: int):
@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 = ModelSupply.query.get(modelsupplyid)
@@ -962,6 +1136,7 @@ def update_model_supply(modelsupplyid: int):
@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 = ModelSupply.query.get(modelsupplyid)

2
plugins/printers/models/__init__.py
View File

@@ -1,6 +1,7 @@
"""Printers plugin models."""
from .printer import Printer, PrinterType # Asset-based models
from .printer_driver import PrinterDriver
from .model_supply import ( # data-driven model -> toner/drum/waste mapping
ModelSupply,
SUPPLY_TYPES,
@@ -11,6 +12,7 @@ from .model_supply import ( # data-driven model -> toner/drum/waste mapping
__all__ = [
'Printer',
'PrinterType',
'PrinterDriver',
'ModelSupply',
'SUPPLY_TYPES',
'SUPPLY_COLORS',

1
plugins/printers/models/printer.py
View File

@@ -15,6 +15,7 @@ class PrinterType(BaseModel):
printertype = db.Column(db.String(100), unique=True, nullable=False)
description = db.Column(db.Text)
icon = db.Column(db.String(50), comment='Icon name for UI')
color = db.Column(db.String(20), comment='CSS color for UI/map markers')
def __repr__(self):
return f"<PrinterType {self.printertype}>"

33
plugins/printers/models/printer_driver.py Normal file
View File

@@ -0,0 +1,33 @@
"""Printer driver: a named link (SMB path or HTTP URL) to driver files."""
from shopdb.api import db
class PrinterDriver(db.Model):
__tablename__ = 'printerdrivers'
driverid = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(150), nullable=False)
# SMB path (\\\\server\\share\\...) or HTTP URL to the driver package
location = db.Column(db.String(500), nullable=False)
description = db.Column(db.Text)
# Optional: attach a driver to a specific printer model
modelnumberid = db.Column(
db.Integer,
db.ForeignKey('models.modelnumberid'),
nullable=True
)
isactive = db.Column(db.Boolean, default=True, nullable=False)
model = db.relationship('Model')
def to_dict(self):
return {
'driverid': self.driverid,
'name': self.name,
'location': self.location,
'description': self.description,
'modelnumberid': self.modelnumberid,
'modelname': self.model.modelnumber if self.model else None,
'isactive': bool(self.isactive),
}

3
plugins/printers/plugin.py
View File

@@ -11,7 +11,7 @@ import click
from shopdb.plugins.base import BasePlugin, PluginMeta
from shopdb.api import db, AssetType
from .models import Printer, PrinterType, ModelSupply
from .models import Printer, PrinterType, ModelSupply, PrinterDriver
from .api import printers_asset_bp
from .services import ZabbixService
@@ -74,6 +74,7 @@ class PrintersPlugin(BasePlugin):
return [
Printer, # Asset-based
PrinterType, # printer type classification
PrinterDriver, # driver links (SMB/HTTP)
ModelSupply, # model -> toner/drum/waste part numbers
]

132
plugins/slides/PLAN-slide-manager.md Normal file
View File

@@ -0,0 +1,132 @@
# Plan: Slides plugin -> full slide manager (Flask)
Port the classic-ASP slide manager (tv-dashboard/slidemanager.asp + apislides.asp)
into the shopdb-flask `slides` plugin. Two surfaces (lobby display, shopfloor
screensaver), upload / reorder / delete, consumed by the lobby TV dashboard and
the EventSaver screensaver over HTTP.
Existing plugin is read-only (single-folder GET). This extends it to full CRUD +
per-surface + a management UI.
## Decisions (locked for this plan)
- SURFACES: fixed allowlist `lobby`, `shopfloor` (mirrors the ASP allowlist).
Not user-defined - keeps validation + paths simple.
- STORAGE: image FILES on disk, ORDER/metadata in the DB.
- Files: `instance/slides/<surface>/<filename>` (outside static; served via a
plugin route so we control content-type + path-traversal, like slide.asp).
- Metadata: one table (order + seconds + surface). Files are the source of
truth for existence; DB rows that lost their file are ignored + pruned
(same self-healing as GetOrderList in the ASP lib).
- Rationale: native, simple, no BLOBs. BLOB-in-DB is the fallback only if the
deploy host cannot give Flask a writable volume.
- MIGRATION: add the table to the CORE alembic chain (ADR-004 / 7c04 -
single authoritative chain, no per-plugin chain).
- AUTH: admin/CRUD routes JWT-protected. The FEED + image-serve routes are
PUBLIC (the screensaver + lobby kiosks have no auth - matches apislides.asp).
- NAMING: locked convention v1 - lowercase concatenated table/columns;
Python/JS vars mirror column names exactly.
## Data model (core migration)
Table `tvslides` (new, added to core chain):
| column | type | notes |
|--------------|--------------|-----------------------------------------|
| slideid | int PK | |
| surface | varchar(20) | 'lobby' | 'shopfloor' (indexed) |
| filename | varchar(255) | safe basename, unique per surface |
| sortorder | int | play order within surface |
| seconds | int | 0 = use default interval |
| uploadeddate | datetime | |
Unique (surface, filename). Model lives in `plugins/slides/models/tvslide.py`,
imported via `shopdb.api` surface only (contract purity). Registered in the core
migration chain per ADR-004.
## Backend - plugins/slides/
```
plugins/slides/
plugin.py # SlidesPlugin: get_blueprint, get_models, get_navigation_items, on_install
manifest.json # api_prefix /api/slides, provides slideshow + slidemanager
models/tvslide.py # TvSlide model
api/routes.py # blueprint slides_bp
```
Routes (prefix `/api/slides`):
- PUBLIC (no auth) - consumed by screensaver + lobby:
- `GET /feed?surface=lobby|shopfloor`
-> flat shape the EventSaver .scr + lobby expect:
`{success, surface, basepath, interval, slides:[{filename, seconds}]}`
basepath -> the image route below. (Deliberately NOT the success_response
{data:{}} wrapper, so the .scr parser needs no change.)
- `GET /img/<surface>/<filename>` -> serve the image (content-type +
path-traversal guard; send_file from instance/slides/<surface>/).
- ADMIN (JWT) - consumed by the Vue manager:
- `POST /<surface>/upload` -> request.files (native multipart), save +
create TvSlide rows at end of order. Skips non-images. Unique-renames.
- `POST /<surface>/order` -> body {order:[filename,...]} rewrite sortorder,
preserve seconds.
- `POST /<surface>/delete` -> body {files:[...]} delete file + row (multi).
- `PATCH /<surface>/<slideid>` -> {seconds} (kept server-side; UI hidden for now).
- `GET /<surface>` -> admin list (ordered, with slideid) for the UI.
Ordering/natural-sort: order comes from `sortorder`; newly-uploaded files append
in natural (numeric-aware) filename order so Slide1..Slide11 land right (port the
ASP NatKey).
## Frontend - core (no frontend plugin system)
Vue pages live in core `frontend/src/` (per project note - plugins own backend,
Vue pages are core). Add:
- `frontend/src/views/SlideManager.vue`
- Surface tabs (Lobby Display / Shopfloor Screensaver).
- Upload (file input -> POST /upload), thumbnail grid/table.
- Drag reorder via vuedraggable -> POST /order.
- Checkbox multi-select + Delete Selected -> POST /delete.
- Inherits AppLayout (sidebar + topbar + theme) automatically -> matches the
site with ZERO styling work (the whole reason to move off ASP).
- Router entry in `frontend/src/router` (e.g. /slides), guarded (admin).
- Nav: plugin `get_navigation_items()` returns the sidebar entry; the frontend
nav consumes plugin nav hooks (as knowledgebase/notifications do).
- `frontend/src/api/slidesApi.js` wrapper for the admin calls.
## Consumers
- Lobby: `TVDashboard.vue` already hits /api/slides; repoint at `/feed?surface=lobby`.
- Screensaver: change `EventSaver.ini` `url=` to
`https://<flask-host>/api/slides/feed?surface=shopfloor`, rehash the ini +
bump the manifest DetectionValue. The .scr HTTP mode is unchanged because
/feed returns the flat shape it already parses.
## Tests (pytest, per plugin conventions)
- test_plugins/test_slides.py: upload (fake file) creates rows + file; feed
returns ordered flat shape; order persists + preserves seconds; delete removes
file+row; surface allowlist rejects junk; path-traversal guard on /img;
natural sort on append; feed is public / admin routes require JWT.
- Guard test already enforces contract-only imports.
## Phasing
1. Model + core migration + manifest bump.
2. Backend routes (feed + img public; upload/order/delete/list admin) + tests.
3. Vue SlideManager.vue + router + nav + slidesApi.js.
4. Repoint TVDashboard.vue to /feed?surface=lobby.
5. Cut over screensaver: EventSaver.ini url -> Flask /feed, rehash + manifest.
6. Retire classic ASP tv-dashboard slide pages once Flask is live for this site.
## Open decisions / gates
- DEPLOY GATE: shopdb-flask must be deployed + reachable by the screensaver PCs
and lobby, with a writable `instance/slides` volume. Live shopdb is still the
classic ASP box; Flask prod target is docker, not yet deployed here. This plan
is buildable now but not cutover-able until Flask is live.
- Image serving: via a plugin route (send_file, guarded) vs Flask static. Route
chosen for the traversal guard + content-type control (parity with slide.asp).
- Seconds field: model + PATCH kept; UI control hidden for now (matches the ASP
decision to hide per-slide delay).
- BLOB fallback: only if the host denies a writable slides volume.

249
plugins/slides/api/routes.py
View File

@@ -1,71 +1,210 @@
"""Slides API for the TV dashboard slideshow."""
"""Slides API: lobby-display + shopfloor-screensaver slideshows.
Image files live on disk at instance/slides/<surface>/; TvSlide rows hold play
order + per-slide seconds. Feed + image routes are PUBLIC (kiosks/screensaver
have no auth); management routes are admin-only.
"""
import os
from flask import Blueprint, current_app
import re
from shopdb.api import success_response, error_response, ErrorCodes
from flask import Blueprint, request, current_app, jsonify, send_from_directory
from flask_jwt_extended import jwt_required
from werkzeug.utils import secure_filename
from shopdb.api import db, success_response, error_response, ErrorCodes, require_role
from ..models import TvSlide
slides_bp = Blueprint('slides', __name__)
# Valid image extensions
SURFACES = frozenset({'lobby', 'shopfloor'})
VALID_EXTENSIONS = {'.jpg', '.jpeg', '.png', '.gif', '.bmp', '.webp'}
DEFAULT_INTERVAL = 10 # seconds per slide when a slide's own seconds is 0
@slides_bp.route('', methods=['GET'])
def get_slides():
"""Get list of slides for the TV dashboard.
def _surface_dir(surface):
return os.path.join(current_app.instance_path, 'slides', surface)
Returns image files from the static/slides directory.
"""
# Look for slides in static folder
static_folder = current_app.static_folder
if not static_folder:
static_folder = os.path.join(current_app.root_path, 'static')
slides_folder = os.path.join(static_folder, 'slides')
def _natkey(name):
"""Natural sort key so Slide1..Slide11 order numerically, not lexically."""
return [int(t) if t.isdigit() else t.lower() for t in re.split(r'(\d+)', name)]
# Also check frontend public folder
frontend_slides = os.path.join(
os.path.dirname(os.path.dirname(os.path.dirname(current_app.root_path))),
'frontend', 'public', 'slides'
)
# Try multiple possible locations
possible_paths = [
slides_folder,
frontend_slides,
'/home/camp/projects/shopdb-flask/shopdb/static/slides',
'/home/camp/projects/shopdb-flask/frontend/public/slides',
def _valid_surface(surface):
return surface in SURFACES
def _is_image(filename):
return os.path.splitext(filename)[1].lower() in VALID_EXTENSIONS
# =============================================================================
# PUBLIC - consumed by the lobby dashboard + the screensaver
# =============================================================================
@slides_bp.route('/feed', methods=['GET'])
def feed():
"""Flat playlist for a surface. NOT wrapped in the success_response envelope
so the screensaver's existing parser needs no change."""
surface = request.args.get('surface', 'lobby')
if not _valid_surface(surface):
surface = 'lobby'
directory = _surface_dir(surface)
rows = (TvSlide.query.filter_by(surface=surface)
.order_by(TvSlide.sortorder, TvSlide.slideid).all())
slides = [
{'filename': r.filename, 'seconds': r.seconds or DEFAULT_INTERVAL}
for r in rows
if os.path.isfile(os.path.join(directory, r.filename))
]
slides_path = None
for path in possible_paths:
if os.path.isdir(path):
slides_path = path
break
if not slides_path:
return success_response({
'slides': [],
'basepath': '/static/slides/',
'message': 'Slides folder not found'
})
# Get list of image files
slides = []
try:
for filename in sorted(os.listdir(slides_path)):
ext = os.path.splitext(filename)[1].lower()
if ext in VALID_EXTENSIONS:
slides.append({'filename': filename})
except Exception as e:
return error_response(
ErrorCodes.INTERNAL_ERROR,
f'Error reading slides: {str(e)}',
http_code=500
)
return success_response({
return jsonify({
'success': True,
'surface': surface,
'basepath': f'/api/slides/img/{surface}/',
'interval': DEFAULT_INTERVAL,
'slides': slides,
'basepath': '/static/slides/'
})
@slides_bp.route('/img/<surface>/<path:filename>', methods=['GET'])
def serve_image(surface, filename):
"""Serve a slide image with a path-traversal guard."""
if not _valid_surface(surface):
return error_response(ErrorCodes.NOT_FOUND, 'Unknown surface', http_code=404)
safe = os.path.basename(filename)
if safe != filename or not safe:
return error_response(ErrorCodes.NOT_FOUND, 'Invalid filename', http_code=404)
directory = _surface_dir(surface)
if not os.path.isfile(os.path.join(directory, safe)):
return error_response(ErrorCodes.NOT_FOUND, 'Slide not found', http_code=404)
return send_from_directory(directory, safe)
# =============================================================================
# ADMIN - the Vue slide manager
# =============================================================================
@slides_bp.route('/<surface>', methods=['GET'])
@jwt_required()
@require_role('admin')
def list_slides(surface):
if not _valid_surface(surface):
return error_response(ErrorCodes.VALIDATION_ERROR, 'Unknown surface')
directory = _surface_dir(surface)
rows = (TvSlide.query.filter_by(surface=surface)
.order_by(TvSlide.sortorder, TvSlide.slideid).all())
out = []
for r in rows:
if os.path.isfile(os.path.join(directory, r.filename)):
item = r.to_dict()
item['url'] = f'/api/slides/img/{surface}/{r.filename}'
out.append(item)
return success_response(out)
@slides_bp.route('/<surface>/upload', methods=['POST'])
@jwt_required()
@require_role('admin')
def upload_slides(surface):
if not _valid_surface(surface):
return error_response(ErrorCodes.VALIDATION_ERROR, 'Unknown surface')
files = request.files.getlist('files')
if not files and 'file' in request.files:
files = [request.files['file']]
if not files:
return error_response(ErrorCodes.VALIDATION_ERROR, 'No files uploaded')
directory = _surface_dir(surface)
os.makedirs(directory, exist_ok=True)
# Append after the current max order; upload in natural filename order.
maxorder = db.session.query(db.func.max(TvSlide.sortorder)).filter_by(surface=surface).scalar()
order = (maxorder or 0)
existing = {r.filename for r in TvSlide.query.filter_by(surface=surface).all()}
added = []
for f in sorted(files, key=lambda x: _natkey(x.filename or '')):
if not f or not f.filename or not _is_image(f.filename):
continue
base = secure_filename(os.path.basename(f.filename))
if not base:
continue
# Unique-rename if the name is taken (on disk or in db).
stem, ext = os.path.splitext(base)
name = base
n = 1
while name in existing or os.path.exists(os.path.join(directory, name)):
name = f"{stem}_{n}{ext}"
n += 1
f.save(os.path.join(directory, name))
existing.add(name)
order += 1
db.session.add(TvSlide(surface=surface, filename=name, sortorder=order, seconds=0))
added.append(name)
db.session.commit()
return success_response({'added': added}, message=f'{len(added)} slide(s) uploaded')
@slides_bp.route('/<surface>/order', methods=['POST'])
@jwt_required()
@require_role('admin')
def reorder_slides(surface):
if not _valid_surface(surface):
return error_response(ErrorCodes.VALIDATION_ERROR, 'Unknown surface')
data = request.get_json() or {}
order = data.get('order') or []
for i, filename in enumerate(order):
row = TvSlide.query.filter_by(surface=surface, filename=filename).first()
if row:
row.sortorder = i
db.session.commit()
return success_response(message='Order saved')
@slides_bp.route('/<surface>/delete', methods=['POST'])
@jwt_required()
@require_role('admin')
def delete_slides(surface):
if not _valid_surface(surface):
return error_response(ErrorCodes.VALIDATION_ERROR, 'Unknown surface')
data = request.get_json() or {}
files = data.get('files') or []
directory = _surface_dir(surface)
removed = 0
for filename in files:
safe = os.path.basename(filename)
row = TvSlide.query.filter_by(surface=surface, filename=safe).first()
fpath = os.path.join(directory, safe)
if os.path.isfile(fpath):
try:
os.remove(fpath)
except OSError:
pass
if row:
db.session.delete(row)
removed += 1
db.session.commit()
return success_response(message=f'{removed} slide(s) deleted')
@slides_bp.route('/<surface>/<int:slideid>', methods=['PATCH'])
@jwt_required()
@require_role('admin')
def update_slide(surface, slideid):
if not _valid_surface(surface):
return error_response(ErrorCodes.VALIDATION_ERROR, 'Unknown surface')
row = TvSlide.query.get(slideid)
if not row or row.surface != surface:
return error_response(ErrorCodes.NOT_FOUND, 'Slide not found', http_code=404)
data = request.get_json() or {}
if 'seconds' in data:
try:
row.seconds = max(0, int(data['seconds']))
except (TypeError, ValueError):
return error_response(ErrorCodes.VALIDATION_ERROR, 'seconds must be an integer')
db.session.commit()
return success_response(row.to_dict(), message='Slide updated')

8
plugins/slides/manifest.json
View File

@@ -1,12 +1,12 @@
{
"name": "slides",
"version": "1.0.0",
"description": "TV dashboard slideshow images served from a static folder",
"version": "2.0.0",
"description": "Slide manager for the lobby display and shopfloor screensaver (upload/reorder/delete per surface)",
"author": "ShopDB Team",
"dependencies": [],
"core_version": ">=0.1.0,<1.0.0",
"core_version": ">=0.2.0,<1.0.0",
"api_prefix": "/api/slides",
"provides": {
"features": ["slideshow"]
"features": ["slideshow", "slidemanager"]
}
}

5
plugins/slides/models/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
"""Slides plugin models."""
from .tvslide import TvSlide
__all__ = ['TvSlide']

35
plugins/slides/models/tvslide.py Normal file
View File

@@ -0,0 +1,35 @@
"""Slide model for the TV dashboard / screensaver slideshows."""
from shopdb.api import db
class TvSlide(db.Model):
"""One slide image in a surface's playlist.
Image files live on disk at instance/slides/<surface>/<filename>; this row
holds the play order + per-slide duration. Files are the source of truth for
existence - rows whose file is gone are ignored and pruned.
"""
__tablename__ = 'tvslides'
slideid = db.Column(db.Integer, primary_key=True)
surface = db.Column(db.String(20), nullable=False, index=True)
filename = db.Column(db.String(255), nullable=False)
sortorder = db.Column(db.Integer, default=0, nullable=False)
seconds = db.Column(db.Integer, default=0, nullable=False) # 0 = use default interval
uploadeddate = db.Column(db.DateTime, default=db.func.now())
# No DB-level unique on (surface, filename): utf8mb4 pushes that index past
# MySQL 5.6's 767-byte limit. Upload unique-renames, so dupes can't occur.
__table_args__ = (
db.Index('idx_tvslide_surface', 'surface'),
)
def to_dict(self):
return {
'slideid': self.slideid,
'surface': self.surface,
'filename': self.filename,
'sortorder': self.sortorder,
'seconds': self.seconds,
}

16
plugins/slides/plugin.py
View File

@@ -15,6 +15,7 @@ from flask import Flask, Blueprint
from shopdb.plugins.base import BasePlugin, PluginMeta
from .api import slides_bp
from .models import TvSlide
logger = logging.getLogger(__name__)
@@ -54,8 +55,19 @@ class SlidesPlugin(BasePlugin):
return slides_bp
def get_models(self) -> List[Type]:
"""No models - slides are read from the filesystem."""
return []
"""Slide playlist metadata (image files live on disk)."""
return [TvSlide]
def get_navigation_items(self) -> List[Dict]:
"""Sidebar entry for the slide manager (admin)."""
return [
{
'name': 'Slides',
'icon': 'image',
'route': '/settings/slides',
'position': 7,
},
]
def init_app(self, app: Flask, db_instance) -> None:
"""Initialize plugin with Flask app."""

865
plugins/usb/api/routes.py
View File

@@ -1,58 +1,242 @@
"""USB plugin API endpoints."""
"""USB plugin API endpoints.
from flask import Blueprint, request
from flask_jwt_extended import jwt_required, get_jwt_identity
Check-in/out state lives in a separate MySQL database (cmmc_usb), reached with
raw parameterized pymysql via cmmc_usb_connection(), mirroring the read-only
employee-directory helper pattern. The SQLAlchemy USBDevice/USBCheckout/
USBDeviceType models in ../models are NOT used by these routes anymore; they
are left in place for any legacy importers.
cmmc_usb schema used here:
devices(device_id PK, device_desc, device_owner, status, locker_location)
status is one of 'checked-in' | 'checked-out' | 'retired'
users(badge_number PK, first_name, last_name)
checkinoutlog(log_id PK auto, badge_number, device_id, action, timestamp,
scanned_viruses, locker_location, sanitized)
action is one of 'check-in' | 'check-out'; timestamp is a
'YYYY-MM-DD HH:MM:SS' string; scanned_viruses/sanitized are '0'|'1'|null.
"""
import logging
import re
from datetime import datetime
from shopdb.api import db, AuditLog, success_response, error_response, paginated_response, ErrorCodes, get_pagination_params, paginate_query
from flask import Blueprint, request
from flask_jwt_extended import jwt_required
from ..models import USBDevice, USBDeviceType, USBCheckout
from shopdb.api import (
cmmc_usb_connection,
employee_connection,
success_response,
error_response,
paginated_response,
ErrorCodes,
get_pagination_params,
require_permission,
)
logger = logging.getLogger(__name__)
usb_bp = Blueprint('usb', __name__)
# public status strings on the devices row
_STATUS_CHECKED_IN = 'checked-in'
_STATUS_CHECKED_OUT = 'checked-out'
_STATUS_RETIRED = 'retired'
# map the ?status= query values to the stored devices.status strings
_STATUS_FILTER = {
'available': _STATUS_CHECKED_IN,
'checkedout': _STATUS_CHECKED_OUT,
'retired': _STATUS_RETIRED,
}
# columns pulled for a devices row
_DEVICE_COLS = 'device_id, device_desc, device_owner, status, locker_location'
# columns pulled for a checkinoutlog row
_LOG_COLS = (
'log_id, badge_number, device_id, action, timestamp, '
'scanned_viruses, locker_location, sanitized'
)
# a badge like 0123456BZ carries a PayNo (the digits) wrapped in 0...BZ
_PAYNO_BADGE = re.compile(r'^0(\d+)BZ$', re.IGNORECASE)
# =============================================================================
# USB Device Types
# Module helpers
# =============================================================================
@usb_bp.route('/types', methods=['GET'])
@jwt_required(optional=True)
def list_device_types():
"""List all USB device types."""
types = USBDeviceType.query.filter_by(isactive=True).order_by(USBDeviceType.typename).all()
return success_response([{
'usbdevicetypeid': t.usbdevicetypeid,
'typename': t.typename,
'description': t.description,
'icon': t.icon
} for t in types])
def _now():
"""Current local time as a 'YYYY-MM-DD HH:MM:SS' string for the log."""
return datetime.now().strftime('%Y-%m-%d %H:%M:%S')
@usb_bp.route('/types', methods=['POST'])
@jwt_required()
def create_device_type():
"""Create a new USB device type."""
data = request.get_json() or {}
def _flag(value):
"""Coerce a truthy/falsy input to a '1'/'0' string, or None when unset."""
if value is None:
return None
if isinstance(value, bool):
return '1' if value else '0'
text = str(value).strip().lower()
if text in ('1', 'true', 'yes', 'y'):
return '1'
if text in ('0', 'false', 'no', 'n', ''):
return '0'
return None
if not data.get('typename'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'typename is required')
if USBDeviceType.query.filter_by(typename=data['typename']).first():
return error_response(ErrorCodes.CONFLICT, 'Type name already exists', http_code=409)
def _lookup_employee_name(badge):
"""Look up "First Last" in the HR directory for a badge. Empty on any miss.
device_type = USBDeviceType(
typename=data['typename'],
description=data.get('description'),
icon=data.get('icon', 'usb')
A numeric badge is an SSO. A 0<digits>BZ badge carries a PayNo (the middle
digits). Any other shape has no HR mapping so we return "".
"""
badge = (badge or '').strip()
if not badge:
return ''
try:
conn = employee_connection()
except Exception:
logger.exception('HR directory connection failed for badge %s', badge)
return ''
try:
with conn.cursor() as ecur:
if badge.isdigit():
ecur.execute(
'SELECT First_Name, Last_Name FROM employees WHERE SSO = %s',
(badge,),
)
else:
match = _PAYNO_BADGE.match(badge)
if not match:
return ''
ecur.execute(
'SELECT First_Name, Last_Name FROM employees WHERE PayNo = %s',
(match.group(1),),
)
emp = ecur.fetchone()
except Exception:
logger.exception('HR directory lookup failed for badge %s', badge)
return ''
finally:
conn.close()
if not emp:
return ''
first = (emp.get('First_Name') or '').strip()
last = (emp.get('Last_Name') or '').strip()
return (first + ' ' + last).strip()
def _resolve_badge_name(cur, badge):
"""Resolve a badge to "First Last". Try cmmc_usb users, then HR directory.
cur is an open cmmc_usb cursor. Returns "" when the badge cannot be
resolved anywhere.
"""
badge = (badge or '').strip() if badge is not None else ''
if not badge:
return ''
cur.execute(
'SELECT first_name, last_name FROM users WHERE badge_number = %s',
(badge,),
)
row = cur.fetchone()
if row:
first = (row.get('first_name') or '').strip()
last = (row.get('last_name') or '').strip()
name = (first + ' ' + last).strip()
if name:
return name
return _lookup_employee_name(badge)
def _ensure_user(cur, badge, name):
"""Insert a badge into cmmc_usb users if absent. Splits name to first/last."""
badge = (badge or '').strip() if badge is not None else ''
if not badge:
return
cur.execute('SELECT badge_number FROM users WHERE badge_number = %s', (badge,))
if cur.fetchone():
return
first, last = '', ''
if name:
parts = name.split()
if parts:
first = parts[0]
last = ' '.join(parts[1:])
cur.execute(
'INSERT INTO users (badge_number, first_name, last_name) VALUES (%s, %s, %s)',
(badge, first, last),
)
db.session.add(device_type)
db.session.commit()
return success_response({
'usbdevicetypeid': device_type.usbdevicetypeid,
'typename': device_type.typename
}, message='Device type created', http_code=201)
def _fetch_device(cur, device_id):
"""Return the devices row for device_id, or None."""
cur.execute(
'SELECT ' + _DEVICE_COLS + ' FROM devices WHERE device_id = %s',
(device_id,),
)
return cur.fetchone()
def _device_to_dict(cur, row):
"""Build the device response dict, resolving owner + current holder names."""
device = {
'device_id': row.get('device_id'),
'device_desc': row.get('device_desc'),
'device_owner': row.get('device_owner'),
'owner_name': _resolve_badge_name(cur, row.get('device_owner')),
'status': row.get('status'),
'locker_location': row.get('locker_location'),
'current_holder': None,
'current_holder_name': None,
'checkout_time': None,
}
if row.get('status') == _STATUS_CHECKED_OUT:
cur.execute(
"SELECT badge_number, timestamp FROM checkinoutlog "
"WHERE device_id = %s AND action = 'check-out' "
"ORDER BY timestamp DESC, log_id DESC LIMIT 1",
(row.get('device_id'),),
)
log = cur.fetchone()
if log:
device['current_holder'] = log.get('badge_number')
device['current_holder_name'] = _resolve_badge_name(
cur, log.get('badge_number')
)
device['checkout_time'] = log.get('timestamp')
return device
def _log_to_dict(cur, row):
"""Build a checkinoutlog response dict with the resolved badge name."""
result = {
'log_id': row.get('log_id'),
'badge_number': row.get('badge_number'),
'badge_name': _resolve_badge_name(cur, row.get('badge_number')),
'device_id': row.get('device_id'),
'action': row.get('action'),
'timestamp': row.get('timestamp'),
'scanned_viruses': row.get('scanned_viruses'),
'locker_location': row.get('locker_location'),
'sanitized': row.get('sanitized'),
}
# /checkouts joins in the current device status; pass it through when present
if 'device_status' in row:
result['device_status'] = row.get('device_status')
return result
# =============================================================================
@@ -62,306 +246,356 @@ def create_device_type():
@usb_bp.route('', methods=['GET'])
@jwt_required(optional=True)
def list_usb_devices():
"""
List all USB devices with checkout status.
"""List USB devices with checkout status.
Query parameters:
- page, per_page: Pagination
- search: Search by serial number, label, or asset number
- available: Filter to only available (not checked out) devices
- typeid: Filter by device type ID
- page, per_page: pagination
- status: available | checkedout | retired
- search: match on device_id or device_desc
"""
page, per_page = get_pagination_params(request)
query = USBDevice.query.filter_by(isactive=True)
where = []
params = []
# Filter by type
if type_id := request.args.get('typeid'):
query = query.filter_by(usbdevicetypeid=int(type_id))
status_arg = request.args.get('status', '').strip().lower()
if status_arg in _STATUS_FILTER:
where.append('status = %s')
params.append(_STATUS_FILTER[status_arg])
# Filter by checkout status
if request.args.get('available', '').lower() == 'true':
query = query.filter_by(ischeckedout=False)
elif request.args.get('checkedout', '').lower() == 'true':
query = query.filter_by(ischeckedout=True)
search = request.args.get('search', '').strip()
if search:
where.append('(device_id LIKE %s OR device_desc LIKE %s)')
like = '%' + search + '%'
params.extend([like, like])
# Search filter
if search := request.args.get('search'):
query = query.filter(
db.or_(
USBDevice.serialnumber.ilike(f'%{search}%'),
USBDevice.label.ilike(f'%{search}%'),
USBDevice.assetnumber.ilike(f'%{search}%'),
USBDevice.manufacturer.ilike(f'%{search}%')
where_sql = (' WHERE ' + ' AND '.join(where)) if where else ''
conn = cmmc_usb_connection()
try:
with conn.cursor() as cur:
cur.execute('SELECT COUNT(*) AS total FROM devices' + where_sql, params)
total = cur.fetchone()['total']
cur.execute(
'SELECT ' + _DEVICE_COLS + ' FROM devices' + where_sql
+ ' ORDER BY device_id LIMIT %s OFFSET %s',
params + [per_page, (page - 1) * per_page],
)
)
query = query.order_by(USBDevice.label, USBDevice.serialnumber)
items, total = paginate_query(query, page, per_page)
data = [device.to_dict() for device in items]
rows = cur.fetchall()
data = [_device_to_dict(cur, row) for row in rows]
finally:
conn.close()
return paginated_response(data, page, per_page, total)
@usb_bp.route('', methods=['POST'])
@jwt_required()
def create_usb_device():
"""Create a new USB device."""
data = request.get_json() or {}
if not data.get('serialnumber'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'serialnumber is required')
if USBDevice.query.filter_by(serialnumber=data['serialnumber']).first():
return error_response(ErrorCodes.CONFLICT, 'Serial number already exists', http_code=409)
device = USBDevice(
serialnumber=data['serialnumber'],
label=data.get('label'),
assetnumber=data.get('assetnumber'),
usbdevicetypeid=data.get('usbdevicetypeid'),
capacitygb=data.get('capacitygb'),
vendorid=data.get('vendorid'),
productid=data.get('productid'),
manufacturer=data.get('manufacturer'),
productname=data.get('productname'),
storagelocation=data.get('storagelocation'),
pin=data.get('pin'),
notes=data.get('notes'),
ischeckedout=False
)
db.session.add(device)
db.session.flush()
AuditLog.log('created', 'USBDevice', entityid=device.usbdeviceid,
entityname=device.label or device.serialnumber)
db.session.commit()
return success_response(device.to_dict(), message='Device created', http_code=201)
@usb_bp.route('/<int:device_id>', methods=['GET'])
@usb_bp.route('/<device_id>', methods=['GET'])
@jwt_required(optional=True)
def get_usb_device(device_id: int):
"""Get a single USB device with checkout history."""
device = USBDevice.query.filter_by(usbdeviceid=device_id, isactive=True).first()
def get_usb_device(device_id):
"""Get a single device plus its last 20 check-in/out log rows."""
conn = cmmc_usb_connection()
try:
with conn.cursor() as cur:
row = _fetch_device(cur, device_id)
if not row:
return error_response(
ErrorCodes.NOT_FOUND,
f'USB device {device_id} not found',
http_code=404,
)
if not device:
return error_response(
ErrorCodes.NOT_FOUND,
f'USB device with ID {device_id} not found',
http_code=404
)
result = _device_to_dict(cur, row)
# Get recent checkout history
checkouts = USBCheckout.query.filter_by(
usbdeviceid=device_id
).order_by(USBCheckout.checkouttime.desc()).limit(20).all()
result = device.to_dict()
result['checkouthistory'] = [c.to_dict() for c in checkouts]
cur.execute(
'SELECT ' + _LOG_COLS + ' FROM checkinoutlog WHERE device_id = %s '
'ORDER BY timestamp DESC, log_id DESC LIMIT 20',
(device_id,),
)
logs = cur.fetchall()
result['checkinoutlog'] = [_log_to_dict(cur, log) for log in logs]
finally:
conn.close()
return success_response(result)
@usb_bp.route('/<int:device_id>', methods=['PUT'])
@usb_bp.route('', methods=['POST'])
@jwt_required()
def update_usb_device(device_id: int):
"""Update a USB device."""
device = USBDevice.query.filter_by(usbdeviceid=device_id, isactive=True).first()
if not device:
return error_response(
ErrorCodes.NOT_FOUND,
f'USB device with ID {device_id} not found',
http_code=404
)
@require_permission('usb.create')
def create_usb_device():
"""Add a device (starts checked-in)."""
data = request.get_json() or {}
# Track changes for audit log
changes = {}
for field in ['label', 'assetnumber', 'usbdevicetypeid', 'capacitygb',
'vendorid', 'productid', 'manufacturer', 'productname',
'storagelocation', 'pin', 'notes']:
if field in data:
old_val = getattr(device, field)
new_val = data[field]
if old_val != new_val:
changes[field] = {'old': old_val, 'new': new_val}
setattr(device, field, data[field])
device_id = (data.get('device_id') or '').strip()
if not device_id:
return error_response(ErrorCodes.VALIDATION_ERROR, 'device_id is required')
if changes:
AuditLog.log('updated', 'USBDevice', entityid=device.usbdeviceid,
entityname=device.label or device.serialnumber, changes=changes)
conn = cmmc_usb_connection()
try:
with conn.cursor() as cur:
cur.execute(
'SELECT device_id FROM devices WHERE device_id = %s', (device_id,)
)
if cur.fetchone():
return error_response(
ErrorCodes.CONFLICT, 'device_id already exists', http_code=409
)
device.modifieddate = datetime.utcnow()
db.session.commit()
cur.execute(
'INSERT INTO devices '
'(device_id, device_desc, device_owner, status, locker_location) '
'VALUES (%s, %s, %s, %s, %s)',
(
device_id,
data.get('device_desc'),
data.get('device_owner'),
_STATUS_CHECKED_IN,
data.get('locker_location'),
),
)
conn.commit()
return success_response(device.to_dict(), message='Device updated')
with conn.cursor() as cur:
row = _fetch_device(cur, device_id)
result = _device_to_dict(cur, row)
finally:
conn.close()
return success_response(result, message='Device created', http_code=201)
@usb_bp.route('/<int:device_id>', methods=['DELETE'])
@usb_bp.route('/<device_id>', methods=['PUT'])
@jwt_required()
def delete_usb_device(device_id: int):
"""Soft delete a USB device."""
device = USBDevice.query.filter_by(usbdeviceid=device_id, isactive=True).first()
if not device:
return error_response(
ErrorCodes.NOT_FOUND,
f'USB device with ID {device_id} not found',
http_code=404
)
if device.ischeckedout:
return error_response(
ErrorCodes.VALIDATION_ERROR,
'Cannot delete a device that is currently checked out',
http_code=400
)
device.isactive = False
device.modifieddate = datetime.utcnow()
AuditLog.log('deleted', 'USBDevice', entityid=device.usbdeviceid,
entityname=device.label or device.serialnumber)
db.session.commit()
return success_response(None, message='Device deleted')
# =============================================================================
# Checkout/Checkin Operations
# =============================================================================
@usb_bp.route('/<int:device_id>/checkout', methods=['POST'])
@jwt_required()
def checkout_device(device_id: int):
"""Check out a USB device."""
device = USBDevice.query.filter_by(usbdeviceid=device_id, isactive=True).first()
if not device:
return error_response(
ErrorCodes.NOT_FOUND,
f'USB device with ID {device_id} not found',
http_code=404
)
if device.ischeckedout:
return error_response(
ErrorCodes.CONFLICT,
f'Device is already checked out to {device.currentusername or device.currentuserid}',
http_code=409
)
@require_permission('usb.edit')
def update_usb_device(device_id):
"""Edit device_desc / device_owner / locker_location / status."""
data = request.get_json() or {}
if not data.get('sso'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'sso is required')
conn = cmmc_usb_connection()
try:
with conn.cursor() as cur:
if not _fetch_device(cur, device_id):
return error_response(
ErrorCodes.NOT_FOUND,
f'USB device {device_id} not found',
http_code=404,
)
# Create checkout record
checkout = USBCheckout(
usbdeviceid=device_id,
machineid=0, # Legacy field, set to 0 for new checkouts
sso=data['sso'],
checkoutname=data.get('checkoutname'),
checkouttime=datetime.utcnow(),
checkoutreason=data.get('checkoutreason'),
waswiped=False
)
fields = []
params = []
for col in ('device_desc', 'device_owner', 'locker_location', 'status'):
if col in data:
fields.append(col + ' = %s')
params.append(data[col])
# Update device status
device.ischeckedout = True
device.currentuserid = data['sso']
device.currentusername = data.get('checkoutname')
device.currentcheckoutdate = datetime.utcnow()
device.modifieddate = datetime.utcnow()
if fields:
params.append(device_id)
cur.execute(
'UPDATE devices SET ' + ', '.join(fields) + ' WHERE device_id = %s',
params,
)
conn.commit()
db.session.add(checkout)
with conn.cursor() as cur:
row = _fetch_device(cur, device_id)
result = _device_to_dict(cur, row)
finally:
conn.close()
AuditLog.log('checked_out', 'USBDevice', entityid=device.usbdeviceid,
entityname=device.label or device.serialnumber,
changes={'checked_out_to': data['sso'], 'reason': data.get('checkoutreason')})
db.session.commit()
return success_response(checkout.to_dict(), message='Device checked out', http_code=201)
return success_response(result, message='Device updated')
@usb_bp.route('/<int:device_id>/checkin', methods=['POST'])
@usb_bp.route('/<device_id>/retire', methods=['POST'])
@jwt_required()
def checkin_device(device_id: int):
"""Check in a USB device."""
device = USBDevice.query.filter_by(usbdeviceid=device_id, isactive=True).first()
@require_permission('usb.edit')
def retire_usb_device(device_id):
"""Retire a device (status -> retired)."""
conn = cmmc_usb_connection()
try:
with conn.cursor() as cur:
if not _fetch_device(cur, device_id):
return error_response(
ErrorCodes.NOT_FOUND,
f'USB device {device_id} not found',
http_code=404,
)
cur.execute(
'UPDATE devices SET status = %s WHERE device_id = %s',
(_STATUS_RETIRED, device_id),
)
conn.commit()
if not device:
return error_response(
ErrorCodes.NOT_FOUND,
f'USB device with ID {device_id} not found',
http_code=404
)
with conn.cursor() as cur:
row = _fetch_device(cur, device_id)
result = _device_to_dict(cur, row)
finally:
conn.close()
if not device.ischeckedout:
return error_response(
ErrorCodes.VALIDATION_ERROR,
'Device is not currently checked out',
http_code=400
)
return success_response(result, message='Device retired')
# Find active checkout
active_checkout = USBCheckout.query.filter_by(
usbdeviceid=device_id,
checkintime=None
).first()
# =============================================================================
# Check-in / check-out operations
# =============================================================================
@usb_bp.route('/<device_id>/checkout', methods=['POST'])
@jwt_required()
@require_permission('usb.create')
def checkout_device(device_id):
"""Check a device out to a badge."""
data = request.get_json() or {}
if active_checkout:
active_checkout.checkintime = datetime.utcnow()
active_checkout.checkinnotes = data.get('checkinnotes', active_checkout.checkinnotes)
active_checkout.waswiped = data.get('waswiped', False)
badge = (data.get('badge') or '').strip()
if not badge:
return error_response(ErrorCodes.VALIDATION_ERROR, 'badge is required')
# Update device status
previous_user = device.currentuserid
device.ischeckedout = False
device.currentuserid = None
device.currentusername = None
device.currentcheckoutdate = None
device.modifieddate = datetime.utcnow()
conn = cmmc_usb_connection()
try:
with conn.cursor() as cur:
device = _fetch_device(cur, device_id)
if not device:
return error_response(
ErrorCodes.NOT_FOUND,
f'USB device {device_id} not found',
http_code=404,
)
if device.get('status') == _STATUS_CHECKED_OUT:
return error_response(
ErrorCodes.CONFLICT,
'Device is already checked out',
http_code=409,
)
AuditLog.log('checked_in', 'USBDevice', entityid=device.usbdeviceid,
entityname=device.label or device.serialnumber,
changes={'returned_by': previous_user, 'wiped': data.get('waswiped', False)})
name = _resolve_badge_name(cur, badge)
_ensure_user(cur, badge, name)
db.session.commit()
locker = data.get('locker_location')
log_locker = locker if locker is not None else device.get('locker_location')
return success_response(
active_checkout.to_dict() if active_checkout else None,
message='Device checked in'
)
cur.execute(
'INSERT INTO checkinoutlog '
'(badge_number, device_id, action, timestamp, locker_location) '
"VALUES (%s, %s, 'check-out', %s, %s)",
(badge, device_id, _now(), log_locker),
)
if locker is not None:
cur.execute(
'UPDATE devices SET status = %s, locker_location = %s '
'WHERE device_id = %s',
(_STATUS_CHECKED_OUT, locker, device_id),
)
else:
cur.execute(
'UPDATE devices SET status = %s WHERE device_id = %s',
(_STATUS_CHECKED_OUT, device_id),
)
conn.commit()
with conn.cursor() as cur:
row = _fetch_device(cur, device_id)
result = _device_to_dict(cur, row)
finally:
conn.close()
return success_response(result, message='Device checked out', http_code=201)
@usb_bp.route('/<device_id>/checkin', methods=['POST'])
@jwt_required()
@require_permission('usb.create')
def checkin_device(device_id):
"""Check a device back in from a badge."""
data = request.get_json() or {}
badge = (data.get('badge') or '').strip()
if not badge:
return error_response(ErrorCodes.VALIDATION_ERROR, 'badge is required')
conn = cmmc_usb_connection()
try:
with conn.cursor() as cur:
device = _fetch_device(cur, device_id)
if not device:
return error_response(
ErrorCodes.NOT_FOUND,
f'USB device {device_id} not found',
http_code=404,
)
if device.get('status') != _STATUS_CHECKED_OUT:
return error_response(
ErrorCodes.VALIDATION_ERROR,
'Device is not currently checked out',
http_code=400,
)
name = _resolve_badge_name(cur, badge)
_ensure_user(cur, badge, name)
locker = data.get('locker_location')
log_locker = locker if locker is not None else device.get('locker_location')
sanitized = _flag(data.get('sanitized'))
scanned_viruses = _flag(data.get('scanned_viruses'))
cur.execute(
'INSERT INTO checkinoutlog '
'(badge_number, device_id, action, timestamp, scanned_viruses, '
'locker_location, sanitized) '
"VALUES (%s, %s, 'check-in', %s, %s, %s, %s)",
(badge, device_id, _now(), scanned_viruses, log_locker, sanitized),
)
if locker is not None:
cur.execute(
'UPDATE devices SET status = %s, locker_location = %s '
'WHERE device_id = %s',
(_STATUS_CHECKED_IN, locker, device_id),
)
else:
cur.execute(
'UPDATE devices SET status = %s WHERE device_id = %s',
(_STATUS_CHECKED_IN, device_id),
)
conn.commit()
with conn.cursor() as cur:
row = _fetch_device(cur, device_id)
result = _device_to_dict(cur, row)
finally:
conn.close()
return success_response(result, message='Device checked in')
# =============================================================================
# Checkout History
# Check-in/out log queries
# =============================================================================
@usb_bp.route('/<int:device_id>/history', methods=['GET'])
@usb_bp.route('/<device_id>/history', methods=['GET'])
@jwt_required(optional=True)
def get_device_history(device_id: int):
"""Get checkout history for a USB device."""
def get_device_history(device_id):
"""Paginated check-in/out log for one device."""
page, per_page = get_pagination_params(request)
query = USBCheckout.query.filter_by(
usbdeviceid=device_id
).order_by(USBCheckout.checkouttime.desc())
conn = cmmc_usb_connection()
try:
with conn.cursor() as cur:
cur.execute(
'SELECT COUNT(*) AS total FROM checkinoutlog WHERE device_id = %s',
(device_id,),
)
total = cur.fetchone()['total']
items, total = paginate_query(query, page, per_page)
data = [c.to_dict() for c in items]
cur.execute(
'SELECT ' + _LOG_COLS + ' FROM checkinoutlog WHERE device_id = %s '
'ORDER BY timestamp DESC, log_id DESC LIMIT %s OFFSET %s',
(device_id, per_page, (page - 1) * per_page),
)
rows = cur.fetchall()
data = [_log_to_dict(cur, row) for row in rows]
finally:
conn.close()
return paginated_response(data, page, per_page, total)
@@ -369,29 +603,48 @@ def get_device_history(device_id: int):
@usb_bp.route('/checkouts', methods=['GET'])
@jwt_required(optional=True)
def list_all_checkouts():
"""
List all checkouts (active and historical).
"""List check-out log rows.
Query parameters:
- active: Filter to only active (not returned) checkouts
- sso: Filter by user SSO
- active: true -> only rows whose device is currently checked out
- badge: filter by badge_number
"""
page, per_page = get_pagination_params(request)
query = USBCheckout.query
where = ["l.action = 'check-out'"]
params = []
# Filter by active only
if request.args.get('active', '').lower() == 'true':
query = query.filter(USBCheckout.checkintime == None)
where.append("d.status = 'checked-out'")
# Filter by user
if sso := request.args.get('sso'):
query = query.filter(USBCheckout.sso == sso)
badge = request.args.get('badge', '').strip()
if badge:
where.append('l.badge_number = %s')
params.append(badge)
query = query.order_by(USBCheckout.checkouttime.desc())
where_sql = ' WHERE ' + ' AND '.join(where)
base = (
'FROM checkinoutlog l JOIN devices d ON l.device_id = d.device_id'
+ where_sql
)
items, total = paginate_query(query, page, per_page)
data = [c.to_dict() for c in items]
conn = cmmc_usb_connection()
try:
with conn.cursor() as cur:
cur.execute('SELECT COUNT(*) AS total ' + base, params)
total = cur.fetchone()['total']
cur.execute(
'SELECT l.log_id, l.badge_number, l.device_id, l.action, '
'l.timestamp, l.scanned_viruses, l.locker_location, l.sanitized, '
'd.status AS device_status ' + base
+ ' ORDER BY l.timestamp DESC, l.log_id DESC LIMIT %s OFFSET %s',
params + [per_page, (page - 1) * per_page],
)
rows = cur.fetchall()
data = [_log_to_dict(cur, row) for row in rows]
finally:
conn.close()
return paginated_response(data, page, per_page, total)
@@ -399,9 +652,27 @@ def list_all_checkouts():
@usb_bp.route('/checkouts/active', methods=['GET'])
@jwt_required(optional=True)
def list_active_checkouts():
"""List all currently active checkouts."""
checkouts = USBCheckout.query.filter(
USBCheckout.checkintime == None
).order_by(USBCheckout.checkouttime.desc()).all()
"""List the latest check-out log row for every currently checked-out device."""
conn = cmmc_usb_connection()
try:
with conn.cursor() as cur:
cur.execute(
'SELECT l.log_id, l.badge_number, l.device_id, l.action, '
'l.timestamp, l.scanned_viruses, l.locker_location, l.sanitized, '
'd.status AS device_status '
'FROM devices d '
'JOIN checkinoutlog l ON l.device_id = d.device_id '
"AND l.action = 'check-out' "
"WHERE d.status = 'checked-out' "
'AND l.log_id = ('
' SELECT MAX(l2.log_id) FROM checkinoutlog l2 '
" WHERE l2.device_id = d.device_id AND l2.action = 'check-out'"
') '
'ORDER BY l.timestamp DESC, l.log_id DESC',
)
rows = cur.fetchall()
data = [_log_to_dict(cur, row) for row in rows]
finally:
conn.close()
return success_response([c.to_dict() for c in checkouts])
return success_response(data)

5
plugins/warranty/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
"""Warranty plugin package."""
from .plugin import WarrantyPlugin
__all__ = ['WarrantyPlugin']

5
plugins/warranty/api/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
"""Warranty plugin API."""
from .routes import warranty_bp
__all__ = ['warranty_bp']

228
plugins/warranty/api/routes.py Normal file
View File

@@ -0,0 +1,228 @@
"""Warranty API: manual CRUD now, provider refresh stubbed for later phases.
Coverage status is derived from enddate at read time (see models.derive_status),
never stored. Warranties link to assets many-to-many via warrantyassets, though
the common case is one warranty per asset.
"""
from datetime import date, datetime
from flask import Blueprint, request
from flask_jwt_extended import jwt_required
from shopdb.api import (
db, Asset,
success_response, error_response, ErrorCodes,
require_permission,
)
from ..models import Warranty, WarrantyAsset
from ..services import get_provider, ProviderNotConfigured
warranty_bp = Blueprint('warranty', __name__)
def _parse_date(value):
"""Accept 'YYYY-MM-DD' (or None/empty) -> date or None."""
if not value:
return None
try:
return datetime.strptime(value[:10], '%Y-%m-%d').date()
except (ValueError, TypeError):
return None
def _asset_summary(asset):
return {
'assetid': asset.assetid,
'assetnumber': asset.assetnumber,
'name': asset.name,
'assettypename': asset.assettype.assettype if asset.assettype else None,
}
def _warranty_payload(warranty, today=None):
"""to_dict plus the linked-asset summaries."""
data = warranty.to_dict(today)
assets = []
for link in warranty.links:
asset = Asset.query.get(link.assetid)
if asset:
assets.append(_asset_summary(asset))
data['assets'] = assets
return data
def _apply_links(warranty, assetids):
"""Replace a warranty's asset links with the given asset id list."""
if assetids is None:
return
wanted = {int(a) for a in assetids if str(a).strip()}
existing = {link.assetid: link for link in warranty.links}
for assetid in wanted - set(existing):
if Asset.query.get(assetid):
warranty.links.append(WarrantyAsset(assetid=assetid))
for assetid in set(existing) - wanted:
warranty.links.remove(existing[assetid])
# =============================================================================
# CRUD
# =============================================================================
@warranty_bp.route('', methods=['GET'])
@jwt_required(optional=True)
def list_warranties():
"""List warranties. Filters: ?status=, ?assetid=, ?active=false."""
query = Warranty.query
if request.args.get('active', 'true').lower() != 'false':
query = query.filter_by(isactive=True)
assetid = request.args.get('assetid', type=int)
if assetid:
query = (query.join(WarrantyAsset, WarrantyAsset.warrantyid == Warranty.warrantyid)
.filter(WarrantyAsset.assetid == assetid))
warranties = query.order_by(Warranty.enddate.is_(None), Warranty.enddate).all()
today = date.today()
items = [_warranty_payload(w, today) for w in warranties]
status_filter = request.args.get('status')
if status_filter:
items = [i for i in items if i['status'] == status_filter]
return success_response(items)
@warranty_bp.route('/asset/<int:assetid>', methods=['GET'])
@jwt_required(optional=True)
def warranties_for_asset(assetid):
"""Warranties covering one asset (for the asset-detail panel)."""
links = WarrantyAsset.query.filter_by(assetid=assetid).all()
today = date.today()
items = []
for link in links:
w = Warranty.query.get(link.warrantyid)
if w and w.isactive:
items.append(_warranty_payload(w, today))
return success_response(items)
@warranty_bp.route('/<int:warrantyid>', methods=['GET'])
@jwt_required(optional=True)
def get_warranty(warrantyid):
warranty = Warranty.query.get(warrantyid)
if not warranty:
return error_response(ErrorCodes.NOT_FOUND, 'Warranty not found', http_code=404)
return success_response(_warranty_payload(warranty))
@warranty_bp.route('', methods=['POST'])
@jwt_required()
@require_permission('warranty.create')
def create_warranty():
data = request.get_json() or {}
vendor = (data.get('vendor') or '').strip()
if not vendor:
return error_response(ErrorCodes.VALIDATION_ERROR, 'vendor is required')
warranty = Warranty(
vendor=vendor,
servicetag=(data.get('servicetag') or '').strip() or None,
provider=(data.get('provider') or 'manual').strip().lower(),
servicelevel=(data.get('servicelevel') or '').strip() or None,
startdate=_parse_date(data.get('startdate')),
enddate=_parse_date(data.get('enddate')),
notes=(data.get('notes') or '').strip() or None,
)
_apply_links(warranty, data.get('assetids'))
db.session.add(warranty)
db.session.commit()
return success_response(_warranty_payload(warranty), message='Warranty created', http_code=201)
@warranty_bp.route('/<int:warrantyid>', methods=['PUT'])
@jwt_required()
@require_permission('warranty.edit')
def update_warranty(warrantyid):
warranty = Warranty.query.get(warrantyid)
if not warranty:
return error_response(ErrorCodes.NOT_FOUND, 'Warranty not found', http_code=404)
data = request.get_json() or {}
if 'vendor' in data:
warranty.vendor = (data['vendor'] or '').strip() or warranty.vendor
if 'servicetag' in data:
warranty.servicetag = (data['servicetag'] or '').strip() or None
if 'provider' in data:
warranty.provider = (data['provider'] or 'manual').strip().lower()
if 'servicelevel' in data:
warranty.servicelevel = (data['servicelevel'] or '').strip() or None
if 'startdate' in data:
warranty.startdate = _parse_date(data['startdate'])
if 'enddate' in data:
warranty.enddate = _parse_date(data['enddate'])
if 'notes' in data:
warranty.notes = (data['notes'] or '').strip() or None
if 'isactive' in data:
warranty.isactive = bool(data['isactive'])
if 'assetids' in data:
_apply_links(warranty, data['assetids'])
db.session.commit()
return success_response(_warranty_payload(warranty), message='Warranty updated')
@warranty_bp.route('/<int:warrantyid>', methods=['DELETE'])
@jwt_required()
@require_permission('warranty.delete')
def delete_warranty(warrantyid):
warranty = Warranty.query.get(warrantyid)
if not warranty:
return error_response(ErrorCodes.NOT_FOUND, 'Warranty not found', http_code=404)
db.session.delete(warranty)
db.session.commit()
return success_response(message='Warranty deleted')
# =============================================================================
# Provider refresh (phase 1: manual only; API providers report not-configured)
# =============================================================================
@warranty_bp.route('/<int:warrantyid>/refresh', methods=['POST'])
@jwt_required()
@require_permission('warranty.edit')
def refresh_warranty(warrantyid):
warranty = Warranty.query.get(warrantyid)
if not warranty:
return error_response(ErrorCodes.NOT_FOUND, 'Warranty not found', http_code=404)
provider = get_provider(warranty.provider)
try:
result = provider.lookup(warranty.servicetag, warranty.vendor)
except ProviderNotConfigured as exc:
return error_response(ErrorCodes.VALIDATION_ERROR, str(exc), http_code=400)
if not result:
return error_response(ErrorCodes.VALIDATION_ERROR,
'This warranty is manual - nothing to refresh.', http_code=400)
if result.get('servicelevel'):
warranty.servicelevel = result['servicelevel']
if result.get('startdate'):
warranty.startdate = _parse_date(result['startdate'])
if result.get('enddate'):
warranty.enddate = _parse_date(result['enddate'])
warranty.lastcheckeddate = datetime.utcnow()
db.session.commit()
return success_response(_warranty_payload(warranty), message='Warranty refreshed')
# =============================================================================
# Report buckets (for the Reports hub)
# =============================================================================
@warranty_bp.route('/report', methods=['GET'])
@jwt_required(optional=True)
def warranty_report():
"""Counts + lists bucketed by derived status."""
today = date.today()
buckets = {'expired': [], 'expiring': [], 'active': [], 'unknown': []}
for w in Warranty.query.filter_by(isactive=True).all():
buckets.setdefault(w.status(today), []).append(_warranty_payload(w, today))
return success_response({
'counts': {k: len(v) for k, v in buckets.items()},
'buckets': buckets,
})

12
plugins/warranty/manifest.json Normal file
View File

@@ -0,0 +1,12 @@
{
"name": "warranty",
"version": "1.0.0",
"description": "Asset warranty tracking - manual entry now, Dell/Lenovo/HP provider lookups later. Derived coverage status + report buckets.",
"author": "ShopDB Team",
"dependencies": [],
"core_version": ">=0.2.0,<1.0.0",
"api_prefix": "/api/warranty",
"provides": {
"features": ["warranty-tracking"]
}
}

5
plugins/warranty/models/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
"""Warranty plugin models."""
from .warranty import Warranty, WarrantyAsset, derive_status, STATUS_COLORS
__all__ = ['Warranty', 'WarrantyAsset', 'derive_status', 'STATUS_COLORS']

96
plugins/warranty/models/warranty.py Normal file
View File

@@ -0,0 +1,96 @@
"""Warranty models.
A Warranty is provider-agnostic (manual entry, or looked up from Dell/Lenovo/HP
later). It links to one or more assets via warrantyassets. Coverage status is
DERIVED from enddate at read time, never stored, so it is always current.
"""
from datetime import date, timedelta
from shopdb.api import db
# Window before enddate where a warranty counts as "expiring soon".
EXPIRING_WINDOW_DAYS = 180
# Derived status -> display color (hex). Reused by the frontend status badge.
STATUS_COLORS = {
'active': '#4CAF50',
'expiring': '#FF9800',
'expired': '#F44336',
'unknown': '#9E9E9E',
}
def derive_status(enddate, today=None):
"""Coverage status from an end date. Never stored - always computed."""
if not enddate:
return 'unknown'
today = today or date.today()
if enddate < today:
return 'expired'
if enddate <= today + timedelta(days=EXPIRING_WINDOW_DAYS):
return 'expiring'
return 'active'
class Warranty(db.Model):
__tablename__ = 'warranties'
warrantyid = db.Column(db.Integer, primary_key=True)
vendor = db.Column(db.String(100), nullable=False)
# Service tag / serial the provider identifies the unit by.
servicetag = db.Column(db.String(100))
# Where the record came from: manual, dell, lenovo, hp.
provider = db.Column(db.String(20), nullable=False, server_default='manual')
servicelevel = db.Column(db.String(150))
startdate = db.Column(db.Date)
enddate = db.Column(db.Date)
# When a provider lookup last refreshed this record.
lastcheckeddate = db.Column(db.DateTime)
notes = db.Column(db.Text)
isactive = db.Column(db.Boolean, nullable=False, server_default='1')
links = db.relationship('WarrantyAsset', back_populates='warranty',
cascade='all, delete-orphan')
def status(self, today=None):
return derive_status(self.enddate, today)
def to_dict(self, today=None):
status = self.status(today)
return {
'warrantyid': self.warrantyid,
'vendor': self.vendor,
'servicetag': self.servicetag,
'provider': self.provider,
'servicelevel': self.servicelevel,
'startdate': self.startdate.isoformat() if self.startdate else None,
'enddate': self.enddate.isoformat() if self.enddate else None,
'lastcheckeddate': self.lastcheckeddate.isoformat() + 'Z' if self.lastcheckeddate else None,
'notes': self.notes,
'isactive': bool(self.isactive),
'status': status,
'statuscolor': STATUS_COLORS.get(status, STATUS_COLORS['unknown']),
}
class WarrantyAsset(db.Model):
__tablename__ = 'warrantyassets'
id = db.Column(db.Integer, primary_key=True)
warrantyid = db.Column(
db.Integer,
db.ForeignKey('warranties.warrantyid', ondelete='CASCADE'),
nullable=False
)
assetid = db.Column(
db.Integer,
db.ForeignKey('assets.assetid', ondelete='CASCADE'),
nullable=False
)
warranty = db.relationship('Warranty', back_populates='links')
__table_args__ = (
db.UniqueConstraint('warrantyid', 'assetid', name='uq_warrantyasset_warranty_asset'),
)

65
plugins/warranty/plugin.py Normal file
View File

@@ -0,0 +1,65 @@
"""Warranty plugin main class.
Asset-general plugin: owns its warranties + warrantyassets tables, an API
surface, and a sidebar entry. Not tied to any one asset type - a warranty can
cover a PC, printer, network device, or equipment.
"""
import json
import logging
from pathlib import Path
from typing import List, Dict, Optional, Type
from flask import Flask, Blueprint
from shopdb.plugins.base import BasePlugin, PluginMeta
from .api import warranty_bp
from .models import Warranty, WarrantyAsset
logger = logging.getLogger(__name__)
class WarrantyPlugin(BasePlugin):
"""Warranty tracking plugin."""
def __init__(self):
self._manifest = self._load_manifest()
def _load_manifest(self) -> Dict:
manifest_path = Path(__file__).parent / 'manifest.json'
if manifest_path.exists():
with open(manifest_path, 'r') as f:
return json.load(f)
return {}
@property
def meta(self) -> PluginMeta:
return PluginMeta(
name=self._manifest.get('name', 'warranty'),
version=self._manifest.get('version', '1.0.0'),
description=self._manifest.get('description', 'Asset warranty tracking'),
author=self._manifest.get('author', 'ShopDB Team'),
dependencies=self._manifest.get('dependencies', []),
core_version=self._manifest.get('core_version', '>=0.2.0,<1.0.0'),
api_prefix=self._manifest.get('api_prefix', '/api/warranty'),
)
def get_blueprint(self) -> Optional[Blueprint]:
return warranty_bp
def get_models(self) -> List[Type]:
return [Warranty, WarrantyAsset]
def get_navigation_items(self) -> List[Dict]:
return [
{
'name': 'Warranties',
'icon': 'shield',
'route': '/warranties',
'position': 8,
},
]
def init_app(self, app: Flask, db_instance) -> None:
logger.info(f"Warranty plugin initialized (v{self.meta.version})")

9
plugins/warranty/services/__init__.py Normal file
View File

@@ -0,0 +1,9 @@
"""Warranty plugin services."""
from .providers import (
get_provider,
provider_names,
ProviderNotConfigured,
)
__all__ = ['get_provider', 'provider_names', 'ProviderNotConfigured']

93
plugins/warranty/services/providers.py Normal file
View File

@@ -0,0 +1,93 @@
"""Warranty provider abstraction.
A provider looks up coverage for a unit by service tag / serial. Phase 1 ships
manual entry plus provider stubs that read per-vendor API config from settings.
When a real API (Dell TechDirect, Lenovo, HP) is wired later, only the matching
provider's lookup() body changes - callers and the API surface stay the same.
"""
from shopdb.api import db
from shopdb.core.models import Setting
class ProviderNotConfigured(Exception):
"""Raised when a provider is asked to look up but has no API config."""
pass
def _setting(key, default=None):
row = Setting.query.filter_by(key=key).first()
return row.value if row and row.value not in (None, '') else default
class WarrantyProvider:
"""Base provider. name matches Warranty.provider values."""
name = 'base'
def lookup(self, servicetag, vendor=None):
"""Return a dict of {servicelevel, startdate, enddate} or None.
Raises ProviderNotConfigured when the provider needs API creds it does
not have.
"""
raise NotImplementedError
class ManualProvider(WarrantyProvider):
"""No external lookup - values are entered by hand."""
name = 'manual'
def lookup(self, servicetag, vendor=None):
return None
class ApiProvider(WarrantyProvider):
"""Common shape for vendor API providers. Reads enabled/url/token from
settings under warranty_<name>_*. Real HTTP call is deferred to a later
phase; today it fails loud if asked to look up so manual entry is unaffected.
"""
def _config(self):
enabled = str(_setting(f'warranty_{self.name}_enabled', 'false')).lower() == 'true'
url = _setting(f'warranty_{self.name}_apiurl')
token = _setting(f'warranty_{self.name}_apitoken')
return enabled, url, token
def lookup(self, servicetag, vendor=None):
enabled, url, token = self._config()
if not (enabled and url and token):
raise ProviderNotConfigured(
f'{self.name} warranty lookup is not configured. '
f'Set warranty_{self.name}_enabled/apiurl/apitoken in Settings.'
)
# Phase 2+: perform the vendor API call here and map the response to
# {servicelevel, startdate, enddate}. Until then, signal not-yet-built.
raise ProviderNotConfigured(
f'{self.name} API lookup not implemented yet (config present).'
)
class DellProvider(ApiProvider):
name = 'dell'
class LenovoProvider(ApiProvider):
name = 'lenovo'
class HpProvider(ApiProvider):
name = 'hp'
_PROVIDERS = {p.name: p for p in (
ManualProvider(), DellProvider(), LenovoProvider(), HpProvider()
)}
def get_provider(name):
"""Return a provider instance, defaulting to manual for unknown names."""
return _PROVIDERS.get((name or 'manual').lower(), _PROVIDERS['manual'])
def provider_names():
return list(_PROVIDERS.keys())