b16f143467
Every asset list shows a Type column (and printers a Model, machines and network a Vendor), but the search filters only looked at the asset number, name, serial and hostname. Searching a type returned zero rows: 'Part Washer' on machines, 'Standard' on PCs, 'Thermal' on printers. Extend the search on machines, computers, printers, network devices, measuring tools and the unified asset list to cover the type name plus the vendor/model where the list shows them. Joins are outer joins so an asset missing a type or vendor still matches on its own fields; the core list uses a correlated EXISTS instead, since its type-name filter already joins AssetType.
937 lines
32 KiB
Python
937 lines
32 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
|
|
})
|