7151b68bdd
Wave one complete. Three cards, no new data and no migrations. Printer supplies reuses the existing low-supplies query and its five-minute cache; a Zabbix round-trip per printer on every dashboard load would make this the slowest page in the app. One row per printer listing every depleted cartridge, criticals first - a row per cartridge would report one printer three times and read as three problems, and showing only the worst class would hide a low cartridge behind a critical one on the same machine when whoever walks out there wants to carry both. While there: the low-supplies REPORT itself was including healthy cartridges. A printer with one empty black and three full colour ones listed all four, so the reader had to find the problem inside the row. It now lists only what needs replacing, and the test that asserted the old behaviour now asserts the new. Expiring warranties keeps already-expired entries on the list rather than dropping them the day they lapse, which is how they get missed. Horizon is warranty_expiringdays, default 90, because that suits a site budgeting quarterly and nobody else. Mis-numbered bays promotes check-shared-machines out of a CLI command nobody will remember to run - it found seven bays that had been wrong for weeks. It reports only numbers with NO child assets, so part markers legitimately sharing an operation stay silent: that distinction is the whole card, and without it it would list correct data beside faults and be ignored. Printers also loses its dead component-named widget; notifications, network and machines still have theirs.
1067 lines
37 KiB
Python
1067 lines
37 KiB
Python
"""Computers plugin API endpoints."""
|
|
|
|
from flask import Blueprint, request
|
|
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, 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')
|
|
or 'device.geaerospace.net').strip().strip('.')
|
|
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),
|
|
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'),
|
|
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', '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 have stopped reporting, and PCs that never started.
|
|
|
|
Two populations, deliberately distinguished rather than merged into one
|
|
count. A PC that reported and went quiet is probably switched off, moved or
|
|
broken. A PC that has NEVER reported is worse: it is either not enrolled, or
|
|
enrolled against the wrong pc-type, so nothing enforces anything on it and
|
|
no backup of it exists. That one hides indefinitely because nothing about it
|
|
is failing loudly - and it is exactly the shape of the bay that carried a
|
|
wrong machine number for weeks.
|
|
|
|
Silence is the only signal available here. There is no heartbeat separate
|
|
from the report, so 'stopped reporting' is measured against the collector's
|
|
own last write.
|
|
"""
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
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)
|
|
|
|
query = (db.session.query(Computer, Asset)
|
|
.join(Asset, Asset.assetid == Computer.assetid)
|
|
.filter(Asset.isactive.is_(True)))
|
|
|
|
rows = []
|
|
for comp, asset in query.all():
|
|
last = comp.lastreporteddate
|
|
if last is None:
|
|
rows.append({
|
|
'computerid': comp.computerid,
|
|
'hostname': comp.hostname,
|
|
'state': 'never reported',
|
|
'lastreported': None,
|
|
'quietdays': None,
|
|
})
|
|
elif last < cutoff:
|
|
rows.append({
|
|
'computerid': comp.computerid,
|
|
'hostname': comp.hostname,
|
|
'state': 'stopped reporting',
|
|
'lastreported': last.isoformat() + 'Z',
|
|
'quietdays': (datetime.now(timezone.utc).replace(tzinfo=None)
|
|
- last).days,
|
|
})
|
|
|
|
# Never-reported first, then longest silence: the order someone should work
|
|
# down the list, not the order the rows came out of the table.
|
|
rows.sort(key=lambda r: (r['quietdays'] is not None, -(r['quietdays'] or 0)))
|
|
return success_response(rows[: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)
|