"""Computers plugin main class.""" import json import logging from pathlib import Path from typing import List, Dict, Optional, Type from flask import Flask, Blueprint import click from shopdb.plugins.base import BasePlugin, PluginMeta from shopdb.api import db, AssetType from .models import Computer, ComputerType, ComputerInstalledApp, AccessProtocol, ComputerAccess from .api import computers_bp logger = logging.getLogger(__name__) # Marker stamped on PC->printer links this collector creates. The stale-link # archive only touches rows carrying this label, so manually-created printer # relationships are never removed by a collector push. Stored in the # assetrelationships.label column (no origin column exists; see BUILD notes). PRINTER_LINK_ORIGIN = 'collector:printers' # Marker stamped on PC->measuringtool "controls" links this collector creates # for metrology PCs (CMM, Keyence, Genspect, wax-and-trace). Same stale-link # discipline as PRINTER_LINK_ORIGIN: only rows carrying this label are archived # by a collector push, so hand-made tool links survive. MEASURINGTOOL_LINK_ORIGIN = 'collector:measuringtool' class ComputersPlugin(BasePlugin): """ Computers plugin - manages PC, server, and workstation assets. Computers include shopfloor PCs, engineer workstations, servers, etc. Uses the new Asset architecture with Computer extension table. """ def __init__(self): self._manifest = self._load_manifest() def _load_manifest(self) -> Dict: """Load plugin manifest from JSON file.""" manifestpath = Path(__file__).parent / 'manifest.json' if manifestpath.exists(): with open(manifestpath, 'r') as f: return json.load(f) return {} @property def meta(self) -> PluginMeta: """Return plugin metadata.""" return PluginMeta( name=self._manifest.get('name', 'computers'), version=self._manifest.get('version', '1.0.0'), description=self._manifest.get( 'description', 'Computer management for PCs, servers, and workstations' ), author=self._manifest.get('author', 'ShopDB Team'), dependencies=self._manifest.get('dependencies', []), core_version=self._manifest.get('core_version', '>=1.0.0'), api_prefix=self._manifest.get('api_prefix', '/api/computers'), ) def get_blueprint(self) -> Optional[Blueprint]: """Return Flask Blueprint with API routes.""" return computers_bp def get_models(self) -> List[Type]: """Return list of SQLAlchemy model classes.""" return [Computer, ComputerType, ComputerInstalledApp, AccessProtocol, ComputerAccess] def init_app(self, app: Flask, db_instance) -> None: """Initialize plugin with Flask app.""" logger.info(f"Computers plugin initialized (v{self.meta.version})") # -- ADR-006 collector contract ----------------------------------------- def get_collector_schema(self) -> Optional[Dict]: """Schema for the PC collector payload (matched by hostname). Aligns with the GE-Enforce status shape (transport may change), using the project naming convention (lowercase concatenated). The caller maps its own field names to these. """ return { 'identityfield': 'hostname', 'fields': { 'hostname': {'type': 'string', 'required': True}, 'machinenumber': {'type': 'string'}, 'pctype': {'type': 'string'}, 'pcsubtype': {'type': 'string'}, 'serialnumber': {'type': 'string'}, 'loggedinuser': {'type': 'string'}, 'lastboottime': {'type': 'string', 'format': 'date-time'}, 'lastcheckin': {'type': 'string', 'format': 'date-time'}, 'ipaddress': {'type': 'string'}, 'vendorname': {'type': 'string'}, 'modelnumber': {'type': 'string'}, 'osname': {'type': 'string'}, 'installedsoftware': { 'type': 'array', 'items': {'name': 'string', 'version': 'string'}, }, # Printer identifiers reported by the GE-Enforce side, which # runs Get-CimInstance Win32_Printer and marks the default with # the Default flag. An identifier is the printer's windows name, # share name, hostname, or port/IP; the collector resolves it # flexibly to a printer asset. Both optional. Presence of either # key drives the PC->printer relationship sync (and stale-link # archive); absence leaves existing printer links untouched. 'defaultprinter': { 'type': 'string', 'description': ('Default printer identifier (Win32_Printer ' 'with Default=true): windows name, share ' 'name, hostname, or port IP.'), }, 'printers': { 'type': 'array', 'items': {'type': 'string'}, 'description': ('All installed network printer identifiers ' '(Win32_Printer): windows name / share / ' 'hostname / IP. Unresolved -> warning.'), }, 'accessprotocols': { 'type': 'array', 'items': {'type': 'string'}, 'description': ('Remote-access protocols this PC actually ' 'exposes, by catalog name (VNC, WinRM, ' 'RDP). Presence of the key drives the sync: ' 'reported protocols are activated and ' 'catalogued ones not reported are ' 'deactivated. Omit the key entirely to ' 'leave existing rows alone - most came from ' 'the legacy isvnc/iswinrm migration.'), }, }, } def apply_collector_payload(self, payload: Dict) -> Dict: """Idempotent upsert of a PC from a collector payload (by hostname).""" from datetime import datetime, timezone from shopdb.api import ( Asset, Application, Communication, CommunicationType, Vendor, Model, OperatingSystem, ) from .pctypemap import pctype_mapping warnings = [] hostname = (payload.get('hostname') or '').strip() if not hostname: raise ValueError('hostname is required') # Machine number is the business identifier (Asset.assetnumber). Skip # the imaging-time placeholder '9999' and fall back to hostname. machinenumber = (payload.get('machinenumber') or '').strip() if machinenumber in ('', '9999'): machinenumber = None comp = Computer.query.filter(Computer.hostname.ilike(hostname)).first() if not comp: comp = (Computer.query.join(Asset, Asset.assetid == Computer.assetid) .filter(Asset.assetnumber.ilike(hostname)).first()) action = 'updated' if not comp: atype = AssetType.query.filter_by(assettype='computer').first() # statusid=1 is the first seeded asset status ("In Use"); a # collector-discovered PC is by definition in use. asset = Asset(assetnumber=machinenumber or hostname, assettypeid=atype.assettypeid, statusid=1) db.session.add(asset) db.session.flush() comp = Computer(assetid=asset.assetid, hostname=hostname) db.session.add(comp) db.session.flush() action = 'created' elif machinenumber and comp.asset: comp.asset.assetnumber = machinenumber comp.lastreporteddate = datetime.now(timezone.utc).replace(tzinfo=None) if payload.get('lastboottime'): try: comp.lastboottime = datetime.fromisoformat( payload['lastboottime'].replace('Z', '+00:00')) except (ValueError, AttributeError): warnings.append('lastboottime not parseable') loggedinuser = payload.get('loggedinuser') or payload.get('currentuser') if loggedinuser: comp.loggedinuser = loggedinuser if payload.get('serialnumber') and comp.asset: comp.asset.serialnumber = payload['serialnumber'] # pc-type -> ComputerType via the configurable settings mapping. pctype = (payload.get('pctype') or '').strip() if pctype: from .models import ComputerType mapped = pctype_mapping().get(pctype) if not mapped: warnings.append(f'no ComputerType mapping for pctype: {pctype}') else: ctype = ComputerType.query.filter_by(computertype=mapped).first() if ctype: comp.computertypeid = ctype.computertypeid else: warnings.append(f'mapped ComputerType not found: {mapped}') # Vendor / model are free vocab - create if missing. vendorname = (payload.get('vendorname') or '').strip() vendor = None if vendorname: vendor = Vendor.query.filter(Vendor.vendor.ilike(vendorname)).first() if not vendor: vendor = Vendor(vendor=vendorname) db.session.add(vendor) db.session.flush() comp.vendorid = vendor.vendorid modelnumber = (payload.get('modelnumber') or '').strip() if modelnumber: model_query = Model.query.filter(Model.modelnumber.ilike(modelnumber)) if vendor: model_query = model_query.filter(Model.vendorid == vendor.vendorid) model = model_query.first() if not model: model = Model(modelnumber=modelnumber, vendorid=vendor.vendorid if vendor else None) db.session.add(model) db.session.flush() comp.modelnumberid = model.modelnumberid # OS is a controlled vocab - look up only, warn if unknown. osname = (payload.get('osname') or '').strip() if osname: os_row = OperatingSystem.query.filter( OperatingSystem.osname.ilike(osname)).first() if os_row: comp.osid = os_row.osid else: warnings.append(f'unknown operating system: {osname}') if payload.get('pcsubtype'): warnings.append('pcsubtype received but not stored (no model field)') if payload.get('ipaddress'): ip_comtype = CommunicationType.query.filter_by(comtype='IP').first() primary = Communication.query.filter_by( assetid=comp.assetid, isprimary=True).first() if primary: primary.ipaddress = payload['ipaddress'] elif ip_comtype: db.session.add(Communication( assetid=comp.assetid, comtypeid=ip_comtype.comtypeid, ipaddress=payload['ipaddress'], isprimary=True)) for app_data in payload.get('installedsoftware', []) or []: name = app_data.get('name') if not name: continue app = Application.query.filter(Application.appname.ilike(name)).first() if not app: warnings.append(f'unknown application: {name}') continue installed = ComputerInstalledApp.query.filter_by( computerid=comp.computerid, appid=app.appid).first() version = app_data.get('version') if installed: installed.installedversion = version installed.isactive = True else: db.session.add(ComputerInstalledApp( computerid=comp.computerid, appid=app.appid, installedversion=version)) # Remote-access protocol sync (only when the payload carried the key). accessprotocols = self._sync_access_protocols(comp, payload, warnings) # Printer relationship sync (only when the payload carried printer data). printerlinks = self._sync_printer_links(comp.asset, payload, warnings) # Measuring-tool sync: metrology PCs (CMM/Keyence/Genspect/wax-trace) # get an attached MeasuringTool asset auto-created and linked. measuringtoollinks = self._sync_measuringtool_link( comp.asset, pctype, hostname, warnings) db.session.commit() return { 'action': action, 'assetid': comp.assetid, 'identityvalue': hostname, 'warnings': warnings, 'extra': { 'printerlinks': printerlinks, 'printerlinkcount': len(printerlinks), 'measuringtoollinks': measuringtoollinks, 'measuringtoollinkcount': len(measuringtoollinks), 'accessprotocols': accessprotocols, }, } # -- printer relationship sync ----------------------------------------- def _sync_access_protocols(self, comp, payload, warnings): """Idempotently sync a PC's remote-access protocols from the collector. Payload key 'accessprotocols' is a list of protocol NAMES as they appear in the accessprotocols catalog ('VNC', 'WinRM', 'RDP'), matched case-insensitively. An unknown name warns and is skipped rather than creating a protocol: the catalog is admin-managed on purpose, so a typo on one bay must not invent a protocol for the whole site. Presence of the key drives the sync, exactly like the printer links: a reported protocol is activated, and a catalogued protocol the PC did NOT report is deactivated (not deleted, so a port override survives a temporary outage). A payload with no 'accessprotocols' key leaves every existing row untouched - most PCs' rows came from the legacy isvnc/iswinrm migration and must not be wiped by a collector that simply does not report them yet. Returns the list of active protocol names after the sync. """ from plugins.computers.models import AccessProtocol, ComputerAccess if 'accessprotocols' not in payload: return [] reported = payload.get('accessprotocols') or [] if not isinstance(reported, list): warnings.append('accessprotocols must be a list of protocol names') return [] wanted = set() for name in reported: name = str(name or '').strip() if not name: continue protocol = AccessProtocol.query.filter( AccessProtocol.name.ilike(name)).first() if not protocol: warnings.append('unknown access protocol: {}'.format(name)) continue wanted.add(protocol.protocolid) existing = {row.protocolid: row for row in ComputerAccess.query.filter_by(computerid=comp.computerid).all()} for protocolid in wanted: row = existing.get(protocolid) if row: row.isactive = True else: db.session.add(ComputerAccess( computerid=comp.computerid, protocolid=protocolid, isactive=True)) # Deactivate what the PC no longer exposes. Kept as rows so a manual # portoverride is not lost the first time a service is briefly down. for protocolid, row in existing.items(): if protocolid not in wanted: row.isactive = False names = [p.name for p in AccessProtocol.query.filter( AccessProtocol.protocolid.in_(wanted)).all()] if wanted else [] return sorted(names) def _sync_printer_links(self, pcasset, payload, warnings): """Idempotently sync PC->printer relationships from collector printer data. Default printer -> 'defaultprinter' (directional). Other reported printers -> 'connectedto' (symmetric). Resolves each identifier to a printer asset by windows name / share / hostname / asset number / name or a communications IP. Unresolved identifiers add a warning and never fail the push. Stale-link archive: on each push, collector-tagged links (label == PRINTER_LINK_ORIGIN) whose (target, type) pair is not in the reported desired set are set inactive. Only tagged rows are touched, so manual links survive. Runs only when the payload carried a printer key ('defaultprinter' or 'printers'); a PC that reports without printer data keeps its existing links. Returns the desired-link list (created + kept). """ from shopdb.api import ( AssetRelationship, RelationshipType, Asset, Communication) has_default = 'defaultprinter' in payload has_list = 'printers' in payload if not has_default and not has_list: return [] try: from plugins.printers.models import Printer except ImportError: warnings.append('printers plugin unavailable; printer links skipped') return [] dp_type = RelationshipType.query.filter_by( relationshiptype='defaultprinter').first() ct_type = RelationshipType.query.filter_by( relationshiptype='connectedto').first() if not dp_type or not ct_type: warnings.append('printer relationship types missing; ' 'run flask seed reference-data') return [] def resolve(identifier): # first match wins: printer text identity, then a printer IP. ident = (identifier or '').strip() if not ident: return None printer = ( Printer.query.join(Asset, Asset.assetid == Printer.assetid) .filter(Asset.isactive == True) .filter(db.or_( Printer.windowsname.ilike(ident), Printer.hostname.ilike(ident), Printer.sharename.ilike(ident), Asset.assetnumber.ilike(ident), Asset.name.ilike(ident), )).first()) if printer: return printer.asset comm = ( db.session.query(Communication) .join(Printer, Printer.assetid == Communication.assetid) .filter(Communication.ipaddress == ident) .first()) if comm: return db.session.get(Asset, comm.assetid) return None pcid = pcasset.assetid desired = set() # (targetassetid, relationshiptypeid) to keep printerlinks = [] default_id = None default_ident = (payload.get('defaultprinter') or '').strip() if default_ident: target = resolve(default_ident) if target: default_id = target.assetid desired.add((default_id, dp_type.relationshiptypeid)) self._sync_one(pcid, default_id, dp_type) printerlinks.append({'assetid': default_id, 'relationshiptype': 'defaultprinter'}) else: warnings.append(f'unresolved default printer: {default_ident}') for ident in payload.get('printers') or []: target = resolve(ident) if not target: warnings.append(f'unresolved printer: {ident}') continue if target.assetid == default_id: continue # already the default link desired.add((target.assetid, ct_type.relationshiptypeid)) self._sync_one(pcid, target.assetid, ct_type) printerlinks.append({'assetid': target.assetid, 'relationshiptype': 'connectedto'}) # Archive collector-tagged links no longer reported (manual links, with # a NULL/other label, are never matched here). collector_rows = AssetRelationship.query.filter( AssetRelationship.sourceassetid == pcid, AssetRelationship.relationshiptypeid.in_( [dp_type.relationshiptypeid, ct_type.relationshiptypeid]), AssetRelationship.isactive == True, AssetRelationship.label == PRINTER_LINK_ORIGIN, ).all() for rel in collector_rows: if (rel.targetassetid, rel.relationshiptypeid) not in desired: rel.isactive = False return printerlinks def _sync_one(self, pcid, printerid, reltype): """Reactivate or create one collector PC->printer link (idempotent).""" from shopdb.api import AssetRelationship existing = AssetRelationship.query.filter_by( sourceassetid=pcid, targetassetid=printerid, relationshiptypeid=reltype.relationshiptypeid).first() if existing: # do not re-stamp label: a pre-existing manual row stays manual. if not existing.isactive: existing.isactive = True return existing rel = AssetRelationship( sourceassetid=pcid, targetassetid=printerid, relationshiptypeid=reltype.relationshiptypeid, label=PRINTER_LINK_ORIGIN) db.session.add(rel) return rel # -- measuring-tool sync ----------------------------------------------- def _sync_measuringtool_link(self, pcasset, pctype, hostname, warnings): """Auto-create + link the MeasuringTool a metrology PC drives. A CMM / Keyence / Genspect / wax-and-trace imaging pc-type means the shopfloor PC controls an attached measuring instrument. This creates that instrument once as a MeasuringTool asset and a directional PC->tool 'controls' relationship, tagged MEASURINGTOOL_LINK_ORIGIN so it is idempotent and self-archiving. The PC's own ComputerType is left alone (it stays a shopfloor PC). A non-metrology pc-type archives any collector-created tool link (e.g. a PC re-imaged to another type) but never deletes the tool asset, which may carry calibration history. Returns the desired-link list. """ from shopdb.api import AssetRelationship, RelationshipType, Asset from .pctypemap import metrology_tool_for tool_spec = metrology_tool_for(pctype) controls = RelationshipType.query.filter_by( relationshiptype='controls').first() if not controls: # No 'controls' type => no tool links can exist. Only a problem for a # metrology PC that needs one; stay quiet for ordinary PCs. if tool_spec: warnings.append("'controls' relationship type missing; " 'run flask seed reference-data') return [] # Collector-created tool links already on this PC (active or archived). existing = AssetRelationship.query.filter( AssetRelationship.sourceassetid == pcasset.assetid, AssetRelationship.relationshiptypeid == controls.relationshiptypeid, AssetRelationship.label == MEASURINGTOOL_LINK_ORIGIN, ).all() if not tool_spec: # Not a metrology PC: archive any collector-created tool link. for rel in existing: if rel.isactive: rel.isactive = False return [] try: from plugins.measuringtools.models import MeasuringTool except ImportError: warnings.append('measuringtools plugin unavailable; ' 'tool link skipped') return [] typename, typedescription = tool_spec tooltype = self._ensure_measuringtool_type(typename, typedescription) # Reuse any prior collector link (reactivate + retype) before creating, # so a re-metrology PC never duplicates the tool asset. reuse = next((rel for rel in existing if rel.isactive), None) \ or (existing[0] if existing else None) if reuse: reuse.isactive = True toolasset = db.session.get(Asset, reuse.targetassetid) if toolasset and toolasset.measuringtool and tooltype: toolasset.measuringtool.measuringtooltypeid = \ tooltype.measuringtooltypeid targetid = reuse.targetassetid else: mt_assettype = AssetType.query.filter_by( assettype='measuring_tool').first() if not mt_assettype: warnings.append('measuring_tool asset type missing; ' 'tool link skipped') return [] suffix = (pctype or '').split('-')[-1].upper() baseasset = pcasset.assetnumber or hostname toolasset = Asset( assetnumber=f'{baseasset}-{suffix}', name=f'{typename} ({hostname})', assettypeid=mt_assettype.assettypeid, statusid=1) db.session.add(toolasset) db.session.flush() db.session.add(MeasuringTool( assetid=toolasset.assetid, measuringtooltypeid=tooltype.measuringtooltypeid if tooltype else None)) db.session.add(AssetRelationship( sourceassetid=pcasset.assetid, targetassetid=toolasset.assetid, relationshiptypeid=controls.relationshiptypeid, label=MEASURINGTOOL_LINK_ORIGIN)) targetid = toolasset.assetid # Only one tool link is desired; archive any other collector rows. for rel in existing: if rel is not reuse and rel.isactive: rel.isactive = False return [{'assetid': targetid, 'relationshiptype': 'controls', 'measuringtooltype': typename}] def _ensure_measuringtool_type(self, name, description): """Find or create a MeasuringToolType (metrology types are not in the measuringtools starter seed).""" from plugins.measuringtools.models import MeasuringToolType tooltype = MeasuringToolType.query.filter_by(name=name).first() if not tooltype: tooltype = MeasuringToolType(name=name, description=description) db.session.add(tooltype) db.session.flush() return tooltype def on_install(self, app: Flask) -> None: """Called when plugin is installed.""" with app.app_context(): self._ensure_asset_type() self._ensure_computer_types() from .pctypemap import seed_pctype_settings seed_pctype_settings() db.session.commit() logger.info("Computers plugin installed") def _ensure_asset_type(self) -> None: """Ensure computer asset type exists.""" existing = AssetType.query.filter_by(assettype='computer').first() if not existing: at = AssetType( assettype='computer', pluginname='computers', tablename='computers', description='PCs, servers, and workstations', icon='desktop' ) db.session.add(at) logger.debug("Created asset type: computer") db.session.commit() def _ensure_computer_types(self) -> None: """Ensure basic computer types exist.""" computer_types = [ ('Shopfloor PC', 'PC located on the shop floor for machine operation', 'desktop'), ('Engineer Workstation', 'Engineering workstation for CAD/CAM work', 'laptop'), ('CMM PC', 'PC dedicated to CMM operation', 'desktop'), ('Server', 'Server system', 'server'), ('Kiosk', 'Kiosk or info display PC', 'tv'), ('Laptop', 'Laptop computer', 'laptop'), ('Virtual Machine', 'Virtual machine', 'cloud'), ('Other', 'Other computer type', 'desktop'), ] for name, description, icon in computer_types: existing = ComputerType.query.filter_by(computertype=name).first() if not existing: ct = ComputerType( computertype=name, description=description, icon=icon ) db.session.add(ct) logger.debug(f"Created computer type: {name}") db.session.commit() def on_uninstall(self, app: Flask) -> None: """Called when plugin is uninstalled.""" logger.info("Computers plugin uninstalled") def get_cli_commands(self) -> List: """Return CLI commands for this plugin.""" @click.group('computers') def computerscli(): """Computers plugin commands.""" pass @computerscli.command('list-types') def list_types(): """List all computer types.""" from flask import current_app with current_app.app_context(): types = ComputerType.query.filter_by(isactive=True).all() if not types: click.echo('No computer types found.') return click.echo('Computer Types:') for t in types: click.echo(f" [{t.computertypeid}] {t.computertype}") @computerscli.command('stats') def stats(): """Show computer statistics.""" from flask import current_app from shopdb.api import Asset with current_app.app_context(): total = db.session.query(Computer).join(Asset).filter( Asset.isactive == True ).count() click.echo(f"Total active computers: {total}") # 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.computertypeid == (sf.computertypeid if sf else -1) ).count() click.echo(f" Shopfloor PCs: {shopfloor}") click.echo(f" Other: {total - shopfloor}") @computerscli.command('find') @click.argument('hostname') def find_by_hostname(hostname): """Find a computer by hostname.""" from flask import current_app with current_app.app_context(): comp = Computer.query.filter( Computer.hostname.ilike(f'%{hostname}%') ).first() if not comp: click.echo(f'No computer found matching hostname: {hostname}') return click.echo(f'Found: {comp.hostname}') click.echo(f' Asset: {comp.asset.assetnumber}') click.echo(f' Type: {comp.computertype.computertype if comp.computertype else "N/A"}') click.echo(f' OS: {comp.operatingsystem.osname if comp.operatingsystem else "N/A"}') click.echo(f' Logged in: {comp.loggedinuser or "N/A"}') return [computerscli] def get_dashboard_widgets(self) -> List[Dict]: """Return dashboard widget definitions.""" return [ { 'name': 'Computer Status', 'component': 'ComputerStatusWidget', 'endpoint': '/api/computers/dashboard/summary', 'size': 'medium', 'position': 6, }, ] def get_navigation_items(self) -> List[Dict]: """Return navigation menu items.""" return [ { 'name': 'PCs', 'icon': 'desktop', 'route': '/pcs', 'position': 15, }, ] def get_permissions(self) -> List: """Return the RBAC permissions this plugin owns.""" return [ ('computers.view', 'View computers', 'computers'), ('computers.create', 'Create computers', 'computers'), ('computers.edit', 'Edit computers', 'computers'), ('computers.delete', 'Delete computers', 'computers'), ]