a846587f39
Email: a stdlib SMTP mailer (settings-first config, graceful no-op when unconfigured), a test-email endpoint wired to the Email settings page, forced first-login password change (users.mustchangepassword, migration 7d23, /change-password flow), new-user welcome mail, and on-demand report/alert delivery (POST /api/reports/email + Email Report buttons) with an external-cron-with-a-scoped-PAT path documented for automation. All tests patch smtplib - no network. Labels: a shared /print/asset-label/<type>/<id> view any asset detail page opens - card or plain style, QR or barcode, configurable encoding. Per-type qr_target_* templates plus label_default_style/codetype/encodes settings on the Printing page. Measuring-tool labels default to encoding their inspection-operation code (derived from the location name, e.g. 0615), so every tool in an area shares the area code - verified by decoding the rendered QR. Machine labels default to the machine number; blank-serial handled gracefully. 808 tests pass; both features verified live. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
286 lines
10 KiB
Python
286 lines
10 KiB
Python
"""Polymorphic Asset models - core of the new asset architecture."""
|
|
|
|
from shopdb.extensions import db
|
|
from .base import BaseModel, SoftDeleteMixin, AuditMixin
|
|
|
|
|
|
class AssetType(BaseModel):
|
|
"""
|
|
Registry of asset categories.
|
|
|
|
Each type maps to a plugin-owned extension table.
|
|
Examples: machine, computer, network_device, printer
|
|
"""
|
|
__tablename__ = 'assettypes'
|
|
|
|
assettypeid = db.Column(db.Integer, primary_key=True)
|
|
assettype = db.Column(
|
|
db.String(50),
|
|
unique=True,
|
|
nullable=False,
|
|
comment='Category name: machine, computer, network_device, printer'
|
|
)
|
|
pluginname = db.Column(
|
|
db.String(100),
|
|
nullable=True,
|
|
comment='Plugin that owns this type'
|
|
)
|
|
tablename = db.Column(
|
|
db.String(100),
|
|
nullable=True,
|
|
comment='Extension table name for this type'
|
|
)
|
|
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"<AssetType {self.assettype}>"
|
|
|
|
|
|
class AssetStatus(BaseModel):
|
|
"""Asset status options."""
|
|
__tablename__ = 'assetstatuses'
|
|
|
|
statusid = db.Column(db.Integer, primary_key=True)
|
|
status = db.Column(db.String(50), unique=True, nullable=False)
|
|
description = db.Column(db.Text)
|
|
color = db.Column(db.String(20), comment='CSS color for UI')
|
|
|
|
def __repr__(self):
|
|
return f"<AssetStatus {self.status}>"
|
|
|
|
|
|
class Asset(BaseModel, SoftDeleteMixin, AuditMixin):
|
|
"""
|
|
Core asset model - minimal shared fields.
|
|
|
|
Category-specific data lives in plugin extension tables
|
|
(machines, computers, network_devices, printers).
|
|
The assetid matches original machineid for migration compatibility.
|
|
"""
|
|
__tablename__ = 'assets'
|
|
|
|
assetid = db.Column(db.Integer, primary_key=True)
|
|
|
|
# Identification
|
|
assetnumber = db.Column(
|
|
db.String(50),
|
|
unique=True,
|
|
nullable=False,
|
|
index=True,
|
|
comment='Business identifier (e.g., CMM01, G5QX1GT3ESF)'
|
|
)
|
|
name = db.Column(
|
|
db.String(100),
|
|
comment='Display name/alias'
|
|
)
|
|
gaugelabreference = db.Column(
|
|
db.String(50),
|
|
index=True,
|
|
comment='Gauge lab asset reference (authoritative tag the gauge lab '
|
|
'assigns to machines); distinct from assetnumber'
|
|
)
|
|
maintenancereference = db.Column(
|
|
db.String(50),
|
|
index=True,
|
|
comment='Maintenance system asset reference; distinct from assetnumber'
|
|
)
|
|
serialnumber = db.Column(
|
|
db.String(100),
|
|
index=True,
|
|
comment='Hardware serial number'
|
|
)
|
|
|
|
# Classification
|
|
assettypeid = db.Column(
|
|
db.Integer,
|
|
db.ForeignKey('assettypes.assettypeid'),
|
|
nullable=False
|
|
)
|
|
statusid = db.Column(
|
|
db.Integer,
|
|
db.ForeignKey('assetstatuses.statusid'),
|
|
default=1,
|
|
comment='In Use, Spare, Retired, etc.'
|
|
)
|
|
|
|
# Location and organization
|
|
locationid = db.Column(
|
|
db.Integer,
|
|
db.ForeignKey('locations.locationid'),
|
|
nullable=True
|
|
)
|
|
businessunitid = db.Column(
|
|
db.Integer,
|
|
db.ForeignKey('businessunits.businessunitid'),
|
|
nullable=True
|
|
)
|
|
|
|
# Floor map position (ADR-001: asset-specific override; nullable)
|
|
mapx = db.Column(db.Integer, comment='X coordinate on floor map (ADR-001)')
|
|
mapy = db.Column(db.Integer, comment='Y coordinate on floor map (ADR-001)')
|
|
|
|
# Notes
|
|
notes = db.Column(db.Text, nullable=True)
|
|
|
|
# Relationships
|
|
assettype = db.relationship('AssetType', backref='assets')
|
|
status = db.relationship('AssetStatus', backref='assets')
|
|
location = db.relationship('Location', backref='assets')
|
|
businessunit = db.relationship('BusinessUnit', backref='assets')
|
|
|
|
# Communications (one-to-many) - will be migrated to use assetid
|
|
communications = db.relationship(
|
|
'Communication',
|
|
foreign_keys='Communication.assetid',
|
|
backref='asset',
|
|
cascade='all, delete-orphan',
|
|
lazy='dynamic'
|
|
)
|
|
|
|
# Indexes
|
|
__table_args__ = (
|
|
db.Index('idx_asset_type_bu', 'assettypeid', 'businessunitid'),
|
|
db.Index('idx_asset_location', 'locationid'),
|
|
db.Index('idx_asset_active', 'isactive'),
|
|
db.Index('idx_asset_status', 'statusid'),
|
|
)
|
|
|
|
def __repr__(self):
|
|
return f"<Asset {self.assetnumber}>"
|
|
|
|
@property
|
|
def display_name(self):
|
|
"""Get display name (name if set, otherwise assetnumber)."""
|
|
return self.name or self.assetnumber
|
|
|
|
@property
|
|
def primary_ip(self):
|
|
"""Get primary IP address from communications."""
|
|
comm = self.communications.filter_by(
|
|
isprimary=True,
|
|
comtypeid=1 # IP type
|
|
).first()
|
|
if comm:
|
|
return comm.ipaddress
|
|
# Fall back to any IP
|
|
comm = self.communications.filter_by(comtypeid=1).first()
|
|
return comm.ipaddress if comm else None
|
|
|
|
def get_inherited_location(self):
|
|
"""
|
|
Get location data from a related asset if this asset has none.
|
|
|
|
Returns dict with locationid, location_name, mapx, mapy, and
|
|
inherited_from (assetnumber of source asset) if location was inherited.
|
|
Returns None if no location data available.
|
|
"""
|
|
if self.locationid is not None or (self.mapx is not None and self.mapy is not None):
|
|
return None
|
|
|
|
related_assets = []
|
|
|
|
if hasattr(self, 'incoming_relationships'):
|
|
for rel in self.incoming_relationships:
|
|
if rel.sourceasset and rel.isactive:
|
|
related_assets.append(rel.sourceasset)
|
|
|
|
if hasattr(self, 'outgoing_relationships'):
|
|
for rel in self.outgoing_relationships:
|
|
if rel.targetasset and rel.isactive:
|
|
related_assets.append(rel.targetasset)
|
|
|
|
for related in related_assets:
|
|
if related.locationid is not None or (related.mapx is not None and related.mapy is not None):
|
|
return {
|
|
'locationid': related.locationid,
|
|
'locationname': related.location.locationname if related.location else None,
|
|
'mapx': related.mapx,
|
|
'mapy': related.mapy,
|
|
'inheritedfrom': related.assetnumber
|
|
}
|
|
|
|
return None
|
|
|
|
def to_dict(self, include_type_data=False, include_inherited_location=True):
|
|
"""
|
|
Convert model to dictionary.
|
|
|
|
Args:
|
|
include_type_data: If True, include category-specific data from extension table
|
|
include_inherited_location: If True, include location from related assets when missing
|
|
"""
|
|
result = super().to_dict()
|
|
|
|
# Add related object names for convenience
|
|
if self.assettype:
|
|
result['assettypename'] = self.assettype.assettype
|
|
result['assettypecolor'] = getattr(self.assettype, 'color', None)
|
|
if self.status:
|
|
result['statusname'] = self.status.status
|
|
result['statuscolor'] = self.status.color
|
|
if self.location:
|
|
result['locationname'] = self.location.locationname
|
|
if self.businessunit:
|
|
result['businessunitname'] = self.businessunit.businessunit
|
|
|
|
# Add plugin-specific ID for navigation purposes
|
|
if hasattr(self, 'machine') and self.machine:
|
|
result['pluginid'] = self.machine.machineid
|
|
elif hasattr(self, 'computer') and self.computer:
|
|
result['pluginid'] = self.computer.computerid
|
|
elif hasattr(self, 'network_device') and self.network_device:
|
|
result['pluginid'] = self.network_device.networkdeviceid
|
|
elif hasattr(self, 'printer') and self.printer:
|
|
result['pluginid'] = self.printer.printerid
|
|
elif hasattr(self, 'measuringtool') and self.measuringtool:
|
|
result['pluginid'] = self.measuringtool.measuringtoolid
|
|
|
|
# Include inherited location if this asset has no location data
|
|
if include_inherited_location:
|
|
inherited = self.get_inherited_location()
|
|
if inherited:
|
|
result['inheritedlocation'] = inherited
|
|
# Also set the location fields if they're missing
|
|
if result.get('locationid') is None:
|
|
result['locationid'] = inherited['locationid']
|
|
result['locationname'] = inherited['locationname']
|
|
if result.get('mapx') is None:
|
|
result['mapx'] = inherited['mapx']
|
|
if result.get('mapy') is None:
|
|
result['mapy'] = inherited['mapy']
|
|
|
|
# Operation/short code of the resolved location (own or inherited).
|
|
# Derived from the location name's leading token; labels can encode a
|
|
# tool's inspection operation instead of the tool. None when unplaced.
|
|
from .location import derive_locationcode
|
|
result['locationcode'] = derive_locationcode(result.get('locationname'))
|
|
|
|
# Include extension data if requested
|
|
if include_type_data:
|
|
ext_data = self._get_extension_data()
|
|
if ext_data:
|
|
result['typedata'] = ext_data
|
|
|
|
return result
|
|
|
|
def _get_extension_data(self):
|
|
"""Get category-specific data from extension table."""
|
|
# Check for machine extension
|
|
if hasattr(self, 'machine') and self.machine:
|
|
return self.machine.to_dict()
|
|
# Check for computer extension
|
|
if hasattr(self, 'computer') and self.computer:
|
|
return self.computer.to_dict()
|
|
# Check for network_device extension
|
|
if hasattr(self, 'network_device') and self.network_device:
|
|
return self.network_device.to_dict()
|
|
# Check for printer extension
|
|
if hasattr(self, 'printer') and self.printer:
|
|
return self.printer.to_dict()
|
|
# Check for measuring-tool extension
|
|
if hasattr(self, 'measuringtool') and self.measuringtool:
|
|
return self.measuringtool.to_dict()
|
|
return None
|