2083029ff2
A site adopting ShopDB had to be handed two files and told what to edit in them. Both are now the product's, and one of them the server writes for you. GET /api/computers/client-script (admin) returns Report-AssetToShopDB.ps1 with this site's values already in it: site_base_url becomes the -ApiUrl default and the new computers_routableranges setting becomes -AllowedRanges. Only the PARAMETER DEFAULTS are substituted - the copy in plugins/computers/client/ stays runnable, so there is no second version to drift from the first - and everything stamped stays overridable by argument or registry, because a bay may need to differ from its site. Settings > Computers > Asset reporter edits the ranges, downloads the script and shows its SHA-256. The collector key is deliberately not stamped in, and a test fails if it ever is. That file lands on every shop-floor PC, and a token spread across hundreds of bays cannot be rotated quietly; it stays in the registry, provisioned per ADOPTING-AT-ANOTHER-SITE.md. The routable ranges are the last thing that was hardcoded in that script. They are now a setting, so West Jefferson's two CIDRs move out of source code and into that site's own configuration - which is what ADR-015 asks for - and a site that sets nothing still works, because the script falls back to the NIC carrying the default route. EventSaver joins it in plugins/slides/client/, source only: EventSaver.cs and EventSaver.ini, no compiled .scr - a binary is a release asset, like the installer exe. The share path that was compiled into Config.Folder is gone. It used to be the fallback when the ini was missing, which silently pointed a new site at the reference site's file server; it is now empty, and failing visibly beats displaying another site's slides. Verified by compiling the edited source in the Windows VM with the in-box csc.exe: 15,872 bytes, exit 0. Also: the DSC example in the adoption guide gains a CollectorRanges resource and stops passing -ApiUrl to a script that already reads BaseUrl from the registry the same example writes, and the guide points at the generated download instead of hand-editing a URL. The contract test caught the endpoint importing shopdb directly for the version string, which ADR-002 forbids a plugin from doing. The product and contract versions are in app.config now, which a plugin reads through current_app. Adds docs/proposals/printer-assignment.md: assign printers to a PC in ShopDB and let the bay install them, with what the fleet data says about drivers - HP and Xerox cover 41 of 44 printers with universal drivers, there are no Brother printers at all despite 208 files of Brother inkjet drivers in the installer, and printerdrivers holds one row pointing at a per-model folder instead of a universal driver.
1176 lines
42 KiB
Python
1176 lines
42 KiB
Python
"""Computers plugin API endpoints."""
|
|
|
|
from flask import Blueprint, request, Response, current_app
|
|
from flask_jwt_extended import jwt_required
|
|
|
|
from shopdb.api import db, Asset, AssetType, OperatingSystem, Application, AuditLog, Communication, CommunicationType, Setting, success_response, error_response, paginated_response, ErrorCodes, get_pagination_params, paginate_query
|
|
|
|
from ..models import Computer, ComputerType, ComputerInstalledApp, AccessProtocol, ComputerAccess
|
|
|
|
from shopdb.api import require_permission, require_role, apply_import_timestamps
|
|
|
|
computers_bp = Blueprint('computers', __name__)
|
|
|
|
|
|
# =============================================================================
|
|
# Computer Types
|
|
# =============================================================================
|
|
|
|
@computers_bp.route('/types', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def list_computer_types():
|
|
"""List all computer types."""
|
|
page, per_page = get_pagination_params(request)
|
|
|
|
query = ComputerType.query
|
|
|
|
if request.args.get('active', 'true').lower() != 'false':
|
|
query = query.filter(ComputerType.isactive == True)
|
|
|
|
if search := request.args.get('search'):
|
|
query = query.filter(ComputerType.computertype.ilike(f'%{search}%'))
|
|
|
|
query = query.order_by(ComputerType.computertype)
|
|
|
|
items, total = paginate_query(query, page, per_page)
|
|
data = [t.to_dict() for t in items]
|
|
|
|
return paginated_response(data, page, per_page, total)
|
|
|
|
|
|
@computers_bp.route('/types/<int:type_id>', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def get_computer_type(type_id: int):
|
|
"""Get a single computer type."""
|
|
t = db.session.get(ComputerType, type_id)
|
|
|
|
if not t:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Computer type with ID {type_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
return success_response(t.to_dict())
|
|
|
|
|
|
@computers_bp.route('/types', methods=['POST'])
|
|
@jwt_required()
|
|
@require_permission('computers.create')
|
|
def create_computer_type():
|
|
"""Create a new computer type."""
|
|
data = request.get_json()
|
|
|
|
if not data or not data.get('computertype'):
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'computertype is required')
|
|
|
|
existing = ComputerType.query.filter_by(computertype=data['computertype']).first()
|
|
if existing:
|
|
if not existing.isactive:
|
|
# Adding a name that matches a deactivated type revives it.
|
|
existing.isactive = True
|
|
for key in ('description', 'icon', 'color'):
|
|
if data.get(key) is not None:
|
|
setattr(existing, key, data[key])
|
|
db.session.commit()
|
|
return success_response(existing.to_dict(), message='Reactivated existing computer type')
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"Computer type '{data['computertype']}' already exists",
|
|
http_code=409
|
|
)
|
|
|
|
t = ComputerType(
|
|
computertype=data['computertype'],
|
|
description=data.get('description'),
|
|
icon=data.get('icon'), color=data.get('color')
|
|
)
|
|
|
|
db.session.add(t)
|
|
db.session.commit()
|
|
|
|
return success_response(t.to_dict(), message='Computer type created', http_code=201)
|
|
|
|
|
|
@computers_bp.route('/types/<int:type_id>', methods=['PUT'])
|
|
@jwt_required()
|
|
@require_permission('computers.edit')
|
|
def update_computer_type(type_id: int):
|
|
"""Update a computer type."""
|
|
t = db.session.get(ComputerType, type_id)
|
|
|
|
if not t:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Computer type with ID {type_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
data = request.get_json()
|
|
if not data:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
|
|
|
|
if 'computertype' in data and data['computertype'] != t.computertype:
|
|
if ComputerType.query.filter_by(computertype=data['computertype']).first():
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"Computer type '{data['computertype']}' already exists",
|
|
http_code=409
|
|
)
|
|
|
|
for key in ['computertype', 'description', 'icon', 'color', 'isactive']:
|
|
if key in data:
|
|
setattr(t, key, data[key])
|
|
|
|
db.session.commit()
|
|
return success_response(t.to_dict(), message='Computer type updated')
|
|
|
|
|
|
@computers_bp.route('/types/<int:type_id>', methods=['DELETE'])
|
|
@jwt_required()
|
|
@require_permission('computers.delete')
|
|
def delete_computer_type(type_id: int):
|
|
"""Delete a computer type. Refused if any PC still uses it."""
|
|
t = db.session.get(ComputerType, type_id)
|
|
if not t:
|
|
return error_response(ErrorCodes.NOT_FOUND, 'Computer type not found', http_code=404)
|
|
inuse = Computer.query.filter_by(computertypeid=type_id).count()
|
|
if inuse:
|
|
return error_response(ErrorCodes.CONFLICT,
|
|
f"Cannot delete: {inuse} PC(s) still use this type", http_code=409)
|
|
db.session.delete(t)
|
|
db.session.commit()
|
|
return success_response(message='Computer type deleted')
|
|
|
|
|
|
# =============================================================================
|
|
# Access protocol catalog (VNC / WinRM / RDP / ...) - admin-managed
|
|
# =============================================================================
|
|
|
|
@computers_bp.route('/protocols', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def list_protocols():
|
|
"""List access protocols. ?active=false includes disabled ones."""
|
|
query = AccessProtocol.query
|
|
if request.args.get('active', 'true').lower() != 'false':
|
|
query = query.filter(AccessProtocol.isactive == True)
|
|
protocols = query.order_by(AccessProtocol.name).all()
|
|
return success_response([p.to_dict() for p in protocols])
|
|
|
|
|
|
@computers_bp.route('/protocols', methods=['POST'])
|
|
@jwt_required()
|
|
@require_permission('computers.edit')
|
|
def create_protocol():
|
|
data = request.get_json() or {}
|
|
if not (data.get('name') and data.get('scheme') and data.get('linktemplate')):
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'name, scheme and linktemplate are required')
|
|
if AccessProtocol.query.filter_by(name=data['name']).first():
|
|
return error_response(ErrorCodes.CONFLICT, f"Protocol '{data['name']}' already exists", http_code=409)
|
|
p = AccessProtocol(
|
|
name=data['name'],
|
|
scheme=data['scheme'],
|
|
defaultport=data.get('defaultport') or None,
|
|
linktemplate=data['linktemplate'],
|
|
isactive=data.get('isactive', True),
|
|
)
|
|
db.session.add(p)
|
|
db.session.commit()
|
|
return success_response(p.to_dict(), message='Protocol created', http_code=201)
|
|
|
|
|
|
@computers_bp.route('/protocols/<int:protocol_id>', methods=['PUT', 'PATCH'])
|
|
@jwt_required()
|
|
@require_permission('computers.edit')
|
|
def update_protocol(protocol_id):
|
|
p = db.session.get(AccessProtocol, protocol_id)
|
|
if not p:
|
|
return error_response(ErrorCodes.NOT_FOUND, 'Protocol not found', http_code=404)
|
|
data = request.get_json() or {}
|
|
for field in ('name', 'scheme', 'linktemplate'):
|
|
if data.get(field):
|
|
setattr(p, field, data[field])
|
|
if 'defaultport' in data:
|
|
p.defaultport = data['defaultport'] or None
|
|
if 'isactive' in data:
|
|
p.isactive = bool(data['isactive'])
|
|
db.session.commit()
|
|
return success_response(p.to_dict(), message='Protocol updated')
|
|
|
|
|
|
@computers_bp.route('/protocols/<int:protocol_id>', methods=['DELETE'])
|
|
@jwt_required()
|
|
@require_permission('computers.edit')
|
|
def delete_protocol(protocol_id):
|
|
p = db.session.get(AccessProtocol, protocol_id)
|
|
if not p:
|
|
return error_response(ErrorCodes.NOT_FOUND, 'Protocol not found', http_code=404)
|
|
# If any PC still references it, deactivate rather than hard-delete.
|
|
if ComputerAccess.query.filter_by(protocolid=protocol_id).first():
|
|
p.isactive = False
|
|
db.session.commit()
|
|
return success_response(message='Protocol is in use; deactivated instead of deleted')
|
|
db.session.delete(p)
|
|
db.session.commit()
|
|
return success_response(message='Protocol deleted')
|
|
|
|
|
|
def _pc_access_domain():
|
|
# Contract-pure read of the pc_access_domain setting (no core.api import).
|
|
row = Setting.query.filter_by(key='pc_access_domain').first()
|
|
return ((row.value if row else '') or '').strip()
|
|
|
|
|
|
def _computer_access_links(comp, domain=None):
|
|
"""Resolved remote-access links for a computer: each enabled protocol's
|
|
template filled with the PC hostname joined to the pc_access_domain setting.
|
|
A hostname that is already an FQDN (has a dot) is used as-is. Pass domain
|
|
when calling in a loop to avoid one settings lookup per computer."""
|
|
if domain is None:
|
|
domain = _pc_access_domain()
|
|
hostname = (comp.hostname or '').strip()
|
|
if not hostname:
|
|
host = ''
|
|
elif '.' in hostname or not domain:
|
|
host = hostname
|
|
else:
|
|
host = f"{hostname}.{domain}"
|
|
|
|
links = []
|
|
for am in comp.accessmethods:
|
|
protocol = am.protocol
|
|
if not (am.isactive and protocol and protocol.isactive):
|
|
continue
|
|
port = am.portoverride or protocol.defaultport
|
|
link = None
|
|
if host:
|
|
try:
|
|
link = protocol.linktemplate.format(
|
|
host=host,
|
|
port=(port if port is not None else ''),
|
|
scheme=protocol.scheme,
|
|
)
|
|
except (KeyError, IndexError, ValueError):
|
|
link = None
|
|
links.append({
|
|
'id': am.id,
|
|
'protocolid': protocol.protocolid,
|
|
'name': protocol.name,
|
|
'scheme': protocol.scheme,
|
|
'port': port,
|
|
'portoverride': am.portoverride,
|
|
'link': link,
|
|
})
|
|
return links
|
|
|
|
|
|
def _sync_access_methods(comp, data):
|
|
"""Replace a computer's enabled protocols from data['accessmethods'] (a list
|
|
of {protocolid, portoverride?}). No-op if the key is absent, so callers that
|
|
don't touch access aren't affected."""
|
|
if 'accessmethods' not in data:
|
|
return
|
|
desired = data.get('accessmethods') or []
|
|
ComputerAccess.query.filter_by(computerid=comp.computerid).delete()
|
|
seen = set()
|
|
for m in desired:
|
|
try:
|
|
pid = int(m.get('protocolid'))
|
|
except (TypeError, ValueError):
|
|
continue
|
|
if pid in seen:
|
|
continue
|
|
seen.add(pid)
|
|
port = m.get('portoverride')
|
|
try:
|
|
port = int(port) if port not in (None, '') else None
|
|
except (TypeError, ValueError):
|
|
port = None
|
|
db.session.add(ComputerAccess(
|
|
computerid=comp.computerid,
|
|
protocolid=pid,
|
|
portoverride=port,
|
|
isactive=True,
|
|
))
|
|
|
|
|
|
# =============================================================================
|
|
# Computers CRUD
|
|
# =============================================================================
|
|
|
|
@computers_bp.route('/display-kiosks', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def list_display_kiosks():
|
|
"""Reporting display kiosks, for the Dashboard Defaults picker.
|
|
|
|
Returns the computers whose type is the one gea-shopfloor-display maps to
|
|
(default 'Kiosk'), each with its DERIVED FQDN (F<serial>.<domain>, domain
|
|
from the display_fqdn_domain setting) so an admin picks a kiosk from a
|
|
dropdown instead of typing an IP/FQDN. Keeps the same F<serial>.<domain>
|
|
format as core derive_display_fqdn (duplicated to avoid a contract bump).
|
|
"""
|
|
from ..pctypemap import pctype_mapping
|
|
from shopdb.api import Setting
|
|
|
|
display_type_name = pctype_mapping().get('gea-shopfloor-display', 'Kiosk')
|
|
ctype = ComputerType.query.filter_by(computertype=display_type_name).first()
|
|
if not ctype:
|
|
return success_response([])
|
|
domain = (Setting.get('display_fqdn_domain', 'device.geaerospace.net') # ADR-015-OK: GE Aerospace-wide domain, and only the DEFAULT of a documented setting every site can override.
|
|
or 'device.geaerospace.net').strip().strip('.') # ADR-015-OK: GE Aerospace-wide domain, and only the DEFAULT of a documented setting every site can override.
|
|
rows = (db.session.query(Computer).join(Asset)
|
|
.filter(Computer.computertypeid == ctype.computertypeid,
|
|
Asset.isactive == True)
|
|
.order_by(Computer.hostname).all())
|
|
out = []
|
|
for comp in rows:
|
|
serial = (comp.asset.serialnumber or '').strip() if comp.asset else ''
|
|
out.append({
|
|
'computerid': comp.computerid,
|
|
'hostname': comp.hostname,
|
|
'serialnumber': serial or None,
|
|
'fqdn': f'F{serial}.{domain}'.lower() if serial else None,
|
|
})
|
|
return success_response(out)
|
|
|
|
|
|
@computers_bp.route('', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def list_computers():
|
|
"""
|
|
List all computers with filtering and pagination.
|
|
|
|
Query parameters:
|
|
- page, per_page: Pagination
|
|
- active: Filter by active status
|
|
- search: Search by asset number, name, or hostname
|
|
- type_id: Filter by computer type ID
|
|
- os_id: Filter by operating system ID
|
|
- location_id: Filter by location ID
|
|
- businessunit_id: Filter by business unit ID
|
|
- shopfloor: Filter by shopfloor flag (true/false)
|
|
"""
|
|
page, per_page = get_pagination_params(request)
|
|
|
|
# Join Computer with Asset
|
|
query = db.session.query(Computer).join(Asset)
|
|
|
|
# Active filter
|
|
if request.args.get('active', 'true').lower() != 'false':
|
|
query = query.filter(Asset.isactive == True)
|
|
|
|
# Exact-match natural-key lookup for idempotent import (asset number).
|
|
if exactassetnumber := request.args.get('assetnumber'):
|
|
query = query.filter(Asset.assetnumber == exactassetnumber)
|
|
|
|
# Search filter. Covers the type name too - it is a column in the list, so
|
|
# searching 'Standard' must find the PCs of that type. Outer join so a PC
|
|
# with no type still matches on its own fields.
|
|
if search := request.args.get('search'):
|
|
pattern = f'%{search}%'
|
|
query = query.outerjoin(
|
|
ComputerType, Computer.computertypeid == ComputerType.computertypeid
|
|
).filter(
|
|
db.or_(
|
|
Asset.assetnumber.ilike(pattern),
|
|
Asset.name.ilike(pattern),
|
|
Asset.serialnumber.ilike(pattern),
|
|
# The optional identifiers too (ADR-001). Global search matches
|
|
# these, and a tag read off the machine has to find it here as
|
|
# well - this box is where someone holding the label looks.
|
|
Asset.gaugelabreference.ilike(pattern),
|
|
Asset.maintenancereference.ilike(pattern),
|
|
Computer.hostname.ilike(pattern),
|
|
ComputerType.computertype.ilike(pattern)
|
|
)
|
|
)
|
|
|
|
# Computer type filter
|
|
if type_id := request.args.get('typeid', request.args.get('type_id')):
|
|
query = query.filter(Computer.computertypeid == int(type_id))
|
|
|
|
# OS filter
|
|
if os_id := request.args.get('osid', request.args.get('os_id')):
|
|
query = query.filter(Computer.osid == int(os_id))
|
|
|
|
# Location filter
|
|
if location_id := request.args.get('locationid', request.args.get('location_id')):
|
|
query = query.filter(Asset.locationid == int(location_id))
|
|
|
|
# Business unit filter
|
|
if bu_id := request.args.get('businessunitid', request.args.get('businessunit_id')):
|
|
query = query.filter(Asset.businessunitid == int(bu_id))
|
|
|
|
# Shopfloor filter (by the Shopfloor computer type)
|
|
if shopfloor := request.args.get('shopfloor'):
|
|
sf = ComputerType.query.filter_by(computertype='Shopfloor').first()
|
|
sf_id = sf.computertypeid if sf else -1
|
|
if shopfloor.lower() == 'true':
|
|
query = query.filter(Computer.computertypeid == sf_id)
|
|
else:
|
|
query = query.filter(db.or_(Computer.computertypeid != sf_id,
|
|
Computer.computertypeid.is_(None)))
|
|
|
|
# Sorting
|
|
sort_by = request.args.get('sort', 'hostname')
|
|
sort_dir = request.args.get('dir', 'asc')
|
|
|
|
if sort_by == 'hostname':
|
|
col = Computer.hostname
|
|
elif sort_by == 'assetnumber':
|
|
col = Asset.assetnumber
|
|
elif sort_by == 'name':
|
|
col = Asset.name
|
|
elif sort_by == 'lastreporteddate':
|
|
col = Computer.lastreporteddate
|
|
else:
|
|
col = Computer.hostname
|
|
|
|
query = query.order_by(col.desc() if sort_dir == 'desc' else col)
|
|
|
|
items, total = paginate_query(query, page, per_page)
|
|
|
|
# Build response with both asset and computer data
|
|
data = []
|
|
accessdomain = _pc_access_domain()
|
|
for comp in items:
|
|
item = comp.asset.to_dict() if comp.asset else {}
|
|
item['computer'] = comp.to_dict()
|
|
item['accessmethods'] = _computer_access_links(comp, domain=accessdomain)
|
|
data.append(item)
|
|
|
|
return paginated_response(data, page, per_page, total)
|
|
|
|
|
|
@computers_bp.route('/<int:computer_id>', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def get_computer(computer_id: int):
|
|
"""Get a single computer with full details."""
|
|
comp = db.session.get(Computer, computer_id)
|
|
|
|
if not comp:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Computer with ID {computer_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
result = comp.asset.to_dict() if comp.asset else {}
|
|
result['computer'] = comp.to_dict()
|
|
result['communications'] = [
|
|
c.to_dict() for c in
|
|
Communication.query.filter_by(assetid=comp.assetid).all()
|
|
]
|
|
result['accessmethods'] = _computer_access_links(comp)
|
|
|
|
return success_response(result)
|
|
|
|
|
|
@computers_bp.route('/by-asset/<int:asset_id>', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def get_computer_by_asset(asset_id: int):
|
|
"""Get computer data by asset ID."""
|
|
comp = Computer.query.filter_by(assetid=asset_id).first()
|
|
|
|
if not comp:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Computer for asset {asset_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
result = comp.asset.to_dict() if comp.asset else {}
|
|
result['computer'] = comp.to_dict()
|
|
|
|
return success_response(result)
|
|
|
|
|
|
@computers_bp.route('/by-hostname/<hostname>', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def get_computer_by_hostname(hostname: str):
|
|
"""Get computer by hostname."""
|
|
comp = Computer.query.filter_by(hostname=hostname).first()
|
|
|
|
if not comp:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Computer with hostname {hostname} not found',
|
|
http_code=404
|
|
)
|
|
|
|
result = comp.asset.to_dict() if comp.asset else {}
|
|
result['computer'] = comp.to_dict()
|
|
|
|
return success_response(result)
|
|
|
|
|
|
@computers_bp.route('', methods=['POST'])
|
|
@jwt_required()
|
|
@require_permission('computers.create')
|
|
def create_computer():
|
|
"""
|
|
Create new computer (creates both Asset and Computer records).
|
|
|
|
Required fields:
|
|
- assetnumber: Business identifier
|
|
|
|
Optional fields:
|
|
- name, serialnumber, statusid, locationid, businessunitid
|
|
- computertypeid, hostname, osid
|
|
- mapx, mapy, notes
|
|
"""
|
|
data = request.get_json()
|
|
|
|
if not data:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
|
|
|
|
if not data.get('assetnumber'):
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'assetnumber is required')
|
|
|
|
# Check for duplicate assetnumber
|
|
if Asset.query.filter_by(assetnumber=data['assetnumber']).first():
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"Asset with number '{data['assetnumber']}' already exists",
|
|
http_code=409
|
|
)
|
|
|
|
# Check for duplicate hostname
|
|
if data.get('hostname'):
|
|
if Computer.query.filter_by(hostname=data['hostname']).first():
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"Computer with hostname '{data['hostname']}' already exists",
|
|
http_code=409
|
|
)
|
|
|
|
# Get computer asset type
|
|
computer_type = AssetType.query.filter_by(assettype='computer').first()
|
|
if not computer_type:
|
|
return error_response(
|
|
ErrorCodes.INTERNAL_ERROR,
|
|
'Computer asset type not found. Plugin may not be properly installed.',
|
|
http_code=500
|
|
)
|
|
|
|
# Create the core asset
|
|
asset = Asset(
|
|
assetnumber=data['assetnumber'],
|
|
name=data.get('name'),
|
|
serialnumber=data.get('serialnumber'),
|
|
gaugelabreference=data.get('gaugelabreference'),
|
|
maintenancereference=data.get('maintenancereference'),
|
|
assettypeid=computer_type.assettypeid,
|
|
statusid=data.get('statusid', 1),
|
|
locationid=data.get('locationid'),
|
|
businessunitid=data.get('businessunitid'),
|
|
mapx=data.get('mapx'),
|
|
levelid=data.get('levelid'),
|
|
mapy=data.get('mapy'),
|
|
notes=data.get('notes')
|
|
)
|
|
|
|
db.session.add(asset)
|
|
db.session.flush() # Get the assetid
|
|
|
|
# Create the computer extension
|
|
comp = Computer(
|
|
assetid=asset.assetid,
|
|
computertypeid=data.get('computertypeid'),
|
|
hostname=data.get('hostname'),
|
|
osid=data.get('osid'),
|
|
vendorid=data.get('vendorid'),
|
|
modelnumberid=data.get('modelnumberid'),
|
|
loggedinuser=data.get('loggedinuser'),
|
|
lastreporteddate=data.get('lastreporteddate'),
|
|
lastboottime=data.get('lastboottime')
|
|
)
|
|
|
|
db.session.add(comp)
|
|
db.session.flush()
|
|
|
|
# Optional primary IP communication
|
|
if data.get('ipaddress'):
|
|
ip_comtype = CommunicationType.query.filter_by(comtype='IP').first()
|
|
if ip_comtype:
|
|
db.session.add(Communication(
|
|
assetid=asset.assetid,
|
|
comtypeid=ip_comtype.comtypeid,
|
|
ipaddress=data['ipaddress'],
|
|
isprimary=True,
|
|
))
|
|
|
|
# Remote-access protocols
|
|
_sync_access_methods(comp, data)
|
|
|
|
# Preserve legacy timestamps in import mode (no-op otherwise)
|
|
apply_import_timestamps(asset, data)
|
|
|
|
# Audit log
|
|
AuditLog.log('created', 'Computer', entityid=comp.computerid,
|
|
entityname=data.get('hostname') or data['assetnumber'])
|
|
|
|
db.session.commit()
|
|
|
|
result = asset.to_dict()
|
|
result['computer'] = comp.to_dict()
|
|
result['accessmethods'] = _computer_access_links(comp)
|
|
|
|
return success_response(result, message='Computer created', http_code=201)
|
|
|
|
|
|
@computers_bp.route('/<int:computer_id>', methods=['PUT'])
|
|
@jwt_required()
|
|
@require_permission('computers.edit')
|
|
def update_computer(computer_id: int):
|
|
"""Update computer (both Asset and Computer records)."""
|
|
comp = db.session.get(Computer, computer_id)
|
|
|
|
if not comp:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Computer with ID {computer_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
data = request.get_json()
|
|
if not data:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
|
|
|
|
asset = comp.asset
|
|
|
|
# Check for conflicting assetnumber
|
|
if 'assetnumber' in data and data['assetnumber'] != asset.assetnumber:
|
|
if Asset.query.filter_by(assetnumber=data['assetnumber']).first():
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"Asset with number '{data['assetnumber']}' already exists",
|
|
http_code=409
|
|
)
|
|
|
|
# Check for conflicting hostname
|
|
if 'hostname' in data and data['hostname'] != comp.hostname:
|
|
existing = Computer.query.filter_by(hostname=data['hostname']).first()
|
|
if existing and existing.computerid != computer_id:
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"Computer with hostname '{data['hostname']}' already exists",
|
|
http_code=409
|
|
)
|
|
|
|
# Track changes for audit log
|
|
changes = {}
|
|
|
|
# Update asset fields
|
|
asset_fields = ['assetnumber', 'name', 'serialnumber', 'gaugelabreference',
|
|
'maintenancereference', 'statusid',
|
|
'locationid', 'businessunitid', 'mapx', 'mapy', 'levelid', 'notes', 'isactive']
|
|
for key in asset_fields:
|
|
if key in data:
|
|
old_val = getattr(asset, key)
|
|
new_val = data[key]
|
|
if old_val != new_val:
|
|
changes[key] = {'old': old_val, 'new': new_val}
|
|
setattr(asset, key, data[key])
|
|
|
|
# Update computer fields
|
|
computer_fields = ['computertypeid', 'hostname', 'osid', 'vendorid',
|
|
'modelnumberid', 'loggedinuser', 'lastreporteddate',
|
|
'lastboottime']
|
|
for key in computer_fields:
|
|
if key in data:
|
|
old_val = getattr(comp, key)
|
|
new_val = data[key]
|
|
if old_val != new_val:
|
|
changes[key] = {'old': old_val, 'new': new_val}
|
|
setattr(comp, key, data[key])
|
|
|
|
# Upsert the primary IP communication so a single PUT covers it
|
|
if 'ipaddress' in data:
|
|
ip = (data.get('ipaddress') or '').strip()
|
|
primary = Communication.query.filter_by(
|
|
assetid=asset.assetid, isprimary=True).first()
|
|
if ip:
|
|
if primary:
|
|
primary.ipaddress = ip
|
|
else:
|
|
ip_comtype = CommunicationType.query.filter_by(comtype='IP').first()
|
|
if ip_comtype:
|
|
db.session.add(Communication(
|
|
assetid=asset.assetid, comtypeid=ip_comtype.comtypeid,
|
|
ipaddress=ip, isprimary=True))
|
|
elif primary:
|
|
primary.ipaddress = None
|
|
|
|
# Remote-access protocols
|
|
_sync_access_methods(comp, data)
|
|
|
|
# Audit log if there were changes
|
|
if changes:
|
|
AuditLog.log('updated', 'Computer', entityid=comp.computerid,
|
|
entityname=comp.hostname or asset.assetnumber, changes=changes)
|
|
|
|
apply_import_timestamps(asset, data)
|
|
db.session.commit()
|
|
|
|
result = asset.to_dict()
|
|
result['computer'] = comp.to_dict()
|
|
result['accessmethods'] = _computer_access_links(comp)
|
|
|
|
return success_response(result, message='Computer updated')
|
|
|
|
|
|
@computers_bp.route('/<int:computer_id>', methods=['DELETE'])
|
|
@jwt_required()
|
|
@require_permission('computers.delete')
|
|
def delete_computer(computer_id: int):
|
|
"""Delete (soft delete) computer."""
|
|
comp = db.session.get(Computer, computer_id)
|
|
|
|
if not comp:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Computer with ID {computer_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
# Soft delete the asset
|
|
comp.asset.isactive = False
|
|
|
|
# Audit log
|
|
AuditLog.log('deleted', 'Computer', entityid=comp.computerid,
|
|
entityname=comp.hostname or comp.asset.assetnumber)
|
|
|
|
db.session.commit()
|
|
|
|
return success_response(message='Computer deleted')
|
|
|
|
|
|
# =============================================================================
|
|
# Installed Applications
|
|
# =============================================================================
|
|
|
|
@computers_bp.route('/<int:computer_id>/apps', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def get_installed_apps(computer_id: int):
|
|
"""Get all installed applications for a computer."""
|
|
comp = db.session.get(Computer, computer_id)
|
|
|
|
if not comp:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Computer with ID {computer_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
apps = ComputerInstalledApp.query.filter_by(
|
|
computerid=computer_id,
|
|
isactive=True
|
|
).all()
|
|
|
|
data = [app.to_dict() for app in apps]
|
|
|
|
return success_response(data)
|
|
|
|
|
|
@computers_bp.route('/<int:computer_id>/apps', methods=['POST'])
|
|
@jwt_required()
|
|
@require_permission('computers.create')
|
|
def add_installed_app(computer_id: int):
|
|
"""Add an installed application to a computer."""
|
|
comp = db.session.get(Computer, computer_id)
|
|
|
|
if not comp:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Computer with ID {computer_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
data = request.get_json()
|
|
if not data or not data.get('appid'):
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'appid is required')
|
|
|
|
appid = data['appid']
|
|
|
|
# Validate app exists
|
|
if not db.session.get(Application, appid):
|
|
return error_response(ErrorCodes.NOT_FOUND, f'Application {appid} not found', http_code=404)
|
|
|
|
# Check for duplicate
|
|
existing = ComputerInstalledApp.query.filter_by(
|
|
computerid=computer_id,
|
|
appid=appid
|
|
).first()
|
|
|
|
if existing:
|
|
if existing.isactive:
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
'This application is already installed on this computer',
|
|
http_code=409
|
|
)
|
|
else:
|
|
# Reactivate
|
|
existing.isactive = True
|
|
existing.appversionid = data.get('appversionid')
|
|
db.session.commit()
|
|
return success_response(existing.to_dict(), message='Application reinstalled')
|
|
|
|
# Create new installation record
|
|
installed = ComputerInstalledApp(
|
|
computerid=computer_id,
|
|
appid=appid,
|
|
appversionid=data.get('appversionid')
|
|
)
|
|
|
|
db.session.add(installed)
|
|
db.session.commit()
|
|
|
|
return success_response(installed.to_dict(), message='Application installed', http_code=201)
|
|
|
|
|
|
@computers_bp.route('/<int:computer_id>/apps/<int:app_id>', methods=['DELETE'])
|
|
@jwt_required()
|
|
@require_permission('computers.delete')
|
|
def remove_installed_app(computer_id: int, app_id: int):
|
|
"""Remove an installed application from a computer."""
|
|
installed = ComputerInstalledApp.query.filter_by(
|
|
computerid=computer_id,
|
|
appid=app_id,
|
|
isactive=True
|
|
).first()
|
|
|
|
if not installed:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
'Installation record not found',
|
|
http_code=404
|
|
)
|
|
|
|
installed.isactive = False
|
|
db.session.commit()
|
|
|
|
return success_response(message='Application uninstalled')
|
|
|
|
|
|
# =============================================================================
|
|
# Status Reporting
|
|
# =============================================================================
|
|
|
|
@computers_bp.route('/<int:computer_id>/report', methods=['POST'])
|
|
@jwt_required()
|
|
@require_permission('computers.create')
|
|
def report_status(computer_id: int):
|
|
"""
|
|
Report computer status (for agent-based reporting).
|
|
|
|
This endpoint can be called periodically by a client agent
|
|
to update status information.
|
|
"""
|
|
comp = db.session.get(Computer, computer_id)
|
|
|
|
if not comp:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Computer with ID {computer_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
data = request.get_json() or {}
|
|
|
|
# Update status fields
|
|
from datetime import datetime, timezone
|
|
comp.lastreporteddate = datetime.now(timezone.utc).replace(tzinfo=None)
|
|
|
|
if 'loggedinuser' in data:
|
|
comp.loggedinuser = data['loggedinuser']
|
|
if 'lastboottime' in data:
|
|
comp.lastboottime = data['lastboottime']
|
|
|
|
db.session.commit()
|
|
|
|
return success_response(message='Status reported')
|
|
|
|
|
|
# =============================================================================
|
|
# Dashboard
|
|
# =============================================================================
|
|
|
|
@computers_bp.route('/dashboard/summary', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def dashboard_summary():
|
|
"""Get computer dashboard summary data."""
|
|
# Total active computers
|
|
total = db.session.query(Computer).join(Asset).filter(
|
|
Asset.isactive == True
|
|
).count()
|
|
|
|
# Count by computer type
|
|
by_type = db.session.query(
|
|
ComputerType.computertype,
|
|
db.func.count(Computer.computerid)
|
|
).join(Computer, Computer.computertypeid == ComputerType.computertypeid
|
|
).join(Asset, Asset.assetid == Computer.assetid
|
|
).filter(Asset.isactive == True
|
|
).group_by(ComputerType.computertype
|
|
).all()
|
|
|
|
# Count by OS
|
|
by_os = db.session.query(
|
|
OperatingSystem.osname,
|
|
db.func.count(Computer.computerid)
|
|
).join(Computer, Computer.osid == OperatingSystem.osid
|
|
).join(Asset, Asset.assetid == Computer.assetid
|
|
).filter(Asset.isactive == True
|
|
).group_by(OperatingSystem.osname
|
|
).all()
|
|
|
|
# Count shopfloor vs non-shopfloor
|
|
sf = ComputerType.query.filter_by(computertype='Shopfloor').first()
|
|
shopfloor_count = db.session.query(Computer).join(Asset).filter(
|
|
Asset.isactive == True,
|
|
Computer.computertypeid == (sf.computertypeid if sf else -1)
|
|
).count()
|
|
|
|
return success_response({
|
|
'total': total,
|
|
'bytype': [{'type': t, 'count': c} for t, c in by_type],
|
|
'byos': [{'os': o, 'count': c} for o, c in by_os],
|
|
'shopfloor': shopfloor_count,
|
|
'nonshopfloor': total - shopfloor_count
|
|
})
|
|
|
|
|
|
@computers_bp.route('/dashboard/quiet', methods=['GET'])
|
|
@jwt_required()
|
|
@require_permission('computers.view')
|
|
def dashboard_quiet():
|
|
"""PCs that WERE reporting, have stopped, and are still meant to be in use.
|
|
|
|
Three filters, and each removes a population that would otherwise be noise:
|
|
|
|
Never reported at all is EXCLUDED. Those are usually records created by
|
|
hand or imported, not bays that broke - listing them buries the PCs that
|
|
actually changed state behind rows nobody is going to act on.
|
|
|
|
Not In Use is EXCLUDED. A PC in Repair, Inventory or Retired is silent on
|
|
purpose; that is the status doing its job, not a fault.
|
|
|
|
Soft-deleted is excluded for the same reason.
|
|
|
|
What is left is the real signal: a machine that was working, is not now, and
|
|
nobody has marked as anything else. Silence is the only evidence available -
|
|
there is no heartbeat separate from the report.
|
|
"""
|
|
from datetime import datetime, timedelta, timezone
|
|
from shopdb.api import AssetStatus
|
|
|
|
hours = 24
|
|
setting = Setting.query.filter_by(key='computers_quietreporthours').first()
|
|
if setting and (setting.value or '').strip():
|
|
try:
|
|
hours = int(setting.value)
|
|
except (TypeError, ValueError):
|
|
pass
|
|
cutoff = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(hours=hours)
|
|
now = datetime.now(timezone.utc).replace(tzinfo=None)
|
|
|
|
rows = (db.session.query(Computer, Asset)
|
|
.join(Asset, Asset.assetid == Computer.assetid)
|
|
.join(AssetStatus, AssetStatus.statusid == Asset.statusid)
|
|
.filter(Asset.isactive.is_(True),
|
|
AssetStatus.status == 'In Use',
|
|
Computer.lastreporteddate.isnot(None),
|
|
Computer.lastreporteddate < cutoff)
|
|
.all())
|
|
|
|
out = [{
|
|
'computerid': comp.computerid,
|
|
'hostname': comp.hostname,
|
|
'lastreported': comp.lastreporteddate.isoformat() + 'Z',
|
|
'quietdays': (now - comp.lastreporteddate).days,
|
|
} for comp, _asset in rows]
|
|
|
|
out.sort(key=lambda r: -r['quietdays'])
|
|
return success_response(out[:50])
|
|
|
|
|
|
@computers_bp.route('/dashboard/sharedmachines', methods=['GET'])
|
|
@jwt_required()
|
|
@require_permission('computers.view')
|
|
def dashboard_sharedmachines():
|
|
"""Machine numbers claimed by more than one PC, with nothing filed under
|
|
them.
|
|
|
|
Several devices genuinely sharing a number is legitimate - part markers do
|
|
it - and those are modelled: each device is its own asset filed `partof` the
|
|
operation, so the operation has CHILD ASSETS. Two PCs carrying the same
|
|
machine number by mistake looks identical from a count and has none. That
|
|
distinction is the whole card; without it this would list correct data
|
|
alongside faults and get ignored.
|
|
|
|
Promoted from `flask relationships check-shared-machines`, which answers the
|
|
same question and which nobody will remember to run. This one found seven
|
|
mis-numbered bays that had been that way for weeks.
|
|
"""
|
|
from shopdb.api import AssetRelationship, RelationshipType
|
|
from sqlalchemy.orm import aliased
|
|
|
|
controls = RelationshipType.query.filter_by(
|
|
relationshiptype='controls').first()
|
|
partof = RelationshipType.query.filter_by(relationshiptype='partof').first()
|
|
if not controls:
|
|
return success_response([])
|
|
|
|
pcasset = aliased(Asset)
|
|
machineasset = aliased(Asset)
|
|
|
|
rows = (db.session.query(machineasset.assetid, machineasset.assetnumber,
|
|
pcasset.assetnumber)
|
|
.select_from(AssetRelationship)
|
|
.join(pcasset, AssetRelationship.sourceassetid == pcasset.assetid)
|
|
.join(machineasset,
|
|
AssetRelationship.targetassetid == machineasset.assetid)
|
|
.filter(AssetRelationship.relationshiptypeid ==
|
|
controls.relationshiptypeid,
|
|
AssetRelationship.label == 'collector:machine',
|
|
AssetRelationship.isactive.is_(True))
|
|
.all())
|
|
|
|
bymachine = {}
|
|
for assetid, machinenumber, pcnumber in rows:
|
|
bymachine.setdefault((assetid, machinenumber), []).append(pcnumber)
|
|
|
|
out = []
|
|
for (assetid, machinenumber), pcs in bymachine.items():
|
|
if len(pcs) < 2:
|
|
continue
|
|
children = 0
|
|
if partof:
|
|
children = (AssetRelationship.query
|
|
.filter_by(targetassetid=assetid,
|
|
relationshiptypeid=partof.relationshiptypeid,
|
|
isactive=True)
|
|
.count())
|
|
if children:
|
|
continue # modelled: the devices are assets in their own right
|
|
out.append({
|
|
'assetid': assetid,
|
|
'machinenumber': machinenumber,
|
|
'pccount': len(pcs),
|
|
'pcs': ', '.join(sorted(p for p in pcs if p)),
|
|
})
|
|
|
|
out.sort(key=lambda r: -r['pccount'])
|
|
return success_response(out)
|
|
|
|
|
|
# =============================================================================
|
|
# Collector client script
|
|
# =============================================================================
|
|
|
|
CLIENT_SCRIPT_NAME = 'Report-AssetToShopDB.ps1'
|
|
|
|
|
|
def _client_script_path():
|
|
"""The reporter shipped with this plugin, which is the single source."""
|
|
import os
|
|
return os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
|
'client', CLIENT_SCRIPT_NAME)
|
|
|
|
|
|
def _setting_value(key):
|
|
row = Setting.query.filter_by(key=key).first()
|
|
return ((row.value if row else '') or '').strip()
|
|
|
|
|
|
def _generate_client_script(source: str, baseurl: str, ranges: str,
|
|
version: str, generatedon: str) -> str:
|
|
"""Stamp a site's own values into the reporter's parameter defaults.
|
|
|
|
ONLY the defaults are substituted, never the body: the file in the repo
|
|
stays runnable as-is, so there is no second copy to drift. Everything
|
|
stamped here is overridable at runtime - the parameter still wins, then the
|
|
registry - because a bay may need to differ from its site.
|
|
|
|
The collector key is NOT stamped in. This file lands on every shop-floor PC,
|
|
and a token in a file on hundreds of bays cannot be rotated quietly; it is
|
|
read from HKLM:\\SOFTWARE\\GE\\ShopDB CollectorKey, provisioned per
|
|
ADOPTING-AT-ANOTHER-SITE.md.
|
|
"""
|
|
apiurl = baseurl.rstrip('/') + '/api/collector/computers' if baseurl else ''
|
|
header = (
|
|
'# GENERATED by ShopDB {version} on {generatedon}\n'
|
|
'# for {baseurl}\n'
|
|
'#\n'
|
|
'# Re-download after upgrading ShopDB: this copy matches that server\'s\n'
|
|
'# collector contract. Edits here are lost on the next download - change\n'
|
|
'# the site settings instead, or pass -ApiUrl / -AllowedRanges.\n'
|
|
'#\n'
|
|
'# The collector key is deliberately NOT in this file. Provision it as\n'
|
|
'# HKLM:\\SOFTWARE\\GE\\ShopDB CollectorKey - see the adoption guide.\n'
|
|
'\n'
|
|
).format(version=version, generatedon=generatedon,
|
|
baseurl=baseurl or 'an unconfigured site (set site_base_url)')
|
|
|
|
out = source
|
|
if apiurl:
|
|
old = "[string]$ApiUrl = ''"
|
|
assert old in out, 'the reporter no longer declares $ApiUrl as expected'
|
|
out = out.replace(old, "[string]$ApiUrl = '{0}'".format(apiurl), 1)
|
|
if ranges:
|
|
old = "[string]$AllowedRanges = ''"
|
|
assert old in out, 'the reporter no longer declares $AllowedRanges as expected'
|
|
out = out.replace(old, "[string]$AllowedRanges = '{0}'".format(ranges), 1)
|
|
return header + out
|
|
|
|
|
|
@computers_bp.route('/client-script', methods=['GET'])
|
|
@jwt_required()
|
|
@require_role('admin')
|
|
def download_client_script():
|
|
"""The collector reporter, stamped with THIS site's values.
|
|
|
|
Admin-only. It carries no secret, but it does state a site's URL and its
|
|
internal ranges, which is configuration rather than something to hand out.
|
|
"""
|
|
import datetime
|
|
import hashlib
|
|
import os
|
|
|
|
path = _client_script_path()
|
|
if not os.path.isfile(path):
|
|
return error_response(ErrorCodes.NOT_FOUND,
|
|
'The collector script is not present in this install',
|
|
http_code=404)
|
|
|
|
with open(path, 'r', encoding='utf-8') as handle:
|
|
source = handle.read()
|
|
|
|
# A site that has not set its public URL still gets a usable script: the
|
|
# browsing origin is the server the admin is talking to right now.
|
|
baseurl = _setting_value('site_base_url') or request.url_root
|
|
# From config, not an import: a plugin reaching into core is an ADR-002
|
|
# violation and the contract test fails the build for it.
|
|
version = current_app.config.get('VERSION') or 'unknown'
|
|
generatedon = datetime.datetime.now(datetime.timezone.utc).strftime('%Y-%m-%d')
|
|
|
|
try:
|
|
body = _generate_client_script(
|
|
source, baseurl.strip(), _setting_value('computers_routableranges'),
|
|
version, generatedon)
|
|
except AssertionError as exc:
|
|
return error_response(ErrorCodes.INTERNAL_ERROR, str(exc), http_code=500)
|
|
|
|
digest = hashlib.sha256(body.encode('utf-8')).hexdigest()
|
|
return Response(
|
|
body,
|
|
mimetype='text/plain; charset=utf-8',
|
|
headers={
|
|
'Content-Disposition': 'attachment; filename={0}'.format(CLIENT_SCRIPT_NAME),
|
|
# Published so a deployment can verify what it fetched, the same way
|
|
# the installer publishes one.
|
|
'X-Script-Sha256': digest,
|
|
})
|