ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
720f4ddbb2294d98d86b882360d94c56ddc8dcd5
shopdb-flask/plugins/computers/plugin.py
cproudlock 27b2817fd7 Import the wilm endpoint-monitor and channel-distribution work 
Everything on origin/wilm that is not on the published main, applied as one
change because the published history has rewritten hashes: per-commit
cherry-pick conflicts against our main even though the squashed diff applies
clean.

Contents: endpoint monitor catalog + policy attachment + warn-only drift
reconciler, operator issue reports from the tray, agent launch-path allowlist,
and the GHCR channel-image distribution path for dev workstations.

NOT READY FOR main. Open items:
  - 10 new routes are absent from docs/api-inventory.json, so
    test_api_inventory_parity fails. That gate SKIPS on the published mirror
    (the test ships, its data file is scrubbed), which is why wilm never saw it.
  - dev-bootstrap.ps1 and Start-ShopDBDev.ps1 hardcode the ge-aero GitHub org.
  - plugins/endpoints/plugin.py carries a whitespace-only re-indent.

Scope: product
2026-09-02 10:05:44 -04:00

2310 lines
108 KiB
Python
Raw Blame History

"""Computers plugin main class."""
import ipaddress
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 .models import (
ConfigurationPolicy,
ConfigurationPolicyAction,
BusinessPolicy,
BusinessPolicyItem,
)
from .api import computers_bp
from .policyflag import policy_flag_enabled as _policy_flag_enabled
logger = logging.getLogger(__name__)
# Marker stamped on PC->measuringtool "controls" links this collector creates
# for metrology PCs (CMM, Keyence, Genspect, wax-and-trace). Only rows carrying
# this label are archived by a collector push, so hand-made tool links survive.
# Stored in assetrelationships.label (no origin column exists; see BUILD notes).
#
# There is deliberately no printer equivalent any more. PRINTER_LINK_ORIGIN went
# with _sync_printer_links: printer ASSIGNMENT is never collected, so there are
# no collector-owned printer rows left to label or archive.
MEASURINGTOOL_LINK_ORIGIN = 'collector:measuringtool'
# Marker stamped on the PC->machine "controls" link built from the reported
# machine number. Same discipline as the two above: only rows carrying this
# label are archived by a collector push, so a link made by hand is never
# touched.
MACHINE_LINK_ORIGIN = 'collector:machine'
# Marker stamped on the PC->machine link built from a CSV import's machine
# number column. A DISTINCT origin from the collector's so the two provenances
# stay separable: an import establishes a link once from a spreadsheet, the
# collector maintains it continuously from the floor, and a site may use one,
# the other, or both. Same archive discipline - only rows carrying this label
# are touched by a re-import.
MACHINE_LINK_IMPORT_ORIGIN = 'import:machine'
# Marker stamped on both links a part-marker PC produces: PC controls marker,
# and marker partof the operation it serves. Same archive discipline again.
PARTMARKER_LINK_ORIGIN = 'collector:partmarker'
# How long a PC holding a machine may go without reporting before a second PC
# claiming that machine is treated as its replacement. A swap resolves itself
# within a day; a PC off overnight or behind a network outage keeps its bay.
# Shorter than this and a spare imaged on the bench could steal a live machine.
MACHINE_CLAIM_QUIET_HOURS = 24
def _corporate_ranges():
"""CIDRs this site considers its routable network, from settings.
Empty means the site has not said, and the fallback below applies. Kept
server-side rather than pushed to the bay: a bay cannot know a site's
addressing, and baking a range into the collector script would be an
ADR-015 violation the moment a second site ran it.
"""
from shopdb.api import Setting
raw = ''
try:
raw = Setting.get('collector_corporate_ranges') or ''
except Exception:
return []
ranges = []
for cidr in str(raw).split(','):
cidr = cidr.strip()
if not cidr:
continue
try:
ranges.append(ipaddress.ip_network(cidr, strict=False))
except ValueError:
continue
return ranges
def _pick_primary(interfaces, ranges):
"""Which reported NIC is the PC's address.
In order: inside a configured corporate range; else the one the bay says
carries the default route; else the first. The last is a guess and is the
reason configuring the ranges is worth doing - it is what a site sets so
the answer stops being a guess.
"""
def addr(entry):
try:
return ipaddress.ip_address((entry.get('ipaddress') or '').strip())
except ValueError:
return None
if ranges:
for entry in interfaces:
parsed = addr(entry)
if parsed and any(parsed in net for net in ranges):
return entry
for entry in interfaces:
if entry.get('isdefaultroute'):
return entry
return interfaces[0] if interfaces else None
def _record_interfaces(comp, interfaces):
"""Write one Communication row per reported NIC, flagging exactly one.
Keyed on the MAC where the bay sent one, because an address moves between
adapters and a MAC does not. Falls back to the address for a bay that
reports no MAC, which is what the single-ipaddress payload looks like.
Rows this PC has that the bay did NOT report are left alone: a NIC that is
down at check-in time is not evidence the record is wrong, and hand-entered
detail (gateway, DNS, serial settings) lives on these same rows.
"""
# Imported here for the same reason the caller does: these come through the
# shopdb.api contract surface, not from core internals.
from shopdb.api import Communication, CommunicationType, db
warnings = []
ip_comtype = CommunicationType.query.filter_by(comtype='IP').first()
if not ip_comtype:
return ['no IP communication type seeded; network not recorded']
usable = [entry for entry in interfaces if (entry.get('ipaddress') or '').strip()]
if not usable:
return warnings
primary = _pick_primary(usable, _corporate_ranges())
existing = Communication.query.filter_by(assetid=comp.assetid).all()
bymac = {(c.macaddress or '').strip().lower(): c for c in existing if c.macaddress}
byip = {(c.ipaddress or '').strip(): c for c in existing if c.ipaddress}
touched = []
for entry in usable:
addr = (entry.get('ipaddress') or '').strip()
mac = (entry.get('macaddress') or '').strip()
row = bymac.get(mac.lower()) if mac else None
if row is None:
row = byip.get(addr)
if row is None:
row = Communication(assetid=comp.assetid, comtypeid=ip_comtype.comtypeid)
db.session.add(row)
row.ipaddress = addr
if mac:
row.macaddress = mac
touched.append((row, entry is primary))
# Exactly one primary, and only among rows this report touched. An untouched
# row keeps its flag only if nothing here claimed it, which is what stops a
# check-in from silently demoting an address somebody set by hand.
for row, isprimary in touched:
row.isprimary = isprimary
if any(isprimary for _, isprimary in touched):
claimed = {id(row) for row, _ in touched}
for row in existing:
if id(row) not in claimed:
row.isprimary = False
return warnings
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,
ConfigurationPolicy, ConfigurationPolicyAction, BusinessPolicy, BusinessPolicyItem]
def init_app(self, app: Flask, db_instance) -> None:
"""Initialize plugin with Flask app."""
self._register_import_hook()
logger.info(f"Computers plugin initialized (v{self.meta.version})")
# -- bulk-import machine link -------------------------------------------
def _register_import_hook(self):
"""Own a `machineassetnumber` column on computers.csv.
A site with a spreadsheet knows "this PC drives machine 3015" and wants
to write that number in the PC row, not hand-build an assetrelationships
sheet keyed on ids it never sees. The core importer stays ignorant of
what the column means: it accepts the header, keeps it out of the
computers write, and hands the value to this hook after each PC row is
written, so the PC's asset id exists. Registration is idempotent, so a
plugin reload or a test re-init does not double-fire.
"""
from shopdb.api import register_import_hook
register_import_hook(
'computers',
{'machineassetnumber':
'the asset number of the machine this PC drives (from '
'machines.csv); builds the PC-controls-machine link. A number '
'that has no machine, or one already driven by another PC, is '
'reported as a warning and skipped - the PC still imports.'},
self._import_machine_link)
def _import_machine_link(self, rowvalues, virtualvalues, line):
"""Build a PC->machine controls link from an import row's machine number.
Runs inside the importer's transaction - a dry run rolls it back - so it
reuses the collector's mature _sync_machine_link rather than inventing a
second linker, but under the import origin (so an import never archives
a collector's rows and vice versa) and in preview mode (so a bulk import
never emails an alert per row; the findings are the report's amber
warnings instead).
Returns RowWarning list, never raises: a machine number that resolves to
no asset is an amber note beside a PC that still imports, not a failed
file. Every warning names the PC, the number, and the one fix.
"""
from shopdb.api import RowWarning
machinenumber = (virtualvalues.get('machineassetnumber') or '').strip()
if not machinenumber:
return []
assetid = rowvalues.get('assetid')
comp = Computer.query.filter_by(assetid=assetid).first() if assetid else None
if comp is None:
# The PC row was written but its Computer extension is not readable
# here - nothing to link to. Amber, not fatal.
return [RowWarning(
line, 'machineassetnumber',
reason='PC row not found to link to machine {!r}'.format(
machinenumber),
action='re-run the import; if it recurs, check the computers '
'row imported',
target=machinenumber)]
hostname = comp.hostname or (assetid and str(assetid)) or '?'
# IN-FILE FAN-OUT GUARD. Two rows in one computers.csv both naming the
# same machine is a spreadsheet mistake - a copied cell, a wrong
# workstation number - not a swap. But _incumbent_has_yielded reads a
# freshly-imported PC (no lastreporteddate) as having yielded, so
# without this the SECOND row would silently archive the first row's
# link and steal the machine, and the file would import clean with the
# wrong PC on the bay. So the first claim on a machine this run wins and
# is remembered; a later row for the same machine is warned and skipped.
# The claim set lives on db.session.info, which Flask-SQLAlchemy resets
# at request teardown, so it is scoped to exactly this one import and
# never leaks into the next.
machine = self._asset_by_number(machinenumber)
if machine is not None and machine.assetid == assetid:
# The number names this very PC. Happens on a PC first seen before
# the machine-number convention, whose own asset number IS the
# machine number. Skip - a PC does not control itself.
return [RowWarning(
line, 'machineassetnumber',
reason='machine number {!r} is this PC\'s own asset number; '
'not linked'.format(machinenumber),
action='give this PC a hostname-based asset number, or point '
'machineassetnumber at the machine it drives',
subject=hostname, target=machinenumber)]
if machine is not None:
# A location-only machine is a virtual bay marker, not a real
# machine, so nothing controls it. Skip rather than build a link
# that would read as a PC driving a location.
skip = self._location_only_skip(
machine, hostname, machinenumber, line)
if skip is not None:
return [skip]
if machine is not None and machine.assetid != assetid:
# ALREADY-CONTROLS-DIFFERENT. Unlike the collector - which reads a
# new machine number as a PC moved to a new bay and silently
# re-bays it - an import naming a machine when this PC already
# actively controls a DIFFERENT one is far more likely a wrong
# workstation number in the sheet than a real move. Warn and skip
# rather than archive a link a person may have set on purpose.
other = self._active_controlled_machine(assetid, machine.assetid)
if other is not None:
return [RowWarning(
line, 'machineassetnumber',
reason='{} already controls machine {}; not moved to {}'
.format(hostname, other, machine.assetnumber),
action='if this PC really moved, clear the existing link '
'first; otherwise correct the machine number on '
'this row',
subject=hostname, target=machinenumber)]
if machine is not None and machine.assetid != assetid:
claimed = db.session.info.setdefault('import_machine_claims', {})
# Keyed by machine asset id -> (claimant PC asset id, its hostname).
# The id is what the guard compares on (the same PC re-stating its
# own link is not a conflict); the hostname is only for the message.
firstclaimant = claimed.get(machine.assetid)
if firstclaimant is not None and firstclaimant[0] != assetid:
return [RowWarning(
line, 'machineassetnumber',
reason='machine {} was already claimed by {} earlier in '
'this import; {} not linked'.format(
machine.assetnumber, firstclaimant[1], hostname),
action='a machine drives one PC - correct the machine '
'number on one of these rows so each PC names its '
'own machine',
subject=hostname, target=machinenumber)]
claimed[machine.assetid] = (assetid, hostname)
# _sync_machine_link appends plain strings to its warnings list. Collect
# them and translate to structured, actionable RowWarnings so the UI can
# show the PC, the machine number, and the corrective step.
rawwarnings = []
self._sync_machine_link(
comp, machinenumber, rawwarnings,
origin_label=MACHINE_LINK_IMPORT_ORIGIN, preview=True)
out = []
for text in rawwarnings:
out.append(RowWarning(
line, 'machineassetnumber',
reason=text,
action=self._machine_link_action(text, machinenumber),
subject=hostname, target=machinenumber))
return out
@staticmethod
def _active_controlled_machine(pcassetid, excludeassetid):
"""Asset number of a machine this PC actively controls under the import
origin, other than excludeassetid, or None.
Scoped to MACHINE_LINK_IMPORT_ORIGIN so an import only reasons about its
own links: a collector link or a hand-made one is the collector's or a
person's to move, not the import's to block on.
"""
from shopdb.api import AssetRelationship, RelationshipType, Asset
controls = RelationshipType.query.filter_by(
relationshiptype='controls').first()
if not controls:
return None
rel = AssetRelationship.query.filter(
AssetRelationship.sourceassetid == pcassetid,
AssetRelationship.relationshiptypeid == controls.relationshiptypeid,
AssetRelationship.label == MACHINE_LINK_IMPORT_ORIGIN,
AssetRelationship.targetassetid != excludeassetid,
AssetRelationship.isactive.is_(True),
).first()
if rel is None:
return None
asset = db.session.get(Asset, rel.targetassetid)
return asset.assetnumber if asset else str(rel.targetassetid)
@staticmethod
def _location_only_skip(machine, hostname, machinenumber, line):
"""A RowWarning if the resolved asset is a location-only marker, else None.
A machine flagged islocationonly is a virtual bay, not a real machine,
so a PC does not control it. Guarded behind the machines plugin being
installed - a site without it cannot have a location-only machine, and
the import must not fail because the model is absent.
"""
from shopdb.api import RowWarning
try:
from plugins.machines.models import Machine
except ImportError:
return None
row = Machine.query.filter_by(assetid=machine.assetid).first()
if row is None or not row.islocationonly:
return None
return RowWarning(
line, 'machineassetnumber',
reason='machine {} is a location-only marker, not a real machine; '
'not linked'.format(machine.assetnumber),
action='point machineassetnumber at a real machine, or clear it '
'for this PC',
subject=hostname, target=machinenumber)
@staticmethod
def _machine_link_action(text, machinenumber):
"""The one corrective step for a machine-link warning, by its shape."""
lowered = text.lower()
if 'no asset for machine number' in lowered:
return ('create the machine asset with number {!r} (import '
'machines.csv first), or correct the number on this PC '
'row'.format(machinenumber))
if 'still held by' in lowered:
return ('confirm which PC drives this machine; retire or move the '
'old PC off In Use to hand the machine over')
if 'taken over from' in lowered:
return ('expected on a PC swap; set the old PC to Retired, '
'Inventory or In Repair')
if 'relationship type missing' in lowered:
return 'run flask seed reference-data, then re-import'
return 'review this PC and its machine number'
# -- 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'},
# The DEVICE hanging off this PC, named by the enrollment file
# asset-id.txt. One file for every bay that has no NTLARS/eDNC
# MachineNo registry to identify it, and it does NOT say what
# kind of device it is: the pc-type already does, through
# SUBORDINATE_DEVICE_MAP.
#
# SEPARATE from machinenumber on purpose: "which bay is this"
# and "which device is this" are different facts, and
# machinenumber is what GE-Enforce TargetMachineNumbers gates
# on - repointing it at a device would silently stop every
# bay-gated manifest entry matching.
#
# Named deviceid rather than assetid because assetid is this
# contract's own RESPONSE field for a PC's integer primary key.
'deviceid': {'type': 'string'},
# Shipped in 0.12.0, before the file was generalised. Accepted
# as an alias so a bay staged with measuringtool-id.txt keeps
# reporting; deviceid wins when both arrive.
'measuringtoolid': {'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'},
# Kept for bays on an older collector. A payload carrying
# both wins on `interfaces`.
'ipaddress': {'type': 'string'},
# EVERY routable IPv4 the bay can see, and no opinion about
# which one matters. Choosing was the bay's job and it was the
# wrong place for it: the bay cannot know a site's addressing,
# so a CMM with a 10Gbit link to its controller reported the
# machine-side address as the PC's, and the corporate address
# was never sent at all. The server picks the primary now,
# using collector_corporate_ranges, and records the rest.
'interfaces': {
'type': 'array',
'items': {
'ipaddress': 'string',
'macaddress': 'string',
'adaptername': 'string',
'isdefaultroute': 'boolean',
},
'description': ('All connected hardware NICs with a '
'routable IPv4. Order carries no meaning.'),
},
'vendorname': {'type': 'string'},
'modelnumber': {'type': 'string'},
'osname': {'type': 'string'},
'installedsoftware': {
'type': 'array',
'items': {'name': 'string', 'version': 'string'},
},
# RETIRED: 'defaultprinter' and 'printers' used to live here and
# wrote PC->printer assignment rows from a report. That is the
# one thing this path must never do - see _RETIRED_PRINTER_KEYS.
# A client still sending them is warned, not rejected.
'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 get_settings_cards(self) -> List[dict]:
"""The asset reporter's own settings page.
It is a settings card rather than a docs page because it does two things
an operator needs at the same moment: name this site's routable ranges,
and download the reporter that carries them.
"""
return [
{
'group': 'Computers',
'to': '/settings/collector',
'icon': 'download',
'title': 'Asset reporter',
'description': 'Download the collector script stamped with this '
'site\'s URL and ranges',
'position': 26,
},
] + self._policy_settings_cards()
def _policy_settings_cards(self):
"""Policy admin cards, shown only while the feature flag is on.
Gating the cards on the switch keeps the settings menu honest: the
Configuration/Business Policy screens appear exactly when the feature is
enabled and vanish on rollback, matching the 404 the routes return.
"""
if not self._configpolicies_enabled_safe():
return []
return [
{
'group': 'Computers',
'to': '/settings/configuration-policies',
'icon': 'shield',
'title': 'Configuration policies',
'description': 'Reusable rules binding endpoint actions to PC '
'types and applications',
'position': 27,
},
{
'group': 'Computers',
'to': '/settings/site-policies',
'icon': 'layers',
'title': 'Site policies',
'description': 'Ordered collections of configuration policies '
'assigned by priority',
'position': 28,
},
]
def get_settings_defaults(self) -> List[dict]:
"""Settings this plugin owns.
The framework seeds these at install, at enable, and on every
`flask plugin upgrade-all`, so a key added in a later version reaches a
site that installed an earlier one.
"""
return [
{
# Which addresses are THIS site's routable network. The
# collector reports every NIC and takes no view; this decides
# which one is the PC's address.
#
# Empty falls back to the NIC the bay says carries the default
# route, then to the first reported - both guesses. A CMM with
# a 10Gbit link to its controller is the case that shows why:
# that link can carry a default route too, and the machine-side
# address then wins over the corporate one.
#
# Comma-separated CIDRs, e.g. '10.48.0.0/16,10.20.0.0/23'.
'key': 'collector_corporate_ranges',
'value': '',
'valuetype': 'string',
'category': 'computers',
'description': ('Comma-separated CIDRs naming this site routable '
'network. The collector reports every NIC; this '
'picks which is the primary address.'),
},
{
'key': 'computers_quietreporthours',
'value': '24',
'valuetype': 'integer',
'category': 'computers',
'description': 'Hours without a collector report before a PC '
'is listed as not reporting on the dashboard.',
},
{
'key': 'computers_routableranges',
'value': '',
'valuetype': 'string',
'category': 'computers',
'description': 'Comma-separated CIDRs naming this site\'s '
'routable ranges, e.g. 10.20.0.0/23,10.21.4.0/26. '
'Stamped into the collector script this server '
'generates, so a bay with a controller NIC and a '
'corporate NIC reports the right one. Blank uses '
'the NIC carrying the default route, which needs '
'no knowledge of a site\'s addressing.',
},
{
'key': 'computers_machinelink_alerts',
'value': 'false',
'valuetype': 'boolean',
'category': 'computers',
'description': 'Email and webhook alerts when a PC takes over '
'a machine or claims one another PC still runs. '
'Off while machine numbers are shared between '
'PCs; the collector response still warns.',
},
{
'key': 'computers_configpolicies_enabled',
'value': 'false',
'valuetype': 'boolean',
'category': 'computers',
'description': 'Master switch for RMM-style Configuration and '
'Site Policies. Off by default: models exist '
'but the CRUD routes, WebUI, and endpoint '
'enforcement stay inert so the whole feature can '
'be rolled back without a schema change.',
},
]
def _configpolicies_enabled(self):
"""Whether the Configuration/Business Policy feature is turned on.
Off by default. This is the single exit gate for the policy scope: the
tables and models load regardless (SQLAlchemy needs them), but every
route, panel, and enforcement path checks this first and no-ops while it
is false, so the feature can be disabled without a migration.
"""
return _policy_flag_enabled()
def _configpolicies_enabled_safe(self):
"""Flag read that never raises when the settings table is absent.
get_settings_cards can be called in contexts without a database (the
plugin contract tests build hooks against a bare app), so a missing
settings table must read as "off" rather than propagate.
"""
try:
return _policy_flag_enabled()
except Exception:
return False
def _machinelink_alerts_enabled(self):
"""Whether the PC-to-machine alerts may be sent.
Off by default, deliberately. A site may legitimately run several PCs
on one machine number - part markers do at the reference site - and there
both the handover and the contested case fire on normal, correct data,
which is noise rather than news. Turn it on at a site where a machine
number means exactly one PC.
The links, the warnings in the collector response, and the archived
history all continue regardless. Only the sending is gated.
"""
from shopdb.api import Setting
setting = Setting.query.filter_by(
key='computers_machinelink_alerts').first()
if not setting:
return False
return (setting.value or '').strip().lower() in ('true', '1', 'yes')
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')
# The machine number identifies the MACHINE, not this PC. It is
# reported so the PC can be related to its machine; it is deliberately
# NOT used as the PC's assetnumber. assets.assetnumber is uniquely
# indexed and the machine already owns that value, so assigning it here
# raised "Duplicate entry '3015' for key 'ix_assets_assetnumber'" and
# returned 500 to the bay - forever, since every retry did the same
# thing. The convention this restores is what the data already shows:
# of 289 computers, none has a numeric assetnumber and 214 use their
# 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=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'
# NOTE: an existing PC's assetnumber is left alone. Overwriting it with
# the machine number renamed the PC onto the machine's identifier, which
# either collided with the unique index or silently changed how that PC
# is identified everywhere else.
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)')
# NETWORK. A bay reports every NIC it can see; THIS decides which one
# is the PC's address, because only the server knows the site's
# addressing. A CMM whose 10Gbit link to its controller sits on
# 100.0.0.1 used to report that as the PC's address and never send the
# corporate one at all - the bay had no way to tell them apart.
interfaces = payload.get('interfaces') or []
if not interfaces and payload.get('ipaddress'):
# An older bay sends one address and no opinion beyond it.
interfaces = [{'ipaddress': payload['ipaddress']}]
if interfaces:
warnings.extend(_record_interfaces(comp, interfaces))
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)
# Part-marker PCs get a marker asset of their own, which is what the
# machine number then hangs off. Done BEFORE the machine link because a
# marker PC must not also claim the operation directly: several markers
# serve one operation number, so direct claims would fight over it. The
# marker is partof the operation and control propagates along that rail.
# asset-id.txt, falling back to the 0.12.0 field name. One value feeds
# BOTH device paths: the file names the device and nothing else, so
# which sync consumes it follows from the pc-type.
deviceid = (payload.get('deviceid')
or payload.get('measuringtoolid') or '').strip() or None
partmarkers = self._sync_partmarker(comp, pctype, machinenumber,
warnings, deviceid=deviceid)
# PC -> machine link from the reported machine number. Only a PC that
# is NOT a marker PC (None above) takes it: a marker PC reaches its
# operation through `partof` on the marker, never by claiming it.
if partmarkers is None:
machinelinks = self._sync_machine_link(comp, machinenumber, warnings)
else:
machinelinks = []
# Printer assignment is NOT collected. See _warn_retired_printer_keys.
self._warn_retired_printer_keys(payload, warnings)
printerlinks = []
# 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,
measuringtoolid=deviceid,
machinenumber=machinenumber)
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,
'machinelinks': machinelinks,
# `or []`: None is the internal "not a marker PC" signal and
# has no business in the response, where it would read as a
# different state from "a marker PC that linked nothing".
'partmarkers': partmarkers or [],
},
}
# -- 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 _warn_retired_printer_keys(self, payload, warnings):
"""Refuse to turn a report into an assignment, out loud.
RETIRED 2026-08-24. This used to be _sync_printer_links, which wrote
'defaultprinter' and 'connectedto' rows straight from a collector
payload. That contradicted the rule the whole printer feature rests on:
ASSIGNED and OBSERVED never mix. A bay that installed the wrong printer
would have made itself right simply by reporting it, and enforcement
would stop meaning anything.
Worse than dormant, it was ADVERTISED: the collector schema listed both
keys and COLLECTOR-INTEGRATION.md's example client set defaultprinter,
while the same document promised on another page that nothing on this
path writes an assignment row. The document contradicted itself and the
code sided with the wrong half.
Observed queues have their own path (POST /api/collector/printers) and
become assignments only when a person asks, through
seed-from-observed after reviewing the comparison.
WARNS RATHER THAN REJECTS. Unknown keys are not validated here, so a
client written against the old docs keeps pushing successfully; only its
printer keys stop doing anything. Silently ignoring them would leave an
integrator believing assignments were being written.
"""
sent = [key for key in ('defaultprinter', 'printers') if key in payload]
if sent:
warnings.append(
'%s ignored: a collector report never sets printer assignment. '
'Report queues to POST /api/collector/printers, then adopt them '
'with POST /api/printers/assignments/seed-from-observed.'
% ' and '.join(sent))
def _sync_machine_link(self, comp, machinenumber, warnings,
origin_label=MACHINE_LINK_ORIGIN, preview=False):
"""Link a PC to the machine it drives, from the reported machine number.
Until this existed the machine number was collected and then discarded,
so a replaced PC never took over its bay: the retired PC kept the link
and the new one got none. Everything that walks PC->machine (the
warranty machine column, the DNC info card) therefore pointed at
hardware that had been pulled out.
ARCHIVES, never deletes. A superseded link stays with isactive=False so
"which PC ran 3015 in June" is still answerable. Status on the old PC is
deliberately NOT changed: the collector cannot tell whether it was
shelved, broken or re-imaged for another bay, and guessing would
overwrite whatever a person deliberately set.
A second PC reporting a machine another PC already holds is a CLAIM, not
proof of replacement. A PC imaged on the bench for machine 3010 carries
that number before it ever reaches the floor, and treating the claim as
a handover made the two PCs trade the link back and forth at collector
cadence, alerting on every pass. The incumbent therefore keeps the
machine while it is still alive, and the challenger is recorded as a
dormant link. See _incumbent_has_yielded for what alive means.
origin_label separates provenance: the collector stamps
MACHINE_LINK_ORIGIN, a CSV import stamps MACHINE_LINK_IMPORT_ORIGIN, and
each only ever archives its own rows so the two sources coexist.
preview=True is for a dry-run import: DB writes still happen in the
caller's transaction (which it rolls back), but the alert fan-out - the
one side effect a rollback cannot undo - is suppressed so a preview
never emails anyone.
"""
from shopdb.api import AssetRelationship, RelationshipType, Asset
pcasset = comp.asset if comp else None
if not machinenumber or not pcasset:
return []
controls = RelationshipType.query.filter_by(
relationshiptype='controls').first()
if not controls:
warnings.append("'controls' relationship type missing; "
'run flask seed reference-data')
return []
machine = self._asset_by_number(machinenumber)
if not machine:
# Reported a machine ShopDB does not know. Warn rather than invent
# an asset: a mistyped number would create a machine nobody can
# account for.
warnings.append(
'no asset for machine number {!r}; PC not linked'.format(
machinenumber))
return []
if machine.assetid == pcasset.assetid:
return []
# This PC's own collector links: the one for the reported machine is
# settled below, any other is archived (the PC moved bays).
mine = AssetRelationship.query.filter(
AssetRelationship.sourceassetid == pcasset.assetid,
AssetRelationship.relationshiptypeid == controls.relationshiptypeid,
AssetRelationship.label == origin_label,
).all()
link = None
for rel in mine:
if rel.targetassetid == machine.assetid:
link = rel
else:
rel.isactive = False
# The unique constraint on assetrelationships is (source, target, type)
# and does NOT include label, so an import (import:machine) and the
# collector (collector:machine) cannot each hold their OWN row for the
# same PC->machine control. If our label-scoped lookup found nothing but
# a row for this exact triple exists under the other provenance, adopt
# it rather than insert a duplicate that would violate the constraint.
if link is None:
link = AssetRelationship.query.filter_by(
sourceassetid=pcasset.assetid,
targetassetid=machine.assetid,
relationshiptypeid=controls.relationshiptypeid).first()
# Whoever actively holds this machine now, if it is not this PC.
held = AssetRelationship.query.filter(
AssetRelationship.targetassetid == machine.assetid,
AssetRelationship.relationshiptypeid == controls.relationshiptypeid,
AssetRelationship.label == origin_label,
AssetRelationship.sourceassetid != pcasset.assetid,
AssetRelationship.isactive.is_(True),
).all()
blocking = [rel for rel in held
if not self._incumbent_has_yielded(rel.sourceassetid)]
if blocking:
# Contested. The incumbent is still reporting and still In Use, so
# this is a claim on a bay it has not taken over yet. Record the
# claim dormant and leave the live link where it is; the dormant row
# is also the marker that says this was already announced, which is
# what stops an alert on every report.
names = ', '.join(self._assetname(rel.sourceassetid)
for rel in blocking)
warnings.append(
'machine {} is still held by {}; {} claim recorded but not '
'linked'.format(machine.assetnumber, names, comp.hostname))
if link is None:
db.session.add(AssetRelationship(
sourceassetid=pcasset.assetid,
targetassetid=machine.assetid,
relationshiptypeid=controls.relationshiptypeid,
label=origin_label,
isactive=False))
if not preview:
self._alert_machine_contested(names, comp, machine)
else:
link.isactive = False
return [{'assetid': machine.assetid,
'machinenumber': machine.assetnumber,
'contestedby': names,
'superseded': 0}]
# Uncontested: take the machine.
if link is None:
link = AssetRelationship(
sourceassetid=pcasset.assetid,
targetassetid=machine.assetid,
relationshiptypeid=controls.relationshiptypeid,
label=origin_label,
isactive=True)
db.session.add(link)
else:
link.isactive = True
# Anything still holding it has yielded: gone quiet, or taken off In
# Use by a person. That is a replacement, so archive and say so.
for rel in held:
rel.isactive = False
oldname = self._assetname(rel.sourceassetid)
warnings.append(
'machine {} was taken over from {}; check that PC'.format(
machine.assetnumber, oldname))
if not preview:
self._alert_pc_superseded(oldname, comp, machine)
return [{'assetid': machine.assetid,
'machinenumber': machine.assetnumber,
'superseded': len(held)}]
def _is_device_of_type(self, asset, spec):
"""Is this asset the kind of device the pc-type's spec describes?
asset-id.txt names a device and says nothing about its type, so the
type is what proves the file points somewhere sensible. A measuring
tool is checked against the measuring-tool set; anything else is
checked against its machine type, which is where a Part Marker lives.
"""
if asset is None:
return False
if spec.get('assettype') == 'measuring_tool':
return asset.assetid in self._measuringtool_assetids()
try:
from plugins.machines.models import Machine, MachineType
except ImportError:
return False
row = (db.session.query(MachineType.machinetype)
.join(Machine, Machine.machinetypeid == MachineType.machinetypeid)
.filter(Machine.assetid == asset.assetid).first())
return bool(row) and row[0] == spec['typename']
def _ensure_device_rows(self, deviceasset, spec, pcasset, controls, label,
warnings, active=True):
"""Get-or-create the extension row and the PC -> device control link.
Shared by the named-device and minted paths: an adopted asset may have
no extension row, and the control link may already exist under another
label, so both are get-or-create rather than insert.
"""
from shopdb.api import AssetRelationship
try:
from plugins.machines.models import Machine, MachineType
except ImportError:
warnings.append('machines plugin unavailable; {} device skipped'
.format(spec['typename']))
return
if not Machine.query.filter_by(assetid=deviceasset.assetid).first():
devicetype = MachineType.query.filter_by(
machinetype=spec['typename']).first()
if not devicetype:
devicetype = MachineType(machinetype=spec['typename'],
description=spec.get('description'))
db.session.add(devicetype)
db.session.flush()
db.session.add(Machine(assetid=deviceasset.assetid,
machinetypeid=devicetype.machinetypeid))
controlrow = AssetRelationship.query.filter_by(
sourceassetid=pcasset.assetid,
targetassetid=deviceasset.assetid,
relationshiptypeid=controls.relationshiptypeid).first()
if controlrow is not None:
controlrow.isactive = active
if not controlrow.label:
controlrow.label = label
else:
db.session.add(AssetRelationship(
sourceassetid=pcasset.assetid,
targetassetid=deviceasset.assetid,
relationshiptypeid=controls.relationshiptypeid,
label=label,
isactive=active))
def _device_incumbents_yield(self, deviceasset, pcasset, label, warnings):
"""Settle a device already controlled by a DIFFERENT PC.
A physical instrument or marker hangs off one PC. Before this, both
paths only ever looked at links whose source was THIS PC, so a bay
naming a device another PC still holds produced two active `controls`
rows with no warning - the replaced PC kept its link forever, and a
file copied to a second bay claimed the device silently.
The machine link settled this years ago and this reuses its rule rather
than inventing a second one: an incumbent that has gone quiet past the
claim window, or been moved off In Use, has yielded, and its link is
ARCHIVED (never deleted, so "which PC drove this in June" stays
answerable). An incumbent still alive keeps the device, and the
challenger is recorded dormant instead of contesting a link that can
only have one holder.
Returns True when this PC may hold the device actively.
"""
from shopdb.api import AssetRelationship, RelationshipType, Asset
controls = RelationshipType.query.filter_by(
relationshiptype='controls').first()
if not controls or not deviceasset or not pcasset:
return True
# EVERY collector-owned holder, not just ones under this label. On a
# CMM the instrument IS the reported bay, so the incumbent's link is the
# machine sync's row under MACHINE_LINK_ORIGIN. Filtering on our own
# label found no incumbent there, handed the challenger an active link,
# and left two live holders of one instrument - each invisible to the
# other because they were labelled differently.
#
# A row made by hand carries neither label and is deliberately NOT
# counted: a person's link is not the collector's to archive.
ourlabels = (label, MACHINE_LINK_ORIGIN, MACHINE_LINK_IMPORT_ORIGIN,
MEASURINGTOOL_LINK_ORIGIN, PARTMARKER_LINK_ORIGIN)
others = AssetRelationship.query.filter(
AssetRelationship.targetassetid == deviceasset.assetid,
AssetRelationship.relationshiptypeid == controls.relationshiptypeid,
AssetRelationship.label.in_(ourlabels),
AssetRelationship.isactive.is_(True),
AssetRelationship.sourceassetid != pcasset.assetid,
).all()
if not others:
return True
holding = []
for rel in others:
if self._incumbent_has_yielded(rel.sourceassetid):
rel.isactive = False
else:
holder = db.session.get(Asset, rel.sourceassetid)
holding.append(holder.assetnumber if holder else rel.sourceassetid)
if holding:
warnings.append(
'{} is still controlled by {}; recorded but not activated'
.format(deviceasset.assetnumber, ', '.join(str(h) for h in holding)))
return False
return True
def _sync_partmarker(self, comp, pctype, machinenumber, warnings,
deviceid=None):
"""Give a part-marker PC a marker asset of its own, under its operation.
Several Telesis markers serve one operation number - 0613, 0615 and
WJPRT each have more than one - so treating the operation as the marker
collapsed separate devices into one record. Their configs, which differ
by COM port, then overwrote each other in the backup history, and no
question about an individual marker (how many are there, which port,
which one failed) could be asked at all.
One marker per PC. The marker is a machine asset of type Part Marker,
the PC `controls` it, and the marker is `partof` the operation it
serves.
IDENTITY COMES FROM asset-id.txt FIRST, because everything else here is
derived from the PC and a PC is not permanent. Reuse looks for a prior
link from THIS PC asset and adoption looks up `<PC number>-PARTMARKER`,
so both survive a re-image and neither survives a PC SWAP: a new
hostname is a new PC asset with no prior link and a predicted number
that has never existed, so the same physical Telesis unit gets a second
record while the first keeps its config and backup history under a dead
PC's name. That is the failure the metrology path already learned from,
where minting from the hostname left 43 instruments shadowed by twins.
That last rail is why the PC does not also claim the operation directly:
`controls` propagates through `partof` (seeded in reference-data), so
control of the operation follows from controlling its marker, and two
markers on one operation no longer contest a link that can only have one
holder.
Returns [] for any PC that does not drive a marker, which leaves the
ordinary machine link to run.
"""
from shopdb.api import AssetRelationship, RelationshipType, Asset
from .pctypemap import subordinate_device_for
spec = subordinate_device_for(pctype)
if not spec or not spec.get('partof') or not comp or not comp.asset:
# Only a device that FILES UNDER an operation goes through here.
# A measuring tool is a subordinate device too, but it does not
# share a machine number, so it keeps the simpler path.
#
# NONE, not []. None is the ONLY answer that means "not a marker PC,
# run the ordinary machine link". Every other exit below means "this
# IS a marker PC" - possibly one that linked nothing - and returning
# [] there let the caller fall through to the machine link, which
# for a marker PC points at the OPERATION. That is the direct claim
# this whole path exists to prevent: several markers serve one
# operation, so two marker PCs would contest a link that can only
# have one holder, while the warning said "not linked".
return None
pcasset = comp.asset
controls = RelationshipType.query.filter_by(
relationshiptype='controls').first()
if not controls:
warnings.append("'controls' relationship type missing; "
'run flask seed reference-data')
return []
try:
from plugins.machines.models import Machine, MachineType
except ImportError:
warnings.append('machines plugin unavailable; {} device skipped'
.format(spec['typename']))
return []
label = spec['label']
# --- 1. an explicitly named device wins over everything -------------
#
# NAMING A DEVICE IS A COMMITMENT. If the bay says which device it
# drives and that name does not resolve, the answer is to link nothing
# and say so - NOT to fall through and mint <HOST>-PARTMARKER, which is
# the hostname-derived twin this whole path exists to prevent. The
# warning used to be true about the typo'd number and false about the
# twin: it warned, then minted anyway.
#
# Silence still means "work it out", so a bay with no file keeps the
# reuse-then-mint behaviour it has always had.
named = (deviceid or '').strip()
namedasset = None
if named:
candidate = self._asset_by_number(named)
if candidate is None:
# Warn rather than invent: a typo must not mint a phantom
# device that nobody can account for.
warnings.append(
'no asset for device {!r}; not linked'.format(named))
return []
if not self._is_device_of_type(candidate, spec):
# The name resolved, but not to this pc-type's device. Refuse
# and say which asset it hit, rather than filing an unrelated
# asset under a device label the collector also owns.
warnings.append(
'asset {!r} is not a {}; not linked'.format(
candidate.assetnumber, spec['typename']))
return []
namedasset = candidate
existing = AssetRelationship.query.filter(
AssetRelationship.sourceassetid == pcasset.assetid,
AssetRelationship.relationshiptypeid == controls.relationshiptypeid,
AssetRelationship.label == label,
).all()
reuse = next((rel for rel in existing if rel.isactive), None) \
or (existing[0] if existing else None)
if namedasset is not None:
# The file is authoritative FOR THIS PC, but the device may still be
# held by another one. Settle that first: a yielded incumbent is
# archived, a live one keeps it and this link is recorded dormant.
mayhold = self._device_incumbents_yield(
namedasset, pcasset, label, warnings)
# The file is authoritative. A PC that previously minted its own
# marker and is now told the real one keeps only the named link;
# the stale link is archived by the one-marker-per-PC sweep below.
markerasset = namedasset
reuse = next((rel for rel in existing
if rel.targetassetid == namedasset.assetid), None)
if reuse is not None:
reuse.isactive = True
self._ensure_device_rows(markerasset, spec, pcasset, controls,
label, warnings, active=mayhold)
elif reuse:
# Re-check incumbency HERE too, not only on the named path. A
# challenger recorded dormant keeps its row, so the cycle after its
# asset-id.txt goes missing found that row as `reuse` and flipped it
# active with no incumbent check and no warning - two live holders,
# arrived at by deleting a file. _sync_machine_link never had this
# hole because it re-checks unconditionally.
markerasset = db.session.get(Asset, reuse.targetassetid)
reuse.isactive = self._device_incumbents_yield(
markerasset, pcasset, label, warnings)
else:
coretype = AssetType.query.filter_by(
assettype=spec['assettype']).first()
if not coretype:
warnings.append('{} asset type missing; {} skipped'.format(
spec['assettype'], spec['typename']))
return []
devicetype = MachineType.query.filter_by(
machinetype=spec['typename']).first()
if not devicetype:
devicetype = MachineType(machinetype=spec['typename'],
description=spec.get('description'))
db.session.add(devicetype)
db.session.flush()
hostname = comp.hostname
assetnumber = '{}-{}'.format(
pcasset.assetnumber or hostname, spec['suffix'])
# ADOPT AN EXISTING ASSET OF THAT NUMBER before minting. The number
# is derived from the PC, so it is entirely predictable and may
# already exist - left by an earlier run whose link was archived, or
# created by a person. assetnumber is unique, so inserting a second
# one raised and the whole report 500'd, on every cycle, forever.
#
# Deliberately NOT filtered on isactive: an inactive asset still
# holds the number, and it is the constraint that decides this.
markerasset = Asset.query.filter(
db.func.lower(Asset.assetnumber) == assetnumber.lower()).first()
if markerasset is None:
markerasset = Asset(
assetnumber=assetnumber,
name='{} ({})'.format(spec['typename'], hostname),
assettypeid=coretype.assettypeid,
statusid=1)
db.session.add(markerasset)
db.session.flush()
self._ensure_device_rows(markerasset, spec, pcasset, controls,
label, warnings)
# One marker per PC: archive any other collector marker link.
for rel in existing:
if rel is not reuse and rel.isactive:
rel.isactive = False
operation = self._link_marker_to_operation(
markerasset, machinenumber, pcasset, label, warnings)
return [{'assetid': markerasset.assetid,
'assetnumber': markerasset.assetnumber,
'operationassetid': operation}]
def _link_marker_to_operation(self, markerasset, machinenumber, pcasset,
label, warnings):
"""Make a marker `partof` the operation whose number its PC reports.
Unlike the PC-to-machine link this does NOT contest: an operation can
hold any number of markers, which is the whole point. Moving a marker to
another operation archives the old membership rather than deleting it,
so where a marker used to live stays answerable.
Refuses to file a marker under the reporting PC. A PC first seen before
the machine-number fix was created with the machine number as its OWN
asset number, and that is deliberately never overwritten, so looking up
the number can return the PC itself. Filing the marker partof its own PC
would read, on the machine page, as the PC being the operation.
"""
from shopdb.api import AssetRelationship, RelationshipType, Asset
if not machinenumber:
return None
partof = RelationshipType.query.filter_by(
relationshiptype='partof').first()
if not partof:
warnings.append("'partof' relationship type missing; "
'run flask seed reference-data')
return None
operation = self._asset_by_number(machinenumber)
if not operation:
warnings.append(
'no asset for machine number {!r}; marker not filed under an '
'operation'.format(machinenumber))
return None
if operation.assetid == markerasset.assetid:
return None
if pcasset is not None and operation.assetid == pcasset.assetid:
warnings.append(
'machine number {!r} is this PC\'s own asset number; marker '
'not filed under an operation. Rename the PC asset to its '
'hostname, or create the operation asset.'.format(
machinenumber))
return None
# Our own rows first: a marker that moved to another operation has its
# old membership archived, never deleted, so where it used to live stays
# answerable.
for rel in AssetRelationship.query.filter(
AssetRelationship.sourceassetid == markerasset.assetid,
AssetRelationship.relationshiptypeid == partof.relationshiptypeid,
AssetRelationship.label == label).all():
if rel.targetassetid != operation.assetid and rel.isactive:
rel.isactive = False
# Then get-or-create KEYED ON THE UNIQUE TRIPLE, not on the label.
# assetrelationships is unique on (source, target, type), so a row made
# by hand or by the legacy import - carrying a different label, or none
# at all - was invisible to the label-filtered lookup this used to do,
# and the insert then violated that constraint. The PC reported 200 on
# the cycle that minted the marker and 500 on every cycle after.
found = AssetRelationship.query.filter_by(
sourceassetid=markerasset.assetid,
targetassetid=operation.assetid,
relationshiptypeid=partof.relationshiptypeid).first()
if found is not None:
found.isactive = True
if not found.label:
# Unlabelled means the import or a person made it. Stamping ours
# is what makes the next cycle reuse it. A row carrying somebody
# else's label is left as theirs.
found.label = label
else:
db.session.add(AssetRelationship(
sourceassetid=markerasset.assetid,
targetassetid=operation.assetid,
relationshiptypeid=partof.relationshiptypeid,
label=label,
isactive=True))
return operation.assetid
def _assetname(self, assetid):
"""Readable name for an asset id, for warnings and alerts."""
from shopdb.api import Asset
asset = db.session.get(Asset, assetid)
return asset.assetnumber if asset else str(assetid)
def _incumbent_has_yielded(self, assetid):
"""True when the PC currently holding a machine has given it up.
Two ways to yield, and both are evidence rather than a guess:
It went quiet. A PC pulled off a machine stops reporting, so silence
past MACHINE_CLAIM_QUIET_HOURS is the handover signal. The window is
long enough that a PC switched off overnight, or one behind a network
outage, never loses its bay to a spare sitting on the bench.
Or a person moved it off In Use. Setting the old PC to Retired,
Inventory or In Repair is a deliberate statement that it no longer runs
the machine, and it is the one-step way to force a handover the moment
the swap happens instead of waiting out the window.
An asset with no computer extension row cannot report at all, so it
cannot be alive; it yields.
"""
from datetime import datetime, timedelta, timezone
from shopdb.api import Asset
computer = Computer.query.filter_by(assetid=assetid).first()
if not computer:
return True
asset = db.session.get(Asset, assetid)
status = asset.status.status if asset and asset.status else None
if status and status != 'In Use':
return True
reported = computer.lastreporteddate
if not reported:
return True
quiet = datetime.now(timezone.utc).replace(tzinfo=None) - reported
return quiet > timedelta(hours=MACHINE_CLAIM_QUIET_HOURS)
def _alert_machine_contested(self, holdername, newcomp, machine):
"""Say that a PC claims a machine another PC is still running.
Fires ONCE, on the report that first records the dormant claim. The
dormant relationship row is the marker: while it exists the claim is
already known, so a PC sitting on the bench for a fortnight does not
alert every collector cycle.
Nothing is changed in the data by this. It exists because the two
legitimate readings - a swap in progress, or a machine number typed
onto the wrong PC at imaging - look identical to the collector, and
only a person can tell them apart.
"""
import logging as _logging
if not self._machinelink_alerts_enabled():
return
try:
from shopdb.api import send_alert, Setting
newname = newcomp.hostname if newcomp else 'a new PC'
machinename = machine.assetnumber or machine.name
subject = 'Machine {} claimed by {}, still run by {}'.format(
machinename, newname, holdername)
base = (Setting.get('site_base_url') or '').rstrip('/')
link = '{}/pcs/{}'.format(base, newcomp.computerid) if base else ''
linkhtml = ('<p><a href="{0}">View {1}</a></p>'.format(link, newname)
if link else '')
html = (
'<p><strong>{0}</strong> reports that it runs machine '
'<strong>{1}</strong>, but <strong>{2}</strong> is still '
'reporting on that machine and is still In Use, so the link '
'has been left where it is.</p>'
'<p>If this is a swap in progress, nothing needs doing: {0} '
'takes the machine once {2} stops reporting for a day, or '
'straight away if you set {2} to Retired, Inventory or In '
'Repair. If instead {0} was imaged with the wrong machine '
'number, correct it on {0}.</p>{3}'.format(
newname, machinename, holdername, linkhtml))
send_alert(subject, html)
except Exception:
_logging.getLogger(__name__).exception(
'machine-contested alert failed for machine %s',
getattr(machine, 'assetnumber', '?'))
def _alert_pc_superseded(self, oldname, newcomp, machine):
"""Tell a human a PC was replaced on a machine. Best effort, never raises.
Deliberately an ALERT and not a status change: the collector cannot tell
whether the old PC was shelved, sent for repair or re-imaged for another
bay, so it says what happened and lets a person decide.
Goes through send_alert, which is the site's configured alert fan-out:
email to the SMTP settings' alert recipients plus the alert webhook.
Resolving those recipients by hand would have missed the
SMTP_ALERT_RECIPIENTS environment fallback, so a site that configures
SMTP by environment rather than in the UI would have got the webhook
and no email.
NOT a shopfloor notification: that board is for operators (General,
Recertification, Recognition), and an IT asset message does not belong
in front of the floor.
"""
import logging as _logging
if not self._machinelink_alerts_enabled():
return
try:
from shopdb.api import send_alert, Setting
newname = newcomp.hostname if newcomp else 'a new PC'
machinename = machine.assetnumber or machine.name
subject = 'PC replaced on machine {}: {} to {}'.format(
machinename, oldname, newname)
# /pcs/:id is keyed on computerid, not assetid - an assetid here
# opens someone else's PC or a 404.
base = (Setting.get('site_base_url') or '').rstrip('/')
link = '{}/pcs/{}'.format(base, newcomp.computerid) if base else ''
linkhtml = ('<p><a href="{0}">View {1}</a></p>'.format(link, newname)
if link else '')
html = (
'<p><strong>{0}</strong> is now reporting machine '
'<strong>{1}</strong>, which was previously driven by '
'<strong>{2}</strong>.</p>'
'<p>The old link has been archived. {2} has NOT had its status '
'changed - set it to Inventory, In Repair or Retired as '
'appropriate.</p>{3}'.format(newname, machinename, oldname,
linkhtml))
send_alert(subject, html)
except Exception:
_logging.getLogger(__name__).exception(
'PC-superseded alert failed for machine %s',
getattr(machine, 'assetnumber', '?'))
def _sync_measuringtool_link(self, pcasset, pctype, hostname, warnings,
measuringtoolid=None, machinenumber=None):
"""Link the MeasuringTool a metrology PC drives. ADOPT before minting.
A CMM / Keyence / Genspect / wax-and-trace pc-type means the PC controls
an attached instrument. This links it with a directional PC->tool
'controls' relationship tagged MEASURINGTOOL_LINK_ORIGIN.
RESOLUTION ORDER, most stable identity first:
1. measuringtoolid - the instrument named by the enrollment file
measuringtool-id.txt. Explicit, survives a PC swap, and it is what
a Keyence or Genspect bay has instead of a bay id.
2. a prior collector link on this PC (reactivate + retype).
3. an EXISTING measuring tool this PC already controls that the
collector did not create - a legacy MT-#### row, or one somebody
made by hand. Adopting stamps the label so it is ours from then on.
4. the reported machine number, when it resolves to a measuring tool.
This is the CMM case: cmmid.txt already reports CMM4.
5. mint, only when none of the above found anything.
WHY THIS ORDER. Minting derives an asset number from the HOSTNAME, so a
permanent instrument inherited the identity of whichever PC drove it that
week - replace the PC and either the number lies or a second tool appears
for the same physical unit. It also could not see a tool it had not
created itself, so on prod 43 legacy MT-#### tools were shadowed by
minted <HOST>-CMM twins, three records deep in places. Minting is now the
last resort, and what it produces should be treated as a placeholder
until a real identifier is recorded.
A non-metrology pc-type archives any collector-created tool link (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)
# --- 1. an explicitly named instrument wins over everything ---------
adopted = None
named = (measuringtoolid or '').strip()
if named:
candidate = self._asset_by_number(named)
if candidate is None:
# Warn rather than invent: a typo in the file must not mint a
# phantom instrument that nobody can account for. RETURN, do
# not fall through - minting <HOST>-CMM here would create the
# hostname-derived twin this resolution order exists to avoid,
# while the warning claimed nothing was linked.
warnings.append(
'no asset for measuring tool {!r}; not linked'.format(named))
return []
if candidate.assetid not in self._measuringtool_assetids():
# The name resolved, but not to an instrument. measuringtool-id
# .txt holding a machine number would otherwise link the PC to
# that MACHINE under a measuring-tool label - a link that reads
# as an instrument everywhere downstream, on a row the machine
# sync also owns. Refuse and say which asset it hit.
warnings.append(
'asset {!r} is not a measuring tool; not linked'.format(
candidate.assetnumber))
return []
adopted = candidate
# --- 2. a prior collector link (reactivate + retype) -----------------
reuse = next((rel for rel in existing if rel.isactive), None) \
or (existing[0] if existing else None)
# --- 3/4. adopt a tool this PC already controls that we did not make --
# Ordered after the collector's own link so a settled PC keeps what it
# has, and before minting so an existing instrument is never twinned.
if adopted is None and reuse is None:
adopted = self._adoptable_measuringtool(
pcasset, controls, machinenumber)
if adopted is not None:
# Settle any OTHER PC still holding this instrument first: a yielded
# incumbent is archived, a live one keeps it and this link is
# recorded dormant rather than becoming a second active holder.
mayhold = self._device_incumbents_yield(
adopted, pcasset, MEASURINGTOOL_LINK_ORIGIN, warnings)
# CLAIM the existing link rather than adding a second one. A row for
# (pc, tool, controls) usually already exists - that is how the tool
# was found - and assetrelationships is unique on exactly that
# triple, so inserting would raise.
claimed = AssetRelationship.query.filter_by(
sourceassetid=pcasset.assetid,
targetassetid=adopted.assetid,
relationshiptypeid=controls.relationshiptypeid).first()
if claimed is not None and claimed.label in (
MACHINE_LINK_ORIGIN, MACHINE_LINK_IMPORT_ORIGIN):
# LEAVE IT ALONE. On a CMM the instrument IS the reported bay,
# so the machine sync (or a CSV import's machine link) has
# already made the row for this exact triple - and each finds
# its own rows BY that label. Relabelling hid the link from it,
# so the next cycle built a second row for the same triple and
# MySQL rejected it: the PC reported 200, then 500 forever
# after. The row's lifecycle (including staying dormant while a
# bay is contested) belongs to the machine sync; adoption only
# needs the identity, which it has.
pass
elif claimed is not None:
# Relabelling is what adoption MEANS for any other row: the link
# is ours from here, so a later cycle reuses it instead of
# minting.
claimed.label = MEASURINGTOOL_LINK_ORIGIN
claimed.isactive = mayhold
else:
db.session.add(AssetRelationship(
sourceassetid=pcasset.assetid,
targetassetid=adopted.assetid,
relationshiptypeid=controls.relationshiptypeid,
label=MEASURINGTOOL_LINK_ORIGIN,
isactive=mayhold))
if adopted.measuringtool and tooltype:
adopted.measuringtool.measuringtooltypeid = \
tooltype.measuringtooltypeid
targetid = adopted.assetid
elif reuse:
# Same re-check as the marker path: a dormant challenger must not be
# promoted just because its enrollment file stopped being readable.
toolasset = db.session.get(Asset, reuse.targetassetid)
reuse.isactive = self._device_incumbents_yield(
toolasset, pcasset, MEASURINGTOOL_LINK_ORIGIN, warnings)
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 row.
#
# Keyed on the TARGET, not on object identity. `rel is not reuse` spared
# the row reuse pointed at even when adoption had just resolved a
# different instrument - so a bay whose measuringtool-id.txt named the
# real tool kept its minted twin link alive alongside the adopted one,
# and the PC read as controlling two instruments. That is the exact
# duplicate this whole adoption path exists to remove.
for rel in existing:
if rel.targetassetid != targetid and rel.isactive:
rel.isactive = False
return [{'assetid': targetid, 'relationshiptype': 'controls',
'measuringtooltype': typename}]
@staticmethod
def _asset_by_number(assetnumber):
"""The active asset with exactly this asset number, case-insensitively.
NOT `ilike`. Every identifier resolved here arrives from a file on a
shopfloor PC - cmmid.txt, measuringtool-id.txt, the reported machine
number - and ilike reads `_` and `%` in that value as WILDCARDS. A
reported '%' matched whatever active asset happened to come first, and
'MT-600_' matched MT-6001: an identifier silently resolving to an asset
it does not name, then linked to as though it did.
"""
from shopdb.api import Asset
number = (assetnumber or '').strip()
if not number:
return None
return Asset.query.filter(
db.func.lower(Asset.assetnumber) == number.lower(),
Asset.isactive.is_(True)).first()
def _measuringtool_assetids(self):
"""Asset ids that really are measuring tools.
Empty when the measuringtools plugin is not installed, which makes both
callers fall through rather than guess: an asset number alone does not
say what kind of thing it names.
"""
try:
from plugins.measuringtools.models import MeasuringTool
except ImportError:
return set()
return {row.assetid for row in
MeasuringTool.query.with_entities(MeasuringTool.assetid)}
def _adoptable_measuringtool(self, pcasset, controls, machinenumber):
"""A measuring tool this PC already controls that we did not create.
Two sources, in order:
a. any ACTIVE controls link from this PC to a measuring_tool asset
that carries a different label - the legacy import's MT-#### rows,
or one somebody added by hand. These were invisible to the
collector, which keyed idempotency on its OWN label and so minted a
twin for the same physical instrument on every fresh PC.
b. the reported machine number, when it names a measuring tool. That
is the CMM case: cmmid.txt reports CMM4, which already resolves to
a real asset through the machine-link path.
Returns the Asset to adopt, or None. Never creates anything.
"""
from shopdb.api import AssetRelationship, Asset
toolassetids = self._measuringtool_assetids()
if not toolassetids:
return None
# a. already linked, by someone else
for rel in (AssetRelationship.query
.filter(AssetRelationship.sourceassetid == pcasset.assetid,
AssetRelationship.relationshiptypeid
== controls.relationshiptypeid,
AssetRelationship.isactive.is_(True))
.order_by(AssetRelationship.relationshipid.asc()).all()):
if rel.label == MEASURINGTOOL_LINK_ORIGIN:
continue
if rel.targetassetid in toolassetids:
return db.session.get(Asset, rel.targetassetid)
# b. the reported machine number naming a tool
number = (machinenumber or '').strip()
if number:
candidate = self._asset_by_number(number)
if candidate is not None and candidate.assetid in toolassetids:
return candidate
return None
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('retype-by-role')
@click.option('--apply', 'apply_changes', is_flag=True,
help='Actually write. Without this the command only reports.')
@click.option('--typename', default='Metrology PC', show_default=True,
help='ComputerType to assign to PCs that drive a measuring tool.')
def retype_by_role(apply_changes, typename):
"""Type PCs that drive a measuring tool as metrology machines.
WHY THE RELATIONSHIP AND NOT THE NAME. A PC's role is already
recorded in what it controls: a metrology box has an active
relationship to a measuring_tool asset. Matching on hostnames or on
the tool's name would guess; this reads the fact.
The imaging scope (gea-shopfloor-keyence and friends) is the OTHER
source, and it is the authoritative one - but only 17 of ~292 PCs
have ever reported an enforcement scope, so it cannot answer for the
fleet today. Relationships can.
"""
from shopdb.api import Asset
target = ComputerType.query.filter_by(computertype=typename).first()
if not target:
if not apply_changes:
click.echo(click.style(
"Type '%s' does not exist yet; it would be created."
% typename, fg='yellow'))
else:
target = ComputerType(computertype=typename, isactive=True)
db.session.add(target)
db.session.flush()
click.echo("Created ComputerType '%s'." % typename)
rows = db.session.execute(db.text("""
SELECT DISTINCT c.computerid, a.assetnumber, c.hostname,
ct.computertype
FROM computers c
JOIN assets a ON a.assetid = c.assetid
LEFT JOIN computertypes ct ON ct.computertypeid = c.computertypeid
JOIN assetrelationships r ON r.sourceassetid = c.assetid
JOIN assets ta ON ta.assetid = r.targetassetid
JOIN assettypes at ON at.assettypeid = ta.assettypeid
WHERE at.assettype = 'measuring_tool' AND r.isactive = 1
ORDER BY ct.computertype, a.assetnumber
""")).fetchall()
moving = [r for r in rows if (r[3] or '') != typename]
click.echo('%d PC(s) drive a measuring tool; %d would change type.'
% (len(rows), len(moving)))
if not moving:
return
for r in moving:
click.echo(' %-14s %-14s %-16s -> %s'
% (r[1], r[2] or '(no hostname)', r[3] or '(none)', typename))
if not apply_changes:
click.echo(click.style(
'Dry run. Nothing was written. Re-run with --apply.', fg='yellow'))
return
for r in moving:
db.session.get(Computer, r[0]).computertypeid = target.computertypeid
db.session.commit()
click.echo(click.style('Retyped %d PC(s).' % len(moving), fg='green'))
@computerscli.command('prune-types')
@click.option('--apply', 'apply_changes', is_flag=True,
help='Actually delete. Without this the command only reports.')
def prune_types(apply_changes):
"""Delete computer types nothing uses and nothing maps to.
Two guards, because a type is cheap to keep and expensive to delete
by mistake. A type is KEPT when any computer is on it, or when the
live pctype map names it - the map is what the collector assigns
from at check-in, so deleting a mapped type would make every future
check-in log 'mapped ComputerType not found' instead of typing the
machine.
"""
from .pctypemap import pctype_mapping
mapped = {str(v).strip().lower() for v in pctype_mapping().values()}
counts = dict(db.session.execute(db.text("""
SELECT ct.computertypeid, COUNT(c.computerid)
FROM computertypes ct
LEFT JOIN computers c ON c.computertypeid = ct.computertypeid
GROUP BY ct.computertypeid
""")).fetchall())
removable, kept = [], []
for ctype in ComputerType.query.order_by(ComputerType.computertype).all():
inuse = counts.get(ctype.computertypeid, 0)
ismapped = ctype.computertype.strip().lower() in mapped
if inuse:
kept.append((ctype.computertype, '%d computer(s)' % inuse))
elif ismapped:
kept.append((ctype.computertype, 'named in the pctype map'))
else:
removable.append(ctype)
click.echo('KEEPING %d:' % len(kept))
for name, why in kept:
click.echo(' %-24s %s' % (name, why))
click.echo('')
click.echo('DELETING %d:' % len(removable))
for ctype in removable:
click.echo(' %s' % ctype.computertype)
if not removable:
return
if not apply_changes:
click.echo(click.style(
'Dry run. Nothing was deleted. Re-run with --apply.', fg='yellow'))
return
for ctype in removable:
db.session.delete(ctype)
db.session.commit()
click.echo(click.style('Deleted %d type(s).' % len(removable), fg='green'))
@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 PC' - NOT 'Shopfloor', which is an empty
# duplicate this counted for as long as both existed, so the
# number reported was always 0.
sf = ComputerType.query.filter_by(computertype='Shopfloor PC').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]:
"""Dashboard cards this plugin contributes.
REPLACES a declaration that named a Vue component which was never
written - as did four other plugins - so nothing rendered and nobody
noticed, because the frontend never called the widgets endpoint either.
The contract now carries data and a renderer; core draws it.
"""
return [
{
'id': 'computers-quiet',
'viewall': '/pcs',
'title': 'PCs gone silent',
'shortname': 'PC silent',
'endpoint': '/api/computers/dashboard/quiet',
'render': 'exceptions',
'severity': 'warning',
'permission': 'computers.view',
'empty': 'hide',
'position': 20,
'map': {
'title': 'hostname',
'detail': '',
'meta': [{'key': 'quietdays', 'label': 'silent',
'suffix': ' days'}],
'link': '/pcs/{computerid}',
},
},
{
'id': 'computers-sharedmachines',
'viewall': '/machines',
'title': 'Machine numbers on two PCs',
'shortname': 'Duplicate machine no.',
'endpoint': '/api/computers/dashboard/sharedmachines',
'render': 'exceptions',
'severity': 'critical',
'permission': 'computers.view',
'empty': 'hide',
'position': 15,
'map': {
'title': 'machinenumber',
'detail': 'pcs',
'meta': [{'key': 'pccount', 'suffix': ' PCs'}],
'link': '/machines/{assetid}',
},
},
]
def get_asset_panels(self) -> List[Dict]:
"""Policy provenance on the PC detail page.
The effective policy already narrows which commands an endpoint will
accept, so without this the reason a control is unavailable is not
visible anywhere. Right column, near the services it governs: this is
provenance, not a 6am triage signal.
The route returns an empty field list when the policy feature is off,
so the card disappears on sites that have not enabled it rather than
rendering an error.
"""
return [
{
'id': 'computers-policy',
'title': 'Policy',
'assettypes': ['computer'],
'endpoint': '/api/computers/asset/{assetid}/policypanel',
'render': 'keyvalue',
'position': 40,
},
]
def get_navigation_items(self) -> List[Dict]:
"""Return navigation menu items."""
return [
{
'name': 'PCs',
'icon': 'desktop',
'route': '/pcs',
'permission': 'computers.view',
'position': 15,
'section': 'assets',
},
]
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'),
]
def get_guest_grantable_permissions(self) -> List[str]:
"""Declared because a site chose it: the floor looks PCs up by number.
KNOWN GAP, tracked separately. The computer serializer is
BaseModel.to_dict(), so it publishes every column - serialnumber, the
Azure device and object ids - and the detail route resolves
accessmethods into launch links, each built from that protocol's own
linktemplate, plus the full Communication row. A guest granted this
gets all of it, on the LIST route as well, so it is one request rather
than one per asset.
Do not name schemes here. The templates are site-editable rows in
accessprotocols, so what a link looks like is data, not a constant -
the shipped VNC row is com.realvnc.vncviewer.connect://, not vnc://,
and an earlier version of this docstring said otherwise for months.
The fix is a field allowlist on Computer.to_dict() plus suppressing the
resolved links unless the caller holds a remote-access permission. Do
not widen this hook until that lands; a site enabling guest access
today should weigh it.
"""
return ['computers.view']
Reference in New Issue View Git Blame Copy Permalink