94d8d6c9b6
"All assets 704" sat beside "all assets in use 737", and both were correct about different populations. The totals summed five specific asset types and subtracted dual-bay secondaries; the status counts took every asset row of any type with no collapse, so USB devices and hidden secondary bays inflated one side of a comparison the layout invites. Status is now counted over exactly the same assets the totals describe. Warranty rows fell back to asset.name when the covered asset had no hostname, and an asset's name is usually the MACHINE's descriptive name - which is how a column meant to identify a PC ended up showing a machine. Hostname, else the asset number, never the name. The machine number loses its label too: the row is hostname, machine, state, and "machine 3015" spends a word on what position already conveys. Printer names now carry the floor-plan preview on hover, the same LocationMapTooltip the printer's own page uses - a location name tells you the room, the map tells you where to walk. Declared as map.maphover on the card, so any card with coordinates gets it; a row without them shows a plain link rather than being dropped. Cards are four across rather than five. At five columns a row holding a hostname, a machine number and a state truncates on exactly the rows that matter. auto-fit, so two cards fill the width instead of leaving empty tracks. Not covered by a test: the count fix. I started one and it was interrupted, and I have not gone back for it - the assertion worth having is that in-use can never exceed the total.
469 lines
19 KiB
Python
469 lines
19 KiB
Python
"""Warranty API: manual CRUD now, provider refresh stubbed for later phases.
|
|
|
|
Coverage status is derived from enddate at read time (see models.derive_status),
|
|
never stored. Warranties link to assets many-to-many via warrantyassets, though
|
|
the common case is one warranty per asset.
|
|
"""
|
|
|
|
from datetime import date, datetime, timedelta, timezone
|
|
|
|
from flask import Blueprint, request
|
|
from flask_jwt_extended import jwt_required
|
|
from sqlalchemy.orm import joinedload
|
|
|
|
from shopdb.api import (
|
|
db, Asset,
|
|
success_response, error_response, ErrorCodes,
|
|
require_permission,
|
|
)
|
|
|
|
from ..models import Warranty, WarrantyAsset
|
|
from ..services import get_provider, ProviderNotConfigured, WarrantyLookupError
|
|
|
|
warranty_bp = Blueprint('warranty', __name__)
|
|
|
|
|
|
def _parse_date(value):
|
|
"""Accept 'YYYY-MM-DD' (or None/empty) -> date or None."""
|
|
if not value:
|
|
return None
|
|
try:
|
|
return datetime.strptime(value[:10], '%Y-%m-%d').date()
|
|
except (ValueError, TypeError):
|
|
return None
|
|
|
|
|
|
def _related_machine(asset):
|
|
"""The machine a covered asset is associated with, or None.
|
|
|
|
A shopfloor PC is bought, warranted and replaced as a PC, but it is FOUND
|
|
by the machine it drives - nobody walks the floor looking for an asset
|
|
number. So a warranty row for a PC carries the machine's number, and its
|
|
map position so the row can point at where to go.
|
|
|
|
Walks the asset relationship graph in BOTH directions: the canonical edge is
|
|
PC --controls--> machine, but a dual-bay pair carries controls on both bays
|
|
and hand-made links are not guaranteed to be oriented. Returns the first
|
|
active related asset whose type is 'machine'.
|
|
"""
|
|
from shopdb.api import Asset as CoreAsset
|
|
|
|
related = []
|
|
for rel in (getattr(asset, 'outgoing_relationships', None) or []):
|
|
if rel.isactive and rel.targetasset:
|
|
related.append(rel.targetasset)
|
|
for rel in (getattr(asset, 'incoming_relationships', None) or []):
|
|
if rel.isactive and rel.sourceasset:
|
|
related.append(rel.sourceasset)
|
|
|
|
for candidate in related:
|
|
typename = candidate.assettype.assettype if candidate.assettype else None
|
|
if typename != 'machine':
|
|
continue
|
|
return {
|
|
'assetid': candidate.assetid,
|
|
'machinenumber': candidate.assetnumber,
|
|
'name': candidate.name,
|
|
# Map position for the hover preview. Either may be None: a machine
|
|
# that has never been placed on the floor map still has a number
|
|
# worth showing, so the caller decides what to do with a missing
|
|
# position rather than the row being dropped.
|
|
'mapx': candidate.mapx,
|
|
'mapy': candidate.mapy,
|
|
'locationid': candidate.locationid,
|
|
'locationname': (candidate.location.locationname
|
|
if candidate.location else None),
|
|
}
|
|
return None
|
|
|
|
|
|
def _asset_summary(asset):
|
|
return {
|
|
'assetid': asset.assetid,
|
|
'assetnumber': asset.assetnumber,
|
|
'name': asset.name,
|
|
'assettypename': asset.assettype.assettype if asset.assettype else None,
|
|
'machine': _related_machine(asset),
|
|
}
|
|
|
|
|
|
def _warranty_payload(warranty, today=None, assetmap=None):
|
|
"""to_dict plus the linked-asset summaries. Pass assetmap (assetid -> Asset)
|
|
to avoid a per-link query when serializing a list; without it, falls back to
|
|
a per-link get (fine for a single warranty)."""
|
|
data = warranty.to_dict(today)
|
|
assets = []
|
|
for link in warranty.links:
|
|
asset = assetmap.get(link.assetid) if assetmap is not None \
|
|
else db.session.get(Asset, link.assetid)
|
|
if asset:
|
|
assets.append(_asset_summary(asset))
|
|
data['assets'] = assets
|
|
return data
|
|
|
|
|
|
def _apply_links(warranty, assetids):
|
|
"""Replace a warranty's asset links with the given asset id list."""
|
|
if assetids is None:
|
|
return
|
|
wanted = {int(a) for a in assetids if str(a).strip()}
|
|
existing = {link.assetid: link for link in warranty.links}
|
|
for assetid in wanted - set(existing):
|
|
if db.session.get(Asset, assetid):
|
|
warranty.links.append(WarrantyAsset(assetid=assetid))
|
|
for assetid in set(existing) - wanted:
|
|
warranty.links.remove(existing[assetid])
|
|
|
|
|
|
# =============================================================================
|
|
# CRUD
|
|
# =============================================================================
|
|
|
|
@warranty_bp.route('', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def list_warranties():
|
|
"""List warranties. Filters: ?status=, ?assetid=, ?active=false."""
|
|
query = Warranty.query
|
|
if request.args.get('active', 'true').lower() != 'false':
|
|
query = query.filter_by(isactive=True)
|
|
# Exact-match natural-key lookup for idempotent import (servicetag + vendor).
|
|
if exactservicetag := request.args.get('servicetag'):
|
|
query = query.filter(Warranty.servicetag == exactservicetag)
|
|
if exactvendor := request.args.get('vendor'):
|
|
query = query.filter(Warranty.vendor == exactvendor)
|
|
assetid = request.args.get('assetid', type=int)
|
|
if assetid:
|
|
query = (query.join(WarrantyAsset, WarrantyAsset.warrantyid == Warranty.warrantyid)
|
|
.filter(WarrantyAsset.assetid == assetid))
|
|
warranties = (query.options(joinedload(Warranty.links))
|
|
.order_by(Warranty.enddate.is_(None), Warranty.enddate).all())
|
|
|
|
# Batch-fetch every linked asset in ONE query (was N+1: a db.session.get per
|
|
# link per warranty, ~1.8s for the full list).
|
|
assetids = {link.assetid for w in warranties for link in w.links}
|
|
assetmap = ({a.assetid: a for a in Asset.query.filter(Asset.assetid.in_(assetids)).all()}
|
|
if assetids else {})
|
|
|
|
today = date.today()
|
|
items = [_warranty_payload(w, today, assetmap) for w in warranties]
|
|
|
|
status_filter = request.args.get('status')
|
|
if status_filter:
|
|
items = [i for i in items if i['status'] == status_filter]
|
|
return success_response(items)
|
|
|
|
|
|
@warranty_bp.route('/asset/<int:assetid>', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def warranties_for_asset(assetid):
|
|
"""Warranties covering one asset (for the asset-detail panel)."""
|
|
links = WarrantyAsset.query.filter_by(assetid=assetid).all()
|
|
today = date.today()
|
|
items = []
|
|
for link in links:
|
|
w = db.session.get(Warranty, link.warrantyid)
|
|
if w and w.isactive:
|
|
items.append(_warranty_payload(w, today))
|
|
return success_response(items)
|
|
|
|
|
|
@warranty_bp.route('/<int:warrantyid>', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def get_warranty(warrantyid):
|
|
warranty = db.session.get(Warranty, warrantyid)
|
|
if not warranty:
|
|
return error_response(ErrorCodes.NOT_FOUND, 'Warranty not found', http_code=404)
|
|
return success_response(_warranty_payload(warranty))
|
|
|
|
|
|
@warranty_bp.route('', methods=['POST'])
|
|
@jwt_required()
|
|
@require_permission('warranty.create')
|
|
def create_warranty():
|
|
data = request.get_json() or {}
|
|
vendor = (data.get('vendor') or '').strip()
|
|
if not vendor:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'vendor is required')
|
|
warranty = Warranty(
|
|
vendor=vendor,
|
|
servicetag=(data.get('servicetag') or '').strip() or None,
|
|
provider=(data.get('provider') or 'manual').strip().lower(),
|
|
servicelevel=(data.get('servicelevel') or '').strip() or None,
|
|
startdate=_parse_date(data.get('startdate')),
|
|
enddate=_parse_date(data.get('enddate')),
|
|
notes=(data.get('notes') or '').strip() or None,
|
|
)
|
|
_apply_links(warranty, data.get('assetids'))
|
|
db.session.add(warranty)
|
|
db.session.commit()
|
|
return success_response(_warranty_payload(warranty), message='Warranty created', http_code=201)
|
|
|
|
|
|
@warranty_bp.route('/<int:warrantyid>', methods=['PUT'])
|
|
@jwt_required()
|
|
@require_permission('warranty.edit')
|
|
def update_warranty(warrantyid):
|
|
warranty = db.session.get(Warranty, warrantyid)
|
|
if not warranty:
|
|
return error_response(ErrorCodes.NOT_FOUND, 'Warranty not found', http_code=404)
|
|
data = request.get_json() or {}
|
|
if 'vendor' in data:
|
|
warranty.vendor = (data['vendor'] or '').strip() or warranty.vendor
|
|
if 'servicetag' in data:
|
|
warranty.servicetag = (data['servicetag'] or '').strip() or None
|
|
if 'provider' in data:
|
|
warranty.provider = (data['provider'] or 'manual').strip().lower()
|
|
if 'servicelevel' in data:
|
|
warranty.servicelevel = (data['servicelevel'] or '').strip() or None
|
|
if 'startdate' in data:
|
|
warranty.startdate = _parse_date(data['startdate'])
|
|
if 'enddate' in data:
|
|
warranty.enddate = _parse_date(data['enddate'])
|
|
if 'notes' in data:
|
|
warranty.notes = (data['notes'] or '').strip() or None
|
|
if 'isactive' in data:
|
|
warranty.isactive = bool(data['isactive'])
|
|
if 'assetids' in data:
|
|
_apply_links(warranty, data['assetids'])
|
|
db.session.commit()
|
|
return success_response(_warranty_payload(warranty), message='Warranty updated')
|
|
|
|
|
|
@warranty_bp.route('/<int:warrantyid>', methods=['DELETE'])
|
|
@jwt_required()
|
|
@require_permission('warranty.delete')
|
|
def delete_warranty(warrantyid):
|
|
warranty = db.session.get(Warranty, warrantyid)
|
|
if not warranty:
|
|
return error_response(ErrorCodes.NOT_FOUND, 'Warranty not found', http_code=404)
|
|
db.session.delete(warranty)
|
|
db.session.commit()
|
|
return success_response(message='Warranty deleted')
|
|
|
|
|
|
# =============================================================================
|
|
# Provider refresh (phase 1: manual only; API providers report not-configured)
|
|
# =============================================================================
|
|
|
|
@warranty_bp.route('/<int:warrantyid>/refresh', methods=['POST'])
|
|
@jwt_required()
|
|
@require_permission('warranty.edit')
|
|
def refresh_warranty(warrantyid):
|
|
warranty = db.session.get(Warranty, warrantyid)
|
|
if not warranty:
|
|
return error_response(ErrorCodes.NOT_FOUND, 'Warranty not found', http_code=404)
|
|
provider = get_provider(warranty.provider)
|
|
try:
|
|
result = provider.lookup(warranty.servicetag, warranty.vendor)
|
|
except (ProviderNotConfigured, WarrantyLookupError) as exc:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, str(exc), http_code=400)
|
|
if not result:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR,
|
|
'This warranty is manual - nothing to refresh.', http_code=400)
|
|
if result.get('servicelevel'):
|
|
warranty.servicelevel = result['servicelevel']
|
|
if result.get('startdate'):
|
|
warranty.startdate = _parse_date(result['startdate'])
|
|
if result.get('enddate'):
|
|
warranty.enddate = _parse_date(result['enddate'])
|
|
warranty.lastcheckeddate = datetime.now(timezone.utc).replace(tzinfo=None)
|
|
db.session.commit()
|
|
return success_response(_warranty_payload(warranty), message='Warranty refreshed')
|
|
|
|
|
|
# =============================================================================
|
|
# Bulk Dell sync - auto-detect Dell hardware by service tag and pull coverage
|
|
# =============================================================================
|
|
|
|
@warranty_bp.route('/sync/dell', methods=['POST'])
|
|
@jwt_required()
|
|
@require_permission('warranty.edit')
|
|
def sync_dell():
|
|
"""Look up every asset serial (= Dell service tag) that has no warranty yet
|
|
and create warranties for the ones Dell recognizes.
|
|
|
|
Only assets that lack a dated warranty are looked up, so re-running is cheap
|
|
and non-Dell serials are filtered out by Dell (they return no coverage).
|
|
"""
|
|
provider = get_provider('dell')
|
|
# ?all=true re-checks assets that already have a dated warranty too.
|
|
recheck_all = request.args.get('all', 'false').lower() == 'true'
|
|
|
|
# Assets already covered by a dated warranty - skipped unless recheck_all.
|
|
covered = set()
|
|
if not recheck_all:
|
|
for link in WarrantyAsset.query.all():
|
|
w = db.session.get(Warranty, link.warrantyid)
|
|
if w and w.isactive and w.enddate:
|
|
covered.add(link.assetid)
|
|
|
|
# Candidate assets: active, have a serial, not already covered.
|
|
candidates = (Asset.query
|
|
.filter(Asset.isactive.is_(True),
|
|
Asset.serialnumber.isnot(None),
|
|
Asset.serialnumber != '')
|
|
.all())
|
|
by_tag = {}
|
|
for asset in candidates:
|
|
if asset.assetid in covered:
|
|
continue
|
|
by_tag.setdefault(asset.serialnumber.strip().upper(), []).append(asset.assetid)
|
|
|
|
if not by_tag:
|
|
return success_response({
|
|
'candidates': 0, 'matched': 0, 'created': 0, 'updated': 0,
|
|
}, message='No PCs need a warranty lookup.')
|
|
|
|
try:
|
|
results = provider.bulk_lookup(list(by_tag.keys()))
|
|
except (ProviderNotConfigured, WarrantyLookupError) as exc:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, str(exc), http_code=400)
|
|
|
|
created = updated = matched = 0
|
|
now = datetime.now(timezone.utc).replace(tzinfo=None)
|
|
for tag, assetids in by_tag.items():
|
|
found = results.get(tag)
|
|
if not found:
|
|
continue
|
|
matched += 1
|
|
for assetid in assetids:
|
|
# Reuse an existing Dell warranty for this asset if there is one.
|
|
existing = None
|
|
for link in WarrantyAsset.query.filter_by(assetid=assetid).all():
|
|
candidate = db.session.get(Warranty, link.warrantyid)
|
|
# Reuse a warranty that is Dell by ANY signal, not just
|
|
# provider: a manually-added or imported Dell warranty carries
|
|
# provider 'manual' (create default) but the same service tag or
|
|
# a "Dell" vendor. Matching only provider=='dell' made re-check
|
|
# duplicate every one of those instead of updating it.
|
|
if candidate and (
|
|
candidate.provider == 'dell'
|
|
or (candidate.servicetag or '').strip().upper() == tag
|
|
or (candidate.vendor or '').strip().lower() == 'dell'):
|
|
existing = candidate
|
|
break
|
|
if existing:
|
|
# Canonicalize to Dell so later re-checks match by provider and
|
|
# never fall through to a duplicate.
|
|
existing.provider = 'dell'
|
|
existing.vendor = existing.vendor or 'Dell'
|
|
existing.servicelevel = found.get('servicelevel')
|
|
existing.startdate = _parse_date(found.get('startdate'))
|
|
existing.enddate = _parse_date(found.get('enddate'))
|
|
existing.servicetag = tag
|
|
existing.lastcheckeddate = now
|
|
updated += 1
|
|
else:
|
|
warranty = Warranty(
|
|
vendor='Dell', provider='dell', servicetag=tag,
|
|
servicelevel=found.get('servicelevel'),
|
|
startdate=_parse_date(found.get('startdate')),
|
|
enddate=_parse_date(found.get('enddate')),
|
|
lastcheckeddate=now,
|
|
)
|
|
warranty.links.append(WarrantyAsset(assetid=assetid))
|
|
db.session.add(warranty)
|
|
created += 1
|
|
|
|
db.session.commit()
|
|
return success_response({
|
|
'candidates': sum(len(v) for v in by_tag.values()),
|
|
'tags': len(by_tag),
|
|
'matched': matched,
|
|
'created': created,
|
|
'updated': updated,
|
|
}, message=f'Dell sync: {created} added, {updated} updated, {matched} tags matched.')
|
|
|
|
|
|
# =============================================================================
|
|
# Report buckets (for the Reports hub)
|
|
# =============================================================================
|
|
|
|
@warranty_bp.route('/report', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def warranty_report():
|
|
"""Counts + lists bucketed by derived status."""
|
|
today = date.today()
|
|
buckets = {'expired': [], 'expiring': [], 'active': [], 'unknown': []}
|
|
for w in Warranty.query.filter_by(isactive=True).all():
|
|
buckets.setdefault(w.status(today), []).append(_warranty_payload(w, today))
|
|
return success_response({
|
|
'counts': {k: len(v) for k, v in buckets.items()},
|
|
'buckets': buckets,
|
|
})
|
|
|
|
|
|
@warranty_bp.route('/dashboard/expiring', methods=['GET'])
|
|
@jwt_required()
|
|
@require_permission('warranty.view')
|
|
def dashboard_expiring():
|
|
"""Warranties running out, and ones that already have.
|
|
|
|
Identified the way the floor identifies them: the PC's hostname and the
|
|
MACHINE it drives. Nobody walks out looking for an asset number, and a
|
|
warranty row for a bay PC is only actionable if you know which bay.
|
|
|
|
No dates. "Expired" or "expiring" is the whole decision - the exact day
|
|
matters when you are placing the order, which is what the warranty report
|
|
is for, not when you are scanning a board.
|
|
|
|
Already-expired stays listed rather than dropping off the day it lapses,
|
|
which is how one gets missed entirely.
|
|
"""
|
|
from shopdb.api import Setting
|
|
|
|
days = 90
|
|
setting = Setting.query.filter_by(key='warranty_expiringdays').first()
|
|
if setting and (setting.value or '').strip():
|
|
try:
|
|
days = int(setting.value)
|
|
except (TypeError, ValueError):
|
|
pass
|
|
|
|
today = date.today()
|
|
horizon = today + timedelta(days=days)
|
|
|
|
rows = []
|
|
query = (db.session.query(Warranty, Asset)
|
|
.join(WarrantyAsset, WarrantyAsset.warrantyid == Warranty.warrantyid)
|
|
.join(Asset, Asset.assetid == WarrantyAsset.assetid)
|
|
.filter(Warranty.isactive.is_(True),
|
|
Warranty.enddate.isnot(None),
|
|
Warranty.enddate <= horizon,
|
|
Asset.isactive.is_(True)))
|
|
|
|
for warranty, asset in query.all():
|
|
remaining = (warranty.enddate - today).days
|
|
# Reuses the machine lookup behind the warranty page's machine column,
|
|
# so the board and the report agree about which bay a PC belongs to.
|
|
machine = _related_machine(asset)
|
|
rows.append({
|
|
'assetid': asset.assetid,
|
|
# Hostname, else the asset number - never asset.name. The name
|
|
# of a covered PC's ASSET is often the machine's descriptive name
|
|
# ("Haas VF-2"), so falling back to it put a machine name in a
|
|
# column that is supposed to identify the PC.
|
|
'hostname': _hostname(asset) or asset.assetnumber or str(asset.assetid),
|
|
'machinenumber': machine['machinenumber'] if machine else None,
|
|
'state': 'expired' if remaining < 0 else 'expiring',
|
|
'daysleft': remaining,
|
|
})
|
|
|
|
rows.sort(key=lambda r: r['daysleft'])
|
|
return success_response(rows[:50])
|
|
|
|
|
|
def _hostname(asset):
|
|
"""The PC's hostname, when the covered asset is a PC.
|
|
|
|
Optional import: a lean site may not install the computers plugin, and a
|
|
warranty on a printer or a machine has no hostname at all. Falls back to
|
|
the asset number, which is what the row shows either way.
|
|
"""
|
|
try:
|
|
from plugins.computers.models import Computer
|
|
except ImportError:
|
|
return None
|
|
computer = Computer.query.filter_by(assetid=asset.assetid).first()
|
|
return computer.hostname if computer else None
|